交接：客户支持

概述

状态机模式描述了代理在任务的不同状态之间切换时行为发生变化的工作流。本教程展示了如何通过工具调用来动态改变单个代理的配置——根据当前状态更新其可用的工具和指令——从而实现一个状态机。状态可以从多个来源确定：代理的过去动作（工具调用）、外部状态（如API调用结果），甚至是初始用户输入（例如，通过运行分类器确定用户意图）。

在本教程中，您将构建一个客户支持代理，实现以下功能：

在处理之前收集保修信息。
将问题分类为硬件或软件问题。
提供解决方案或上报给人工支持。
跨多轮对话维护会话状态。

与子代理模式（子代理作为工具被调用）不同，状态机模式使用一个单一的代理，其配置根据工作流的进展而变化。每个"步骤"只是同一底层代理的不同配置（系统提示 + 工具），根据状态动态选择。

以下是您将要构建的工作流：

mermaid

flowchart TD
    Start([💬 客户报告问题]) --> Warranty{设备是否<br>在保修期内？}

    Warranty -->|✅ 是| IssueType{问题类型？}
    Warranty -->|❌ 否| OutOfWarranty{问题类型？}

    IssueType -->|🔩 硬件| Repair[提供保修<br>维修说明]
    IssueType -->|💻 软件| Troubleshoot[提供故障<br>排除步骤]

    OutOfWarranty -->|🔩 硬件| Escalate[上报人工<br>自费维修选项]
    OutOfWarranty -->|💻 软件| Troubleshoot

    Troubleshoot --> Close([✅ 问题已解决])
    Repair --> Close
    Escalate --> Close

设置

安装

本教程需要 langchain 包：

bash

pip install langchain

更多详情请参阅我们的安装指南。

LangSmith

设置 LangSmith 以观察代理内部的运行情况：

python

import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

选择模型

选择一个聊天模型：

python

from langchain.chat_models import init_chat_model

# OpenAI
model = init_chat_model("openai:gpt-5.4")

# Anthropic
model = init_chat_model("anthropic:claude-sonnet-4-6")

# Google Gemini
model = init_chat_model("google_genai:gemini-2.5-flash-lite")

1. 定义自定义状态

首先，定义一个自定义状态模式，追踪当前激活的是哪个步骤：

python

from langchain.agents import AgentState
from typing_extensions import NotRequired
from typing import Literal

# 定义可能的工作流步骤
SupportStep = Literal["warranty_collector", "issue_classifier", "resolution_specialist"]

class SupportState(AgentState):
    """客户支持工作流的状态。"""
    current_step: NotRequired[SupportStep]
    warranty_status: NotRequired[Literal["in_warranty", "out_of_warranty"]]
    issue_type: NotRequired[Literal["hardware", "software"]]

current_step 字段是状态机模式的核心——它决定了每轮对话中加载哪个配置（提示 + 工具）。

2. 创建工作流状态管理工具

创建更新工作流状态的工具。这些工具允许代理记录信息并转换到下一步。

关键在于使用 Command 来更新状态，包括 current_step 字段：

python

from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage
from langgraph.types import Command

@tool
def record_warranty_status(
    status: Literal["in_warranty", "out_of_warranty"],
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    """记录客户的保修状态并转换到问题分类步骤。"""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"保修状态已记录为：{status}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "warranty_status": status,
            "current_step": "issue_classifier",
        }
    )


@tool
def record_issue_type(
    issue_type: Literal["hardware", "software"],
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    """记录问题类型并转换到解决方案专家步骤。"""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"问题类型已记录为：{issue_type}",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "issue_type": issue_type,
            "current_step": "resolution_specialist",
        }
    )


@tool
def escalate_to_human(reason: str) -> str:
    """将案例上报给人工支持专家。"""
    # 在真实系统中，这将会创建工单、通知员工等
    return f"正在上报给人工支持。原因：{reason}"


@tool
def provide_solution(solution: str) -> str:
    """为客户的问题提供解决方案。"""
    return f"已提供解决方案：{solution}"

注意 record_warranty_status 和 record_issue_type 返回的 Command 对象同时更新了数据（warranty_status、issue_type）和 current_step。这就是状态机的工作方式——工具控制工作流的推进。

3. 定义步骤配置

为每个步骤定义提示词和工具。

python

# 定义每个步骤的提示词
WARRANTY_COLLECTOR_PROMPT = """你是一个处理设备问题的客户支持代理。

当前阶段：保修验证

在此步骤中，你需要：
1. 热情地问候客户
2. 询问他们的设备是否在保修期内
3. 使用 record_warranty_status 记录他们的回答并进入下一步

保持对话式、友好的语气。不要同时问多个问题。"""

ISSUE_CLASSIFIER_PROMPT = """你是一个处理设备问题的客户支持代理。

当前阶段：问题分类
客户信息：保修状态为 {warranty_status}

在此步骤中，你需要：
1. 请客户描述他们的问题
2. 判断是硬件问题（物理损坏、零件故障）还是软件问题（应用崩溃、性能问题）
3. 使用 record_issue_type 记录分类并进入下一步

如果不确定，请先提出澄清性问题再进行分类。"""

RESOLUTION_SPECIALIST_PROMPT = """你是一个处理设备问题的客户支持代理。

当前阶段：解决方案
客户信息：保修状态为 {warranty_status}，问题类型为 {issue_type}

在此步骤中，你需要：
1. 对于软件问题：使用 provide_solution 提供故障排除步骤
2. 对于硬件问题：
   - 如果在保修期内：使用 provide_solution 说明保修维修流程
   - 如果不在保修期内：使用 escalate_to_human 获取自费维修选项

在你的解决方案中要具体且有帮助。"""

