最近在尝试将大语言模型LLM与多智能体系统结合构建能够协作完成复杂任务的自动化流程时你是否也遇到过这些困扰智能体之间如何有效沟通任务状态如何持久化并流转LangChain 和 LangGraph 看起来功能强大但官方文档庞杂上手门槛高网上资料又多是零散的概念介绍缺乏一个从零到一的完整项目闭环。本文将为你彻底解决这些问题。我们将基于 LangChain 和 LangGraph 框架手把手构建一个名为 “DeepAgents” 的多智能体协作项目。这个项目将模拟一个简易的“技术内容创作团队”包含策划、写作、审核三个角色智能体它们将协作完成一篇技术博客的生成。通过这个实战案例你将系统掌握多智能体系统的核心设计思想、LangGraph 的状态机管理、以及如何将独立的 LLM 调用串联成智能的工作流。无论你是想入门智能体开发还是希望将现有单智能体应用升级为协作系统本文提供的代码和架构都可直接复用助你避开初期 99% 的配置与集成弯路。1. 多智能体系统与 LangChain/LangGraph 核心概念在开始实战之前我们有必要厘清几个核心概念这有助于理解我们为什么要用这些工具以及它们各自扮演什么角色。1.1 什么是多智能体系统多智能体系统指由多个具有一定自主性的智能体组成的集合它们通过交互、协作或竞争共同完成单个智能体难以解决的复杂任务。在我们的语境下每个“智能体”就是一个封装了特定能力如调用 LLM、执行工具的程序模块。例如一个“翻译智能体”专精语言转换一个“数据分析智能体”擅长处理表格。让它们协作就能完成“翻译一份数据分析报告”这样的复合任务。1.2 LangChain智能体的“工具箱”与“连接器”你可以把 LangChain 理解为一个强大的“粘合剂”框架。它的核心价值在于标准化交互它提供了统一的接口来调用各种 LLM如 OpenAI GPT, Anthropic Claude 国内 DeepSeek, 智谱 GLM 等让你无需关心不同厂商 API 的细节差异。模块化组件提供了提示词模板、输出解析器、记忆模块、文档加载器等标准化组件方便你快速构建智能体的“大脑”。工具调用智能体可以声明和使用“工具”Tool比如搜索网络、查询数据库、执行代码等极大地扩展了 LLM 的能力边界。简单说LangChain 帮你快速造出一个个功能各异的“智能体工人”。1.3 LangGraph智能体团队的“调度中心”与“流程图”当你有多个“智能体工人”后如何安排他们有序工作谁先谁后工作结果如何传递出了问题怎么处理这就是 LangGraph 要解决的问题。基于图的编程模型LangGraph 允许你用“图”来定义智能体之间的协作流程。节点Node代表一个操作如调用一个智能体边Edge代表操作之间的流转条件。状态管理它维护一个共享的“状态”对象智能体们可以读取和修改这个状态从而实现信息的传递和共享。这是多智能体协作的基石。控制流支持条件分支、循环、并行等复杂逻辑使得工作流可以动态调整而不仅仅是简单的线性管道。所以LangChain 负责制造智能体而 LangGraph 负责组织智能体协同工作。两者结合才能构建出真正强大、可应对复杂场景的多智能体应用。2. 环境准备与项目初始化我们的实战项目将使用 Python 作为开发语言。请确保你的环境满足以下要求。2.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。Python 版本 3.10。这是 LangChain 相关库比较稳定的支持版本。包管理工具pip。LLM 服务你需要一个可用的 LLM API 密钥。本文示例将使用OpenAI GPT-4和DeepSeek作为备选。你也可以替换为任何 LangChain 支持的模型如 Anthropic, 智谱AI等。2.2 创建项目与安装依赖首先创建一个新的项目目录并初始化虚拟环境这是管理 Python 项目依赖的最佳实践。# 创建项目目录 mkdir deepagents-tutorial cd deepagents-tutorial # 创建虚拟环境 (Python 3.10) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate接下来安装核心依赖库。我们将安装langchain,langgraph, 以及对应 LLM 的 SDK。# 安装 LangChain 核心和 OpenAI 集成 (如果使用 OpenAI) pip install langchain langchain-openai langgraph # 如果你计划使用 DeepSeek需要安装 langchain-community 和 requests pip install langchain-community requests # 安装环境变量管理库推荐 pip install python-dotenv2.3 项目结构规划一个清晰的项目结构有助于代码维护。我们的项目结构如下deepagents-tutorial/ ├── .env # 存储API密钥等敏感信息 ├── requirements.txt # 项目依赖列表 ├── main.py # 应用主入口 ├── agents/ # 智能体模块目录 │ ├── __init__.py │ ├── planner_agent.py # 策划智能体 │ ├── writer_agent.py # 写作智能体 │ └── reviewer_agent.py # 审核智能体 ├── graph/ # LangGraph 工作流定义目录 │ ├── __init__.py │ └── blog_workflow.py # 博客生成工作流图 ├── state/ # 状态模型定义目录 │ ├── __init__.py │ └── blog_state.py # 博客生成流程的共享状态 └── utils/ # 工具函数目录 ├── __init__.py └── llm_client.py # LLM 客户端初始化封装你可以使用以下命令快速创建这个结构mkdir -p agents graph state utils touch .env requirements.txt main.py touch agents/__init__.py agents/planner_agent.py agents/writer_agent.py agents/reviewer_agent.py touch graph/__init__.py graph/blog_workflow.py touch state/__init__.py state/blog_state.py touch utils/__init__.py utils/llm_client.py2.4 配置 API 密钥在项目根目录创建.env文件用于安全地存储你的 API 密钥。切记不要将此文件提交到版本控制系统如 Git。# .env # 使用 OpenAI OPENAI_API_KEYyour_openai_api_key_here # 或者使用 DeepSeek (注意DeepSeek API Base URL 可能不同请查阅最新文档) DEEPSEEK_API_KEYyour_deepseek_api_key_here DEEPSEEK_API_BASEhttps://api.deepseek.com然后在utils/llm_client.py中我们创建一个统一的 LLM 客户端初始化工具。# utils/llm_client.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI # 示例如果需要使用 DeepSeek # from langchain_community.chat_models import ChatOpenAI # 部分社区模型兼容OpenAI接口 # 加载 .env 文件中的环境变量 load_dotenv() def get_llm(model_typeopenai, temperature0.7): 获取 LLM 客户端。 Args: model_type: openai 或 deepseek temperature: 创造性0-1越低越确定。 Returns: 配置好的 LLM 实例。 if model_type.lower() openai: api_key os.getenv(OPENAI_API_KEY) if not api_key: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY) # 使用 gpt-4 或 gpt-3.5-turbo return ChatOpenAI(modelgpt-4-turbo-preview, api_keyapi_key, temperaturetemperature) elif model_type.lower() deepseek: api_key os.getenv(DEEPSEEK_API_KEY) api_base os.getenv(DEEPSEEK_API_BASE) if not api_key: raise ValueError(请在 .env 文件中设置 DEEPSEEK_API_KEY) # 注意DeepSeek 的 LangChain 集成方式可能随版本变化此为示例。 # 一种常见方式是通过 openai 兼容接口调用。 return ChatOpenAI( modeldeepseek-chat, openai_api_keyapi_key, openai_api_baseapi_base, temperaturetemperature ) else: raise ValueError(f不支持的 model_type: {model_type})3. 定义协作状态与智能体多智能体协作的核心是共享状态。我们首先定义在整个工作流中传递的数据结构。3.1 设计状态模型状态是一个字典或 Pydantic 模型包含了工作流从开始到结束所需的所有信息。我们在state/blog_state.py中定义。# state/blog_state.py from typing import List, Optional, Dict, Any from typing_extensions import TypedDict class BlogState(TypedDict): 博客生成工作流的共享状态。 使用 TypedDict 可以提供更好的类型提示。 # 用户输入 topic: str # 博客主题 requirements: str # 额外要求 # 策划阶段产出 outline: Optional[str] # 大纲 keywords: Optional[List[str]] # 关键词 target_audience: Optional[str] # 目标读者 # 写作阶段产出 draft: Optional[str] # 博客草稿 # 审核阶段产出 feedback: Optional[str] # 审核反馈 revised_draft: Optional[str] # 修订后的草稿 # 系统状态 current_step: str # 当前步骤如 planning, writing, reviewing, end max_revisions: int # 最大修订次数 revision_count: int # 当前修订计数3.2 实现策划智能体策划智能体负责根据主题生成大纲、关键词和受众分析。在agents/planner_agent.py中实现。# agents/planner_agent.py from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from utils.llm_client import get_llm import json class PlannerAgent: def __init__(self, llm_model_typeopenai): self.llm get_llm(llm_model_type, temperature0.3) # 策划需要更确定性 self.chain self._create_chain() def _create_chain(self): 创建策划链提示词 LLM 输出解析器 prompt ChatPromptTemplate.from_messages([ (system, 你是一位资深技术博客策划专家。你的任务是根据用户提供的主题和需求制定一份详细、结构清晰的博客大纲并提取核心关键词和分析目标读者。 请严格按照以下 JSON 格式输出不要包含任何其他解释性文字 {{ outline: 博客的详细大纲用 Markdown 格式的列表表示至少包含三级标题如 ##, ###。, keywords: [关键词1, 关键词2, 关键词3, 关键词4, 关键词5], target_audience: 描述这篇博客的目标读者群体例如‘初级后端开发者’、‘对机器学习感兴趣的学生’等。 }}), (human, 博客主题{topic}\n额外要求或背景{requirements}) ]) # 使用 StrOutputParser然后解析 JSON return prompt | self.llm | StrOutputParser() def run(self, topic: str, requirements: str ) - dict: 执行策划任务 print(f[策划智能体] 开始为主题 {topic} 制定大纲...) result_str self.chain.invoke({topic: topic, requirements: requirements}) try: result json.loads(result_str) print([策划智能体] 大纲制定完成。) return result except json.JSONDecodeError as e: print(f[策划智能体] 解析输出失败: {e}\n原始输出:\n{result_str}) # 提供一个兜底结构 return { outline: f# {topic}\n\n## 概述\n\n## 主要内容\n\n## 总结, keywords: [topic], target_audience: 技术爱好者 } # 示例单独测试策划智能体 if __name__ __main__: agent PlannerAgent(openai) # 或 deepseek plan agent.run(如何使用 LangGraph 构建多智能体系统, 面向有 Python 基础的中级开发者) print(json.dumps(plan, indent2, ensure_asciiFalse))3.3 实现写作智能体写作智能体根据策划阶段产出的大纲和关键词撰写博客正文。在agents/writer_agent.py中实现。# agents/writer_agent.py from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from utils.llm_client import get_llm class WriterAgent: def __init__(self, llm_model_typeopenai): self.llm get_llm(llm_model_type, temperature0.7) # 写作需要一定创造性 self.chain self._create_chain() def _create_chain(self): prompt ChatPromptTemplate.from_messages([ (system, 你是一位优秀的科技博客作者文风清晰、逻辑严谨、示例丰富。 请根据提供的大纲、关键词和目标读者撰写一篇完整的技术博客正文。 要求 1. 严格遵循提供的大纲结构。 2. 自然地融入提供的关键词。 3. 语言风格要符合目标读者的认知水平。 4. 在合适的地方添加代码示例用 标记和说明。 5. 输出为完整的 Markdown 格式文本不要包含大纲本身以外的元信息。), (human, 请撰写博客正文。 主题背景{topic} 目标读者{target_audience} 博客大纲 {outline} 核心关键词{keywords} ) ]) return prompt | self.llm | StrOutputParser() def run(self, topic: str, outline: str, keywords: list, target_audience: str) - str: 执行写作任务 print(f[写作智能体] 开始根据大纲撰写博客 {topic}...) draft self.chain.invoke({ topic: topic, outline: outline, keywords: , .join(keywords), target_audience: target_audience }) print([写作智能体] 初稿撰写完成。) return draft3.4 实现审核智能体审核智能体负责检查博客草稿的质量提出修改意见。在agents/reviewer_agent.py中实现。# agents/reviewer_agent.py from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from utils.llm_client import get_llm class ReviewerAgent: def __init__(self, llm_model_typeopenai): self.llm get_llm(llm_model_type, temperature0.2) # 审核需要严谨 self.chain self._create_chain() def _create_chain(self): prompt ChatPromptTemplate.from_messages([ (system, 你是一位严格的技术编辑负责审核技术博客草稿。 请从以下维度进行审核并给出具体的、可操作的修改反馈 1. **技术准确性**概念、代码示例是否有误 2. **逻辑结构**是否清晰是否符合大纲 3. **可读性**语言是否流畅术语解释是否到位 4. **实用性**对目标读者是否有实际帮助 5. **格式规范**Markdown 格式是否正确代码块是否标注语言 如果草稿质量很高无需修改请反馈“审核通过无需修改”。 否则请清晰列出需要修改的点并给出修改建议。反馈应直接、简洁。), (human, 请审核以下博客草稿 博客主题{topic} 目标读者{target_audience} 草稿内容 {draft} ) ]) return prompt | self.llm | StrOutputParser() def run(self, topic: str, draft: str, target_audience: str) - str: 执行审核任务 print(f[审核智能体] 开始审核博客 {topic}...) feedback self.chain.invoke({ topic: topic, draft: draft, target_audience: target_audience }) print([审核智能体] 审核完成。) return feedback4. 使用 LangGraph 编排多智能体工作流现在我们有了三个独立的智能体。接下来使用 LangGraph 将它们连接成一个有序的、带状态的工作流。4.1 定义工作流图在graph/blog_workflow.py中我们将创建整个协作流程图。# graph/blog_workflow.py from typing import Literal from langgraph.graph import StateGraph, END from state.blog_state import BlogState from agents.planner_agent import PlannerAgent from agents.writer_agent import WriterAgent from agents.reviewer_agent import ReviewerAgent class BlogWorkflow: def __init__(self, llm_model_typeopenai): self.llm_model_type llm_model_type self.planner PlannerAgent(llm_model_type) self.writer WriterAgent(llm_model_type) self.reviewer ReviewerAgent(llm_model_type) self.graph self._build_graph() def _build_graph(self): 构建博客生成工作流图 workflow StateGraph(BlogState) # 1. 定义节点每个智能体或操作都是一个节点 workflow.add_node(planning, self._planning_node) workflow.add_node(writing, self._writing_node) workflow.add_node(reviewing, self._reviewing_node) workflow.add_node(revising, self._revising_node) # 2. 设置入口点 workflow.set_entry_point(planning) # 3. 添加边定义节点之间的流转逻辑 workflow.add_edge(planning, writing) # 策划完直接写作 workflow.add_edge(writing, reviewing) # 写作完进入审核 # 4. 添加条件边审核后根据反馈决定下一步 workflow.add_conditional_edges( reviewing, self._decide_after_review, # 这是一个判断函数 { revise: revising, # 需要修订 accept: END, # 审核通过结束 max_revisions: END, # 修订次数超限强制结束 } ) workflow.add_edge(revising, reviewing) # 修订后重新进入审核 return workflow.compile() # --- 节点函数定义 --- def _planning_node(self, state: BlogState) - BlogState: 策划节点生成大纲、关键词、受众分析 print(\n 进入 [策划] 阶段 ) plan_result self.planner.run(state[topic], state.get(requirements, )) # 更新状态 state[outline] plan_result.get(outline) state[keywords] plan_result.get(keywords, []) state[target_audience] plan_result.get(target_audience) state[current_step] planning return state def _writing_node(self, state: BlogState) - BlogState: 写作节点根据策划结果撰写草稿 print(\n 进入 [写作] 阶段 ) draft self.writer.run( state[topic], state[outline], state[keywords], state[target_audience] ) state[draft] draft state[current_step] writing return state def _reviewing_node(self, state: BlogState) - BlogState: 审核节点审核草稿并生成反馈 print(\n 进入 [审核] 阶段 ) feedback self.reviewer.run( state[topic], state[draft], state[target_audience] ) state[feedback] feedback state[current_step] reviewing return state def _revising_node(self, state: BlogState) - BlogState: 修订节点根据反馈修改草稿这里简化处理实际可让写作智能体根据反馈重写 print(\n 进入 [修订] 阶段 ) # 简化逻辑将反馈附加到草稿前提示写作智能体下一轮优化。 # 更复杂的实现可以调用一个专门的“修订智能体”。 revision_instruction f根据以下审核反馈修改之前的草稿\n{state[feedback]}\n\n请输出修改后的完整博客内容。 # 这里我们为了简化直接让写作智能体基于原大纲和反馈重写。 # 注意实际项目中修订逻辑需要更精细的设计。 revised_draft self.writer.run( state[topic], state[outline], state[keywords], state[target_audience] ) state[revised_draft] revised_draft state[draft] revised_draft # 用修订稿覆盖原草稿进入下一轮审核 state[revision_count] state.get(revision_count, 0) 1 state[current_step] revising print(f[系统] 第 {state[revision_count]} 次修订完成。) return state # --- 条件判断函数 --- def _decide_after_review(self, state: BlogState) - Literal[revise, accept, max_revisions]: 审核后决策判断是需要修订、接受还是终止 feedback state[feedback] revision_count state.get(revision_count, 0) max_revisions state.get(max_revisions, 3) # 默认最多修订3次 # 检查是否超限 if revision_count max_revisions: print(f[系统] 已达到最大修订次数 ({max_revisions})流程终止。) return max_revisions # 简单规则如果反馈包含“审核通过”或类似字样则接受 if 审核通过 in feedback or 无需修改 in feedback: print([系统] 审核通过流程结束。) return accept else: print([系统] 审核反馈需要修订进入修订阶段。) return revise # --- 对外调用接口 --- def run(self, topic: str, requirements: str , max_revisions: int 3) - dict: 运行工作流 # 初始化状态 initial_state: BlogState { topic: topic, requirements: requirements, outline: None, keywords: None, target_audience: None, draft: None, feedback: None, revised_draft: None, current_step: start, max_revisions: max_revisions, revision_count: 0, } print(*50) print(f开始执行博客生成工作流主题{topic}) print(*50) # 执行编译好的图 final_state self.graph.invoke(initial_state) print(\n *50) print(工作流执行完毕) print(*50) return final_state5. 整合与运行从主程序调用最后我们创建一个主程序main.py来整合所有模块并提供用户交互界面。# main.py import sys from graph.blog_workflow import BlogWorkflow def main(): print(欢迎使用 DeepAgents 多智能体博客生成系统) print(本系统将自动完成策划、写作、审核流程。\n) # 用户输入 topic input(请输入博客主题 (例如Python 异步编程入门): ).strip() if not topic: print(主题不能为空程序退出。) sys.exit(1) requirements input(请输入额外要求或背景可选直接回车跳过: ).strip() # 选择 LLM 模型 print(\n请选择使用的 LLM 模型) print(1. OpenAI GPT-4 (需要 OPENAI_API_KEY)) print(2. DeepSeek (需要 DEEPSEEK_API_KEY)) choice input(请输入选项 (1 或 2): ).strip() model_type openai if choice 2: model_type deepseek elif choice ! 1: print(输入无效默认使用 OpenAI。) # 初始化并运行工作流 try: workflow BlogWorkflow(llm_model_typemodel_type) final_state workflow.run(topic, requirements, max_revisions2) # 设置最大修订2次 # 输出最终结果 print(\n *60) print(最终生成的博客内容) print(*60) # 优先输出修订后的草稿如果没有则输出初始草稿 final_content final_state.get(revised_draft) or final_state.get(draft) if final_content: print(final_content) # 可选保存到文件 filename fblog_{topic[:20]}.md.replace( , _).replace(/, _) with open(filename, w, encodingutf-8) as f: f.write(final_content) print(f\n博客内容已保存至文件{filename}) else: print(生成失败未获得有效内容。) print(\n *60) print(流程状态摘要) print(f主题{final_state[topic]}) print(f最终步骤{final_state[current_step]}) print(f修订次数{final_state[revision_count]}) if final_state[feedback]: print(f最后审核反馈\n{final_state[feedback][:200]}...) # 只显示前200字符 except Exception as e: print(f\n程序运行出错: {e}) # 可以在这里添加更详细的错误日志 import traceback traceback.print_exc() if __name__ __main__: main()5.1 运行你的多智能体项目确保你的.env文件已正确配置 API 密钥然后在项目根目录下运行python main.py按照提示输入博客主题和要求选择模型系统将自动运行策划、写作、审核、修订的完整流程并将最终生成的 Markdown 博客保存到本地文件。6. 常见问题与排查思路在实际运行中你可能会遇到以下问题。这里提供排查思路。问题现象可能原因解决思路ModuleNotFoundError: No module named langchain_openai依赖未正确安装。1. 确认虚拟环境已激活。2. 运行pip install -r requirements.txt或pip install langchain-openai。AuthenticationError或Invalid API KeyAPI 密钥错误或未设置。1. 检查.env文件是否存在格式是否正确无空格无引号。2. 确认环境变量已加载load_dotenv()。3. 在llm_client.py中打印os.getenv(“KEY”)验证是否读取成功。RateLimitErrorAPI 调用频率超限。1. 检查对应平台的用量和频率限制。2. 在代码中添加延迟如time.sleep(1)。3. 考虑使用更低成本的模型进行开发测试。智能体输出格式不符合预期JSON解析失败LLM 没有严格按照提示词中的格式输出。1. 降低temperature参数如设为 0.2。2. 在提示词中更严格地规定格式使用json.dumps示例。3. 实现一个更健壮的输出解析器如JsonOutputParser。工作流陷入无限循环审核逻辑_decide_after_review判断条件不明确总是返回”revise”。1. 检查审核反馈的判断逻辑确保有明确的“通过”条件。2. 设置合理的max_revisions。3. 在状态中增加revision_count并打印监控循环次数。DeepSeek 模型调用失败DeepSeek 的 API 端点或参数与 OpenAI 不完全兼容。1. 查阅 DeepSeek 官方最新的 API 文档。2. 确认openai_api_base和model参数是否正确。3. 考虑使用 LangChain 的ChatDeepSeek类如果官方提供。程序中途退出无错误信息可能某个智能体调用 LLM 超时或网络中断。1. 在关键函数调用处添加try…except捕获异常并打印。2. 为 LLM 调用增加超时参数如ChatOpenAI(…, request_timeout60)。3. 检查网络连接。7. 最佳实践与项目扩展建议掌握了基础框架后你可以遵循以下最佳实践来优化项目并探索更高级的功能。7.1 工程化最佳实践配置管理将模型类型、温度、最大修订次数等参数提取到配置文件如config.yaml中避免硬编码。日志记录使用 Python 的logging模块替代print可以输出到文件并设置不同级别INFO, DEBUG, ERROR方便调试和监控。异常处理与重试LLM API 调用可能不稳定使用tenacity库为关键操作添加重试机制和退避策略。状态持久化将BlogState在关键节点序列化保存到数据库或文件以便工作流中断后可以恢复。测试为每个智能体编写单元测试模拟 LLM 响应确保业务逻辑正确。为工作流编写集成测试。7.2 性能与成本优化模型选择策划和审核可以使用较小、较快的模型如 GPT-3.5-Turbo写作使用能力更强的模型如 GPT-4以平衡效果与成本/速度。缓存对相对固定的提示词和 LLM 响应使用缓存如langchain.cache减少重复调用和费用。异步调用如果智能体间没有严格的先后依赖可以使用asyncio实现并行执行大幅缩短总耗时。7.3 功能扩展方向增加更多智能体SEO优化智能体在发布前分析并优化内容的关键词密度和元描述。配图建议智能体根据文章内容生成配图的关键词描述或调用文生图 API。社交媒体摘要智能体从长文中提取要点生成适合 Twitter、微博的短文案。增强工作流逻辑并行分支让“写作”和“寻找参考链接”两个智能体同时工作。人工审核节点在自动审核后引入一个等待人工输入HumanApproval的节点只有人工确认后才能继续。动态工具调用让智能体在运行过程中自主决定是否调用网络搜索、代码执行等工具来获取信息。集成外部系统向量数据库让策划智能体先检索相关的历史文章或知识库使大纲更有依据。CMS发布工作流最后增加一个自动发布到 WordPress、Confluence 等系统的智能体。监控与告警将工作流的执行状态、耗时、成本指标发送到 Prometheus 或 Datadog。通过这个从零开始的“DeepAgents”项目实战你已经掌握了使用 LangChain 构建智能体、并用 LangGraph 编排复杂多智能体工作流的核心方法。这套模式具有很强的通用性你可以轻松地将“博客生成团队”替换为“客服问答团队”、“代码审查团队”或“数据分析团队”只需定义新的智能体角色和状态结构即可。多智能体系统的魅力在于其模块化和可编排性希望本文能成为你探索更复杂 AI 应用架构的坚实起点。如果在实践过程中遇到问题欢迎在评论区交流探讨。
基于LangGraph与LangChain构建多智能体协作系统实战
最近在尝试将大语言模型LLM与多智能体系统结合构建能够协作完成复杂任务的自动化流程时你是否也遇到过这些困扰智能体之间如何有效沟通任务状态如何持久化并流转LangChain 和 LangGraph 看起来功能强大但官方文档庞杂上手门槛高网上资料又多是零散的概念介绍缺乏一个从零到一的完整项目闭环。本文将为你彻底解决这些问题。我们将基于 LangChain 和 LangGraph 框架手把手构建一个名为 “DeepAgents” 的多智能体协作项目。这个项目将模拟一个简易的“技术内容创作团队”包含策划、写作、审核三个角色智能体它们将协作完成一篇技术博客的生成。通过这个实战案例你将系统掌握多智能体系统的核心设计思想、LangGraph 的状态机管理、以及如何将独立的 LLM 调用串联成智能的工作流。无论你是想入门智能体开发还是希望将现有单智能体应用升级为协作系统本文提供的代码和架构都可直接复用助你避开初期 99% 的配置与集成弯路。1. 多智能体系统与 LangChain/LangGraph 核心概念在开始实战之前我们有必要厘清几个核心概念这有助于理解我们为什么要用这些工具以及它们各自扮演什么角色。1.1 什么是多智能体系统多智能体系统指由多个具有一定自主性的智能体组成的集合它们通过交互、协作或竞争共同完成单个智能体难以解决的复杂任务。在我们的语境下每个“智能体”就是一个封装了特定能力如调用 LLM、执行工具的程序模块。例如一个“翻译智能体”专精语言转换一个“数据分析智能体”擅长处理表格。让它们协作就能完成“翻译一份数据分析报告”这样的复合任务。1.2 LangChain智能体的“工具箱”与“连接器”你可以把 LangChain 理解为一个强大的“粘合剂”框架。它的核心价值在于标准化交互它提供了统一的接口来调用各种 LLM如 OpenAI GPT, Anthropic Claude 国内 DeepSeek, 智谱 GLM 等让你无需关心不同厂商 API 的细节差异。模块化组件提供了提示词模板、输出解析器、记忆模块、文档加载器等标准化组件方便你快速构建智能体的“大脑”。工具调用智能体可以声明和使用“工具”Tool比如搜索网络、查询数据库、执行代码等极大地扩展了 LLM 的能力边界。简单说LangChain 帮你快速造出一个个功能各异的“智能体工人”。1.3 LangGraph智能体团队的“调度中心”与“流程图”当你有多个“智能体工人”后如何安排他们有序工作谁先谁后工作结果如何传递出了问题怎么处理这就是 LangGraph 要解决的问题。基于图的编程模型LangGraph 允许你用“图”来定义智能体之间的协作流程。节点Node代表一个操作如调用一个智能体边Edge代表操作之间的流转条件。状态管理它维护一个共享的“状态”对象智能体们可以读取和修改这个状态从而实现信息的传递和共享。这是多智能体协作的基石。控制流支持条件分支、循环、并行等复杂逻辑使得工作流可以动态调整而不仅仅是简单的线性管道。所以LangChain 负责制造智能体而 LangGraph 负责组织智能体协同工作。两者结合才能构建出真正强大、可应对复杂场景的多智能体应用。2. 环境准备与项目初始化我们的实战项目将使用 Python 作为开发语言。请确保你的环境满足以下要求。2.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。Python 版本 3.10。这是 LangChain 相关库比较稳定的支持版本。包管理工具pip。LLM 服务你需要一个可用的 LLM API 密钥。本文示例将使用OpenAI GPT-4和DeepSeek作为备选。你也可以替换为任何 LangChain 支持的模型如 Anthropic, 智谱AI等。2.2 创建项目与安装依赖首先创建一个新的项目目录并初始化虚拟环境这是管理 Python 项目依赖的最佳实践。# 创建项目目录 mkdir deepagents-tutorial cd deepagents-tutorial # 创建虚拟环境 (Python 3.10) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate接下来安装核心依赖库。我们将安装langchain,langgraph, 以及对应 LLM 的 SDK。# 安装 LangChain 核心和 OpenAI 集成 (如果使用 OpenAI) pip install langchain langchain-openai langgraph # 如果你计划使用 DeepSeek需要安装 langchain-community 和 requests pip install langchain-community requests # 安装环境变量管理库推荐 pip install python-dotenv2.3 项目结构规划一个清晰的项目结构有助于代码维护。我们的项目结构如下deepagents-tutorial/ ├── .env # 存储API密钥等敏感信息 ├── requirements.txt # 项目依赖列表 ├── main.py # 应用主入口 ├── agents/ # 智能体模块目录 │ ├── __init__.py │ ├── planner_agent.py # 策划智能体 │ ├── writer_agent.py # 写作智能体 │ └── reviewer_agent.py # 审核智能体 ├── graph/ # LangGraph 工作流定义目录 │ ├── __init__.py │ └── blog_workflow.py # 博客生成工作流图 ├── state/ # 状态模型定义目录 │ ├── __init__.py │ └── blog_state.py # 博客生成流程的共享状态 └── utils/ # 工具函数目录 ├── __init__.py └── llm_client.py # LLM 客户端初始化封装你可以使用以下命令快速创建这个结构mkdir -p agents graph state utils touch .env requirements.txt main.py touch agents/__init__.py agents/planner_agent.py agents/writer_agent.py agents/reviewer_agent.py touch graph/__init__.py graph/blog_workflow.py touch state/__init__.py state/blog_state.py touch utils/__init__.py utils/llm_client.py2.4 配置 API 密钥在项目根目录创建.env文件用于安全地存储你的 API 密钥。切记不要将此文件提交到版本控制系统如 Git。# .env # 使用 OpenAI OPENAI_API_KEYyour_openai_api_key_here # 或者使用 DeepSeek (注意DeepSeek API Base URL 可能不同请查阅最新文档) DEEPSEEK_API_KEYyour_deepseek_api_key_here DEEPSEEK_API_BASEhttps://api.deepseek.com然后在utils/llm_client.py中我们创建一个统一的 LLM 客户端初始化工具。# utils/llm_client.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI # 示例如果需要使用 DeepSeek # from langchain_community.chat_models import ChatOpenAI # 部分社区模型兼容OpenAI接口 # 加载 .env 文件中的环境变量 load_dotenv() def get_llm(model_typeopenai, temperature0.7): 获取 LLM 客户端。 Args: model_type: openai 或 deepseek temperature: 创造性0-1越低越确定。 Returns: 配置好的 LLM 实例。 if model_type.lower() openai: api_key os.getenv(OPENAI_API_KEY) if not api_key: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY) # 使用 gpt-4 或 gpt-3.5-turbo return ChatOpenAI(modelgpt-4-turbo-preview, api_keyapi_key, temperaturetemperature) elif model_type.lower() deepseek: api_key os.getenv(DEEPSEEK_API_KEY) api_base os.getenv(DEEPSEEK_API_BASE) if not api_key: raise ValueError(请在 .env 文件中设置 DEEPSEEK_API_KEY) # 注意DeepSeek 的 LangChain 集成方式可能随版本变化此为示例。 # 一种常见方式是通过 openai 兼容接口调用。 return ChatOpenAI( modeldeepseek-chat, openai_api_keyapi_key, openai_api_baseapi_base, temperaturetemperature ) else: raise ValueError(f不支持的 model_type: {model_type})3. 定义协作状态与智能体多智能体协作的核心是共享状态。我们首先定义在整个工作流中传递的数据结构。3.1 设计状态模型状态是一个字典或 Pydantic 模型包含了工作流从开始到结束所需的所有信息。我们在state/blog_state.py中定义。# state/blog_state.py from typing import List, Optional, Dict, Any from typing_extensions import TypedDict class BlogState(TypedDict): 博客生成工作流的共享状态。 使用 TypedDict 可以提供更好的类型提示。 # 用户输入 topic: str # 博客主题 requirements: str # 额外要求 # 策划阶段产出 outline: Optional[str] # 大纲 keywords: Optional[List[str]] # 关键词 target_audience: Optional[str] # 目标读者 # 写作阶段产出 draft: Optional[str] # 博客草稿 # 审核阶段产出 feedback: Optional[str] # 审核反馈 revised_draft: Optional[str] # 修订后的草稿 # 系统状态 current_step: str # 当前步骤如 planning, writing, reviewing, end max_revisions: int # 最大修订次数 revision_count: int # 当前修订计数3.2 实现策划智能体策划智能体负责根据主题生成大纲、关键词和受众分析。在agents/planner_agent.py中实现。# agents/planner_agent.py from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from utils.llm_client import get_llm import json class PlannerAgent: def __init__(self, llm_model_typeopenai): self.llm get_llm(llm_model_type, temperature0.3) # 策划需要更确定性 self.chain self._create_chain() def _create_chain(self): 创建策划链提示词 LLM 输出解析器 prompt ChatPromptTemplate.from_messages([ (system, 你是一位资深技术博客策划专家。你的任务是根据用户提供的主题和需求制定一份详细、结构清晰的博客大纲并提取核心关键词和分析目标读者。 请严格按照以下 JSON 格式输出不要包含任何其他解释性文字 {{ outline: 博客的详细大纲用 Markdown 格式的列表表示至少包含三级标题如 ##, ###。, keywords: [关键词1, 关键词2, 关键词3, 关键词4, 关键词5], target_audience: 描述这篇博客的目标读者群体例如‘初级后端开发者’、‘对机器学习感兴趣的学生’等。 }}), (human, 博客主题{topic}\n额外要求或背景{requirements}) ]) # 使用 StrOutputParser然后解析 JSON return prompt | self.llm | StrOutputParser() def run(self, topic: str, requirements: str ) - dict: 执行策划任务 print(f[策划智能体] 开始为主题 {topic} 制定大纲...) result_str self.chain.invoke({topic: topic, requirements: requirements}) try: result json.loads(result_str) print([策划智能体] 大纲制定完成。) return result except json.JSONDecodeError as e: print(f[策划智能体] 解析输出失败: {e}\n原始输出:\n{result_str}) # 提供一个兜底结构 return { outline: f# {topic}\n\n## 概述\n\n## 主要内容\n\n## 总结, keywords: [topic], target_audience: 技术爱好者 } # 示例单独测试策划智能体 if __name__ __main__: agent PlannerAgent(openai) # 或 deepseek plan agent.run(如何使用 LangGraph 构建多智能体系统, 面向有 Python 基础的中级开发者) print(json.dumps(plan, indent2, ensure_asciiFalse))3.3 实现写作智能体写作智能体根据策划阶段产出的大纲和关键词撰写博客正文。在agents/writer_agent.py中实现。# agents/writer_agent.py from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from utils.llm_client import get_llm class WriterAgent: def __init__(self, llm_model_typeopenai): self.llm get_llm(llm_model_type, temperature0.7) # 写作需要一定创造性 self.chain self._create_chain() def _create_chain(self): prompt ChatPromptTemplate.from_messages([ (system, 你是一位优秀的科技博客作者文风清晰、逻辑严谨、示例丰富。 请根据提供的大纲、关键词和目标读者撰写一篇完整的技术博客正文。 要求 1. 严格遵循提供的大纲结构。 2. 自然地融入提供的关键词。 3. 语言风格要符合目标读者的认知水平。 4. 在合适的地方添加代码示例用 标记和说明。 5. 输出为完整的 Markdown 格式文本不要包含大纲本身以外的元信息。), (human, 请撰写博客正文。 主题背景{topic} 目标读者{target_audience} 博客大纲 {outline} 核心关键词{keywords} ) ]) return prompt | self.llm | StrOutputParser() def run(self, topic: str, outline: str, keywords: list, target_audience: str) - str: 执行写作任务 print(f[写作智能体] 开始根据大纲撰写博客 {topic}...) draft self.chain.invoke({ topic: topic, outline: outline, keywords: , .join(keywords), target_audience: target_audience }) print([写作智能体] 初稿撰写完成。) return draft3.4 实现审核智能体审核智能体负责检查博客草稿的质量提出修改意见。在agents/reviewer_agent.py中实现。# agents/reviewer_agent.py from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from utils.llm_client import get_llm class ReviewerAgent: def __init__(self, llm_model_typeopenai): self.llm get_llm(llm_model_type, temperature0.2) # 审核需要严谨 self.chain self._create_chain() def _create_chain(self): prompt ChatPromptTemplate.from_messages([ (system, 你是一位严格的技术编辑负责审核技术博客草稿。 请从以下维度进行审核并给出具体的、可操作的修改反馈 1. **技术准确性**概念、代码示例是否有误 2. **逻辑结构**是否清晰是否符合大纲 3. **可读性**语言是否流畅术语解释是否到位 4. **实用性**对目标读者是否有实际帮助 5. **格式规范**Markdown 格式是否正确代码块是否标注语言 如果草稿质量很高无需修改请反馈“审核通过无需修改”。 否则请清晰列出需要修改的点并给出修改建议。反馈应直接、简洁。), (human, 请审核以下博客草稿 博客主题{topic} 目标读者{target_audience} 草稿内容 {draft} ) ]) return prompt | self.llm | StrOutputParser() def run(self, topic: str, draft: str, target_audience: str) - str: 执行审核任务 print(f[审核智能体] 开始审核博客 {topic}...) feedback self.chain.invoke({ topic: topic, draft: draft, target_audience: target_audience }) print([审核智能体] 审核完成。) return feedback4. 使用 LangGraph 编排多智能体工作流现在我们有了三个独立的智能体。接下来使用 LangGraph 将它们连接成一个有序的、带状态的工作流。4.1 定义工作流图在graph/blog_workflow.py中我们将创建整个协作流程图。# graph/blog_workflow.py from typing import Literal from langgraph.graph import StateGraph, END from state.blog_state import BlogState from agents.planner_agent import PlannerAgent from agents.writer_agent import WriterAgent from agents.reviewer_agent import ReviewerAgent class BlogWorkflow: def __init__(self, llm_model_typeopenai): self.llm_model_type llm_model_type self.planner PlannerAgent(llm_model_type) self.writer WriterAgent(llm_model_type) self.reviewer ReviewerAgent(llm_model_type) self.graph self._build_graph() def _build_graph(self): 构建博客生成工作流图 workflow StateGraph(BlogState) # 1. 定义节点每个智能体或操作都是一个节点 workflow.add_node(planning, self._planning_node) workflow.add_node(writing, self._writing_node) workflow.add_node(reviewing, self._reviewing_node) workflow.add_node(revising, self._revising_node) # 2. 设置入口点 workflow.set_entry_point(planning) # 3. 添加边定义节点之间的流转逻辑 workflow.add_edge(planning, writing) # 策划完直接写作 workflow.add_edge(writing, reviewing) # 写作完进入审核 # 4. 添加条件边审核后根据反馈决定下一步 workflow.add_conditional_edges( reviewing, self._decide_after_review, # 这是一个判断函数 { revise: revising, # 需要修订 accept: END, # 审核通过结束 max_revisions: END, # 修订次数超限强制结束 } ) workflow.add_edge(revising, reviewing) # 修订后重新进入审核 return workflow.compile() # --- 节点函数定义 --- def _planning_node(self, state: BlogState) - BlogState: 策划节点生成大纲、关键词、受众分析 print(\n 进入 [策划] 阶段 ) plan_result self.planner.run(state[topic], state.get(requirements, )) # 更新状态 state[outline] plan_result.get(outline) state[keywords] plan_result.get(keywords, []) state[target_audience] plan_result.get(target_audience) state[current_step] planning return state def _writing_node(self, state: BlogState) - BlogState: 写作节点根据策划结果撰写草稿 print(\n 进入 [写作] 阶段 ) draft self.writer.run( state[topic], state[outline], state[keywords], state[target_audience] ) state[draft] draft state[current_step] writing return state def _reviewing_node(self, state: BlogState) - BlogState: 审核节点审核草稿并生成反馈 print(\n 进入 [审核] 阶段 ) feedback self.reviewer.run( state[topic], state[draft], state[target_audience] ) state[feedback] feedback state[current_step] reviewing return state def _revising_node(self, state: BlogState) - BlogState: 修订节点根据反馈修改草稿这里简化处理实际可让写作智能体根据反馈重写 print(\n 进入 [修订] 阶段 ) # 简化逻辑将反馈附加到草稿前提示写作智能体下一轮优化。 # 更复杂的实现可以调用一个专门的“修订智能体”。 revision_instruction f根据以下审核反馈修改之前的草稿\n{state[feedback]}\n\n请输出修改后的完整博客内容。 # 这里我们为了简化直接让写作智能体基于原大纲和反馈重写。 # 注意实际项目中修订逻辑需要更精细的设计。 revised_draft self.writer.run( state[topic], state[outline], state[keywords], state[target_audience] ) state[revised_draft] revised_draft state[draft] revised_draft # 用修订稿覆盖原草稿进入下一轮审核 state[revision_count] state.get(revision_count, 0) 1 state[current_step] revising print(f[系统] 第 {state[revision_count]} 次修订完成。) return state # --- 条件判断函数 --- def _decide_after_review(self, state: BlogState) - Literal[revise, accept, max_revisions]: 审核后决策判断是需要修订、接受还是终止 feedback state[feedback] revision_count state.get(revision_count, 0) max_revisions state.get(max_revisions, 3) # 默认最多修订3次 # 检查是否超限 if revision_count max_revisions: print(f[系统] 已达到最大修订次数 ({max_revisions})流程终止。) return max_revisions # 简单规则如果反馈包含“审核通过”或类似字样则接受 if 审核通过 in feedback or 无需修改 in feedback: print([系统] 审核通过流程结束。) return accept else: print([系统] 审核反馈需要修订进入修订阶段。) return revise # --- 对外调用接口 --- def run(self, topic: str, requirements: str , max_revisions: int 3) - dict: 运行工作流 # 初始化状态 initial_state: BlogState { topic: topic, requirements: requirements, outline: None, keywords: None, target_audience: None, draft: None, feedback: None, revised_draft: None, current_step: start, max_revisions: max_revisions, revision_count: 0, } print(*50) print(f开始执行博客生成工作流主题{topic}) print(*50) # 执行编译好的图 final_state self.graph.invoke(initial_state) print(\n *50) print(工作流执行完毕) print(*50) return final_state5. 整合与运行从主程序调用最后我们创建一个主程序main.py来整合所有模块并提供用户交互界面。# main.py import sys from graph.blog_workflow import BlogWorkflow def main(): print(欢迎使用 DeepAgents 多智能体博客生成系统) print(本系统将自动完成策划、写作、审核流程。\n) # 用户输入 topic input(请输入博客主题 (例如Python 异步编程入门): ).strip() if not topic: print(主题不能为空程序退出。) sys.exit(1) requirements input(请输入额外要求或背景可选直接回车跳过: ).strip() # 选择 LLM 模型 print(\n请选择使用的 LLM 模型) print(1. OpenAI GPT-4 (需要 OPENAI_API_KEY)) print(2. DeepSeek (需要 DEEPSEEK_API_KEY)) choice input(请输入选项 (1 或 2): ).strip() model_type openai if choice 2: model_type deepseek elif choice ! 1: print(输入无效默认使用 OpenAI。) # 初始化并运行工作流 try: workflow BlogWorkflow(llm_model_typemodel_type) final_state workflow.run(topic, requirements, max_revisions2) # 设置最大修订2次 # 输出最终结果 print(\n *60) print(最终生成的博客内容) print(*60) # 优先输出修订后的草稿如果没有则输出初始草稿 final_content final_state.get(revised_draft) or final_state.get(draft) if final_content: print(final_content) # 可选保存到文件 filename fblog_{topic[:20]}.md.replace( , _).replace(/, _) with open(filename, w, encodingutf-8) as f: f.write(final_content) print(f\n博客内容已保存至文件{filename}) else: print(生成失败未获得有效内容。) print(\n *60) print(流程状态摘要) print(f主题{final_state[topic]}) print(f最终步骤{final_state[current_step]}) print(f修订次数{final_state[revision_count]}) if final_state[feedback]: print(f最后审核反馈\n{final_state[feedback][:200]}...) # 只显示前200字符 except Exception as e: print(f\n程序运行出错: {e}) # 可以在这里添加更详细的错误日志 import traceback traceback.print_exc() if __name__ __main__: main()5.1 运行你的多智能体项目确保你的.env文件已正确配置 API 密钥然后在项目根目录下运行python main.py按照提示输入博客主题和要求选择模型系统将自动运行策划、写作、审核、修订的完整流程并将最终生成的 Markdown 博客保存到本地文件。6. 常见问题与排查思路在实际运行中你可能会遇到以下问题。这里提供排查思路。问题现象可能原因解决思路ModuleNotFoundError: No module named langchain_openai依赖未正确安装。1. 确认虚拟环境已激活。2. 运行pip install -r requirements.txt或pip install langchain-openai。AuthenticationError或Invalid API KeyAPI 密钥错误或未设置。1. 检查.env文件是否存在格式是否正确无空格无引号。2. 确认环境变量已加载load_dotenv()。3. 在llm_client.py中打印os.getenv(“KEY”)验证是否读取成功。RateLimitErrorAPI 调用频率超限。1. 检查对应平台的用量和频率限制。2. 在代码中添加延迟如time.sleep(1)。3. 考虑使用更低成本的模型进行开发测试。智能体输出格式不符合预期JSON解析失败LLM 没有严格按照提示词中的格式输出。1. 降低temperature参数如设为 0.2。2. 在提示词中更严格地规定格式使用json.dumps示例。3. 实现一个更健壮的输出解析器如JsonOutputParser。工作流陷入无限循环审核逻辑_decide_after_review判断条件不明确总是返回”revise”。1. 检查审核反馈的判断逻辑确保有明确的“通过”条件。2. 设置合理的max_revisions。3. 在状态中增加revision_count并打印监控循环次数。DeepSeek 模型调用失败DeepSeek 的 API 端点或参数与 OpenAI 不完全兼容。1. 查阅 DeepSeek 官方最新的 API 文档。2. 确认openai_api_base和model参数是否正确。3. 考虑使用 LangChain 的ChatDeepSeek类如果官方提供。程序中途退出无错误信息可能某个智能体调用 LLM 超时或网络中断。1. 在关键函数调用处添加try…except捕获异常并打印。2. 为 LLM 调用增加超时参数如ChatOpenAI(…, request_timeout60)。3. 检查网络连接。7. 最佳实践与项目扩展建议掌握了基础框架后你可以遵循以下最佳实践来优化项目并探索更高级的功能。7.1 工程化最佳实践配置管理将模型类型、温度、最大修订次数等参数提取到配置文件如config.yaml中避免硬编码。日志记录使用 Python 的logging模块替代print可以输出到文件并设置不同级别INFO, DEBUG, ERROR方便调试和监控。异常处理与重试LLM API 调用可能不稳定使用tenacity库为关键操作添加重试机制和退避策略。状态持久化将BlogState在关键节点序列化保存到数据库或文件以便工作流中断后可以恢复。测试为每个智能体编写单元测试模拟 LLM 响应确保业务逻辑正确。为工作流编写集成测试。7.2 性能与成本优化模型选择策划和审核可以使用较小、较快的模型如 GPT-3.5-Turbo写作使用能力更强的模型如 GPT-4以平衡效果与成本/速度。缓存对相对固定的提示词和 LLM 响应使用缓存如langchain.cache减少重复调用和费用。异步调用如果智能体间没有严格的先后依赖可以使用asyncio实现并行执行大幅缩短总耗时。7.3 功能扩展方向增加更多智能体SEO优化智能体在发布前分析并优化内容的关键词密度和元描述。配图建议智能体根据文章内容生成配图的关键词描述或调用文生图 API。社交媒体摘要智能体从长文中提取要点生成适合 Twitter、微博的短文案。增强工作流逻辑并行分支让“写作”和“寻找参考链接”两个智能体同时工作。人工审核节点在自动审核后引入一个等待人工输入HumanApproval的节点只有人工确认后才能继续。动态工具调用让智能体在运行过程中自主决定是否调用网络搜索、代码执行等工具来获取信息。集成外部系统向量数据库让策划智能体先检索相关的历史文章或知识库使大纲更有依据。CMS发布工作流最后增加一个自动发布到 WordPress、Confluence 等系统的智能体。监控与告警将工作流的执行状态、耗时、成本指标发送到 Prometheus 或 Datadog。通过这个从零开始的“DeepAgents”项目实战你已经掌握了使用 LangChain 构建智能体、并用 LangGraph 编排复杂多智能体工作流的核心方法。这套模式具有很强的通用性你可以轻松地将“博客生成团队”替换为“客服问答团队”、“代码审查团队”或“数据分析团队”只需定义新的智能体角色和状态结构即可。多智能体系统的魅力在于其模块化和可编排性希望本文能成为你探索更复杂 AI 应用架构的坚实起点。如果在实践过程中遇到问题欢迎在评论区交流探讨。
