1. 从“链”到“图”为什么我们需要 LangGraph如果你用过 LangChain那你一定对“链”Chain这个概念不陌生。它就像一条流水线把大语言模型LLM和工具Tools串联起来按顺序处理任务。比如先让LLM理解用户问题再调用搜索引擎查资料最后让LLM总结答案。这在处理线性、确定性的任务时非常高效。但现实世界的任务尤其是那些复杂的、需要决策和状态管理的任务往往不是一条直线。想象一下一个客服机器人用户说“我想订一张明天去北京的机票”。这看似简单但背后可能涉及多个步骤和分支查询航班、确认时间、选择舱位、获取乘客信息、创建订单、处理支付……而且用户可能在任意环节改变主意“等等我改后天吧”或者“我先看看酒店”。用传统的“链”来建模这种对话代码会变得异常复杂和脆弱充满了大量的if-else判断来处理状态跳转。这就是 LangGraph 诞生的背景。它不是要取代 LangChain而是一个强有力的补充专门用来构建有状态的、多步骤的、带循环和条件分支的智能体Agent应用。LangGraph 的核心思想是“图”Graph。在计算机科学中图由“节点”Nodes和“边”Edges组成。在这里节点可以是一个LLM调用、一个工具执行、或者一段自定义的逻辑函数。边定义了工作流的方向一个节点执行完后接下来该去哪个节点这种图结构天然适合描述带有分支、循环和状态的工作流。而让这张图“活”起来让信息在不同节点间有序流动的关键就是Graph 的通讯机制。理解了这个机制你才能真正掌握 LangGraph 的精髓而不仅仅是照猫画虎地拼接几个节点。本文将深入拆解 LangGraph 的通讯核心State对象和Runnable接口并通过一个从零开始的订票机器人案例带你从入门到精通。2. 通讯基石深入理解 State 对象在 LangGraph 中State是整个工作流运行时唯一的核心数据容器。所有节点都读取它、修改它并通过它来传递信息。你可以把它想象成一个共享的、可变的“白板”或者“上下文背包”工作流中的每个参与者都在上面读写数据。2.1 State 的本质与定义LangGraph 中的 State 通常是一个 Pydantic BaseModel 的子类。使用 Pydantic 的好处是提供了强大的类型提示和数据验证。我们来看一个订票场景下最基础的 State 定义from typing import Optional, List, Annotated from typing_extensions import TypedDict from pydantic import BaseModel, Field from langgraph.graph.message import add_messages # 方式一使用 Pydantic BaseModel经典且功能强大 class AgentState(BaseModel): # 对话消息历史。Annotated 和 add_messages 是 LangGraph 提供的特殊类型用于自动管理消息列表的合并。 messages: Annotated[List[BaseMessage], add_messages] Field(default_factorylist) # 用户原始输入 user_input: str # 解析出的用户意图如 “book_flight”, “change_order”, “inquire” intent: Optional[str] None # 从对话中提取的航班信息 flight_info: Optional[dict] None # 当前工作流执行到的步骤 current_step: str start # 是否需要人工接管 needs_human: bool False # 方式二使用 TypedDict更轻量Python原生 class AgentState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] user_input: str intent: Optional[str] flight_info: Optional[dict] current_step: str needs_human: bool为什么这么设计类型安全与自文档化每个字段都有明确的类型比如flight_info是Optional[dict]这比使用一个普通的字典state[“flight_info”]要清晰得多IDE 也能提供自动补全和错误检查。结构化数据流State 定义了工作流中所有可能被传递的数据结构。这迫使开发者在设计工作流之初就思考清楚数据的生命周期避免了后期数据混乱的问题。序列化友好Pydantic 模型可以轻松地转换为 JSON这对于持久化工作流状态比如存入数据库、调试和可视化至关重要。2.2 消息Messages的特殊处理你可能注意到了messages字段的奇怪类型Annotated[List[BaseMessage], add_messages]。这是 LangGraph 通讯机制的第一个精妙之处。BaseMessage来自 LangChain是聊天消息的基类常见子类有HumanMessage用户输入、AIMessageAI回复、ToolMessage工具执行结果等。Annotated与add_messages这是一个“缩减器”Reducer。它的作用是定义当多个节点同时向state.messages字段写入数据时如何合并这些数据。add_messages这个缩减器的逻辑很简单将新的消息列表追加到旧的消息列表后面。这意味着你不需要在节点函数里写state.messages.append(new_message)。LangGraph 会自动帮你处理。你只需要在函数中返回一个包含messages键的字典例如{“messages”: [AIMessage(content“…”)]}框架会自动将其合并到全局的state.messages中。这保证了对话历史的完整性和有序性是构建多轮对话智能体的基础。2.3 State 的流转与修改规则State 对象在节点间是可变且持续更新的。每个节点函数接收当前的状态State执行逻辑然后返回一个更新字典。这个更新字典只会包含它想要修改的那些字段。def intent_classifier(state: AgentState): 节点函数识别用户意图 latest_message state.messages[-1] # 获取最新的一条用户消息 user_text latest_message.content # 调用LLM或规则进行意图分类 # 假设我们有一个简单的分类逻辑 if “订票” in user_text or “航班” in user_text: intent “book_flight” elif “取消” in user_text: intent “cancel_flight” else: intent “chitchat” # 返回一个更新字典只更新 intent 字段。 # LangGraph 会用这个字典去更新全局的 State其他字段保持不变。 return {“intent”: intent, “current_step”: “intent_classified”}关键理解节点函数返回的不是全新的 State 对象而是一个“补丁”。LangGraph 内部会应用这个补丁到当前的 State 上。这种设计非常高效也符合直觉——每个节点只关心和修改自己负责的那部分数据。3. 节点的契约Runnable 接口与函数签名节点是图的工作单元而Runnable接口是 LangGraph 中所有可执行组件的统一抽象。无论是调用一个LLM、执行一个工具还是运行你自定义的Python函数在 LangGraph 眼里都是“可运行”的。这为图的组合提供了极大的灵活性。3.1 将各种组件包装成节点一个节点本质上是一个接受输入通常是 State并返回输出更新字典的可调用对象。以下是如何将不同组件变成节点from langchain_core.runnables import RunnableLambda from langchain_openai import ChatOpenAI from langchain.tools import Tool # 1. 最简单的自定义函数节点 def echo_node(state: AgentState): print(f“当前步骤{state.current_step}”) return {“current_step”: “echo_done”} # 2. 使用 RunnableLambda 包装自定义函数更规范 custom_node RunnableLambda(echo_node) # 3. 将 LangChain LCEL 链作为节点 llm ChatOpenAI(model“gpt-4”) prompt_template … # 定义一个Prompt chain prompt_template | llm # 这是一个 LCEL 链 def chain_node(state: AgentState): # 从state中构造链的输入 result chain.invoke({“question”: state.user_input}) # 将结果封装成更新字典 return {“messages”: [AIMessage(contentresult.content)]} # 4. 将 Tool 的执行作为节点 search_tool Tool(name“Search”, funcsearch_function, description“…”) def tool_node(state: AgentState): # 假设我们从state里解析出了查询词 query state.flight_info.get(“query”) tool_result search_tool.invoke(query) return {“messages”: [ToolMessage(contenttool_result, tool_call_id“1”)]}核心要点节点的函数签名通常是def node_function(state: StateType) - dict。它从state中读取所需数据经过处理返回一个字典指明要更新state的哪些部分。RunnableLambda是一个非常有用的包装器它能让你的普通函数完全融入 LangGraph 的生态体系。3.2 边Edges与条件路由节点定义了“做什么”边则定义了“接下来去哪”。这是通讯机制的控制流部分。LangGraph 提供了两种主要的边普通边add_edge无条件地从源节点指向目标节点。条件边add_conditional_edges根据当前 State 的内容动态决定下一个节点。条件边是实现分支和循环的关键。它需要一个“路由函数”这个函数接收 State并返回下一个要执行的节点的名称字符串。from langgraph.graph import StateGraph, END workflow StateGraph(AgentState) # 先添加几个节点 workflow.add_node(“classify_intent”, intent_classifier) workflow.add_node(“handle_booking”, booking_handler) workflow.add_node(“handle_cancellation”, cancellation_handler) workflow.add_node(“make_small_talk”, small_talk_handler) # 设置入口点 workflow.set_entry_point(“classify_intent”) # 添加条件边根据 classify_intent 节点输出的 intent 字段决定路由 def route_by_intent(state: AgentState): # 这个函数在 classify_intent 节点执行后被调用 # 此时 state 已经被 classify_intent 节点的返回结果更新过 intent state.get(“intent”) if intent “book_flight”: return “handle_booking” elif intent “cancel_flight”: return “handle_cancellation” else: return “make_small_talk” # 将条件边从 classify_intent 节点引出 workflow.add_conditional_edges( “classify_intent”, # 源节点 route_by_intent, # 路由函数 { “handle_booking”: “handle_booking”, “handle_cancellation”: “handle_cancellation”, “make_small_talk”: “make_small_talk” } ) # 为其他节点添加普通边指向结束或下一个环节 workflow.add_edge(“handle_booking”, END) workflow.add_edge(“handle_cancellation”, END) workflow.add_edge(“make_small_talk”, END)通讯视角解读classify_intent节点通过修改state.intent字段向图“发送”了一个信号。route_by_intent这个路由函数“监听”了这个信号读取该字段并据此决定消息流的下一个目的地。这完美体现了通过共享 State 进行通讯和控制的范式。4. 实战构建一个订票机器人工作流现在我们将前面所有的概念整合起来构建一个简化但完整的机票预订机器人。这个机器人能处理意图分类、信息收集、确认和异常处理。4.1 定义完整的 State 与节点首先我们定义一个更丰富的 State 和所需的节点函数。from enum import Enum from pydantic import BaseModel, Field from typing import List, Optional, Annotated from langchain_core.messages import HumanMessage, AIMessage, BaseMessage from langgraph.graph.message import add_messages class Step(str, Enum): START “start” INTENT_CLASSIFIED “intent_classified” COLLECTING_INFO “collecting_info” CONFIRMING “confirming” COMPLETED “completed” NEEDS_HUMAN “needs_human” class BookingState(BaseModel): messages: Annotated[List[BaseMessage], add_messages] Field(default_factorylist) current_step: Step Step.START intent: Optional[str] None extracted_info: dict Field(default_factorydict) # 存放提取的出发地、目的地、时间等 confirmation_pending: bool False needs_human: bool False # 节点1欢迎与初始询问 def welcome_node(state: BookingState): welcome_msg AIMessage(content“您好我是机票预订助手。请问您需要什么帮助例如订票、查询、取消”) return { “messages”: [welcome_msg], “current_step”: Step.START } # 节点2意图分类使用一个模拟的LLM调用 def classify_intent_node(state: BookingState): last_msg state.messages[-1].content # 模拟一个简单的分类逻辑实践中这里应调用LLM if any(word in last_msg for word in [“订”, “买”, “预订”, “flight”, “book”]): intent “book” reply “好的我将为您办理机票预订。请告诉我您的出发城市、目的地和出行日期例如北京到上海明天。 elif “取消” in last_msg: intent “cancel” reply “我将帮您处理订单取消。请提供您的订单号。” else: intent “inquire” reply “我将为您查询航班信息。请告诉我行程详情。” return { “messages”: [AIMessage(contentreply)], “intent”: intent, “current_step”: Step.INTENT_CLASSIFIED } # 节点3信息收集与填充模拟信息提取 def collect_info_node(state: BookingState): # 这是一个简化示例。真实场景可能需要多轮对话和NLU解析。 user_input state.messages[-1].content extracted state.extracted_info.copy() # 非常简单的关键词提取 if “到” in user_input: parts user_input.split(“到”) extracted[“departure”] parts[0].strip() extracted[“destination”] parts[1].split(“”)[0].strip() if “明天” in user_input: extracted[“date”] “2023-10-27” # 模拟日期 reply f“好的正在为您查找航班。\n已识别信息{extracted}。\n请问出行人数是” return { “messages”: [AIMessage(contentreply)], “extracted_info”: extracted, # 更新提取的信息 “current_step”: Step.COLLECTING_INFO } # 节点4信息确认 def confirm_node(state: BookingState): info state.extracted_info confirm_msg f“请确认您的预订信息\n{info}\n回复‘确认’以继续或提出修改。” return { “messages”: [AIMessage(contentconfirm_msg)], “confirmation_pending”: True, “current_step”: Step.CONFIRMING } # 节点5最终处理模拟订票成功 def finalize_node(state: BookingState): order_id “ORD123456” success_msg f“预订成功您的订单号是 {order_id}。祝您旅途愉快” return { “messages”: [AIMessage(contentsuccess_msg)], “current_step”: Step.COMPLETED } # 节点6人工接管 def human_agent_node(state: BookingState): escalation_msg “您的问题比较复杂我将为您转接人工客服请稍候。” return { “messages”: [AIMessage(contentescalation_msg)], “needs_human”: True, “current_step”: Step.NEEDS_HUMAN }4.2 构建图并定义复杂的路由逻辑接下来我们用StateGraph将这些节点连接起来并设置条件路由。from langgraph.graph import StateGraph, END # 初始化图 workflow StateGraph(BookingState) # 添加所有节点 workflow.add_node(“welcome”, welcome_node) workflow.add_node(“classify_intent”, classify_intent_node) workflow.add_node(“collect_info”, collect_info_node) workflow.add_node(“confirm”, confirm_node) workflow.add_node(“finalize”, finalize_node) workflow.add_node(“human_agent”, human_agent_node) # 设置入口点 workflow.set_entry_point(“welcome”) # 1. 从 welcome 无条件到 classify_intent workflow.add_edge(“welcome”, “classify_intent”) # 2. 从 classify_intent 根据意图路由 def route_after_intent(state: BookingState): intent state.intent if intent “book”: # 如果是订票进入信息收集环节 return “collect_info” elif intent “cancel”: # 如果是取消我们简化处理直接转人工或另一个取消流程节点 return “human_agent” else: # 查询或其他意图也转人工/特定处理节点 return “human_agent” workflow.add_conditional_edges( “classify_intent”, route_after_intent, {“collect_info”: “collect_info”, “human_agent”: “human_agent”} ) # 3. 从 collect_info 节点出来判断信息是否足够 def route_after_collect(state: BookingState): info state.extracted_info # 简单判断如果已收集到出发地、目的地和日期则认为信息足够进入确认环节 if all(k in info for k in [“departure”, “destination”, “date”]): return “confirm” else: # 信息不足返回自身继续收集这里会产生循环 return “collect_info” workflow.add_conditional_edges( “collect_info”, route_after_collect, {“confirm”: “confirm”, “collect_info”: “collect_info”} ) # 4. 从 confirm 节点出来根据用户最新回复决定下一步 def route_after_confirm(state: BookingState): if not state.confirmation_pending: return END last_msg_content state.messages[-1].content.lower() if “确认” in last_msg_content: return “finalize” elif “修改” in last_msg_content: # 用户要修改清空部分信息回到收集节点 return “collect_info” else: # 其他未识别的回复转人工 return “human_agent” workflow.add_conditional_edges( “confirm”, route_after_confirm, {“finalize”: “finalize”, “collect_info”: “collect_info”, “human_agent”: “human_agent”} ) # 5. 设置最终节点和人工节点的出口 workflow.add_edge(“finalize”, END) workflow.add_edge(“human_agent”, END) # 编译图 app workflow.compile()4.3 运行与调试观察 State 的流动现在让我们运行这个工作流并观察 State 是如何在节点间传递和演变的。from langchain_core.messages import HumanMessage # 初始化状态 initial_state {“messages”: [HumanMessage(content“我想订一张票”)]} # 使用 stream 方式运行可以观察每个步骤后的状态 for event in app.stream(initial_state, stream_mode“values”): node_name list(event.keys())[0] state event[node_name] print(f“\n 节点 [{node_name}] 执行后 ) print(f“当前步骤: {state.current_step}”) print(f“意图: {state.intent}”) print(f“提取信息: {state.extracted_info}”) print(f“最新消息: {state.messages[-1].content if state.messages else ‘None’}”) print(“-” * 50)运行上述代码你会看到类似以下的输出清晰地展示了通讯过程 节点 [welcome] 执行后 当前步骤: start 意图: None 提取信息: {} 最新消息: 您好我是机票预订助手。请问您需要什么帮助例如订票、查询、取消 -------------------------------------------------- 节点 [classify_intent] 执行后 当前步骤: intent_classified 意图: book 提取信息: {} 最新消息: 好的我将为您办理机票预订。请告诉我您的出发城市、目的地和出行日期例如北京到上海明天。 -------------------------------------------------- 节点 [collect_info] 执行后 当前步骤: collecting_info 意图: book 提取信息: {‘departure’: ‘北京’ ‘destination’: ‘上海’ ‘date’: ‘2023-10-27’} 最新消息: 好的正在为您查找航班。已识别信息{‘departure’: ‘北京’ ‘destination’: ‘上海’ ‘date’: ‘2023-10-27’}。请问出行人数是 -------------------------------------------------- ...通过这个流式输出你可以像看日志一样追踪BookingState对象中每个字段是如何随着每个节点的执行而变化的。intent在classify_intent节点被赋值extracted_info在collect_info节点被填充current_step驱动着路由逻辑。这就是 LangGraph 通讯机制的生动体现State 是共享的上下文节点是修改者边是路由控制器三者协同完成复杂的工作流。5. 高级通讯模式与最佳实践掌握了基础机制后我们来看看更高级的用法和实践中容易踩的坑。5.1 子图Subgraphs与模块化对于复杂系统你可以将一部分节点和边打包成一个子图。子图本身也是一个可编译、可运行的图它可以作为父图的一个节点。这是实现模块化和复用的关键。from langgraph.graph import StateGraph def create_booking_subgraph(): “”“创建一个负责订票核心流程的子图”“” subgraph StateGraph(BookingState) subgraph.add_node(“collect_info”, collect_info_node) subgraph.add_node(“confirm”, confirm_node) subgraph.add_node(“finalize”, finalize_node) subgraph.set_entry_point(“collect_info”) subgraph.add_conditional_edges(“collect_info”, route_after_collect) subgraph.add_conditional_edges(“confirm”, route_after_confirm) subgraph.add_edge(“finalize”, END) return subgraph.compile() # 在主图中将子图作为一个节点添加 main_workflow StateGraph(BookingState) main_workflow.add_node(“welcome”, welcome_node) main_workflow.add_node(“classify”, classify_intent_node) main_workflow.add_node(“booking_workflow”, create_booking_subgraph()) # 子图作为节点 main_workflow.add_node(“human”, human_agent_node) # … 设置主图的路由逻辑例如 classify 后路由到 booking_workflow 节点通讯意义子图内部的节点通过子图自己的 State 进行通讯对外部而言子图是一个黑盒。主图只关心传递给子图什么初始状态以及接收子图返回的最终状态更新。这极大地简化了复杂工作流的管理和调试。5.2 中断Interruption与长期记忆LangGraph 支持“中断”机制允许工作流在特定节点暂停等待外部输入如用户反馈、异步API回调然后再继续。这通常通过结合pregelLangGraph的底层执行引擎的checkpoint功能和外部存储来实现。其通讯本质是将当前的完整State序列化后保存到数据库如Redis。当外部事件触发继续时从数据库加载该 State并从中断的节点继续执行。这为实现需要等待人类审核或长时间运行任务的智能体提供了可能。5.3 避坑指南与实操心得State 设计要前瞻在项目初期花时间设计一个好的 State 模型。字段名要清晰类型要明确。避免后期不断添加字段导致结构混乱。考虑使用Optional类型来处理字段的渐进式填充。路由函数要保持纯净路由函数应该只读取 State 并返回下一个节点名。不要在路由函数里修改 State。修改 State 是节点的职责。混用职责会导致状态变更难以追踪和调试。善用stream模式进行调试如上文所示app.stream()是调试 LangGraph 应用的神器。它能让你可视化工作流的执行路径和 State 的每一次变化快速定位逻辑错误或路由问题。注意消息列表的合并由于使用了add_messages缩减器你返回的{“messages”: [new_message]}会被追加。如果你需要替换或修改某条特定历史消息需要更精细的操作比如在 State 中维护一个独立的message_history列表和current_prompt字段。子图入口和出口的 State 映射当子图作为节点时要清楚主图的 State 如何映射到子图所需的 State。有时可能需要一个适配器函数来提取或转换字段。错误处理节点函数可能抛出异常。LangGraph 本身不提供复杂的错误处理节点。一种模式是在关键节点外包裹try…except并在出错时返回一个如{“error”: str(e), “current_step”: “error”}的更新然后通过条件边路由到一个专门的“错误处理”节点。理解 LangGraph 的通讯机制就是理解其以State 为中心的数据流和以条件边为驱动的控制流。这不同于传统的函数调用链它提供了一种声明式的、图形化的方式来构建复杂的、有状态的应用程序。从简单的线性流程到带有循环、分支和中断的复杂智能体LangGraph 的这套机制都能提供清晰、可维护的建模方式。 LangGraph通讯机制解析：State与Runnable构建智能体工作流
1. 从“链”到“图”为什么我们需要 LangGraph如果你用过 LangChain那你一定对“链”Chain这个概念不陌生。它就像一条流水线把大语言模型LLM和工具Tools串联起来按顺序处理任务。比如先让LLM理解用户问题再调用搜索引擎查资料最后让LLM总结答案。这在处理线性、确定性的任务时非常高效。但现实世界的任务尤其是那些复杂的、需要决策和状态管理的任务往往不是一条直线。想象一下一个客服机器人用户说“我想订一张明天去北京的机票”。这看似简单但背后可能涉及多个步骤和分支查询航班、确认时间、选择舱位、获取乘客信息、创建订单、处理支付……而且用户可能在任意环节改变主意“等等我改后天吧”或者“我先看看酒店”。用传统的“链”来建模这种对话代码会变得异常复杂和脆弱充满了大量的if-else判断来处理状态跳转。这就是 LangGraph 诞生的背景。它不是要取代 LangChain而是一个强有力的补充专门用来构建有状态的、多步骤的、带循环和条件分支的智能体Agent应用。LangGraph 的核心思想是“图”Graph。在计算机科学中图由“节点”Nodes和“边”Edges组成。在这里节点可以是一个LLM调用、一个工具执行、或者一段自定义的逻辑函数。边定义了工作流的方向一个节点执行完后接下来该去哪个节点这种图结构天然适合描述带有分支、循环和状态的工作流。而让这张图“活”起来让信息在不同节点间有序流动的关键就是Graph 的通讯机制。理解了这个机制你才能真正掌握 LangGraph 的精髓而不仅仅是照猫画虎地拼接几个节点。本文将深入拆解 LangGraph 的通讯核心State对象和Runnable接口并通过一个从零开始的订票机器人案例带你从入门到精通。2. 通讯基石深入理解 State 对象在 LangGraph 中State是整个工作流运行时唯一的核心数据容器。所有节点都读取它、修改它并通过它来传递信息。你可以把它想象成一个共享的、可变的“白板”或者“上下文背包”工作流中的每个参与者都在上面读写数据。2.1 State 的本质与定义LangGraph 中的 State 通常是一个 Pydantic BaseModel 的子类。使用 Pydantic 的好处是提供了强大的类型提示和数据验证。我们来看一个订票场景下最基础的 State 定义from typing import Optional, List, Annotated from typing_extensions import TypedDict from pydantic import BaseModel, Field from langgraph.graph.message import add_messages # 方式一使用 Pydantic BaseModel经典且功能强大 class AgentState(BaseModel): # 对话消息历史。Annotated 和 add_messages 是 LangGraph 提供的特殊类型用于自动管理消息列表的合并。 messages: Annotated[List[BaseMessage], add_messages] Field(default_factorylist) # 用户原始输入 user_input: str # 解析出的用户意图如 “book_flight”, “change_order”, “inquire” intent: Optional[str] None # 从对话中提取的航班信息 flight_info: Optional[dict] None # 当前工作流执行到的步骤 current_step: str start # 是否需要人工接管 needs_human: bool False # 方式二使用 TypedDict更轻量Python原生 class AgentState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] user_input: str intent: Optional[str] flight_info: Optional[dict] current_step: str needs_human: bool为什么这么设计类型安全与自文档化每个字段都有明确的类型比如flight_info是Optional[dict]这比使用一个普通的字典state[“flight_info”]要清晰得多IDE 也能提供自动补全和错误检查。结构化数据流State 定义了工作流中所有可能被传递的数据结构。这迫使开发者在设计工作流之初就思考清楚数据的生命周期避免了后期数据混乱的问题。序列化友好Pydantic 模型可以轻松地转换为 JSON这对于持久化工作流状态比如存入数据库、调试和可视化至关重要。2.2 消息Messages的特殊处理你可能注意到了messages字段的奇怪类型Annotated[List[BaseMessage], add_messages]。这是 LangGraph 通讯机制的第一个精妙之处。BaseMessage来自 LangChain是聊天消息的基类常见子类有HumanMessage用户输入、AIMessageAI回复、ToolMessage工具执行结果等。Annotated与add_messages这是一个“缩减器”Reducer。它的作用是定义当多个节点同时向state.messages字段写入数据时如何合并这些数据。add_messages这个缩减器的逻辑很简单将新的消息列表追加到旧的消息列表后面。这意味着你不需要在节点函数里写state.messages.append(new_message)。LangGraph 会自动帮你处理。你只需要在函数中返回一个包含messages键的字典例如{“messages”: [AIMessage(content“…”)]}框架会自动将其合并到全局的state.messages中。这保证了对话历史的完整性和有序性是构建多轮对话智能体的基础。2.3 State 的流转与修改规则State 对象在节点间是可变且持续更新的。每个节点函数接收当前的状态State执行逻辑然后返回一个更新字典。这个更新字典只会包含它想要修改的那些字段。def intent_classifier(state: AgentState): 节点函数识别用户意图 latest_message state.messages[-1] # 获取最新的一条用户消息 user_text latest_message.content # 调用LLM或规则进行意图分类 # 假设我们有一个简单的分类逻辑 if “订票” in user_text or “航班” in user_text: intent “book_flight” elif “取消” in user_text: intent “cancel_flight” else: intent “chitchat” # 返回一个更新字典只更新 intent 字段。 # LangGraph 会用这个字典去更新全局的 State其他字段保持不变。 return {“intent”: intent, “current_step”: “intent_classified”}关键理解节点函数返回的不是全新的 State 对象而是一个“补丁”。LangGraph 内部会应用这个补丁到当前的 State 上。这种设计非常高效也符合直觉——每个节点只关心和修改自己负责的那部分数据。3. 节点的契约Runnable 接口与函数签名节点是图的工作单元而Runnable接口是 LangGraph 中所有可执行组件的统一抽象。无论是调用一个LLM、执行一个工具还是运行你自定义的Python函数在 LangGraph 眼里都是“可运行”的。这为图的组合提供了极大的灵活性。3.1 将各种组件包装成节点一个节点本质上是一个接受输入通常是 State并返回输出更新字典的可调用对象。以下是如何将不同组件变成节点from langchain_core.runnables import RunnableLambda from langchain_openai import ChatOpenAI from langchain.tools import Tool # 1. 最简单的自定义函数节点 def echo_node(state: AgentState): print(f“当前步骤{state.current_step}”) return {“current_step”: “echo_done”} # 2. 使用 RunnableLambda 包装自定义函数更规范 custom_node RunnableLambda(echo_node) # 3. 将 LangChain LCEL 链作为节点 llm ChatOpenAI(model“gpt-4”) prompt_template … # 定义一个Prompt chain prompt_template | llm # 这是一个 LCEL 链 def chain_node(state: AgentState): # 从state中构造链的输入 result chain.invoke({“question”: state.user_input}) # 将结果封装成更新字典 return {“messages”: [AIMessage(contentresult.content)]} # 4. 将 Tool 的执行作为节点 search_tool Tool(name“Search”, funcsearch_function, description“…”) def tool_node(state: AgentState): # 假设我们从state里解析出了查询词 query state.flight_info.get(“query”) tool_result search_tool.invoke(query) return {“messages”: [ToolMessage(contenttool_result, tool_call_id“1”)]}核心要点节点的函数签名通常是def node_function(state: StateType) - dict。它从state中读取所需数据经过处理返回一个字典指明要更新state的哪些部分。RunnableLambda是一个非常有用的包装器它能让你的普通函数完全融入 LangGraph 的生态体系。3.2 边Edges与条件路由节点定义了“做什么”边则定义了“接下来去哪”。这是通讯机制的控制流部分。LangGraph 提供了两种主要的边普通边add_edge无条件地从源节点指向目标节点。条件边add_conditional_edges根据当前 State 的内容动态决定下一个节点。条件边是实现分支和循环的关键。它需要一个“路由函数”这个函数接收 State并返回下一个要执行的节点的名称字符串。from langgraph.graph import StateGraph, END workflow StateGraph(AgentState) # 先添加几个节点 workflow.add_node(“classify_intent”, intent_classifier) workflow.add_node(“handle_booking”, booking_handler) workflow.add_node(“handle_cancellation”, cancellation_handler) workflow.add_node(“make_small_talk”, small_talk_handler) # 设置入口点 workflow.set_entry_point(“classify_intent”) # 添加条件边根据 classify_intent 节点输出的 intent 字段决定路由 def route_by_intent(state: AgentState): # 这个函数在 classify_intent 节点执行后被调用 # 此时 state 已经被 classify_intent 节点的返回结果更新过 intent state.get(“intent”) if intent “book_flight”: return “handle_booking” elif intent “cancel_flight”: return “handle_cancellation” else: return “make_small_talk” # 将条件边从 classify_intent 节点引出 workflow.add_conditional_edges( “classify_intent”, # 源节点 route_by_intent, # 路由函数 { “handle_booking”: “handle_booking”, “handle_cancellation”: “handle_cancellation”, “make_small_talk”: “make_small_talk” } ) # 为其他节点添加普通边指向结束或下一个环节 workflow.add_edge(“handle_booking”, END) workflow.add_edge(“handle_cancellation”, END) workflow.add_edge(“make_small_talk”, END)通讯视角解读classify_intent节点通过修改state.intent字段向图“发送”了一个信号。route_by_intent这个路由函数“监听”了这个信号读取该字段并据此决定消息流的下一个目的地。这完美体现了通过共享 State 进行通讯和控制的范式。4. 实战构建一个订票机器人工作流现在我们将前面所有的概念整合起来构建一个简化但完整的机票预订机器人。这个机器人能处理意图分类、信息收集、确认和异常处理。4.1 定义完整的 State 与节点首先我们定义一个更丰富的 State 和所需的节点函数。from enum import Enum from pydantic import BaseModel, Field from typing import List, Optional, Annotated from langchain_core.messages import HumanMessage, AIMessage, BaseMessage from langgraph.graph.message import add_messages class Step(str, Enum): START “start” INTENT_CLASSIFIED “intent_classified” COLLECTING_INFO “collecting_info” CONFIRMING “confirming” COMPLETED “completed” NEEDS_HUMAN “needs_human” class BookingState(BaseModel): messages: Annotated[List[BaseMessage], add_messages] Field(default_factorylist) current_step: Step Step.START intent: Optional[str] None extracted_info: dict Field(default_factorydict) # 存放提取的出发地、目的地、时间等 confirmation_pending: bool False needs_human: bool False # 节点1欢迎与初始询问 def welcome_node(state: BookingState): welcome_msg AIMessage(content“您好我是机票预订助手。请问您需要什么帮助例如订票、查询、取消”) return { “messages”: [welcome_msg], “current_step”: Step.START } # 节点2意图分类使用一个模拟的LLM调用 def classify_intent_node(state: BookingState): last_msg state.messages[-1].content # 模拟一个简单的分类逻辑实践中这里应调用LLM if any(word in last_msg for word in [“订”, “买”, “预订”, “flight”, “book”]): intent “book” reply “好的我将为您办理机票预订。请告诉我您的出发城市、目的地和出行日期例如北京到上海明天。 elif “取消” in last_msg: intent “cancel” reply “我将帮您处理订单取消。请提供您的订单号。” else: intent “inquire” reply “我将为您查询航班信息。请告诉我行程详情。” return { “messages”: [AIMessage(contentreply)], “intent”: intent, “current_step”: Step.INTENT_CLASSIFIED } # 节点3信息收集与填充模拟信息提取 def collect_info_node(state: BookingState): # 这是一个简化示例。真实场景可能需要多轮对话和NLU解析。 user_input state.messages[-1].content extracted state.extracted_info.copy() # 非常简单的关键词提取 if “到” in user_input: parts user_input.split(“到”) extracted[“departure”] parts[0].strip() extracted[“destination”] parts[1].split(“”)[0].strip() if “明天” in user_input: extracted[“date”] “2023-10-27” # 模拟日期 reply f“好的正在为您查找航班。\n已识别信息{extracted}。\n请问出行人数是” return { “messages”: [AIMessage(contentreply)], “extracted_info”: extracted, # 更新提取的信息 “current_step”: Step.COLLECTING_INFO } # 节点4信息确认 def confirm_node(state: BookingState): info state.extracted_info confirm_msg f“请确认您的预订信息\n{info}\n回复‘确认’以继续或提出修改。” return { “messages”: [AIMessage(contentconfirm_msg)], “confirmation_pending”: True, “current_step”: Step.CONFIRMING } # 节点5最终处理模拟订票成功 def finalize_node(state: BookingState): order_id “ORD123456” success_msg f“预订成功您的订单号是 {order_id}。祝您旅途愉快” return { “messages”: [AIMessage(contentsuccess_msg)], “current_step”: Step.COMPLETED } # 节点6人工接管 def human_agent_node(state: BookingState): escalation_msg “您的问题比较复杂我将为您转接人工客服请稍候。” return { “messages”: [AIMessage(contentescalation_msg)], “needs_human”: True, “current_step”: Step.NEEDS_HUMAN }4.2 构建图并定义复杂的路由逻辑接下来我们用StateGraph将这些节点连接起来并设置条件路由。from langgraph.graph import StateGraph, END # 初始化图 workflow StateGraph(BookingState) # 添加所有节点 workflow.add_node(“welcome”, welcome_node) workflow.add_node(“classify_intent”, classify_intent_node) workflow.add_node(“collect_info”, collect_info_node) workflow.add_node(“confirm”, confirm_node) workflow.add_node(“finalize”, finalize_node) workflow.add_node(“human_agent”, human_agent_node) # 设置入口点 workflow.set_entry_point(“welcome”) # 1. 从 welcome 无条件到 classify_intent workflow.add_edge(“welcome”, “classify_intent”) # 2. 从 classify_intent 根据意图路由 def route_after_intent(state: BookingState): intent state.intent if intent “book”: # 如果是订票进入信息收集环节 return “collect_info” elif intent “cancel”: # 如果是取消我们简化处理直接转人工或另一个取消流程节点 return “human_agent” else: # 查询或其他意图也转人工/特定处理节点 return “human_agent” workflow.add_conditional_edges( “classify_intent”, route_after_intent, {“collect_info”: “collect_info”, “human_agent”: “human_agent”} ) # 3. 从 collect_info 节点出来判断信息是否足够 def route_after_collect(state: BookingState): info state.extracted_info # 简单判断如果已收集到出发地、目的地和日期则认为信息足够进入确认环节 if all(k in info for k in [“departure”, “destination”, “date”]): return “confirm” else: # 信息不足返回自身继续收集这里会产生循环 return “collect_info” workflow.add_conditional_edges( “collect_info”, route_after_collect, {“confirm”: “confirm”, “collect_info”: “collect_info”} ) # 4. 从 confirm 节点出来根据用户最新回复决定下一步 def route_after_confirm(state: BookingState): if not state.confirmation_pending: return END last_msg_content state.messages[-1].content.lower() if “确认” in last_msg_content: return “finalize” elif “修改” in last_msg_content: # 用户要修改清空部分信息回到收集节点 return “collect_info” else: # 其他未识别的回复转人工 return “human_agent” workflow.add_conditional_edges( “confirm”, route_after_confirm, {“finalize”: “finalize”, “collect_info”: “collect_info”, “human_agent”: “human_agent”} ) # 5. 设置最终节点和人工节点的出口 workflow.add_edge(“finalize”, END) workflow.add_edge(“human_agent”, END) # 编译图 app workflow.compile()4.3 运行与调试观察 State 的流动现在让我们运行这个工作流并观察 State 是如何在节点间传递和演变的。from langchain_core.messages import HumanMessage # 初始化状态 initial_state {“messages”: [HumanMessage(content“我想订一张票”)]} # 使用 stream 方式运行可以观察每个步骤后的状态 for event in app.stream(initial_state, stream_mode“values”): node_name list(event.keys())[0] state event[node_name] print(f“\n 节点 [{node_name}] 执行后 ) print(f“当前步骤: {state.current_step}”) print(f“意图: {state.intent}”) print(f“提取信息: {state.extracted_info}”) print(f“最新消息: {state.messages[-1].content if state.messages else ‘None’}”) print(“-” * 50)运行上述代码你会看到类似以下的输出清晰地展示了通讯过程 节点 [welcome] 执行后 当前步骤: start 意图: None 提取信息: {} 最新消息: 您好我是机票预订助手。请问您需要什么帮助例如订票、查询、取消 -------------------------------------------------- 节点 [classify_intent] 执行后 当前步骤: intent_classified 意图: book 提取信息: {} 最新消息: 好的我将为您办理机票预订。请告诉我您的出发城市、目的地和出行日期例如北京到上海明天。 -------------------------------------------------- 节点 [collect_info] 执行后 当前步骤: collecting_info 意图: book 提取信息: {‘departure’: ‘北京’ ‘destination’: ‘上海’ ‘date’: ‘2023-10-27’} 最新消息: 好的正在为您查找航班。已识别信息{‘departure’: ‘北京’ ‘destination’: ‘上海’ ‘date’: ‘2023-10-27’}。请问出行人数是 -------------------------------------------------- ...通过这个流式输出你可以像看日志一样追踪BookingState对象中每个字段是如何随着每个节点的执行而变化的。intent在classify_intent节点被赋值extracted_info在collect_info节点被填充current_step驱动着路由逻辑。这就是 LangGraph 通讯机制的生动体现State 是共享的上下文节点是修改者边是路由控制器三者协同完成复杂的工作流。5. 高级通讯模式与最佳实践掌握了基础机制后我们来看看更高级的用法和实践中容易踩的坑。5.1 子图Subgraphs与模块化对于复杂系统你可以将一部分节点和边打包成一个子图。子图本身也是一个可编译、可运行的图它可以作为父图的一个节点。这是实现模块化和复用的关键。from langgraph.graph import StateGraph def create_booking_subgraph(): “”“创建一个负责订票核心流程的子图”“” subgraph StateGraph(BookingState) subgraph.add_node(“collect_info”, collect_info_node) subgraph.add_node(“confirm”, confirm_node) subgraph.add_node(“finalize”, finalize_node) subgraph.set_entry_point(“collect_info”) subgraph.add_conditional_edges(“collect_info”, route_after_collect) subgraph.add_conditional_edges(“confirm”, route_after_confirm) subgraph.add_edge(“finalize”, END) return subgraph.compile() # 在主图中将子图作为一个节点添加 main_workflow StateGraph(BookingState) main_workflow.add_node(“welcome”, welcome_node) main_workflow.add_node(“classify”, classify_intent_node) main_workflow.add_node(“booking_workflow”, create_booking_subgraph()) # 子图作为节点 main_workflow.add_node(“human”, human_agent_node) # … 设置主图的路由逻辑例如 classify 后路由到 booking_workflow 节点通讯意义子图内部的节点通过子图自己的 State 进行通讯对外部而言子图是一个黑盒。主图只关心传递给子图什么初始状态以及接收子图返回的最终状态更新。这极大地简化了复杂工作流的管理和调试。5.2 中断Interruption与长期记忆LangGraph 支持“中断”机制允许工作流在特定节点暂停等待外部输入如用户反馈、异步API回调然后再继续。这通常通过结合pregelLangGraph的底层执行引擎的checkpoint功能和外部存储来实现。其通讯本质是将当前的完整State序列化后保存到数据库如Redis。当外部事件触发继续时从数据库加载该 State并从中断的节点继续执行。这为实现需要等待人类审核或长时间运行任务的智能体提供了可能。5.3 避坑指南与实操心得State 设计要前瞻在项目初期花时间设计一个好的 State 模型。字段名要清晰类型要明确。避免后期不断添加字段导致结构混乱。考虑使用Optional类型来处理字段的渐进式填充。路由函数要保持纯净路由函数应该只读取 State 并返回下一个节点名。不要在路由函数里修改 State。修改 State 是节点的职责。混用职责会导致状态变更难以追踪和调试。善用stream模式进行调试如上文所示app.stream()是调试 LangGraph 应用的神器。它能让你可视化工作流的执行路径和 State 的每一次变化快速定位逻辑错误或路由问题。注意消息列表的合并由于使用了add_messages缩减器你返回的{“messages”: [new_message]}会被追加。如果你需要替换或修改某条特定历史消息需要更精细的操作比如在 State 中维护一个独立的message_history列表和current_prompt字段。子图入口和出口的 State 映射当子图作为节点时要清楚主图的 State 如何映射到子图所需的 State。有时可能需要一个适配器函数来提取或转换字段。错误处理节点函数可能抛出异常。LangGraph 本身不提供复杂的错误处理节点。一种模式是在关键节点外包裹try…except并在出错时返回一个如{“error”: str(e), “current_step”: “error”}的更新然后通过条件边路由到一个专门的“错误处理”节点。理解 LangGraph 的通讯机制就是理解其以State 为中心的数据流和以条件边为驱动的控制流。这不同于传统的函数调用链它提供了一种声明式的、图形化的方式来构建复杂的、有状态的应用程序。从简单的线性流程到带有循环、分支和中断的复杂智能体LangGraph 的这套机制都能提供清晰、可维护的建模方式。