LangGraph实战：构建具备状态与决策能力的智能体工作流

1. LangGraph是什么能解决什么问题第一次接触LangGraph时我也被这个名词唬住了。后来在实际项目中用它搭建客服系统才发现它其实就是个智能乐高——通过拼接不同功能模块节点和连接规则边就能组装出能自主决策的AI工作流。举个生活中的例子就像快递分拣系统包裹状态经过扫描节点后根据目的地条件自动分流到不同传送带边。LangGraph最核心的能力是状态管理和动态路由。传统编程中要实现一个多步骤流程往往需要写大量if-else嵌套。而用LangGraph你可以定义全局状态对象比如包含用户问题、历史对话、临时数据的字典创建独立的功能节点如意图识别、数据库查询、人工转接用条件边决定什么情况下该走哪条路最近帮某电商客户实现的工单系统就是个典型场景。当用户反馈订单没收到时系统会自动提取订单号状态更新查询物流信息调用外部API根据结果决定是自动退款节点A还是转人工节点B 整个过程完全自动化还能保留完整上下文。2. 从零搭建客服工单工作流2.1 定义你的状态机状态State是LangGraph的灵魂相当于工作流的记忆中枢。建议从最简单的结构开始from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class CustomerServiceState(TypedDict): # 对话历史 messages: Annotated[list, add_messages] # 工单元数据 ticket_meta: dict # 临时存储查询结果 temp_data: dict实际项目中我发现三个设计技巧分层存储像ticket_meta这种核心数据要和临时缓存temp_data分开类型标注用Annotated声明特殊处理逻辑如自动合并消息最小化原则避免在状态里存大文件等非序列化数据2.2 构建核心功能节点节点就是你的功能积木块。以工单分类节点为例def intent_classification(state: CustomerServiceState): from langchain_core.prompts import ChatPromptTemplate # 用对话历史判断意图 prompt ChatPromptTemplate.from_messages([ (system, 你是一名专业客服请分类用户问题\n 1. 物流问题\n2. 产品质量\n3. 退款申请\n4. 其他), (human, {input}) ]) chain prompt | chat_model # chat_model是预先初始化的模型 response chain.invoke({input: state[messages][-1][content]}) # 更新状态 return { ticket_meta: { category: response.content, priority: high if 退款 in response.content else medium } }踩坑提醒节点函数应该像纯函数一样工作——只依赖输入state输出state更新。我曾犯过的错误是在节点内修改全局变量导致工作流不可预测。2.3 设计决策逻辑边条件边让工作流会思考。比如根据分类结果路由from langgraph.graph import StateGraph builder StateGraph(CustomerServiceState) # 添加节点 builder.add_node(classify, intent_classification) builder.add_node(handle_logistics, logistics_handler) builder.add_node(handle_refund, refund_handler) # 设置路由规则 def route_based_on_category(state): category state[ticket_meta][category] if 物流 in category: return handle_logistics elif 退款 in category: return handle_refund else: return human_escalation builder.add_conditional_edges( classify, route_based_on_category )实测发现条件判断最好用明确的字符串匹配如 物流问题不要依赖LLM的模糊判断否则可能产生蝴蝶效应。3. 高级技巧状态持久化与人工接管3.1 实现长期记忆生产环境必须考虑中断恢复。LangGraph的检查点机制可以完美解决from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(:memory:) # 可用真实数据库路径 builder StateGraph( CustomerServiceState, checkpointermemory ) # 运行时需要指定thread_id config {configurable: {thread_id: ticket_123}} graph builder.compile() for event in graph.stream( {messages: [{role: user, content: 我的包裹丢了}]}, config ): print(event)我在银行项目中使用PostgreSQL做持久层时发现三个优化点为高频访问的state字段建立索引设置自动清理过期检查点的定时任务对敏感数据加密存储3.2 人工接管流程当AI无法处理时需要平滑转人工。这需要两个特殊设计暂停机制在特定节点等待人工输入from langgraph.graph import Pause def human_intervention(state): return Pause(wait_for_agent) builder.add_node(human_escalation, human_intervention)恢复逻辑人工处理后继续流程# 人工处理完成后 graph.stream( Command(resume包裹已补发运单号SF12345), {configurable: {thread_id: ticket_123}} )某次线上事故教会我一定要设置超时自动提醒避免工单卡在等待状态。4. 调试与性能优化实战4.1 可视化工作流LangGraph内置可视化工具这对复杂流程至关重要from IPython.display import Image # 生成流程图 Image(graph.get_graph().draw_mermaid_png())某次排查内存泄漏时我通过流程图发现有个循环边忘记设终止条件导致状态对象不断膨胀。4.2 性能压测技巧针对高并发场景我总结的优化方案节点级缓存对耗时的检索操作from functools import lru_cache lru_cache(maxsize100) def query_knowledge_base(question: str): # 向量检索等耗时操作 return results异步执行适合I/O密集型节点async def async_node(state): import asyncio await asyncio.sleep(0.1) return {data: result}批量处理合并同类请求def batch_process_node(state): # 假设state包含多个待处理项 bulk_results external_api.batch_call(state[items]) return {results: bulk_results}在最近的双十一大促中通过这三点优化我们的客服系统成功应对了每秒300的工单峰值。