# Week 3 | 第3课：实战 —— 多步骤决策流程

**课程编号**: 3.3
**时长**: 45 分钟
**前置**: 3.2 LangGraph 基础

---

## 学习目标

- 搭建一个完整的客服分类系统（意图分类 → 路由 → 处理 → 满意度检查 → 循环）
- 掌握 LangGraph 中的循环实现和循环次数限制
- 理解如何在 State 中累积消息历史
- 学会跟踪不同执行路径

---

## 1. 项目概述

我们要构建一个完整的客服工单处理系统，流程如下：

```
┌──────────────┐
│  Start       │
└──────┬───────┘
       │ 用户输入
       ▼
┌──────────────┐
│  意图分类     │ ── 调用 LLM 判断问题类型
└──────┬───────┘
       │
   ┌───┴──────┬──────────┐
   ▼          ▼          ▼
┌─────┐   ┌──────┐   ┌────────┐
│账单  │   │技术   │   │ 其他    │
└──┬──┘   └──┬───┘   └───┬────┘
   │         │           │
   └─────────┼───────────┘
             ▼
     ┌──────────────┐
     │  生成回复     │ ── 根据类别生成专业回复
     └──────┬───────┘
            │
            ▼
     ┌──────────────┐
     │ 满意度检查    │ ── 判断用户是否满意
     └──────┬───────┘
            │
      ┌─────┴─────┐
      ▼           ▼
   满意 End    不满意 ──► 回到意图分类（循环，最多3次）
```

---

## 2. 完整实现

### 2.1 定义 State

```python
"""
lesson3_3_customer_triage.py
完整的客服分类系统：意图分类 → 路由 → 处理 → 满意度 → 循环
"""

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# ========== State 定义 ==========
class TriageState(TypedDict):
    # 消息历史（使用 add_messages 自动合并，避免覆盖）
    messages: Annotated[List, add_messages]
    # 当前分类结果
    intent: str
    # 当前回复内容
    reply: str
    # 满意度：True=满意, False=不满意
    satisfied: bool
    # 循环计数器（防止无限循环）
    retry_count: int
    # 最大重试次数
    max_retries: int
```

**关键点**：`messages` 字段使用了 `Annotated[List, add_messages]`。这是 LangGraph 提供的 reducer，它会自动将新消息追加到列表，而不是覆盖。这样我们就有了完整的对话历史。

### 2.2 定义所有 Node

```python
# ========== 初始化 LLM ==========
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ========== Node 1: 意图分类 ==========
def classify_intent(state: TriageState) -> dict:
    """使用 LLM 将用户问题分类为 billing / tech / support"""

    # 获取最新的用户消息
    last_message = state["messages"][-1]
    if isinstance(last_message, HumanMessage):
        question = last_message.content
    else:
        question = str(last_message)

    system_prompt = """你是一个客服意图分类器。将用户问题分类为以下三类之一：
- billing: 账单、付款、发票、费用、退款相关
- tech: 技术故障、功能使用、API、集成、bug 相关
- support: 其他一般性客服问题

只返回类别名称（billing/tech/support），不要返回其他内容。"""

    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=question)
    ])

    intent = response.content.strip().lower()
    print(f"[分类] 意图: {intent}")

    return {"intent": intent}


# ========== Node 2/3/4: 各类别的处理 ==========
def handle_billing(state: TriageState) -> dict:
    """处理账单类问题"""
    system_prompt = """你是账单专家。请专业、友好地回答用户关于账单、付款、发票的问题。
如果涉及具体金额，请仔细核对该信息。如果不确定，建议用户查看账单详情或联系财务。"""

    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=state["messages"][-1].content)
    ])

    print(f"[账单处理] 生成回复")
    return {"reply": response.content}


def handle_tech(state: TriageState) -> dict:
    """处理技术类问题"""
    system_prompt = """你是技术支持专家。请按以下步骤回答：
1. 确认用户遇到的具体问题
2. 提供清晰的解决步骤（用编号列表）
3. 如果问题无法远程解决，建议联系技术团队

回答要专业、条理清晰。"""

    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=state["messages"][-1].content)
    ])

    print(f"[技术处理] 生成回复")
    return {"reply": response.content}


def handle_support(state: TriageState) -> dict:
    """处理一般客服问题"""
    system_prompt = """你是友好的客服代表。请热情地回答用户的问题。
如果问题超出你的知识范围，礼貌地建议用户转接专人处理。"""

    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=state["messages"][-1].content)
    ])

    print(f"[一般处理] 生成回复")
    return {"reply": response.content}


# ========== Node 5: 发送回复并检查满意度 ==========
def check_satisfaction(state: TriageState) -> dict:
    """将回复发送给用户，并检查是否满意"""

    reply = state["reply"]
    print(f"\n{'='*50}")
    print(f"客服: {reply}")
    print(f"{'='*50}")

    # 模拟用户满意度判断
    # 实际项目中，这里应该：
    # 方案A: 直接询问用户（交互式）
    # 方案B: 用 LLM 分析用户后续消息判断是否满意
    # 这里我们演示方案B的简化版本

    # 简单规则：如果用户消息中包含负面词汇，认为不满意
    last_user_msg = ""
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            last_user_msg = msg.content
            break

    negative_words = ["不行", "没解决", "还是不对", "不满意", "不对", "错误", "没用", "help"]
    is_satisfied = not any(w in last_user_msg for w in negative_words)

    print(f"[满意度] {'满意' if is_satisfied else '不满意'}")

    return {
        "satisfied": is_satisfied,
        "messages": [AIMessage(content=reply)],
    }


# ========== Node 6: 处理不满意的情况 ==========
def handle_retry(state: TriageState) -> dict:
    """用户不满意时的处理：记录重试"""
    retry_count = state["retry_count"] + 1
    print(f"[重试] 第 {retry_count} 次重试")

    # 添加一条系统消息，提示重新分类
    retry_message = HumanMessage(content="用户表示不满意，请重新分析问题并提供更好的解决方案。")

    return {
        "retry_count": retry_count,
        "messages": [retry_message],
    }
```

