电商客服智能体状态机编排：基于 LangGraph 的双轨意图识别与分级降级工程实践

本文基于一个电商订单客服示例项目的真实实现整理。该项目演示了一条完整的工程化路径：把 LangGraph 官方脚手架改造成一个可测试、可降级、可可视化调试的订单处理 Agent，是"从 LangChain 语法走向生产级状态编排"的典型样例。

1. 项目缘起：从官方脚手架到订单客服系统

LangGraph 官方提供了 new-langgraph-project-python 脚手架，它只有一个"输入什么就返回什么"的单节点图，作用是让开发者快速理解 LangGraph Server / Studio 的接入方式。脚手架的价值在于骨架，而不在于逻辑——真正的学习发生在"往骨架上填充业务"的过程，本文剖析的项目就是一份完整的填充答卷。

该项目瞄准的是一段非常典型的电商客服对话：用户抛出一句话，系统需要判断他想干什么，再决定要不要执行系统操作、由谁来回答。业务上预先定义了 7 类意图，形成领域内的"意图词典"：

意图标识	业务含义	典型触发语
query_order	订单查询	“查询订单状态、物流到哪了”
modify_order	订单修改	“改收货地址、换电话”
cancel_order	订单取消	“取消订单、退货”
customer_service	客服咨询/投诉	“需要人工、投诉”
payment_issue	支付问题	“支付失败、退款、发票”
product_inquiry	商品咨询	“这款商品参数怎么样”
unknown	无法归类（兜底）	闲聊等

整条流水线回答一个核心问题：用户输入 → 意图 → 动作 → 话术，中间每一步都可能失败，因此系统围绕"永远要有答案"这一原则设计——这正是下文所有降级逻辑的出发点。

图 1 业务管道总览：实线为主流程，虚线为"外部依赖不可用"时的兜底旁路——两条路径都收敛于"回复用户"，这正是"永远要有答案"原则的直观体现。

2. 理论基石：状态、图与条件路由

在深入代码之前，先建立理论坐标系。LangGraph 构建的 Agent 本质是一个有状态的状态机计算模型，三个概念贯穿始终：

• 状态（State）：工作流运行时的"白板"，所有节点读写它完成数据传递。LangGraph 支持两种状态契约：TypedDict 与 Pydantic 模型，字段可声明 reducer 函数（规约函数）来决定"新值到来时是覆盖还是合并"。例如 add_messages reducer 会把新消息追加到历史而非覆盖，这是多轮对话能够累积上下文的关键机制。未声明 reducer 的字段则遵循"后写覆盖先写"的语义。
• 节点（Node）：工作单元，接收当前状态、返回状态增量。图的计算以**超步（Superstep）**为单位推进：每个超步中，处于激活态的节点执行完毕后向邻居传播状态变更，直至没有新消息产生、全体节点进入 halted 状态。
• 边与条件路由：静态边表达"必然发生"的转移；条件边（add_conditional_edges）把"走哪条路"的决策交给路由函数，返回目标节点名。条件路由是让图具备业务智能的分水岭——路由逻辑可以基于规则、向量相似度或 LLM 判断。

把状态机思想迁移到客服场景，收益是结构化的：意图识别、订单处理、话术生成被拆成独立可测的节点；状态随流程逐步补全（intent → order_info → response），天然形成审计轨迹；任何环节的失败都可以在节点边界上被捕获并降级，而不是让整个流程崩溃。

3. 系统架构：职责分层的四模块设计

本项目没有把逻辑全部塞进图里，而是采用了工程上成熟的分层：

┌────────────────────────────────────────────────────┐
│ graph.py     编排层：节点拓扑 + 条件路由，只做"流程"      │
├────────────────────────────────────────────────────┤
│ services.py  服务层：意图识别 / 通义调用 / 订单操作       │
├────────────────────────────────────────────────────┤
│ models.py    契约层：Pydantic 领域模型                 │
├────────────────────────────────────────────────────┤
│ config.py    配置层：密钥、阈值、意图路由表、节点元数据     │
└────────────────────────────────────────────────────┘

