【langgraph 从入门到精通】全面实践项目：智能客服工单系统

文章目录

• 第 11 章：全面实践项目：智能客服工单系统
• • 11.1 本章目标
  • 11.2 项目概述
  • • 11.2.1 系统描述
    • 11.2.2 系统架构图
    • 11.2.3 项目结构
  • 11.3 核心概念（图文讲解）
  • • 11.3.1 生动类比：智能客服中心
    • 11.3.2 意图识别流程图
  • 11.4 实战：完整代码
  • • 11.4.1 models.py — 数据模型
    • 11.4.2 knowledge_base.py — 知识库
    • 11.4.3 agents.py — 多 Agent 定义
    • 11.4.4 main.py — 主程序
    • 11.4.5 代码逐段解析
  • 11.5 API 速查
  • 11.6 单元测试
  • • 11.6.1 tests/test_workflow.py
    • 11.6.2 requirements.txt
    • 11.6.3 Dockerfile
    • 11.6.4 docker-compose.yml
  • 11.7 常见错误与避坑指南
  • • 错误 1：忘记在 Entrypoint 配置 checkpointer 就使用 interrupt()
    • 错误 2：恢复执行时使用了错误的 thread_id
    • 错误 3：RAG 检索时没有考虑意图分类，检索结果不准确
    • 错误 4：工单创建后没有更新状态，导致工单一直处于 OPEN 状态
    • 错误 5：没有对用户输入做验证，导致恶意输入或异常输入
  • 11.8 最佳实践总结
  • 结语：从零到 Hero 的旅程

第 11 章：全面实践项目：智能客服工单系统

11.1 本章目标

学完本章后，你将能够：

• 从零构建一个完整的生产级智能客服工单系统
• 掌握意图识别、RAG 检索、工单创建、人工客服介入等核心能力
• 实现多 Agent 协作（技术支持 + 账单咨询 + 一般客服）的完整工作流
• 使用 Checkpointing 实现对话历史持久化
• 编写单元测试和 Docker 部署配置
• 将前 10 章所学的所有知识点融会贯通

11.2 项目概述

11.2.1 系统描述

本章将构建一个智能客服工单系统（Smart Ticket System），它的核心功能是：

11.2.2 系统架构图

#mermaid-svg-OClaFPYaby4b4YLn{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-OClaFPYaby4b4YLn .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-OClaFPYaby4b4YLn .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-OClaFPYaby4b4YLn .error-icon{fill:#552222;}#mermaid-svg-OClaFPYaby4b4YLn .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-OClaFPYaby4b4YLn .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-OClaFPYaby4b4YLn .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-OClaFPYaby4b4YLn .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-OClaFPYaby4b4YLn .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-OClaFPYaby4b4YLn .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-OClaFPYaby4b4YLn .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-OClaFPYaby4b4YLn .marker{fill:#333333;stroke:#333333;}#mermaid-svg-OClaFPYaby4b4YLn .marker.cross{stroke:#333333;}#mermaid-svg-OClaFPYaby4b4YLn svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-OClaFPYaby4b4YLn p{margin:0;}#mermaid-svg-OClaFPYaby4b4YLn .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-OClaFPYaby4b4YLn .cluster-label text{fill:#333;}#mermaid-svg-OClaFPYaby4b4YLn .cluster-label span{color:#333;}#mermaid-svg-OClaFPYaby4b4YLn .cluster-label span p{background-color:transparent;}#mermaid-svg-OClaFPYaby4b4YLn .label text,#mermaid-svg-OClaFPYaby4b4YLn span{fill:#333;color:#333;}#mermaid-svg-OClaFPYaby4b4YLn .node rect,#mermaid-svg-OClaFPYaby4b4YLn .node circle,#mermaid-svg-OClaFPYaby4b4YLn .node ellipse,#mermaid-svg-OClaFPYaby4b4YLn .node polygon,#mermaid-svg-OClaFPYaby4b4YLn .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-OClaFPYaby4b4YLn .rough-node .label text,#mermaid-svg-OClaFPYaby4b4YLn .node .label text,#mermaid-svg-OClaFPYaby4b4YLn .image-shape .label,#mermaid-svg-OClaFPYaby4b4YLn .icon-shape .label{text-anchor:middle;}#mermaid-svg-OClaFPYaby4b4YLn .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-OClaFPYaby4b4YLn .rough-node .label,#mermaid-svg-OClaFPYaby4b4YLn .node .label,#mermaid-svg-OClaFPYaby4b4YLn .image-shape .label,#mermaid-svg-OClaFPYaby4b4YLn .icon-shape .label{text-align:center;}#mermaid-svg-OClaFPYaby4b4YLn .node.clickable{cursor:pointer;}#mermaid-svg-OClaFPYaby4b4YLn .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-OClaFPYaby4b4YLn .arrowheadPath{fill:#333333;}#mermaid-svg-OClaFPYaby4b4YLn .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-OClaFPYaby4b4YLn .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-OClaFPYaby4b4YLn .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-OClaFPYaby4b4YLn .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-OClaFPYaby4b4YLn .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-OClaFPYaby4b4YLn .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-OClaFPYaby4b4YLn .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-OClaFPYaby4b4YLn .cluster text{fill:#333;}#mermaid-svg-OClaFPYaby4b4YLn .cluster span{color:#333;}#mermaid-svg-OClaFPYaby4b4YLn div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-OClaFPYaby4b4YLn .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-OClaFPYaby4b4YLn rect.text{fill:none;stroke-width:0;}#mermaid-svg-OClaFPYaby4b4YLn .icon-shape,#mermaid-svg-OClaFPYaby4b4YLn .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-OClaFPYaby4b4YLn .icon-shape p,#mermaid-svg-OClaFPYaby4b4YLn .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-OClaFPYaby4b4YLn .icon-shape .label rect,#mermaid-svg-OClaFPYaby4b4YLn .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-OClaFPYaby4b4YLn .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-OClaFPYaby4b4YLn .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-OClaFPYaby4b4YLn :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

持久化层

多 Agent 协作层

Agent 引擎层

接入层

用户层

技术问题

账单问题

一般咨询

AI 可解决

AI 无法解决

用户

FastAPI 服务

Entrypoint客服工作流入口

Task: 意图识别classify_intent

Task: RAG 检索search_knowledge_base

Task: 自动回复generate_response

Task: 创建工单create_ticket

Task: 人工介入wait_for_human

Supervisor路由决策

技术支持 AgentTech Support

账单咨询 AgentBilling Support

一般客服 AgentGeneral Support

Checkpointer对话历史

知识库文档检索

工单数据库SQLite

11.2.3 项目结构

smart_ticket_system/
├── main.py # 主程序：Agent 工作流定义
├── models.py # 数据模型：State 定义
├── knowledge_base.py # 知识库：文档索引和检索
├── agents.py # 多 Agent 定义
├── api.py # FastAPI 服务
├── tests/
│ └── test_workflow.py # 单元测试
├── requirements.txt # 依赖清单
├── Dockerfile # Docker 部署配置
└── docker-compose.yml # Docker Compose 配置

11.3 核心概念（图文讲解）

11.3.1 生动类比：智能客服中心

