第 27 章 实战项目一：多轮智能客服系统

27.1 需求与场景

多轮智能客服系统面向电商与 SaaS 产品的在线客服场景。用户通过 Web 或移动端会话入口发起咨询，系统在单个会话内处理订单查询、物流跟踪、退货退款、常见问题解答、工单创建与转人工六类任务。能力边界限定于已接入的业务域，不处理账户注销、合同变更、投诉升级等需人工坐席或后台审核的操作，此类请求统一路由至人工坐席。

业务场景覆盖三个典型流程。其一为订单类咨询，用户询问"我的订单什么时候发货""订单 12345 现在什么状态"，系统需要识别意图、抽取订单号（若未提供则追问）、调用订单中心接口返回状态。其二为退款类高风险操作，用户提出"要退货退款"，系统校验退款资格后必须中断等待审批，不得直接执行资金操作。其三为多轮混合诉求，用户先查物流、再问退货政策、最后要求退款，系统需在同一会话内保留上下文实体并在不同子图间切换。

功能需求涵盖七个维度。其一为多轮对话上下文记忆，系统需在同一 session 内跨轮次保留用户问题与已提取实体，避免重复追问订单号等信息，messages 列表以 add_messages 归约器追加。其二为意图识别与路由（Router，路由器模式），将用户输入分发至订单、物流、退款、FAQ（Frequently Asked Questions，常见问题）、工单、人工六个专业 Agent（智能体）。其三为业务工具调用，封装内部订单中心、物流网关、退款中心、工单系统的 HTTP 或 RPC（Remote Procedure Call，远程过程调用）接口，工具须具备超时、重试与幂等能力。其四为高风险操作人工审批（Human-in-the-Loop，HITL，人在回路），退款类操作必须在执行前由坐席或风控审批人确认，金额超过阈值还需二级审批。其五为兜底转人工，无法识别意图、置信度低于阈值或用户明确要求时移交坐席，并附带会话摘要以缩短坐席应答时间。其六为会话结束自动归档摘要，生成结构化摘要写入长期存储，用于后续会话召回与客服质量分析。其七为全程流式回复，前端以 SSE（Server-Sent Events，服务器推送事件）逐 token 渲染并接收自定义进度事件，降低首字延迟。

非功能需求包括：单会话 P95 响应延迟低于 2 秒（不含外部工具调用）；工具调用统一 30 秒超时；单租户支持 500 并发会话；对话记录与工具调用全程可追溯；所有工具调用与审批决策写入审计日志；系统支持灰度发版与快速回滚；知识库覆盖度 ≥ 85% FAQ 高频问题；意图识别准确率 ≥ 92%；高风险操作零误触发（100% 走 HITL）。

典型用户旅程示意：用户打开客服窗口输入"我的订单怎么还没到"，intent_router 识别 logistics 意图但未抽取 order_id，logistics 子图内部追问"请问您要查询的订单号是？"；用户回复订单号后，物流 Agent 调用 logistics_track_tool 返回轨迹；若用户继续追问"这个能退吗"，子图结束后回到 intent_router 识别 refund 意图，进入 refund 子图；退款资格通过后 interrupt 暂停等待审批；审批通过后退款执行；最后用户说"没别的事了"，进入 summary_node 生成摘要并存入 Store。

27.2 架构设计

系统采用 Router 模式与 Handoffs（交接）混合架构。顶层为客服 Supervisor（监督者）图，内部以 intent_router 为入口分发至六个业务子图（Subgraph），每个业务 Agent 封装为独立子图以隔离状态、工具集与提示词，处理完成后返回 Supervisor 以判断是否继续多轮对话或进入归档节点结束。

