这次我们来看一个关于多 Agent 编排模式的技术话题。当单一 AI Agent 的能力不足以应对复杂任务时将任务拆解由多个专业 Agent 协同完成已成为提升系统智能与可靠性的关键路径。但随之而来的核心问题是多个 Agent 之间谁说了算如何高效、有序地协作这篇文章将直接切入四种主流的多 Agent 编排模式顺序链、并行执行、分层监督与动态工作流。我们不空谈理论而是重点关注每种模式的实现逻辑、适用场景、技术门槛以及如何在实际项目中落地。无论你是想构建一个能自动处理多步骤任务的智能助手还是设计一个需要多个 AI 角色协作的复杂系统这篇文章都能为你提供清晰的架构选型思路和可操作的实践参考。1. 核心能力速览四种编排模式对比在深入细节之前我们先通过一个表格快速把握这四种模式的核心特征与差异这有助于你快速判断哪种模式更适合你的项目。编排模式核心思想控制权归属适用场景技术复杂度典型框架/工具参考顺序链 (Sequential Chain)任务按预定流水线执行前一个 Agent 的输出是后一个的输入。流程预设无中心决策者。文档处理流水线如摘要 - 翻译 - 格式化、固定的多步骤任务。低LangChain Expression Language, AutoGen (Sequential Chat)并行执行 (Parallel Execution)多个 Agent 同时处理同一任务的不同部分或不同任务。任务分发器或主控 Agent。同时调用多个工具/API、多路信息检索、方案对比生成。中AutoGen (GroupChat), CrewAI分层监督 (Hierarchical Supervisor)引入一个“主管”Agent负责任务分解、分配子任务给“员工”Agent并汇总结果。中心化的 Supervisor Agent。复杂项目规划如开发一个软件、需要动态任务拆解的场景。高LangGraph (StateGraph Supervisor), Microsoft Autogen (GroupChat Manager)动态工作流 (Dynamic Workflow)基于当前状态和规则动态决定下一个执行哪个 Agent支持循环、条件分支。工作流引擎或基于规则的 Router。复杂对话、诊断系统、交互式任务需多次往返确认。高LangGraph, Windmill, Temporal简单总结想跑通固定流程选顺序链简单直接。需要同时干多件事选并行执行提升效率。任务复杂需动态规划选分层监督让“主管”来指挥。流程充满判断和循环选动态工作流灵活应对变化。2. 适用场景与使用边界在决定采用哪种模式之前必须明确你的任务属性和技术边界。1. 顺序链适合谁场景任务步骤清晰、固定且前后依赖性强。例如“获取新闻 - 提取关键信息 - 生成简报 - 发送邮件”。这一步失败了下一步就没必要执行。边界缺乏灵活性。一旦中间某个环节因为意外输入而失败整个链条就会中断不具备自我修复或绕过的能力。2. 并行执行适合谁场景任务可被独立拆分或需要多源信息聚合。例如让三个不同的 Agent 同时去分析同一份市场报告的技术、财务和风险层面或者同时调用搜索引擎、数据库查询和内部知识库。边界需要设计良好的结果聚合逻辑。并行产生的多个结果可能需要去重、排序、投票或总结这本身可能又是一个挑战。同时资源消耗API调用成本、计算资源会成倍增加。3. 分层监督适合谁场景面对一个模糊的顶层目标如“开发一个贪吃蛇游戏”需要先分解成“设计游戏逻辑”、“编写前端界面”、“测试”等子任务再分配给不同的专家 Agent 执行。这模仿了人类项目经理的工作方式。边界对“主管”Agent 的能力要求极高。它需要具备强大的任务分解、规划、协调和结果评估能力。如果“主管”决策失误整个系统效率会很低。4. 动态工作流适合谁场景任务路径不确定需要根据中间结果动态调整。例如一个客服对话 Agent用户提问 - 分类 Agent 判断意图 - 如果是技术问题路由到技术支持 Agent如果是账单问题路由到财务 Agent如果问题不清晰则触发澄清 Agent 反问用户。边界设计和调试复杂。需要明确定义状态转移的条件规则或基于 LLM 的路由容易陷入循环或状态混乱。对系统设计者的抽象能力要求高。通用安全与合规边界 无论采用哪种模式当你的 Agent 系统涉及对外部工具/API的调用需确保有合法授权并处理调用失败、超时、费用超支等问题。处理用户数据必须遵守隐私政策避免在 Agent 间传递或日志中泄露敏感信息。生成内容需建立审核机制防止生成有害、偏见或侵权内容尤其是在多个 Agent 协作的“黑箱”中。关键决策不应完全依赖未经严格验证的多 Agent 系统做金融、医疗、法律等领域的最终决策应设有人工复核环节。3. 环境准备与前置条件要实践多 Agent 编排你需要一个基础的 AI 应用开发环境。以下是一个通用性较高的准备清单具体项目可能只需其中一部分。Python 环境这是大多数 Agent 框架的首选语言。建议使用 Python 3.9。使用conda或venv创建独立的虚拟环境是最佳实践。# 创建并激活虚拟环境 (以 conda 为例) conda create -n multi-agent python3.10 conda activate multi-agent大模型访问权限Agent 的核心是 LLM。你需要准备OpenAI API Key如果你使用 GPT 系列模型。或其他云端 LLM 服务的 Key如 Anthropic Claude, Google Gemini, 国内的通义千问、文心一言等。本地模型如果追求隐私和成本可部署 Llama、Qwen 等开源模型并通过ollama、vLLM或LM Studio提供 API 服务。这需要一定的 GPU 资源。关键 Python 包根据你选择的框架安装。# 基础包 pip install openai anthropic langchain langchain-community # 如果你探索 LangGraph (用于动态工作流/监督) pip install langgraph langchain-openai # 如果你探索 AutoGen pip install pyautogen # 如果你探索 CrewAI pip install crewai crewai-tools代码编辑器或 IDEVS Code, PyCharm 等用于编写和调试复杂的协作逻辑。网络与代理设置如需要确保你的开发环境能够稳定访问你所选 LLM 的 API 端点。4. 模式一顺序链的实现与验证顺序链是最直观的模式。我们以使用 LangChain 构建一个“新闻摘要 - 翻译 - 情感分析”流水线为例。测试目的验证多个 Agent 能否严格按照顺序执行并将数据流向下传递。操作步骤定义各个环节的 Agent每个 Agent 可以是一个简单的 LLMChain也可以是一个配备了工具的复杂 Agent。使用 LangChain Expression Language (LCEL)将它们连接起来。代码示例from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 1. 初始化模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 2. 定义各个“环节”的提示词 summary_prompt ChatPromptTemplate.from_template( 请用中文对以下新闻内容进行摘要保留核心事实\n{news} ) translation_prompt ChatPromptTemplate.from_template( 将以下中文摘要翻译成流畅的英文\n{summary} ) sentiment_prompt ChatPromptTemplate.from_template( 分析以下英文文本的情感倾向正面/负面/中性并简要说明原因\n{translated_text} ) # 3. 创建各个链可视为简单Agent summary_chain summary_prompt | llm | StrOutputParser() translation_chain translation_prompt | llm | StrOutputParser() sentiment_chain sentiment_prompt | llm | StrOutputParser() # 4. 构建顺序链 sequential_workflow ( {summary: summary_chain} # 第一步生成摘要 | RunnablePassthrough.assign(translated_texttranslation_chain) # 第二步翻译并将结果存入translated_text键 | RunnablePassthrough.assign(sentimentsentiment_chain) # 第三步情感分析结果存入sentiment键 ) # 5. 测试 input_news 北京时间今日凌晨某科技公司发布了新一代人工智能芯片宣称其性能提升十倍而功耗减半。市场分析师普遍看好此举认为将巩固其行业领先地位。 result sequential_workflow.invoke({news: input_news}) print(最终结果:, result)预期输出与判断result应该是一个字典包含summary中文摘要、translated_text英文翻译和sentiment情感分析三个键。成功标志是每个环节都产生了符合提示词要求的、连贯的输出。常见失败原因API 密钥未设置设置环境变量OPENAI_API_KEY。网络超时增加超时设置或在ChatOpenAI初始化时配置request_timeout。输出解析错误如果某个环节的输出格式不符合下一个环节的输入预期链会中断。需要检查提示词或引入输出解析器进行清洗。5. 模式二并行执行的实现与验证并行模式用于同时执行多个独立任务。我们模拟一个“多专家评审”场景。测试目的验证主控程序能同时发起多个任务并正确收集所有结果。操作步骤定义一个任务列表和对应的处理 Agent或链。使用asyncio或线程池并发执行。聚合所有结果。代码示例import asyncio from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 定义三个不同“专家”的提示词 tech_review_prompt ChatPromptTemplate.from_template( 你是一位技术专家。请从技术可行性、创新性角度评审以下项目创意{idea} ) business_review_prompt ChatPromptTemplate.from_template( 你是一位商业分析师。请从市场规模、盈利模式角度评审以下项目创意{idea} ) risk_review_prompt ChatPromptTemplate.from_template( 你是一位风险投资经理。请从潜在风险、投资回报角度评审以下项目创意{idea} ) # 创建三个链 tech_chain tech_review_prompt | llm | StrOutputParser() business_chain business_review_prompt | llm | StrOutputParser() risk_chain risk_review_prompt | llm | StrOutputParser() async def parallel_review(project_idea): # 创建异步任务 tasks [ tech_chain.ainvoke({idea: project_idea}), business_chain.ainvoke({idea: project_idea}), risk_chain.ainvoke({idea: project_idea}) ] # 并行执行 tech_review, business_review, risk_review await asyncio.gather(*tasks) return { 技术评审: tech_review, 商业评审: business_review, 风险评审: risk_review } # 运行测试 project_idea 开发一个基于AI的个性化健身教练APP通过手机摄像头实时纠正用户动作。 results asyncio.run(parallel_review(project_idea)) for role, review in results.items(): print(f {role} ) print(review[:200] ...) # 打印前200字符 print()预期输出与判断 程序应几乎同时打印出三段来自不同视角的评审意见。成功标志是三个任务独立完成总耗时接近于耗时最长的那个单个任务而不是三个任务耗时的总和。资源与性能观察API 成本与限流并行调用会瞬间消耗大量 Token需注意云端 API 的 RPM/TPM 限制可能需要实现限流或重试机制。异步编程使用asyncio可以高效处理 I/O 密集型任务如网络请求。如果 Agent 涉及大量本地计算可能需要使用concurrent.futures.ThreadPoolExecutor。6. 模式三分层监督的实现与验证以 LangGraph 为例分层监督模式引入了“主管”Supervisor。我们使用 LangGraph 来构建一个简单的“主管 员工”架构。测试目的验证一个主管 Agent 能根据目标动态创建任务列表并分配给不同的员工 Agent 执行最后汇总。操作步骤定义“员工”Agent 节点函数。定义“主管”Agent 节点负责规划和分配。使用 LangGraph 的StateGraph定义节点和边流程。让主管节点根据状态决定下一个要执行的员工节点。代码示例from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 1. 定义状态结构 class AgentState(TypedDict): goal: str # 总体目标 tasks: List[str] # 任务列表 current_task: str # 当前正在执行的任务 results: Annotated[List[str], operator.add] # 累积结果 next: str # 下一步该谁执行 # 2. 初始化模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 3. 定义“员工”节点函数 def writer_node(state: AgentState): 作家员工撰写内容 prompt ChatPromptTemplate.from_template( 你是一位专业作家。请根据以下主题撰写一段约150字的文章{task} ) chain prompt | llm | StrOutputParser() result chain.invoke({task: state[current_task]}) return {results: [f【作家完成】: {result}]} def researcher_node(state: AgentState): 研究员员工搜集资料 prompt ChatPromptTemplate.from_template( 你是一位研究员。请为以下主题列出3个最关键的事实或数据点{task} ) chain prompt | llm | StrOutputParser() result chain.invoke({task: state[current_task]}) return {results: [f【研究员完成】: {result}]} # 4. 定义“主管”节点函数 def supervisor_node(state: AgentState): 主管规划任务并分配 if not state.get(tasks): # 如果任务列表为空先创建任务 planning_prompt ChatPromptTemplate.from_template( 请将以下目标分解为2-3个具体的子任务 目标{goal} 请直接输出任务列表用‘- ’开头。 ) chain planning_prompt | llm | StrOutputParser() tasks_text chain.invoke({goal: state[goal]}) tasks [line.strip()[2:] for line in tasks_text.split(\n) if line.startswith(- )] next_task tasks[0] return {tasks: tasks, current_task: next_task, next: assigner} else: # 分配下一个任务 completed_tasks len(state[results]) all_tasks state[tasks] if completed_tasks len(all_tasks): next_task all_tasks[completed_tasks] return {current_task: next_task, next: assigner} else: return {next: END} # 所有任务完成结束 def assigner_node(state: AgentState): 分配器根据任务类型决定派给哪个员工 task state[current_task] # 简单的基于关键词的分配逻辑实际应用中可用更智能的Router if 撰写 in task or 文章 in task: return {next: writer} elif 研究 in task or 数据 in task: return {next: researcher} else: # 默认分配给作家 return {next: writer} # 5. 构建工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(supervisor, supervisor_node) workflow.add_node(assigner, assigner_node) workflow.add_node(writer, writer_node) workflow.add_node(researcher, researcher_node) # 设置边 workflow.set_entry_point(supervisor) workflow.add_edge(supervisor, assigner) workflow.add_conditional_edges( assigner, lambda x: x[next], # 根据 state 中的 next 字段决定去向 {writer: writer, researcher: researcher} ) workflow.add_edge(writer, supervisor) workflow.add_edge(researcher, supervisor) # 编译图 app workflow.compile() # 6. 测试运行 initial_state {goal: 制作一份关于‘可再生能源发展现状’的简短报告, results: []} final_state app.invoke(initial_state) print(最终目标:, initial_state[goal]) print(\n生成的任务列表:, final_state.get(tasks, [])) print(\n所有执行结果:) for r in final_state.get(results, []): print(r)预期输出与判断 程序应输出一个由主管分解出的任务列表如[‘撰写一篇关于可再生能源发展的引言’, ‘研究太阳能和风能的最新装机容量数据’]以及每个任务对应的员工执行结果。成功标志是主管能正确分解目标并将任务路由给合适的员工最后收集所有结果。核心观察点主管的决策能力本例使用了简单的关键词路由。在实际中主管可以是一个更强大的 LLM通过分析任务描述来决定派发给哪个专家 Agent。状态的维护AgentState是整个工作流共享的内存它记录了目标、任务列表、当前任务、结果和下一步动作是协调多个 Agent 的关键。7. 模式四动态工作流的实现与验证以条件路由为例动态工作流强调基于运行时的状态做出决策。我们在 LangGraph 的基础上实现一个更灵活的路由机制。测试目的验证工作流能根据中间结果如用户输入、Agent 输出选择不同的执行分支。操作步骤定义多个处理不同意图的 Agent 节点。定义一个路由节点Router根据输入内容决定下一个节点。构建一个支持循环如返回路由节点重新判断的图。代码示例模拟一个简单的客服路由。from typing import Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 定义状态 class RouterState(TypedDict): user_input: str classification: str # 分类结果如 tech, billing, general response: str # 定义分类器节点 def classifier_node(state: RouterState): prompt ChatPromptTemplate.from_template( 请将以下用户问题分类为 tech技术问题、billing账单问题或 general一般咨询 用户问题{input} 只输出分类标签不要输出其他任何文字。 ) chain prompt | llm | StrOutputParser() classification chain.invoke({input: state[user_input]}).strip().lower() return {classification: classification} # 定义各个处理节点 def tech_support_node(state: RouterState): prompt ChatPromptTemplate.from_template(你是一名技术客服。请专业地回答以下技术问题{input}) chain prompt | llm | StrOutputParser() response chain.invoke({input: state[user_input]}) return {response: f[技术客服] {response}, classification: done} def billing_support_node(state: RouterState): prompt ChatPromptTemplate.from_template(你是一名财务客服。请耐心地回答以下账单问题{input}) chain prompt | llm | StrOutputParser() response chain.invoke({input: state[user_input]}) return {response: f[财务客服] {response}, classification: done} def general_support_node(state: RouterState): prompt ChatPromptTemplate.from_template(你是一名通用客服。请友好地回答以下咨询{input}) chain prompt | llm | StrOutputParser() response chain.invoke({input: state[user_input]}) return {response: f[通用客服] {response}, classification: done} # 定义路由决策函数 def route_decision(state: RouterState) - Literal[tech, billing, general, __end__]: 根据分类结果决定下一个节点 classification state.get(classification) if classification tech: return tech elif classification billing: return billing elif classification general: return general else: # 如果分类不是三者之一结束流程或进入澄清节点 return __end__ # 构建图 workflow StateGraph(RouterState) workflow.add_node(classifier, classifier_node) workflow.add_node(tech, tech_support_node) workflow.add_node(billing, billing_support_node) workflow.add_node(general, general_support_node) workflow.set_entry_point(classifier) # 分类后根据决策路由 workflow.add_conditional_edges( classifier, route_decision, { tech: tech, billing: billing, general: general, __end__: END } ) # 处理节点直接结束 workflow.add_edge(tech, END) workflow.add_edge(billing, END) workflow.add_edge(general, END) app workflow.compile() # 测试不同输入 test_inputs [ 我的软件突然无法启动了错误代码是0x80070005。, 我上个月的账单金额好像不对能帮我查一下吗, 你们的营业时间是什么时候 ] for inp in test_inputs: print(f\n用户输入: {inp}) result app.invoke({user_input: inp}) print(f分类: {result.get(classification)}) print(f回复: {result.get(response)})预期输出与判断 对于不同的用户输入系统应正确分类tech, billing, general并路由到对应的客服节点给出相应风格的回复。成功标志是路由逻辑正确且整个流程是动态的由classifier_node的输出决定路径。动态性的体现条件分支add_conditional_edges是关键它允许根据状态值选择不同的下游节点。可扩展性可以轻松添加新的分类标签和处理节点如“complaint”-escalation_node。循环如果需要可以将节点重新连回classifier或一个新的clarify_node用于处理不明确的输入实现多轮交互。8. 接口 API 与批量任务集成当多 Agent 系统成熟后通常需要封装成 API 服务供其他系统调用或处理批量任务。1. 封装为 Web API使用 FastAPI 示例from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from typing import List import asyncio # 假设 sequential_workflow 是之前定义好的顺序链 # from your_agent_module import sequential_workflow, parallel_review, app (LangGraph应用) api_app FastAPI() class NewsInput(BaseModel): content: str class IdeaInput(BaseModel): idea: str class GoalInput(BaseModel): goal: str api_app.post(/process_news) async def process_news(input_data: NewsInput): 顺序链API result await sequential_workflow.ainvoke({news: input_data.content}) return {status: success, data: result} api_app.post(/review_idea) async def review_idea(input_data: IdeaInput): 并行评审API results await parallel_review(input_data.idea) return {status: success, reviews: results} api_app.post(/supervise_task) async def supervise_task(input_data: GoalInput, background_tasks: BackgroundTasks): 分层监督API可能耗时较长可放入后台 def run_supervision(goal: str): # 注意这里需要根据你的LangGraph app的调用方式调整 final_state app.invoke({goal: goal, results: []}) # 将结果存入数据库或发送通知 print(f任务完成: {goal}, 结果: {final_state[results]}) background_tasks.add_task(run_supervision, input_data.goal) return {status: accepted, message: 任务已提交后台处理} # 运行: uvicorn api_main:api_app --host 0.0.0.0 --port 80002. 批量任务处理 对于批量文件处理如一个文件夹里的多份文档核心是循环调用你的 Agent 工作流并做好任务管理和错误处理。import os import json from pathlib import Path import asyncio from your_agent_module import sequential_workflow # 导入你的工作流 async def batch_process_news(input_dir: Path, output_dir: Path): 批量处理新闻文件 output_dir.mkdir(parentsTrue, exist_okTrue) tasks [] for file_path in input_dir.glob(*.txt): with open(file_path, r, encodingutf-8) as f: content f.read() # 为每个文件创建异步任务 task sequential_workflow.ainvoke({news: content}) task_with_meta (file_path.stem, task) # 保留文件名用于输出 tasks.append(task_with_meta) # 并发执行限制并发数避免过量请求 semaphore asyncio.Semaphore(5) # 最大5个并发 async def process_with_semaphore(name, task): async with semaphore: try: result await task output_file output_dir / f{name}_result.json with open(output_file, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) print(f处理成功: {name}) except Exception as e: print(f处理失败 {name}: {e}) # 记录失败日志 with open(output_dir / error.log, a) as f: f.write(f{name}: {e}\n) await asyncio.gather(*[process_with_semaphore(n, t) for n, t in tasks]) # 调用示例 # asyncio.run(batch_process_news(Path(./news_input), Path(./news_output)))关键点并发控制使用Semaphore限制同时发起的 API 调用数量避免触发限流。错误处理与重试必须捕获单个任务失败避免整个批量作业中断。可以实现简单的重试逻辑。结果持久化每个任务的结果应及时保存如写入文件或数据库避免内存溢出。任务队列对于超大规模批量任务应考虑引入专业的任务队列如 Celery, Dramatiq。9. 资源占用、性能观察与优化多 Agent 系统的性能瓶颈通常不在本地计算而在网络 I/OLLM API 调用和复杂工作流的状态管理。1. 主要资源消耗点LLM API 调用这是最大的耗时和成本来源。Token 消耗与 Agent 间的对话轮次、每次交互的上下文长度成正比。工作流引擎开销LangGraph 等框架本身会引入一些状态管理开销但对于大多数应用来说可忽略不计。内存维护复杂的 State 对象和中间结果会占用内存但在处理单个请求时通常不是问题。批量处理时需注意。2. 性能观察方法记录与监控在每个 Agent 节点或链的调用前后记录时间戳。import time class TimedAgent: def __init__(self, chain, name): self.chain chain self.name name async def arun(self, input): start time.time() result await self.chain.ainvoke(input) end time.time() print(f[{self.name}] 耗时: {end - start:.2f}秒) return resultToken 计数使用 LangChain 的 Callback 或 LLM 提供商的后台监控统计每次交互的 Token 使用量。3. 优化策略减少不必要的交互设计高效的工作流避免 Agent 间来回传递无关信息。让每个 Agent 的职责尽可能单一、明确。压缩上下文在将历史对话或中间结果传递给下一个 Agent 前考虑进行摘要Summarization。缓存对于相同或相似的查询可以考虑缓存 LLM 的响应结果。LangChain 提供了LLMCache组件。设置超时与重试为每个 API 调用设置合理的超时时间并实现重试机制注意使用指数退避。选择合适模型对于不需要顶级创造力的路由、分类等任务使用更小、更快的模型如gpt-3.5-turbo可以大幅降低成本并提升速度。10. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 执行因错误终止(Agent execution terminated due to error)1. LLM API 调用失败网络、鉴权、限流。2. 某个节点的代码抛出未处理的异常。3. 状态State格式不符合下一个节点的预期。1. 查看完整错误堆栈。2. 检查 API 密钥、网络连接。3. 在关键节点添加try...except并打印状态。1. 实现全局错误处理或将失败任务路由到“补救”节点。2. 在 LangGraph 中可以使用try...except包装节点函数并在出错时返回一个特定的错误状态由工作流决定下一步如重试或终止。工作流陷入无限循环1. 条件路由逻辑有误导致状态在几个节点间来回跳转。2. 没有设置明确的终止条件。1. 打印每次循环后的状态检查next字段的变化。2. 在图中设置最大循环次数。1. 仔细检查add_conditional_edges的条件函数。2. 在 State 中增加iteration_count字段并在主管节点或路由节点中检查超过阈值则导向END。显存/内存溢出本地模型1. 同时运行多个本地大模型实例。2. 批量任务数据未及时释放。1. 使用nvidia-smi或psutil监控资源。2. 检查代码中是否有全局变量累积大量数据。1. 限制并发数。2. 使用流式处理处理完一个任务后及时清理中间变量。3. 考虑使用 GPU 内存更小的量化模型。API 调用速率超限并行任务过多触发云服务商的 RPM/TPM 限制。查看 API 返回的错误信息通常为429状态码。1. 在批量/并行处理中引入asyncio.Semaphore或令牌桶算法进行限流。2. 实现带退避机制的自动重试。最终输出质量差1. 单个 Agent 的提示词设计不佳。2. Agent 间传递的信息有损失或歧义。3. 主管的决策任务分解、分配不合理。1. 单独测试每个 Agent 的功能。2. 打印并检查每个环节的输入和输出。1. 迭代优化每个 Agent 的提示词Prompt Engineering。2. 在 State 中传递更结构化、更明确的信息而非纯自然语言。3. 为主管 Agent 提供更详细的上下文和示例Few-shot。11. 最佳实践与使用建议从简单开始逐步复杂不要一开始就设计包含 10 个 Agent 的复杂系统。先用顺序链实现核心流程再逐步引入并行、路由和监督。为每个 Agent 明确职责给每个 Agent 一个清晰、单一的角色如“翻译专家”、“数据分析师”、“安全检查员”并通过提示词强化其角色。设计健壮的状态结构使用 TypedDict 明确定义 State 的格式。这是多 Agent 间通信的“合同”避免混乱。实现全面的日志记录记录每个 Agent 的输入、输出、耗时和 Token 使用。这对调试、优化和成本核算至关重要。为关键决策设置人工审核点在涉及重要业务结果如内容发布、金额计算的环节设计流程将结果提交给人审核而不是完全自动化。进行彻底的集成测试模拟各种正常和异常输入测试你的多 Agent 系统确保其不会在边缘情况下崩溃或产生有害输出。成本监控与优化将 Token 消耗和 API 调用次数纳入监控定期审查工作流看是否有环节可以简化或合并。多 Agent 编排不是银弹它引入了更高的复杂性和协调成本。但对于需要组合多种能力、应对不确定路径的复杂任务它提供了强大的抽象能力和灵活性。从理解这四种基础模式开始选择最适合你当前场景的入手在实践中不断迭代你就能构建出真正智能、可靠的 AI 应用系统。建议将本文中的代码示例作为实验起点结合你的具体需求进行修改和扩展。 多Agent编排模式详解：顺序链、并行执行、分层监督与动态工作流
这次我们来看一个关于多 Agent 编排模式的技术话题。当单一 AI Agent 的能力不足以应对复杂任务时将任务拆解由多个专业 Agent 协同完成已成为提升系统智能与可靠性的关键路径。但随之而来的核心问题是多个 Agent 之间谁说了算如何高效、有序地协作这篇文章将直接切入四种主流的多 Agent 编排模式顺序链、并行执行、分层监督与动态工作流。我们不空谈理论而是重点关注每种模式的实现逻辑、适用场景、技术门槛以及如何在实际项目中落地。无论你是想构建一个能自动处理多步骤任务的智能助手还是设计一个需要多个 AI 角色协作的复杂系统这篇文章都能为你提供清晰的架构选型思路和可操作的实践参考。1. 核心能力速览四种编排模式对比在深入细节之前我们先通过一个表格快速把握这四种模式的核心特征与差异这有助于你快速判断哪种模式更适合你的项目。编排模式核心思想控制权归属适用场景技术复杂度典型框架/工具参考顺序链 (Sequential Chain)任务按预定流水线执行前一个 Agent 的输出是后一个的输入。流程预设无中心决策者。文档处理流水线如摘要 - 翻译 - 格式化、固定的多步骤任务。低LangChain Expression Language, AutoGen (Sequential Chat)并行执行 (Parallel Execution)多个 Agent 同时处理同一任务的不同部分或不同任务。任务分发器或主控 Agent。同时调用多个工具/API、多路信息检索、方案对比生成。中AutoGen (GroupChat), CrewAI分层监督 (Hierarchical Supervisor)引入一个“主管”Agent负责任务分解、分配子任务给“员工”Agent并汇总结果。中心化的 Supervisor Agent。复杂项目规划如开发一个软件、需要动态任务拆解的场景。高LangGraph (StateGraph Supervisor), Microsoft Autogen (GroupChat Manager)动态工作流 (Dynamic Workflow)基于当前状态和规则动态决定下一个执行哪个 Agent支持循环、条件分支。工作流引擎或基于规则的 Router。复杂对话、诊断系统、交互式任务需多次往返确认。高LangGraph, Windmill, Temporal简单总结想跑通固定流程选顺序链简单直接。需要同时干多件事选并行执行提升效率。任务复杂需动态规划选分层监督让“主管”来指挥。流程充满判断和循环选动态工作流灵活应对变化。2. 适用场景与使用边界在决定采用哪种模式之前必须明确你的任务属性和技术边界。1. 顺序链适合谁场景任务步骤清晰、固定且前后依赖性强。例如“获取新闻 - 提取关键信息 - 生成简报 - 发送邮件”。这一步失败了下一步就没必要执行。边界缺乏灵活性。一旦中间某个环节因为意外输入而失败整个链条就会中断不具备自我修复或绕过的能力。2. 并行执行适合谁场景任务可被独立拆分或需要多源信息聚合。例如让三个不同的 Agent 同时去分析同一份市场报告的技术、财务和风险层面或者同时调用搜索引擎、数据库查询和内部知识库。边界需要设计良好的结果聚合逻辑。并行产生的多个结果可能需要去重、排序、投票或总结这本身可能又是一个挑战。同时资源消耗API调用成本、计算资源会成倍增加。3. 分层监督适合谁场景面对一个模糊的顶层目标如“开发一个贪吃蛇游戏”需要先分解成“设计游戏逻辑”、“编写前端界面”、“测试”等子任务再分配给不同的专家 Agent 执行。这模仿了人类项目经理的工作方式。边界对“主管”Agent 的能力要求极高。它需要具备强大的任务分解、规划、协调和结果评估能力。如果“主管”决策失误整个系统效率会很低。4. 动态工作流适合谁场景任务路径不确定需要根据中间结果动态调整。例如一个客服对话 Agent用户提问 - 分类 Agent 判断意图 - 如果是技术问题路由到技术支持 Agent如果是账单问题路由到财务 Agent如果问题不清晰则触发澄清 Agent 反问用户。边界设计和调试复杂。需要明确定义状态转移的条件规则或基于 LLM 的路由容易陷入循环或状态混乱。对系统设计者的抽象能力要求高。通用安全与合规边界 无论采用哪种模式当你的 Agent 系统涉及对外部工具/API的调用需确保有合法授权并处理调用失败、超时、费用超支等问题。处理用户数据必须遵守隐私政策避免在 Agent 间传递或日志中泄露敏感信息。生成内容需建立审核机制防止生成有害、偏见或侵权内容尤其是在多个 Agent 协作的“黑箱”中。关键决策不应完全依赖未经严格验证的多 Agent 系统做金融、医疗、法律等领域的最终决策应设有人工复核环节。3. 环境准备与前置条件要实践多 Agent 编排你需要一个基础的 AI 应用开发环境。以下是一个通用性较高的准备清单具体项目可能只需其中一部分。Python 环境这是大多数 Agent 框架的首选语言。建议使用 Python 3.9。使用conda或venv创建独立的虚拟环境是最佳实践。# 创建并激活虚拟环境 (以 conda 为例) conda create -n multi-agent python3.10 conda activate multi-agent大模型访问权限Agent 的核心是 LLM。你需要准备OpenAI API Key如果你使用 GPT 系列模型。或其他云端 LLM 服务的 Key如 Anthropic Claude, Google Gemini, 国内的通义千问、文心一言等。本地模型如果追求隐私和成本可部署 Llama、Qwen 等开源模型并通过ollama、vLLM或LM Studio提供 API 服务。这需要一定的 GPU 资源。关键 Python 包根据你选择的框架安装。# 基础包 pip install openai anthropic langchain langchain-community # 如果你探索 LangGraph (用于动态工作流/监督) pip install langgraph langchain-openai # 如果你探索 AutoGen pip install pyautogen # 如果你探索 CrewAI pip install crewai crewai-tools代码编辑器或 IDEVS Code, PyCharm 等用于编写和调试复杂的协作逻辑。网络与代理设置如需要确保你的开发环境能够稳定访问你所选 LLM 的 API 端点。4. 模式一顺序链的实现与验证顺序链是最直观的模式。我们以使用 LangChain 构建一个“新闻摘要 - 翻译 - 情感分析”流水线为例。测试目的验证多个 Agent 能否严格按照顺序执行并将数据流向下传递。操作步骤定义各个环节的 Agent每个 Agent 可以是一个简单的 LLMChain也可以是一个配备了工具的复杂 Agent。使用 LangChain Expression Language (LCEL)将它们连接起来。代码示例from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 1. 初始化模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 2. 定义各个“环节”的提示词 summary_prompt ChatPromptTemplate.from_template( 请用中文对以下新闻内容进行摘要保留核心事实\n{news} ) translation_prompt ChatPromptTemplate.from_template( 将以下中文摘要翻译成流畅的英文\n{summary} ) sentiment_prompt ChatPromptTemplate.from_template( 分析以下英文文本的情感倾向正面/负面/中性并简要说明原因\n{translated_text} ) # 3. 创建各个链可视为简单Agent summary_chain summary_prompt | llm | StrOutputParser() translation_chain translation_prompt | llm | StrOutputParser() sentiment_chain sentiment_prompt | llm | StrOutputParser() # 4. 构建顺序链 sequential_workflow ( {summary: summary_chain} # 第一步生成摘要 | RunnablePassthrough.assign(translated_texttranslation_chain) # 第二步翻译并将结果存入translated_text键 | RunnablePassthrough.assign(sentimentsentiment_chain) # 第三步情感分析结果存入sentiment键 ) # 5. 测试 input_news 北京时间今日凌晨某科技公司发布了新一代人工智能芯片宣称其性能提升十倍而功耗减半。市场分析师普遍看好此举认为将巩固其行业领先地位。 result sequential_workflow.invoke({news: input_news}) print(最终结果:, result)预期输出与判断result应该是一个字典包含summary中文摘要、translated_text英文翻译和sentiment情感分析三个键。成功标志是每个环节都产生了符合提示词要求的、连贯的输出。常见失败原因API 密钥未设置设置环境变量OPENAI_API_KEY。网络超时增加超时设置或在ChatOpenAI初始化时配置request_timeout。输出解析错误如果某个环节的输出格式不符合下一个环节的输入预期链会中断。需要检查提示词或引入输出解析器进行清洗。5. 模式二并行执行的实现与验证并行模式用于同时执行多个独立任务。我们模拟一个“多专家评审”场景。测试目的验证主控程序能同时发起多个任务并正确收集所有结果。操作步骤定义一个任务列表和对应的处理 Agent或链。使用asyncio或线程池并发执行。聚合所有结果。代码示例import asyncio from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 定义三个不同“专家”的提示词 tech_review_prompt ChatPromptTemplate.from_template( 你是一位技术专家。请从技术可行性、创新性角度评审以下项目创意{idea} ) business_review_prompt ChatPromptTemplate.from_template( 你是一位商业分析师。请从市场规模、盈利模式角度评审以下项目创意{idea} ) risk_review_prompt ChatPromptTemplate.from_template( 你是一位风险投资经理。请从潜在风险、投资回报角度评审以下项目创意{idea} ) # 创建三个链 tech_chain tech_review_prompt | llm | StrOutputParser() business_chain business_review_prompt | llm | StrOutputParser() risk_chain risk_review_prompt | llm | StrOutputParser() async def parallel_review(project_idea): # 创建异步任务 tasks [ tech_chain.ainvoke({idea: project_idea}), business_chain.ainvoke({idea: project_idea}), risk_chain.ainvoke({idea: project_idea}) ] # 并行执行 tech_review, business_review, risk_review await asyncio.gather(*tasks) return { 技术评审: tech_review, 商业评审: business_review, 风险评审: risk_review } # 运行测试 project_idea 开发一个基于AI的个性化健身教练APP通过手机摄像头实时纠正用户动作。 results asyncio.run(parallel_review(project_idea)) for role, review in results.items(): print(f {role} ) print(review[:200] ...) # 打印前200字符 print()预期输出与判断 程序应几乎同时打印出三段来自不同视角的评审意见。成功标志是三个任务独立完成总耗时接近于耗时最长的那个单个任务而不是三个任务耗时的总和。资源与性能观察API 成本与限流并行调用会瞬间消耗大量 Token需注意云端 API 的 RPM/TPM 限制可能需要实现限流或重试机制。异步编程使用asyncio可以高效处理 I/O 密集型任务如网络请求。如果 Agent 涉及大量本地计算可能需要使用concurrent.futures.ThreadPoolExecutor。6. 模式三分层监督的实现与验证以 LangGraph 为例分层监督模式引入了“主管”Supervisor。我们使用 LangGraph 来构建一个简单的“主管 员工”架构。测试目的验证一个主管 Agent 能根据目标动态创建任务列表并分配给不同的员工 Agent 执行最后汇总。操作步骤定义“员工”Agent 节点函数。定义“主管”Agent 节点负责规划和分配。使用 LangGraph 的StateGraph定义节点和边流程。让主管节点根据状态决定下一个要执行的员工节点。代码示例from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 1. 定义状态结构 class AgentState(TypedDict): goal: str # 总体目标 tasks: List[str] # 任务列表 current_task: str # 当前正在执行的任务 results: Annotated[List[str], operator.add] # 累积结果 next: str # 下一步该谁执行 # 2. 初始化模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 3. 定义“员工”节点函数 def writer_node(state: AgentState): 作家员工撰写内容 prompt ChatPromptTemplate.from_template( 你是一位专业作家。请根据以下主题撰写一段约150字的文章{task} ) chain prompt | llm | StrOutputParser() result chain.invoke({task: state[current_task]}) return {results: [f【作家完成】: {result}]} def researcher_node(state: AgentState): 研究员员工搜集资料 prompt ChatPromptTemplate.from_template( 你是一位研究员。请为以下主题列出3个最关键的事实或数据点{task} ) chain prompt | llm | StrOutputParser() result chain.invoke({task: state[current_task]}) return {results: [f【研究员完成】: {result}]} # 4. 定义“主管”节点函数 def supervisor_node(state: AgentState): 主管规划任务并分配 if not state.get(tasks): # 如果任务列表为空先创建任务 planning_prompt ChatPromptTemplate.from_template( 请将以下目标分解为2-3个具体的子任务 目标{goal} 请直接输出任务列表用‘- ’开头。 ) chain planning_prompt | llm | StrOutputParser() tasks_text chain.invoke({goal: state[goal]}) tasks [line.strip()[2:] for line in tasks_text.split(\n) if line.startswith(- )] next_task tasks[0] return {tasks: tasks, current_task: next_task, next: assigner} else: # 分配下一个任务 completed_tasks len(state[results]) all_tasks state[tasks] if completed_tasks len(all_tasks): next_task all_tasks[completed_tasks] return {current_task: next_task, next: assigner} else: return {next: END} # 所有任务完成结束 def assigner_node(state: AgentState): 分配器根据任务类型决定派给哪个员工 task state[current_task] # 简单的基于关键词的分配逻辑实际应用中可用更智能的Router if 撰写 in task or 文章 in task: return {next: writer} elif 研究 in task or 数据 in task: return {next: researcher} else: # 默认分配给作家 return {next: writer} # 5. 构建工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(supervisor, supervisor_node) workflow.add_node(assigner, assigner_node) workflow.add_node(writer, writer_node) workflow.add_node(researcher, researcher_node) # 设置边 workflow.set_entry_point(supervisor) workflow.add_edge(supervisor, assigner) workflow.add_conditional_edges( assigner, lambda x: x[next], # 根据 state 中的 next 字段决定去向 {writer: writer, researcher: researcher} ) workflow.add_edge(writer, supervisor) workflow.add_edge(researcher, supervisor) # 编译图 app workflow.compile() # 6. 测试运行 initial_state {goal: 制作一份关于‘可再生能源发展现状’的简短报告, results: []} final_state app.invoke(initial_state) print(最终目标:, initial_state[goal]) print(\n生成的任务列表:, final_state.get(tasks, [])) print(\n所有执行结果:) for r in final_state.get(results, []): print(r)预期输出与判断 程序应输出一个由主管分解出的任务列表如[‘撰写一篇关于可再生能源发展的引言’, ‘研究太阳能和风能的最新装机容量数据’]以及每个任务对应的员工执行结果。成功标志是主管能正确分解目标并将任务路由给合适的员工最后收集所有结果。核心观察点主管的决策能力本例使用了简单的关键词路由。在实际中主管可以是一个更强大的 LLM通过分析任务描述来决定派发给哪个专家 Agent。状态的维护AgentState是整个工作流共享的内存它记录了目标、任务列表、当前任务、结果和下一步动作是协调多个 Agent 的关键。7. 模式四动态工作流的实现与验证以条件路由为例动态工作流强调基于运行时的状态做出决策。我们在 LangGraph 的基础上实现一个更灵活的路由机制。测试目的验证工作流能根据中间结果如用户输入、Agent 输出选择不同的执行分支。操作步骤定义多个处理不同意图的 Agent 节点。定义一个路由节点Router根据输入内容决定下一个节点。构建一个支持循环如返回路由节点重新判断的图。代码示例模拟一个简单的客服路由。from typing import Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 定义状态 class RouterState(TypedDict): user_input: str classification: str # 分类结果如 tech, billing, general response: str # 定义分类器节点 def classifier_node(state: RouterState): prompt ChatPromptTemplate.from_template( 请将以下用户问题分类为 tech技术问题、billing账单问题或 general一般咨询 用户问题{input} 只输出分类标签不要输出其他任何文字。 ) chain prompt | llm | StrOutputParser() classification chain.invoke({input: state[user_input]}).strip().lower() return {classification: classification} # 定义各个处理节点 def tech_support_node(state: RouterState): prompt ChatPromptTemplate.from_template(你是一名技术客服。请专业地回答以下技术问题{input}) chain prompt | llm | StrOutputParser() response chain.invoke({input: state[user_input]}) return {response: f[技术客服] {response}, classification: done} def billing_support_node(state: RouterState): prompt ChatPromptTemplate.from_template(你是一名财务客服。请耐心地回答以下账单问题{input}) chain prompt | llm | StrOutputParser() response chain.invoke({input: state[user_input]}) return {response: f[财务客服] {response}, classification: done} def general_support_node(state: RouterState): prompt ChatPromptTemplate.from_template(你是一名通用客服。请友好地回答以下咨询{input}) chain prompt | llm | StrOutputParser() response chain.invoke({input: state[user_input]}) return {response: f[通用客服] {response}, classification: done} # 定义路由决策函数 def route_decision(state: RouterState) - Literal[tech, billing, general, __end__]: 根据分类结果决定下一个节点 classification state.get(classification) if classification tech: return tech elif classification billing: return billing elif classification general: return general else: # 如果分类不是三者之一结束流程或进入澄清节点 return __end__ # 构建图 workflow StateGraph(RouterState) workflow.add_node(classifier, classifier_node) workflow.add_node(tech, tech_support_node) workflow.add_node(billing, billing_support_node) workflow.add_node(general, general_support_node) workflow.set_entry_point(classifier) # 分类后根据决策路由 workflow.add_conditional_edges( classifier, route_decision, { tech: tech, billing: billing, general: general, __end__: END } ) # 处理节点直接结束 workflow.add_edge(tech, END) workflow.add_edge(billing, END) workflow.add_edge(general, END) app workflow.compile() # 测试不同输入 test_inputs [ 我的软件突然无法启动了错误代码是0x80070005。, 我上个月的账单金额好像不对能帮我查一下吗, 你们的营业时间是什么时候 ] for inp in test_inputs: print(f\n用户输入: {inp}) result app.invoke({user_input: inp}) print(f分类: {result.get(classification)}) print(f回复: {result.get(response)})预期输出与判断 对于不同的用户输入系统应正确分类tech, billing, general并路由到对应的客服节点给出相应风格的回复。成功标志是路由逻辑正确且整个流程是动态的由classifier_node的输出决定路径。动态性的体现条件分支add_conditional_edges是关键它允许根据状态值选择不同的下游节点。可扩展性可以轻松添加新的分类标签和处理节点如“complaint”-escalation_node。循环如果需要可以将节点重新连回classifier或一个新的clarify_node用于处理不明确的输入实现多轮交互。8. 接口 API 与批量任务集成当多 Agent 系统成熟后通常需要封装成 API 服务供其他系统调用或处理批量任务。1. 封装为 Web API使用 FastAPI 示例from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from typing import List import asyncio # 假设 sequential_workflow 是之前定义好的顺序链 # from your_agent_module import sequential_workflow, parallel_review, app (LangGraph应用) api_app FastAPI() class NewsInput(BaseModel): content: str class IdeaInput(BaseModel): idea: str class GoalInput(BaseModel): goal: str api_app.post(/process_news) async def process_news(input_data: NewsInput): 顺序链API result await sequential_workflow.ainvoke({news: input_data.content}) return {status: success, data: result} api_app.post(/review_idea) async def review_idea(input_data: IdeaInput): 并行评审API results await parallel_review(input_data.idea) return {status: success, reviews: results} api_app.post(/supervise_task) async def supervise_task(input_data: GoalInput, background_tasks: BackgroundTasks): 分层监督API可能耗时较长可放入后台 def run_supervision(goal: str): # 注意这里需要根据你的LangGraph app的调用方式调整 final_state app.invoke({goal: goal, results: []}) # 将结果存入数据库或发送通知 print(f任务完成: {goal}, 结果: {final_state[results]}) background_tasks.add_task(run_supervision, input_data.goal) return {status: accepted, message: 任务已提交后台处理} # 运行: uvicorn api_main:api_app --host 0.0.0.0 --port 80002. 批量任务处理 对于批量文件处理如一个文件夹里的多份文档核心是循环调用你的 Agent 工作流并做好任务管理和错误处理。import os import json from pathlib import Path import asyncio from your_agent_module import sequential_workflow # 导入你的工作流 async def batch_process_news(input_dir: Path, output_dir: Path): 批量处理新闻文件 output_dir.mkdir(parentsTrue, exist_okTrue) tasks [] for file_path in input_dir.glob(*.txt): with open(file_path, r, encodingutf-8) as f: content f.read() # 为每个文件创建异步任务 task sequential_workflow.ainvoke({news: content}) task_with_meta (file_path.stem, task) # 保留文件名用于输出 tasks.append(task_with_meta) # 并发执行限制并发数避免过量请求 semaphore asyncio.Semaphore(5) # 最大5个并发 async def process_with_semaphore(name, task): async with semaphore: try: result await task output_file output_dir / f{name}_result.json with open(output_file, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) print(f处理成功: {name}) except Exception as e: print(f处理失败 {name}: {e}) # 记录失败日志 with open(output_dir / error.log, a) as f: f.write(f{name}: {e}\n) await asyncio.gather(*[process_with_semaphore(n, t) for n, t in tasks]) # 调用示例 # asyncio.run(batch_process_news(Path(./news_input), Path(./news_output)))关键点并发控制使用Semaphore限制同时发起的 API 调用数量避免触发限流。错误处理与重试必须捕获单个任务失败避免整个批量作业中断。可以实现简单的重试逻辑。结果持久化每个任务的结果应及时保存如写入文件或数据库避免内存溢出。任务队列对于超大规模批量任务应考虑引入专业的任务队列如 Celery, Dramatiq。9. 资源占用、性能观察与优化多 Agent 系统的性能瓶颈通常不在本地计算而在网络 I/OLLM API 调用和复杂工作流的状态管理。1. 主要资源消耗点LLM API 调用这是最大的耗时和成本来源。Token 消耗与 Agent 间的对话轮次、每次交互的上下文长度成正比。工作流引擎开销LangGraph 等框架本身会引入一些状态管理开销但对于大多数应用来说可忽略不计。内存维护复杂的 State 对象和中间结果会占用内存但在处理单个请求时通常不是问题。批量处理时需注意。2. 性能观察方法记录与监控在每个 Agent 节点或链的调用前后记录时间戳。import time class TimedAgent: def __init__(self, chain, name): self.chain chain self.name name async def arun(self, input): start time.time() result await self.chain.ainvoke(input) end time.time() print(f[{self.name}] 耗时: {end - start:.2f}秒) return resultToken 计数使用 LangChain 的 Callback 或 LLM 提供商的后台监控统计每次交互的 Token 使用量。3. 优化策略减少不必要的交互设计高效的工作流避免 Agent 间来回传递无关信息。让每个 Agent 的职责尽可能单一、明确。压缩上下文在将历史对话或中间结果传递给下一个 Agent 前考虑进行摘要Summarization。缓存对于相同或相似的查询可以考虑缓存 LLM 的响应结果。LangChain 提供了LLMCache组件。设置超时与重试为每个 API 调用设置合理的超时时间并实现重试机制注意使用指数退避。选择合适模型对于不需要顶级创造力的路由、分类等任务使用更小、更快的模型如gpt-3.5-turbo可以大幅降低成本并提升速度。10. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 执行因错误终止(Agent execution terminated due to error)1. LLM API 调用失败网络、鉴权、限流。2. 某个节点的代码抛出未处理的异常。3. 状态State格式不符合下一个节点的预期。1. 查看完整错误堆栈。2. 检查 API 密钥、网络连接。3. 在关键节点添加try...except并打印状态。1. 实现全局错误处理或将失败任务路由到“补救”节点。2. 在 LangGraph 中可以使用try...except包装节点函数并在出错时返回一个特定的错误状态由工作流决定下一步如重试或终止。工作流陷入无限循环1. 条件路由逻辑有误导致状态在几个节点间来回跳转。2. 没有设置明确的终止条件。1. 打印每次循环后的状态检查next字段的变化。2. 在图中设置最大循环次数。1. 仔细检查add_conditional_edges的条件函数。2. 在 State 中增加iteration_count字段并在主管节点或路由节点中检查超过阈值则导向END。显存/内存溢出本地模型1. 同时运行多个本地大模型实例。2. 批量任务数据未及时释放。1. 使用nvidia-smi或psutil监控资源。2. 检查代码中是否有全局变量累积大量数据。1. 限制并发数。2. 使用流式处理处理完一个任务后及时清理中间变量。3. 考虑使用 GPU 内存更小的量化模型。API 调用速率超限并行任务过多触发云服务商的 RPM/TPM 限制。查看 API 返回的错误信息通常为429状态码。1. 在批量/并行处理中引入asyncio.Semaphore或令牌桶算法进行限流。2. 实现带退避机制的自动重试。最终输出质量差1. 单个 Agent 的提示词设计不佳。2. Agent 间传递的信息有损失或歧义。3. 主管的决策任务分解、分配不合理。1. 单独测试每个 Agent 的功能。2. 打印并检查每个环节的输入和输出。1. 迭代优化每个 Agent 的提示词Prompt Engineering。2. 在 State 中传递更结构化、更明确的信息而非纯自然语言。3. 为主管 Agent 提供更详细的上下文和示例Few-shot。11. 最佳实践与使用建议从简单开始逐步复杂不要一开始就设计包含 10 个 Agent 的复杂系统。先用顺序链实现核心流程再逐步引入并行、路由和监督。为每个 Agent 明确职责给每个 Agent 一个清晰、单一的角色如“翻译专家”、“数据分析师”、“安全检查员”并通过提示词强化其角色。设计健壮的状态结构使用 TypedDict 明确定义 State 的格式。这是多 Agent 间通信的“合同”避免混乱。实现全面的日志记录记录每个 Agent 的输入、输出、耗时和 Token 使用。这对调试、优化和成本核算至关重要。为关键决策设置人工审核点在涉及重要业务结果如内容发布、金额计算的环节设计流程将结果提交给人审核而不是完全自动化。进行彻底的集成测试模拟各种正常和异常输入测试你的多 Agent 系统确保其不会在边缘情况下崩溃或产生有害输出。成本监控与优化将 Token 消耗和 API 调用次数纳入监控定期审查工作流看是否有环节可以简化或合并。多 Agent 编排不是银弹它引入了更高的复杂性和协调成本。但对于需要组合多种能力、应对不确定路径的复杂任务它提供了强大的抽象能力和灵活性。从理解这四种基础模式开始选择最适合你当前场景的入手在实践中不断迭代你就能构建出真正智能、可靠的 AI 应用系统。建议将本文中的代码示例作为实验起点结合你的具体需求进行修改和扩展。