想象你打电话给一家大公司的客服热线。你的体验是这样的：

11.3.2 意图识别流程图

多Agent路由

人工介入

创建工单

自动回复

RAG检索

意图识别

Entrypoint

用户

多Agent路由

人工介入

创建工单

自动回复

RAG检索

意图识别

Entrypoint

用户

#mermaid-svg-Nczv1VHHRYwRkfCP{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-Nczv1VHHRYwRkfCP .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-Nczv1VHHRYwRkfCP .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-Nczv1VHHRYwRkfCP .error-icon{fill:#552222;}#mermaid-svg-Nczv1VHHRYwRkfCP .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-Nczv1VHHRYwRkfCP .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-Nczv1VHHRYwRkfCP .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-Nczv1VHHRYwRkfCP .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-Nczv1VHHRYwRkfCP .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-Nczv1VHHRYwRkfCP .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-Nczv1VHHRYwRkfCP .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-Nczv1VHHRYwRkfCP .marker{fill:#333333;stroke:#333333;}#mermaid-svg-Nczv1VHHRYwRkfCP .marker.cross{stroke:#333333;}#mermaid-svg-Nczv1VHHRYwRkfCP svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-Nczv1VHHRYwRkfCP p{margin:0;}#mermaid-svg-Nczv1VHHRYwRkfCP .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-Nczv1VHHRYwRkfCP text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-Nczv1VHHRYwRkfCP .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-Nczv1VHHRYwRkfCP .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-Nczv1VHHRYwRkfCP .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-Nczv1VHHRYwRkfCP .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-Nczv1VHHRYwRkfCP #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-Nczv1VHHRYwRkfCP .sequenceNumber{fill:white;}#mermaid-svg-Nczv1VHHRYwRkfCP #sequencenumber{fill:#333;}#mermaid-svg-Nczv1VHHRYwRkfCP #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-Nczv1VHHRYwRkfCP .messageText{fill:#333;stroke:none;}#mermaid-svg-Nczv1VHHRYwRkfCP .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-Nczv1VHHRYwRkfCP .labelText,#mermaid-svg-Nczv1VHHRYwRkfCP .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-Nczv1VHHRYwRkfCP .loopText,#mermaid-svg-Nczv1VHHRYwRkfCP .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-Nczv1VHHRYwRkfCP .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-Nczv1VHHRYwRkfCP .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-Nczv1VHHRYwRkfCP .noteText,#mermaid-svg-Nczv1VHHRYwRkfCP .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-Nczv1VHHRYwRkfCP .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-Nczv1VHHRYwRkfCP .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-Nczv1VHHRYwRkfCP .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-Nczv1VHHRYwRkfCP .actorPopupMenu{position:absolute;}#mermaid-svg-Nczv1VHHRYwRkfCP .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-Nczv1VHHRYwRkfCP .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-Nczv1VHHRYwRkfCP .actor-man circle,#mermaid-svg-Nczv1VHHRYwRkfCP line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-Nczv1VHHRYwRkfCP :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

alt

[AI 无法解决]

"我的订单支付失败了"

classify_intent()

intent="billing", confidence=0.95

search_knowledge_base("支付失败")

检索到 3 条相关文档

generate_response("支付失败", documents)

auto_reply + can_resolve=False

create_ticket("支付失败", "billing")

ticket_id="TK-20260712-001"

wait_for_human(ticket)

human_response

route_to_agent(intent="billing")

路由到账单咨询 Agent

billing_agent_response

最终回复 + 工单编号

11.4 实战：完整代码

11.4.1 models.py — 数据模型

# ============================================================
# models.py — 智能客服工单系统的数据模型
# ============================================================

from pydantic import BaseModel, Field # 导入 Pydantic 数据模型
from typing import Annotated, List, Optional, Dict, Any # 导入类型注解
import operator # 导入 operator 模块，用于 reducer
from enum import Enum # 导入枚举类型

# —- 意图枚举 —-
class IntentType(str, Enum):
"""用户意图类型枚举
使用 Enum 确保意图类型是有限且确定的，避免拼写错误。
"""
TECHNICAL = "technical" # 技术问题：如系统故障、API 报错
BILLING = "billing" # 账单问题：如支付失败、退款、发票
GENERAL = "general" # 一般咨询：如产品使用、功能询问

# —- 工单状态枚举 —-
class TicketStatus(str, Enum):
"""工单状态枚举"""
OPEN = "open" # 已创建，等待处理
IN_PROGRESS = "in_progress" # 处理中
WAITING_HUMAN = "waiting_human" # 等待人工介入
RESOLVED = "resolved" # 已解决
CLOSED = "closed" # 已关闭

# —- 工单模型 —-
class Ticket(BaseModel):
"""工单数据模型
每个工单代表一个用户的客服请求，包含完整的问题信息和处理状态。
"""
ticket_id: str = Field(default="", description="工单编号，格式：TK-YYYYMMDD-NNN")
intent: IntentType = Field(default=IntentType.GENERAL, description="意图类型")
subject: str = Field(default="", description="工单主题")
description: str = Field(default="", description="问题详细描述")
status: TicketStatus = Field(default=TicketStatus.OPEN, description="工单状态")
priority: int = Field(default=3, description="优先级：1-最高，5-最低")
created_at: str = Field(default="", description="创建时间")
assigned_agent: str = Field(default="", description="分配的客服 Agent")
resolution: str = Field(default="", description="解决方案")

# —- 知识库文档模型 —-
class KnowledgeDocument(BaseModel):
"""知识库文档模型"""
doc_id: str = Field(default="", description="文档 ID")
title: str = Field(default="", description="文档标题")
content: str = Field(default="", description="文档内容")
category: IntentType = Field(default=IntentType.GENERAL, description="文档分类")
tags: List[str] = Field(default_factory=list, description="标签列表")

# —- 主 State 模型 —-
class TicketSystemState(BaseModel):
"""智能客服工单系统的主状态
这个 State 贯穿整个工作流，所有 Task 共享和修改它。
"""
# 对话历史（使用 operator.add 作为 reducer，新消息追加到列表）
messages: Annotated[List[Dict[str, Any]], operator.add] = Field(
default_factory=list,
description="完整对话历史，每条消息包含 role 和 content",
)
# 用户输入
user_input: str = Field(default="", description="当前用户输入")
# 意图识别结果
intent: IntentType = Field(default=IntentType.GENERAL, description="识别的用户意图")
intent_confidence: float = Field(default=0.0, description="意图识别的置信度")
# RAG 检索结果
retrieved_docs: List[KnowledgeDocument] = Field(
default_factory=list,
description="从知识库检索到的相关文档",
)
# 自动回复
auto_reply: str = Field(default="", description="AI 自动生成的回复")
can_resolve: bool = Field(default=True, description="AI 是否能解决此问题")
# 工单
ticket: Optional[Ticket] = Field(default=None, description="创建的工单")
# 人工介入
human_response: str = Field(default="", description="人工客服的回复")
needs_human: bool = Field(default=False, description="是否需要人工介入")
# 多 Agent 协作
agent_response: str = Field(default="", description="专业 Agent 的回复")
# 最终输出
final_output: str = Field(default="", description="最终返回给用户的回复")