### 2.3 路由函数

```python
# ========== 路由函数 1: 按意图路由 ==========
def route_by_intent(state: TriageState) -> str:
    """根据意图分类路由到对应的处理节点"""
    return state["intent"]


# ========== 路由函数 2: 满意度检查后路由 ==========
def route_by_satisfaction(state: TriageState) -> str:
    """根据满意度决定：结束、重试还是强制结束"""
    if state["satisfied"]:
        return "end"

    # 不满意，检查是否超过最大重试次数
    if state["retry_count"] >= state.get("max_retries", 3):
        return "max_retries"

    # 不满意但未超过重试次数，回到分类
    return "retry"
```

### 2.4 构建完整的 Graph

```python
# ========== 构建 Graph ==========
builder = StateGraph(TriageState)

# 1. 添加所有节点
builder.add_node("classify", classify_intent)
builder.add_node("billing", handle_billing)
builder.add_node("tech", handle_tech)
builder.add_node("support", handle_support)
builder.add_node("check_satisfaction", check_satisfaction)
builder.add_node("handle_retry", handle_retry)

# 2. 设置入口点
builder.set_entry_point("classify")

# 3. 意图分类后的条件路由
builder.add_conditional_edges(
    "classify",
    route_by_intent,
    {
        "billing": "billing",
        "tech": "tech",
        "support": "support",
    }
)

# 4. 各处理节点完成后，都进入满意度检查
builder.add_edge("billing", "check_satisfaction")
builder.add_edge("tech", "check_satisfaction")
builder.add_edge("support", "check_satisfaction")

# 5. 满意度检查后的条件路由
builder.add_conditional_edges(
    "check_satisfaction",
    route_by_satisfaction,
    {
        "end": END,
        "retry": "handle_retry",
        "max_retries": END,  # 超过最大重试次数也结束
    }
)

# 6. 重试后回到分类节点（形成循环）
builder.add_edge("handle_retry", "classify")

# 7. 编译
triage_app = builder.compile()
```

### 2.5 运行和测试

```python
# ========== 运行测试 ==========
def run_triage(question: str, max_retries: int = 3):
    """运行一次完整的客服分类流程"""
    print(f"\n{'#'*60}")
    print(f"用户: {question}")
    print(f"{'#'*60}")

    initial_state = {
        "messages": [HumanMessage(content=question)],
        "intent": "",
        "reply": "",
        "satisfied": False,
        "retry_count": 0,
        "max_retries": max_retries,
    }

    result = triage_app.invoke(initial_state)

    print(f"\n{'='*60}")
    print(f"最终回复: {result['reply']}")
    print(f"满意度: {'满意' if result['satisfied'] else '不满意（已达最大重试次数）'}")
    print(f"重试次数: {result['retry_count']}")
    print(f"{'='*60}")

    return result


if __name__ == "__main__":
    print("=" * 60)
    print("       客服分类系统演示")
    print("=" * 60)

    # 测试场景 1: 账单问题（满意）
    run_triage("我这个月的账单怎么多了50块钱？能帮我查一下吗？")

    # 测试场景 2: 技术问题（满意）
    run_triage("你们的 API 一直返回 500 错误，已经持续半小时了")

    # 测试场景 3: 一般问题
    run_triage("请问你们的办公时间是什么？")

    # 测试场景 4: 不满意场景（模拟）
    # 注意：这个需要交互式运行，下面用 mock 方式演示
    print("\n\n===== 不满意场景演示（Mock）=====")
    print("用户: 这个功能完全不能用，太糟糕了")
    print("  → 分类: tech")
    print("  → 技术处理: 生成解决方案")
    print("  → 满意度: 不满意（检测到负面词汇）")
    print("  → 重试第1次 → 重新分类 → 再次处理")
    print("  → 如果仍然不满意 → 重试第2次...")
    print("  → 超过 max_retries → 强制结束，转人工")
```