┌────────┐   HTTP/SSE   ┌──────────────┐   invoke/stream   ┌────────────────────────────────────────┐
│ Client │ ───────────▶ │  API Layer   │ ────────────────▶ │         Customer Service Graph         │
│ (Web)  │ ◀─────────── │ (FastAPI)    │ ◀──────────────── │  ┌──────────┐                          │
└────────┘              └──────────────┘                   │  │ intent_  │  Command(goto=...)       │
                                                           │  │ router   │ ──┬──┬──┬──┬──┬──┬──┐   │
                                                           │  └──────────┘   │  │  │  │  │  │  │   │
                                                           │                 ▼  ▼  ▼  ▼  ▼  ▼  ▼   │
                                                           │         ┌───────────────────────────┐  │
                                                           │         │ order/logistics/refund/   │  │
                                                           │         │ faq/ticket/handoff sub-   │  │
                                                           │         │ graphs (each a Subgraph)  │  │
                                                           │         └─────────┬─────────────────┘  │
                                                           │                   ▼                    │
                                                           │           ┌──────────────┐             │
                                                           │           │ summary_node │──▶ END      │
                                                           │           └──────────────┘             │
                                                           └───────────────┬────────────────────────┘
                                                                           │
                              ┌──────────────────────┬─────────────────────┼─────────────────────┐
                              ▼                      ▼                     ▼                     ▼
                       ┌────────────┐        ┌─────────────┐       ┌──────────────┐      ┌────────────┐
                       │ Postgres   │        │ Postgres    │       │ LangSmith    │      │ Business   │
                       │ Saver      │        │ Store       │       │ Tracing      │      │ APIs       │
                       │(checkpoints)│       │(user profile)│      │(runs/traces) │      │(order/...) │
                       └────────────┘        └─────────────┘       └──────────────┘      └────────────┘

技术选型：LangGraph 1.2.x 作为编排框架，Supervisor 为顶层 StateGraph，业务 Agent 以 Subgraph 形式通过 add_node 注册，子图间通过 Command(goto=...) 实现显式跳转。Checkpointer（检查点器）使用 PostgresSaver 保存会话检查点，支持断点续跑与 HITL 暂停恢复。Store（长期存储）使用 PostgresStore 保存用户画像与长期偏好，跨 session 复用。追踪接入 LangSmith，记录每次 run 的节点输入输出、token 消耗与延迟。模型层默认使用 gpt-4o-mini 做意图识别、FAQ 与摘要，gpt-4o 做退款资格推理、工单信息提取等重任务；按租户维度配置可切换主备模型。

项目目录结构：

customer_service/
├── app/
│   ├── main.py              # FastAPI 入口，SSE 与审批回调
│   ├── graph.py             # Supervisor 图构建
│   ├── state.py             # 共享 State 定义
│   ├── nodes/
│   │   ├── router.py        # intent_router 节点
│   │   ├── post_dispatch.py # Agent 结束后的路由分发
│   │   ├── summary.py       # summary 节点
│   │   └── agents/
│   │       ├── faq.py       # FAQ 子图（RAG）
│   │       ├── order.py     # 订单子图
│   │       ├── logistics.py # 物流子图
│   │       ├── refund.py    # 退款子图（含 HITL）
│   │       ├── ticket.py    # 工单子图
│   │       └── handoff.py   # 转人工子图（含 HITL）
│   ├── tools/
│   │   ├── order_api.py     # 订单查询工具
│   │   ├── logistics_api.py # 物流跟踪工具
│   │   ├── refund_api.py    # 退款资格与执行工具
│   │   ├── ticket_api.py    # 工单创建工具
│   │   └── faq_retriever.py # FAQ 向量检索工具
│   └── utils/
│       ├── db.py            # 数据库连接池
│       ├── http.py          # 统一 httpx 客户端与重试
│       └── security.py      # 脱敏与审批校验
├── prompts/                 # 版本化 YAML 提示词
│   ├── intent_router.yaml
│   ├── refund.yaml
│   └── summary.yaml
├── tests/
│   ├── test_intent.py
│   ├── test_refund_approval.py
│   ├── test_session_isolation.py
│   └── test_tool_fallback.py
├── docker-compose.yml
├── requirements.txt
└── .env

依赖安装命令：

pip install "langgraph>=1.2,<1.3" langchain langchain-openai langchain-core \
    psycopg[binary,pool] fastapi uvicorn httpx pydantic pyyaml sse-starlette

各子图职责边界划分：faq 子图仅挂载 faq_retriever 向量检索工具，不访问业务写接口，其职责是从知识库中返回标准问答对，无法命中时可回退到 LLM 通用回答并提示答案仅供参考；order 子图挂载 order_query_tool，只读不写，未取得 order_id 时通过追问节点向用户索取，不会反复重试；logistics 子图挂载 logistics_track_tool，负责格式化轨迹节点为可读文本，对于运输异常（滞留、退回）节点自动高亮；refund 子图挂载 refund_eligibility_tool 与 refund_execute_tool，是唯一包含写操作与 HITL 审批的子图；ticket 子图挂载 create_ticket_tool，需要与用户多轮收集 title/category/description/priority 四项必填字段，缺失字段逐项追问；human_handoff 子图不挂载业务工具，只负责通过 interrupt 通知坐席并等待 accept/close 信号，坐席接管后进入"人工回复"通道，图本身不再调用 LLM 产生自动回复，仅在坐席关闭会话时回到 summary_node。

