10 实战：用 Handoffs 搭一套客服工单流转（一个 Agent + 状态机式步骤）

1
from langchain.agents import AgentState
2
from typing_extensions import NotRequired
3
from typing import Literal
4

5
# 工作流里允许出现的步骤名，写成 Literal，拼错立刻报错
6
SupportStep = Literal["warranty_collector", "issue_classifier", "resolution_specialist"]
7

8

9
class SupportState(AgentState):
10
    """客服工作流的状态。"""
11

12
    current_step: NotRequired[SupportStep]          # 核心：当前处于哪一步
13
    warranty_status: NotRequired[Literal["in_warranty", "out_of_warranty"]]
14
    issue_type: NotRequired[Literal["hardware", "software"]]

1
from langchain.tools import tool, ToolRuntime
2
from langchain.messages import ToolMessage
3
from langgraph.types import Command
4

5
@tool
6
def record_warranty_status(
7
    status: Literal["in_warranty", "out_of_warranty"],
8
    runtime: ToolRuntime[None, SupportState],
9
) -> Command:
10
    """记录客户保修状态，并流转到问题分类步骤。"""
11
    return Command(
12
        update={
13
            # 必须自己补一条 ToolMessage，模型才看得到工具执行结果
14
            "messages": [
15
                ToolMessage(
16
                    content=f"Warranty status recorded as: {status}",
17
                    tool_call_id=runtime.tool_call_id,
18
                )
19
            ],
20
            "warranty_status": status,
21
            "current_step": "issue_classifier",   # 状态迁移就发生在这一行
22
        }
23
    )

1
WARRANTY_COLLECTOR_PROMPT = """你是客服，负责处理设备问题。
2

3
当前步骤：核实保修状态（Warranty verification）
4

5
这一步你要做：先友好打招呼 → 询问设备是否在保修期内 → 调用 record_warranty_status
6
记录答案并进入下一步。语气自然，一次只问一个问题，不要连珠炮。"""
7

8
# 另两份提示词结构相同，只是「已知信息」和「要做什么」不同（完整版见原文）
9
ISSUE_CLASSIFIER_PROMPT = """……
10
当前步骤：问题分类（Issue classification）
11
已知信息：保修状态 = {warranty_status}
12
……（让客户描述问题 → 判断硬件/软件 → 调 record_issue_type）"""
13

14
RESOLUTION_SPECIALIST_PROMPT = """……
15
当前步骤：解决方案（Resolution）
16
已知信息：保修状态 = {warranty_status}，问题类型 = {issue_type}
17
……（软件给排障步骤；硬件在保给保修维修；硬件过保转人工）"""

1
# 步骤配置：步骤名 → (提示词, 工具集, 前置必需字段)
2
STEP_CONFIG = {
3
    "warranty_collector": {
4
        "prompt": WARRANTY_COLLECTOR_PROMPT,
5
        "tools": [record_warranty_status],
6
        "requires": [],
7
    },
8
    "issue_classifier": {
9
        "prompt": ISSUE_CLASSIFIER_PROMPT,
10
        "tools": [record_issue_type],
11
        "requires": ["warranty_status"],          # 没拿到保修状态不许进来
12
    },
13
    "resolution_specialist": {
14
        "prompt": RESOLUTION_SPECIALIST_PROMPT,
15
        "tools": [provide_solution, escalate_to_human],
16
        "requires": ["warranty_status", "issue_type"],
17
    },
18
}

1
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
2
from typing import Callable
3

4

5
@wrap_model_call
6
def apply_step_config(
7
    request: ModelRequest,
8
    handler: Callable[[ModelRequest], ModelResponse],
9
) -> ModelResponse:
10
    """按当前步骤配置 Agent 行为。"""
11
    # 取当前步骤；首轮没有该字段，默认从保修登记开始
12
    current_step = request.state.get("current_step", "warranty_collector")
13
    step_config = STEP_CONFIG[current_step]          # 查表
14
    # 前置依赖校验：该有的数据没有就直接报错，别让模型瞎猜
15
    for key in step_config["requires"]:
16
        if request.state.get(key) is None:
17
            raise ValueError(f"{key} must be set before reaching {current_step}")
18
    # 用 state 里的值渲染提示词模板
19
    system_prompt = step_config["prompt"].format(**request.state)
20
    # 覆盖系统提示词与可用工具集：不新建 Agent，只改这一次调用的配置
21
    request = request.override(
22
        system_prompt=system_prompt,
23
        tools=step_config["tools"],
24
    )
25
    return handler(request)

1
from langchain.agents import create_agent
2
from langgraph.checkpoint.memory import InMemorySaver
3

4
# 所有步骤用到的工具汇总注册；注册 ≠ 可用，实际可见范围由中间件裁剪
5
all_tools = [
6
    record_warranty_status,
7
    record_issue_type,
8
    provide_solution,
9
    escalate_to_human,
10
]
11

12
agent = create_agent(
13
    model,
14
    tools=all_tools,
15
    state_schema=SupportState,     # 用自定义 state
16
    middleware=[apply_step_config],
17
    checkpointer=InMemorySaver(),  # 跨轮次保持状态
18
)

1
from langchain.messages import HumanMessage
2
from langchain_core.utils.uuid import uuid7
3

4
thread_id = str(uuid7())
5
config = {"configurable": {"thread_id": thread_id}}   # 同一个 thread 才算同一通对话
6

7
# 第 1 轮：客户只说症状，流程还没到分类那一步
8
result = agent.invoke({"messages": [HumanMessage("Hi, my phone screen is cracked")]}, config)
9

10
# 第 2 轮：回答保修问题 → record_warranty_status 写回 current_step = issue_classifier
11
result = agent.invoke({"messages": [HumanMessage("Yes, it's still under warranty")]}, config)
12
print(f"Current step: {result.get('current_step')}")
13

14
# 第 3 轮：描述问题 → record_issue_type 写回 current_step = resolution_specialist
15
# ...（原文还有第 4 轮：用户问 What should I do?，Agent 给出保修维修方案）
16
for msg in result["messages"]:
17
    msg.pretty_print()

1
# 第 2 轮：record_warranty_status 的产物
2
Command(update={
3
    "warranty_status": "in_warranty",
4
    "current_step": "issue_classifier",       # 状态迁移！
5
})
6
# 第 3 轮：record_issue_type 的产物
7
Command(update={
8
    "issue_type": "hardware",
9
    "current_step": "resolution_specialist",  # 状态迁移！
10
})

1
from langchain.agents.middleware import SummarizationMiddleware
2

3
# 在 §5 的 create_agent 上，middleware 里再加一个摘要中间件
4
agent = create_agent(
5
    model, tools=all_tools, state_schema=SupportState,
6
    middleware=[
7
        apply_step_config,                     # 先切步骤配置
8
        SummarizationMiddleware(               # 再做历史压缩：用便宜的小模型摘要
9
            model="gpt-5.4-mini",
10
            trigger=("tokens", 4000),          # 超过 4000 token 才触发
11
            keep=("messages", 10),             # 最近 10 条原样保留
12
        ),
13
    ],
14
    checkpointer=InMemorySaver(),
15
)

1
@tool
2
def go_back_to_warranty() -> Command:
3
    """回到保修核实步骤。"""
4
    return Command(update={"current_step": "warranty_collector"})
5

6
@tool
7
def go_back_to_classification() -> Command:
8
    """回到问题分类步骤。"""
9
    return Command(update={"current_step": "issue_classifier"})
10

11
# 把这两个工具挂到解决方案这一步上
12
STEP_CONFIG["resolution_specialist"]["tools"].extend([
13
    go_back_to_warranty,
14
    go_back_to_classification,
15
])

1
如果客户表示之前提供的信息有误，使用：
2
- go_back_to_warranty 纠正保修状态
3
- go_back_to_classification 纠正问题类型

1
result = agent.invoke(
2
    {"messages": [HumanMessage("Actually, I made a mistake - my device is out of warranty")]},
3
    config,
4
)  # 触发回退