11.4.2 knowledge_base.py — 知识库

# ============================================================
# knowledge_base.py — 知识库（简化版 RAG）
# 在生产环境中，应替换为 Chroma/Pinecone/Weaviate 等向量数据库
# ============================================================

from models import KnowledgeDocument, IntentType # 导入数据模型
from typing import List # 导入类型注解

# ============================================================
# 模拟知识库（生产环境替换为向量数据库 + 嵌入模型）
# ============================================================
# 这里我们使用一个简单的文档列表来模拟知识库
# 在实际生产环境中，你需要：
# 1. 使用嵌入模型（如 text-embedding-3-small）将文档转换为向量
# 2. 将向量存储在 Chroma/Pinecone/Weaviate 中
# 3. 使用余弦相似度或欧几里得距离进行语义搜索
KNOWLEDGE_DOCS: List[KnowledgeDocument] = [
# —- 技术问题文档 —-
KnowledgeDocument(
doc_id="KB-TECH-001",
title="API 连接超时问题排查",
content="如果 API 连接超时，请检查：1. 网络连接是否正常；2. API Key 是否有效；3. 请求频率是否超过限制（默认 60 次/分钟）；4. 服务器是否正常运行。如果问题持续，请提供错误日志。",
category=IntentType.TECHNICAL,
tags=["API", "超时", "连接", "错误排查"],
),
KnowledgeDocument(
doc_id="KB-TECH-002",
title="系统 500 错误处理指南",
content="遇到 500 错误通常表示服务器内部错误。处理步骤：1. 刷新页面重试；2. 清除浏览器缓存；3. 检查请求参数是否正确；4. 如果持续出现，请记录错误时间和操作步骤，提交工单给技术团队。",
category=IntentType.TECHNICAL,
tags=["500", "服务器错误", "错误排查"],
),
KnowledgeDocument(
doc_id="KB-TECH-003",
title="SDK 安装与配置常见问题",
content="SDK 安装失败常见原因：1. Python 版本不兼容（需要 3.10+）；2. pip 版本过旧（运行 pip install –upgrade pip）；3. 依赖冲突（使用虚拟环境 venv）；4. 网络问题（使用国内镜像源）。",
category=IntentType.TECHNICAL,
tags=["SDK", "安装", "配置", "Python"],
),
# —- 账单问题文档 —-
KnowledgeDocument(
doc_id="KB-BILL-001",
title="支付失败常见原因与解决方案",
content="支付失败可能的原因：1. 银行卡余额不足；2. 银行卡已过期；3. 银行风控拦截；4. 支付限额超限。解决方案：1. 检查银行卡状态；2. 尝试其他支付方式；3. 联系发卡银行；4. 如果是系统问题，请提交工单处理。",
category=IntentType.BILLING,
tags=["支付失败", "银行卡", "风控", "退款"],
),
KnowledgeDocument(
doc_id="KB-BILL-002",
title="退款政策与流程",
content="退款政策：1. 购买后 7 天内可全额退款；2. 7-30 天按比例退款；3. 超过 30 天不可退款。退款流程：登录账户 → 订单管理 → 选择订单 → 申请退款 → 填写原因 → 提交。退款通常在 3-5 个工作日内到账。",
category=IntentType.BILLING,
tags=["退款", "退款政策", "退款流程", "订单"],
),
KnowledgeDocument(
doc_id="KB-BILL-003",
title="发票开具指南",
content="发票开具流程：1. 登录账户进入订单详情；2. 点击「申请发票」；3. 填写发票抬头和税号；4. 选择发票类型（电子/纸质）；5. 提交申请。电子发票一般在 1 小时内发送到注册邮箱，纸质发票 3-5 个工作日寄出。",
category=IntentType.BILLING,
tags=["发票", "电子发票", "纸质发票", "税号"],
),
# —- 一般咨询文档 —-
KnowledgeDocument(
doc_id="KB-GEN-001",
title="产品功能介绍",
content="我们的产品提供以下核心功能：1. 智能数据分析（支持 CSV/Excel/JSON 导入）；2. 可视化图表（支持 20+ 图表类型）；3. 团队协作（实时协作编辑）；4. API 集成（RESTful API 和 SDK）；5. 自动化报告（定时生成和发送）。",
category=IntentType.GENERAL,
tags=["产品介绍", "功能", "数据分析", "图表"],
),
KnowledgeDocument(
doc_id="KB-GEN-002",
title="账号注册与登录问题",
content="注册流程：1. 访问官网点击注册；2. 输入邮箱和密码；3. 验证邮箱；4. 完成注册。如果收不到验证邮件，请检查垃圾邮件箱，或联系客服。登录问题：如果忘记密码，点击「忘记密码」重置。如果账号被锁定，等待 15 分钟后重试。",
category=IntentType.GENERAL,
tags=["注册", "登录", "账号", "密码", "验证"],
),
KnowledgeDocument(
doc_id="KB-GEN-003",
title="价格方案与套餐对比",
content="我们提供三种套餐：1. 免费版（基础功能，每月 1000 次 API 调用）；2. 专业版（¥99/月，完整功能，每月 10000 次 API 调用）；3. 企业版（¥499/月，全部功能 + 专属支持，无限制 API 调用）。所有套餐均支持 7 天免费试用。",
category=IntentType.GENERAL,
tags=["价格", "套餐", "免费", "专业版", "企业版"],
),
]

# ============================================================
# 简易关键词检索函数（生产环境替换为语义搜索）
# ============================================================
def search_knowledge(query: str, top_k: int = 3) –> List[KnowledgeDocument]:
"""基于关键词匹配的简易检索函数
在生产环境中，应替换为向量相似度搜索（语义搜索）。
这里使用简单的关键词匹配作为演示。

参数：
query: 用户查询文本
top_k: 返回的最相关文档数量

返回：
匹配的文档列表，按匹配度排序
"""
# 将查询转为小写，便于匹配
query_lower = query.lower()
# 将查询拆分为关键词列表
query_keywords = set(query_lower.split())

# 计算每个文档的匹配分数
scored_docs = []
for doc in KNOWLEDGE_DOCS:
score = 0 # 匹配分数初始化
# 检查标题匹配
if any(kw in doc.title.lower() for kw in query_keywords):
score += 3 # 标题匹配权重最高
# 检查标签匹配
for tag in doc.tags:
if tag.lower() in query_lower:
score += 2 # 标签匹配权重次之
# 检查内容关键词匹配
content_lower = doc.content.lower()
for kw in query_keywords:
if len(kw) > 1 and kw in content_lower: # 忽略单字匹配
score += 1 # 内容匹配权重最低

if score > 0:
scored_docs.append((score, doc))

# 按分数降序排序，取 top_k 个
scored_docs.sort(key=lambda x: x[0], reverse=True)
return [doc for score, doc in scored_docs[:top_k]]

11.4.3 agents.py — 多 Agent 定义

# ============================================================
# agents.py — 多 Agent 协作层
# 使用 Supervisor 模式管理三个专业 Agent
# ============================================================

from langgraph.func import entrypoint, task # 导入 LangGraph Functional API
from langgraph.types import Command # 导入 Command 类型
from langgraph.checkpoint.memory import MemorySaver # 导入内存检查点
from langchain_openai import ChatOpenAI # 导入 LLM

