如果你正在寻找一套能真正跑起来的 LangGraph 多智能体实战方案而不是停留在概念讲解那么这篇文章就是为你准备的。LangGraph 作为 LangChain 生态中构建复杂、有状态多智能体应用的核心框架其价值在于能用清晰的图结构定义智能体之间的协作与决策流程。本文将直接切入实战带你从零搭建一个可运行的多智能体系统重点剖析其架构设计、核心组件并通过完整的代码示例让你快速掌握如何让多个 AI 智能体协同工作解决复杂任务。本文的核心是“能用”和“怎么用”。我们将重点关注 LangGraph 的架构思想、关键组件的实际作用、代码如何组织以及如何在你自己的开发环境中快速启动和验证。无论你是想构建一个自动化工作流还是研究多智能体协同的潜力这里提供的思路和代码都能直接复用。1. 核心能力速览在深入代码之前我们先快速了解 LangGraph 在多智能体场景下的核心价值与能力边界。能力项说明与实战价值项目类型基于 Python 的、用于构建有状态、多智能体工作流的框架库。核心开源方LangChain AI。它与 LangChain 深度集成但也可独立用于构建智能体图。主要功能1.定义工作流图将多个智能体节点和决策逻辑边组织成有向图。2.管理复杂状态在整个工作流执行过程中持久化并传递共享状态State。3.支持循环与分支实现智能体间的多轮对话、条件判断和动态路由。硬件/环境门槛纯代码库无特殊硬件要求。运行依赖主要是 Python 环境和所需的大模型 API 密钥如 OpenAI、 Anthropic 等或本地模型。对 CPU/GPU 无强制要求性能取决于背后调用的模型服务。启动与运行方式通过 Python 脚本启动。核心是定义一个StateGraph添加节点和边编译为可执行的Graph对象然后传入初始状态调用。是否支持 API 服务原生是一个库但你可以轻松地将编译好的Graph对象封装成 FastAPI、 Flask 等 Web 服务对外提供 API。是否支持批量/异步任务支持。Graph的ainvoke或abatch方法可用于异步处理多个并行的任务流。适合场景1.复杂任务拆解与分配如一个智能体分析需求另一个写代码第三个进行审查。2.模拟辩论与评审多个智能体围绕一个议题进行多轮讨论形成最终结论。3.自动化业务流程需要条件判断、循环审批的多步骤自动化流程。2. 适用场景与使用边界LangGraph 不是万能的理解其适用场景和边界能帮助你更好地决策。它非常适合以下情况任务需要多角色协同例如一个“产品经理”智能体生成需求文档一个“工程师”智能体编写代码一个“测试员”智能体检查代码质量。流程包含决策与循环例如根据代码审查结果决定是“通过”、“返回修改”还是“需要更多评审”。LangGraph 的“条件边”可以优雅地处理这种逻辑。需要持久化对话上下文在多轮交互中需要记住之前的对话历史、中间结果或工具调用记录。LangGraph 的State是管理这类共享上下文的理想容器。它可能不是最佳选择简单的单次问答如果只是调用一次大模型 API 并返回结果直接使用ChatOpenAI等客户端更简单。无状态的线性管道如果只是将数据依次通过几个处理函数A - B - C且中间不需要复杂的条件分支或状态共享使用简单的函数组合或 LangChain Expression Language (LCEL) 可能更轻量。对图形化编排有强依赖虽然 LangGraph Studio 提供了可视化界面但其核心仍是代码定义。如果你追求完全零代码的拖拽式工作流设计可能需要考虑其他专门的低代码平台。合规与安全边界API 密钥管理本文示例会使用大模型 API如 OpenAI。请务必妥善保管你的 API 密钥不要在代码或版本控制中明文提交。内容安全由多智能体生成的内容如代码、文档、建议需进行人工审核特别是用于生产环境时。智能体可能产生错误或不符合预期的输出。成本控制多智能体工作流意味着多次调用大模型 API需密切关注使用量避免意外成本。3. 环境准备与前置条件让我们开始准备实战环境。你需要的是一个干净的 Python 开发环境。Python 版本推荐使用 Python 3.10 或 3.11。确保python和pip命令可用。虚拟环境强烈推荐使用venv或conda创建独立环境避免包冲突。# 使用 venv python -m venv langgraph-env # 激活环境 # Windows: langgraph-env\Scripts\activate # Linux/Mac: source langgraph-env/bin/activate大模型 API 访问权限准备一个可用的 OpenAI API 密钥或其他如 Anthropic、Groq 等支持的模型提供商密钥。我们将以此作为智能体的“大脑”。基础工具一个你喜欢的代码编辑器如 VS Code和终端。4. 安装部署与启动方式安装过程非常简单主要通过pip完成。# 安装 langgraph 核心库 pip install langgraph # 安装 langchain 和 openai 库用于构建智能体和连接模型 pip install langchain-openai langchain # 可选安装 langchain-community包含更多社区工具和集成 # pip install langchain-community验证安装在 Python 交互环境中执行import langgraph若无报错则安装成功。启动方式的核心LangGraph 应用的“启动”不是运行一个服务而是执行一个 Python 脚本。这个脚本定义了你的工作流图Graph并调用它。一个最简单的启动脚本骨架如下# app.py from langgraph.graph import StateGraph, END # ... 导入其他必要的模块 # 1. 定义状态State的结构 class AgentState(TypedDict): messages: list # 可以添加其他共享字段如 task, results 等 # 2. 定义各个智能体节点函数 def node_agent_1(state: AgentState): # 处理逻辑更新 state return {messages: [新的消息]} def node_agent_2(state: AgentState): # 处理逻辑更新 state return {messages: [新的消息]} # 3. 构建图 workflow StateGraph(AgentState) workflow.add_node(agent_1, node_agent_1) workflow.add_node(agent_2, node_agent_2) # 4. 定义边执行顺序和条件 workflow.set_entry_point(agent_1) workflow.add_edge(agent_1, agent_2) workflow.add_edge(agent_2, END) # 5. 编译图 app workflow.compile() # 6. 运行图这才是“启动” initial_state {messages: [{role: user, content: 初始任务描述}]} final_state app.invoke(initial_state) print(final_state)执行这个脚本就是启动你的多智能体应用python app.py。5. 功能测试与效果验证构建一个代码评审多智能体系统现在我们构建一个实用的例子一个由“开发者”和“评审者”两个智能体组成的代码评审系统。流程是用户提出需求 - 开发者写代码 - 评审者审查代码 - 若评审不通过返回给开发者修改 - 循环直至通过或超时。5.1 定义共享状态与智能体节点首先定义整个工作流需要共享的状态。我们将使用TypedDict来获得更好的类型提示。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage import os # 设置你的 OpenAI API Key (请勿提交到代码仓库) os.environ[OPENAI_API_KEY] 你的-api-key-here # 1. 定义状态结构 class CodeReviewState(TypedDict): 多智能体代码评审工作流的共享状态。 original_requirement: str # 原始需求 messages: List # 完整的对话历史 current_code: str # 当前版本的代码 review_result: str # 最近的评审意见 iteration_count: int # 迭代次数用于防止无限循环 is_approved: bool # 代码是否已通过评审 # 2. 初始化大模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) # 使用一个性价比较高的模型 # 3. 定义“开发者”智能体节点 def developer_node(state: CodeReviewState): 根据需求或评审意见编写或修改代码。 system_prompt 你是一名资深软件开发工程师。你的任务是根据用户需求或代码评审意见编写高质量的、可运行的Python代码。 只输出代码本身除非特别要求否则不要包含任何解释性文字。确保代码逻辑正确、简洁高效。 # 构建对话历史 chat_history state[messages] # 如果是第一次迭代根据原始需求写代码 if state[iteration_count] 0: user_input f请根据以下需求编写Python代码\n{state[original_requirement]} else: # 如果不是第一次则根据评审意见修改代码 user_input f这是你上一版的代码\npython\n{state[current_code]}\n\n\n这是评审意见\n{state[review_result]}\n\n请根据评审意见修改代码。 messages [ SystemMessage(contentsystem_prompt), *chat_history, # 包含历史对话让智能体有上下文 HumanMessage(contentuser_input) ] # 调用大模型 response llm.invoke(messages) new_code response.content.strip() # 更新状态 new_messages chat_history [HumanMessage(contentuser_input), AIMessage(contentnew_code)] return { current_code: new_code, messages: new_messages, iteration_count: state[iteration_count] 1 } # 4. 定义“评审者”智能体节点 def reviewer_node(state: CodeReviewState): 评审开发者提交的代码给出通过/不通过的意见。 system_prompt 你是一名严格的代码评审专家。你的任务是评审Python代码检查其是否正确、高效、符合需求并遵循最佳实践。 请给出明确的评审结论APPROVED 或 NEEDS_MODIFICATION。 如果结论是 NEEDS_MODIFICATION必须提供具体、可操作的修改意见。 user_input f请评审以下Python代码。 需求描述{state[original_requirement]} 待评审代码 python {state[current_code]} messages [ SystemMessage(contentsystem_prompt), HumanMessage(contentuser_input) ] response llm.invoke(messages) review_text response.content # 简单解析评审结论在实际应用中可能需要更复杂的解析逻辑 is_approved APPROVED in review_text.upper() return { review_result: review_text, is_approved: is_approved }5.2 构建图与条件路由这是 LangGraph 的核心魅力所在定义智能体之间的流转逻辑。# 5. 构建状态图 workflow StateGraph(CodeReviewState) # 添加节点 workflow.add_node(developer, developer_node) workflow.add_node(reviewer, reviewer_node) # 设置入口点从开发者开始 workflow.set_entry_point(developer) # 添加边开发者完成后总是交给评审者 workflow.add_edge(developer, reviewer) # 6. 定义条件边评审者之后根据结果决定下一步 def should_continue(state: CodeReviewState): 根据评审结果决定下一步是循环修改还是结束。 if state[is_approved]: return end # 通过结束 elif state[iteration_count] 3: # 设置最大迭代次数防止无限循环 return end # 超时强制结束 else: return developer # 未通过返回给开发者修改 # 添加条件边 workflow.add_conditional_edges( reviewer, # 从哪个节点出发 should_continue, # 条件判断函数 { developer: developer, # 返回修改 end: END # 结束流程 } ) # 7. 编译图 app workflow.compile()5.3 运行测试与效果验证现在让我们用一个实际的需求来测试这个多智能体系统。# 8. 定义初始状态并运行图 initial_state: CodeReviewState { original_requirement: 编写一个函数接收一个整数列表返回列表中所有偶数的平方组成的新列表。, messages: [], current_code: , review_result: , iteration_count: 0, is_approved: False } print( 开始多智能体代码评审流程 ) print(f原始需求: {initial_state[original_requirement]}\n) # 运行工作流 try: final_state app.invoke(initial_state) print(f\n 流程结束 ) print(f总迭代次数: {final_state[iteration_count]}) print(f最终评审结果: {通过 if final_state[is_approved] else 未通过可能已达最大迭代次数}) print(f\n最终生成的代码:\npython\n{final_state[current_code]}\n) if final_state[review_result]: print(f\n最终评审意见:\n{final_state[review_result]}) except Exception as e: print(f运行过程中出现错误: {e})预期结果与判断标准成功标准脚本能正常运行不报错。在控制台能看到“开始流程”和“流程结束”的日志。最终会输出一段符合需求的 Python 代码get_even_squares函数。效果验证观察输出。理想情况下经过 1-2 轮“开发-评审”循环is_approved会变为True并输出简洁高效的代码。评审意见会指出初始代码可能存在的不足如未处理空列表、变量命名不清晰等开发者智能体会据此修改。常见失败原因API 密钥错误OPENAI_API_KEY未设置或无效会导致llm.invoke调用失败。网络问题无法连接到 OpenAI API。依赖包版本冲突确保langgraph,langchain,langchain-openai版本兼容。条件判断函数逻辑错误should_continue函数返回的字符串必须与add_conditional_edges中映射的键完全一致。6. 接口 API 与批量任务封装将编译好的app(即Graph对象) 封装成 Web 服务可以轻松对外提供 API并处理批量任务。6.1 使用 FastAPI 封装为 HTTP 服务# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional import asyncio from your_workflow_module import app # 导入前面编译好的 app 图对象 # 假设你的状态定义在同一个文件或已导入 app_fastapi FastAPI(titleLangGraph 代码评审多智能体 API) class CodeReviewRequest(BaseModel): requirement: str max_iterations: Optional[int] 3 # 允许客户端自定义最大迭代次数 class CodeReviewResponse(BaseModel): success: bool final_code: Optional[str] None review_history: Optional[list] None iteration_count: int message: str app_fastapi.post(/review, response_modelCodeReviewResponse) async def code_review_endpoint(request: CodeReviewRequest): 接收代码需求启动多智能体评审流程。 try: # 1. 构建初始状态 initial_state { original_requirement: request.requirement, messages: [], current_code: , review_result: , iteration_count: 0, is_approved: False } # 2. 异步调用图LangGraph 支持异步 # 注意这里需要根据你的图定义可能需要在编译时配置中断条件以支持 max_iterations # 为了简化我们假设图内部已处理最大次数 final_state await app.ainvoke(initial_state) # 3. 构造响应 return CodeReviewResponse( successfinal_state.get(is_approved, False), final_codefinal_state.get(current_code), review_historyfinal_state.get(messages), iteration_countfinal_state.get(iteration_count, 0), message流程执行完毕 if final_state.get(is_approved) else 流程结束但代码未获批准可能已达最大迭代次数 ) except Exception as e: raise HTTPException(status_code500, detailf工作流执行失败: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app_fastapi, host0.0.0.0, port8000)启动服务python api_server.py。现在你可以通过POST /review接口提交代码需求。6.2 批量任务处理对于批量需求可以利用异步批量调用。# batch_processor.py import asyncio from your_workflow_module import app async def process_batch_requirements(requirements_list): 批量处理多个代码需求。 tasks [] for req in requirements_list: initial_state { original_requirement: req, messages: [], current_code: , review_result: , iteration_count: 0, is_approved: False } # 创建异步任务 task app.ainvoke(initial_state) tasks.append(task) # 并发执行所有任务 results await asyncio.gather(*tasks, return_exceptionsTrue) processed_results [] for req, result in zip(requirements_list, results): if isinstance(result, Exception): processed_results.append({requirement: req, error: str(result)}) else: processed_results.append({ requirement: req, approved: result.get(is_approved), final_code: result.get(current_code), iterations: result.get(iteration_count) }) return processed_results # 使用示例 if __name__ __main__: batch_reqs [ 写一个函数计算斐波那契数列第n项。, 写一个函数验证输入的字符串是否是有效的电子邮件格式。, 写一个简单的爬虫获取网页标题。 ] results asyncio.run(process_batch_requirements(batch_reqs)) for res in results: print(res)7. 资源占用与性能观察由于 LangGraph 本身是一个轻量的编排框架其资源占用主要取决于两点你定义的智能体节点函数的复杂度。底层大模型 API 调用的延迟和成本。性能观察要点延迟主要来自大模型 API 的网络往返时间RTT和模型推理时间。使用gpt-4o-mini会比gpt-4快很多。可以在节点函数中添加计时逻辑来监控每个智能体的耗时。并发与吞吐量当你使用abatch进行批量处理时注意 API 的速率限制。需要实现适当的重试和退避机制。内存LangGraph 会在内存中维护整个State对象和历史消息。对于极长的多轮对话需注意内存增长。可以考虑将历史消息定期摘要或持久化到外部存储。图编译时间workflow.compile()在启动时执行一次开销很小可忽略不计。优化建议模型选择在效果和速度/成本间权衡。对于智能体协作反应速度快的模型如gpt-4o-mini,claude-3-haiku往往体验更好。状态设计只将必要的共享信息放入State。避免存储过大的中间文件如图片、长文档可以存储文件路径或引用。异步调用尽可能使用ainvoke和abatch避免在 Web 服务中阻塞主线程。8. 常见问题与排查方法在开发和运行 LangGraph 多智能体应用时你可能会遇到以下问题问题现象可能原因排查方式解决方案导入StateGraph失败langgraph未正确安装或版本不兼容。在终端执行 pip listgrep langgraph和python -c “import langgraph; print(langgraph.version)”。调用app.invoke时报状态字段错误传递给invoke的初始状态字典的键与StateGraph定义时使用的State类型声明不匹配。检查class YourState(TypedDict)定义的字段与initial_state字典的键是否完全一致。确保initial_state包含State类中定义的所有非可选字段。智能体节点函数没有更新状态节点函数返回值中的字典键没有覆盖或添加到State中。打印节点函数的输入state和返回值确认返回值格式。节点函数必须返回一个字典其键是State中定义的字段名值用于更新该字段。条件边 (add_conditional_edges) 不生效1. 条件函数返回的值不在映射字典的键中。2. 条件函数逻辑错误始终返回同一个值。在条件函数should_continue中添加print语句观察其输入和返回值。确保条件函数返回的字符串如”continue”,”end”与add_conditional_edges映射字典的键完全匹配。工作流陷入无限循环退出条件设置不当例如should_continue逻辑永远返回循环路径。在State中添加iteration_count字段并在每次循环时递增在条件函数中判断是否超过最大限制。务必设置安全阀如在State中设置max_iterations在条件边中判断if state[‘iteration_count’] state[‘max_iterations’]: return “end”。大模型 API 调用超时或报错网络问题、API 密钥无效、模型服务不可用、请求速率超限。检查 API 密钥环境变量尝试用简单的ChatOpenAI().invoke(“Hello”)测试连通性。查看模型提供商的状态页。1. 确认网络。2. 检查 API 密钥和额度。3. 在代码中添加重试机制和错误处理。4. 考虑使用更稳定的模型或备用模型。ainvoke异步调用报错未在异步上下文如async def函数内中调用。确保调用ainvoke的函数被async定义并使用await。将调用封装在async函数中并使用asyncio.run()或在其它的异步框架如 FastAPI中执行。9. 最佳实践与使用建议基于实战经验以下建议能帮助你更稳健地使用 LangGraph从简单开始逐步复杂化先构建一个两个节点、一条边的线性图并跑通。再逐步添加条件分支、循环和更多节点。不要一开始就设计过于复杂的图。精心设计 StateState是你的应用的“全局内存”。仔细规划哪些数据需要共享哪些是节点局部变量。使用TypedDict或Pydantic模型来获得类型安全和清晰的结构。为每个节点编写纯净的函数节点函数应尽可能只依赖于输入State和其内部逻辑避免副作用。这便于测试和调试。实现可视化利用LangGraph Studio或手动将图导出为 PNGapp.get_graph().draw_mermaid_png()来可视化你的工作流。这对于理解复杂流程和向他人解释至关重要。加入人工干预点对于关键决策可以设计一个节点其功能是暂停工作流等待人工输入例如通过 API 回调或消息队列。LangGraph 的interrupt机制支持这一点。日志与可观测性在每个节点函数的开始和结束添加详细的日志记录输入状态、输出状态和关键决策。这对于调试复杂工作流和监控运行状态必不可少。版本控制你的图工作流图的定义代码StateGraph的构建部分应该被纳入版本控制。当业务逻辑变更时图的版本也应随之更新。测试策略对单个节点函数进行单元测试。对整个图进行集成测试使用模拟Mock的大模型响应来验证不同分支路径如批准 vs 驳回是否能正确执行。10. 总结与下一步通过本文的实战拆解你应该已经掌握了 LangGraph 构建多智能体系统的核心脉络定义状态、创建节点、构建图表、设置路由、编译运行。我们构建的“开发者-评审者”代码评审系统虽然简单但完整展示了多智能体协作、状态共享和条件循环的核心模式。最值得尝试的下一步替换智能体能力将示例中的“开发者”和“评审者”换成“数据分析师”和“报告撰写员”就能构建一个数据分析流水线。集成外部工具让智能体不仅能“思考”调用 LLM还能“行动”。使用tool装饰器或 LangChain Tool 集成让智能体可以调用搜索引擎、数据库、内部 API 等。探索 LangGraph Studio这是 LangGraph 的可视化开发工具能让你以更直观的方式编辑、调试和监控你的图。实现更复杂的状态管理尝试使用Annotation和Reducer来更精细地控制状态中列表、字典等结构的合并方式。LangGraph 将多智能体系统的复杂性封装在清晰的图抽象之下是构建下一代 AI 应用的有力武器。从今天这个可运行的代码示例开始逐步扩展它的边界你将能设计出真正自主、协同、高效的智能体工作流。建议将本文的代码作为模板收藏在遇到复杂任务编排需求时它或许能提供关键的解决思路。