最近在尝试将大语言模型(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_KEY=your_openai_api_key_here # 或者使用 DeepSeek (注意:DeepSeek API Base URL 可能不同,请查阅最新文档) DEEPSEEK_API_KEY=your_deepseek_api_key_here DEEPSEEK_API_BASE=https://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_type="openai", temperature=0.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(model="gpt-4-turbo-preview", api_key=api_key, temperature=temperature) 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( model="deepseek-chat", openai_api_key=api_key, openai_api_base=api_base, temperature=temperature ) 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_type="openai"): self.llm = get_llm(llm_model_type, temperature=0.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, indent=2, ensure_ascii=False))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_type="openai"): self.llm = get_llm(llm_model_type, temperature=0.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_type="openai"): self.llm = get_llm(llm_model_type, temperature=0.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_type="openai"): 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_type=model_type) final_state = workflow.run(topic, requirements, max_revisions=2) # 设置最大修订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 = f"blog_{topic[:20]}.md".replace(" ", "_").replace("/", "_") with open(filename, 'w', encoding='utf-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 Key | API 密钥错误或未设置。 | 1. 检查.env文件是否存在,格式是否正确(无空格,无引号)。2. 确认环境变量已加载( load_dotenv())。3. 在 llm_client.py中打印os.getenv(“KEY”)验证是否读取成功。 |
RateLimitError | API 调用频率超限。 | 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_timeout=60)。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 应用架构的坚实起点。如果在实践过程中遇到问题,欢迎在评论区交流探讨。