from models import TicketSystemState, IntentType, Ticket, TicketStatus # 导入数据模型
from knowledge_base import search_knowledge # 导入知识库检索函数

import os # 导入操作系统模块

# ============================================================
# 初始化 LLM 和检查点
# ============================================================
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7) # 初始化 LLM
checkpointer = MemorySaver() # 初始化检查点存储（生产环境替换为 SqliteSaver）

# ============================================================
# 定义三个专业 Agent（每个都是独立的 @entrypoint 子图）
# ============================================================

# —- Agent 1：技术支持 —-
@entrypoint()
def tech_support_agent(state: TicketSystemState) –> TicketSystemState:
"""技术支持 Agent：专门处理技术问题
当用户的问题涉及技术故障、API 错误、系统问题等时，
Supervisor 会将任务路由到这个 Agent。
"""
# 构建系统提示词
system_prompt = (
"你是一位经验丰富的技术支持工程师。你的职责是：\\n"
"1. 分析用户遇到的技术问题\\n"
"2. 提供详细的排查步骤和解决方案\\n"
"3. 如果问题超出了你的能力范围，建议用户升级到高级支持\\n"
"4. 使用专业但易懂的语言回复用户\\n"
"请基于以下知识库内容回答问题。"
)

# 如果工单存在，加入工单上下文
ticket_context = ""
if state.ticket:
ticket_context = f"\\n当前工单编号：{state.ticket.ticket_id}\\n工单主题：{state.ticket.subject}"

# 构建完整消息
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"用户问题：{state.user_input}\\n{ticket_context}"},
]

# 调用 LLM 生成回复
response = llm.invoke(messages)
# 更新 state
state.agent_response = response.content
state.messages.append({"role": "assistant", "content": response.content})
return state

# —- Agent 2：账单咨询 —-
@entrypoint()
def billing_support_agent(state: TicketSystemState) –> TicketSystemState:
"""账单咨询 Agent：专门处理支付、退款、发票等账单问题"""
system_prompt = (
"你是一位专业的账单客服专员。你的职责是：\\n"
"1. 处理用户的支付、退款、发票等账单相关问题\\n"
"2. 准确解释退款政策和流程\\n"
"3. 帮助用户理解账单明细\\n"
"4. 对于需要人工处理的操作（如手动退款），告知用户预期处理时间\\n"
"请以专业、耐心的态度回复用户。"
)

ticket_context = ""
if state.ticket:
ticket_context = f"\\n当前工单编号：{state.ticket.ticket_id}\\n工单主题：{state.ticket.subject}"

messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"用户问题：{state.user_input}\\n{ticket_context}"},
]

response = llm.invoke(messages)
state.agent_response = response.content
state.messages.append({"role": "assistant", "content": response.content})
return state

# —- Agent 3：一般客服 —-
@entrypoint()
def general_support_agent(state: TicketSystemState) –> TicketSystemState:
"""一般客服 Agent：处理产品咨询、账号问题等一般性问题"""
system_prompt = (
"你是一位专业的一般客服代表。你的职责是：\\n"
"1. 回答用户关于产品功能、使用方法的咨询\\n"
"2. 帮助用户解决账号注册、登录等问题\\n"
"3. 介绍产品价格方案和套餐对比\\n"
"4. 以友好、热情的态度服务每一位用户\\n"
"如果用户的问题涉及技术或账单，请告知用户你会帮忙转接。"
)

ticket_context = ""
if state.ticket:
ticket_context = f"\\n当前工单编号：{state.ticket.ticket_id}\\n工单主题：{state.ticket.subject}"

messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"用户问题：{state.user_input}\\n{ticket_context}"},
]

response = llm.invoke(messages)
state.agent_response = response.content
state.messages.append({"role": "assistant", "content": response.content})
return state

# ============================================================
# Agent 路由函数：根据意图类型决定使用哪个 Agent
# ============================================================
def route_to_agent(state: TicketSystemState) –> TicketSystemState:
"""根据意图类型路由到对应的专业 Agent
这是 Supervisor 模式的核心路由逻辑。
规则：
– TECHNICAL（技术问题） → tech_support_agent
– BILLING（账单问题） → billing_support_agent
– GENERAL（一般咨询） → general_support_agent
"""
# 根据意图类型选择对应的 Agent
if state.intent == IntentType.TECHNICAL:
result = tech_support_agent(state).result() # 调用技术支持 Agent
elif state.intent == IntentType.BILLING:
result = billing_support_agent(state).result() # 调用账单咨询 Agent
else:
result = general_support_agent(state).result() # 调用一般客服 Agent

return result

11.4.4 main.py — 主程序

# ============================================================
# main.py — 智能客服工单系统主程序
# 这是系统的核心，将所有模块串联成完整的工作流
# 保存为 main.py 后运行：python main.py
# 运行前请确保已设置：export OPENAI_API_KEY="your-key"
# ============================================================

import os # 导入操作系统模块
import uuid # 导入 UUID 模块，用于生成唯一 ID
from datetime import datetime # 导入日期时间模块
from typing import Optional # 导入类型注解

# —- LangGraph 核心 —-
from langgraph.func import entrypoint, task # Functional API 核心
from langgraph.types import Command, interrupt # Command 和 interrupt
from langgraph.checkpoint.memory import MemorySaver # 检查点存储

# —- LangChain —-
from langchain_openai import ChatOpenAI # LLM 客户端

# —- 项目模块 —-
from models import ( # 导入数据模型
TicketSystemState,
IntentType,
Ticket,
TicketStatus,
KnowledgeDocument,
)
from knowledge_base import search_knowledge # 导入知识库检索
from agents import route_to_agent # 导入 Agent 路由

# ============================================================
# 初始化 LLM 和检查点
# ============================================================
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7) # 初始化 LLM
checkpointer = MemorySaver() # 初始化检查点（生产环境替换为 SqliteSaver）

# ============================================================
# Task 1：意图识别
# ============================================================
@task
def classify_intent(state: TicketSystemState) –> TicketSystemState:
"""识别用户意图：技术问题、账单问题还是一般咨询
这是工作流的第一个 Task，其输出决定了后续的执行路径。
"""
# 构建意图识别的提示词
prompt = f"""你是一个客服意图识别系统。请分析以下用户消息，判断其意图类型。

用户消息："{state.user_input}"

意图类型（三选一）：
– technical：技术问题（系统故障、API 错误、SDK 安装、性能问题等）
– billing：账单问题（支付失败、退款、发票、价格咨询等）
– general：一般咨询（产品功能、账号注册、使用方法等）

请只返回意图类型（一个单词），不要返回其他内容。
示例：technical
"""
# 调用 LLM 进行意图识别
response = llm.invoke(prompt)
intent_str = response.content.strip().lower() # 清理输出

# 将 LLM 返回的字符串映射到 IntentType 枚举
intent_map = {
"technical": IntentType.TECHNICAL,
"billing": IntentType.BILLING,
"general": IntentType.GENERAL,
}
state.intent = intent_map.get(intent_str, IntentType.GENERAL) # 默认归为一般咨询
state.intent_confidence = 0.9 # 在实际项目中，可以从 LLM 的 logprobs 获取真实置信度