Supervisor 与子图的 State 隔离策略：顶层 ServiceState 包含所有跨子图共享字段；子图内部若需临时字段（如 refund 子图的 refund_amount、ticket 子图的 ticket_form），可在子图 State 中以 TypedDict 扩展，并在子图输出时只回写顶层关心的字段（messages、current_intent 等），私有字段不回写，避免 checkpoint 膨胀。子图通过 input/output 字段映射（compile 时的 config 或 add_node 时的通道对齐）选择性暴露通道。

27.3 State 与节点路由设计

顶层 State 承载跨子图共享的会话上下文。messages 使用 add_messages 归约器追加消息，避免子图间覆盖。业务实体字段在 intent_router 阶段填充，子图节点按需读取，不在子图间传递冗余业务数据。

from typing import Annotated, Optional, Literal
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class ServiceState(TypedDict):
    """客服会话共享状态。所有业务子图共享该结构。"""
    messages: Annotated[list[BaseMessage], add_messages]
    user_id: str
    session_id: str
    order_id: Optional[str]               # 从用户话语中提取的订单号
    current_intent: Optional[Literal[
        "order","logistics","refund","faq","ticket","human"]]
    human_approval_pending: bool          # 是否存在未完成的人工审批
    conversation_summary: str             # 会话结束时回填
    handoff_reason: Optional[str]         # 转人工原因（低置信度/用户主动/失败降级）
    retry_count: int                      # 当前轮工具失败重试计数
    last_tool_error: Optional[str]        # 最近一次工具错误信息

子图内部可扩展私有字段（如 refund 子图的 refund_amount 与 refund_reason），私有字段定义在子图 State 中，不回写到顶层 State，避免状态污染与 checkpoint 体积膨胀。

节点设计遵循单一职责原则：

• intent_router：使用 with_structured_output 调用 LLM 识别意图并提取实体（order_id、退款金额、工单分类等），输出 Command(goto=target_agent, update=...)。置信度低于 0.6 时直接路由至 human_handoff_agent，并在 handoff_reason 中记录"低置信度"或"无法识别"。
• faq_agent：基于向量检索知识库回答常见问题，子图内挂载 faq_retriever 工具，无业务写操作，仅做 RAG（Retrieval-Augmented Generation，检索增强生成）。
• order_agent：调用 order_query_tool，将结构化返回转为自然语言状态描述；若未提供 order_id 则追问用户。
• logistics_agent：调用 logistics_track_tool，解析轨迹节点列表为按时间倒序的可读文本。
• refund_agent：先调用 refund_eligibility_tool 校验资格，校验通过后调用 interrupt() 携带退款详情 payload 等待审批；审批通过后再调用 refund_execute_tool；驳回则告知原因并建议转人工。
• ticket_agent：多轮收集工单标题、问题分类、问题描述与紧急度，信息齐全后调用 create_ticket_tool 返回工单号。
• human_handoff_agent：调用 interrupt() 通知坐席端有会话待接入，payload 携带会话摘要与用户基础信息，等待坐席 accept 后恢复。
• summary_node：会话结束触发时，对 messages 做结构化摘要（问题类型、解决结果、用户情绪），写入 PostgresStore 的用户画像条目。

条件路由规则：每个业务子图结束节点返回 Command(goto=...)，统一跳到 post_agent_dispatch 路由节点；该节点根据最后一条用户消息判断是否继续多轮。若用户消息包含"没有其他问题""结束""没事了"等结束标记，路由到 summary_node 后 END；否则回到 intent_router 继续识别下一个意图。

# graph.py 核心图构建
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from .state import ServiceState
from .nodes.router import intent_router
from .nodes.summary import summary_node
from .nodes.post_dispatch import post_agent_dispatch
from .nodes.agents.faq import build_faq_graph
from .nodes.agents.order import build_order_graph
from .nodes.agents.logistics import build_logistics_graph
from .nodes.agents.refund import build_refund_graph
from .nodes.agents.ticket import build_ticket_graph
from .nodes.agents.handoff import build_handoff_graph