模块	核心职责	设计动机
graph.py	定义 State/Context，组装节点与（条件）边	图只回答"流程怎么走"，可整体可视化
services.py	三类服务：关键词意图识别、通义 LLM 调用、模拟订单库	业务细节与编排解耦，可独立单测
models.py	OrderInfo、IntentResult、LLMResponse、WorkflowState 等 Pydantic 模型	字段契约即文档，运行时自动校验
config.py	API 密钥/模型/阈值 + INTENT_MAPPING/NODE_CONFIG	新增意图只改配置不动代码（开闭原则）

这套分层的直接回报是：services / models / config 全部可以被 LLM 之外的单元测试独立验证，图中真正依赖外部模型的只有薄薄一层。

4. 图即流程：状态与条件路由的代码精讲

4.1 状态的显式建模

class Context(TypedDict):
    """运行时上下文：每次调用（每段会话）可注入的参数"""
    user_id: str
    session_id: str

@dataclass
class State:
    user_input: str = ""
    intent: str = ""            # 意图识别结果
    order_info: Dict[str, Any] = None   # 订单处理结果
    response: str = ""          # 最终话术
    next_action: str = ""       # 处理后的下一步动作标记
    messages: list = None       # 消息历史

    def __post_init__(self):
        if self.order_info is None:
            self.order_info = {}
        if self.messages is None:
            self.messages = []

讲解：代码刻意区分了两种输入通道——State 是业务状态（随节点流转、被各节点读写），Context 是会话上下文（user_id/session_id，由平台或调用方注入，用于多租户隔离与跟踪）。__post_init__ 保证了可变默认值的正确初始化，避免共享引用陷阱。这里的 messages 与 LangGraph 官方 add_messages 的 reducer 追加语义不同（整列表覆盖），适合"消息仅作审计"的轻量场景——状态 schema 的取舍应服务于业务语义，而非机械照搬。

4.2 条件边：把"业务判断"显式化为路由

def should_continue(state: State) -> Literal["order_processing", "__end__"]:
    """意图识别后的路由：识别到的意图 → 订单处理；未知 → 直接结束"""
    if state.intent in ["query_order", "modify_order", "cancel_order", "customer_service"]:
        return "order_processing"
    return "__end__"

def route_after_processing(state: State) -> Literal["tongyi_llm", "__end__"]:
    """订单处理后的路由：客服类场景 → 转人工说明由 LLM 生成；其余结束"""
    if state.next_action == "transfer_to_human":
        return "tongyi_llm"
    return "__end__"

graph = (
    StateGraph(State, context_schema=Context)
    .add_node("intent_recognition", intent_recognition_node)
    .add_node("tongyi_llm", tongyi_llm_node)
    .add_node("order_processing", order_processing_node)
    .add_edge("__start__", "intent_recognition")
    .add_conditional_edges("intent_recognition", should_continue,
                           {"order_processing": "order_processing", "__end__": "__end__"})
    .add_conditional_edges("order_processing", route_after_processing,
                           {"tongyi_llm": "tongyi_llm", "__end__": "__end__"})
    .add_edge("tongyi_llm", "__end__")
    .compile(name="电商订单处理工作流")
)

讲解：这是全项目最值得反复读的 20 行。三个节点表达"分析 → 执行 → 表达"的业务三段论；两条条件边则把两个决策点显式化：“该不该处理”（未知意图不浪费系统动作）与**“处理完谁收尾”**（客服场景需要 LLM 生成转人工说明）。相比在单个大函数里写 if/else 串行调用，图式编排的收益在于：节点可独立调试、路由可独立测试、整图可在 LangGraph Studio 中逐超步观察状态——这就是状态机建模对复杂交互流程的"结构化红利"。

把上述代码翻译成图，就是下面这张拓扑（与 LangGraph Studio 中渲染出的视图一致）：