# 记录意图识别结果
state.messages.append({
"role": "system",
"content": f"[系统] 意图识别结果：{state.intent.value}（置信度：{state.intent_confidence}）",
})
return state

# ============================================================
# Task 2：RAG 检索
# ============================================================
@task
def search_knowledge_base(state: TicketSystemState) –> TicketSystemState:
"""从知识库检索相关文档（RAG）
根据用户输入和识别的意图，从知识库中检索最相关的文档。
检索到的文档将作为后续 LLM 调用的上下文，提高回答质量。
"""
# 调用知识库检索函数
state.retrieved_docs = search_knowledge(state.user_input, top_k=3)

# 记录检索结果
doc_titles = [doc.title for doc in state.retrieved_docs]
state.messages.append({
"role": "system",
"content": f"[系统] RAG 检索到 {len(state.retrieved_docs)} 条相关文档：{', '.join(doc_titles)}",
})
return state

# ============================================================
# Task 3：自动回复生成
# ============================================================
@task
def generate_response(state: TicketSystemState) –> TicketSystemState:
"""基于知识库检索结果生成自动回复
如果检索到了相关文档，LLM 将基于这些文档生成回复。
同时，LLM 还会判断自己是否能解决这个问题。
"""
# 构建包含知识库内容的提示词
kb_context = "" # 知识库上下文
if state.retrieved_docs:
kb_context = "相关知识库内容：\\n"
for i, doc in enumerate(state.retrieved_docs, 1):
kb_context += f"\\n文档 {i}：{doc.title}\\n{doc.content}\\n"

# 提示词要求 LLM 同时返回回复和是否能解决的判断
prompt = f"""你是一个智能客服助手。请根据知识库内容回答用户的问题。

{kb_context}

用户问题：{state.user_input}

请按以下格式回复（严格遵守格式）：

【能否解决】：是 或 否
【回复内容】：
（你的回复内容，基于知识库信息。如果知识库中没有相关信息，请诚实说明。）

注意：
– 如果知识库中有明确答案，标记【能否解决】：是
– 如果知识库中没有相关信息，或者问题需要人工处理（如退款操作、技术深度排查），标记【能否解决】：否
– 回复内容要友好、专业、有帮助
"""
# 调用 LLM 生成回复
response = llm.invoke(prompt)
response_text = response.content

# 解析 LLM 返回的内容，提取「能否解决」和「回复内容」
can_resolve = True # 默认认为可以解决
reply_content = response_text # 默认整个回复都是回复内容

# 尝试解析「能否解决」标记
if "【能否解决】：" in response_text:
parts = response_text.split("【能否解决】：")
if len(parts) > 1:
after_marker = parts[1]
if "\\n" in after_marker:
can_resolve_str = after_marker.split("\\n")[0].strip()
can_resolve = "是" in can_resolve_str # 包含「是」则认为是可解决的
# 提取回复内容
if "【回复内容】：" in response_text:
reply_parts = response_text.split("【回复内容】：")
if len(reply_parts) > 1:
reply_content = reply_parts[1].strip()

# 更新 state
state.auto_reply = reply_content
state.can_resolve = can_resolve
state.messages.append({
"role": "assistant",
"content": reply_content,
})
return state

# ============================================================
# Task 4：创建工单
# ============================================================
@task
def create_ticket(state: TicketSystemState) –> TicketSystemState:
"""创建工单
当 AI 无法解决问题时，自动创建工单，记录问题信息。
工单会分配给对应类型的专业 Agent 处理。
"""
# 生成工单编号：TK-YYYYMMDD-NNN
today = datetime.now().strftime("%Y%m%d") # 当前日期
ticket_id = f"TK-{today}–{uuid.uuid4().hex[:6].upper()}" # 唯一工单编号

# 创建工单对象
ticket = Ticket(
ticket_id=ticket_id,
intent=state.intent,
subject=f"【{state.intent.value}】{state.user_input[:50]}", # 截取前 50 字符作为主题
description=state.user_input,
status=TicketStatus.OPEN, # 初始状态：已创建
priority=3, # 默认优先级
created_at=datetime.now().isoformat(), # 创建时间
assigned_agent=state.intent.value, # 根据意图类型分配 Agent
)

# 更新 state
state.ticket = ticket
state.needs_human = True # 标记需要人工介入
state.messages.append({
"role": "system",
"content": f"[系统] 已创建工单 {ticket_id}，状态：{ticket.status.value}，分配给：{ticket.assigned_agent}",
})
return state

# ============================================================
# Task 5：人工客服介入（Human-in-the-loop）
# ============================================================
@task
def wait_for_human(state: TicketSystemState) –> TicketSystemState:
"""等待人工客服介入
使用 interrupt() 暂停工作流，等待人类客服查看工单并回复。
这是 Human-in-the-loop 模式的核心实现。
"""
# 构建中断信息（人类客服看到的内容）
interrupt_message = {
"action": "需要人工客服介入",
"ticket_id": state.ticket.ticket_id if state.ticket else "N/A",
"intent": state.intent.value,
"user_question": state.user_input,
"auto_reply": state.auto_reply,
"instruction": "请审核自动回复并提供人工回复。如需修改，输入修改后的回复。",
}

# 调用 interrupt() 暂停工作流
# 工作流会在此处暂停，直到外部调用 Command(resume=…) 恢复
human_decision = interrupt(interrupt_message)

# 人类客服的回复（通过 Command(resume=…) 传入）
if isinstance(human_decision, dict):
state.human_response = human_decision.get("response", str(human_decision))
# 更新工单状态
if state.ticket:
state.ticket.status = TicketStatus.IN_PROGRESS
state.ticket.resolution = human_decision.get("resolution", "")
elif isinstance(human_decision, str):
state.human_response = human_decision
if state.ticket:
state.ticket.status = TicketStatus.IN_PROGRESS
else:
state.human_response = str(human_decision)

state.messages.append({
"role": "human",
"content": f"[人工客服] {state.human_response}",
})
return state

# ============================================================
# Task 6：多 Agent 协作处理
# ============================================================
@task
def process_with_agent(state: TicketSystemState) –> TicketSystemState:
"""将任务路由到对应的专业 Agent 处理
根据意图类型，将工单分配给技术支持、账单咨询或一般客服 Agent。
"""
# 调用路由函数，根据意图类型选择 Agent
state = route_to_agent(state)
return state

# ============================================================
# Entrypoint：主工作流
# ============================================================
@entrypoint(checkpointer=checkpointer) # 配置检查点，支持多轮对话和中断恢复
def ticket_system_workflow(state: TicketSystemState) –> TicketSystemState:
"""智能客服工单系统主工作流
工作流步骤：
1. 意图识别 → 2. 知识库检索 → 3. 自动回复 → 4. 判断是否需要人工介入
→ 5a. 不需要 → 直接返回 AI 回复
→ 5b. 需要 → 创建工单 → 人工介入 → 多 Agent 协作 → 返回最终回复
"""
# —- 步骤 1：意图识别 —-
state = classify_intent(state).result()