builder = StateGraph(ServiceState)
builder.add_node("intent_router", intent_router)
builder.add_node("faq", build_faq_graph())
builder.add_node("order", build_order_graph())
builder.add_node("logistics", build_logistics_graph())
builder.add_node("refund", build_refund_graph())
builder.add_node("ticket", build_ticket_graph())
builder.add_node("human_handoff", build_handoff_graph())
builder.add_node("post_agent_dispatch", post_agent_dispatch)
builder.add_node("summary", summary_node)

builder.add_edge(START, "intent_router")
for agent in ["faq","order","logistics","refund","ticket","human_handoff"]:
    builder.add_edge(agent, "post_agent_dispatch")
builder.add_conditional_edges("post_agent_dispatch",
    lambda s: "summary" if s.get("_finish") else "intent_router",
    {"summary":"summary", "intent_router":"intent_router"})
builder.add_edge("summary", END)

intent_router 节点实现采用结构化输出 IntentResult。系统提示词存储于 prompts/intent_router.yaml 并按版本加载，模型选择上优先 gpt-4o-mini 以降低延迟与成本，若租户配置中要求高准确率可切换到 gpt-4o。路由时同时考虑当前轮消息与近几轮历史，以识别省略主语或代词指代的情况（如"那这个能退吗"中的"这个"需要通过上下文判断指代上一个订单）。若用户消息出现"人工""转人工""找人"等关键词，直接路由 human_handoff 并跳过置信度判断，尊重用户选择权。

# nodes/router.py
from pydantic import BaseModel, Field
from typing import Optional, Literal
from langchain.chat_models import init_chat_model
from langchain_core.messages import AIMessage
from langgraph.types import Command

class IntentResult(BaseModel):
    intent: Literal["order","logistics","refund","faq","ticket","human","fallback"]
    order_id: Optional[str] = None
    confidence: float = Field(ge=0, le=1)
    rationale: str

async def intent_router(state: ServiceState):
    llm = init_chat_model("openai:gpt-4o-mini").with_structured_output(IntentResult)
    history = state["messages"][-10:]  # 裁剪近 10 条节省 token
    result: IntentResult = await llm.ainvoke([
        {"role":"system","content":"该角色为客服意图识别器，根据对话历史判定当前轮意图，提取订单号，返回置信度与理由。"},
        *history
    ])
    if result.intent in ("fallback","human") or result.confidence < 0.6:
        return Command(
            goto="human_handoff",
            update={"current_intent":"human",
                    "handoff_reason":f"置信度{result.confidence:.2f}或主动转人工",
                    "order_id": result.order_id or state.get("order_id")})
    return Command(
        goto=result.intent,
        update={"current_intent": result.intent,
                "order_id": result.order_id or state.get("order_id"),
                "retry_count": 0,
                "last_tool_error": None})

FAQ 子图结构示意（RAG 检索 + 回答）：

# nodes/agents/faq.py
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from ...tools.faq_retriever import faq_retriever_tool

async def faq_retrieve_node(state):
    # 先用检索工具找最相关的 5 条 FAQ
    res = await faq_retriever_tool.ainvoke(state["messages"][-1].content)
    return {"_faq_context": res}

async def faq_answer_node(state):
    llm = init_chat_model("openai:gpt-4o-mini")
    ctx = state.get("_faq_context","")
    ans = await llm.ainvoke([
        {"role":"system","content":"基于以下 FAQ 知识库片段用中文回答用户问题，若知识库无相关信息，请如实说明并建议转人工或提交工单。\n"+ctx},
        *state["messages"][-6:]])
    return {"messages": [ans], "_faq_context": ""}

def build_faq_graph():
    sub = StateGraph(ServiceState)
    sub.add_node("faq_retrieve", faq_retrieve_node)
    sub.add_node("faq_answer", faq_answer_node)
    sub.add_edge(START, "faq_retrieve")
    sub.add_edge("faq_retrieve", "faq_answer")
    sub.add_edge("faq_answer", END)
    return sub.compile()

订单子图与物流子图结构类似，均为"参数校验 → 追问或调用工具 → 格式化回复"三段式，若 order_id 缺失则由 ask_order_id 节点发起追问并 END，等待用户下一轮消息重新进入意图识别，避免子图内部阻塞等待用户输入（用户输入永远通过顶层 graph.astream 注入）。