图 2 StateGraph 拓扑：菱形之外的所有判断都收敛到两个条件边上——should_continue 决定"是否动手处理"，route_after_processing 决定"处理完由谁收尾"；__end__ 是图中唯一的终止态。

5. 意图识别双轨：加权关键词与 LLM 的互补

5.1 理论：意图识别与置信度

意图识别是对话系统的"输入侧路由器"。工程上存在两条路线：规则引擎（快、可解释、零成本，但覆盖有限）与 LLM 判别（泛化强，但有延迟、成本与不可用风险）。该项目的可贵之处在于不选边站，而是让两者以 “LLM 优先 + 规则兜底” 的双轨结构协同，并在规则引擎内部引入置信度概念——每种方法输出得分，系统依据得分决策，当什么都匹配不上时，unknown 兜底意图保证系统永远给出响应而非崩溃。

5.2 精品代码：关键词加权打分（services.py）

def recognize_intent(self, user_input: str) -> IntentResult:
    user_input_lower = user_input.lower()
    intent_scores = {}
    # 计算每个意图的匹配分数：命中关键词数 / 该意图关键词总数
    for intent, keywords in self.intent_keywords.items():
        score = sum(1 for keyword in keywords if keyword in user_input_lower)
        if score > 0:
            intent_scores[intent] = score / len(keywords)

    if not intent_scores:
        return IntentResult(intent="unknown", confidence=0.0, entities={})

    best_intent = max(intent_scores, key=intent_scores.get)
    confidence = intent_scores[best_intent]
    entities = self._extract_entities(user_input, best_intent)

    return IntentResult(intent=best_intent, confidence=confidence, entities=entities)

讲解：三处设计值得品味。其一，归一化打分——每类意图命中数除以该类的关键词总数，使"查询类有 6 个词、商品类只有 5 个词"这种类别不平衡不会扭曲比较；其二，可解释性——IntentResult 携带 confidence 与 entities，上游可以依据阈值做二次决策（config 中 INTENT_CONFIDENCE_THRESHOLD = 0.3），低置信时宁可降级到 LLM 或 unknown；其三，结构化输出——意图、置信度、实体三元组本身就是后续节点的输入契约，这与第 7 章的数据模型设计一脉相承。

5.3 双轨协作策略（graph.py 意图识别节点）

图中节点采用"先 LLM 后规则"的执行顺序：用通义千问识别意图并校验返回值是否落在合法意图集合内（valid_intents 白名单校验），一旦 API 异常或返回非法值，立刻降级到关键词匹配：

try:
    intent_result = llm.invoke(intent_prompt).strip()
    intent = intent_result if intent_result in valid_intents else "unknown"
except Exception as e:
    print(f"通义千问意图识别失败: {e}")     # 降级到关键词匹配
    if any(keyword in user_input.lower() for keyword in ["查询", "查看", "订单状态"]):
        intent = "query_order"
    elif any(keyword in user_input.lower() for keyword in ["修改", "更改", "变更"]):
        intent = "modify_order"
    ...

讲解：except 不是"错误处理"而是业务路径——它保证在模型服务不可用时，系统从"智能客服"平滑退化为"规则客服"，而不是抛出异常终止整张图。白名单校验同样关键：LLM 的自由文本输出必须先经集合约束才能进入路由，防止幻觉意图污染 should_continue 的决策。

整条识别链的决策路径如下：

图 3 双轨识别决策流：右侧规则轨仅在左侧 LLM 轨"异常"时启动，而白名单校验在两条轨的输出上又做了一层约束——任何一步失败都有可观测的收敛终点（unknown）。

6. 大模型服务层：意图感知提示与优雅降级

6.1 意图即提示上下文

LLM 的回复质量高度依赖提示中的角色与约束。服务层把"每个意图配一套系统提示"建模为字典映射——提示模板的维护成本随意图数量线性增长，且每个意图的提示可独立迭代：

