LangGraph教程（四）：智能客服全流程实战指南

2025年11月26日 互联网

LangGraph教程（四）：智能客服项目示例（流程+完整代码）

一、项目背景与核心价值

在AI技术深度渗透的当下，智能客服已成为企业降本增效的关键工具。传统客服系统存在三大痛点：意图识别不精准导致无效对话、多轮对话管理混乱影响用户体验、知识库更新滞后无法应对新问题。本教程基于LangGraph框架，通过状态机模式构建可扩展的智能客服系统，重点解决以下技术挑战：

  • 动态对话流程控制：支持分支跳转、循环提问等复杂场景
  • 多工具协同调用：集成检索增强生成（RAG）、函数调用等能力
  • 可观测性设计：实时监控对话状态与模型调用情况

二、系统架构设计

2.1 技术栈选型

组件 技术选型 核心优势
流程编排 LangGraph 0.2.0 基于状态机的显式流程控制
大模型 GPT-4o/Qwen2.5 强语义理解与多轮对话能力
检索引擎 ChromaDB 语义向量检索与混合查询
监控系统 Prometheus+Grafana 实时指标可视化与告警

2.2 核心流程设计

  1. graph TD
  2.     A[用户输入] --> B{意图分类}
  3.     B -->|查询类| C[知识库检索]
  4.     B -->|任务类| D[函数调用]
  5.     B -->|闲聊类| E[生成式回复]
  6.     C --> F{检索置信度}
  7.     F -->|高| G[返回结果]
  8.     F -->|低| H[转人工]
  9.     D --> I[执行结果校验]
  10.     I -->|成功| J[返回数据]
  11.     I -->|失败| K[错误处理]

三、LangGraph实现详解

3.1 状态机定义

  1. from langgraph.prebuilt import StateGraph
  2. from langgraph.graph import END
  4. class CustomerServiceGraph(StateGraph):
  5.     def __init__(self, llm, vector_store):
  6.         super().__init__()
  7.         self.llm = llm
  8.         self.vector_store = vector_store
  9.         self._build_graph()
  11.     def _build_graph(self):
  12.         # 初始状态
  13.         with self.state("start") as s:
  14.             s.input("user_query")
  15.             s.output("intent")
  16.             s.node(self.classify_intent)
  17.             s.to("route")
  19.         # 路由分支
  20.         with self.state("route") as s:
  21.             s.input("intent")
  22.             s.output("action")
  23.             s.condition(
  24.                 lambda state: state.intent == "query",
  25.                 "knowledge_search",
  26.                 condition2=lambda state: state.intent == "task",
  27.                 to2="function_call",
  28.                 default="generative_reply"
  29.             )
  31.         # 查询处理分支
  32.         with self.state("knowledge_search") as s:
  33.             s.input("user_query")
  34.             s.output("answer")
  35.             s.node(self.search_knowledge)
  36.             s.condition(
  37.                 lambda state: state.search_score > 0.8,
  38.                 "return_answer",
  39.                 default="escalate_human"
  40.             )
  42.         # 终止状态
  43.         self.add_state("return_answer", END)
  44.         self.add_state("escalate_human", END)

3.2 关键节点实现

意图分类节点

  1. from langchain_community.llms import OpenAI
  2. from langchain_core.prompts import ChatPromptTemplate
  4. class IntentClassifier:
  5.     def __init__(self, llm):
  6.         self.llm = llm
  7.         self.prompt = ChatPromptTemplate.from_template("""
  8.             用户问题: {query}
  9.             请从以下类别中选择最匹配的意图：
  10.             1. 查询类（产品信息、政策等）
  11.             2. 任务类（下单、退款等）
  12.             3. 闲聊类
  13.             直接返回分类编号
  14.         """)
  16.     def __call__(self, state):
  17.         response = self.llm(self.prompt.format(query=state.user_query))
  18.         intent_map = {"1": "query", "2": "task", "3": "chat"}
  19.         state.intent = intent_map.get(response.content.strip(), "chat")

