1. 从零到一为什么选择 LangGraph 来构建客服 Agent最近在折腾一个智能客服的 Demo核心需求很明确能处理多轮对话能根据用户意图调用外部工具比如查订单、查物流。市面上 Agent 框架不少LangChain 名气大但说实话对于复杂的状态流转和循环逻辑用它的 Chain 来搭代码结构容易变得很“面条化”状态管理得自己费不少心思。而 LangGraph 的出现正好解决了这个痛点。它不是一个独立的框架而是构建在 LangChain 之上的一个库专门用来创建有状态、多步骤的、带循环的工作流。你可以把它想象成一个有向图编辑器节点是处理单元LLM调用、工具执行、条件判断边是状态流转的路径。用它来搭客服 Agent流程清晰得像画流程图状态管理完全交给框架开发者只需要关心每个节点的业务逻辑。我第一次用 LangGraph 重构一个旧的客服原型时感受最深的就是“清爽”。之前用纯 LangChain 搭各种RunnableBranch、RunnableLambda嵌套调试时追踪状态流转得在日志里大海捞针。换成 LangGraph 后整个对话流程一目了然哪个节点出问题状态卡在哪里通过可视化工具没错它自带看得一清二楚。这对于需要精准控制多轮对话逻辑比如先确认订单号再查询最后询问是否还需要其他帮助的场景简直是降维打击。所以如果你正在为如何优雅地管理一个会“记忆”、会“思考下一步”、会“执行动作”的智能体而头疼LangGraph 值得你花时间深入一下。2. 核心概念拆解State、Node、Edge 与 Graph在动手写代码之前必须把 LangGraph 的几个核心概念吃透。这就像玩乐高得先认识清楚各种基础积木块。2.1 State对话的“记忆中枢”在 LangGraph 中State是一个贯穿整个图执行过程的共享数据结构。它通常是一个 Pydantic 模型定义了工作流中需要传递和更新的所有信息。对于客服 Agent我们的 State 至少需要包含from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 对话历史LangGraph 提供了 add_messages 操作符来简化消息列表的追加 messages: Annotated[List, add_messages] # 用户当前输入 user_input: str # LLM 的原始响应包含可能的工具调用 llm_response: str # 已调用的工具及其结果列表 tool_calls: List[dict] # 当前对话轮次可用于控制流程 turn_count: int # 一个标志位指示是否应该结束对话 should_end: bool这里的关键是Annotated[List, add_messages]。add_messages是一个归约器reducer它定义了当新消息到来时如何更新messages字段。默认行为是追加到列表末尾。这种声明式的方式把状态更新的规则写在了数据结构里非常清晰。2.2 Node执行具体任务的“工作单元”Node 就是一个普通的 Python 函数或可调用对象它接收当前的State执行一些操作然后返回一个包含要更新字段的字典。这个字典会与原始 State 根据归约器规则进行合并。一个典型的 LLM 调用节点可能长这样from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(model“gpt-4o-mini”, temperature0) prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个专业的电商客服助手。请根据对话历史和工具调用结果友好、准确地回应用户。如果使用了工具请将工具返回的结果整合到回复中。”), (“placeholder”, “{messages}”) # 这里会自动注入完整的对话历史 ]) def llm_node(state: AgentState) - dict: # 1. 准备调用LLM chain prompt | llm # 2. 调用这里 messages 已由 State 提供 response chain.invoke({“messages”: state[“messages”]}) # 3. 返回要更新的状态 return {“llm_response”: response.content, “messages”: [response]}这个节点做了三件事构建调用链、调用 LLM、将 LLM 的响应内容更新到 State 中。注意我们把response对象本身也追加到了messages里这样下一次调用时LLM 就能看到自己上一次的回复形成连贯的对话历史。2.3 Edge决定流程走向的“路标”Edge 定义了在节点执行完毕后下一步应该走到哪个节点。它有两种主要类型条件边Conditional Edge根据 State 中的某个条件决定流向。这是实现多轮对话和循环的核心。普通边Normal Edge无条件地指向下一个节点。条件边通常由一个路由函数来实现def should_continue(state: AgentState) - str: # 检查 LLM 的响应中是否包含工具调用请求 last_message state[“messages”][-1] if hasattr(last_message, ‘tool_calls’) and len(last_message.tool_calls) 0: return “call_tool” # 需要调用工具 elif state.get(“should_end”, False): return “end” # 结束对话 else: return “respond” # 直接生成最终回复给用户这个路由函数检查最后一个消息是否包含tool_calls属性这是 OpenAI 格式消息中表示工具调用的字段如果有就路由到call_tool节点否则根据其他标志位决定是结束还是直接回复。2.4 Graph将一切组装起来最后我们用StateGraph这个类把 State、Node 和 Edge 组装成一个完整的工作流。from langgraph.graph import StateGraph, END # 初始化图并指定 State 的类型 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“process_input”, process_input_node) # 预处理用户输入 workflow.add_node(“call_llm”, llm_node) workflow.add_node(“call_tool”, tool_node) workflow.add_node(“format_response”, format_response_node) # 设置入口点 workflow.set_entry_point(“process_input”) # 添加普通边 workflow.add_edge(“process_input”, “call_llm”) workflow.add_edge(“format_response”, END) # END 是 LangGraph 内置的终止节点 # 添加条件边 workflow.add_conditional_edges( “call_llm”, # 来源节点 should_continue, # 路由函数 { “call_tool”: “call_tool”, # 如果返回 “call_tool”则跳转到 call_tool 节点 “respond”: “format_response”, “end”: END } ) workflow.add_edge(“call_tool”, “call_llm”) # 工具调用后重新咨询 LLM # 编译图 app workflow.compile()这个图定义了一个经典循环process_input - call_llm - (条件判断) - call_tool - call_llm … - format_response - END。工具调用后状态会带着工具执行结果流回call_llm节点让 LLM 根据新信息生成回复从而实现多轮交互。3. 实战构建一个具备查询和确认功能的客服 Agent现在我们用一个具体的电商客服场景来串联上述概念。假设我们的 Agent 需要处理两类请求查询订单状态和查询物流信息。每类查询都需要先确认订单号。3.1 定义工具并绑定给 LLM首先我们定义两个简单的工具函数并使用 LangChain 的tool装饰器来包装它们。from langchain.tools import tool from typing import Optional tool def get_order_status(order_id: str) - str: “”“根据订单号查询订单状态。这是一个模拟函数。”“” # 模拟数据库查询 mock_db { “ORD-12345”: “已发货”, “ORD-67890”: “待付款”, } status mock_db.get(order_id, “未找到该订单”) return f“订单 {order_id} 的状态是{status}” tool def get_logistics_info(order_id: str) - str: “”“根据订单号查询物流信息。这是一个模拟函数。”“” # 模拟 API 调用 mock_logistics { “ORD-12345”: “【XX快递】运单号 SF123456789最新轨迹已到达上海虹桥中转站”, “ORD-67890”: “该订单尚未发货无物流信息”, } info mock_logistics.get(order_id, “未找到该订单的物流信息”) return info接下来将工具绑定到 LLM 对象上。这样当 LLM 认为需要调用工具时它会在响应中返回结构化的工具调用请求。llm_with_tools llm.bind_tools([get_order_status, get_logistics_info])在之前的llm_node函数中我们需要将llm替换为llm_with_tools。这样当用户问“我的订单 12345 到哪了”时LLM 可能会返回一个包含tool_calls的响应指示需要调用get_logistics_info工具。3.2 实现工具调用节点工具调用节点call_tool需要解析 LLM 响应中的工具调用指令执行相应的工具并将结果格式化后存入 State。def call_tool_node(state: AgentState) - dict: messages state[“messages”] last_message messages[-1] # 获取 LLM 的响应消息 tool_calls_result [] # 遍历 LLM 要求的所有工具调用 for tool_call in last_message.tool_calls: tool_name tool_call[‘name’].lower() tool_args tool_call[‘args’] # 根据工具名选择执行哪个工具 if tool_name “get_order_status”: result get_order_status.invoke(tool_args) elif tool_name “get_logistics_info”: result get_logistics_invoke(tool_args) else: result f“未知工具调用{tool_name}” # 记录结果 tool_calls_result.append({ “call_id”: tool_call.get(‘id’), “name”: tool_name, “args”: tool_args, “result”: result }) # 关键步骤将工具执行结果构造为一个 ToolMessage并添加到消息历史中 # 这样 LLM 在下一轮就能看到工具返回的数据 from langchain_core.messages import ToolMessage tool_message ToolMessage( contentstr(result), nametool_name, tool_call_idtool_call.get(‘id’) ) messages.append(tool_message) # 返回更新后的状态 return { “tool_calls”: tool_calls_result, “messages”: messages # 注意这里返回的是整个更新后的 messages 列表 }这里有个非常重要的细节工具执行后必须将结果包装成ToolMessage并追加到messages列表。ToolMessage有一个tool_call_id字段与之前 LLM 请求中的id对应这帮助 LLM 将结果与请求关联起来。这是实现 LLM 与工具间正确对话上下文的关键。3.3 设计多轮对话的 State 与流程控制基本的工具调用循环有了但一个成熟的客服需要多轮对话来澄清意图。例如用户只说“查一下订单”我们需要反问“请问您的订单号是多少”。这需要更精细的 State 设计和流程控制。我们升级一下AgentState和流程class EnhancedAgentState(TypedDict): messages: Annotated[List, add_messages] user_input: str # 新增当前对话的“目标”或“阶段” current_goal: Optional[str] # 例如“query_order_status”, “query_logistics”, “clarify_order_id” # 新增临时存储的信息比如等待确认的订单号 pending_info: dict turn_count: int should_end: bool然后我们修改入口节点process_input_node和 LLM 的 System Prompt使其具备意图识别和状态管理能力。def process_input_node(state: EnhancedAgentState) - dict: user_input state[“user_input”] # 简单的规则式意图识别生产环境可用更复杂的 NLU 模型 if “订单状态” in user_input or “订单到哪了” in user_input: current_goal “query_order_status” elif “物流” in user_input or “快递” in user_input: current_goal “query_logistics” else: current_goal “general_chat” # 将用户输入转为 HumanMessage 存入历史 from langchain_core.messages import HumanMessage human_message HumanMessage(contentuser_input) return { “current_goal”: current_goal, “messages”: [human_message], “turn_count”: state.get(“turn_count”, 0) 1 } # 更新 LLM 的 System Prompt prompt ChatPromptTemplate.from_messages([ (“system”, “”” 你是一个电商客服助手。请根据以下规则进行对话 1. 如果用户想查询订单状态或物流但未提供订单号你必须主动、友好地询问订单号。 2. 只有当你获得了明确的订单号后才可以使用相应的工具进行查询。 3. 使用工具后请将结果清晰、友好地告知用户。 4. 当前对话阶段{current_goal}。待确认信息{pending_info}。 “””), (“placeholder”, “{messages}”) ]) def llm_node(state: EnhancedAgentState) - dict: # 将 current_goal 和 pending_info 也注入提示词 response (prompt | llm_with_tools).invoke({ “messages”: state[“messages”], “current_goal”: state.get(“current_goal”, “”), “pending_info”: str(state.get(“pending_info”, {})) }) return {“llm_response”: response.content, “messages”: [response]}同时我们需要修改路由函数should_continue加入对“澄清意图”阶段的判断。def should_continue(state: EnhancedAgentState) - str: last_message state[“messages”][-1] current_goal state.get(“current_goal”) # 情况1LLM 要求调用工具 if hasattr(last_message, ‘tool_calls’) and len(last_message.tool_calls) 0: return “call_tool” # 情况2处于需要澄清信息的阶段如等待订单号且 LLM 的回复是反问 # 这里用一个简单判断如果 LLM 回复包含“”且 current_goal 是查询类 if current_goal in [“query_order_status”, “query_logistics”] and “?” in last_message.content: # 此时不应该结束也不应该直接回复而是应该等待用户下一次输入。 # 但我们的图是单次触发所以这里返回 “respond”将 LLM 的反问直接返回给用户。 # 实际上对话暂停了等待下一次用户输入触发新的 graph 执行。 # 更复杂的实现可以引入“等待用户输入”的专用节点和外部事件驱动。 return “respond” # 情况3正常结束或需要直接回复 if state.get(“should_end”, False): return “end” else: return “respond”这个设计使得 Agent 具备了简单的多轮对话能力识别用户意图 - 若信息不足则反问 - 等待用户再次输入新的 graph 执行- 获取信息后调用工具 - 回复结果。3.4 编译、运行与可视化完成所有节点和边的定义后编译并运行它。app workflow.compile() # 运行一次对话 initial_state { “user_input”: “我的订单 12345 发货了吗”, “messages”: [], “current_goal”: None, “pending_info”: {}, “turn_count”: 0, “should_end”: False } final_state app.invoke(initial_state) print(final_state[“messages”][-1].content) # 打印最终回复 # LangGraph 的强大功能可视化 from langchain_core.runnables.graph import MermaidDrawer try: # 生成 Mermaid 图代码 graph_data MermaidDrawer().draw(app.graph) print(graph_data) # 可以将 graph_data 复制到 Mermaid 在线编辑器中查看 except: print(“可视化功能需要额外依赖。”)可视化功能让你能清晰看到整个工作流的全貌对于调试复杂流程至关重要。4. 避坑指南与进阶技巧在实际开发中你会遇到一些预料之外的问题。下面分享几个我踩过的坑和对应的解决方案。4.1 状态更新冲突与归约器Reducer的深入理解这是 LangGraph 初学者最容易困惑的地方。State 的更新不是简单的覆盖而是由归约器控制的合并。比如我们定义了messages: Annotated[List, add_messages]。add_messages这个归约器决定了当多个节点返回的更新中都包含messages字段时如何合并。假设节点 A 返回{“messages”: [msg1]}节点 B 返回{“messages”: [msg2]}。最终 State 中的messages会是[msg1, msg2]追加而不是只有msg2覆盖。这通常是我们想要的。但如果你错误地在某个节点里直接赋值了一个全新的列表比如return {“messages”: [new_msg]}这可能会破坏历史。最佳实践是在节点函数中总是基于传入的state[“messages”]进行修改并返回修改后的整个列表或增量列表。对于非列表的简单字段如turn_count直接返回新值即可因为默认的归约器是“覆盖”。4.2 工具调用格式的兼容性问题不同的 LLM 提供商OpenAI, Anthropic, 本地模型返回的工具调用格式可能略有不同。上面的代码示例假设了 OpenAI 的格式tool_calls列表。如果你使用其他模型需要适配。# 通用性更强的工具调用解析示例 def call_tool_node(state: AgentState) - dict: last_message state[“messages”][-1] tool_calls [] # 检查 OpenAI 格式 if hasattr(last_message, ‘tool_calls’): tool_calls last_message.tool_calls # 检查 LangChain 的 AIMessage 中附加的 tool_call 信息另一种常见格式 elif hasattr(last_message, ‘additional_kwargs’) and ‘tool_calls’ in last_message.additional_kwargs: tool_calls last_message.additional_kwargs[‘tool_calls’] # … 后续执行工具的逻辑 …建议在开发初期就使用print(last_message)或查看其type和dir()弄清楚你使用的 LLM 返回的消息对象具体结构。4.3 处理流式输出与中断LangGraph 的app.stream()方法支持流式输出这对于构建响应式的聊天应用很重要。同时你可能需要处理用户中途打断或取消任务的情况。# 流式调用 inputs {“user_input”: “查询订单ORD-12345的状态”, “messages”: []} for output in app.stream(inputs): for node_name, node_output in output.items(): if node_name “call_llm”: # node_output 可能是一个 AIMessage 的 Chunk if hasattr(node_output, ‘content’): # 可以逐块输出 content print(node_output.content, end“”, flushTrue) elif node_name “call_tool”: print(f“\n[正在调用工具{node_output.get(‘tool_name’)}]”)对于中断LangGraph 本身没有内置的“暂停”机制。一种常见的模式是在 State 中设置一个interrupted标志并在每个节点开始执行时检查它。或者在外部应用层控制不发起下一次app.invoke调用即可。4.4 引入长期记忆与对话总结基本的messages历史会随着对话变长而增加 token 消耗。一个优化策略是引入“对话总结”节点。在对话轮次达到一定数量或者检测到话题切换时触发一个子图让 LLM 对之前的对话进行摘要然后将摘要作为一条系统消息放入新的消息历史开头并清空或截断旧的历史消息。这可以通过在图中添加一个summarize_conversation节点并在特定条件下通过条件边路由到该节点来实现。这个节点会读取state[“messages”]调用 LLM 生成摘要然后返回更新后的state其中messages列表被替换为[SystemMessage(content摘要), 最近一两轮对话]。4.5 子图Subgraph管理复杂逻辑当客服 Agent 的逻辑变得非常复杂时比如处理退货流程包含多个确认步骤你可以使用子图。子图允许你将一个复杂的、多步骤的逻辑如“退货流程”封装成一个独立的图然后在主图中将其作为一个节点来调用。这极大地提升了代码的模块化和可维护性。from langgraph.graph import StateGraph as SubStateGraph # 1. 定义一个处理退货的子图 def create_return_subgraph(): graph SubStateGraph(ReturnSubState) graph.add_node(“validate_return”, validate_return_node) graph.add_node(“generate_rma”, generate_rma_node) graph.add_edge(“validate_return”, “generate_rma”) graph.set_entry_point(“validate_return”) return graph.compile() # 2. 在主图中将子图作为一个节点添加 return_subgraph create_return_subgraph() main_workflow.add_node(“handle_return”, return_subgraph)主图执行到handle_return节点时会进入子图执行子图有自己的状态定义和流程执行完毕后再带着结果返回主图。构建一个功能完善的客服 Agent 远不止调用 API 那么简单它涉及对对话状态、工具执行、异常流程的精细管理。LangGraph 提供的图抽象恰好为这种复杂、有状态的工作流提供了清晰、可维护的编程模型。从简单的工具调用循环开始逐步引入意图识别、多轮澄清、记忆管理等模块你能像搭积木一样构建出能力强大的智能体。最关键的是它的可视化特性和明确的状态流让调试和协作变得前所未有的直观。