system_prompts = {
    "query_order":    "你是一个专业的电商客服，专门处理订单查询问题。请提供准确、友好的回复。",
    "cancel_order":   "你是一个电商客服，处理订单取消请求。请说明取消流程和相关政策。",
    "customer_service": "你是一个专业的电商客服代表，提供优质的客户服务。",
    "payment_issue":  "你是一个电商支付问题处理专家，帮助解决支付相关问题。",
    ...
}
system_prompt = system_prompts.get(intent, "你是一个专业的电商客服助手。")

6.2 精品代码：优雅降级与可观测性（services.py）

def generate_response(self, prompt: str, intent: str = "") -> LLMResponse:
    start_time = time.time()
    ...
    try:
        from langchain_community.llms import Tongyi
        llm = Tongyi(dashscope_api_key=self.api_key,
                     model_name="qwen-turbo", temperature=0.7)
        full_prompt = system_prompt + "\n\n用户问题: " + prompt + "\n\n请提供专业回复："
        response_content = llm.invoke(full_prompt).strip()
    except Exception as e:
        print(f"通义千问API调用失败: {e}")
        response_content = self._generate_mock_response(prompt, intent)  # 预设话术兜底

    response_time = time.time() - start_time
    return LLMResponse(content=response_content, model="qwen-turbo",
                       tokens_used=len(prompt) + len(response_content),
                       response_time=response_time)

讲解：这段代码是"降级三级跳"的最末一级——LLM → 意图模板话术 → 通用话术，保证任何情况下 content 非空（测试 test_tongyi_llm_service 正是断言这一点）。更值得学习的是结果对象的元数据设计：LLMResponse 不只携带 content，还记录 model、tokens_used（近似估算）与 response_time。这让"一次模型调用"从不可观测的黑盒变成可度量的事件——为后续接入 LangSmith、做成本与延迟分析预留了契约位。

完整的"降级三级跳"结构如下：

图 4 回复生成的三级降级：降级产物与成功产物共用同一结构契约（LLMResponse），上层节点无需区分来源——"结构统一、实现可换"是服务封装的核心收益。

7. 契约与配置：数据模型与声明式路由表

7.1 领域模型的"自文档化"（models.py）

class WorkflowState(BaseModel):
    session_id: str = Field(..., description="会话ID")
    user_input: str = Field(..., description="用户输入")
    intent_result: Optional[IntentResult] = None
    llm_response: Optional[LLMResponse] = None
    processing_result: Optional[ProcessingResult] = None
    current_step: str = Field(default="start", description="当前步骤")
    created_at: datetime = Field(default_factory=datetime.now, description="创建时间")
    updated_at: datetime = Field(default_factory=datetime.now, description="更新时间")

讲解：Pydantic 模型在此扮演"领域契约层"——IntentResult、LLMResponse、ProcessingResult 的组合定义了工作流各阶段的产物格式，字段级 description 让模型直接充当文档；Optional 表达"该阶段产物尚未产生"的合法中间态；时间戳字段则为审计留痕。先定契约、再写节点的顺序，让图的所有节点天然对齐输入输出。

7.2 声明式配置：意图路由表（config.py）

INTENT_MAPPING = {
    "query_order":      {"name": "订单查询", "next_nodes": ["tongyi_llm", "order_processing"]},
    "customer_service": {"name": "客服咨询", "next_nodes": ["tongyi_llm", "order_processing"]},
    "payment_issue":    {"name": "支付问题", "next_nodes": ["tongyi_llm"]},
    "unknown":          {"name": "未知意图", "next_nodes": ["tongyi_llm"]},
}

NODE_CONFIG = {
    "intent_recognition": {"name": "意图识别", "timeout": 30},
    "tongyi_llm":         {"name": "通义大模型", "timeout": 60},
    "order_processing":   {"name": "订单处理", "timeout": 45},
}

