LangGraph教程（四）：智能客服系统实战指南（流程+代码）

2025年12月29日 互联网

LangGraph教程（四）：智能客服系统实战指南（流程+代码）

智能客服系统是LangGraph框架的典型应用场景，其多轮对话、状态管理和工具调用能力能够有效解决传统客服系统的流程僵化问题。本文将通过一个完整的智能客服案例，从流程设计到代码实现进行系统讲解。

一、系统架构设计

1.1 核心功能模块

智能客服系统包含三大核心模块：

  • 意图识别模块：通过NLP模型理解用户输入
  • 对话管理模块：维护对话状态并控制流程
  • 工具集成模块：调用知识库、订单系统等外部服务
  1. graph TD
  2.     A[用户输入] --> B[意图识别]
  3.     B --> C{意图类型}
  4.     C -->|查询类| D[调用知识库]
  5.     C -->|操作类| E[调用业务系统]
  6.     C -->|闲聊类| F[生成回复]
  7.     D --> G[结果包装]
  8.     E --> G
  9.     F --> G
  10.     G --> H[输出回复]

1.2 LangGraph适配方案

采用状态机模式实现对话控制，每个节点代表一个对话状态：

  • 初始状态：欢迎语和主菜单
  • 查询状态：处理知识库查询
  • 操作状态：处理业务操作
  • 结束状态：对话总结

二、完整代码实现

2.1 环境准备

  1. # 安装必要依赖
  2. pip install langgraph langchain-community pydantic

2.2 核心组件定义

  1. from langgraph.prebuilt import StateGraph
  2. from pydantic import BaseModel
  3. from typing import Optional
  5. class DialogueState(BaseModel):
  6.     current_step: str = "welcome"
  7.     query_context: Optional[str] = None
  8.     session_id: str = "default"
  10. class ToolResult(BaseModel):
  11.     success: bool
  12.     message: str
  13.     data: Optional[dict] = None

2.3 工具集成实现

  1. class KnowledgeBaseTool:
  2.     def query(self, query: str) -> ToolResult:
  3.         # 模拟知识库查询
  4.         if "退款" in query:
  5.             return ToolResult(
  6.                 success=True,
  7.                 message="退款政策查询成功",
  8.                 data={"policy": "支持7天无理由退款"}
  9.             )
  10.         return ToolResult(
  11.             success=False,
  12.             message="未找到相关政策"
  13.         )
  15. class OrderSystemTool:
  16.     def check_status(self, order_id: str) -> ToolResult:
  17.         # 模拟订单查询
  18.         return ToolResult(
  19.             success=True,
  20.             message="订单状态查询成功",
  21.             data={"status": "已发货", "tracking": "123456"}
  22.         )

2.4 状态机配置

  1. from langgraph.prebuilt import create_state_graph
  3. app = create_state_graph(DialogueState)
  5. # 欢迎节点
  6. @app.node("welcome")
  7. def welcome_node(state: DialogueState):
  8.     return {"reply": "您好，欢迎使用智能客服，请问需要什么帮助？"}
  10. # 意图分类节点
  11. @app.node("intent_classification")
  12. def classify_intent(state: DialogueState, input: str):
  13.     if "政策" in input or "规则" in input:
  14.         return {"next_step": "knowledge_query"}
  15.     elif "订单" in input or "发货" in input:
  16.         return {"next_step": "order_query"}
  17.     else:
  18.         return {"next_step": "fallback"}
  20. # 知识查询流程
  21. @app.node("knowledge_query")
  22. def handle_knowledge(state: DialogueState, input: str):
  23.     kb = KnowledgeBaseTool()
  24.     result = kb.query(input)
  25.     if result.success:
  26.         return {"reply": result.message + f"\n详情：{result.data['policy']}"}
  27.     return {"reply": result.message}
  29. # 订单查询流程
  30. @app.node("order_query")
  31. def handle_order(state: DialogueState, input: str):
  32.     # 简单提取订单号（实际应使用正则）
  33.     order_id = input.split("订单")[1].strip() if "订单" in input else "1001"
  34.     os = OrderSystemTool()
  35.     result = os.check_status(order_id)
  36.     return {
  37.         "reply": f"{result.message}\n状态：{result.data['status']}\n运单号：{result.data['tracking']}"
  38.     }
  40. # 配置状态转移
  41. app.add_edge("welcome", "intent_classification")
  42. app.add_edge("intent_classification", "knowledge_query", condition=lambda x: x["next_step"] == "knowledge_query")
  43. app.add_edge("intent_classification", "order_query", condition=lambda x: x["next_step"] == "order_query")
  44. app.add_edge("intent_classification", "fallback", condition=lambda x: x["next_step"] == "fallback")