# —- 步骤 2：知识库检索 —-
state = search_knowledge_base(state).result()

# —- 步骤 3：自动回复生成 —-
state = generate_response(state).result()

# —- 步骤 4：判断是否需要人工介入 —-
if not state.can_resolve:
# —- 步骤 5a：AI 无法解决，需要人工介入 —-
# 创建工单
state = create_ticket(state).result()
# 等待人工客服介入（interrupt）
state = wait_for_human(state).result()
# 多 Agent 协作处理
state = process_with_agent(state).result()

# 生成最终输出（包含人工回复和 Agent 回复）
final_parts = []
if state.ticket:
final_parts.append(f"工单编号：{state.ticket.ticket_id}")
if state.human_response:
final_parts.append(f"人工客服回复：{state.human_response}")
if state.agent_response:
final_parts.append(f"专业顾问建议：{state.agent_response}")
state.final_output = "\\n\\n".join(final_parts)
else:
# —- 步骤 5b：AI 可以解决，直接返回自动回复 —-
state.final_output = state.auto_reply
state.messages.append({
"role": "system",
"content": "[系统] AI 自动解决，无需人工介入。",
})

return state

# ============================================================
# 主程序入口：演示三个场景
# ============================================================
if __name__ == "__main__":
print("=" * 70)
print("智能客服工单系统 v1.0")
print("=" * 70)
print()

# —- 场景 1：技术问题（AI 可解决） —-
print("【场景 1】技术问题 — AI 可解决")
print("-" * 50)
state1 = TicketSystemState(
user_input="我的 API 调用总是超时，怎么排查？",
)
result1 = ticket_system_workflow.invoke(state1)
print(f"意图：{result1.intent.value}")
print(f"AI 可解决：{result1.can_resolve}")
print(f"回复：\\n{result1.final_output[:300]}…")
print()

# —- 场景 2：账单问题（AI 无法解决，需要人工介入） —-
print("【场景 2】账单问题 — 需要人工介入")
print("-" * 50)
# 使用唯一的 thread_id 确保会话隔离
thread_id = f"session-{uuid.uuid4().hex[:8]}"
config = {"configurable": {"thread_id": thread_id}}

state2 = TicketSystemState(
user_input="我的订单支付失败了，钱已经扣了但订单没生成，请帮我退款",
)

try:
# 第一次调用：会触发 interrupt 并抛出 GraphInterrupt 异常
result2 = ticket_system_workflow.invoke(state2, config=config)
except Exception as e:
# 捕获 GraphInterrupt 异常（interrupt 中断时抛出）
print(f"工作流已暂停，等待人工客服介入…")
print(f"异常类型：{type(e).__name__}")

# 模拟人工客服回复
human_response = {
"response": "您好，我们已经收到您的退款请求。经核实，您的支付确实已扣款但订单未生成。我们已为您发起退款流程，预计 3-5 个工作日内到账。",
"resolution": "已发起退款，退款编号：RF-20260712-001",
}

# 使用 Command(resume=…) 恢复工作流
resume_cmd = Command(resume=human_response)
result2 = ticket_system_workflow.invoke(resume_cmd, config=config)

print(f"意图：{result2.intent.value}")
print(f"工单编号：{result2.ticket.ticket_id if result2.ticket else 'N/A'}")
print(f"最终回复：\\n{result2.final_output[:300]}…")
print()

# —- 场景 3：一般咨询（AI 可解决） —-
print("【场景 3】一般咨询 — AI 可解决")
print("-" * 50)
state3 = TicketSystemState(
user_input="你们的产品有哪些功能？价格是多少？",
)
result3 = ticket_system_workflow.invoke(state3)
print(f"意图：{result3.intent.value}")
print(f"AI 可解决：{result3.can_resolve}")
print(f"回复：\\n{result3.final_output[:300]}…")
print()

print("=" * 70)
print("智能客服工单系统演示完成！")
print("=" * 70)

11.4.5 代码逐段解析

第一部分：意图识别（classify_intent）

@task
def classify_intent(state: TicketSystemState) –> TicketSystemState:
prompt = f"""你是一个客服意图识别系统。请分析以下用户消息…

意图类型（三选一）：
– technical：技术问题
– billing：账单问题
– general：一般咨询

请只返回意图类型（一个单词），不要返回其他内容。"""
response = llm.invoke(prompt)
intent_str = response.content.strip().lower()
state.intent = intent_map.get(intent_str, IntentType.GENERAL)

意图识别是整个工作流的第一道关卡。它的输出决定了下游所有 Task 的行为。我们使用 LLM 来做意图识别，因为 LLM 能理解自然语言中的细微差异。例如，「API 调用超时」和「支付超时」虽然都包含「超时」，但一个是技术问题，一个是账单问题。

第二部分：RAG 检索（search_knowledge_base）

@task
def search_knowledge_base(state: TicketSystemState) –> TicketSystemState:
state.retrieved_docs = search_knowledge(state.user_input, top_k=3)

RAG（Retrieval-Augmented Generation，检索增强生成）是提高 AI 回答质量的关键技术。它通过从知识库中检索相关文档，将检索结果作为 LLM 的上下文，让 LLM 能基于真实、准确的信息来回答。在本教程中，我们使用简单的关键词匹配来模拟检索，生产环境中应该使用向量数据库和语义搜索。

第三部分：Human-in-the-loop（wait_for_human）

@task
def wait_for_human(state: TicketSystemState) –> TicketSystemState:
interrupt_message = {
"action": "需要人工客服介入",
"ticket_id": state.ticket.ticket_id,
...
}
human_decision = interrupt(interrupt_message) # 暂停工作流
# 工作流恢复后，human_decision 包含人类客服的回复

interrupt() 是 LangGraph 的 Human-in-the-loop 核心 API。它暂停工作流并保存检查点，向外部发送中断信号。外部调用方通过 Command(resume=…) 恢复执行。在恢复时，interrupt() 的返回值是 Command(resume=…) 中传入的值。

第四部分：多 Agent 协作（process_with_agent）

@task
def process_with_agent(state: TicketSystemState) –> TicketSystemState:
state = route_to_agent(state) # 根据意图路由到对应的 Agent
return state

route_to_agent 函数内部根据意图类型选择对应的 Agent。这是 Supervisor 模式的简化实现——不使用 langgraph-supervisor 库，而是直接在代码中实现路由逻辑。这种方式更灵活，但需要手动维护路由规则。

11.5 API 速查

11.6 单元测试

11.6.1 tests/test_workflow.py

# ============================================================
# tests/test_workflow.py — 智能客服工单系统的单元测试
# 运行方式：pytest tests/test_workflow.py -v
# 运行前请确保已设置：export OPENAI_API_KEY="your-key"
# ============================================================

import pytest # 导入 pytest 测试框架
import sys # 导入系统模块
import os # 导入操作系统模块

# 将项目根目录添加到 Python 路径，确保可以导入项目模块
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from models import TicketSystemState, IntentType, Ticket, TicketStatus # 导入数据模型
from knowledge_base import search_knowledge # 导入知识库检索函数
from main import ticket_system_workflow # 导入主工作流

# ============================================================
# 测试 1：知识库检索功能
# ============================================================
class TestKnowledgeBase:
"""知识库检索功能的单元测试"""

