如果你正在构建AI Agent应用可能会遇到这样的困境单个Agent能力有限但多个Agent协作时状态流转混乱、工具调用冲突、流程难以可视化最终代码变成一团难以维护的“面条式”逻辑。这不仅是代码复杂度问题更是工程化Agent系统的核心瓶颈。LangGraph的出现正是为了解决这个痛点。它不是一个全新的框架而是LangChain生态中用于构建有状态、多环节工作流的库。其核心价值在于将复杂的多Agent协作、工具调用序列、条件分支和循环抽象为一张清晰可控的有向图。这彻底改变了我们设计和实现Agent系统的方式——从过程式的代码编写转变为声明式的“绘图”与“组装”。本文将彻底拆解LangGraph不仅告诉你它是什么更会深入剖析它为何能成为构建企业级多智能体Multi-Agent系统的首选架构。我们将从核心概念入手通过一个完整的“旅行规划助手”多Agent实战项目手把手带你搭建环境、编写代码、理解状态管理并最终部署运行。你会看到如何用LangGraph将任务拆解、分配给不同的专家Agent如行程规划师、预算管理员、本地通并让它们有序协作产出可靠结果。1. 这篇文章真正要解决的问题很多开发者初次接触LangGraph时容易产生一个误解它只是LangChain的一个“图形化插件”或“可视化工具”。这个理解过于表面也低估了它的价值。LangGraph解决的根本问题是如何将AI能力LLM调用、工具使用组织成稳定、可靠、可维护的复杂业务流程。在没有LangGraph之前实现一个包含条件判断、循环、并行分支的多Agent系统你需要手动管理大量的中间状态、设计复杂的回调机制、处理可能出现的死循环代码很快就会变得臃肿且脆弱。举个例子一个客服场景可能需要1意图识别Agent2根据意图路由到查询Agent或投诉处理Agent3查询Agent可能需要调用知识库工具和数据库工具4最终由回复生成Agent汇总结果。这个流程中的状态用户问题、识别出的意图、查询到的信息、生成的回复如何传递和持久化某个环节失败如何回退或重试LangGraph通过其StateGraph和MessageState等核心抽象提供了标准化的解决方案。因此本文的目标读者是已经熟悉LangChain基础但苦恼于构建复杂工作流的开发者。正在设计多AI智能体协作系统的架构师。希望将AI能力更深度、更可靠地集成到业务系统中的工程师。读完本文你将能清晰地回答LangGraph的核心模型是什么它与LangChain是什么关系如何用它设计一个多Agent系统以及在实际项目中如何避免常见的“坑”。2. 基础概念与核心原理要理解LangGraph必须先理清几个关键概念以及它们是如何映射到实际编程模型中的。2.1 图Graph与节点Node这是LangGraph最核心的比喻。你可以把整个工作流想象成一张有向图。节点Node代表工作流中的一个步骤或一个计算单元。最常见的就是一个Agent一个LLM调用可能伴随工具使用也可以是一个纯函数如数据格式化、条件判断。边Edge定义了节点之间的执行顺序和条件。决定了上一个节点执行完后下一步该去哪个节点。这种抽象的好处是复杂的逻辑流程变得可视化且易于理解。循环就是指向之前节点的边条件分支就是根据状态值选择不同的边。2.2 状态State这是LangGraph的“灵魂”。整个图在执行过程中所有节点共享并修改同一个状态对象。状态是一个字典或Pydantic模型包含了工作流运行所需的所有数据。初始状态工作流开始时的输入。状态流转每个节点读取状态中的部分数据执行计算如调用LLM然后将结果写回状态。最终状态工作流结束时的输出。LangGraph提供了几种预定义的状态类最常用的是MessageState它专门为基于消息类似聊天记录的对话式Agent工作流设计内置了消息列表的管理。2.3 智能体Agent与工具Tool在LangGraph的语境下一个“智能体”通常被实现为一个节点。这个节点封装了提示词Prompt定义该Agent的角色和任务。语言模型LLM如OpenAI的GPT、Anthropic的Claude或本地的Ollama模型。工具ToolsAgent可以调用的函数如搜索、计算、查询数据库等。LangGraph本身不定义Agent它利用LangChain Core中成熟的Runnable协议来构建节点这意味着你可以无缝使用LangChain生态中大量的现有组件。2.4 监督者Supervisor这是一个高级但至关重要的模式尤其在多Agent系统中。Supervisor本身也是一个Agent一个节点它的职责是统筹规划接收总任务并将其拆解成子任务。任务分发根据子任务类型决定调用哪个专家Agent节点。结果汇总收集各专家Agent的结果进行整合或判断是否完成。在多Agent架构中Supervisor节点通常作为图的“中枢”或“路由器”是实现复杂任务协调的关键。LangGraph vs. LangChain澄清关系这是一个常见的困惑点。简单来说LangChain是一个用于开发由LLM驱动的应用程序的完整框架。它提供了模型I/O、提示词管理、记忆、索引、链Chains和代理Agents等大量组件。LangGraph是LangChain生态系统中的一个库专注于一件事构建有状态、多步骤的工作流图。它是对LangChain中“链”和“代理”概念的增强和规范化特别擅长处理比简单线性链更复杂的逻辑。你可以把LangChain看作你的“工具箱”而LangGraph是工具箱里一把专门用于组装复杂机械工作流的“多功能扳手”。在项目中你通常会同时使用两者。3. 环境准备与前置条件在开始实战之前我们需要搭建开发环境。本项目将构建一个基于本地Ollama模型的多Agent旅行规划系统因此你需要准备以下内容。3.1 软件与工具Python 3.10 或以上这是大多数AI框架的稳定支持版本。包管理工具pip或conda。代码编辑器VS Code、PyCharm等均可。Ollama用于在本地运行开源大模型。我们将使用llama3.2或qwen2.5等较小模型进行演示。3.2 安装Ollama访问 Ollama官网 下载并安装对应操作系统的版本。打开终端拉取一个模型例如ollama pull llama3.2运行模型服务确保它在后台运行。Ollama默认会在http://localhost:11434提供API服务。3.3 创建项目与安装依赖创建一个新的项目目录并建立虚拟环境。# 创建项目目录并进入 mkdir langgraph-travel-planner cd langgraph-travel-planner # 创建虚拟环境 (以venv为例) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate安装必要的Python包pip install langgraph langchain langchain-community langchain-corelanggraph: 本文的核心。langchain和langchain-core: 提供基础组件LLM、提示词、工具等。langchain-community: 包含社区维护的集成如与Ollama的连接器。3.4 验证环境创建一个简单的Python脚本test_env.py来测试环境是否正常。# test_env.py from langchain_community.llms import Ollama # 连接到本地Ollama服务 llm Ollama(modelllama3.2) # 尝试一个简单生成 response llm.invoke(请用一句话介绍你自己。) print(模型回复, response)运行脚本python test_env.py如果看到模型返回了一句自我介绍说明Ollama和LangChain连接成功。4. 核心流程拆解构建多Agent旅行规划图我们将构建一个包含三个专家Agent和一个监督者Supervisor的旅行规划系统行程规划师ItineraryPlanner负责生成详细的每日行程安排。预算管理员BudgetManager负责估算行程花费并提供省钱建议。本地通LocalExpert负责提供目的地的小贴士、文化禁忌和必备物品。旅行规划监督者TravelSupervisor接收用户请求协调以上三个专家工作并汇总最终报告。整个工作流的逻辑图如下文字描述用户输入 - TravelSupervisor - 根据任务类型分发 ├── 需要详细行程 - ItineraryPlanner ├── 需要预算 - BudgetManager └── 需要本地建议 - LocalExpert 可能循环调用直到Supervisor认为信息完备 - 各专家结果汇总至Supervisor - 最终报告 - 输出给用户接下来我们将分步实现这个图。5. 完整示例与代码实现我们将代码组织在多个文件中保持清晰度。5.1 定义共享状态State首先我们需要定义在整个工作流中传递的状态。我们将使用TypedDict来创建类型化的状态字典。# state.py from typing import TypedDict, List, Annotated import operator from langgraph.graph.message import add_messages class TravelState(TypedDict): 旅行规划工作流的共享状态。 # 用户原始输入 user_request: str # 消息历史用于记录Agent间的对话 messages: Annotated[List[str], add_messages] # 行程规划师输出的详细行程 itinerary: str # 预算管理员输出的预算估算 budget_estimate: str # 本地通输出的本地建议 local_advice: str # 监督者汇总的最终报告 final_report: str # 一个标志位指示工作流是否应继续 should_continue: bool这里的关键是Annotated[List[str], add_messages]。add_messages是LangGraph提供的一个归约器Reducer它定义了当多个节点都想修改messages字段时如何合并它们的修改这里是追加到列表。这是管理对话历史的最佳实践。5.2 实现专家Agent节点每个专家Agent都是一个函数它接收当前State调用LLM和工具然后返回更新后的State片段。# agents.py from langchain_core.prompts import ChatPromptTemplate from langchain_community.llms import Ollama from state import TravelState import json # 初始化LLM llm Ollama(modelllama3.2) # 可根据你的Ollama模型调整 def itinerary_planner_node(state: TravelState) - dict: 行程规划师节点。 print([ItineraryPlanner] 正在规划行程...) prompt ChatPromptTemplate.from_messages([ (system, 你是一位专业的旅行行程规划师。请根据用户的需求生成一份详细到每天上午、下午、晚上的旅行行程安排。请考虑交通、景点、餐饮和休息的合理性。), (human, 用户需求{request}\n\n请生成详细行程) ]) chain prompt | llm itinerary chain.invoke({request: state[user_request]}) # 返回要更新到状态中的字段 return {itinerary: itinerary, messages: [f行程规划师已生成行程{itinerary[:100]}...]} def budget_manager_node(state: TravelState) - dict: 预算管理员节点。 print([BudgetManager] 正在估算预算...) prompt ChatPromptTemplate.from_messages([ (system, 你是一位精打细算的旅行预算管理员。请根据行程如果提供和用户需求估算大致的总花费并拆分为交通、住宿、餐饮、门票等类别。同时提供3条省钱建议。), (human, 用户需求{request}\n\n相关行程{itinerary}\n\n请估算预算并提供建议) ]) chain prompt | llm # 注意这里我们尝试使用行程信息即使它可能为空初次调用时 budget_info chain.invoke({request: state[user_request], itinerary: state.get(itinerary, 暂无详细行程)}) return {budget_estimate: budget_info, messages: [f预算管理员已完成估算{budget_info[:100]}...]} def local_expert_node(state: TravelState) - dict: 本地通节点。 print([LocalExpert] 正在提供本地建议...) prompt ChatPromptTemplate.from_messages([ (system, 你是一位目的地本地通熟知当地文化、习俗、隐藏景点和实用信息。请提供旅行小贴士、文化禁忌、必备物品和推荐的非网红体验。), (human, 用户计划前往的目的地相关信息{request}\n\n请提供本地化建议) ]) chain prompt | llm local_tips chain.invoke({request: state[user_request]}) return {local_advice: local_tips, messages: [f本地通已提供建议{local_tips[:100]}...]}5.3 实现监督者Supervisor节点监督者是一个更复杂的节点它需要决定下一步调用哪个专家或者判断任务是否完成。# supervisor.py from langchain_core.prompts import ChatPromptTemplate from agents import llm # 复用同一个LLM实例 from state import TravelState def travel_supervisor_node(state: TravelState) - dict: 旅行规划监督者节点。负责协调和汇总。 print([TravelSupervisor] 正在分析任务并协调专家...) # 1. 首先判断需要调用哪些专家 planning_prompt ChatPromptTemplate.from_messages([ (system, 你是一个旅行规划监督者。你需要分析用户请求并决定需要调用哪些专家行程规划师、预算管理员、本地通来完成任务。 你拥有以下专家 - ItineraryPlanner: 当用户需要具体行程安排时调用。 - BudgetManager: 当用户关心花费或需要预算时调用。 - LocalExpert: 当用户需要本地文化、小贴士、禁忌等信息时调用。 请只返回一个JSON对象格式如下 {{needs_itinerary: true/false, needs_budget: true/false, needs_local_advice: true/false, summary: 一句话任务总结}} ), (human, 用户请求{request}) ]) planning_chain planning_prompt | llm decision_str planning_chain.invoke({request: state[user_request]}) # 简单解析LLM的返回生产环境需要更健壮的解析 try: # 尝试提取JSON部分 import re json_match re.search(r\{.*\}, decision_str, re.DOTALL) if json_match: decision json.loads(json_match.group()) else: decision {needs_itinerary: True, needs_budget: True, needs_local_advice: True, summary: 默认调用所有专家} except json.JSONDecodeError: decision {needs_itinerary: True, needs_budget: True, needs_local_advice: True, summary: 解析失败默认调用所有专家} print(f[Supervisor] 决策结果{decision}) # 2. 根据决策设置下一个要执行的节点。 # LangGraph通过修改状态中的特殊字段来路由这里我们用 next_node 来演示。 # 更优雅的方式是使用 send 到不同的节点但为简化我们先在状态里存标志。 # 实际项目中Supervisor节点后通常会连接一个路由逻辑Conditional Edge。 # 3. 检查是否所有需要的专家都已提供信息这是一个简化逻辑 # 我们假设Supervisor在协调一圈后自己生成最终报告。 has_itinerary bool(state.get(itinerary)) has_budget bool(state.get(budget_estimate)) has_advice bool(state.get(local_advice)) # 简单的完成条件如果状态中已有所需信息则生成报告 needs_itinerary decision.get(needs_itinerary, False) needs_budget decision.get(needs_budget, False) needs_local_advice decision.get(needs_local_advice, False) should_generate_report True if needs_itinerary and not has_itinerary: should_generate_report False if needs_budget and not has_budget: should_generate_report False if needs_local_advice and not has_advice: should_generate_report False if should_generate_report: print([Supervisor] 信息已齐备生成最终报告...) report_prompt ChatPromptTemplate.from_messages([ (system, 你是一位旅行报告整合专家。请将以下行程、预算和本地建议整合成一份完整、流畅、用户友好的最终旅行规划报告。), (human, 用户原始需求{request} 详细行程 {itinerary} 预算估算 {budget} 本地建议 {advice} 请生成最终报告) ]) report_chain report_prompt | llm final_report report_chain.invoke({ request: state[user_request], itinerary: state.get(itinerary, 暂无), budget: state.get(budget_estimate, 暂无), advice: state.get(local_advice, 暂无) }) return { final_report: final_report, should_continue: False, # 工作流结束 messages: [f监督者已生成最终报告。] } else: # 信息不全告诉图需要继续执行例如去调用缺失信息的专家 # 这里我们返回一个标志后续通过图的边逻辑来处理 next_task [] if needs_itinerary and not has_itinerary: next_task.append(itinerary) if needs_budget and not has_budget: next_task.append(budget) if needs_local_advice and not has_advice: next_task.append(local_advice) print(f[Supervisor] 仍需执行{next_task}) return { should_continue: True, next_tasks: next_task, # 指示下一步做什么 messages: [f监督者决定下一步执行{next_task}] }5.4 构建并编译LangGraph图这是将所有节点和路由逻辑组装起来的关键步骤。# graph_builder.py from langgraph.graph import StateGraph, END from state import TravelState from agents import itinerary_planner_node, budget_manager_node, local_expert_node from supervisor import travel_supervisor_node def build_travel_planning_graph(): 构建并返回旅行规划工作流图。 # 1. 创建图构建器并指定状态模式 workflow StateGraph(TravelState) # 2. 添加节点 workflow.add_node(supervisor, travel_supervisor_node) workflow.add_node(planner, itinerary_planner_node) workflow.add_node(budget_manager, budget_manager_node) workflow.add_node(local_expert, local_expert_node) # 3. 设置入口点 workflow.set_entry_point(supervisor) # 4. 定义边路由逻辑 # 这是一个简化的路由Supervisor之后根据状态中的next_tasks决定下一步 # 我们使用一个条件函数来实现路由 def route_after_supervisor(state: TravelState) - str: 根据supervisor的输出决定下一个节点。 if not state.get(should_continue, True): # 如果should_continue为False表示工作流结束 return END # 否则查看需要执行什么任务 next_tasks state.get(next_tasks, []) if not next_tasks: # 没有任务了也结束 return END # 这里我们简单选择第一个任务实际可以更复杂如并行 task next_tasks[0] if task itinerary: return planner elif task budget: return budget_manager elif task local_advice: return local_expert else: # 未知任务回到supervisor重新决策 return supervisor # 5. 添加从supervisor出发的条件边 workflow.add_conditional_edges( supervisor, route_after_supervisor, { planner: planner, budget_manager: budget_manager, local_expert: local_expert, END: END } ) # 6. 设置专家节点执行完后都回到supervisor进行下一轮协调 workflow.add_edge(planner, supervisor) workflow.add_edge(budget_manager, supervisor) workflow.add_edge(local_expert, supervisor) # 7. 编译图 app workflow.compile() return app if __name__ __main__: # 测试图构建 app build_travel_planning_graph() print(旅行规划图构建成功) # 可视化图结构需要安装额外的库如graphviz try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: print(无法显示图形但图已成功编译。)5.5 主程序运行工作流创建一个主文件来运行整个多Agent系统。# main.py from graph_builder import build_travel_planning_graph from state import TravelState def main(): # 1. 构建图 print(正在初始化旅行规划多Agent系统...) app build_travel_planning_graph() # 2. 准备初始状态用户输入 user_request input(请输入您的旅行需求例如我计划下个月去杭州旅行3天预算5000元喜欢自然风光和历史文化\n) # user_request 我计划下个月去杭州旅行3天预算5000元喜欢自然风光和历史文化 # 也可以写死用于测试 initial_state: TravelState { user_request: user_request, messages: [], itinerary: , budget_estimate: , local_advice: , final_report: , should_continue: True, next_tasks: [] # 初始为空由supervisor填充 } # 3. 运行图 print(\n开始执行多Agent规划流程...) print(*50) final_state app.invoke(initial_state) # 4. 输出结果 print(\n *50) print(✅ 旅行规划完成) print(*50) print(\n 最终报告) print(-*30) print(final_state[final_report]) print(-*30) # 可选查看中间状态 # print(\n 中间消息记录) # for msg in final_state.get(messages, []): # print(f - {msg}) if __name__ __main__: main()6. 运行结果与效果验证现在让我们运行这个系统看看多Agent是如何协作的。启动程序在终端中确保虚拟环境已激活并运行python main.py输入请求程序会提示你输入旅行需求。输入一个具体描述例如“我计划国庆假期去西安旅行4天主要想看兵马俑、陕西历史博物馆品尝当地小吃总预算控制在4000元左右。”观察执行过程在控制台你会看到类似以下的日志清晰地展示了Agent间的协作流程正在初始化旅行规划多Agent系统... 请输入您的旅行需求... [用户输入] 开始执行多Agent规划流程... [TravelSupervisor] 正在分析任务并协调专家... [Supervisor] 决策结果{needs_itinerary: True, needs_budget: True, needs_local_advice: True, summary: ...} [Supervisor] 仍需执行[itinerary, budget, local_advice] [ItineraryPlanner] 正在规划行程... [TravelSupervisor] 正在分析任务并协调专家... [Supervisor] 仍需执行[budget, local_advice] [BudgetManager] 正在估算预算... [TravelSupervisor] 正在分析任务并协调专家... [Supervisor] 仍需执行[local_advice] [LocalExpert] 正在提供本地建议... [TravelSupervisor] 正在分析任务并协调专家... [Supervisor] 信息已齐备生成最终报告... ✅ 旅行规划完成 最终报告 ------------------------------ 【西安4天3晚文化美食之旅规划报告】 一、行程概览 ... (详细的整合报告) ------------------------------验证结果检查最终输出的报告是否完整性包含了行程、预算、本地建议三大块。一致性预算与行程天数、活动相匹配。实用性本地建议具有可操作性如交通提示、美食推荐。如何判断成功程序没有报错正常执行完毕。控制台日志显示了清晰的“Supervisor - 专家 - Supervisor”的循环协调过程。最终输出了一个结构完整、信息丰富的旅行规划报告。状态final_state中的itinerary、budget_estimate、local_advice字段都被正确填充。如果失败第一步排查检查Ollama服务运行ollama list确认模型已下载运行curl http://localhost:11434/api/generate -d {model:llama3.2, prompt:hello}测试API是否通畅。检查Python依赖确认langgraph,langchain,langchain-community版本兼容。查看错误日志Python的完整错误栈会指出是代码语法错误、导入错误还是LLM API调用错误。7. 常见问题与排查思路在开发和运行LangGraph多Agent系统时你可能会遇到以下典型问题问题现象可能原因排查方式解决方案RuntimeError: ...图编译错误节点函数签名与State定义不匹配循环引用未正确终止。1. 检查所有node函数是否接收State并返回dict。2. 检查图是否有通往END的路径避免死循环。确保函数返回的字典键名与State的字段名对应。使用add_conditional_edges或add_edge明确指向END。工作流陷入无限循环条件边Conditional Edge逻辑有误导致状态无法满足结束条件。1. 打印should_continue等控制变量。2. 检查Supervisor节点的决策逻辑是否在某些边界情况下无法将should_continue设为False。在状态中添加iteration_count字段在Supervisor节点中对其递增并在达到阈值如10次后强制结束。LLM调用超时或返回空Ollama服务未启动模型名称错误网络问题。1. 运行ollama serve查看服务状态。2. 用简单脚本如test_env.py测试LLM基础调用。确保Ollama在运行且模型名与代码中Ollama(model...)一致。考虑增加LLM调用的超时参数。状态更新未生效节点函数修改了状态但后续节点读取的是旧值。1. 在每个节点开始和结束时打印关键状态字段。2. 确认使用了Annotated和正确的Reducer如add_messages。理解LangGraph的状态更新是增量合并的。节点应返回需要修改的字段而不是完整状态。确保Reducer使用正确。多个Agent输出格式混乱每个Agent的LLM返回格式不一导致Supervisor解析失败。打印每个Agent节点的原始输出。为每个Agent设计更严格的提示词要求其以特定格式如JSON、Markdown章节返回。使用LangChain的OutputParser来结构化输出。MessageState消息列表异常直接对messages列表进行append操作而不是通过Reducer。检查代码中是否有state[“messages”].append(...)。永远不要直接修改状态中的列表/字典。对于messages应返回{“messages”: [new_message]}LangGraph会用add_messages自动合并。8. 最佳实践与工程建议基于实战经验以下是构建生产级LangGraph多Agent系统的关键建议状态设计要精简而明确只将工作流真正需要共享和传递的数据放入State。避免将临时计算变量也塞进去。优先使用TypedDict或Pydantic模型来定义State以获得类型提示和验证。对于列表类字段如messages务必使用Annotated和官方Reducer如add_messages来管理并发更新。节点设计遵循单一职责每个节点Agent应只做一件事并把它做好。例如一个节点负责“信息提取”另一个负责“格式校验”。节点函数应保持纯净输入是State输出是State的更新片段。避免在节点内部维护全局变量或产生副作用如写入文件除非这是其明确职责。强化Supervisor的决策可靠性本示例中的Supervisor决策逻辑较为简单。在生产环境中应通过以下方式增强结构化输出强制Supervisor的LLM调用返回可解析的JSON。验证与重试对LLM的决策结果进行校验如果不符合预期设计重试或降级逻辑。上下文感知让Supervisor能参考完整的对话历史messages来做决策而不仅仅是初始请求。实现可观测性与调试在每个节点的开始和结束处添加日志记录关键输入输出。利用LangGraph的 检查点Checkpoint 功能持久化中间状态便于调试和实现“暂停/继续”。考虑集成langgraph studio进行可视化调试它能直观展示图的执行路径和状态变化。处理错误与边界情况LLM调用失败在节点函数中使用try...except包裹LLM调用在失败时返回错误信息到State并由Supervisor或特定错误处理节点决定重试或终止。超时控制为长时间运行的工作流设置总体超时或为每个LLM调用设置单独超时。用户干预设计“人工审核”节点在关键决策点如大额预算批准将状态暂停等待外部输入后再继续。性能优化并行执行如果多个专家节点间没有依赖关系可以使用langgraph的StateGraph的add_edge实现并发。例如让Supervisor同时将任务发送给Planner和LocalExpert。缓存对于昂贵的LLM调用如查询相同的目的地信息可以考虑集成缓存层。流式输出如果最终报告很长可以利用LLM的流式响应能力逐步输出给用户提升体验。通过遵循这些实践你的LangGraph多Agent系统将不仅能够运行更能达到可靠、可维护、可观测的企业级应用标准。从简单的任务自动化到复杂的业务决策流程LangGraph提供的图抽象都能为你提供一个清晰且强大的框架。