最近在尝试构建复杂的AI应用时你是否遇到过这样的困境单个大模型能力有限处理复杂任务时逻辑混乱、容易出错想引入多个AI角色协同工作却发现状态管理、流程编排、错误处理等底层逻辑异常复杂代码迅速变得难以维护这正是传统AI应用开发中普遍存在的痛点。LangGraph的出现正是为了解决多智能体系统Multi-Agent System, MAS的工程化难题。它不是一个全新的框架而是对LangChain的强力补充专注于用清晰、可维护的图Graph结构来定义和运行由多个智能体Agent和工具Tool组成的复杂工作流。本文将为你彻底拆解LangGraph从核心概念、架构设计到手把手的代码实战带你构建一个功能完整的本地多智能体系统。无论你是想深入理解Agent架构还是急需一个可落地的项目方案这篇文章都能提供一条清晰的路径。1. LangGraph与多智能体系统核心概念在深入代码之前我们必须厘清几个关键概念这是理解后续所有内容的基础。1.1 什么是多智能体系统MAS多智能体系统是由多个自治的、相互作用的智能体Agent组成的计算系统。每个智能体具备特定的能力、目标和知识它们通过通信、协作或竞争来完成单个智能体无法解决的复杂任务。类比到人类世界就像一个项目团队有项目经理、开发、测试、产品等不同角色各司其职又紧密配合。在AI语境下一个“智能体”通常是一个具备特定指令Prompt和大模型调用能力的模块它可以理解用户意图、调用工具如搜索、计算、写代码、进行推理并做出决策。1.2 LangGraph 是什么它和 LangChain 是什么关系这是最容易混淆的点。LangGraph不是用来替代LangChain的而是构建在LangChain之上专门用于创建有状态、多参与者的工作流。LangChain是一个用于开发由语言模型驱动的应用程序的框架。它提供了大量的模块如Models, Prompts, Chains, Agents, Tools, Memory像“乐高积木”一样你可以用它们组装成各种应用。它的核心抽象是“链”Chain用于顺序执行。LangGraph它的核心抽象是“图”Graph和“状态”State。它将工作流中的每个步骤节点和步骤之间的流转条件边显式地定义出来特别适合包含循环、分支、多角色交互的复杂场景。你可以把LangGraph看作是专门为构建复杂、有状态的智能体工作流而设计的“编排引擎”。简单说LangChain提供了建造智能体的“材料”模型、工具、记忆而LangGraph提供了设计和组装复杂智能体协作流水线的“蓝图”和“控制器”。1.3 LangGraph 的核心优势显式的工作流将复杂的智能体交互逻辑可视化、代码化易于理解、调试和维护。循环与状态原生支持基于条件的循环比如“直到答案满意为止”并能优雅地管理整个工作流运行过程中的状态。多参与者轻松定义多个具有不同角色和能力的智能体并编排它们之间的交互。Human-in-the-Loop内置支持在特定节点暂停等待人工审核或输入这对于关键业务场景至关重要。可持久化与可恢复工作流状态可以保存和加载允许长时间运行的任务或失败后的恢复。2. 环境准备与项目初始化我们将构建一个本地化的多智能体系统使用轻量级的本地大模型避免对网络API的依赖。2.1 环境与工具清单操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 22.04)。本文以macOS/Linux命令行示例为主Windows用户可在PowerShell或WSL中操作。Python版本 3.8 - 3.11。推荐使用3.10或3.11以获得最佳兼容性。包管理工具pip(Python自带) 或poetry(推荐用于项目管理)。本地大模型服务Ollama。这是一个强大的工具可以让你在本地轻松运行如Llama 3.1、Mistral、Qwen2.5等开源模型。IDEVS Code, PyCharm 或任何你熟悉的代码编辑器。2.2 安装 Ollama 并拉取模型首先我们需要一个本地运行的大模型作为智能体的“大脑”。安装Ollama 访问 Ollama 官网下载并安装对应操作系统的版本。安装后打开终端验证ollama --version拉取一个轻量级模型例如 Llama 3.2 的 3B 参数版本对硬件要求低ollama pull llama3.2:3b你也可以选择其他模型如mistral:7b,qwen2.5:7b。模型越大能力越强所需内存也越多7B模型约需14GB内存3B模型约需6GB。测试模型ollama run llama3.2:3b在出现的提示符后输入“Hello”看是否能得到正常回复。按CtrlD退出。2.3 创建项目并安装Python依赖我们创建一个新的项目目录并安装必要的Python库。# 创建项目目录 mkdir langgraph-multi-agent-demo cd langgraph-multi-agent-demo # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install langgraph langchain langchain-community ollama # langgraph: 核心工作流编排 # langchain: 基础框架 # langchain-community: 社区集成包含Ollama集成 # ollama: Ollama的Python客户端2.4 项目结构预览在开始编码前我们先规划一下项目结构这有助于理解代码组织。langgraph-multi-agent-demo/ ├── main.py # 主程序入口定义并运行图 ├── agents/ # 智能体定义模块 │ ├── __init__.py │ ├── planner.py # 规划者智能体 │ ├── researcher.py # 研究者智能体 │ └── writer.py # 写作者智能体 ├── tools/ # 工具定义模块 │ ├── __init__.py │ └── web_search.py # 模拟网络搜索工具 ├── state.py # 定义共享状态Schema └── requirements.txt # 依赖列表可由 pip freeze requirements.txt 生成3. LangGraph 核心组件深度剖析理解LangGraph的四大核心组件是构建任何工作流的关键。3.1 状态State状态是工作流的“记忆体”它是一个贯穿整个图执行过程的共享数据结构。LangGraph使用Pydantic模型或TypedDict来定义状态的结构。作用在不同节点间传递数据。例如用户的原始问题、规划者的计划、研究者的发现、写作者的草稿都存放在状态中。定义方式通常继承自TypedDict并标注每个字段的类型。3.2 节点Node节点是工作流中的基本执行单元一个节点就是一个Python函数或可调用对象。输入当前的状态State。输出对状态State的更新。LangGraph使用“更新字典”的方式只修改需要改变的字段。职责每个节点应只完成一项明确的任务例如“调用规划者智能体”、“执行网络搜索”、“格式化最终答案”。3.3 边Edge边定义了节点之间的流转逻辑即“接下来执行哪个节点”。这是LangGraph实现条件分支和循环的核心。条件边Conditional Edge根据状态的某个值或某个函数的返回值决定下一步走向哪个节点。这是实现“if-else”逻辑的关键。普通边无条件地指向下一个节点。3.4 图Graph图是节点和边的容器它定义了工作流的整体拓扑结构。编译Compile将定义好的节点、边添加进图后需要调用graph.compile()来生成一个可执行的工作流对象。执行通过compiled_graph.invoke(initial_state)来启动工作流。工作流生命周期定义State - 创建Graph - 添加Node - 连接Edge - 编译 - 传入初始State执行。4. 实战构建一个“研究-写作”多智能体系统我们将构建一个模拟的“研究助手”系统。用户提出一个复杂主题如“解释量子计算的基本原理”系统将协调三个智能体完成研究并生成报告规划者Planner分析问题制定研究大纲和步骤。研究者Researcher根据大纲使用工具模拟搜索查找信息。写作者Writer整合研究结果撰写结构清晰、语言流畅的最终报告。4.1 定义共享状态State首先在state.py中定义整个工作流需要共享的数据结构。# state.py from typing import TypedDict, List, Optional, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): 多智能体工作流的共享状态。 # 用户输入的原问题 input: str # 规划者生成的研究计划 plan: Optional[str] # 研究者收集到的资料列表 research_materials: List[str] # 写作者生成的最终报告 final_report: Optional[str] # 用于记录智能体间对话的消息历史LangGraph内置支持 messages: Annotated[list, add_messages]关键解释TypedDict从typing模块导入用于定义类型化的字典让IDE和类型检查器能提供更好的支持。Annotated和add_messages这是LangGraph的一个高级特性用于自动管理消息列表类似聊天历史。add_messages是一个归约器reducer确保对messages列表的更新如追加新消息是线程安全的。这对于多轮对话智能体非常重要。4.2 创建智能体Agents接下来我们在agents/目录下创建三个智能体。每个智能体本质上是一个配备了特定系统提示词System Prompt和聊天模型的LangChainChatPromptTemplate。首先创建一个基础的工具函数来初始化Ollama模型在agents/__init__.py中# agents/__init__.py from langchain_community.chat_models import ChatOllama from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder def create_llm(model_name: str llama3.2:3b): 创建连接到本地Ollama服务的聊天模型。 # 温度temperature控制创造性0.1较低输出更确定max_tokens限制单次生成长度 return ChatOllama(modelmodel_name, temperature0.1, max_tokens2048) def create_agent_prompt(system_message: str): 创建智能体的提示词模板。 return ChatPromptTemplate.from_messages([ (system, system_message), MessagesPlaceholder(variable_namemessages), # 这里会注入状态中的messages ])现在分别创建三个智能体# agents/planner.py from . import create_llm, create_agent_prompt from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 1. 规划者智能体 planner_system_prompt 你是一个资深的项目规划师。你的任务是根据用户提出的复杂问题制定一个清晰、可执行的研究计划。 计划应该分解为具体的步骤并说明每个步骤的目标和可能需要的信息来源。 请直接输出计划内容不要添加“计划”这样的前缀。 planner_prompt create_agent_prompt(planner_system_prompt) planner_llm create_llm() # 构建一个可运行的“链”提示词 - 模型 - 字符串输出解析器 planner_chain planner_prompt | planner_llm | StrOutputParser()# agents/researcher.py from . import create_llm, create_agent_prompt from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 2. 研究者智能体 researcher_system_prompt 你是一个严谨的研究员。你会收到一个具体的研究步骤或问题。 你的职责是调用搜索工具查找相关信息并对找到的信息进行归纳、总结和核实确保其准确性。 请用简洁明了的语言总结你的发现。 researcher_prompt create_agent_prompt(researcher_system_prompt) researcher_llm create_llm() researcher_chain researcher_prompt | researcher_llm | StrOutputParser()# agents/writer.py from . import create_llm, create_agent_prompt from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 3. 写作者智能体 writer_system_prompt 你是一位优秀的科技作家。你将收到用户的问题、研究计划以及一系列研究材料。 你的任务是根据这些材料撰写一份结构完整、逻辑清晰、语言流畅的最终报告。 报告应包括引言、主体根据计划分节和结论。请确保引用研究材料中的关键信息。 writer_prompt create_agent_prompt(writer_system_prompt) writer_llm create_llm() writer_chain writer_prompt | writer_llm | StrOutputParser()4.3 创建模拟工具Tools在真实场景中研究者会调用真实的搜索引擎API如SerperAPI、Tavily。为了简化演示我们创建一个模拟搜索工具。在tools/web_search.py中# tools/web_search.py from langchain.tools import tool import random import time # 模拟一个网络搜索工具 tool def web_search(query: str) - str: 模拟网络搜索。传入搜索查询词返回模拟的搜索结果摘要。 在实际应用中这里应替换为真正的搜索API调用如SerperAPI, Tavily等。 print(f[工具调用] 正在搜索: {query}) # 模拟网络延迟 time.sleep(0.5) # 模拟根据不同查询返回不同的结果 mock_responses { 量子计算原理: 量子计算利用量子比特qubit的叠加和纠缠特性进行信息处理。与传统比特的0或1不同量子比特可以同时处于0和1的叠加态这使得量子计算机在处理特定问题如因数分解、优化时具有指数级优势。基本原理包括叠加态、纠缠、量子门操作和量子测量。, 量子比特类型: 主流量子比特物理实现包括超导量子比特如IBM、Google使用、离子阱量子比特、拓扑量子比特微软研究方向、光量子比特等。每种类型在相干时间、操控精度和可扩展性上各有优劣。, 量子算法例子: 著名的量子算法有Shor算法用于大数质因数分解威胁RSA加密、Grover算法用于无序数据库搜索提供平方级加速、量子模拟算法用于模拟分子和材料特性。, default: f关于{query}的搜索结果这是一个模拟的搜索结果。在实际系统中这里会包含从互联网获取的真实、最新的相关信息摘要。建议结合权威教科书和最新研究论文进行深入理解。 } for key in mock_responses: if key in query: return mock_responses[key] return mock_responses[default]关键解释tool装饰器来自LangChain它将一个普通Python函数转换为LangChain智能体可以识别的“工具”。工具的作用智能体可以通过描述来知道工具的功能并在需要时决定调用它。研究者智能体将学习调用这个工具。4.4 构建研究者节点集成工具调用研究者智能体需要能够自主决定何时调用搜索工具。我们需要创建一个更复杂的“工具调用链”。更新agents/researcher.py# agents/researcher.py (更新版) from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from . import create_llm # 假设web_search工具已经导入或可通过其他方式获取 from tools.web_search import web_search def create_researcher_agent_executor(): 创建集成了工具调用能力的研究者智能体执行器。 llm create_llm() # 研究者可以使用的工具列表 tools [web_search] # 智能体提示词包含工具描述 prompt ChatPromptTemplate.from_messages([ (system, researcher_system_prompt), # 使用之前定义的system_prompt (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 用于记录工具调用和结果的历史 ]) # 创建支持工具调用的智能体 agent create_tool_calling_agent(llmllm, toolstools, promptprompt) # 创建执行器负责运行智能体并处理工具调用 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) return agent_executor # 创建一个全局的执行器实例 researcher_agent_executor create_researcher_agent_executor()4.5 定义工作流节点Nodes现在我们将在主文件main.py中定义图的工作节点。每个节点是一个函数它接收状态更新状态并返回更新后的状态。# main.py from typing import Dict, Any from langgraph.graph import StateGraph, END from state import State from agents.planner import planner_chain from agents.researcher import researcher_agent_executor from agents.writer import writer_chain # ---------- 节点函数定义 ---------- def planner_node(state: State) - Dict[str, Any]: 规划者节点分析问题制定计划。 print(\n 规划者开始工作 ) # 从状态中获取用户输入 user_input state[input] # 准备给规划者的消息历史初始只有用户输入 messages [{role: user, content: f请为以下问题制定研究计划\n{user_input}}] # 调用规划者链 plan planner_chain.invoke({messages: messages}) print(f生成的计划\n{plan}) # 返回要更新的状态字段 return {plan: plan, messages: messages [{role: assistant, content: plan}]} def researcher_node(state: State) - Dict[str, Any]: 研究者节点执行计划中的研究步骤这里简化为一个步骤。 print(\n 研究者开始工作 ) plan state[plan] # 在实际应用中这里可以解析计划拆分成多个研究子任务循环执行。 # 本例简化将整个计划作为一个研究查询。 research_query f根据以下研究计划查找相关资料{plan} # 调用研究者智能体执行器它会自动决定是否及如何调用工具 result researcher_agent_executor.invoke({input: research_query}) research_findings result[output] print(f研究结果\n{research_findings}) # 将研究发现添加到材料列表 current_materials state.get(research_materials, []) new_materials current_materials [research_findings] return {research_materials: new_materials, messages: state[messages] [{role: user, content: research_query}, {role: assistant, content: research_findings}]} def writer_node(state: State) - Dict[str, Any]: 写作者节点整合所有信息生成最终报告。 print(\n 写作者开始工作 ) user_input state[input] plan state[plan] materials state[research_materials] # 准备给写作者的上下文 context f 原始问题{user_input} 研究计划{plan} 收集到的研究材料 {chr(10).join([f- {m} for m in materials])} 请基于以上信息撰写最终报告。 messages [{role: user, content: context}] final_report writer_chain.invoke({messages: messages}) print(f生成的最终报告\n{final_report}) return {final_report: final_report} def human_review_node(state: State) - Dict[str, Any]: 人工审核节点示例在此暂停等待人工输入。 # 在实际应用中这里可以连接到一个Web界面、发送邮件或等待命令行输入。 print(\n 等待人工审核 ) print(f当前生成的报告草稿\n{state.get(final_report, 无)}) # 模拟人工输入例如从某个接口获取 human_feedback 内容不错请在结论部分补充一下当前面临的挑战。 # 这里应是动态获取的 print(f收到人工反馈{human_feedback}) # 将反馈加入状态后续节点可以处理 return {human_feedback: human_feedback}4.6 构建并连接图Graph在main.py中继续添加图的构建逻辑# main.py (续) # ---------- 构建图 ---------- def create_workflow_graph(): 创建并编译研究-写作工作流图。 # 1. 初始化一个状态图指定状态类型 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(researcher, researcher_node) workflow.add_node(writer, writer_node) workflow.add_node(human_review, human_review_node) # 可选的人工审核节点 # 3. 设置入口点 workflow.set_entry_point(planner) # 4. 连接边定义执行流程 workflow.add_edge(planner, researcher) # 规划完就去研究 workflow.add_edge(researcher, writer) # 研究完就去写作 # 5. 添加条件边示例根据报告长度决定是否需要人工审核 # 定义一个路由函数 def should_review(state: State) - str: report state.get(final_report, ) # 简单逻辑如果报告超过500字符则进入人工审核 if len(report) 500: return human_review else: return END # 直接结束 # 从writer节点出来后根据条件决定下一步 workflow.add_conditional_edges( writer, should_review, # 路由函数 { human_review: human_review, # 如果返回human_review则跳转到human_review节点 END: END # 如果返回END则直接结束图 } ) # 6. 从人工审核节点连接到结束或者可以连接回writer进行修改 workflow.add_edge(human_review, END) # 7. 编译图 compiled_graph workflow.compile() return compiled_graph # ---------- 主程序 ---------- if __name__ __main__: # 创建图 graph create_workflow_graph() # 定义初始状态 initial_state: State { input: 请解释量子计算的基本原理、主要量子比特类型以及一个著名的量子算法例子。, plan: None, research_materials: [], final_report: None, messages: [] # 初始消息为空 } print( * 50) print(开始执行多智能体工作流...) print( * 50) # 执行图 try: final_state graph.invoke(initial_state) print(\n * 50) print(工作流执行完成) print( * 50) print(\n最终报告) print(- * 30) print(final_state[final_report]) except Exception as e: print(f\n工作流执行出错{e})4.7 运行与验证保存所有文件后在项目根目录下运行python main.py你应该会在终端看到类似以下的输出具体内容因模型随机性而异 开始执行多智能体工作流... 规划者开始工作 生成的计划 1. 理解问题核心明确量子计算基本原理、量子比特类型和算法实例三个部分。 2. 搜集基本原理资料包括叠加态、纠缠、量子门、测量等概念。 3. 调研量子比特物理实现超导、离子阱、拓扑、光量子等。 4. 查找著名量子算法以Shor算法或Grover算法为例解释其原理和意义。 5. 信息整合与验证交叉核对不同来源确保准确性。 研究者开始工作 [工具调用] 正在搜索: 根据以下研究计划查找相关资料1. 理解问题核心...计划内容 ... 研究结果 根据搜索量子计算原理基于...模拟的搜索结果摘要 写作者开始工作 生成的最终报告 关于量子计算的综合报告 引言量子计算是...生成的完整报告5. 核心机制与高级特性详解通过上面的实战你已经搭建了一个基础的多智能体工作流。下面我们深入剖析LangGraph的几个高级特性和核心机制。5.1 状态管理State Management状态是LangGraph的“脊柱”。每个节点函数接收完整的当前状态但只返回需要更新的部分一个字典。LangGraph内部会将这些更新安全地合并到全局状态中。这种模式类似于React/Redux中的reducer使得每个节点职责单一且状态变更可预测。5.2 条件路由与循环Conditional Routing Loopingadd_conditional_edges是实现复杂逻辑的关键。路由函数 (should_review) 检查状态并返回下一个节点的名称。你可以用它实现循环例如研究者节点完成后检查材料是否充足。如果不足返回researcher节点让其继续搜索直到满足条件。分支根据用户问题的类型如“编程” vs “写作”路由到不同的专家智能体节点。5.3 Human-in-the-Loop人工介入human_review_node展示了人工介入的简单模式。在实际生产环境中这个节点可以将状态持久化到数据库。向审核系统发送通知邮件、Slack、内部工单。等待外部API回调包含人工审核结果。收到结果后更新状态并让工作流继续执行例如回到writer节点进行修改。5.4 持久化与检查点Persistence Checkpoints对于长时间运行的工作流LangGraph支持将状态和图的执行进度保存到外部存储数据库、Redis、文件。这允许暂停与恢复工作流可以在人工审核节点暂停几天后从该点恢复。错误恢复如果系统崩溃可以从上一个检查点重启而不是从头开始。调试与审计完整记录工作流的每一步状态变化。5.5 多智能体协作模式我们的例子是顺序管道式Planner - Researcher - Writer。LangGraph还支持更复杂的模式广播/聚合式一个协调者节点将任务分发给多个并行的专家节点然后聚合结果。辩论式多个智能体就一个问题提出不同观点另一个智能体进行评判和总结。竞争式多个智能体尝试解决同一问题选择最优解。6. 常见问题与排查思路在开发和运行LangGraph应用时你可能会遇到以下问题问题现象可能原因排查思路与解决方案GraphCompilationError或节点未定义1. 节点函数未用add_node添加到图中。2. 边引用了不存在的节点名。1. 检查workflow.add_node调用确保所有节点都已添加。2. 检查add_edge和add_conditional_edges中的节点名称拼写。状态字段更新不生效节点函数返回的字典键名与State中定义的字段名不匹配。确保返回的字典键如{plan: plan}与StateTypedDict 中的字段名完全一致。Ollama连接超时或模型未加载1. Ollama服务未启动。2. 指定的模型未通过ollama pull下载。1. 终端运行ollama serve确保服务在运行。2. 运行ollama list确认模型存在或用ollama pull model_name下载。智能体不调用工具1. 工具未正确传递给智能体执行器。2. 模型能力不足无法理解工具描述。3. 提示词未引导模型使用工具。1. 检查create_tool_calling_agent和AgentExecutor的tools参数。2. 尝试更大的模型如7B, 13B。3. 在系统提示词中明确要求“使用搜索工具查找信息”。工作流陷入无限循环条件边逻辑有误导致在几个节点间来回跳转无法满足结束条件。1. 在条件路由函数中添加打印语句调试返回值。2. 确保存在一条路径能到达END。3. 考虑设置最大循环次数在状态中维护一个step_count并在节点中检查。内存占用过高1. 模型太大。2. 状态中累积了过多历史消息或中间数据。1. 使用更小的模型或量化版模型。2. 定期清理状态中非必需的历史数据或使用LangGraph的“内存管理”功能限制消息历史长度。7. 工程最佳实践与进阶建议当你准备将LangGraph多智能体系统用于实际项目时请考虑以下建议7.1 项目结构组织按功能模块化就像我们示例中的agents/,tools/,state.py。将节点函数、智能体定义、工具、状态Schema分开提高可读性和可维护性。配置外部化将模型名称、API密钥、超时时间等配置项放入.env文件或配置类中不要硬编码。依赖注入考虑使用依赖注入框架如dependency-injector来管理智能体、工具等组件的创建便于测试和切换实现。7.2 智能体设计明确职责单一每个智能体应专注于一个特定的任务。避免创建“全能”智能体。设计清晰的提示词系统提示词是指引智能体行为的关键。明确角色、职责、输出格式和约束条件。迭代优化提示词是提升效果性价比最高的方式。工具设计规范化工具函数应有清晰的文档字符串Docstring描述其功能和参数。这能帮助大模型更好地理解何时以及如何调用它。7.3 状态与流程设计最小化状态只在状态中保存必要的数据。过大的状态会影响性能且难以调试。规划错误处理路径在图设计中考虑错误节点。当某个节点执行失败时可以路由到一个专门的error_handler节点记录日志、发送警报或尝试恢复。实现超时与重试对于调用外部API或耗时长的节点在节点逻辑或执行器层面添加超时和重试机制。7.4 可观测性与监控全面日志记录在每个节点的开始、结束、关键决策点记录日志。区分INFO,DEBUG,WARNING等级别。可视化工作流LangGraph提供了graph.get_graph().draw_mermaid_png()等方法可以将你定义的图生成Mermaid图表便于团队理解和沟通架构。追踪与评估集成像LangSmith这样的平台可以追踪每次调用、查看链的输入输出、分析延迟和成本并对智能体的输出进行质量评估。7.5 生产环境部署容器化使用Docker将你的应用及其依赖包括Ollama打包确保环境一致性。异步执行对于I/O密集型操作如网络请求考虑使用asyncio和LangChain的异步接口提升并发性能。状态持久化集成数据库如PostgreSQL, Redis来持久化工作流状态支持长时间运行和故障恢复。API化使用FastAPI或Flask将你的工作流包装成REST API方便与其他系统集成。从理解多智能体系统的价值到掌握LangGraph以图为核心的全新编程范式再到亲手搭建一个具备规划、研究、写作能力的协同系统你已经走完了从概念到实战的关键一步。LangGraph将复杂的智能体协作逻辑变得直观和可管理这是构建下一代AI应用的基础设施。真正的掌握源于实践和迭代。建议你以本文的示例为起点尝试以下挑战为研究者添加真实的搜索API如Serper Dev引入一个“评审者”智能体来评估报告质量并触发修改循环或者设计一个处理客户服务请求的多智能体流程包含“理解意图”、“查询知识库”、“生成回复”、“情感安抚”等节点。
LangGraph实战：构建本地多智能体AI协作系统
最近在尝试构建复杂的AI应用时你是否遇到过这样的困境单个大模型能力有限处理复杂任务时逻辑混乱、容易出错想引入多个AI角色协同工作却发现状态管理、流程编排、错误处理等底层逻辑异常复杂代码迅速变得难以维护这正是传统AI应用开发中普遍存在的痛点。LangGraph的出现正是为了解决多智能体系统Multi-Agent System, MAS的工程化难题。它不是一个全新的框架而是对LangChain的强力补充专注于用清晰、可维护的图Graph结构来定义和运行由多个智能体Agent和工具Tool组成的复杂工作流。本文将为你彻底拆解LangGraph从核心概念、架构设计到手把手的代码实战带你构建一个功能完整的本地多智能体系统。无论你是想深入理解Agent架构还是急需一个可落地的项目方案这篇文章都能提供一条清晰的路径。1. LangGraph与多智能体系统核心概念在深入代码之前我们必须厘清几个关键概念这是理解后续所有内容的基础。1.1 什么是多智能体系统MAS多智能体系统是由多个自治的、相互作用的智能体Agent组成的计算系统。每个智能体具备特定的能力、目标和知识它们通过通信、协作或竞争来完成单个智能体无法解决的复杂任务。类比到人类世界就像一个项目团队有项目经理、开发、测试、产品等不同角色各司其职又紧密配合。在AI语境下一个“智能体”通常是一个具备特定指令Prompt和大模型调用能力的模块它可以理解用户意图、调用工具如搜索、计算、写代码、进行推理并做出决策。1.2 LangGraph 是什么它和 LangChain 是什么关系这是最容易混淆的点。LangGraph不是用来替代LangChain的而是构建在LangChain之上专门用于创建有状态、多参与者的工作流。LangChain是一个用于开发由语言模型驱动的应用程序的框架。它提供了大量的模块如Models, Prompts, Chains, Agents, Tools, Memory像“乐高积木”一样你可以用它们组装成各种应用。它的核心抽象是“链”Chain用于顺序执行。LangGraph它的核心抽象是“图”Graph和“状态”State。它将工作流中的每个步骤节点和步骤之间的流转条件边显式地定义出来特别适合包含循环、分支、多角色交互的复杂场景。你可以把LangGraph看作是专门为构建复杂、有状态的智能体工作流而设计的“编排引擎”。简单说LangChain提供了建造智能体的“材料”模型、工具、记忆而LangGraph提供了设计和组装复杂智能体协作流水线的“蓝图”和“控制器”。1.3 LangGraph 的核心优势显式的工作流将复杂的智能体交互逻辑可视化、代码化易于理解、调试和维护。循环与状态原生支持基于条件的循环比如“直到答案满意为止”并能优雅地管理整个工作流运行过程中的状态。多参与者轻松定义多个具有不同角色和能力的智能体并编排它们之间的交互。Human-in-the-Loop内置支持在特定节点暂停等待人工审核或输入这对于关键业务场景至关重要。可持久化与可恢复工作流状态可以保存和加载允许长时间运行的任务或失败后的恢复。2. 环境准备与项目初始化我们将构建一个本地化的多智能体系统使用轻量级的本地大模型避免对网络API的依赖。2.1 环境与工具清单操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 22.04)。本文以macOS/Linux命令行示例为主Windows用户可在PowerShell或WSL中操作。Python版本 3.8 - 3.11。推荐使用3.10或3.11以获得最佳兼容性。包管理工具pip(Python自带) 或poetry(推荐用于项目管理)。本地大模型服务Ollama。这是一个强大的工具可以让你在本地轻松运行如Llama 3.1、Mistral、Qwen2.5等开源模型。IDEVS Code, PyCharm 或任何你熟悉的代码编辑器。2.2 安装 Ollama 并拉取模型首先我们需要一个本地运行的大模型作为智能体的“大脑”。安装Ollama 访问 Ollama 官网下载并安装对应操作系统的版本。安装后打开终端验证ollama --version拉取一个轻量级模型例如 Llama 3.2 的 3B 参数版本对硬件要求低ollama pull llama3.2:3b你也可以选择其他模型如mistral:7b,qwen2.5:7b。模型越大能力越强所需内存也越多7B模型约需14GB内存3B模型约需6GB。测试模型ollama run llama3.2:3b在出现的提示符后输入“Hello”看是否能得到正常回复。按CtrlD退出。2.3 创建项目并安装Python依赖我们创建一个新的项目目录并安装必要的Python库。# 创建项目目录 mkdir langgraph-multi-agent-demo cd langgraph-multi-agent-demo # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install langgraph langchain langchain-community ollama # langgraph: 核心工作流编排 # langchain: 基础框架 # langchain-community: 社区集成包含Ollama集成 # ollama: Ollama的Python客户端2.4 项目结构预览在开始编码前我们先规划一下项目结构这有助于理解代码组织。langgraph-multi-agent-demo/ ├── main.py # 主程序入口定义并运行图 ├── agents/ # 智能体定义模块 │ ├── __init__.py │ ├── planner.py # 规划者智能体 │ ├── researcher.py # 研究者智能体 │ └── writer.py # 写作者智能体 ├── tools/ # 工具定义模块 │ ├── __init__.py │ └── web_search.py # 模拟网络搜索工具 ├── state.py # 定义共享状态Schema └── requirements.txt # 依赖列表可由 pip freeze requirements.txt 生成3. LangGraph 核心组件深度剖析理解LangGraph的四大核心组件是构建任何工作流的关键。3.1 状态State状态是工作流的“记忆体”它是一个贯穿整个图执行过程的共享数据结构。LangGraph使用Pydantic模型或TypedDict来定义状态的结构。作用在不同节点间传递数据。例如用户的原始问题、规划者的计划、研究者的发现、写作者的草稿都存放在状态中。定义方式通常继承自TypedDict并标注每个字段的类型。3.2 节点Node节点是工作流中的基本执行单元一个节点就是一个Python函数或可调用对象。输入当前的状态State。输出对状态State的更新。LangGraph使用“更新字典”的方式只修改需要改变的字段。职责每个节点应只完成一项明确的任务例如“调用规划者智能体”、“执行网络搜索”、“格式化最终答案”。3.3 边Edge边定义了节点之间的流转逻辑即“接下来执行哪个节点”。这是LangGraph实现条件分支和循环的核心。条件边Conditional Edge根据状态的某个值或某个函数的返回值决定下一步走向哪个节点。这是实现“if-else”逻辑的关键。普通边无条件地指向下一个节点。3.4 图Graph图是节点和边的容器它定义了工作流的整体拓扑结构。编译Compile将定义好的节点、边添加进图后需要调用graph.compile()来生成一个可执行的工作流对象。执行通过compiled_graph.invoke(initial_state)来启动工作流。工作流生命周期定义State - 创建Graph - 添加Node - 连接Edge - 编译 - 传入初始State执行。4. 实战构建一个“研究-写作”多智能体系统我们将构建一个模拟的“研究助手”系统。用户提出一个复杂主题如“解释量子计算的基本原理”系统将协调三个智能体完成研究并生成报告规划者Planner分析问题制定研究大纲和步骤。研究者Researcher根据大纲使用工具模拟搜索查找信息。写作者Writer整合研究结果撰写结构清晰、语言流畅的最终报告。4.1 定义共享状态State首先在state.py中定义整个工作流需要共享的数据结构。# state.py from typing import TypedDict, List, Optional, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): 多智能体工作流的共享状态。 # 用户输入的原问题 input: str # 规划者生成的研究计划 plan: Optional[str] # 研究者收集到的资料列表 research_materials: List[str] # 写作者生成的最终报告 final_report: Optional[str] # 用于记录智能体间对话的消息历史LangGraph内置支持 messages: Annotated[list, add_messages]关键解释TypedDict从typing模块导入用于定义类型化的字典让IDE和类型检查器能提供更好的支持。Annotated和add_messages这是LangGraph的一个高级特性用于自动管理消息列表类似聊天历史。add_messages是一个归约器reducer确保对messages列表的更新如追加新消息是线程安全的。这对于多轮对话智能体非常重要。4.2 创建智能体Agents接下来我们在agents/目录下创建三个智能体。每个智能体本质上是一个配备了特定系统提示词System Prompt和聊天模型的LangChainChatPromptTemplate。首先创建一个基础的工具函数来初始化Ollama模型在agents/__init__.py中# agents/__init__.py from langchain_community.chat_models import ChatOllama from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder def create_llm(model_name: str llama3.2:3b): 创建连接到本地Ollama服务的聊天模型。 # 温度temperature控制创造性0.1较低输出更确定max_tokens限制单次生成长度 return ChatOllama(modelmodel_name, temperature0.1, max_tokens2048) def create_agent_prompt(system_message: str): 创建智能体的提示词模板。 return ChatPromptTemplate.from_messages([ (system, system_message), MessagesPlaceholder(variable_namemessages), # 这里会注入状态中的messages ])现在分别创建三个智能体# agents/planner.py from . import create_llm, create_agent_prompt from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 1. 规划者智能体 planner_system_prompt 你是一个资深的项目规划师。你的任务是根据用户提出的复杂问题制定一个清晰、可执行的研究计划。 计划应该分解为具体的步骤并说明每个步骤的目标和可能需要的信息来源。 请直接输出计划内容不要添加“计划”这样的前缀。 planner_prompt create_agent_prompt(planner_system_prompt) planner_llm create_llm() # 构建一个可运行的“链”提示词 - 模型 - 字符串输出解析器 planner_chain planner_prompt | planner_llm | StrOutputParser()# agents/researcher.py from . import create_llm, create_agent_prompt from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 2. 研究者智能体 researcher_system_prompt 你是一个严谨的研究员。你会收到一个具体的研究步骤或问题。 你的职责是调用搜索工具查找相关信息并对找到的信息进行归纳、总结和核实确保其准确性。 请用简洁明了的语言总结你的发现。 researcher_prompt create_agent_prompt(researcher_system_prompt) researcher_llm create_llm() researcher_chain researcher_prompt | researcher_llm | StrOutputParser()# agents/writer.py from . import create_llm, create_agent_prompt from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 3. 写作者智能体 writer_system_prompt 你是一位优秀的科技作家。你将收到用户的问题、研究计划以及一系列研究材料。 你的任务是根据这些材料撰写一份结构完整、逻辑清晰、语言流畅的最终报告。 报告应包括引言、主体根据计划分节和结论。请确保引用研究材料中的关键信息。 writer_prompt create_agent_prompt(writer_system_prompt) writer_llm create_llm() writer_chain writer_prompt | writer_llm | StrOutputParser()4.3 创建模拟工具Tools在真实场景中研究者会调用真实的搜索引擎API如SerperAPI、Tavily。为了简化演示我们创建一个模拟搜索工具。在tools/web_search.py中# tools/web_search.py from langchain.tools import tool import random import time # 模拟一个网络搜索工具 tool def web_search(query: str) - str: 模拟网络搜索。传入搜索查询词返回模拟的搜索结果摘要。 在实际应用中这里应替换为真正的搜索API调用如SerperAPI, Tavily等。 print(f[工具调用] 正在搜索: {query}) # 模拟网络延迟 time.sleep(0.5) # 模拟根据不同查询返回不同的结果 mock_responses { 量子计算原理: 量子计算利用量子比特qubit的叠加和纠缠特性进行信息处理。与传统比特的0或1不同量子比特可以同时处于0和1的叠加态这使得量子计算机在处理特定问题如因数分解、优化时具有指数级优势。基本原理包括叠加态、纠缠、量子门操作和量子测量。, 量子比特类型: 主流量子比特物理实现包括超导量子比特如IBM、Google使用、离子阱量子比特、拓扑量子比特微软研究方向、光量子比特等。每种类型在相干时间、操控精度和可扩展性上各有优劣。, 量子算法例子: 著名的量子算法有Shor算法用于大数质因数分解威胁RSA加密、Grover算法用于无序数据库搜索提供平方级加速、量子模拟算法用于模拟分子和材料特性。, default: f关于{query}的搜索结果这是一个模拟的搜索结果。在实际系统中这里会包含从互联网获取的真实、最新的相关信息摘要。建议结合权威教科书和最新研究论文进行深入理解。 } for key in mock_responses: if key in query: return mock_responses[key] return mock_responses[default]关键解释tool装饰器来自LangChain它将一个普通Python函数转换为LangChain智能体可以识别的“工具”。工具的作用智能体可以通过描述来知道工具的功能并在需要时决定调用它。研究者智能体将学习调用这个工具。4.4 构建研究者节点集成工具调用研究者智能体需要能够自主决定何时调用搜索工具。我们需要创建一个更复杂的“工具调用链”。更新agents/researcher.py# agents/researcher.py (更新版) from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from . import create_llm # 假设web_search工具已经导入或可通过其他方式获取 from tools.web_search import web_search def create_researcher_agent_executor(): 创建集成了工具调用能力的研究者智能体执行器。 llm create_llm() # 研究者可以使用的工具列表 tools [web_search] # 智能体提示词包含工具描述 prompt ChatPromptTemplate.from_messages([ (system, researcher_system_prompt), # 使用之前定义的system_prompt (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 用于记录工具调用和结果的历史 ]) # 创建支持工具调用的智能体 agent create_tool_calling_agent(llmllm, toolstools, promptprompt) # 创建执行器负责运行智能体并处理工具调用 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) return agent_executor # 创建一个全局的执行器实例 researcher_agent_executor create_researcher_agent_executor()4.5 定义工作流节点Nodes现在我们将在主文件main.py中定义图的工作节点。每个节点是一个函数它接收状态更新状态并返回更新后的状态。# main.py from typing import Dict, Any from langgraph.graph import StateGraph, END from state import State from agents.planner import planner_chain from agents.researcher import researcher_agent_executor from agents.writer import writer_chain # ---------- 节点函数定义 ---------- def planner_node(state: State) - Dict[str, Any]: 规划者节点分析问题制定计划。 print(\n 规划者开始工作 ) # 从状态中获取用户输入 user_input state[input] # 准备给规划者的消息历史初始只有用户输入 messages [{role: user, content: f请为以下问题制定研究计划\n{user_input}}] # 调用规划者链 plan planner_chain.invoke({messages: messages}) print(f生成的计划\n{plan}) # 返回要更新的状态字段 return {plan: plan, messages: messages [{role: assistant, content: plan}]} def researcher_node(state: State) - Dict[str, Any]: 研究者节点执行计划中的研究步骤这里简化为一个步骤。 print(\n 研究者开始工作 ) plan state[plan] # 在实际应用中这里可以解析计划拆分成多个研究子任务循环执行。 # 本例简化将整个计划作为一个研究查询。 research_query f根据以下研究计划查找相关资料{plan} # 调用研究者智能体执行器它会自动决定是否及如何调用工具 result researcher_agent_executor.invoke({input: research_query}) research_findings result[output] print(f研究结果\n{research_findings}) # 将研究发现添加到材料列表 current_materials state.get(research_materials, []) new_materials current_materials [research_findings] return {research_materials: new_materials, messages: state[messages] [{role: user, content: research_query}, {role: assistant, content: research_findings}]} def writer_node(state: State) - Dict[str, Any]: 写作者节点整合所有信息生成最终报告。 print(\n 写作者开始工作 ) user_input state[input] plan state[plan] materials state[research_materials] # 准备给写作者的上下文 context f 原始问题{user_input} 研究计划{plan} 收集到的研究材料 {chr(10).join([f- {m} for m in materials])} 请基于以上信息撰写最终报告。 messages [{role: user, content: context}] final_report writer_chain.invoke({messages: messages}) print(f生成的最终报告\n{final_report}) return {final_report: final_report} def human_review_node(state: State) - Dict[str, Any]: 人工审核节点示例在此暂停等待人工输入。 # 在实际应用中这里可以连接到一个Web界面、发送邮件或等待命令行输入。 print(\n 等待人工审核 ) print(f当前生成的报告草稿\n{state.get(final_report, 无)}) # 模拟人工输入例如从某个接口获取 human_feedback 内容不错请在结论部分补充一下当前面临的挑战。 # 这里应是动态获取的 print(f收到人工反馈{human_feedback}) # 将反馈加入状态后续节点可以处理 return {human_feedback: human_feedback}4.6 构建并连接图Graph在main.py中继续添加图的构建逻辑# main.py (续) # ---------- 构建图 ---------- def create_workflow_graph(): 创建并编译研究-写作工作流图。 # 1. 初始化一个状态图指定状态类型 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(researcher, researcher_node) workflow.add_node(writer, writer_node) workflow.add_node(human_review, human_review_node) # 可选的人工审核节点 # 3. 设置入口点 workflow.set_entry_point(planner) # 4. 连接边定义执行流程 workflow.add_edge(planner, researcher) # 规划完就去研究 workflow.add_edge(researcher, writer) # 研究完就去写作 # 5. 添加条件边示例根据报告长度决定是否需要人工审核 # 定义一个路由函数 def should_review(state: State) - str: report state.get(final_report, ) # 简单逻辑如果报告超过500字符则进入人工审核 if len(report) 500: return human_review else: return END # 直接结束 # 从writer节点出来后根据条件决定下一步 workflow.add_conditional_edges( writer, should_review, # 路由函数 { human_review: human_review, # 如果返回human_review则跳转到human_review节点 END: END # 如果返回END则直接结束图 } ) # 6. 从人工审核节点连接到结束或者可以连接回writer进行修改 workflow.add_edge(human_review, END) # 7. 编译图 compiled_graph workflow.compile() return compiled_graph # ---------- 主程序 ---------- if __name__ __main__: # 创建图 graph create_workflow_graph() # 定义初始状态 initial_state: State { input: 请解释量子计算的基本原理、主要量子比特类型以及一个著名的量子算法例子。, plan: None, research_materials: [], final_report: None, messages: [] # 初始消息为空 } print( * 50) print(开始执行多智能体工作流...) print( * 50) # 执行图 try: final_state graph.invoke(initial_state) print(\n * 50) print(工作流执行完成) print( * 50) print(\n最终报告) print(- * 30) print(final_state[final_report]) except Exception as e: print(f\n工作流执行出错{e})4.7 运行与验证保存所有文件后在项目根目录下运行python main.py你应该会在终端看到类似以下的输出具体内容因模型随机性而异 开始执行多智能体工作流... 规划者开始工作 生成的计划 1. 理解问题核心明确量子计算基本原理、量子比特类型和算法实例三个部分。 2. 搜集基本原理资料包括叠加态、纠缠、量子门、测量等概念。 3. 调研量子比特物理实现超导、离子阱、拓扑、光量子等。 4. 查找著名量子算法以Shor算法或Grover算法为例解释其原理和意义。 5. 信息整合与验证交叉核对不同来源确保准确性。 研究者开始工作 [工具调用] 正在搜索: 根据以下研究计划查找相关资料1. 理解问题核心...计划内容 ... 研究结果 根据搜索量子计算原理基于...模拟的搜索结果摘要 写作者开始工作 生成的最终报告 关于量子计算的综合报告 引言量子计算是...生成的完整报告5. 核心机制与高级特性详解通过上面的实战你已经搭建了一个基础的多智能体工作流。下面我们深入剖析LangGraph的几个高级特性和核心机制。5.1 状态管理State Management状态是LangGraph的“脊柱”。每个节点函数接收完整的当前状态但只返回需要更新的部分一个字典。LangGraph内部会将这些更新安全地合并到全局状态中。这种模式类似于React/Redux中的reducer使得每个节点职责单一且状态变更可预测。5.2 条件路由与循环Conditional Routing Loopingadd_conditional_edges是实现复杂逻辑的关键。路由函数 (should_review) 检查状态并返回下一个节点的名称。你可以用它实现循环例如研究者节点完成后检查材料是否充足。如果不足返回researcher节点让其继续搜索直到满足条件。分支根据用户问题的类型如“编程” vs “写作”路由到不同的专家智能体节点。5.3 Human-in-the-Loop人工介入human_review_node展示了人工介入的简单模式。在实际生产环境中这个节点可以将状态持久化到数据库。向审核系统发送通知邮件、Slack、内部工单。等待外部API回调包含人工审核结果。收到结果后更新状态并让工作流继续执行例如回到writer节点进行修改。5.4 持久化与检查点Persistence Checkpoints对于长时间运行的工作流LangGraph支持将状态和图的执行进度保存到外部存储数据库、Redis、文件。这允许暂停与恢复工作流可以在人工审核节点暂停几天后从该点恢复。错误恢复如果系统崩溃可以从上一个检查点重启而不是从头开始。调试与审计完整记录工作流的每一步状态变化。5.5 多智能体协作模式我们的例子是顺序管道式Planner - Researcher - Writer。LangGraph还支持更复杂的模式广播/聚合式一个协调者节点将任务分发给多个并行的专家节点然后聚合结果。辩论式多个智能体就一个问题提出不同观点另一个智能体进行评判和总结。竞争式多个智能体尝试解决同一问题选择最优解。6. 常见问题与排查思路在开发和运行LangGraph应用时你可能会遇到以下问题问题现象可能原因排查思路与解决方案GraphCompilationError或节点未定义1. 节点函数未用add_node添加到图中。2. 边引用了不存在的节点名。1. 检查workflow.add_node调用确保所有节点都已添加。2. 检查add_edge和add_conditional_edges中的节点名称拼写。状态字段更新不生效节点函数返回的字典键名与State中定义的字段名不匹配。确保返回的字典键如{plan: plan}与StateTypedDict 中的字段名完全一致。Ollama连接超时或模型未加载1. Ollama服务未启动。2. 指定的模型未通过ollama pull下载。1. 终端运行ollama serve确保服务在运行。2. 运行ollama list确认模型存在或用ollama pull model_name下载。智能体不调用工具1. 工具未正确传递给智能体执行器。2. 模型能力不足无法理解工具描述。3. 提示词未引导模型使用工具。1. 检查create_tool_calling_agent和AgentExecutor的tools参数。2. 尝试更大的模型如7B, 13B。3. 在系统提示词中明确要求“使用搜索工具查找信息”。工作流陷入无限循环条件边逻辑有误导致在几个节点间来回跳转无法满足结束条件。1. 在条件路由函数中添加打印语句调试返回值。2. 确保存在一条路径能到达END。3. 考虑设置最大循环次数在状态中维护一个step_count并在节点中检查。内存占用过高1. 模型太大。2. 状态中累积了过多历史消息或中间数据。1. 使用更小的模型或量化版模型。2. 定期清理状态中非必需的历史数据或使用LangGraph的“内存管理”功能限制消息历史长度。7. 工程最佳实践与进阶建议当你准备将LangGraph多智能体系统用于实际项目时请考虑以下建议7.1 项目结构组织按功能模块化就像我们示例中的agents/,tools/,state.py。将节点函数、智能体定义、工具、状态Schema分开提高可读性和可维护性。配置外部化将模型名称、API密钥、超时时间等配置项放入.env文件或配置类中不要硬编码。依赖注入考虑使用依赖注入框架如dependency-injector来管理智能体、工具等组件的创建便于测试和切换实现。7.2 智能体设计明确职责单一每个智能体应专注于一个特定的任务。避免创建“全能”智能体。设计清晰的提示词系统提示词是指引智能体行为的关键。明确角色、职责、输出格式和约束条件。迭代优化提示词是提升效果性价比最高的方式。工具设计规范化工具函数应有清晰的文档字符串Docstring描述其功能和参数。这能帮助大模型更好地理解何时以及如何调用它。7.3 状态与流程设计最小化状态只在状态中保存必要的数据。过大的状态会影响性能且难以调试。规划错误处理路径在图设计中考虑错误节点。当某个节点执行失败时可以路由到一个专门的error_handler节点记录日志、发送警报或尝试恢复。实现超时与重试对于调用外部API或耗时长的节点在节点逻辑或执行器层面添加超时和重试机制。7.4 可观测性与监控全面日志记录在每个节点的开始、结束、关键决策点记录日志。区分INFO,DEBUG,WARNING等级别。可视化工作流LangGraph提供了graph.get_graph().draw_mermaid_png()等方法可以将你定义的图生成Mermaid图表便于团队理解和沟通架构。追踪与评估集成像LangSmith这样的平台可以追踪每次调用、查看链的输入输出、分析延迟和成本并对智能体的输出进行质量评估。7.5 生产环境部署容器化使用Docker将你的应用及其依赖包括Ollama打包确保环境一致性。异步执行对于I/O密集型操作如网络请求考虑使用asyncio和LangChain的异步接口提升并发性能。状态持久化集成数据库如PostgreSQL, Redis来持久化工作流状态支持长时间运行和故障恢复。API化使用FastAPI或Flask将你的工作流包装成REST API方便与其他系统集成。从理解多智能体系统的价值到掌握LangGraph以图为核心的全新编程范式再到亲手搭建一个具备规划、研究、写作能力的协同系统你已经走完了从概念到实战的关键一步。LangGraph将复杂的智能体协作逻辑变得直观和可管理这是构建下一代AI应用的基础设施。真正的掌握源于实践和迭代。建议你以本文的示例为起点尝试以下挑战为研究者添加真实的搜索API如Serper Dev引入一个“评审者”智能体来评估报告质量并触发修改循环或者设计一个处理客户服务请求的多智能体流程包含“理解意图”、“查询知识库”、“生成回复”、“情感安抚”等节点。