# nodes/post_dispatch.py
async def post_agent_dispatch(state: ServiceState):
    # 读取最近一条用户消息（若存在）
    user_msgs = [m for m in state["messages"] if m.type == "human"]
    if not user_msgs:
        return {"_finish": False}
    last_user = user_msgs[-1].content
    end_markers = ["没有其他问题", "结束", "谢谢", "没事了", "再见"]
    finish = any(m in last_user for m in end_markers)
    # 若已转人工且坐席已结束会话，也进入 summary
    if state.get("current_intent") == "human" and state.get("_handoff_closed"):
        finish = True
    return {"_finish": finish}

27.4 工具实现与 HITL

业务工具统一以 @tool 封装，外层通过 with_retry 装饰器加入超时、指数退避重试与结构化错误返回。工具返回结构化 dict，键包括 ok、data、error，便于下游节点判断分支，避免将自然语言错误信息交给 LLM 解析。

# utils/http.py 与 tools/order_api.py 合并示意
from langchain_core.tools import tool
import httpx, asyncio
from functools import wraps

def with_retry(max_retry: int = 2, timeout: float = 10.0):
    """HTTP 工具统一重试装饰器。"""
    def decorator(fn):
        @wraps(fn)
        async def wrapper(*args, **kwargs):
            last_exc = None
            for i in range(max_retry + 1):
                try:
                    async with httpx.AsyncClient(timeout=timeout) as client:
                        return await fn(client, *args, **kwargs)
                except (httpx.HTTPError, asyncio.TimeoutError) as e:
                    last_exc = e
                    await asyncio.sleep(0.5 * (i + 1))
            return {"ok": False,
                    "error": {"code": "tool_timeout_or_error", "message": str(last_exc)}}
        return wrapper
    return decorator

@tool
@with_retry()
async def order_query_tool(client, order_id: str) -> dict:
    """查询订单状态。参数 order_id 为订单编号字符串。"""
    r = await client.get(f"http://order-svc.internal/orders/{order_id}")
    r.raise_for_status()
    return {"ok": True, "data": r.json()}

@tool
@with_retry()
async def logistics_track_tool(client, order_id: str) -> dict:
    """查询物流轨迹节点列表。参数 order_id 为订单编号。"""
    r = await client.get(f"http://logistics-svc.internal/track/{order_id}")
    r.raise_for_status()
    return {"ok": True, "data": r.json()}

@tool
@with_retry(max_retry=0)  # 退款执行禁止自动重试，避免重复退款
async def refund_execute_tool(client, order_id: str, amount: float, reason: str, idem_key: str) -> dict:
    """执行退款。需审批通过后方可调用，idem_key 为幂等键。"""
    r = await client.post("http://refund-svc.internal/execute",
        json={"order_id": order_id, "amount": amount, "reason": reason,
              "idem_key": idem_key})
    r.raise_for_status()
    return {"ok": True, "data": r.json()}

@tool
@with_retry()
async def refund_eligibility_tool(client, order_id: str) -> dict:
    """校验订单是否满足退款条件，返回可退金额与原因。"""
    r = await client.get(f"http://refund-svc.internal/eligibility/{order_id}")
    r.raise_for_status()
    return {"ok": True, "data": r.json()}

@tool
@with_retry()
async def create_ticket_tool(client, user_id: str, title: str, category: str,
                              description: str, priority: str = "normal") -> dict:
    """创建客服工单。category 取值 billing/technical/account/other。"""
    r = await client.post("http://ticket-svc.internal/tickets",
        json={"user_id": user_id, "title": title, "category": category,
              "description": description, "priority": priority})
    r.raise_for_status()
    return {"ok": True, "data": r.json()}

退款子图的 HITL 流程是系统安全核心。eligibility 校验通过后，节点调用 interrupt() 将退款详情抛出给外部审批系统，运行图在此暂停，checkpoint 自动持久化至 Postgres。审批人在后台完成 approve/reject/modify 操作后，通过 API 以 Command(resume=decision) 恢复图执行，refund_request 节点从 interrupt 返回值处获得审批结果并继续执行。