讲解：这是典型的数据驱动设计——意图的种类、展示名、后续可达节点被整体外置为声明式映射，业务新增一个意图（例如"发票申请"）只需在此追加一行，图中的路由逻辑与提示模板字典都不必改动。timeout 字段则把"节点最坏执行时间"显式化，为分布式编排中的超时治理（如 config 中 MAX_WORKFLOW_STEPS = 10、WORKFLOW_TIMEOUT = 300）提供了依据。配置集中、行为声明化，是这类多意图系统控制复杂度的第一手段。

8. 可信交付：测试、演示与已知问题

8.1 测试即规格（tests/test_workflow.py）

项目的测试覆盖了三层：服务层（关键词识别与 LLM 兜底）、单场景工作流（查询/取消/客服/未知）、回归层（多次运行）：

def test_workflow_customer_service(self):
    """客服咨询：应转人工，且由 LLM 生成转接说明"""
    initial_state = State(user_input="订单有问题，需要客服帮助", intent="", order_info={},
                          response="", next_action="", messages=[])
    context = Context(user_id="test_user", session_id="test_session")
    result = self.workflow.invoke(initial_state, {"configurable": context})

    assert result["intent"] == "customer_service"
    assert result["order_info"]["next_action"] == "transfer_to_human"

讲解：注意测试对 next_action（内部流转信号）与 response（最终话术）分别断言——前者验证图的路由正确性，后者验证输出完备性。断言内部流转信号而非只看最终话术，是工作流测试的要点：它能精确定位"路由错了"还是"话术错了"。

8.2 运行三部曲

export DASHSCOPE_API_KEY="your_key"          # 1. 配置密钥（不配则走降级模式）
python demo_tongyi_integration.py            # 2. 综合演示：意图服务 → LLM 服务 → 完整工作流
pytest tests/test_workflow.py -v             # 3. 执行测试
langgraph dev                                # 4. 在项目根目录启动 LangGraph Studio 可视化调试整图

8.3 已知问题与修复建议（工程博客的诚实条款）

经语法校验（python -m py_compile），当前 graph.py 中 tongyi_llm_node 函数体存在 IndentationError（第 159 行 unexpected indent），导致整图无法导入运行——这是"LLM 生成代码后未经编译验证"的典型事故，恰好反向印证了测试与编译检查的必要性。修复思路有两个层次：

1. 最小修复：纠正该节点内的缩进（函数体内统一 4 空格层级），恢复 try/except 块结构；
2. 更优重构：图中节点应"变薄"——直接调用第 6 章已验证的 TongyiLLMService，消除节点内联大模型调用与 services.py 的重复实现：

def tongyi_llm_node(state: State, runtime: Runtime[Context]) -> Dict[str, Any]:
    """通义大模型节点：委托服务层生成回复（薄节点 + 厚服务）"""
    service = TongyiLLMService()
    llm_response = service.generate_response(state.user_input, state.intent)
    return {
        "response": llm_response.content,
        "messages": state.messages + [{"role": "assistant", "content": llm_response.content}],
    }

9. 总结：可复用的模式与演进方向

复盘整个实现，可以提炼出三条可迁移到任何 Agent 工程的模式：

1. 双轨降级（Degradation by Design）：LLM 优先、规则兜底、模板话术保底，把"外部依赖不可用"当成正常业务路径来设计，而非异常分支；
2. 薄图厚服务（Thin Graph, Fat Services）：graph.py 只负责拓扑与路由，算法与外部调用沉淀在 services.py，使编排可可视化、逻辑可单测；
3. 契约先行（Contract First）：Pydantic 模型定义阶段产物，声明式 INTENT_MAPPING 定义意图-节点关系，让"加一个意图"从改代码退化为改配置。

在此基础上，向生产迈进的方向也清晰可见：以 MemorySaver/Redis 接入持久化 checkpoint（对应 LangGraph 短期/长期记忆）；以 interrupt 机制加入人工审批节点（HITL）；将模拟 OrderService 替换为真实订单系统并补充幂等设计；以及用 LangSmith 对双轨识别的准确率做离线评测与阈值调优。脚手架到业务系统的距离，往往就是这几层工程化习惯的距离。