在构建复杂AI应用时你是否遇到过这样的困境单个大模型智能体Agent能力有限难以处理需要多步骤协作、状态持久化或动态决策流的任务比如一个客服系统需要先理解用户意图再查询知识库最后生成个性化回复并记录对话历史。传统的链式调用如LangChain虽然简单但在处理循环、分支和复杂状态管理时往往力不从心代码变得冗长且难以维护。LangGraph 正是为解决这类问题而生。它并非要取代 LangChain而是作为其强大的补充专注于用图Graph的思维来编排多个智能体或工具构建出具备复杂工作流和状态管理能力的多智能体系统。本文将为你彻底拆解 LangGraph从核心概念、架构设计到手把手代码实战带你构建一个可运行的多智能体协作项目。无论你是想入门智能体开发还是希望将现有单智能体应用升级为更强大的协同系统这篇文章都能提供一条清晰的路径。1. LangGraph 核心概念为什么是“图”在深入代码之前我们必须理解 LangGraph 的核心理念。它将一个工作流抽象为一个有向图。图中的节点Node代表一个执行单元可以是一个智能体、一个工具调用或一个函数边Edge代表执行路径和条件。1.1 与 LangChain 的对比很多人会混淆 LangChain 和 LangGraph。简单来说LangChain是一个用于构建基于大模型应用的框架提供了连接模型、提示词模板、记忆、索引和大量工具链的组件。它的核心是“链”Chain一种线性的、预定义的执行序列。LangGraph是 LangChain 生态系统中的一个库专门用于构建有状态、多分支、可循环的工作流。它的核心是“图”Graph支持更复杂的控制逻辑。你可以把 LangChain 看作建造房屋的“工具箱”提供砖瓦、水泥、木材而 LangGraph 则是专门设计复杂“电路图”或“水管图”的“绘图板”它决定了能量或信息如何在各个房间智能体之间流动。1.2 核心组件解析理解以下几个关键概念是掌握 LangGraph 的基础State状态这是 LangGraph 的灵魂。它是一个共享的数据结构在整个图的工作流执行过程中传递和修改。通常定义为 Pydantic 模型或 TypedDict包含了所有节点需要读取和写入的信息例如用户输入、模型响应、中间结果、对话历史等。Node节点工作流中的一个步骤。它是一个函数接收当前的State作为输入执行操作如调用LLM、运行工具并返回一个对State的更新。关键点节点只关心如何修改状态。Edge边决定工作流下一步走向哪个节点。分为两种条件边Conditional Edge根据State中的某个条件例如模型输出的某个字段决定下一个节点。这实现了“分支”逻辑。普通边Normal Edge无条件地指向下一个节点。这实现了“顺序”逻辑。Graph图由节点和边组成的整体结构。LangGraph 允许你定义“开始”和“结束”节点并支持“循环”将一个边指向之前的某个节点从而实现多轮交互。这种基于图的设计使得构建像“评审-修改”循环、多专家协作路由等复杂智能体系统变得异常清晰和模块化。2. 环境准备与项目搭建我们将在 Python 环境中进行实战。请确保你的 Python 版本 3.8。2.1 安装依赖首先创建一个新的项目目录并安装必要的包。我们将使用 OpenAI 的模型作为智能体的“大脑”因此需要其 API Key。# 创建项目目录并进入 mkdir langgraph-multi-agent-demo cd langgraph-multi-agent-demo # 创建虚拟环境可选但推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装核心库 pip install langgraph langchain-openai langchain-community pydantic # 安装可选库用于示例中的工具 pip install wikipedia requests2.2 设置 API Key为了调用 OpenAI 模型你需要设置环境变量。强烈建议不要将密钥硬编码在代码中。# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here或者在代码中通过os.environ设置仅用于测试生产环境请用上述方式import os os.environ[OPENAI_API_KEY] your-api-key-here2.3 项目结构预览我们的示例项目将包含以下核心文件langgraph-multi-agent-demo/ ├── agents.py # 定义不同的智能体节点 ├── tools.py # 定义智能体可用的工具 ├── graph_builder.py # 构建并编译 LangGraph ├── state.py # 定义共享的 State 结构 └── main.py # 运行工作流的入口文件3. 定义共享状态State与工具State 是工作流中流动的“血液”必须首先设计。3.1 创建 State 模型在state.py中我们使用 Pydantic 来定义一个强类型的 State。# state.py from typing import List, Dict, Any, Optional from pydantic import BaseModel, Field class AgentState(BaseModel): 多智能体工作流的共享状态。 # 用户原始输入 input: str Field(description用户的问题或指令) # 所有智能体产生的对话历史 messages: List[Dict[str, Any]] Field( default_factorylist, description消息历史记录格式为 [{role: user/assistant/tool, content: ...}] ) # 由“路由智能体”决定当前应该由哪个专家处理 next_agent: Optional[str] Field( defaultNone, description决定下一步由哪个专家智能体执行如 writer, coder, researcher ) # 存储最终输出 final_output: Optional[str] Field(defaultNone, description工作流的最终输出结果)这个AgentState包含了工作流所需的所有信息输入、对话上下文、路由决策和最终结果。3.2 创建工具智能体通过工具与外界交互。在tools.py中我们定义几个简单的工具。# tools.py import wikipedia import requests from datetime import datetime def search_wikipedia(query: str) - str: 在维基百科中搜索一个主题。 try: # 设置语言为中文 wikipedia.set_lang(zh) summary wikipedia.summary(query, sentences2) return f维基百科摘要{summary} except wikipedia.exceptions.DisambiguationError as e: return f查询 {query} 可能指代多个条目请更具体一些。选项{e.options[:5]} except wikipedia.exceptions.PageError: return f未找到关于 {query} 的维基百科页面。 except Exception as e: return f搜索维基百科时出错{str(e)} def fetch_webpage(url: str) - str: 获取一个网页的标题和部分内容模拟。 try: response requests.get(url, timeout5) response.raise_for_status() # 简单提取标题实际应用可用BeautifulSoup from html.parser import HTMLParser class TitleParser(HTMLParser): def __init__(self): super().__init__() self.in_title False self.title def handle_starttag(self, tag, attrs): if tag title: self.in_title True def handle_endtag(self, tag): if tag title: self.in_title False def handle_data(self, data): if self.in_title: self.title data parser TitleParser() parser.feed(response.text[:5000]) # 只解析前5000字符以提高速度 title parser.title.strip() or 无标题 return f网页标题{title}。URL{url} except Exception as e: return f获取网页失败{str(e)} def get_current_time(*args) - str: 返回当前日期和时间。 now datetime.now() return f当前时间是{now.strftime(%Y-%m-%d %H:%M:%S)}4. 构建智能体节点Nodes智能体本质上是“具备能力的节点”。我们将创建三个专家智能体和一个路由智能体。4.1 创建专家智能体在agents.py中我们利用langchain_openai的ChatOpenAI和上一步定义的工具来构建智能体。# agents.py from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from .tools import search_wikipedia, fetch_webpage, get_current_time from .state import AgentState # 初始化大语言模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 定义不同专家的工具集 writer_tools [] # 写作专家可能不需要外部工具专注于文本生成 coder_tools [] # 编码专家未来可集成代码执行工具 researcher_tools [search_wikipedia, fetch_webpage, get_current_time] # 为研究员智能体创建提示词 RESEARCHER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个专业的研究员助手。你的职责是使用提供的工具严谨、准确地查找信息来回答用户的问题。 你必须根据问题类型选择合适的工具。例如 - 询问概念或人物 - 使用 search_wikipedia - 询问某个网站内容 - 使用 fetch_webpage - 询问时间 - 使用 get_current_time 你的回答应基于工具返回的事实并注明信息来源。如果工具无法找到答案请如实告知用户。), MessagesPlaceholder(variable_namemessages), # 这里会注入历史消息 ]) # 创建研究员智能体执行器 researcher_agent_runnable create_tool_calling_agent( llmllm, toolsresearcher_tools, promptRESEARCHER_PROMPT ) researcher_executor AgentExecutor(agentresearcher_agent_runnable, toolsresearcher_tools, verboseFalse) def researcher_node(state: AgentState) - dict: 研究员智能体节点使用工具查询信息。 # 从状态中提取最新的用户消息通常是路由智能体转述的问题 last_message state.messages[-1] if state.messages else {content: state.input} human_input last_message.get(content, state.input) # 调用智能体执行器 result researcher_executor.invoke({ messages: [(user, human_input)] }) # 更新消息历史 new_messages state.messages [ {role: assistant, content: result[output]} ] # 返回更新后的状态LangGraph 期望返回一个字典其键对应 State 的字段 return {messages: new_messages} def writer_node(state: AgentState) - dict: 写作专家智能体节点润色、总结或创作文本。 # 模拟一个简单的写作专家实际应用中会使用更复杂的提示词 last_message_content state.messages[-1][content] if state.messages else state.input # 调用LLM进行写作润色 prompt f请将以下内容进行润色、总结使其更加专业、流畅和清晰 原始内容 {last_message_content} 润色后的文本 response llm.invoke(prompt) new_messages state.messages [ {role: assistant, content: response.content} ] return {messages: new_messages} def coder_node(state: AgentState) - dict: 编码专家智能体节点生成或解释代码。 last_message_content state.messages[-1][content] if state.messages else state.input prompt f你是一个资深的软件开发工程师。请根据用户需求生成或解释相关的代码片段。 用户需求 {last_message_content} 请提供代码并附上简要说明。 response llm.invoke(prompt) new_messages state.messages [ {role: assistant, content: response.content} ] return {messages: new_messages}4.2 创建路由智能体路由智能体负责分析用户输入并决定将任务派发给哪个专家。# agents.py (续) def router_node(state: AgentState) - dict: 路由智能体节点分析输入决定下一个执行的专家。 user_input state.input # 使用LLM进行意图分类 router_prompt f请分析用户的请求并决定由哪类专家处理最合适。选项如下 - writer: 如果请求涉及文本润色、总结、创作、翻译等。 - coder: 如果请求涉及编写代码、解释代码、调试、算法等。 - researcher: 如果请求涉及查询事实、获取实时信息、搜索资料等。 - end: 如果请求已得到充分回答或只是一个问候/结束语。 用户请求{user_input} 只返回一个单词必须是 writer, coder, researcher, end 中的一个。 response llm.invoke(router_prompt) decision response.content.strip().lower() # 确保决策在预期范围内 if decision not in [writer, coder, researcher, end]: decision researcher # 默认回退到研究员 # 将路由决策和用户输入添加到消息历史以便专家智能体知晓上下文 new_messages state.messages [ {role: user, content: f路由分配{decision}{user_input}} ] # 更新状态消息历史和下一步要执行的智能体 return {messages: new_messages, next_agent: decision}5. 组装工作流图Graph这是 LangGraph 最核心的部分我们将节点和边连接起来形成一个完整的工作流。在graph_builder.py中完成。# graph_builder.py from langgraph.graph import StateGraph, END from .state import AgentState from .agents import router_node, researcher_node, writer_node, coder_node def build_multi_agent_graph(): 构建并编译多智能体工作流图。 # 1. 创建一个图并指定其状态结构为 AgentState workflow StateGraph(AgentState) # 2. 添加节点 # 第一个节点是路由决策器 workflow.add_node(router, router_node) # 三个专家节点 workflow.add_node(researcher, researcher_node) workflow.add_node(writer, writer_node) workflow.add_node(coder, coder_node) # 3. 设置入口点所有流程都从路由开始 workflow.set_entry_point(router) # 4. 从“router”节点出发根据其返回的 state.next_agent 值动态决定下一个节点 # 这里使用 conditional_edge def decide_next_step(state: AgentState): 根据 router 节点的决策返回下一个节点的名称。 # 如果 router 决定结束则流向 END if state.next_agent end: return END # 否则流向对应的专家节点 return state.next_agent # 添加从 router 出发的条件边 workflow.add_conditional_edges( router, decide_next_step, # 这个函数决定了下一个节点 # 列出所有可能的目的地包括 END { researcher: researcher, writer: writer, coder: coder, END: END } ) # 5. 为每个专家节点添加普通边执行完后直接结束工作流。 # 这是一个简单的设计。更复杂的图可以让专家节点再流回router进行下一轮判断。 workflow.add_edge(researcher, END) workflow.add_edge(writer, END) workflow.add_edge(coder, END) # 6. 编译图得到一个可执行的对象 compiled_graph workflow.compile() return compiled_graph这个图的结构非常清晰用户输入触发进入router节点。router分析意图将next_agent设置为researcher、writer、coder或end。根据next_agent的值工作流被路由到对应的专家节点。专家节点完成任务后工作流结束。6. 运行与测试多智能体系统现在让我们创建一个主程序来运行这个工作流。# main.py import asyncio from graph_builder import build_multi_agent_graph from state import AgentState async def main(): # 编译图 print(正在编译多智能体工作流图...) app build_multi_agent_graph() print(编译完成) # 测试用例 test_inputs [ Python中如何用requests库发送一个POST请求, 帮我总结一下量子计算的主要原理。, 今天的日期是什么, 你好请介绍下你自己。, ] for i, input_text in enumerate(test_inputs): print(f\n{*50}) print(f测试用例 {i1}: {input_text}) print(f{*50}) # 初始化状态 initial_state AgentState(inputinput_text) # 运行图异步 try: # LangGraph 的 compiled_graph 可以同步或异步调用 # 我们使用异步的 ainvoke final_state await app.ainvoke(initial_state) # 打印最终结果 print(f\n[最终输出]:) # 取最后一条助手消息作为输出 if final_state.messages: last_msg final_state.messages[-1] if last_msg[role] assistant: print(last_msg[content]) else: print(未生成助手回复。) else: print(无消息历史。) # 打印执行路径可视化 print(f\n[执行路径]: Router - {final_state.next_agent or END}) except Exception as e: print(f运行工作流时出错{e}) if __name__ __main__: asyncio.run(main())运行这个程序python main.py你将看到类似以下的输出表明不同的用户输入被路由到了不同的专家智能体进行处理正在编译多智能体工作流图... 编译完成 测试用例 1: Python中如何用requests库发送一个POST请求 [最终输出]: 在Python中使用requests库发送POST请求非常简单以下是一个基本的示例 python import requests url https://httpbin.org/post data {key1: value1, key2: value2} headers {Content-Type: application/json} response requests.post(url, jsondata, headersheaders) print(response.status_code) print(response.json())说明1. 导入requests模块。2. 定义目标URL、要发送的数据字典格式和可选的请求头。3. 调用requests.post()方法传入URL、数据和头部信息。4. 返回的response对象包含状态码、响应内容等可以通过.json()方法将JSON响应解析为Python字典。[执行路径]: Router - coder 测试用例 2: 帮我总结一下量子计算的主要原理。量子比特Qubit与传统比特0或1不同量子比特可以同时处于0和1的叠加态直到被测量。叠加允许量子比特同时表示多种状态并行处理大量可能性。纠缠两个或多个量子比特可以形成关联一个的状态变化会瞬时影响另一个无论距离多远。量子门对量子比特进行操作的基本单元通过改变叠加和纠缠状态来实现计算。量子测量测量会使量子比特坍缩到一个确定状态0或1得到计算结果。这些原理使得量子计算机在解决特定问题如大数分解、优化、模拟量子系统上具有超越经典计算机的潜力。[执行路径]: Router - writer## 7. 进阶实现循环与多轮对话 上面的示例是一个简单的单向图。LangGraph 的强大之处在于支持循环可以构建多轮对话或评审流程。例如我们可以让 writer 节点处理完后不直接结束而是流回 router让用户决定是否继续提问或让其他专家处理。 修改 graph_builder.py 中的边定义即可 python # graph_builder.py (进阶版 - 支持循环) def build_multi_agent_graph_with_loop(): workflow StateGraph(AgentState) workflow.add_node(router, router_node) workflow.add_node(researcher, researcher_node) workflow.add_node(writer, writer_node) workflow.add_node(coder, coder_node) workflow.set_entry_point(router) def decide_next_step(state: AgentState): if state.next_agent end: return END return state.next_agent workflow.add_conditional_edges( router, decide_next_step, { researcher: researcher, writer: writer, coder: coder, END: END } ) # 关键修改专家节点执行后不结束而是流回router等待用户下一轮输入 # 这需要修改 State 和节点逻辑来支持持续对话这里展示图的连接方式 workflow.add_edge(researcher, router) workflow.add_edge(writer, router) workflow.add_edge(coder, router) # 注意为了让循环工作我们需要一个机制在流回router前更新state.input为用户的新问题。 # 这通常通过一个额外的“用户输入”节点或修改router逻辑来实现。 return workflow.compile()这种循环结构非常适合构建聊天机器人、多步骤任务分解如规划-执行-检查等复杂场景。8. 常见问题与排查思路在开发 LangGraph 应用时你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案编译图时出错State模型定义错误或节点函数返回值格式不对。1. 检查State的字段是否都有默认值或default_factory。2. 确保节点函数返回的是字典且字典的键是State的字段名。运行时状态不更新节点函数修改了局部变量但没有正确返回更新字典。节点函数必须返回{field_name: new_value}。对列表等可变对象的修改最好创建新对象返回如{messages: old_messages [new_message]}。条件边不生效总是走默认路径decide_next_step函数返回的值不在预设的映射中或者state.next_agent的值不符合预期。1. 打印state.next_agent的值检查路由逻辑。2. 确保add_conditional_edges的映射字典包含了所有可能的返回值。智能体不调用工具工具没有正确绑定到智能体或者提示词没有引导模型使用工具。1. 检查create_tool_calling_agent的tools参数是否正确传入工具列表。2. 在系统提示词中明确要求模型使用工具并描述工具用途。图陷入无限循环边的连接形成了环但没有设置终止条件。1. 检查图形结构确保存在通往END的路径。2. 在状态中设置一个计数器或标志位在router节点判断是否超过最大轮次。langchain相关导入错误版本不兼容或包未安装。1. 使用pip list | grep langchain检查版本。2. LangGraph 和 LangChain 版本需匹配建议使用较新稳定版。3. 确认安装的是langchain-openai,langchain-community。9. 最佳实践与工程建议将 LangGraph 用于实际项目时遵循以下建议可以提升系统的可维护性和鲁棒性精心设计 StateState 是所有节点共享的全局上下文设计要深思熟虑。避免放入过于庞大或频繁变更的数据。使用 Pydantic 进行数据验证和类型提示能提前发现许多数据格式错误。考虑将 State 持久化如存入数据库以实现工作流的暂停、恢复和追溯。保持节点功能单一每个节点应只负责一件事。例如一个节点调用 LLM另一个节点处理数据库查询再一个节点格式化结果。单一职责的节点更容易测试、复用和调试。实现完善的错误处理在节点函数内部使用try...except捕获可能发生的异常如 API 调用失败、工具执行错误。可以考虑设计一个专门的error_handler节点接收出错的 State并决定是重试、转人工还是优雅失败。为图添加可视化LangGraph 提供了get_graph().draw_mermaid()方法可以生成 Mermaid 图表代码便于你理解、评审和文档化工作流。# 在 graph_builder.py 末尾添加 compiled_graph workflow.compile() # 打印 Mermaid 代码可复制到支持 Mermaid 的编辑器如 Typora, Notion中查看 print(compiled_graph.get_graph().draw_mermaid())测试策略单元测试单独测试每个节点函数模拟输入 State断言输出 State。集成测试测试整个图对于特定输入能否产生预期的最终输出和执行路径。Mock 外部依赖在测试中Mock LLM 的返回值和工具的执行结果使测试快速且稳定。性能与成本优化缓存对于频繁且结果不变的 LLM 调用或工具查询如某些知识检索可以考虑引入缓存机制。异步执行如果节点之间没有严格的先后依赖可以考虑使用 LangGraph 的异步支持或并发节点来提升性能。流式输出对于需要长时间运行的图考虑使用 LangGraph 的流式接口逐步输出结果提升用户体验。LangGraph 为我们提供了一种声明式、可视化且强大的方式来编排 AI 智能体。它解决了复杂工作流中的状态管理和控制流难题。通过本文的实战你应该已经掌握了从零构建一个多智能体系统的核心步骤定义状态、创建工具与节点、组装图、运行与调试。接下来你可以尝试将其应用到你的具体场景中例如构建一个智能客服、一个自动化内容创作流水线或一个复杂的决策支持系统。记住好的设计始于清晰的 State 和模块化的节点。