# nodes/agents/refund.py 退款子图核心节点
from langgraph.types import interrupt
from langchain_core.messages import AIMessage
from langgraph.graph import StateGraph, START, END
from ...tools.refund_api import refund_eligibility_tool, refund_execute_tool
from ...state import ServiceState
import uuid

async def refund_check_node(state: ServiceState):
    oid = state.get("order_id")
    if not oid:
        return {"messages": [AIMessage(content="请提供需要退款的订单号。")],
                "_refund_done": False}
    elig = await refund_eligibility_tool.ainvoke({"order_id": oid})
    if not elig.get("ok"):
        return {"messages":[AIMessage(content="暂时无法校验退款资格，请稍后再试或转人工。")],
                "last_tool_error": elig["error"]["message"], "_refund_done": True}
    if not elig["data"].get("eligible"):
        return {"messages":[AIMessage(
            content=f"订单 {oid} 暂不符合退款条件：{elig['data'].get('reason','未知原因')}")],
                "_refund_done": True}
    amount = elig["data"]["refundable_amount"]
    # 触发中断，等待审批人操作，payload 包含审批 UI 所需全部信息
    decision = interrupt({
        "type": "refund_approval",
        "order_id": oid,
        "amount": amount,
        "user_id": state["user_id"],
        "reason_snapshot": state["messages"][-1].content if state["messages"] else "",
        "awaiting": "approver_decision"
    })
    # decision 期望结构：{"action":"approve"|"reject"|"modify",
    #                    "amount":float,"comment":str}
    if decision.get("action") == "reject":
        return {"messages":[AIMessage(
            content=f"退款申请未通过：{decision.get('comment','审批驳回')}，可申请转人工进一步处理。")],
                "human_approval_pending": False, "_refund_done": True}
    final_amount = decision.get("amount", amount)
    idem_key = f"refund:{state['session_id']}:{oid}:{uuid.uuid4().hex[:8]}"
    exe = await refund_execute_tool.ainvoke({
        "order_id": oid, "amount": final_amount,
        "reason": "customer_request", "idem_key": idem_key})
    if exe.get("ok"):
        return {"messages":[AIMessage(
            content=f"退款 {final_amount:.2f} 元已提交，原路返回预计 1-3 个工作日到账。")],
                "human_approval_pending": False, "_refund_done": True}
    return {"messages":[AIMessage(content="退款执行失败，已为您转人工处理。")],
            "human_approval_pending": False, "handoff_reason":"退款执行失败",
            "_refund_done": False, "_escalate_human": True}

def build_refund_graph():
    sub = StateGraph(ServiceState)
    sub.add_node("refund_check", refund_check_node)
    sub.add_edge(START, "refund_check")
    sub.add_edge("refund_check", END)
    return sub.compile()

工单与转人工子图要点：ticket_agent 使用 FormFilling 思路，用 Pydantic 模型 TicketForm 校验已收集字段，若有缺失则生成对应追问，全部字段齐备后调用工具；human_handoff_agent 的 interrupt payload 中包含最近 10 条消息的摘要、用户已识别意图、当前业务实体（order_id）以及失败原因，帮助坐席快速切入。坐席端接收 interrupt 事件后有 60 秒响应窗口，超时自动升级到下一组坐席并在 payload 中标记 escalated=true。

工具幂等与超时策略：所有写工具（refund_execute_tool、create_ticket_tool）必须接受 idem_key 参数，由调用节点用 f"{session_id}:{action}:{entity_id}:{attempt}" 生成幂等键传入后端，后端按 idem_key 去重；读工具可重试 2 次，指数退避间隔 0.5s/1s；工具超时时间按接口 SLA 差异化配置，订单查询 5s、物流跟踪 8s、退款执行 15s、工单创建 10s；所有工具返回值必须包含 ok 字段，错误码 code 使用稳定字符串（tool_timeout、permission_denied、business_rejected、network_error）便于上层节点做分支判断。

27.5 持久化、流式与错误处理

持久化层使用 PostgresSaver 与 PostgresStore，二者共享同一连接池以减少连接数。thread_id 按会话维度设置为 sess:{session_id}，确保同一 session 的所有 checkpoint 连续且可随时通过 thread_id 恢复。Store 以 ("users", user_id) 为 namespace，保存用户长期偏好（称呼、常用地址、对响应长度的偏好）、历史会话摘要、常见问题命中记录等跨会话信息。

# graph.py 编译
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
from psycopg_pool import ConnectionPool

