LangGraph实战：用Python构建一个带状态管理的智能客服工作流

LangGraph实战：用Python构建一个带状态管理的智能客服工作流

想象一下这样的场景：用户打开客服对话框，输入"我的订单怎么还没到？"，系统不仅能理解这句话的字面意思，还能记住之前的对话上下文，自动查询订单状态，并根据物流情况给出个性化的回复。这种流畅的多轮对话体验背后，正是LangGraph的状态管理和工作流引擎在发挥作用。

作为Python开发者，你可能已经熟悉了LangChain的基础组件，但当业务逻辑变得复杂时，单纯链式调用的局限性就显现出来了。LangGraph提供的图结构工作流和状态管理能力，让我们能够构建真正具备上下文感知能力的智能客服系统。本文将从一个具体的订单查询场景出发，手把手带你实现一个完整的解决方案。

1. 环境准备与基础架构

在开始编码前，我们需要明确几个核心概念。LangGraph的工作流由节点(Node)和边(Edge)组成，节点负责具体业务逻辑处理，边定义执行路径。状态(State)则贯穿整个工作流生命周期，保存对话上下文和中间结果。

首先安装必要的依赖：

pip install langgraph langchain-openai

基础架构代码框架如下：

from typing import TypedDict, List from langgraph.graph import StateGraph, END # 定义状态结构 class AgentState(TypedDict): user_input: str intent: str db_result: dict response: str chat_history: List[dict] # 初始化工作流 workflow = StateGraph(AgentState)

这个基础框架定义了一个包含用户输入、意图识别结果、数据库查询结果、响应内容和对话历史的状态容器。StateGraph是LangGraph的核心类，负责管理工作流的执行。

2. 构建智能客服的核心节点

2.1 意图识别节点

意图识别是对话系统的第一道关卡。我们使用LLM来分析用户输入：

from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-3.5-turbo") def intent_detection(state: AgentState): prompt = ChatPromptTemplate.from_template(""" 你是一个专业的意图分类器。请分析以下用户输入，判断其意图： - ORDER_STATUS: 查询订单状态 - REFUND: 退款申请 - COMPLAINT: 投诉建议 - OTHER: 其他 用户输入: {input} 只返回意图类型，不要解释。 """) chain = prompt | llm intent = chain.invoke({"input": state["user_input"]}).content return {"intent": intent} # 将节点加入工作流 workflow.add_node("intent_detection", intent_detection)

这个节点使用了ChatGPT作为分类器，将用户输入映射到预定义的意图类别。在实际应用中，你可能需要：

• 更细致的意图分类体系
• 支持多语言处理
• 添加异常处理逻辑

2.2 数据库查询节点

根据识别出的意图，我们需要从业务系统获取相关数据：

def query_database(state: AgentState): # 模拟数据库查询 - 实际项目中替换为真实数据源 if state["intent"] == "ORDER_STATUS": return {"db_result": { "order_id": "12345", "status": "shipped", "estimated_delivery": "2024-05-20" }} elif state["intent"] == "REFUND": return {"db_result": { "eligible": True, "process_days": 7 }} return {"db_result": {}} workflow.add_node("db_query", query_database)

提示：生产环境中应该使用LangChain的工具集成功能连接真实数据库，并添加适当的错误处理和超时机制。

2.3 响应生成节点

最后，我们需要将原始数据转化为自然语言回复：

def generate_response(state: AgentState): prompt = ChatPromptTemplate.from_template(""" 你是一个专业的客服代表。根据以下信息生成友好、专业的回复： 用户问题: {user_input} 意图: {intent} 查询结果: {db_result} 历史对话: {history} 回复要求: - 使用用户的语言(中文/英文) - 保持专业但友好的语气 - 不超过100字 """) chain = prompt | llm response = chain.invoke({ "user_input": state["user_input"], "intent": state["intent"], "db_result": state["db_result"], "history": state["chat_history"][-3:] # 取最近3条历史 }).content return {"response": response, "chat_history": state["chat_history"] + [ {"role": "user", "content": state["user_input"]}, {"role": "assistant", "content": response} ]} workflow.add_node("generate_response", generate_response)

这个节点不仅生成当前回复，还更新了对话历史，为后续的多轮对话提供上下文。

3. 设计工作流路由逻辑

有了各个功能节点后，我们需要定义它们之间的执行顺序和条件分支：

# 设置入口点 workflow.set_entry_point("intent_detection") # 定义节点连接 workflow.add_edge("intent_detection", "db_query") workflow.add_edge("db_query", "generate_response") workflow.add_edge("generate_response", END) # 编译工作流 app = workflow.compile()

这个基本流程是线性的：意图识别 → 数据库查询 → 回复生成。但实际业务中，我们可能需要更复杂的路由：

# 添加条件分支示例 def should_check_details(state: AgentState): if "details" in state["user_input"].lower(): return True return False workflow.add_conditional_edges( "intent_detection", should_check_details, { True: "check_details_node", False: "db_query" } )

条件路由让工作流能够根据对话内容动态调整执行路径，实现更智能的交互。

4. 完整工作流测试与优化

现在我们可以测试整个工作流了：

# 初始化状态 initial_state = { "user_input": "我的订单12345到哪了？", "chat_history": [] } # 执行工作流 for step in app.stream(initial_state): print(f"== 步骤 {step['step']} ==") print(f"节点: {step['node']}") print(f"状态: {step['keys']}") print("---")

执行结果会显示每个节点的处理过程和状态变化。为了提升系统性能，我们可以考虑以下优化策略：

实际部署时还需要考虑：

• 错误处理和重试机制
• 对话超时和会话管理
• 性能监控和日志记录

5. 扩展多轮对话能力

真正的智能客服需要记住上下文。让我们增强状态管理：

class EnhancedState(TypedDict): user_input: str intent: str confirmed_intent: bool # 是否确认过意图 slots: dict # 提取的实体信息 db_result: dict response: str chat_history: List[dict] retry_count: int # 重试次数 def confirm_intent(state: EnhancedState): if state["confirmed_intent"]: return {"response": "继续处理您的请求..."} prompt = """请确认是否理解正确： 用户意图: {intent} 提取信息: {slots} 请回答是/否""" # ... 实现确认逻辑

这种增强后的状态管理可以支持更复杂的对话场景，比如：

1. 信息确认："您是要查询订单12345对吗？"
2. 槽位填充："请问您的订单号是多少？"
3. 流程切换："您想改为申请退款吗？"

6. 生产环境部署建议

将开发好的工作流部署到生产环境需要考虑以下几个关键因素：

性能优化

• 使用异步版本的工作流(AsyncStateGraph)
• 对LLM调用实现批处理
• 设置合理的超时时间

可观测性

• 集成LangSmith进行执行追踪
• 添加自定义监控指标
• 记录完整的对话日志

安全考虑

• 对用户输入进行过滤和清理
• 敏感数据脱敏处理
• 实现访问控制和限流

一个简单的FastAPI集成示例：

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class ChatRequest(BaseModel): message: str session_id: str @app.post("/chat") async def handle_chat(request: ChatRequest): # 从缓存获取或初始化状态 state = cache.get(request.session_id) or initial_state # 更新用户输入 state["user_input"] = request.message # 执行工作流 async for step in app.astream(state): # 处理每一步... pass # 保存状态 cache.set(request.session_id, state) return {"response": state["response"]}

在实际项目中，我们还需要处理会话超时、状态持久化、多租户隔离等复杂需求。