def test_search_technical_issue(self):
"""测试：搜索技术问题应该返回技术相关文档"""
docs = search_knowledge("API 调用超时", top_k=3)
# 验证返回了文档
assert len(docs) > 0, "应该返回至少 1 条文档"
# 验证返回的文档包含技术相关标签
tech_docs = [d for d in docs if d.category == IntentType.TECHNICAL]
assert len(tech_docs) > 0, "应该返回技术类文档"

def test_search_billing_issue(self):
"""测试：搜索账单问题应该返回账单相关文档"""
docs = search_knowledge("支付失败退款", top_k=3)
assert len(docs) > 0, "应该返回至少 1 条文档"
billing_docs = [d for d in docs if d.category == IntentType.BILLING]
assert len(billing_docs) > 0, "应该返回账单类文档"

def test_search_no_match(self):
"""测试：搜索不存在的关键词应该返回空列表"""
docs = search_knowledge("xyzabc123不存在的关键词", top_k=3)
assert len(docs) == 0, "不存在的关键词应该返回空列表"

def test_search_top_k_limit(self):
"""测试：top_k 参数应该限制返回文档数量"""
docs = search_knowledge("问题", top_k=2)
assert len(docs) <= 2, "返回的文档数量不应超过 top_k"

# ============================================================
# 测试 2：数据模型
# ============================================================
class TestModels:
"""数据模型的单元测试"""

def test_ticket_creation(self):
"""测试：创建工单对象"""
ticket = Ticket(
ticket_id="TK-20260712-ABC123",
intent=IntentType.TECHNICAL,
subject="API 超时",
description="API 调用超时",
)
assert ticket.ticket_id == "TK-20260712-ABC123"
assert ticket.intent == IntentType.TECHNICAL
assert ticket.status == TicketStatus.OPEN # 默认状态应为 OPEN
assert ticket.priority == 3 # 默认优先级应为 3

def test_ticket_status_transition(self):
"""测试：工单状态转换"""
ticket = Ticket(
ticket_id="TK-TEST-001",
intent=IntentType.BILLING,
subject="退款",
description="请退款",
)
# 初始状态
assert ticket.status == TicketStatus.OPEN
# 状态转换
ticket.status = TicketStatus.IN_PROGRESS
assert ticket.status == TicketStatus.IN_PROGRESS
ticket.status = TicketStatus.RESOLVED
assert ticket.status == TicketStatus.RESOLVED

def test_state_initialization(self):
"""测试：State 初始化"""
state = TicketSystemState(
user_input="测试问题",
)
assert state.user_input == "测试问题"
assert state.intent == IntentType.GENERAL # 默认意图
assert state.messages == [] # 默认消息列表为空
assert state.retrieved_docs == [] # 默认检索结果为空
assert state.can_resolve is True # 默认 AI 可解决
assert state.needs_human is False # 默认不需要人工
assert state.ticket is None # 默认没有工单

# ============================================================
# 测试 3：意图识别（集成测试，需要 LLM）
# ============================================================
@pytest.mark.integration # 标记为集成测试（需要 LLM API）
class TestIntentClassification:
"""意图识别功能的集成测试"""

@pytest.mark.skipif(
not os.environ.get("OPENAI_API_KEY"),
reason="需要设置 OPENAI_API_KEY 环境变量",
)
def test_technical_intent(self):
"""测试：技术问题应该被识别为 TECHNICAL"""
state = TicketSystemState(
user_input="我的 API 调用返回了 500 错误，怎么解决？",
)
result = ticket_system_workflow.invoke(state)
assert result.intent == IntentType.TECHNICAL, (
f"预期意图为 TECHNICAL，实际为 {result.intent}"
)

@pytest.mark.skipif(
not os.environ.get("OPENAI_API_KEY"),
reason="需要设置 OPENAI_API_KEY 环境变量",
)
def test_billing_intent(self):
"""测试：账单问题应该被识别为 BILLING"""
state = TicketSystemState(
user_input="我的订单扣款了但是没有成功，请帮我退款",
)
result = ticket_system_workflow.invoke(state)
assert result.intent == IntentType.BILLING, (
f"预期意图为 BILLING，实际为 {result.intent}"
)

@pytest.mark.skipif(
not os.environ.get("OPENAI_API_KEY"),
reason="需要设置 OPENAI_API_KEY 环境变量",
)
def test_general_intent(self):
"""测试：一般咨询应该被识别为 GENERAL"""
state = TicketSystemState(
user_input="你们的产品有哪些功能？",
)
result = ticket_system_workflow.invoke(state)
assert result.intent == IntentType.GENERAL, (
f"预期意图为 GENERAL，实际为 {result.intent}"
)

# ============================================================
# 运行测试的说明
# ============================================================
if __name__ == "__main__":
# 运行所有测试
# pytest tests/test_workflow.py -v
# 只运行单元测试（不依赖 LLM）：
# pytest tests/test_workflow.py -v -m "not integration"
# 运行集成测试（需要 LLM）：
# pytest tests/test_workflow.py -v -m integration
print("请使用 pytest 运行测试：")
print(" 单元测试：pytest tests/test_workflow.py -v -m 'not integration'")
print(" 集成测试：pytest tests/test_workflow.py -v -m integration")
print(" 全部测试：pytest tests/test_workflow.py -v")

11.6.2 requirements.txt

# ============================================================
# requirements.txt — 智能客服工单系统依赖清单
# 安装：pip install -r requirements.txt
# ============================================================

# LangGraph 核心
langgraph>=1.2.9

# LangChain 集成
langchain>=0.3.0
langchain-openai>=0.3.0
langchain-core>=0.3.0

# 数据模型
pydantic>=2.0

# Web 服务（API 部署）
fastapi>=0.115.0
uvicorn>=0.30.0

# 测试
pytest>=8.0
pytest-asyncio>=0.24.0

# 环境变量管理
python-dotenv>=1.0.0

11.6.3 Dockerfile

# ============================================================
# Dockerfile — 智能客服工单系统 Docker 镜像
# 构建：docker build -t smart-ticket-system .
# 运行：docker run -p 8000:8000 -e OPENAI_API_KEY=your-key smart-ticket-system
# ============================================================

# 使用 Python 3.11 官方镜像（slim 版本更小）
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 设置环境变量
# 不生成 .pyc 文件；不缓冲输出（日志实时显示）；设置 Python 路径
ENV PYTHONDONTWRITEBYTECODE=1 \\
PYTHONUNBUFFERED=1 \\
PYTHONPATH=/app