conn_pool = ConnectionPool(
    conninfo="postgresql://cs_user:cs_pwd@pg:5432/langgraph",
    max_size=20,
    kwargs={"autocommit": True})
checkpointer = PostgresSaver(conn_pool)
store = PostgresStore(conn_pool)
checkpointer.setup()
store.setup()

graph = builder.compile(
    checkpointer=checkpointer,
    store=store,
    interrupt_before=["human_handoff"],  # 转人工前额外的入口断点（可选）
)

流式输出采用双模式：messages 模式输出模型 token 流供前端逐字渲染；custom 模式输出业务节点进度事件，前端根据 event 类型展示"正在查询订单…""已提交退款审批"等提示。

# main.py 流式入口
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langchain_core.messages import HumanMessage
import json, uuid

app = FastAPI()

class ChatIn(BaseModel):
    message: str
    user_id: str
    session_id: str

async def stream_turn(user_input: str, user_id: str, session_id: str):
    config = {"configurable": {
                  "thread_id": f"sess:{session_id}",
                  "user_id": user_id},
              "recursion_limit": 50}
    async for mode, payload in graph.astream(
        {"messages": [HumanMessage(content=user_input)],
         "user_id": user_id,
         "session_id": session_id,
         "human_approval_pending": False,
         "retry_count": 0},
        config=config,
        stream_mode=["messages", "custom"]):
        if mode == "messages":
            msg, meta = payload
            if msg.content:
                yield {"type":"token", "content": msg.content,
                       "node": meta.get("langgraph_node")}
        elif mode == "custom":
            yield {"type":"progress", "data": payload}

@app.post("/chat/stream")
async def chat_stream(body: ChatIn):
    async def gen():
        async for ev in stream_turn(body.message, body.user_id, body.session_id):
            yield f"data: {json.dumps(ev, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

在工具调用节点内通过 get_stream_writer() 主动推送进度事件：

from langgraph.config import get_stream_writer

async def order_query_node(state):
    writer = get_stream_writer()
    writer({"event":"tool_start","tool":"order_query","message":"正在查询订单信息…"})
    res = await order_query_tool.ainvoke({"order_id": state["order_id"]})
    writer({"event":"tool_end","tool":"order_query"})
    if not res.get("ok"):
        # 降级处理
        ...

错误处理覆盖四类场景。工具调用失败由 with_retry 装饰器自动重试 2 次；若仍失败，节点返回降级文案"系统暂时无法查询订单信息，已为您转人工"并通过 Command(goto="human_handoff") 路由。API 超时由 httpx 超时捕获，同上降级。意图识别置信度低于 0.6 直接转人工。退款执行工具 refund_execute_tool 禁用自动重试，防止重复退款，失败即转人工。summary_node 失败不阻塞会话结束，异常捕获后写入错误日志，摘要字段留空，由夜间批处理任务补救。所有工具错误均写入 last_tool_error 字段，summary 时一并写入审计库。

27.6 部署与测试

部署形态采用 Docker Compose 启动 FastAPI 应用、Postgres 与可选的 LangGraph Server。前端通过 SSE 接入 /chat/stream，审批后台通过 /approval 接口调用 Command(resume=...) 恢复中断的图。

# docker-compose.yml
version: "3.9"
services:
  app:
    build: .
    ports: ["8000:8000"]
    environment:
      - DATABASE_URL=postgresql://cs_user:cs_pwd@pg:5432/langgraph
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LANGSMITH_API_KEY=${LANGSMITH_API_KEY}
      - LANGSMITH_TRACING=true
    depends_on: [pg]
    restart: always
  pg:
    image: postgres:16
    environment:
      POSTGRES_USER: cs_user
      POSTGRES_PASSWORD: cs_pwd
      POSTGRES_DB: langgraph
    ports: ["5432:5432"]
    volumes: ["cs_pgdata:/var/lib/postgresql/data"]
    healthcheck:
      test: ["CMD-SHELL","pg_isready -U cs_user -d langgraph"]
      interval: 5s
      timeout: 3s
      retries: 10
volumes:
  cs_pgdata:

审批恢复接口示例：

class ApprovalIn(BaseModel):
    session_id: str
    action: Literal["approve","reject","modify"]
    amount: float = None
    comment: str = ""

