如果你最近在尝试构建多智能体系统大概率会遇到这样的困境单智能体能力很强但一旦需要多个智能体协作代码就迅速变得复杂难控。状态流转混乱、消息传递无序、异常处理缺失——这些问题让很多开发者止步于“Demo能跑”却难以投入实际使用。这正是 LangGraph 要解决的核心问题。它不是一个替代 LangChain 的框架而是一个专门为多智能体协作设计的编排层。LangGraph 的关键洞察是把多智能体系统看作一个有向图每个节点是一个智能体或决策点边是状态流转路径。这种抽象让复杂协作变得可描述、可调试、可扩展。1. 先理解 LangGraph 要解决的根本问题从单点智能到可控协作在 LangGraph 之前我们构建多智能体系统通常有两种方式一种是硬编码的线性流程另一种是完全自由的对话式交互。前者缺乏灵活性后者难以控制。真正的生产系统需要在这两者之间找到平衡——既要有明确的业务流程又要允许智能体根据情境自主决策。1.1 为什么多智能体编排比单智能体复杂得多单智能体的核心是“输入-处理-输出”的直线流程。而多智能体系统需要处理的是状态共享智能体之间需要传递什么信息哪些状态需要持久化流程控制什么时候该调用哪个智能体什么条件下流程应该结束错误处理某个智能体失败时系统应该如何降级或重试资源管理如何避免多个智能体同时访问冲突资源这些问题的复杂度不是线性增加的。当智能体数量从1个增加到3个时可能的状态路径会呈指数级增长。这就是为什么需要专门的编排框架。1.2 LangGraph 的图抽象如何简化复杂性LangGraph 将整个系统建模为一个有向图Graph。这个抽象的好处是可视化调试你可以直观地看到状态在智能体之间的流转路径明确边界每个节点有明确的输入输出规范可控循环可以在特定条件下让流程循环执行直到满足退出条件条件分支根据中间结果选择不同的执行路径这种思维方式的变化很关键从“我要如何控制每个智能体”转变为“我如何设计智能体之间的协作规则”。2. LangGraph 核心概念拆解从理论到实践的关键桥梁要真正用好 LangGraph需要理解几个核心概念。这些概念构成了 LangGraph 的编程模型也是后续实战的基础。2.1 State 设计多智能体系统的共享内存State 是 LangGraph 中最重要的概念之一它定义了在整个图执行过程中共享的数据结构。设计良好的 State 是构建稳定多智能体系统的前提。from typing import Annotated, List from typing_extensions import TypedDict from langgraph.graph import add_messages class AgentState(TypedDict): messages: Annotated[List[str], add_messages] current_agent: str task_description: str intermediate_results: List[dict] error_count: intState 设计的关键原则最小化共享只共享必要的字段避免不必要的耦合明确语义每个字段应该有清晰的用途定义考虑序列化State 需要能够被持久化避免使用不可序列化的对象错误容忍包含错误计数、重试状态等字段2.2 Node 实现智能体的职责边界在 LangGraph 中每个 Node 代表一个处理单元通常对应一个智能体。Node 的设计应该遵循单一职责原则。def research_agent(state: AgentState) - AgentState: 研究型智能体负责信息搜集和分析 # 从共享状态中获取需要研究的问题 question state[messages][-1] if state[messages] else state[task_description] # 调用研究工具或API research_result research_tool(question) # 更新状态为下一个智能体准备数据 state[intermediate_results].append({ agent: research_agent, result: research_result, timestamp: datetime.now().isoformat() }) # 决定下一个执行的智能体 if research_result.get(needs_deep_analysis): state[current_agent] analysis_agent else: state[current_agent] summarize_agent return state2.3 Edge 配置控制流程的逻辑枢纽Edge 决定了状态在 Node 之间的流转路径。LangGraph 提供了几种类型的 Edge条件边Conditional Edge根据当前状态值选择下一个节点固定边Fixed Edge无条件跳转到指定节点动态边根据节点执行结果动态决定下一跳def should_continue(state: AgentState) - str: 根据当前状态决定下一步执行哪个智能体 if state.get(error_count, 0) 3: return error_handling_agent current_agent state[current_agent] if current_agent research_agent: return analysis_agent elif current_agent analysis_agent: return review_agent else: return __end__3. 构建第一个可用的多智能体系统从零到一的完整路径理论理解之后我们通过一个实际案例来构建完整的多智能体系统。这个案例是一个内容创作助手包含研究、分析、写作、审核四个智能体。3.1 环境准备和基础配置首先确保安装必要的依赖pip install langgraph langchain-openai然后配置基础环境import os from langchain_openai import ChatOpenAI # 配置LLM - 可以根据需要为不同智能体配置不同的模型 research_llm ChatOpenAI(modelgpt-4, temperature0.7) analysis_llm ChatOpenAI(modelgpt-4, temperature0.3) # 分析需要更确定性 writing_llm ChatOpenAI(modelgpt-4, temperature0.8) # 写作需要更多创造性 review_llm ChatOpenAI(modelgpt-4, temperature0.2) # 审核需要严格性 # 状态定义 class ContentCreationState(TypedDict): topic: str research_materials: List[dict] outline: str draft: str feedback: List[str] current_step: str completed_steps: List[str]3.2 实现四个核心智能体节点每个智能体都有明确的职责和输入输出规范def research_agent_node(state: ContentCreationState) - ContentCreationState: 研究智能体搜集主题相关材料 if state[current_step] ! research: return state # 如果不是研究阶段直接返回 topic state[topic] # 模拟研究过程 - 实际项目中这里会调用搜索API等 research_results [ {source: source_1, content: f关于{topic}的关键信息1}, {source: source_2, content: f关于{topic}的关键信息2} ] state[research_materials] research_results state[current_step] analysis state[completed_steps].append(research) return state def analysis_agent_node(state: ContentCreationState) - ContentCreationState: 分析智能体基于研究材料生成大纲 if state[current_step] ! analysis: return state materials state[research_materials] # 基于研究材料生成内容大纲 outline_prompt f 基于以下研究材料为主题{state[topic]}生成一个内容大纲 {materials} 大纲应该包含主要章节和关键点。 response analysis_llm.invoke(outline_prompt) state[outline] response.content state[current_step] writing state[completed_steps].append(analysis) return state def writing_agent_node(state: ContentCreationState) - ContentCreationState: 写作智能体根据大纲撰写初稿 if state[current_step] ! writing: return state outline state[outline] writing_prompt f 根据以下大纲撰写完整内容 {outline} 研究材料参考{state[research_materials]} response writing_llm.invoke(writing_prompt) state[draft] response.content state[current_step] review state[completed_steps].append(writing) return state def review_agent_node(state: ContentCreationState) - ContentCreationState: 审核智能体检查内容质量并提供反馈 if state[current_step] ! review: return state draft state[draft] review_prompt f 审核以下内容提供改进建议 {draft} 重点关注逻辑连贯性、事实准确性、语言表达。 response review_llm.invoke(review_prompt) state[feedback].append(response.content) # 根据反馈质量决定是否重新写作 if 重大修改 in response.content: state[current_step] writing # 返回写作阶段 else: state[current_step] complete state[completed_steps].append(review) return state3.3 构建完整的图和工作流将各个智能体节点连接成完整的工作流from langgraph.graph import StateGraph, END # 创建图结构 workflow StateGraph(ContentCreationState) # 添加节点 workflow.add_node(research, research_agent_node) workflow.add_node(analysis, analysis_agent_node) workflow.add_node(writing, writing_agent_node) workflow.add_node(review, review_agent_node) # 设置边和流转逻辑 workflow.set_entry_point(research) workflow.add_edge(research, analysis) workflow.add_edge(analysis, writing) workflow.add_edge(writing, review) def route_after_review(state: ContentCreationState): 审核后的路由逻辑 if state[current_step] writing: # 需要重新写作 return writing elif state[current_step] complete: # 完成 return END else: # 继续审核 return review workflow.add_conditional_edges( review, route_after_review, { writing: writing, review: review, END: END } ) # 编译图 graph workflow.compile()3.4 执行和测试工作流现在可以测试这个多智能体系统# 初始化状态 initial_state { topic: 人工智能在医疗诊断中的应用, research_materials: [], outline: , draft: , feedback: [], current_step: research, completed_steps: [] } # 执行工作流 final_state graph.invoke(initial_state) print(最终结果) print(f完成步骤: {final_state[completed_steps]}) print(f生成内容长度: {len(final_state[draft])}) print(f审核反馈: {final_state[feedback]})4. 生产环境的关键考量从能跑到好用的差距一个能在笔记本上运行的多智能体系统与能在生产环境稳定运行的系统之间存在巨大的工程化差距。以下是必须考虑的关键问题。4.1 状态持久化和恢复生产环境中工作流执行可能被中断服务器重启、超时等需要能够从断点恢复import json from datetime import datetime class StateManager: def __init__(self, storage_path./state_storage): self.storage_path storage_path os.makedirs(storage_path, exist_okTrue) def save_state(self, workflow_id: str, state: dict): 保存状态到文件 filename f{self.storage_path}/{workflow_id}.json with open(filename, w, encodingutf-8) as f: json.dump({ state: state, timestamp: datetime.now().isoformat(), version: 1.0 }, f, ensure_asciiFalse, indent2) def load_state(self, workflow_id: str) - dict: 从文件加载状态 filename f{self.storage_path}/{workflow_id}.json try: with open(filename, r, encodingutf-8) as f: data json.load(f) return data[state] except FileNotFoundError: return None # 使用示例 state_manager StateManager() def execute_with_persistence(workflow_id, initial_state, graph): # 尝试加载已有状态 saved_state state_manager.load_state(workflow_id) if saved_state: current_state saved_state print(f从断点恢复工作流: {workflow_id}) else: current_state initial_state print(f开始新工作流: {workflow_id}) # 执行并定期保存状态 for step, next_state in graph.stream(current_state): state_manager.save_state(workflow_id, next_state) yield step, next_state4.2 错误处理和重试机制多智能体系统中错误处理比单智能体复杂得多from tenacity import retry, stop_after_attempt, wait_exponential class ErrorHandlingAgent: def __init__(self, max_retries3): self.max_retries max_retries retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def execute_with_retry(self, agent_func, state, agent_name): 带重试机制的智能体执行 try: return agent_func(state) except Exception as e: print(f智能体 {agent_name} 执行失败: {e}) # 更新错误计数 state[error_count] state.get(error_count, 0) 1 state[last_error] { agent: agent_name, error: str(e), timestamp: datetime.now().isoformat() } if state[error_count] self.max_retries: state[current_step] error_recovery return state else: raise # 触发重试 def create_robust_agent(original_agent_func, agent_name): 创建带错误处理的智能体包装器 error_handler ErrorHandlingAgent() def robust_agent(state): return error_handler.execute_with_retry(original_agent_func, state, agent_name) return robust_agent # 使用稳健的智能体 robust_research_agent create_robust_agent(research_agent_node, research_agent)4.3 性能监控和优化生产环境需要监控每个智能体的性能import time from collections import defaultdict class PerformanceMonitor: def __init__(self): self.metrics defaultdict(list) def track_execution(self, agent_name): 跟踪智能体执行时间和资源使用 start_time time.time() start_memory self.get_memory_usage() def callback(result): end_time time.time() end_memory self.get_memory_usage() self.metrics[agent_name].append({ execution_time: end_time - start_time, memory_delta: end_memory - start_memory, timestamp: datetime.now().isoformat() }) return result return callback def get_memory_usage(self): 获取内存使用情况简化版 import psutil process psutil.Process() return process.memory_info().rss / 1024 / 1024 # MB def get_performance_report(self): 生成性能报告 report {} for agent_name, metrics_list in self.metrics.items(): if metrics_list: times [m[execution_time] for m in metrics_list] report[agent_name] { avg_time: sum(times) / len(times), max_time: max(times), min_time: min(times), total_executions: len(metrics_list) } return report # 集成性能监控 monitor PerformanceMonitor() def monitored_agent(original_agent_func, agent_name): 创建带监控的智能体 def wrapper(state): callback monitor.track_execution(agent_name) result original_agent_func(state) return callback(result) return wrapper5. 高级模式和应用场景超越基础工作流掌握了基础的多智能体构建后可以探索更复杂的模式和场景这些模式能够解决更实际的业务问题。5.1 竞争性多智能体模式在某些场景下让多个智能体竞争性地解决同一个问题然后选择最佳方案def competitive_analysis_workflow(topic: str): 竞争性分析多个分析智能体独立工作然后评估最佳结果 # 定义不同专长的分析智能体 specialists { technical_analyst: 专注于技术实现细节, business_analyst: 专注于商业价值和可行性, user_experience_analyst: 专注于用户体验影响 } # 并行执行多个分析 analysis_results {} for specialist, focus in specialists.items(): prompt f 作为{specialist}{focus}分析以下主题 {topic} 请从你的专业角度提供深度分析。 result analysis_llm.invoke(prompt) analysis_results[specialist] { content: result.content, focus: focus, timestamp: datetime.now().isoformat() } # 评估哪个分析最全面 evaluation_prompt f 评估以下三个分析报告选择最全面和深入的一个 {json.dumps(analysis_results, ensure_asciiFalse, indent2)} 请给出选择理由。 evaluation review_llm.invoke(evaluation_prompt) return { all_analyses: analysis_results, selected_analysis: evaluation.content, selection_reason: 基于全面性评估 }5.2 分层决策模式对于复杂决策可以采用分层的方法先由基层智能体处理疑难问题向上级传递class HierarchicalDecisionSystem: def __init__(self): self.levels { level_1: {name: 初级决策, complexity_threshold: 0.3}, level_2: {name: 中级决策, complexity_threshold: 0.6}, level_3: {name: 高级决策, complexity_threshold: 1.0} } def assess_complexity(self, problem: str) - float: 评估问题复杂度 prompt f 评估以下问题的复杂度0-1分数 {problem} 考虑因素技术难度、涉及领域数量、决策影响范围。 只返回0-1之间的数字。 response analysis_llm.invoke(prompt) try: return float(response.content.strip()) except: return 0.5 # 默认值 def route_to_appropriate_level(self, problem: str): 将问题路由到合适的决策层级 complexity self.assess_complexity(problem) for level_id, level_info in self.levels.items(): if complexity level_info[complexity_threshold]: return self.solve_at_level(level_id, problem, complexity) return self.solve_at_level(level_3, problem, complexity) # 默认最高级 def solve_at_level(self, level_id: str, problem: str, complexity: float): 在指定层级解决问题 level_prompt f 作为{self.levels[level_id][name]}智能体解决以下问题复杂度{complexity} {problem} response analysis_llm.invoke(level_prompt) return { solution: response.content, solved_at_level: level_id, complexity_score: complexity, confidence: self.assess_confidence(response.content) }5.3 动态工作流调整根据运行时信息动态调整工作流结构def dynamic_workflow_orchestrator(initial_task, context): 动态工作流编排器根据任务特征调整智能体组合 # 分析任务特征 task_analysis analyze_task_requirements(initial_task, context) # 根据特征选择智能体组合 if task_analysis[requires_creativity] 0.7: agent_sequence [brainstorm_agent, research_agent, creative_writing_agent] elif task_analysis[requires_accuracy] 0.7: agent_sequence [research_agent, fact_check_agent, precision_writing_agent] else: agent_sequence [research_agent, analysis_agent, writing_agent] # 动态构建工作流 dynamic_workflow StateGraph(ContentCreationState) # 动态添加节点 available_agents get_available_agents() # 获取所有注册的智能体 for i, agent_name in enumerate(agent_sequence): if agent_name in available_agents: dynamic_workflow.add_node(agent_name, available_agents[agent_name]) # 动态设置边 for i in range(len(agent_sequence) - 1): current_agent agent_sequence[i] next_agent agent_sequence[i 1] if current_agent in available_agents and next_agent in available_agents: dynamic_workflow.add_edge(current_agent, next_agent) dynamic_workflow.set_entry_point(agent_sequence[0]) dynamic_workflow.add_edge(agent_sequence[-1], END) return dynamic_workflow.compile()6. 实际项目中的经验教训和最佳实践经过多个项目的实践总结出一些关键的经验教训这些经验可以帮助避免常见的陷阱。6.1 智能体设计的单一职责原则每个智能体应该只负责一个明确的职责。过度复杂的智能体会导致调试困难问题定位不明确复用性差难以在其他工作流中使用性能瓶颈单个智能体成为系统瓶颈正确的做法是宁可多设计几个简单的智能体也不要设计一个复杂的万能智能体。6.2 状态设计的版本兼容性State 结构一旦确定修改要非常谨慎。实际项目中建议为 State 添加版本字段重大变更时保持向后兼容提供状态迁移工具在文档中明确每个字段的语义和生命周期class VersionedState(TypedDict): schema_version: str # 如 1.0.0 data: dict metadata: dict6.3 测试策略从单元测试到集成测试多智能体系统的测试需要分层进行智能体单元测试测试单个智能体的输入输出工作流集成测试测试完整的工作流执行性能压力测试测试并发执行和资源使用故障恢复测试测试错误处理和状态恢复import pytest class TestMultiAgentSystem: def test_individual_agent(self): 测试单个智能体 test_state {topic: 测试主题, current_step: research} result research_agent_node(test_state) assert research_materials in result assert result[current_step] analysis def test_workflow_integration(self): 测试完整工作流 initial_state { topic: 集成测试主题, current_step: research, research_materials: [], completed_steps: [] } final_state graph.invoke(initial_state) assert complete in final_state[completed_steps] def test_error_recovery(self): 测试错误恢复机制 # 模拟智能体失败场景 # 验证系统能够正常恢复6.4 监控和可观测性生产环境中监控比功能更重要执行轨迹记录记录每个智能体的输入输出性能指标收集收集执行时间、资源使用等指标业务指标跟踪跟踪工作流成功率、完成时间等业务指标异常报警设置智能的异常检测和报警机制6.5 团队协作和文档维护多智能体系统通常需要团队协作开发为每个智能体编写清晰的接口文档使用版本控制管理工作流定义建立代码审查流程特别是状态结构的变更维护工作流的可视化文档LangGraph 的真正价值不在于让多智能体系统能够运行而在于让复杂协作变得可管理、可调试、可演进。从简单的线性工作流开始逐步引入条件分支、循环、错误处理等复杂逻辑最终构建出能够应对真实业务场景的智能系统。关键是要记住好的多智能体系统不是一次性设计出来的而是通过不断迭代优化出来的。从最小可行产品开始在实际使用中发现问题、调整设计、优化性能这才是构建可持续多智能体系统的正确路径。 LangGraph多智能体系统编排：从图抽象到生产实践
如果你最近在尝试构建多智能体系统大概率会遇到这样的困境单智能体能力很强但一旦需要多个智能体协作代码就迅速变得复杂难控。状态流转混乱、消息传递无序、异常处理缺失——这些问题让很多开发者止步于“Demo能跑”却难以投入实际使用。这正是 LangGraph 要解决的核心问题。它不是一个替代 LangChain 的框架而是一个专门为多智能体协作设计的编排层。LangGraph 的关键洞察是把多智能体系统看作一个有向图每个节点是一个智能体或决策点边是状态流转路径。这种抽象让复杂协作变得可描述、可调试、可扩展。1. 先理解 LangGraph 要解决的根本问题从单点智能到可控协作在 LangGraph 之前我们构建多智能体系统通常有两种方式一种是硬编码的线性流程另一种是完全自由的对话式交互。前者缺乏灵活性后者难以控制。真正的生产系统需要在这两者之间找到平衡——既要有明确的业务流程又要允许智能体根据情境自主决策。1.1 为什么多智能体编排比单智能体复杂得多单智能体的核心是“输入-处理-输出”的直线流程。而多智能体系统需要处理的是状态共享智能体之间需要传递什么信息哪些状态需要持久化流程控制什么时候该调用哪个智能体什么条件下流程应该结束错误处理某个智能体失败时系统应该如何降级或重试资源管理如何避免多个智能体同时访问冲突资源这些问题的复杂度不是线性增加的。当智能体数量从1个增加到3个时可能的状态路径会呈指数级增长。这就是为什么需要专门的编排框架。1.2 LangGraph 的图抽象如何简化复杂性LangGraph 将整个系统建模为一个有向图Graph。这个抽象的好处是可视化调试你可以直观地看到状态在智能体之间的流转路径明确边界每个节点有明确的输入输出规范可控循环可以在特定条件下让流程循环执行直到满足退出条件条件分支根据中间结果选择不同的执行路径这种思维方式的变化很关键从“我要如何控制每个智能体”转变为“我如何设计智能体之间的协作规则”。2. LangGraph 核心概念拆解从理论到实践的关键桥梁要真正用好 LangGraph需要理解几个核心概念。这些概念构成了 LangGraph 的编程模型也是后续实战的基础。2.1 State 设计多智能体系统的共享内存State 是 LangGraph 中最重要的概念之一它定义了在整个图执行过程中共享的数据结构。设计良好的 State 是构建稳定多智能体系统的前提。from typing import Annotated, List from typing_extensions import TypedDict from langgraph.graph import add_messages class AgentState(TypedDict): messages: Annotated[List[str], add_messages] current_agent: str task_description: str intermediate_results: List[dict] error_count: intState 设计的关键原则最小化共享只共享必要的字段避免不必要的耦合明确语义每个字段应该有清晰的用途定义考虑序列化State 需要能够被持久化避免使用不可序列化的对象错误容忍包含错误计数、重试状态等字段2.2 Node 实现智能体的职责边界在 LangGraph 中每个 Node 代表一个处理单元通常对应一个智能体。Node 的设计应该遵循单一职责原则。def research_agent(state: AgentState) - AgentState: 研究型智能体负责信息搜集和分析 # 从共享状态中获取需要研究的问题 question state[messages][-1] if state[messages] else state[task_description] # 调用研究工具或API research_result research_tool(question) # 更新状态为下一个智能体准备数据 state[intermediate_results].append({ agent: research_agent, result: research_result, timestamp: datetime.now().isoformat() }) # 决定下一个执行的智能体 if research_result.get(needs_deep_analysis): state[current_agent] analysis_agent else: state[current_agent] summarize_agent return state2.3 Edge 配置控制流程的逻辑枢纽Edge 决定了状态在 Node 之间的流转路径。LangGraph 提供了几种类型的 Edge条件边Conditional Edge根据当前状态值选择下一个节点固定边Fixed Edge无条件跳转到指定节点动态边根据节点执行结果动态决定下一跳def should_continue(state: AgentState) - str: 根据当前状态决定下一步执行哪个智能体 if state.get(error_count, 0) 3: return error_handling_agent current_agent state[current_agent] if current_agent research_agent: return analysis_agent elif current_agent analysis_agent: return review_agent else: return __end__3. 构建第一个可用的多智能体系统从零到一的完整路径理论理解之后我们通过一个实际案例来构建完整的多智能体系统。这个案例是一个内容创作助手包含研究、分析、写作、审核四个智能体。3.1 环境准备和基础配置首先确保安装必要的依赖pip install langgraph langchain-openai然后配置基础环境import os from langchain_openai import ChatOpenAI # 配置LLM - 可以根据需要为不同智能体配置不同的模型 research_llm ChatOpenAI(modelgpt-4, temperature0.7) analysis_llm ChatOpenAI(modelgpt-4, temperature0.3) # 分析需要更确定性 writing_llm ChatOpenAI(modelgpt-4, temperature0.8) # 写作需要更多创造性 review_llm ChatOpenAI(modelgpt-4, temperature0.2) # 审核需要严格性 # 状态定义 class ContentCreationState(TypedDict): topic: str research_materials: List[dict] outline: str draft: str feedback: List[str] current_step: str completed_steps: List[str]3.2 实现四个核心智能体节点每个智能体都有明确的职责和输入输出规范def research_agent_node(state: ContentCreationState) - ContentCreationState: 研究智能体搜集主题相关材料 if state[current_step] ! research: return state # 如果不是研究阶段直接返回 topic state[topic] # 模拟研究过程 - 实际项目中这里会调用搜索API等 research_results [ {source: source_1, content: f关于{topic}的关键信息1}, {source: source_2, content: f关于{topic}的关键信息2} ] state[research_materials] research_results state[current_step] analysis state[completed_steps].append(research) return state def analysis_agent_node(state: ContentCreationState) - ContentCreationState: 分析智能体基于研究材料生成大纲 if state[current_step] ! analysis: return state materials state[research_materials] # 基于研究材料生成内容大纲 outline_prompt f 基于以下研究材料为主题{state[topic]}生成一个内容大纲 {materials} 大纲应该包含主要章节和关键点。 response analysis_llm.invoke(outline_prompt) state[outline] response.content state[current_step] writing state[completed_steps].append(analysis) return state def writing_agent_node(state: ContentCreationState) - ContentCreationState: 写作智能体根据大纲撰写初稿 if state[current_step] ! writing: return state outline state[outline] writing_prompt f 根据以下大纲撰写完整内容 {outline} 研究材料参考{state[research_materials]} response writing_llm.invoke(writing_prompt) state[draft] response.content state[current_step] review state[completed_steps].append(writing) return state def review_agent_node(state: ContentCreationState) - ContentCreationState: 审核智能体检查内容质量并提供反馈 if state[current_step] ! review: return state draft state[draft] review_prompt f 审核以下内容提供改进建议 {draft} 重点关注逻辑连贯性、事实准确性、语言表达。 response review_llm.invoke(review_prompt) state[feedback].append(response.content) # 根据反馈质量决定是否重新写作 if 重大修改 in response.content: state[current_step] writing # 返回写作阶段 else: state[current_step] complete state[completed_steps].append(review) return state3.3 构建完整的图和工作流将各个智能体节点连接成完整的工作流from langgraph.graph import StateGraph, END # 创建图结构 workflow StateGraph(ContentCreationState) # 添加节点 workflow.add_node(research, research_agent_node) workflow.add_node(analysis, analysis_agent_node) workflow.add_node(writing, writing_agent_node) workflow.add_node(review, review_agent_node) # 设置边和流转逻辑 workflow.set_entry_point(research) workflow.add_edge(research, analysis) workflow.add_edge(analysis, writing) workflow.add_edge(writing, review) def route_after_review(state: ContentCreationState): 审核后的路由逻辑 if state[current_step] writing: # 需要重新写作 return writing elif state[current_step] complete: # 完成 return END else: # 继续审核 return review workflow.add_conditional_edges( review, route_after_review, { writing: writing, review: review, END: END } ) # 编译图 graph workflow.compile()3.4 执行和测试工作流现在可以测试这个多智能体系统# 初始化状态 initial_state { topic: 人工智能在医疗诊断中的应用, research_materials: [], outline: , draft: , feedback: [], current_step: research, completed_steps: [] } # 执行工作流 final_state graph.invoke(initial_state) print(最终结果) print(f完成步骤: {final_state[completed_steps]}) print(f生成内容长度: {len(final_state[draft])}) print(f审核反馈: {final_state[feedback]})4. 生产环境的关键考量从能跑到好用的差距一个能在笔记本上运行的多智能体系统与能在生产环境稳定运行的系统之间存在巨大的工程化差距。以下是必须考虑的关键问题。4.1 状态持久化和恢复生产环境中工作流执行可能被中断服务器重启、超时等需要能够从断点恢复import json from datetime import datetime class StateManager: def __init__(self, storage_path./state_storage): self.storage_path storage_path os.makedirs(storage_path, exist_okTrue) def save_state(self, workflow_id: str, state: dict): 保存状态到文件 filename f{self.storage_path}/{workflow_id}.json with open(filename, w, encodingutf-8) as f: json.dump({ state: state, timestamp: datetime.now().isoformat(), version: 1.0 }, f, ensure_asciiFalse, indent2) def load_state(self, workflow_id: str) - dict: 从文件加载状态 filename f{self.storage_path}/{workflow_id}.json try: with open(filename, r, encodingutf-8) as f: data json.load(f) return data[state] except FileNotFoundError: return None # 使用示例 state_manager StateManager() def execute_with_persistence(workflow_id, initial_state, graph): # 尝试加载已有状态 saved_state state_manager.load_state(workflow_id) if saved_state: current_state saved_state print(f从断点恢复工作流: {workflow_id}) else: current_state initial_state print(f开始新工作流: {workflow_id}) # 执行并定期保存状态 for step, next_state in graph.stream(current_state): state_manager.save_state(workflow_id, next_state) yield step, next_state4.2 错误处理和重试机制多智能体系统中错误处理比单智能体复杂得多from tenacity import retry, stop_after_attempt, wait_exponential class ErrorHandlingAgent: def __init__(self, max_retries3): self.max_retries max_retries retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def execute_with_retry(self, agent_func, state, agent_name): 带重试机制的智能体执行 try: return agent_func(state) except Exception as e: print(f智能体 {agent_name} 执行失败: {e}) # 更新错误计数 state[error_count] state.get(error_count, 0) 1 state[last_error] { agent: agent_name, error: str(e), timestamp: datetime.now().isoformat() } if state[error_count] self.max_retries: state[current_step] error_recovery return state else: raise # 触发重试 def create_robust_agent(original_agent_func, agent_name): 创建带错误处理的智能体包装器 error_handler ErrorHandlingAgent() def robust_agent(state): return error_handler.execute_with_retry(original_agent_func, state, agent_name) return robust_agent # 使用稳健的智能体 robust_research_agent create_robust_agent(research_agent_node, research_agent)4.3 性能监控和优化生产环境需要监控每个智能体的性能import time from collections import defaultdict class PerformanceMonitor: def __init__(self): self.metrics defaultdict(list) def track_execution(self, agent_name): 跟踪智能体执行时间和资源使用 start_time time.time() start_memory self.get_memory_usage() def callback(result): end_time time.time() end_memory self.get_memory_usage() self.metrics[agent_name].append({ execution_time: end_time - start_time, memory_delta: end_memory - start_memory, timestamp: datetime.now().isoformat() }) return result return callback def get_memory_usage(self): 获取内存使用情况简化版 import psutil process psutil.Process() return process.memory_info().rss / 1024 / 1024 # MB def get_performance_report(self): 生成性能报告 report {} for agent_name, metrics_list in self.metrics.items(): if metrics_list: times [m[execution_time] for m in metrics_list] report[agent_name] { avg_time: sum(times) / len(times), max_time: max(times), min_time: min(times), total_executions: len(metrics_list) } return report # 集成性能监控 monitor PerformanceMonitor() def monitored_agent(original_agent_func, agent_name): 创建带监控的智能体 def wrapper(state): callback monitor.track_execution(agent_name) result original_agent_func(state) return callback(result) return wrapper5. 高级模式和应用场景超越基础工作流掌握了基础的多智能体构建后可以探索更复杂的模式和场景这些模式能够解决更实际的业务问题。5.1 竞争性多智能体模式在某些场景下让多个智能体竞争性地解决同一个问题然后选择最佳方案def competitive_analysis_workflow(topic: str): 竞争性分析多个分析智能体独立工作然后评估最佳结果 # 定义不同专长的分析智能体 specialists { technical_analyst: 专注于技术实现细节, business_analyst: 专注于商业价值和可行性, user_experience_analyst: 专注于用户体验影响 } # 并行执行多个分析 analysis_results {} for specialist, focus in specialists.items(): prompt f 作为{specialist}{focus}分析以下主题 {topic} 请从你的专业角度提供深度分析。 result analysis_llm.invoke(prompt) analysis_results[specialist] { content: result.content, focus: focus, timestamp: datetime.now().isoformat() } # 评估哪个分析最全面 evaluation_prompt f 评估以下三个分析报告选择最全面和深入的一个 {json.dumps(analysis_results, ensure_asciiFalse, indent2)} 请给出选择理由。 evaluation review_llm.invoke(evaluation_prompt) return { all_analyses: analysis_results, selected_analysis: evaluation.content, selection_reason: 基于全面性评估 }5.2 分层决策模式对于复杂决策可以采用分层的方法先由基层智能体处理疑难问题向上级传递class HierarchicalDecisionSystem: def __init__(self): self.levels { level_1: {name: 初级决策, complexity_threshold: 0.3}, level_2: {name: 中级决策, complexity_threshold: 0.6}, level_3: {name: 高级决策, complexity_threshold: 1.0} } def assess_complexity(self, problem: str) - float: 评估问题复杂度 prompt f 评估以下问题的复杂度0-1分数 {problem} 考虑因素技术难度、涉及领域数量、决策影响范围。 只返回0-1之间的数字。 response analysis_llm.invoke(prompt) try: return float(response.content.strip()) except: return 0.5 # 默认值 def route_to_appropriate_level(self, problem: str): 将问题路由到合适的决策层级 complexity self.assess_complexity(problem) for level_id, level_info in self.levels.items(): if complexity level_info[complexity_threshold]: return self.solve_at_level(level_id, problem, complexity) return self.solve_at_level(level_3, problem, complexity) # 默认最高级 def solve_at_level(self, level_id: str, problem: str, complexity: float): 在指定层级解决问题 level_prompt f 作为{self.levels[level_id][name]}智能体解决以下问题复杂度{complexity} {problem} response analysis_llm.invoke(level_prompt) return { solution: response.content, solved_at_level: level_id, complexity_score: complexity, confidence: self.assess_confidence(response.content) }5.3 动态工作流调整根据运行时信息动态调整工作流结构def dynamic_workflow_orchestrator(initial_task, context): 动态工作流编排器根据任务特征调整智能体组合 # 分析任务特征 task_analysis analyze_task_requirements(initial_task, context) # 根据特征选择智能体组合 if task_analysis[requires_creativity] 0.7: agent_sequence [brainstorm_agent, research_agent, creative_writing_agent] elif task_analysis[requires_accuracy] 0.7: agent_sequence [research_agent, fact_check_agent, precision_writing_agent] else: agent_sequence [research_agent, analysis_agent, writing_agent] # 动态构建工作流 dynamic_workflow StateGraph(ContentCreationState) # 动态添加节点 available_agents get_available_agents() # 获取所有注册的智能体 for i, agent_name in enumerate(agent_sequence): if agent_name in available_agents: dynamic_workflow.add_node(agent_name, available_agents[agent_name]) # 动态设置边 for i in range(len(agent_sequence) - 1): current_agent agent_sequence[i] next_agent agent_sequence[i 1] if current_agent in available_agents and next_agent in available_agents: dynamic_workflow.add_edge(current_agent, next_agent) dynamic_workflow.set_entry_point(agent_sequence[0]) dynamic_workflow.add_edge(agent_sequence[-1], END) return dynamic_workflow.compile()6. 实际项目中的经验教训和最佳实践经过多个项目的实践总结出一些关键的经验教训这些经验可以帮助避免常见的陷阱。6.1 智能体设计的单一职责原则每个智能体应该只负责一个明确的职责。过度复杂的智能体会导致调试困难问题定位不明确复用性差难以在其他工作流中使用性能瓶颈单个智能体成为系统瓶颈正确的做法是宁可多设计几个简单的智能体也不要设计一个复杂的万能智能体。6.2 状态设计的版本兼容性State 结构一旦确定修改要非常谨慎。实际项目中建议为 State 添加版本字段重大变更时保持向后兼容提供状态迁移工具在文档中明确每个字段的语义和生命周期class VersionedState(TypedDict): schema_version: str # 如 1.0.0 data: dict metadata: dict6.3 测试策略从单元测试到集成测试多智能体系统的测试需要分层进行智能体单元测试测试单个智能体的输入输出工作流集成测试测试完整的工作流执行性能压力测试测试并发执行和资源使用故障恢复测试测试错误处理和状态恢复import pytest class TestMultiAgentSystem: def test_individual_agent(self): 测试单个智能体 test_state {topic: 测试主题, current_step: research} result research_agent_node(test_state) assert research_materials in result assert result[current_step] analysis def test_workflow_integration(self): 测试完整工作流 initial_state { topic: 集成测试主题, current_step: research, research_materials: [], completed_steps: [] } final_state graph.invoke(initial_state) assert complete in final_state[completed_steps] def test_error_recovery(self): 测试错误恢复机制 # 模拟智能体失败场景 # 验证系统能够正常恢复6.4 监控和可观测性生产环境中监控比功能更重要执行轨迹记录记录每个智能体的输入输出性能指标收集收集执行时间、资源使用等指标业务指标跟踪跟踪工作流成功率、完成时间等业务指标异常报警设置智能的异常检测和报警机制6.5 团队协作和文档维护多智能体系统通常需要团队协作开发为每个智能体编写清晰的接口文档使用版本控制管理工作流定义建立代码审查流程特别是状态结构的变更维护工作流的可视化文档LangGraph 的真正价值不在于让多智能体系统能够运行而在于让复杂协作变得可管理、可调试、可演进。从简单的线性工作流开始逐步引入条件分支、循环、错误处理等复杂逻辑最终构建出能够应对真实业务场景的智能系统。关键是要记住好的多智能体系统不是一次性设计出来的而是通过不断迭代优化出来的。从最小可行产品开始在实际使用中发现问题、调整设计、优化性能这才是构建可持续多智能体系统的正确路径。