---

## 3. 不同执行路径演示

### 3.1 路径 1: 账单问题 → 满意

```
用户: "我这个月的账单怎么多了50块钱？"

[分类] 意图: billing
[账单处理] 生成回复

==================================================
客服: 您好！关于账单金额差异的问题，我来帮您核查。
      账单差异可能有以下原因：
      1. 本月有新增的服务项目
      2. 上月优惠本月到期
      3. 用量超出免费额度
      建议您登录系统查看详细账单明细。如需进一步核查，
      我可以帮您转接财务部门。
==================================================

[满意度] 满意
→ END
```

### 3.2 路径 2: 技术问题 → 不满意 → 重试 → 满意

```
用户: "API 一直返回 500 错误"

[分类] 意图: tech
[技术处理] 生成回复
[满意度] 不满意（用户回复"还是不对"）
[重试] 第 1 次重试
[分类] 意图: tech
[技术处理] 生成回复（更详细的方案）
[满意度] 满意
→ END
```

### 3.3 路径 3: 超过最大重试次数

```
用户: "什么都不好用"

[分类] 意图: tech
[技术处理] 生成回复
[满意度] 不满意
[重试] 第 1 次重试
...（重复）
[重试] 第 2 次重试
...（重复）
[重试] 第 3 次重试
[满意度] 不满意
→ END（已达 max_retries，建议转人工）
```

---

## 4. 可视化完整流程

```python
# 生成完整流程图
print(triage_app.get_graph().draw_mermaid())
```

对应的流程图：

```mermaid
graph TD
    __start__ --> classify
    classify -->|billing| billing
    classify -->|tech| tech
    classify -->|support| support
    billing --> check_satisfaction
    tech --> check_satisfaction
    support --> check_satisfaction
    check_satisfaction -->|end| __end__
    check_satisfaction -->|retry| handle_retry
    check_satisfaction -->|max_retries| __end__
    handle_retry --> classify
```

---

## 5. 使用 Mock 模式（无 LLM 也能运行）

如果没有 API Key 或想快速测试流程逻辑，可以用 Mock 模式：