知识检索节点

  1. from langchain_community.vectorstores import Chroma
  2. from langchain_core.retrievers import ContextualCompressionRetriever
  4. class KnowledgeSearcher:
  5.     def __init__(self, vector_store):
  6.         self.vector_store = vector_store
  7.         self.retriever = ContextualCompressionRetriever(
  8.             base_compressor=...,  # 压缩器配置
  9.             base_retriever=vector_store.as_retriever()
  10.         )
  12.     def __call__(self, state):
  13.         docs = self.retriever.get_relevant_documents(state.user_query)
  14.         state.search_results = docs
  15.         state.search_score = max(doc.metadata["score"] for doc in docs) if docs else 0
  16.         state.answer = "\n".join(doc.page_content for doc in docs[:3])

四、完整运行示例

4.1 环境准备

  1. pip install langgraph langchain chromadb openai
  2. export OPENAI_API_KEY="your-api-key"

4.2 主程序实现

  1. from langgraph.graph import GraphApplication
  2. from langchain_community.llms import OpenAI
  3. from chromadb.config import Settings
  4. from chromadb.utils import embedding_functions
  6. # 初始化组件
  7. llm = OpenAI(temperature=0)
  8. vector_store = Chroma(
  9.     embedding_function=embedding_functions.OpenAIEmbeddingFunction(
  10.         api_key=os.getenv("OPENAI_API_KEY")
  11.     ),
  12.     persist_directory="./db"
  13. )
  15. # 构建状态机
  16. graph = CustomerServiceGraph(llm, vector_store)
  17. app = GraphApplication.from_graph(graph)
  19. # 运行示例
  20. state = {
  21.     "user_query": "如何申请退货？"
  22. }
  23. result = app.invoke(state)
  24. print("系统回复:", result.get("answer") or "已转人工处理")

五、优化与扩展建议

5.1 性能优化方案

  1. 缓存层设计：对高频查询结果进行缓存
    ```python
    from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_search(query):
return vector_store.similarity_search(query, k=3)

  2. 2. **异步处理**：使用LangGraph的异步模式处理耗时操作
  3. ```python
  4. async def async_function_call(state):
  5.     # 实现异步API调用
  6.     pass

5.2 功能扩展方向

  1. 多语言支持：集成翻译模型实现语言自动检测
  2. 情绪分析：在对话中实时检测用户情绪并调整回复策略
  3. 主动学习：收集低置信度查询用于模型微调

六、常见问题解决方案

6.1 模型幻觉问题

现象：生成错误信息或虚构数据
解决方案：

  • 启用检索增强生成（RAG）
  • 设置回复置信度阈值 
    1. def validate_response(response):
    2.   if "不确定" in response.lower():
    3.       raise ValueError("低置信度回复")

6.2 对话中断处理

现象：用户长时间无响应导致流程卡死
解决方案：

  • 添加超时控制机制
    ```python
    import asyncio

async def with_timeout(node, state, timeout=30):
try:
return await asyncio.wait_for(node(state), timeout)
except asyncio.TimeoutError:
state.error = “用户响应超时”
return state

  2. ## 七、部署与监控
  3. ### 7.1 Docker化部署
  4. ```dockerfile
  5. FROM python:3.11-slim
  6. WORKDIR /app
  7. COPY requirements.txt .
  8. RUN pip install -r requirements.txt
  9. COPY . .
  10. CMD ["python", "main.py"]

7.2 监控指标设计

指标名称 计算方式 告警阈值
对话成功率 成功完成对话数/总对话数 <85%
平均响应时间 P90响应时长 >3s
模型调用次数 单位时间API调用量 >1000/min

本教程完整实现了基于LangGraph的智能客服系统，覆盖从流程设计到生产部署的全链路。开发者可通过调整状态机配置快速适配不同业务场景，建议结合实际需求逐步添加监控告警、A/B测试等企业级功能。

最新文章