@app.post("/approval")
async def handle_approval(body: ApprovalIn):
    """审批后台在审批人操作后调用此接口恢复中断的图。"""
    config = {"configurable": {"thread_id": f"sess:{body.session_id}"}}
    # 通过 Command(resume=...) 注入审批结果，框架从 interrupt 断点继续执行
    async for _ in graph.astream(
        Command(resume={"action": body.action,
                        "amount": body.amount,
                        "comment": body.comment}),
        config=config,
        stream_mode="updates"):
        pass
    return {"status": "ok"}

部署要点：FastAPI 以 uvicorn worker 启动，worker 数量按 CPU 核数配置（I/O 密集场景建议 2×cores）；数据库连接池上限匹配 worker 数 × 每 worker 最大连接（通常每 worker 2-3 个 PG 连接）；生产环境前置 Nginx 做 TLS 终止与连接复用，SSE 路径关闭 proxy_buffering 设置 proxy_read_timeout 1h 防止长连接断开；LangSmith 项目名按租户分项目（如 cs-tenantA），便于按租户观测 token 与错误率；Postgres 主从部署，checkpointer/Store 使用连接串指向主库写入，若部署只读副本可通过 separate pool 服务读查询；HITL 审批后台单独部署，只暴露审批 API 与坐席 SSE 通道，不直接连接模型。

运维与监控：Prometheus 暴露 /metrics，指标包括 cs_sessions_active、cs_intent_total{intent}、cs_tool_latency_seconds{tool}、cs_human_handoff_total{reason}、cs_refund_approval_total{action}、cs_checkpoint_write_latency、cs_sse_connections_active；关键告警：人工兜底率连续 10 分钟 > 15%、工具错误率连续 5 分钟 > 5%、refund 工具调用次数异常（可能资金风险）、checkpoint 写入 P99 > 500ms（存储瓶颈）；LangSmith dashboard 监控每个节点的平均 token 与延迟，对异常意图节点触发 prompt 审查。

客服系统特有的质量保障：建立坐席反馈闭环，人工坐席接手后将"机器人回答错误/遗漏/不准确"的 case 标记回传，作为 intent_router 与各业务 Agent prompt 的优化样本；每周自动生成机器人解决率、平均会话轮次、一次性解决率、人工兜底率四类核心指标报表；对高频无法识别的问法聚类后补充 FAQ 知识库或新增业务工具，持续提升自动化率。

多轮对话体验优化：使用 trim_messages 控制 messages 长度但保留最近一轮的用户问题与系统回复；当意图切换时（如刚查完订单马上问退款），在新子图首轮消息中自动注入已确认的 order_id 实体，避免用户重复输入；对于需要收集多字段的工单场景，使用 Pydantic Form 模型记录已收集字段，避免子图内部用散乱的 flag 字段维护状态；流式回复先输出"您好，"等开场，再边生成边发送，首字延迟控制在 500ms 以内。

测试方案分为四层。第一层意图识别测试：维护 200+ 条标注语料覆盖六类意图与歧义问法，使用 pytest 参数化批量校验 intent 准确率、order_id 抽取 F1、置信度阈值下的人工兜底率，目标整体准确率 ≥ 0.92，兜底率 ≤ 8%。第二层退款审批流程测试：覆盖 approve/reject/modify/超时四种路径，断言 refund_execute_tool 调用次数与资金接口不被重复触发、checkpoint 状态迁移符合预期、幂等键生效。第三层并发 session 隔离测试：以 asyncio.gather 启动 50 个并发 session，每个 session 使用独立 thread_id，交叉校验 messages 不串流、order_id 不串写、human_approval_pending 不跨会话污染。第四层工具失败降级测试：以 respx/pytest-httpx mock 订单接口 500、超时、空返回，断言降级文案出现且转人工被触发。第五层 HITL 恢复测试：通过 graph.get_state 读取 interrupt 状态，注入 approve/reject 两种 resume 值，断言后续节点按分支执行、历史 messages 不丢失。所有测试以 pytest + pytest-asyncio 编写，纳入 CI，Golden Dataset 回归作为发版门禁。

数据与合规：用户消息中可能包含手机号、地址等 PII（Personally Identifiable Information，个人身份信息），日志落盘前需用正则脱敏；退款执行必须校验审批人权限（角色、金额阈值、与工单的绑定关系）；会话摘要写入 Store 前去除敏感字段；审计日志保留 180 天以上以满足合规要求。