# 安装系统依赖
RUN apt-get update && apt-get install -y –no-install-recommends \\
curl \\ # 健康检查工具
&& rm -rf /var/lib/apt/lists/* # 清理 apt 缓存

# 复制依赖文件并安装 Python 依赖
COPY requirements.txt .
RUN pip install –no-cache-dir -r requirements.txt

# 复制项目代码
COPY . .

# 创建非 root 用户运行应用（安全最佳实践）
RUN useradd –create-home –shell /bin/bash appuser
RUN chown -R appuser:appuser /app
USER appuser

# 暴露端口
EXPOSE 8000

# 健康检查
HEALTHCHECK –interval=30s –timeout=10s –start-period=5s –retries=3 \\
CMD curl -f http://localhost:8000/health || exit 1

# 启动命令
CMD ["uvicorn", "api:app", "–host", "0.0.0.0", "–port", "8000"]

11.6.4 docker-compose.yml

# ============================================================
# docker-compose.yml — 多服务编排配置
# 启动：docker-compose up -d
# 停止：docker-compose down
# ============================================================

version: '3.8'

services:
# 主应用服务
app:
build: .
container_name: smart–ticket–system
ports:
– "8000:8000" # 将容器的 8000 端口映射到宿主机的 8000 端口
environment:
– OPENAI_API_KEY=${OPENAI_API_KEY} # 从宿主机环境变量读取
– LOG_LEVEL=INFO # 日志级别
volumes:
– ./data:/app/data # 持久化数据目录（检查点、工单数据等）
restart: unless–stopped # 除非手动停止，否则自动重启
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3

11.7 常见错误与避坑指南

错误 1：忘记在 Entrypoint 配置 checkpointer 就使用 interrupt()

错误代码：

# 错误：没有配置 checkpointer
@entrypoint() # 缺少 checkpointer=checkpointer
def my_workflow(state):
...
result = wait_for_human(state).result() # wait_for_human 内部有 interrupt()
# 运行时会报错，因为 interrupt() 需要检查点支持

正确代码：

checkpointer = MemorySaver() # 或 SqliteSaver

@entrypoint(checkpointer=checkpointer) # 必须配置 checkpointer
def my_workflow(state):
...
result = wait_for_human(state).result()

原因：interrupt() 需要检查点来保存和恢复执行状态。如果 entrypoint 没有配置 checkpointer，调用 interrupt() 会抛出错误。

错误 2：恢复执行时使用了错误的 thread_id

错误代码：

# 第一次调用
config1 = {"configurable": {"thread_id": "session-abc"}}
try:
result = workflow.invoke(state, config=config1)
except GraphInterrupt:
pass

# 错误：恢复时使用了不同的 thread_id
config2 = {"configurable": {"thread_id": "session-xyz"}} # 不同的 thread_id！
resume_cmd = Command(resume=human_response)
result = workflow.invoke(resume_cmd, config=config2) # 找不到中断的检查点！

正确代码：

# 第一次调用
config = {"configurable": {"thread_id": "session-abc"}}
try:
result = workflow.invoke(state, config=config)
except GraphInterrupt:
pass

# 正确：恢复时使用相同的 config
resume_cmd = Command(resume=human_response)
result = workflow.invoke(resume_cmd, config=config) # 相同的 config！

原因：LangGraph 通过 thread_id 定位中断的检查点。恢复时必须使用相同的 thread_id，否则 LangGraph 找不到中断的会话。

错误 3：RAG 检索时没有考虑意图分类，检索结果不准确

错误代码：

# 错误：只根据用户输入检索，没有利用意图分类结果
# 可能导致「API 超时」检索到「支付超时」的文档
@task
def search_knowledge_base(state):
# 没有传入意图分类，检索结果可能包含不相关的文档
state.retrieved_docs = search_knowledge(state.user_input)
return state

正确代码：

# 正确：结合意图分类进行检索，提高准确率
@task
def search_knowledge_base(state):
# 将意图作为额外的过滤条件
query = f"{state.intent.value} {state.user_input}" # 加入意图作为上下文
docs = search_knowledge(query, top_k=3)
# 进一步过滤：只保留与意图类型匹配的文档
filtered_docs = [
d for d in docs
if d.category == state.intent or d.category == IntentType.GENERAL
]
state.retrieved_docs = filtered_docs if filtered_docs else docs[:2] # 有兜底
return state

原因：意图识别已经告诉我们用户问题的大致类型。利用这个信息进行检索过滤，可以显著提高检索结果的相关性，避免「技术问题的用户看到账单文档」。

错误 4：工单创建后没有更新状态，导致工单一直处于 OPEN 状态

错误代码：

# 错误：创建工单后忘记更新状态
@task
def create_ticket(state):
ticket = Ticket(
ticket_id=f"TK-{today}–{uuid.uuid4().hex[:6]}",
status=TicketStatus.OPEN, # 创建时是 OPEN
...
)
state.ticket = ticket
# 忘记更新状态！工单永远是 OPEN
return state

正确代码：

# 正确：在工单生命周期的关键节点更新状态
@task
def create_ticket(state):
ticket = Ticket(
ticket_id=f"TK-{today}–{uuid.uuid4().hex[:6]}",
status=TicketStatus.OPEN,
...
)
state.ticket = ticket
return state

@task
def wait_for_human(state):
human_decision = interrupt(...)
# 正确：人工介入后更新工单状态
if state.ticket:
state.ticket.status = TicketStatus.IN_PROGRESS # 更新为处理中
return state

@task
def resolve_ticket(state):
# 正确：问题解决后更新工单状态
if state.ticket:
state.ticket.status = TicketStatus.RESOLVED # 更新为已解决
return state

原因：工单状态是客服系统的重要指标。如果状态不更新，管理者和用户都无法知道工单的进展。在工单生命周期的关键节点（创建、处理中、等待人工、已解决、已关闭）更新状态。

错误 5：没有对用户输入做验证，导致恶意输入或异常输入

错误代码：

# 错误：直接使用用户输入，没有验证
@entrypoint(checkpointer=checkpointer)
def workflow(state: TicketSystemState):
# 如果用户输入为空、超长或包含恶意内容，直接传给 LLM
state = classify_intent(state).result() # 可能传入空字符串或超长文本
return state

正确代码：

# 正确：在入口处验证用户输入
MAX_INPUT_LENGTH = 2000 # 最大输入长度
MIN_INPUT_LENGTH = 2 # 最小输入长度

@entrypoint(checkpointer=checkpointer)
def workflow(state: TicketSystemState):
# 输入验证
user_input = state.user_input.strip()

# 检查是否为空
if not user_input:
state.final_output = "请输入您的问题，我们很乐意帮助您。"
return state

# 检查长度
if len(user_input) > MAX_INPUT_LENGTH:
state.final_output = f"您的问题描述过长（当前 {len(user_input)} 字符），请精简到 {MAX_INPUT_LENGTH} 字符以内。"
return state

if len(user_input) < MIN_INPUT_LENGTH:
state.final_output = "请提供更多细节，以便我们更好地帮助您。"
return state

# 更新 state 中的用户输入为清理后的版本
state.user_input = user_input
# 继续正常流程
state = classify_intent(state).result()
return state

原因：用户输入是外部数据，不可信任。在传给 LLM 之前，需要验证长度、格式和内容。这可以防止 LLM 被恶意输入利用（prompt injection），也能避免不必要的 API 调用成本。

11.8 最佳实践总结

结语：从零到 Hero 的旅程

恭喜你完成了全部 11 章的学习！让我们回顾一下这段旅程：

从「什么是 LangGraph」到「构建完整的智能客服工单系统」，你已经掌握了使用 LangGraph Functional API 构建生产级 AI Agent 应用的全部核心技能。

下一步建议：

祝你构建出优秀的 AI Agent 应用！