# 步骤配置：将步骤名称映射到（提示词、工具、必需状态）
STEP_CONFIG = {
    "warranty_collector": {
        "prompt": WARRANTY_COLLECTOR_PROMPT,
        "tools": [record_warranty_status],
        "requires": [],
    },
    "issue_classifier": {
        "prompt": ISSUE_CLASSIFIER_PROMPT,
        "tools": [record_issue_type],
        "requires": ["warranty_status"],
    },
    "resolution_specialist": {
        "prompt": RESOLUTION_SPECIALIST_PROMPT,
        "tools": [provide_solution, escalate_to_human],
        "requires": ["warranty_status", "issue_type"],
    },
}

这种基于字典的配置使得：

所有步骤一目了然
添加新步骤很容易（只需添加一条新记录）
工作流依赖关系清晰（requires 字段）
可以使用状态变量渲染提示模板（如 {warranty_status}）

4. 创建基于步骤的中间件

创建从状态中读取 current_step 并应用适当配置的中间件：

python

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


@wrap_model_call
def apply_step_config(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """根据当前步骤配置代理行为。"""
    # 获取当前步骤（首次交互默认为 warranty_collector）
    current_step = request.state.get("current_step", "warranty_collector")

    # 查找步骤配置
    stage_config = STEP_CONFIG[current_step]

    # 验证必需状态是否存在
    for key in stage_config["requires"]:
        if request.state.get(key) is None:
            raise ValueError(f"{key} 必须在进入 {current_step} 之前设置")

    # 使用状态值格式化提示词（支持 {warranty_status}、{issue_type} 等）
    system_prompt = stage_config["prompt"].format(**request.state)

    # 注入系统提示和步骤特定工具
    request = request.override(
        system_prompt=system_prompt,
        tools=stage_config["tools"],
    )

    return handler(request)

这个中间件：

读取当前步骤：从状态中获取 current_step（默认为 "warranty_collector"）。
查找配置：在 STEP_CONFIG 中找到匹配的条目。
验证依赖关系：确保所有必需的状态字段已设置。
格式化提示词：用状态值替换提示模板中的占位符。
注入配置：用当前步骤的提示词和工具覆盖请求。

5. 创建和配置代理

使用中间件创建代理：

python

from langchain.agents import create_agent

support_agent = create_agent(
    model,
    tools=[],  # 工具由中间件动态注入
    system_prompt="",  # 系统提示由中间件动态注入
    middlewares=[apply_step_config],
)

6. 测试工作流

模拟一个完整的客户支持对话：

python

# 初始状态（无状态，因此默认开始于 warranty_collector）
initial_state = SupportState()

for step in support_agent.stream(
    {"messages": [{"role": "user", "content": "我的笔记本电脑出了问题"}],
     "state": initial_state}
):
    for update in step.values():
        for message in update.get("messages", []):
            message.pretty_print()

代理将在保修验证步骤中开始对话，询问客户设备是否在保修期内。一旦客户确认，代理调用 record_warranty_status，状态机会自动推进到问题分类步骤。在后续对话轮次中，代理将拥有不同的工具和提示词。

状态转换演示

python

# 轮次 1：保修收集器
# 代理："您好！很抱歉听到您的笔记本电脑出了问题。请问您的设备还在保修期内吗？"

# 轮次 2：用户回复 "是的，还在保修期内"
# 代理调用 record_warranty_status(status="in_warranty")
# → current_step 变为 "issue_classifier"

# 轮次 3：问题分类器
# 代理："好的，保修期内。请问您遇到的是硬件问题还是软件问题？"

# 轮次 4：用户回复 "屏幕碎了"
# 代理调用 record_issue_type(issue_type="hardware")
# → current_step 变为 "resolution_specialist"

# 轮次 5：解决方案专家
# 代理（保修期内 + 硬件）：
# "很抱歉听到屏幕碎裂。由于您的设备在保修期内，我们可以提供免费维修。"

概念回顾

在本教程中，您学到了：

状态机模式 — 单个代理根据当前工作流状态改变其配置
手递手交接 — 通过工具调用（Command + current_step）控制状态转换
自定义状态模式 — 扩展 AgentState 以跟踪工作流进度
@wrap_model_call 中间件 — 动态注入步骤特定的提示词和工具
字典式步骤配置 — 所有步骤的清晰、可维护的定义

与子代理模式不同，状态机模式在整个工作流中使用一个单一的代理实例。每次"交接"只是同一代理在不同的配置下运行。这使得对话历史得以保持，同时允许在不同的工作流阶段拥有根本不同的行为。

下一步

探索子代理模式，了解监督者协调多个专业代理的方式
学习路由器模式，构建并行查询多个知识源的系统
深入了解LangGraph 状态图
查看人工审核的高级用法

交接：客户支持 ​

概述 ​

设置 ​

安装 ​

LangSmith ​

选择模型 ​

1. 定义自定义状态 ​

2. 创建工作流状态管理工具 ​

3. 定义步骤配置 ​

4. 创建基于步骤的中间件 ​

5. 创建和配置代理 ​

6. 测试工作流 ​

状态转换演示 ​

概念回顾 ​

下一步 ​

交接：客户支持

概述

设置

安装

LangSmith

选择模型

1. 定义自定义状态

2. 创建工作流状态管理工具

3. 定义步骤配置

4. 创建基于步骤的中间件

5. 创建和配置代理

6. 测试工作流

状态转换演示

概念回顾

下一步