```python
"""
lesson3_3_mock_triage.py
Mock 版本的客服分类系统，无需 LLM API
"""

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage
import time

class MockState(TypedDict):
    messages: Annotated[List, add_messages]
    intent: str
    reply: str
    satisfied: bool
    retry_count: int
    max_retries: int

# Mock 回复模板
MOCK_RESPONSES = {
    "billing": "【账单】关于您的问题，系统显示本月账单有一笔增值服务费用。如需明细，可在'我的账单'中查看。",
    "tech": "【技术】建议您先尝试清除缓存并重新登录。如果问题仍存在，请提供错误截图。",
    "support": "【客服】感谢您的咨询。我们的办公时间是周一至周五 9:00-18:00。",
}

def mock_classify(state: MockState) -> dict:
    """Mock 分类：基于关键词"""
    question = state["messages"][-1].content.lower()
    if "账单" in question or "钱" in question or "费用" in question:
        intent = "billing"
    elif "错误" in question or "不能用" in question or "故障" in question:
        intent = "tech"
    else:
        intent = "support"
    print(f"[分类] 意图: {intent}")
    return {"intent": intent}

def mock_handle(state: MockState) -> dict:
    """Mock 处理：返回预设回复"""
    intent = state["intent"]
    reply = MOCK_RESPONSES.get(intent, "【客服】收到您的问题，正在处理中。")
    print(f"[{intent.upper()}] 生成回复")
    return {"reply": reply}

def mock_satisfaction(state: MockState) -> dict:
    """Mock 满意度：基于关键词判断"""
    last_msg = state["messages"][-1].content if state["messages"] else ""
    negative = ["不行", "没解决", "还是不对", "不满意", "不对", "错误"]
    satisfied = not any(w in last_msg for w in negative)
    print(f"[满意度] {'满意' if satisfied else '不满意'}")
    reply = state["reply"]
    return {
        "satisfied": satisfied,
        "messages": [AIMessage(content=reply)],
    }

def mock_retry(state: MockState) -> dict:
    retry_count = state["retry_count"] + 1
    print(f"[重试] 第 {retry_count} 次重试")
    return {
        "retry_count": retry_count,
        "messages": [HumanMessage(content="用户不满意，请提供更好的方案。")],
    }

def route_by_intent(state): return state["intent"]

def route_by_satisfaction(state):
    if state["satisfied"]: return "end"
    if state["retry_count"] >= state.get("max_retries", 3): return "max_retries"
    return "retry"

# 构建图
builder = StateGraph(MockState)
builder.add_node("classify", mock_classify)
builder.add_node("handle", mock_handle)  # 统一处理节点
builder.add_node("check", mock_satisfaction)
builder.add_node("retry", mock_retry)

builder.set_entry_point("classify")
builder.add_conditional_edges("classify", route_by_intent, {
    "billing": "handle", "tech": "handle", "support": "handle"
})
builder.add_edge("handle", "check")
builder.add_conditional_edges("check", route_by_satisfaction, {
    "end": END, "retry": "retry", "max_retries": END
})
builder.add_edge("retry", "classify")

mock_graph = builder.compile()

if __name__ == "__main__":
    # 测试 1: 账单（满意）
    result = mock_graph.invoke({
        "messages": [HumanMessage(content="我这个月账单多了50块钱？")],
        "intent": "", "reply": "", "satisfied": False,
        "retry_count": 0, "max_retries": 3,
    })
    print(f"结果: {result['reply']}\n")

    # 测试 2: 不满意场景
    result = mock_graph.invoke({
        "messages": [HumanMessage(content="这个功能完全不能用，不行！")],
        "intent": "", "reply": "", "satisfied": False,
        "retry_count": 0, "max_retries": 2,
    })
    print(f"最终: 重试 {result['retry_count']} 次")
```

---

## 6. 动手练习

### 练习 1：扩展分类类别

在现有系统上增加一个新的类别：`emergency`（紧急问题），路由逻辑如下：
- 如果用户消息中包含"紧急"、"严重影响"、"系统崩溃"等关键词，分类为 emergency
- emergency 问题的处理节点应该返回"已升级为紧急工单，技术团队将在 30 分钟内响应"
- emergency 问题**不需要**满意度检查，直接结束

### 练习 2：添加执行路径日志

在 State 中增加一个 `execution_path: List[str]` 字段，在每个节点执行时追加节点名称，最终输出完整的执行路径：

```
执行路径: classify → tech → check_satisfaction → retry → classify → tech → check_satisfaction → end
```

<details>
<summary>提示代码</summary>

```python
class TriageState(TypedDict):
    # ... 其他字段 ...
    execution_path: list  # 记录执行路径

# 在每个节点中追加：
def classify_intent(state: TriageState) -> dict:
    path = state.get("execution_path", [])
    return {"intent": intent, "execution_path": path + ["classify"]}
```

</details>

### 练习 3：实现交互式满意度

将 Mock 满意度改为交互式：使用 `input()` 让用户在实际运行时输入反馈，判断是否满意：

```python
def interactive_satisfaction(state: TriageState) -> dict:
    reply = state["reply"]
    print(f"\n客服: {reply}")
    user_feedback = input("\n是否满意？(y/n): ").strip().lower()
    satisfied = user_feedback == "y"
    return {
        "satisfied": satisfied,
        "messages": [AIMessage(content=reply)],
    }
```

---

## 7. 小结

本课要点：

- **完整的多步骤流程**：分类 → 路由 → 处理 → 满意度 → 循环，每个环节都是独立的节点
- **循环的实现**：通过条件边将 `handle_retry` 连接回 `classify`，形成闭环
- **防止死循环**：用 `retry_count` 和 `max_retries` 控制最大循环次数
- **消息历史累积**：使用 `Annotated[List, add_messages]` 让 State 自动追加消息
- **Mock 模式**：在没有 LLM 的情况下也能测试流程逻辑
- **可视化**：`get_graph().draw_mermaid()` 让复杂流程一目了然

**下节课预告**: 给这个 Agent 加一个 Web UI，让非技术人员也能通过浏览器使用。