2.5 完整运行示例

  1. def run_dialogue():
  2.     # 初始化状态机
  3.     state = DialogueState()
  5.     # 模拟对话轮次
  6.     dialogues = [
  7.         "你好",
  8.         "我想了解退款政策",
  9.         "我的订单123456发货了吗？",
  10.         "谢谢"
  11.     ]
  13.     current_state = state
  14.     for input_text in dialogues:
  15.         if current_state.current_step == "welcome":
  16.             output = app.step(current_state, input=input_text)["welcome"]
  17.         else:
  18.             # 实际应通过NLP模型获取意图，这里简化处理
  19.             intent_output = app.step(current_state, input=input_text)["intent_classification"]
  20.             next_step = intent_output["next_step"]
  21.             output = app.step(current_state, input=input_text)[next_step]
  23.         print(f"用户: {input_text}")
  24.         print(f"系统: {output['reply']}\n")
  26.         # 更新状态（实际应根据工具调用结果更新）
  27.         current_state = DialogueState(
  28.             current_step="welcome",  # 简化处理，实际应动态更新
  29.             session_id=current_state.session_id
  30.         )
  32. if __name__ == "__main__":
  33.     run_dialogue()

三、关键实现细节

3.1 状态持久化方案

建议使用Redis等外部存储实现跨会话状态管理：

  1. import redis
  3. class StateManager:
  4.     def __init__(self):
  5.         self.r = redis.Redis(host='localhost', port=6379, db=0)
  7.     def save_state(self, session_id: str, state: DialogueState):
  8.         self.r.set(f"dialogue:{session_id}", state.json())
  10.     def load_state(self, session_id: str) -> Optional[DialogueState]:
  11.         data = self.r.get(f"dialogue:{session_id}")
  12.         return DialogueState.parse_raw(data) if data else None

3.2 异常处理机制

  1. @app.node("error_handler")
  2. def handle_error(state: DialogueState, error: Exception):
  3.     return {
  4.         "reply": f"系统处理出错：{str(error)}",
  5.         "next_step": "welcome"  # 错误后重置对话
  6.     }
  8. # 在状态转移中添加错误边
  9. app.add_error_edge("knowledge_query", "error_handler")

四、性能优化建议

  1. 工具调用缓存：对频繁查询的知识库内容实施缓存
    ```python
    from functools import lru_cache

class CachedKnowledgeBase(KnowledgeBaseTool):
@lru_cache(maxsize=128)
def query(self, query: str) -> ToolResult:
return super().query(query)

  2. 2. **异步工具调用**：对耗时操作使用异步调用
  3. ```python
  4. import asyncio
  6. async def async_order_query(order_id: str):
  7.     # 模拟异步IO
  8.     await asyncio.sleep(0.5)
  9.     return OrderSystemTool().check_status(order_id)
  1. 状态压缩：对大型状态对象实施序列化优化
    ```python
    import orjson

def compress_state(state: DialogueState) -> bytes:
return orjson.dumps(state.dict())

def decompress_state(data: bytes) -> DialogueState:
return DialogueState.parse_raw(data)

  2. ## 五、扩展性设计
  4. ### 5.1 插件式工具架构
  5. ```python
  6. from abc import ABC, abstractmethod
  8. class ToolBase(ABC):
  9.     @abstractmethod
  10.     def execute(self, input: str) -> ToolResult:
  11.         pass
  13. class ToolRegistry:
  14.     def __init__(self):
  15.         self.tools = {}
  17.     def register(self, name: str, tool: ToolBase):
  18.         self.tools[name] = tool
  20.     def execute(self, name: str, input: str) -> ToolResult:
  21.         return self.tools[name].execute(input)

5.2 多轮对话记忆

  1. class DialogueMemory:
  2.     def __init__(self):
  3.         self.history = []
  5.     def add_turn(self, user_input: str, system_reply: str):
  6.         self.history.append({
  7.             "role": "user",
  8.             "content": user_input
  9.         })
  10.         self.history.append({
  11.             "role": "system",
  12.             "content": system_reply
  13.         })
  15.     def get_context(self, window_size: int = 3):
  16.         return self.history[-window_size*2:]

六、部署最佳实践

  1. 容器化部署：

    1. FROM python:3.9-slim
    2. WORKDIR /app
    3. COPY requirements.txt .
    4. RUN pip install -r requirements.txt
    5. COPY . .
    6. CMD ["python", "app.py"]

  2. 水平扩展方案：

  • 使用消息队列（如Kafka）解耦对话处理
  • 状态存储使用分布式缓存
  • 无状态服务实例可随意扩展
  1. 监控指标：
  • 对话完成率
  • 平均处理时长
  • 工具调用成功率
  • 状态转移错误率

本文提供的完整实现展示了如何使用LangGraph构建企业级智能客服系统。实际开发中，建议从简单流程开始，逐步添加复杂功能，并通过AB测试验证不同对话策略的效果。对于生产环境，还需考虑安全审计、多语言支持等高级特性。

最新文章