如果你正在尝试构建一个能自主决策、协作完成复杂任务的AI智能体系统那么你很可能已经听说过LangChain并体验过其强大但有时略显“笨重”的链式编程。当你试图让多个智能体协同工作处理需要状态记忆、条件分支和循环的任务时传统的链式思维很快就会遇到瓶颈状态管理混乱、流程控制复杂、调试困难。这正是LangGraph要解决的核心痛点。LangGraph并非LangChain的替代品而是其官方推出的、专门用于构建有状态、多智能体工作流的框架。它把智能体协作的复杂逻辑从“写一堆if-else和回调函数”的泥潭中解放出来让你能用图Graph这种直观的方式来定义智能体的交互流程。简单来说LangGraph让你像画流程图一样设计和运行你的多智能体系统。这篇文章将为你彻底拆解LangGraph。我们不只讲“节点”和“边”是什么而是要回答几个更关键的问题为什么在LangChain之后还需要LangGraph它到底解决了哪些LangChain不擅长的问题一个合格的多智能体架构应该具备哪些核心能力以及如何从零开始用LangGraph构建一个能真正跑起来的、解决实际问题的多智能体应用读完本文你将能清晰理解LangGraph的架构思想掌握其核心组件的用法并亲手搭建一个具备任务规划、工具调用和状态记忆的协作型智能体系统。更重要的是你会明白如何避开从单体智能体到多智能体架构升级过程中的那些常见“坑”。1. 这篇文章真正要解决的问题从“链”到“图”的思维跃迁在深入代码之前我们必须先厘清一个根本性问题当我们在谈论“多智能体”时我们在谈论什么以及为什么传统的工具链会力不从心1.1 单体智能体的局限性一个基于LangChain构建的典型智能体其工作模式往往是线性的接收用户输入 - 思考调用LLM- 决定行动调用工具- 观察结果 - 再思考 - 循环直到结束。这个过程通过AgentExecutor来驱动。对于单一、目标明确的任务这很有效。但一旦任务变得复杂需要分工协作问题就来了状态共享困难智能体A计算出的中间结果如何优雅地传递给智能体B使用全局变量那会很快变成一团乱麻。流程控制生硬要实现“如果步骤1成功则执行步骤2A否则执行步骤2B”这样的条件逻辑你需要在代码里写大量的条件判断破坏了智能体本身的抽象。协作与竞争如何让两个智能体就一个方案进行“辩论”如何让一个管理智能体给多个执行智能体分配任务这些动态交互用线性的“链”很难描述。1.2 LangGraph的核心价值将流程可视化、状态中心化LangGraph引入了“图”的概念。在图论中图由节点Nodes和边Edges组成。节点代表一个执行单元。它可以是一个简单的函数、一个LangChain工具调用、一个完整的LLM调用链甚至是一个子图嵌套的工作流。每个智能体都可以被建模为一个或多个节点。边定义了节点之间的执行流向。它决定了当前节点执行完后下一个该执行谁。边的方向可以由条件逻辑Conditional Edge动态决定这就实现了分支和循环。状态State这是LangGraph的灵魂。整个图共享一个中心化的状态对象通常是一个字典或Pydantic模型。每个节点读取状态、修改状态边的条件判断也基于状态。这彻底解决了状态共享的难题。简单比喻如果把构建单体智能体比作编写一个线性的剧本那么用LangGraph构建多智能体系统就像是在设计一个游戏的任务流程图。你可以清晰地看到每个角色节点在什么条件下边执行什么动作并且所有角色的信息都记录在一张公共的“任务白板”状态上。本文要解决的正是如何将你对智能体的理解从“线性剧本思维”升级到“流程图思维”并利用LangGraph这个工具将想法快速实现为可运行、可调试的复杂系统。2. 基础概念与核心原理拆解要玩转LangGraph必须吃透三个核心概念状态State、节点Node、边Edge。它们共同构成了LangGraph工作流的基石。2.1 状态State系统的共享记忆状态是一个贯穿整个工作流执行周期的数据容器。在LangGraph中我们通常使用TypedDict或Pydantic的BaseModel来定义状态的模式Schema这能提供良好的类型提示和验证。一个典型的多智能体状态可能包含messages: 对话消息列表记录用户输入、智能体回复、工具执行结果等。sender: 标识当前消息或下一步该由哪个智能体处理。next: 指示下一步应该执行哪个节点通常由边来设置。intermediate_steps: 存储工具调用的中间结果。任何你自定义的字段如task_plan,results,current_agent等。状态的妙处在于它是可变的但变更被限制在节点的执行过程中。LangGraph内部会管理状态的版本这对于调试和理解执行流程至关重要。2.2 节点Node执行单元节点是一个可调用对象函数它接收当前State作为参数并返回一个包含对State所做更新的字典。from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from typing_extensions import TypedDict import operator # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[list, operator.add] # 关键使用Annotated和operator.add来声明这是一个追加操作的列表 next: str # 2. 定义一个节点函数 def node_agent_a(state: AgentState) - dict: 智能体A的节点处理某种特定任务 # 从状态中获取信息 latest_message state[“messages”][-1] # 这里模拟智能体A的逻辑例如调用LLM或工具 response f“Agent A processed: {latest_message}” # 更新状态追加消息并指定下一步例如交给智能体B return {“messages”: [response], “next”: “agent_b”}注意Annotated[list, operator.add]的用法这是LangGraph的一个关键语法糖它告诉框架对这个字段messages的更新是追加append而不是覆盖。这对于管理对话历史这样的列表数据非常方便。2.3 边Edge流程的导航规则边决定了工作流的走向。LangGraph提供了几种边普通边Normal Edge无条件地从一个节点指向另一个节点。条件边Conditional Edge根据状态的某个值动态决定下一个节点。这是实现分支逻辑的核心。入口点Entry Point工作流的起点。结束点END工作流的终点。条件边的使用示例from langgraph.graph import StateGraph, END from langgraph.graph import START # 假设我们有一个判断任务类型的函数 def route_task(state: AgentState) - str: 根据最新消息内容路由到不同的处理节点 last_msg state[“messages”][-1].content.lower() if “translate” in last_msg: return “translation_agent” elif “calculate” in last_msg: return “calculation_agent” else: return “general_agent” # 在构建图时使用条件边 builder StateGraph(AgentState) builder.add_node(“translation_agent”, translation_node) builder.add_node(“calculation_agent”, calculation_node) builder.add_node(“general_agent”, general_node) # 设置条件路由从START开始根据route_task函数的返回值决定去哪个节点 builder.add_conditional_edges( START, route_task, # 路由函数 { “translation_agent”: “translation_agent”, “calculation_agent”: “calculation_agent”, “general_agent”: “general_agent”, } ) # 然后可以继续添加这些节点到其他节点或END的边通过组合节点和条件边你可以构建出任意复杂的、带循环和分支的工作流这正是多智能体系统所需要的。3. 环境准备与前置条件在开始实战之前请确保你的开发环境已就绪。本文将使用Python进行演示。3.1 基础环境要求Python版本建议使用 Python 3.10 或更高版本。LangGraph对较新的Python版本支持更好。包管理工具使用pip或poetry等均可。3.2 安装核心库打开终端执行以下命令安装必要的包# 安装 LangGraph 和 LangChain 核心库 pip install langgraph langchain langchain-core # 安装一个LLM提供者这里以OpenAI为例你需要有自己的API Key pip install langchain-openai # 可选但推荐用于结构化输出和状态定义 pip install pydantic # 可选用于更清晰的类型提示如果你使用Python 3.9 pip install typing-extensions3.3 配置API密钥为了调用大模型如OpenAI的GPT你需要设置API密钥。切勿将密钥硬编码在代码中提交到版本库。推荐使用环境变量管理# 在Linux/macOS的终端中 export OPENAI_API_KEY‘your-api-key-here’ # 在Windows的PowerShell中 $env:OPENAI_API_KEY‘your-api-key-here’或者在Python代码中临时设置仅用于测试import os os.environ[“OPENAI_API_KEY”] “your-api-key-here”3.4 IDE建议任何你熟悉的Python IDE或编辑器均可如VS Code、PyCharm。确保其支持Python语言服务和虚拟环境管理。4. 核心流程拆解构建你的第一个LangGraph智能体让我们从一个最简单的例子开始构建一个包含两个“智能体”的协作系统。智能体A负责生成任务计划智能体B负责执行该计划中的一项具体任务比如查询天气。我们将分步拆解这个过程。4.1 第一步定义共享状态首先我们需要定义整个工作流共享的数据结构。from typing import TypedDict, List, Annotated, Union from langchain_core.messages import AnyMessage, HumanMessage, AIMessage, ToolMessage import operator class MultiAgentState(TypedDict): 多智能体协作的共享状态 # 消息历史所有智能体和用户的对话、工具调用结果都存储在这里。 # operator.add 表示对这个字段的更新是追加列表合并而非覆盖。 messages: Annotated[List[AnyMessage], operator.add] # 当前任务计划由规划智能体生成 plan: Union[str, None] # 最近一次工具调用的结果 last_tool_output: Union[str, None]这里我们使用了TypedDict和Annotated。messages字段是所有节点通信的媒介。plan和last_tool_output是我们为这个特定工作流自定义的字段。4.2 第二步创建智能体节点接下来我们创建两个节点函数分别代表规划智能体和执行智能体。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 初始化LLM llm ChatOpenAI(model“gpt-3.5-turbo”) # 1. 规划智能体节点 def planning_agent_node(state: MultiAgentState) - dict: 分析用户请求生成任务计划。 # 从状态中获取最新的用户消息 user_input state[“messages”][-1].content # 构建规划提示词 planner_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个任务规划专家。请将用户的复杂请求分解成一个清晰的、可执行的步骤计划。只输出计划本身。”), (“human”, “用户请求是{input}”) ]) planner_chain planner_prompt | llm | StrOutputParser() # 生成计划 plan planner_chain.invoke({“input”: user_input}) # 更新状态存储计划并添加一条AI消息告知计划已生成 return { “plan”: plan, “messages”: [AIMessage(contentf“我已制定计划{plan}”)] } # 2. 执行智能体节点假设它有一个查询天气的工具 from langchain.tools import tool tool def get_weather(city: str) - str: 获取指定城市的天气信息。这是一个模拟工具。 # 在实际应用中这里会调用真实的天气API weather_data { “北京”: “晴15°C”, “上海”: “多云18°C”, “深圳”: “阵雨22°C” } return weather_data.get(city, f“未找到{city}的天气信息”) # 给执行智能体装配工具 from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate executor_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个任务执行者。根据给定的计划步骤调用合适的工具来完成任务。如果你需要的信息不全请询问用户。”), (“placeholder”, “{messages}”) # LangChain会自动将历史消息放入这里 ]) executor_agent create_tool_calling_agent(llm, [get_weather], executor_prompt) executor_agent_executor AgentExecutor(agentexecutor_agent, tools[get_weather], handle_parsing_errorsTrue) def execution_agent_node(state: MultiAgentState) - dict: 执行规划中的具体任务例如查询天气。 # 这里我们简化逻辑假设规划中提到了城市就直接查询 # 在实际中你可能需要从plan或messages中解析出城市名 plan state.get(“plan”, “”) # 简单演示如果计划中提到“北京”就查询北京天气 tool_output “” if “北京” in plan: tool_output get_weather.invoke(“北京”) # 更新状态记录工具输出并添加一条工具消息 return { “last_tool_output”: tool_output, “messages”: [ToolMessage(contenttool_output, tool_call_id“weather_call_1”)] }在这个例子中planning_agent_node使用LLM生成计划execution_agent_node则调用一个模拟的天气工具。注意我们使用了LangChain的create_tool_calling_agent来快速创建一个能调用工具的智能体执行器。4.3 第三步构建图并定义流程现在我们将节点和边组装起来。from langgraph.graph import StateGraph, START, END # 创建图构建器并传入我们定义的状态类型 workflow StateGraph(MultiAgentState) # 添加节点 workflow.add_node(“planner”, planning_agent_node) workflow.add_node(“executor”, execution_agent_node) # 设置流程从START开始先到规划节点 workflow.add_edge(START, “planner”) # 规划完成后自动进入执行节点 workflow.add_edge(“planner”, “executor”) # 执行完成后结束工作流 workflow.add_edge(“executor”, END) # 编译图得到可执行的应用 app workflow.compile()至此一个最简单的线性两阶段工作流就定义好了用户输入 - 规划 - 执行 - 结束。4.4 第四步运行与可视化让我们运行它并查看其内部结构。# 准备初始输入 initial_state { “messages”: [HumanMessage(content“帮我规划一下明天的出行并看看北京的天气怎么样”)], “plan”: None, “last_tool_output”: None } # 运行工作流 final_state app.invoke(initial_state) print(“ 最终状态 ) print(f“生成的计划{final_state[‘plan’]}”) print(f“工具输出{final_state[‘last_tool_output’]}”) print(“\n 完整消息历史 ) for msg in final_state[“messages”]: print(f“{msg.type}: {msg.content}”)要可视化这个图LangGraph提供了便捷的方法需要安装graphviz# 将图导出为PNG图片可选 try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: # 如果没有IPython环境可以打印文本表示 print(app.get_graph().draw_mermaid())这个简单的例子展示了LangGraph的基础用法。但真正的多智能体系统远不止于此。接下来我们将构建一个更贴近实战的、具备动态路由和循环能力的系统。5. 完整示例一个具备动态路由的协作型多智能体系统现在我们来构建一个更复杂的系统包含三个智能体一个主管Supervisor、一个规划师Planner和一个执行者Executor。流程如下用户提出一个复杂请求如“我想去旅游请帮我做一份包含天气查询和景点推荐的计划”。主管接收请求并判断是否需要规划师介入。如果需要则将任务交给规划师。规划师生成详细的任务步骤列表交还给主管。主管根据规划步骤依次将每个步骤派发给执行者。执行者调用相应的工具天气、搜索等完成任务将结果返回给主管。主管收集所有结果整合后回复用户。如果规划中有多个步骤则循环执行第4、5步。这个流程体现了条件路由主管决定是否调用规划师和循环主管循环派发多个任务。5.1 定义增强版状态from typing import TypedDict, List, Annotated, Union, Literal from langchain_core.messages import AnyMessage import operator class CollaborativeState(TypedDict): 协作智能体系统的状态 messages: Annotated[List[AnyMessage], operator.add] # 当前应该由哪个角色来处理‘supervisor‘, ‘planner‘, ‘executor‘, or ‘end‘ next: str # 规划师生成的任务列表 task_list: List[str] # 当前正在执行的任务索引 current_task_index: int # 所有任务的执行结果汇总 all_results: List[str]5.2 实现三个智能体节点from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser, JsonOutputParser from pydantic import BaseModel, Field import json llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0) # --- 工具定义 --- tool def search_web(query: str) - str: 模拟网络搜索工具。 # 模拟返回结果 return f“关于‘{query}‘的搜索结果这是一个模拟的搜索结果摘要。” tool def get_weather(city: str) - str: 模拟天气查询工具。 weather_map {“北京”: “晴10-20°C”, “上海”: “多云15-22°C”, “杭州”: “小雨12-18°C”} return weather_map.get(city, “天气信息暂不可用”) tools [search_web, get_weather] # --- 主管节点 --- def supervisor_node(state: CollaborativeState) - dict: 主管决定下一步由谁处理。 last_message state[“messages”][-1] # 如果是用户初始输入或者执行者返回了结果需要判断下一步 if last_message.type “human” or (last_message.type “tool” and state[“next”] “executor”): # 这里简化逻辑如果还没有任务列表就去找规划师否则派发给执行者。 if not state.get(“task_list”): next_agent “planner” msg_content “主管任务需要规划已转交规划师。” else: # 还有任务未执行 if state[“current_task_index”] len(state[“task_list”]): next_agent “executor” msg_content f“主管正在派发任务 ‘{state[‘task_list’][state[‘current_task_index’]]}‘ 给执行者。” else: # 所有任务完成准备结束 next_agent “end” msg_content “主管所有任务已完成正在生成最终报告。” else: # 其他情况保持当前流向例如规划师刚回复 next_agent state[“next”] msg_content “主管流程继续。” from langchain_core.messages import AIMessage return { “next”: next_agent, “messages”: [AIMessage(contentmsg_content)] } # --- 规划师节点 --- class Plan(BaseModel): tasks: List[str] Field(description“分解后的具体任务列表”) def planner_node(state: CollaborativeState) - dict: 规划师将复杂请求分解为任务列表。 user_request state[“messages”][0].content # 获取初始用户请求 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个任务分解专家。请将用户的复杂请求分解成一个清晰、有序、可执行的任务列表。每个任务应该是一个简单的句子描述要做什么。输出格式必须是JSON包含一个‘tasks‘数组。”), (“human”, “用户请求{request}”) ]) # 使用Pydantic解析器来获得结构化的输出 parser JsonOutputParser(pydantic_objectPlan) chain prompt | llm | parser try: plan_result: Plan chain.invoke({“request”: user_request}) task_list plan_result.tasks except: # 如果解析失败使用备用方案 task_list [“1. 理解用户请求”, “2. 执行必要查询”, “3. 汇总结果”] from langchain_core.messages import AIMessage return { “task_list”: task_list, “current_task_index”: 0, # 重置索引 “next”: “supervisor”, # 完成后交回主管 “messages”: [AIMessage(contentf“规划师我已制定计划共{len(task_list)}个任务{‘ ‘.join(task_list)}”)] } # --- 执行者节点 --- from langchain.agents import create_tool_calling_agent, AgentExecutor executor_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个执行者负责调用工具完成具体任务。根据当前任务描述判断并调用最合适的工具。一次只完成一个任务。”), (“placeholder”, “{messages}”) ]) executor_agent create_tool_calling_agent(llm, tools, executor_prompt) executor_agent_executor AgentExecutor(agentexecutor_agent, toolstools, verboseFalse, handle_parsing_errorsTrue) def executor_node(state: CollaborativeState) - dict: 执行者调用工具完成当前任务。 current_index state[“current_task_index”] task_list state[“task_list”] if current_index len(task_list): # 没有任务了 return {“next”: “supervisor”, “messages”: [AIMessage(content“执行者无任务可执行。”)]} current_task task_list[current_index] # 让智能体执行器来处理这个任务 # 我们需要将任务描述作为新消息传入 from langchain_core.messages import HumanMessage new_messages state[“messages”] [HumanMessage(contentf“请执行以下任务{current_task}”)] try: result executor_agent_executor.invoke({“messages”: new_messages}) tool_output result[“output”] except Exception as e: tool_output f“工具执行出错{str(e)}” # 更新状态任务索引1收集结果下一步交回主管 from langchain_core.messages import ToolMessage return { “current_task_index”: current_index 1, “all_results”: state[“all_results”] [tool_output], “next”: “supervisor”, “messages”: [ToolMessage(contenttool_output, tool_call_idf“task_{current_index}”)] }5.3 构建带条件边和循环的图这是最关键的一步我们将使用条件边来实现动态路由。from langgraph.graph import StateGraph, START, END # 初始化图 builder StateGraph(CollaborativeState) # 添加节点 builder.add_node(“supervisor”, supervisor_node) builder.add_node(“planner”, planner_node) builder.add_node(“executor”, executor_node) # 设置入口点 builder.add_edge(START, “supervisor”) # 定义条件路由函数根据状态的next字段决定下一步 def route_after_supervisor(state: CollaborativeState) - str: 主管节点之后的路由逻辑 return state[“next”] # 主管节点完成后根据其设置的next值动态路由 builder.add_conditional_edges( “supervisor”, route_after_supervisor, { “planner”: “planner”, “executor”: “executor”, “end”: END, # 如果next是‘end‘则结束流程 “supervisor”: “supervisor”, # 也可以指向自己形成循环判断 } ) # 规划师和执行者完成后都固定返回主管节点由主管进行下一轮调度 builder.add_edge(“planner”, “supervisor”) builder.add_edge(“executor”, “supervisor”) # 编译图 collaborative_app builder.compile()5.4 运行复杂工作流现在让我们运行这个更强大的系统。# 初始化状态 initial_state { “messages”: [HumanMessage(content“我想周末去杭州玩请帮我查一下杭州的天气并推荐两个必去的景点。”)], “next”: “supervisor”, “task_list”: [], “current_task_index”: 0, “all_results”: [] } # 运行应用。设置recursion_limit以防止无限循环对于复杂循环是必要的 from langgraph.checkpoint import MemorySaver from langgraph.graph import MessagesState # 为了更好的演示我们可以使用checkpoint来逐步执行 checkpointer MemorySaver() app_with_memory builder.compile(checkpointercheckpointer) # 开始执行 config {“configurable”: {“thread_id”: “test_thread_1”}} final_state None for step in app_with_memory.stream(initial_state, config, stream_mode“values”): node_name list(step.keys())[0] print(f“\n 节点 [{node_name}] 执行完毕 ) print(f“当前 next: {step[node_name].get(‘next’)}”) print(f“当前任务列表: {step[node_name].get(‘task_list’)}”) print(f“当前任务索引: {step[node_name].get(‘current_task_index’)}”) if step[node_name][“messages”]: last_msg step[node_name][“messages”][-1] print(f“最新消息: {last_msg.type}: {last_msg.content[:100]}...”) # 截取前100字符 print(“\n” “”*50) print(“工作流执行结束”) if final_state is None: final_state step[list(step.keys())[-1]] print(f“最终生成的任务列表: {final_state.get(‘task_list’)}”) print(f“收集的所有结果: {final_state.get(‘all_results’)}”)这个例子展示了LangGraph如何优雅地处理多智能体间的复杂协作、条件判断和循环任务。主管节点充当了“调度中心”的角色整个系统的控制流清晰可见。6. 运行结果与效果验证运行上述代码后你应该能看到类似以下的输出具体内容因LLM输出而异 节点 [supervisor] 执行完毕 当前 next: planner 当前任务列表: [] 当前任务索引: 0 最新消息: ai: 主管任务需要规划已转交规划师。... 节点 [planner] 执行完毕 当前 next: supervisor 当前任务列表: [‘查询杭州本周末的天气情况。‘, ‘搜索杭州最受欢迎的两个旅游景点。‘, ‘汇总天气和景点信息给出出行建议。‘] 当前任务索引: 0 最新消息: ai: 规划师我已制定计划共3个任务查询杭州本周末的天气情况。 搜索杭州最受欢迎的两个旅游景点。 汇总天气和景点信息给出出行建议。... 节点 [supervisor] 执行完毕 当前 next: executor 当前任务列表: [‘查询杭州本周末的天气情况。‘, ‘搜索杭州最受欢迎的两个旅游景点。‘, ‘汇总天气和景点信息给出出行建议。‘] 当前任务索引: 0 最新消息: ai: 主管正在派发任务 ‘查询杭州本周末的天气情况。‘ 给执行者。... 节点 [executor] 执行完毕 当前 next: supervisor 当前任务列表: [‘查询杭州本周末的天气情况。‘, ‘搜索杭州最受欢迎的两个旅游景点。‘, ‘汇总天气和景点信息给出出行建议。‘] 当前任务索引: 1 最新消息: tool: 杭州的天气信息小雨12-18°C... ... (后续循环) ... 工作流执行结束 最终生成的任务列表: [‘查询杭州本周末的天气情况。‘, ‘搜索杭州最受欢迎的两个旅游景点。‘, ‘汇总天气和景点信息给出出行建议。‘] 收集的所有结果: [‘杭州的天气信息小雨12-18°C‘, ‘关于‘杭州最受欢迎的两个旅游景点‘的搜索结果1. 西湖 2. 灵隐寺‘, ‘最终报告...‘]如何验证成功流程正确性观察控制台输出确认流程按照“主管-规划师-主管-执行者-主管-执行者-主管-结束”的顺序执行。这验证了条件路由和循环正常工作。状态一致性检查每个节点执行后task_list,current_task_index,all_results等状态字段是否按预期更新。例如current_task_index应该在每次executor完成后递增。任务完成度最终的all_results列表应包含与task_list中任务数量对应的结果证明所有分解出的子任务都被执行了。工具调用确认executor节点成功调用了我们定义的模拟工具get_weather,search_web并返回了模拟结果。如果运行失败首先检查API密钥是否已正确设置OPENAI_API_KEY环境变量。依赖版本langgraph,langchain等库版本是否兼容。建议使用较新的稳定版本。网络问题是否能正常访问OpenAI API如果你使用其他模型则检查对应服务。状态结构确保节点函数返回的字典键名与State定义中的字段名完全匹配。7. 常见问题与排查思路在开发LangGraph多智能体应用时你可能会遇到以下典型问题问题现象可能原因排查方式解决方案KeyError或状态字段未找到1. 节点返回的字典键与State定义不一致。2. 使用了未在State中声明的字段。1. 仔细核对节点函数return的字典键名。2. 检查State的TypedDict定义。确保节点返回的键名与State定义的字段名完全一致。对于列表追加字段使用Annotated[list, operator.add]。图编译错误1. 节点未添加到图中就添加边。2. 使用了未定义的节点名作为边的目标。3.START/END常量拼写错误。1. 检查add_node和add_edge的调用顺序。2. 检查add_edge或add_conditional_edges中引用的节点名是否存在。遵循“先添加节点再添加边”的顺序。使用常量START和END而非字符串“start”。工作流陷入无限循环1. 条件边逻辑有误导致无法到达END。2. 节点未正确更新next字段。1. 在route_after_supervisor这类路由函数中打印日志。2. 使用app.get_graph().draw_mermaid()可视化图结构检查循环路径。1. 确保路由函数在所有可能情况下都返回一个已定义的节点名或END。2. 使用recursion_limit参数限制最大步数app.invoke(..., recursion_limit100)。工具调用失败或LLM调用超时1. API密钥无效或额度不足。2. 网络连接问题。3. 工具函数参数不匹配或抛出异常。1. 先单独测试LLM调用和工具函数。2. 查看LangChain/OpenAI的错误信息。1. 验证API密钥和网络。2. 在工具函数内部做好异常捕获返回明确的错误信息。3. 为AgentExecutor设置handle_parsing_errorsTrue和max_execution_time。状态更新不符合预期如列表被覆盖对标记为Annotated[list, operator.add]的字段执行了赋值操作而非追加。检查节点中是使用return {“messages”: [new_msg]}正确还是return {“messages”: new_list}可能错误。对于追加字段确保返回的是一个列表LangGraph会自动将其与原列表合并。若要替换整个列表需在State中定义另一个字段。Pydantic解析错误1. LLM的输出不符合JsonOutputParser或Pydantic模型定义的格式。2. 模型字段定义与提示词要求不匹配。1. 在解析前打印LLM的原始输出。2. 检查JsonOutputParser的pydantic_object是否正确定义。1. 在提示词中明确要求JSON格式并提供示例。2. 使用handle_parsing_errors参数或在链中增加重试、格式化步骤。可视化图时出错未安装graphviz或pygraphviz。检查错误信息是否提示缺少图形库。安装graphvizpip install pygraphviz可能需要系统级安装Graphviz软件。或者使用app.get_graph().print_ascii()进行文本可视化。8. 最佳实践与工程建议将LangGraph用于实际项目时遵循以下建议可以大幅提升开发效率和系统稳定性8.1 状态设计原则最小化与清晰化只将工作流真正需要共享的数据放入State。避免将临时变量或节点内部状态塞进去。善用Annotated明确每个字段的更新语义。operator.add用于列表追加operator.setitem用于字典更新需从langgraph.graph导入。对于简单覆盖直接定义类型即可如plan: str。使用Pydantic Model对于复杂状态优先使用pydantic.BaseModel代替TypedDict它能提供运行时验证和更丰富的字段类型。8.2 节点设计原则单一职责每个节点应只做一件事。例如一个节点负责调用LLM生成文本另一个节点负责解析文本并更新状态。幂等性与容错假设节点可能被多次执行在错误重试或特定图结构中设计时应尽量保证幂等性。对工具调用和外部API调用做好异常处理。日志与可观测性在节点函数内关键步骤添加日志记录输入、输出和关键决策。这对于调试复杂工作流至关重要。8.3 图结构设计先画图再写码在编码前用纸笔或绘图工具画出工作流的草图明确节点、边和状态流转。这能极大减少逻辑错误。模块化与子图对于非常复杂的工作流利用StateGraph的嵌套能力将相关节点组封装成子图Subgraph使主图结构更清晰。合理使用条件边条件边非常强大但过度使用会使流程难以理解和调试。确保路由函数的逻辑尽可能简单。8.4 生产环境部署持久化检查点使用SqliteSaver或MongoDBSaver等持久化检查点存储而不是MemorySaver。这样可以在应用重启后恢复长时间运行的工作流状态并支持并发执行。超时与限流为LLM调用和工具调用设置合理的超时时间。对于可能被频繁调用的节点考虑增加限流机制。版本控制工作流图的结构也是代码的一部分。当对图进行修改时要有明确的版本管理策略考虑如何平滑迁移正在运行中的旧版本工作流状态。8.5 测试与调试单元测试节点单独测试每个节点函数模拟输入State验证其输出是否符合预期。集成测试工作流编写测试用例针对不同的初始输入验证整个工作流的最终输出和状态。利用LangGraph Studio积极探索LangGraph官方提供的可视化开发工具LangGraph Studio它允许你以低代码方式构建、调试和监控工作流是开发和理解复杂系统的利器。9. 总结与后续学习方向通过本文我们完成了从理解LangGraph为什么出现到掌握其状态、节点、边三大核心概念再到亲手构建从简单到复杂的多智能体工作流的全过程。关键收获在于思维模式的转变用“图”来建模智能体间的协作用“中心化状态”来管理共享信息。LangGraph的强大之处在于它将复杂的控制流逻辑循环、条件分支、并行从杂乱的业务代码中抽离出来变成了清晰、可视化的图定义。这使得多智能体系统的设计、调试和维护难度大大降低。下一步你可以从以下几个方向深入探索深入研究状态管理尝试更复杂的State设计如嵌套的Pydantic模型或使用add_messages等LangGraph内置的Reducer来简化消息列表操作。探索高级图特性并行执行了解如何利用langgraph.graph中的CONCURRENT语义让多个节点同时运行。中断与人工干预学习如何在工作流中设置“中断点”允许人类审核或提供额外输入后再继续。动态图修改研究是否能在运行时根据状态动态添加或移除节点高级用法。集成更强大的工具将LangGraph与数据库、搜索引擎、外部API、代码解释器等更丰富的工具连接起来构建功能更强大的智能体。投入生产学习如何将编译好的app部署为API服务如使用FastAPI并配置持久化的检查点存储如PostgreSQL以处理真实的、长时间运行的业务流程。关注生态LangGraph生态在快速发展关注langgraph-checkpoint、langgraph-elasticsearch等官方或社区扩展它们能提供更企业级的特性。多智能体架构是构建下一代AI应用的关键。LangGraph提供了一个坚实、优雅的底层框架。掌握它意味着你拥有了将多个“AI大脑”组织起来解决复杂现实问题的能力。建议将本文中的示例代码作为起点不断修改和实验逐步搭建出属于你自己的、能够解决实际业务问题的智能体协作系统。
如果你正在尝试构建一个能自主决策、协作完成复杂任务的AI智能体系统那么你很可能已经听说过LangChain并体验过其强大但有时略显“笨重”的链式编程。当你试图让多个智能体协同工作处理需要状态记忆、条件分支和循环的任务时传统的链式思维很快就会遇到瓶颈状态管理混乱、流程控制复杂、调试困难。这正是LangGraph要解决的核心痛点。LangGraph并非LangChain的替代品而是其官方推出的、专门用于构建有状态、多智能体工作流的框架。它把智能体协作的复杂逻辑从“写一堆if-else和回调函数”的泥潭中解放出来让你能用图Graph这种直观的方式来定义智能体的交互流程。简单来说LangGraph让你像画流程图一样设计和运行你的多智能体系统。这篇文章将为你彻底拆解LangGraph。我们不只讲“节点”和“边”是什么而是要回答几个更关键的问题为什么在LangChain之后还需要LangGraph它到底解决了哪些LangChain不擅长的问题一个合格的多智能体架构应该具备哪些核心能力以及如何从零开始用LangGraph构建一个能真正跑起来的、解决实际问题的多智能体应用读完本文你将能清晰理解LangGraph的架构思想掌握其核心组件的用法并亲手搭建一个具备任务规划、工具调用和状态记忆的协作型智能体系统。更重要的是你会明白如何避开从单体智能体到多智能体架构升级过程中的那些常见“坑”。1. 这篇文章真正要解决的问题从“链”到“图”的思维跃迁在深入代码之前我们必须先厘清一个根本性问题当我们在谈论“多智能体”时我们在谈论什么以及为什么传统的工具链会力不从心1.1 单体智能体的局限性一个基于LangChain构建的典型智能体其工作模式往往是线性的接收用户输入 - 思考调用LLM- 决定行动调用工具- 观察结果 - 再思考 - 循环直到结束。这个过程通过AgentExecutor来驱动。对于单一、目标明确的任务这很有效。但一旦任务变得复杂需要分工协作问题就来了状态共享困难智能体A计算出的中间结果如何优雅地传递给智能体B使用全局变量那会很快变成一团乱麻。流程控制生硬要实现“如果步骤1成功则执行步骤2A否则执行步骤2B”这样的条件逻辑你需要在代码里写大量的条件判断破坏了智能体本身的抽象。协作与竞争如何让两个智能体就一个方案进行“辩论”如何让一个管理智能体给多个执行智能体分配任务这些动态交互用线性的“链”很难描述。1.2 LangGraph的核心价值将流程可视化、状态中心化LangGraph引入了“图”的概念。在图论中图由节点Nodes和边Edges组成。节点代表一个执行单元。它可以是一个简单的函数、一个LangChain工具调用、一个完整的LLM调用链甚至是一个子图嵌套的工作流。每个智能体都可以被建模为一个或多个节点。边定义了节点之间的执行流向。它决定了当前节点执行完后下一个该执行谁。边的方向可以由条件逻辑Conditional Edge动态决定这就实现了分支和循环。状态State这是LangGraph的灵魂。整个图共享一个中心化的状态对象通常是一个字典或Pydantic模型。每个节点读取状态、修改状态边的条件判断也基于状态。这彻底解决了状态共享的难题。简单比喻如果把构建单体智能体比作编写一个线性的剧本那么用LangGraph构建多智能体系统就像是在设计一个游戏的任务流程图。你可以清晰地看到每个角色节点在什么条件下边执行什么动作并且所有角色的信息都记录在一张公共的“任务白板”状态上。本文要解决的正是如何将你对智能体的理解从“线性剧本思维”升级到“流程图思维”并利用LangGraph这个工具将想法快速实现为可运行、可调试的复杂系统。2. 基础概念与核心原理拆解要玩转LangGraph必须吃透三个核心概念状态State、节点Node、边Edge。它们共同构成了LangGraph工作流的基石。2.1 状态State系统的共享记忆状态是一个贯穿整个工作流执行周期的数据容器。在LangGraph中我们通常使用TypedDict或Pydantic的BaseModel来定义状态的模式Schema这能提供良好的类型提示和验证。一个典型的多智能体状态可能包含messages: 对话消息列表记录用户输入、智能体回复、工具执行结果等。sender: 标识当前消息或下一步该由哪个智能体处理。next: 指示下一步应该执行哪个节点通常由边来设置。intermediate_steps: 存储工具调用的中间结果。任何你自定义的字段如task_plan,results,current_agent等。状态的妙处在于它是可变的但变更被限制在节点的执行过程中。LangGraph内部会管理状态的版本这对于调试和理解执行流程至关重要。2.2 节点Node执行单元节点是一个可调用对象函数它接收当前State作为参数并返回一个包含对State所做更新的字典。from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from typing_extensions import TypedDict import operator # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[list, operator.add] # 关键使用Annotated和operator.add来声明这是一个追加操作的列表 next: str # 2. 定义一个节点函数 def node_agent_a(state: AgentState) - dict: 智能体A的节点处理某种特定任务 # 从状态中获取信息 latest_message state[“messages”][-1] # 这里模拟智能体A的逻辑例如调用LLM或工具 response f“Agent A processed: {latest_message}” # 更新状态追加消息并指定下一步例如交给智能体B return {“messages”: [response], “next”: “agent_b”}注意Annotated[list, operator.add]的用法这是LangGraph的一个关键语法糖它告诉框架对这个字段messages的更新是追加append而不是覆盖。这对于管理对话历史这样的列表数据非常方便。2.3 边Edge流程的导航规则边决定了工作流的走向。LangGraph提供了几种边普通边Normal Edge无条件地从一个节点指向另一个节点。条件边Conditional Edge根据状态的某个值动态决定下一个节点。这是实现分支逻辑的核心。入口点Entry Point工作流的起点。结束点END工作流的终点。条件边的使用示例from langgraph.graph import StateGraph, END from langgraph.graph import START # 假设我们有一个判断任务类型的函数 def route_task(state: AgentState) - str: 根据最新消息内容路由到不同的处理节点 last_msg state[“messages”][-1].content.lower() if “translate” in last_msg: return “translation_agent” elif “calculate” in last_msg: return “calculation_agent” else: return “general_agent” # 在构建图时使用条件边 builder StateGraph(AgentState) builder.add_node(“translation_agent”, translation_node) builder.add_node(“calculation_agent”, calculation_node) builder.add_node(“general_agent”, general_node) # 设置条件路由从START开始根据route_task函数的返回值决定去哪个节点 builder.add_conditional_edges( START, route_task, # 路由函数 { “translation_agent”: “translation_agent”, “calculation_agent”: “calculation_agent”, “general_agent”: “general_agent”, } ) # 然后可以继续添加这些节点到其他节点或END的边通过组合节点和条件边你可以构建出任意复杂的、带循环和分支的工作流这正是多智能体系统所需要的。3. 环境准备与前置条件在开始实战之前请确保你的开发环境已就绪。本文将使用Python进行演示。3.1 基础环境要求Python版本建议使用 Python 3.10 或更高版本。LangGraph对较新的Python版本支持更好。包管理工具使用pip或poetry等均可。3.2 安装核心库打开终端执行以下命令安装必要的包# 安装 LangGraph 和 LangChain 核心库 pip install langgraph langchain langchain-core # 安装一个LLM提供者这里以OpenAI为例你需要有自己的API Key pip install langchain-openai # 可选但推荐用于结构化输出和状态定义 pip install pydantic # 可选用于更清晰的类型提示如果你使用Python 3.9 pip install typing-extensions3.3 配置API密钥为了调用大模型如OpenAI的GPT你需要设置API密钥。切勿将密钥硬编码在代码中提交到版本库。推荐使用环境变量管理# 在Linux/macOS的终端中 export OPENAI_API_KEY‘your-api-key-here’ # 在Windows的PowerShell中 $env:OPENAI_API_KEY‘your-api-key-here’或者在Python代码中临时设置仅用于测试import os os.environ[“OPENAI_API_KEY”] “your-api-key-here”3.4 IDE建议任何你熟悉的Python IDE或编辑器均可如VS Code、PyCharm。确保其支持Python语言服务和虚拟环境管理。4. 核心流程拆解构建你的第一个LangGraph智能体让我们从一个最简单的例子开始构建一个包含两个“智能体”的协作系统。智能体A负责生成任务计划智能体B负责执行该计划中的一项具体任务比如查询天气。我们将分步拆解这个过程。4.1 第一步定义共享状态首先我们需要定义整个工作流共享的数据结构。from typing import TypedDict, List, Annotated, Union from langchain_core.messages import AnyMessage, HumanMessage, AIMessage, ToolMessage import operator class MultiAgentState(TypedDict): 多智能体协作的共享状态 # 消息历史所有智能体和用户的对话、工具调用结果都存储在这里。 # operator.add 表示对这个字段的更新是追加列表合并而非覆盖。 messages: Annotated[List[AnyMessage], operator.add] # 当前任务计划由规划智能体生成 plan: Union[str, None] # 最近一次工具调用的结果 last_tool_output: Union[str, None]这里我们使用了TypedDict和Annotated。messages字段是所有节点通信的媒介。plan和last_tool_output是我们为这个特定工作流自定义的字段。4.2 第二步创建智能体节点接下来我们创建两个节点函数分别代表规划智能体和执行智能体。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 初始化LLM llm ChatOpenAI(model“gpt-3.5-turbo”) # 1. 规划智能体节点 def planning_agent_node(state: MultiAgentState) - dict: 分析用户请求生成任务计划。 # 从状态中获取最新的用户消息 user_input state[“messages”][-1].content # 构建规划提示词 planner_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个任务规划专家。请将用户的复杂请求分解成一个清晰的、可执行的步骤计划。只输出计划本身。”), (“human”, “用户请求是{input}”) ]) planner_chain planner_prompt | llm | StrOutputParser() # 生成计划 plan planner_chain.invoke({“input”: user_input}) # 更新状态存储计划并添加一条AI消息告知计划已生成 return { “plan”: plan, “messages”: [AIMessage(contentf“我已制定计划{plan}”)] } # 2. 执行智能体节点假设它有一个查询天气的工具 from langchain.tools import tool tool def get_weather(city: str) - str: 获取指定城市的天气信息。这是一个模拟工具。 # 在实际应用中这里会调用真实的天气API weather_data { “北京”: “晴15°C”, “上海”: “多云18°C”, “深圳”: “阵雨22°C” } return weather_data.get(city, f“未找到{city}的天气信息”) # 给执行智能体装配工具 from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate executor_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个任务执行者。根据给定的计划步骤调用合适的工具来完成任务。如果你需要的信息不全请询问用户。”), (“placeholder”, “{messages}”) # LangChain会自动将历史消息放入这里 ]) executor_agent create_tool_calling_agent(llm, [get_weather], executor_prompt) executor_agent_executor AgentExecutor(agentexecutor_agent, tools[get_weather], handle_parsing_errorsTrue) def execution_agent_node(state: MultiAgentState) - dict: 执行规划中的具体任务例如查询天气。 # 这里我们简化逻辑假设规划中提到了城市就直接查询 # 在实际中你可能需要从plan或messages中解析出城市名 plan state.get(“plan”, “”) # 简单演示如果计划中提到“北京”就查询北京天气 tool_output “” if “北京” in plan: tool_output get_weather.invoke(“北京”) # 更新状态记录工具输出并添加一条工具消息 return { “last_tool_output”: tool_output, “messages”: [ToolMessage(contenttool_output, tool_call_id“weather_call_1”)] }在这个例子中planning_agent_node使用LLM生成计划execution_agent_node则调用一个模拟的天气工具。注意我们使用了LangChain的create_tool_calling_agent来快速创建一个能调用工具的智能体执行器。4.3 第三步构建图并定义流程现在我们将节点和边组装起来。from langgraph.graph import StateGraph, START, END # 创建图构建器并传入我们定义的状态类型 workflow StateGraph(MultiAgentState) # 添加节点 workflow.add_node(“planner”, planning_agent_node) workflow.add_node(“executor”, execution_agent_node) # 设置流程从START开始先到规划节点 workflow.add_edge(START, “planner”) # 规划完成后自动进入执行节点 workflow.add_edge(“planner”, “executor”) # 执行完成后结束工作流 workflow.add_edge(“executor”, END) # 编译图得到可执行的应用 app workflow.compile()至此一个最简单的线性两阶段工作流就定义好了用户输入 - 规划 - 执行 - 结束。4.4 第四步运行与可视化让我们运行它并查看其内部结构。# 准备初始输入 initial_state { “messages”: [HumanMessage(content“帮我规划一下明天的出行并看看北京的天气怎么样”)], “plan”: None, “last_tool_output”: None } # 运行工作流 final_state app.invoke(initial_state) print(“ 最终状态 ) print(f“生成的计划{final_state[‘plan’]}”) print(f“工具输出{final_state[‘last_tool_output’]}”) print(“\n 完整消息历史 ) for msg in final_state[“messages”]: print(f“{msg.type}: {msg.content}”)要可视化这个图LangGraph提供了便捷的方法需要安装graphviz# 将图导出为PNG图片可选 try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: # 如果没有IPython环境可以打印文本表示 print(app.get_graph().draw_mermaid())这个简单的例子展示了LangGraph的基础用法。但真正的多智能体系统远不止于此。接下来我们将构建一个更贴近实战的、具备动态路由和循环能力的系统。5. 完整示例一个具备动态路由的协作型多智能体系统现在我们来构建一个更复杂的系统包含三个智能体一个主管Supervisor、一个规划师Planner和一个执行者Executor。流程如下用户提出一个复杂请求如“我想去旅游请帮我做一份包含天气查询和景点推荐的计划”。主管接收请求并判断是否需要规划师介入。如果需要则将任务交给规划师。规划师生成详细的任务步骤列表交还给主管。主管根据规划步骤依次将每个步骤派发给执行者。执行者调用相应的工具天气、搜索等完成任务将结果返回给主管。主管收集所有结果整合后回复用户。如果规划中有多个步骤则循环执行第4、5步。这个流程体现了条件路由主管决定是否调用规划师和循环主管循环派发多个任务。5.1 定义增强版状态from typing import TypedDict, List, Annotated, Union, Literal from langchain_core.messages import AnyMessage import operator class CollaborativeState(TypedDict): 协作智能体系统的状态 messages: Annotated[List[AnyMessage], operator.add] # 当前应该由哪个角色来处理‘supervisor‘, ‘planner‘, ‘executor‘, or ‘end‘ next: str # 规划师生成的任务列表 task_list: List[str] # 当前正在执行的任务索引 current_task_index: int # 所有任务的执行结果汇总 all_results: List[str]5.2 实现三个智能体节点from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser, JsonOutputParser from pydantic import BaseModel, Field import json llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0) # --- 工具定义 --- tool def search_web(query: str) - str: 模拟网络搜索工具。 # 模拟返回结果 return f“关于‘{query}‘的搜索结果这是一个模拟的搜索结果摘要。” tool def get_weather(city: str) - str: 模拟天气查询工具。 weather_map {“北京”: “晴10-20°C”, “上海”: “多云15-22°C”, “杭州”: “小雨12-18°C”} return weather_map.get(city, “天气信息暂不可用”) tools [search_web, get_weather] # --- 主管节点 --- def supervisor_node(state: CollaborativeState) - dict: 主管决定下一步由谁处理。 last_message state[“messages”][-1] # 如果是用户初始输入或者执行者返回了结果需要判断下一步 if last_message.type “human” or (last_message.type “tool” and state[“next”] “executor”): # 这里简化逻辑如果还没有任务列表就去找规划师否则派发给执行者。 if not state.get(“task_list”): next_agent “planner” msg_content “主管任务需要规划已转交规划师。” else: # 还有任务未执行 if state[“current_task_index”] len(state[“task_list”]): next_agent “executor” msg_content f“主管正在派发任务 ‘{state[‘task_list’][state[‘current_task_index’]]}‘ 给执行者。” else: # 所有任务完成准备结束 next_agent “end” msg_content “主管所有任务已完成正在生成最终报告。” else: # 其他情况保持当前流向例如规划师刚回复 next_agent state[“next”] msg_content “主管流程继续。” from langchain_core.messages import AIMessage return { “next”: next_agent, “messages”: [AIMessage(contentmsg_content)] } # --- 规划师节点 --- class Plan(BaseModel): tasks: List[str] Field(description“分解后的具体任务列表”) def planner_node(state: CollaborativeState) - dict: 规划师将复杂请求分解为任务列表。 user_request state[“messages”][0].content # 获取初始用户请求 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个任务分解专家。请将用户的复杂请求分解成一个清晰、有序、可执行的任务列表。每个任务应该是一个简单的句子描述要做什么。输出格式必须是JSON包含一个‘tasks‘数组。”), (“human”, “用户请求{request}”) ]) # 使用Pydantic解析器来获得结构化的输出 parser JsonOutputParser(pydantic_objectPlan) chain prompt | llm | parser try: plan_result: Plan chain.invoke({“request”: user_request}) task_list plan_result.tasks except: # 如果解析失败使用备用方案 task_list [“1. 理解用户请求”, “2. 执行必要查询”, “3. 汇总结果”] from langchain_core.messages import AIMessage return { “task_list”: task_list, “current_task_index”: 0, # 重置索引 “next”: “supervisor”, # 完成后交回主管 “messages”: [AIMessage(contentf“规划师我已制定计划共{len(task_list)}个任务{‘ ‘.join(task_list)}”)] } # --- 执行者节点 --- from langchain.agents import create_tool_calling_agent, AgentExecutor executor_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个执行者负责调用工具完成具体任务。根据当前任务描述判断并调用最合适的工具。一次只完成一个任务。”), (“placeholder”, “{messages}”) ]) executor_agent create_tool_calling_agent(llm, tools, executor_prompt) executor_agent_executor AgentExecutor(agentexecutor_agent, toolstools, verboseFalse, handle_parsing_errorsTrue) def executor_node(state: CollaborativeState) - dict: 执行者调用工具完成当前任务。 current_index state[“current_task_index”] task_list state[“task_list”] if current_index len(task_list): # 没有任务了 return {“next”: “supervisor”, “messages”: [AIMessage(content“执行者无任务可执行。”)]} current_task task_list[current_index] # 让智能体执行器来处理这个任务 # 我们需要将任务描述作为新消息传入 from langchain_core.messages import HumanMessage new_messages state[“messages”] [HumanMessage(contentf“请执行以下任务{current_task}”)] try: result executor_agent_executor.invoke({“messages”: new_messages}) tool_output result[“output”] except Exception as e: tool_output f“工具执行出错{str(e)}” # 更新状态任务索引1收集结果下一步交回主管 from langchain_core.messages import ToolMessage return { “current_task_index”: current_index 1, “all_results”: state[“all_results”] [tool_output], “next”: “supervisor”, “messages”: [ToolMessage(contenttool_output, tool_call_idf“task_{current_index}”)] }5.3 构建带条件边和循环的图这是最关键的一步我们将使用条件边来实现动态路由。from langgraph.graph import StateGraph, START, END # 初始化图 builder StateGraph(CollaborativeState) # 添加节点 builder.add_node(“supervisor”, supervisor_node) builder.add_node(“planner”, planner_node) builder.add_node(“executor”, executor_node) # 设置入口点 builder.add_edge(START, “supervisor”) # 定义条件路由函数根据状态的next字段决定下一步 def route_after_supervisor(state: CollaborativeState) - str: 主管节点之后的路由逻辑 return state[“next”] # 主管节点完成后根据其设置的next值动态路由 builder.add_conditional_edges( “supervisor”, route_after_supervisor, { “planner”: “planner”, “executor”: “executor”, “end”: END, # 如果next是‘end‘则结束流程 “supervisor”: “supervisor”, # 也可以指向自己形成循环判断 } ) # 规划师和执行者完成后都固定返回主管节点由主管进行下一轮调度 builder.add_edge(“planner”, “supervisor”) builder.add_edge(“executor”, “supervisor”) # 编译图 collaborative_app builder.compile()5.4 运行复杂工作流现在让我们运行这个更强大的系统。# 初始化状态 initial_state { “messages”: [HumanMessage(content“我想周末去杭州玩请帮我查一下杭州的天气并推荐两个必去的景点。”)], “next”: “supervisor”, “task_list”: [], “current_task_index”: 0, “all_results”: [] } # 运行应用。设置recursion_limit以防止无限循环对于复杂循环是必要的 from langgraph.checkpoint import MemorySaver from langgraph.graph import MessagesState # 为了更好的演示我们可以使用checkpoint来逐步执行 checkpointer MemorySaver() app_with_memory builder.compile(checkpointercheckpointer) # 开始执行 config {“configurable”: {“thread_id”: “test_thread_1”}} final_state None for step in app_with_memory.stream(initial_state, config, stream_mode“values”): node_name list(step.keys())[0] print(f“\n 节点 [{node_name}] 执行完毕 ) print(f“当前 next: {step[node_name].get(‘next’)}”) print(f“当前任务列表: {step[node_name].get(‘task_list’)}”) print(f“当前任务索引: {step[node_name].get(‘current_task_index’)}”) if step[node_name][“messages”]: last_msg step[node_name][“messages”][-1] print(f“最新消息: {last_msg.type}: {last_msg.content[:100]}...”) # 截取前100字符 print(“\n” “”*50) print(“工作流执行结束”) if final_state is None: final_state step[list(step.keys())[-1]] print(f“最终生成的任务列表: {final_state.get(‘task_list’)}”) print(f“收集的所有结果: {final_state.get(‘all_results’)}”)这个例子展示了LangGraph如何优雅地处理多智能体间的复杂协作、条件判断和循环任务。主管节点充当了“调度中心”的角色整个系统的控制流清晰可见。6. 运行结果与效果验证运行上述代码后你应该能看到类似以下的输出具体内容因LLM输出而异 节点 [supervisor] 执行完毕 当前 next: planner 当前任务列表: [] 当前任务索引: 0 最新消息: ai: 主管任务需要规划已转交规划师。... 节点 [planner] 执行完毕 当前 next: supervisor 当前任务列表: [‘查询杭州本周末的天气情况。‘, ‘搜索杭州最受欢迎的两个旅游景点。‘, ‘汇总天气和景点信息给出出行建议。‘] 当前任务索引: 0 最新消息: ai: 规划师我已制定计划共3个任务查询杭州本周末的天气情况。 搜索杭州最受欢迎的两个旅游景点。 汇总天气和景点信息给出出行建议。... 节点 [supervisor] 执行完毕 当前 next: executor 当前任务列表: [‘查询杭州本周末的天气情况。‘, ‘搜索杭州最受欢迎的两个旅游景点。‘, ‘汇总天气和景点信息给出出行建议。‘] 当前任务索引: 0 最新消息: ai: 主管正在派发任务 ‘查询杭州本周末的天气情况。‘ 给执行者。... 节点 [executor] 执行完毕 当前 next: supervisor 当前任务列表: [‘查询杭州本周末的天气情况。‘, ‘搜索杭州最受欢迎的两个旅游景点。‘, ‘汇总天气和景点信息给出出行建议。‘] 当前任务索引: 1 最新消息: tool: 杭州的天气信息小雨12-18°C... ... (后续循环) ... 工作流执行结束 最终生成的任务列表: [‘查询杭州本周末的天气情况。‘, ‘搜索杭州最受欢迎的两个旅游景点。‘, ‘汇总天气和景点信息给出出行建议。‘] 收集的所有结果: [‘杭州的天气信息小雨12-18°C‘, ‘关于‘杭州最受欢迎的两个旅游景点‘的搜索结果1. 西湖 2. 灵隐寺‘, ‘最终报告...‘]如何验证成功流程正确性观察控制台输出确认流程按照“主管-规划师-主管-执行者-主管-执行者-主管-结束”的顺序执行。这验证了条件路由和循环正常工作。状态一致性检查每个节点执行后task_list,current_task_index,all_results等状态字段是否按预期更新。例如current_task_index应该在每次executor完成后递增。任务完成度最终的all_results列表应包含与task_list中任务数量对应的结果证明所有分解出的子任务都被执行了。工具调用确认executor节点成功调用了我们定义的模拟工具get_weather,search_web并返回了模拟结果。如果运行失败首先检查API密钥是否已正确设置OPENAI_API_KEY环境变量。依赖版本langgraph,langchain等库版本是否兼容。建议使用较新的稳定版本。网络问题是否能正常访问OpenAI API如果你使用其他模型则检查对应服务。状态结构确保节点函数返回的字典键名与State定义中的字段名完全匹配。7. 常见问题与排查思路在开发LangGraph多智能体应用时你可能会遇到以下典型问题问题现象可能原因排查方式解决方案KeyError或状态字段未找到1. 节点返回的字典键与State定义不一致。2. 使用了未在State中声明的字段。1. 仔细核对节点函数return的字典键名。2. 检查State的TypedDict定义。确保节点返回的键名与State定义的字段名完全一致。对于列表追加字段使用Annotated[list, operator.add]。图编译错误1. 节点未添加到图中就添加边。2. 使用了未定义的节点名作为边的目标。3.START/END常量拼写错误。1. 检查add_node和add_edge的调用顺序。2. 检查add_edge或add_conditional_edges中引用的节点名是否存在。遵循“先添加节点再添加边”的顺序。使用常量START和END而非字符串“start”。工作流陷入无限循环1. 条件边逻辑有误导致无法到达END。2. 节点未正确更新next字段。1. 在route_after_supervisor这类路由函数中打印日志。2. 使用app.get_graph().draw_mermaid()可视化图结构检查循环路径。1. 确保路由函数在所有可能情况下都返回一个已定义的节点名或END。2. 使用recursion_limit参数限制最大步数app.invoke(..., recursion_limit100)。工具调用失败或LLM调用超时1. API密钥无效或额度不足。2. 网络连接问题。3. 工具函数参数不匹配或抛出异常。1. 先单独测试LLM调用和工具函数。2. 查看LangChain/OpenAI的错误信息。1. 验证API密钥和网络。2. 在工具函数内部做好异常捕获返回明确的错误信息。3. 为AgentExecutor设置handle_parsing_errorsTrue和max_execution_time。状态更新不符合预期如列表被覆盖对标记为Annotated[list, operator.add]的字段执行了赋值操作而非追加。检查节点中是使用return {“messages”: [new_msg]}正确还是return {“messages”: new_list}可能错误。对于追加字段确保返回的是一个列表LangGraph会自动将其与原列表合并。若要替换整个列表需在State中定义另一个字段。Pydantic解析错误1. LLM的输出不符合JsonOutputParser或Pydantic模型定义的格式。2. 模型字段定义与提示词要求不匹配。1. 在解析前打印LLM的原始输出。2. 检查JsonOutputParser的pydantic_object是否正确定义。1. 在提示词中明确要求JSON格式并提供示例。2. 使用handle_parsing_errors参数或在链中增加重试、格式化步骤。可视化图时出错未安装graphviz或pygraphviz。检查错误信息是否提示缺少图形库。安装graphvizpip install pygraphviz可能需要系统级安装Graphviz软件。或者使用app.get_graph().print_ascii()进行文本可视化。8. 最佳实践与工程建议将LangGraph用于实际项目时遵循以下建议可以大幅提升开发效率和系统稳定性8.1 状态设计原则最小化与清晰化只将工作流真正需要共享的数据放入State。避免将临时变量或节点内部状态塞进去。善用Annotated明确每个字段的更新语义。operator.add用于列表追加operator.setitem用于字典更新需从langgraph.graph导入。对于简单覆盖直接定义类型即可如plan: str。使用Pydantic Model对于复杂状态优先使用pydantic.BaseModel代替TypedDict它能提供运行时验证和更丰富的字段类型。8.2 节点设计原则单一职责每个节点应只做一件事。例如一个节点负责调用LLM生成文本另一个节点负责解析文本并更新状态。幂等性与容错假设节点可能被多次执行在错误重试或特定图结构中设计时应尽量保证幂等性。对工具调用和外部API调用做好异常处理。日志与可观测性在节点函数内关键步骤添加日志记录输入、输出和关键决策。这对于调试复杂工作流至关重要。8.3 图结构设计先画图再写码在编码前用纸笔或绘图工具画出工作流的草图明确节点、边和状态流转。这能极大减少逻辑错误。模块化与子图对于非常复杂的工作流利用StateGraph的嵌套能力将相关节点组封装成子图Subgraph使主图结构更清晰。合理使用条件边条件边非常强大但过度使用会使流程难以理解和调试。确保路由函数的逻辑尽可能简单。8.4 生产环境部署持久化检查点使用SqliteSaver或MongoDBSaver等持久化检查点存储而不是MemorySaver。这样可以在应用重启后恢复长时间运行的工作流状态并支持并发执行。超时与限流为LLM调用和工具调用设置合理的超时时间。对于可能被频繁调用的节点考虑增加限流机制。版本控制工作流图的结构也是代码的一部分。当对图进行修改时要有明确的版本管理策略考虑如何平滑迁移正在运行中的旧版本工作流状态。8.5 测试与调试单元测试节点单独测试每个节点函数模拟输入State验证其输出是否符合预期。集成测试工作流编写测试用例针对不同的初始输入验证整个工作流的最终输出和状态。利用LangGraph Studio积极探索LangGraph官方提供的可视化开发工具LangGraph Studio它允许你以低代码方式构建、调试和监控工作流是开发和理解复杂系统的利器。9. 总结与后续学习方向通过本文我们完成了从理解LangGraph为什么出现到掌握其状态、节点、边三大核心概念再到亲手构建从简单到复杂的多智能体工作流的全过程。关键收获在于思维模式的转变用“图”来建模智能体间的协作用“中心化状态”来管理共享信息。LangGraph的强大之处在于它将复杂的控制流逻辑循环、条件分支、并行从杂乱的业务代码中抽离出来变成了清晰、可视化的图定义。这使得多智能体系统的设计、调试和维护难度大大降低。下一步你可以从以下几个方向深入探索深入研究状态管理尝试更复杂的State设计如嵌套的Pydantic模型或使用add_messages等LangGraph内置的Reducer来简化消息列表操作。探索高级图特性并行执行了解如何利用langgraph.graph中的CONCURRENT语义让多个节点同时运行。中断与人工干预学习如何在工作流中设置“中断点”允许人类审核或提供额外输入后再继续。动态图修改研究是否能在运行时根据状态动态添加或移除节点高级用法。集成更强大的工具将LangGraph与数据库、搜索引擎、外部API、代码解释器等更丰富的工具连接起来构建功能更强大的智能体。投入生产学习如何将编译好的app部署为API服务如使用FastAPI并配置持久化的检查点存储如PostgreSQL以处理真实的、长时间运行的业务流程。关注生态LangGraph生态在快速发展关注langgraph-checkpoint、langgraph-elasticsearch等官方或社区扩展它们能提供更企业级的特性。多智能体架构是构建下一代AI应用的关键。LangGraph提供了一个坚实、优雅的底层框架。掌握它意味着你拥有了将多个“AI大脑”组织起来解决复杂现实问题的能力。建议将本文中的示例代码作为起点不断修改和实验逐步搭建出属于你自己的、能够解决实际业务问题的智能体协作系统。
)
