ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

如何设计多Agent

如何设计多Agent 多 Agent 的设计核心不是“造很多个 Agent”而是把原本塞在一个循环里的职责拆成多个边界清晰、可独立测试、可独立授权的工作单元再用一个受控的编排层把它们串起来。如果单 Agent 状态机已经能稳定解决问题就不要引入多 Agent。多 Agent 的价值在于领域隔离、权限隔离、上下文隔离、并行执行、以及不同模型/成本的按需分配。下面从设计判断、架构模式、状态与通信、编排实现、生产级要点五个方面展开。一、先判断你真的需要多 Agent 吗多 Agent 会带来额外的通信开销、状态同步复杂度和调试难度。以下信号出现时才值得拆领域知识差异大一个 Agent 既要懂 SQL又要懂合同条款还要写营销文案Prompt 会互相干扰。权限差异大查询数据库和发起支付不应由同一个 Agent 持有凭证。上下文窗口压力大不同子任务的中间结果互相污染导致模型注意力分散。需要并行多个独立子任务可以同时执行缩短端到端延迟。需要不同模型档位简单分类用小模型复杂推理用大模型按 Agent 分配。如果只是工具多优先考虑单 Agent 工具分组 状态机路由而不是直接上多 Agent。二、常见多 Agent 架构模式Supervisor / Orchestrator-Workers最推荐用于生产一个中心化的 Supervisor Agent 负责理解任务、拆解、调度 Worker并汇总结果。Worker 只执行自己领域的子任务完成后把结果交回 Supervisor。用户 → Supervisor → Worker A → Supervisor → Worker B → Supervisor → 最终回答优点控制流集中易于审计、中断、重试、限制轮次。缺点Supervisor 可能成为瓶颈需要精心设计路由 Prompt。适用大多数企业级任务如客服、数据分析、审批流程。Hierarchical层级式Supervisor 下面再挂子 Supervisor每个子 Supervisor 管理一组 Worker。适合超大型任务比如“市场分析”下面分“数据采集”“竞品分析”“报告撰写”三个子团队。顶层 Supervisor├── 数据子 Supervisor → SQL Agent / 爬虫 Agent├── 分析子 Supervisor → 统计 Agent / 归因 Agent└── 撰写子 Supervisor → 文案 Agent / 审校 AgentNetwork / Peer-to-Peer去中心化Agent 之间通过“交接工具”Handoff直接转移控制权。例如 OpenAI Swarm 的模式Agent A 调用 transfer_to_agent_B控制权就交给 B。优点灵活适合对话式、边界模糊的场景。缺点全局控制弱容易死循环生产级慎用。适用轻量级、探索性、人工介入频繁的场景。Pipeline / Sequential流水线按固定顺序执行Agent A → Agent B → Agent C每个 Agent 的输出是下一个的输入。优点简单、可预测、易测试。缺点无法动态分支不适合复杂决策。适用文档处理、ETL、内容生成流水线。Blackboard黑板模式所有 Agent 共享一个全局状态黑板每个 Agent 监听自己关心的部分条件满足时触发执行。优点解耦适合事件驱动。缺点状态竞争、调试困难。适用实时协作、多模态感知融合。生产级首选 Supervisor 或 Hierarchical因为它们把控制流显式化便于加入检查点、中断、权限网关和可观测性。三、Agent 的契约设计每个 Agent 是一个“函数”无论哪种模式每个 Agent 都应定义清晰的输入输出契约而不是随意共享全部上下文。一个 Agent 的规格至少包括关键原则Agent 之间传递的是结构化任务包而不是完整的对话历史。例如{“task_id”: “t-123”,“goal”: “查询上季度华东区销售额”,“constraints”: [“仅使用只读权限”, “输出 JSON”],“input_data”: {“region”: “华东”, “quarter”: “2025Q1”},“expected_output”: {“schema”: {“sales”: “number”}}}Worker 返回结果时也只返回结构化结果和简短摘要由 Supervisor 决定是否注入全局状态。这样能避免上下文爆炸和交叉污染。四、状态与通信共享状态 vs 消息传递在 LangGraph 这类状态图框架中多 Agent 通常通过共享状态通信。状态是全局的但每个 Agent 只读写自己关心的字段。全局状态设计示例fromtypingimportTypedDict,AnnotatedimportoperatorclassMultiAgentState(TypedDict):# 全局消息流用于审计和最终回答messages:Annotated[list,operator.add]# 当前任务包task:dict# 各 Agent 产出的结构化结果artifacts:Annotated[dict,lambdaa,b:{**a,**b}]# 下一个要执行的 Agentnext_agent:str# 已执行的 Agent 历史用于循环检测agent_history:Annotated[list,operator.add]# 全局重试/轮次计数handoff_count:int私有上下文每个 Agent 内部可以有自己的私有消息列表不写入全局状态。例如 SQL Agent 调用模型多次生成 SQL、执行、修正这些中间步骤只留在 Agent 内部最终只把 {“sql”: “…”, “result”: […]} 写入 artifacts。通信方式对比方式 实现 优点 缺点共享状态 LangGraph StateGraph 简单、可持久化、可回溯 需设计 reducer避免竞争消息传递 消息队列 / 事件总线 解耦、可扩展 延迟高、调试难Handoff Agent 调用转移工具 灵活、对话自然 控制弱、易循环生产级推荐共享状态 结构化任务包必要时再引入消息队列做异步解耦。五、用 LangGraph 实现 Supervisor 多 Agent下面是一个可运行骨架展示 Supervisor 如何调度多个 Worker。fromtypingimportTypedDict,Annotated,Literalimportoperatorfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportMemorySaver1. 定义全局状态classState(TypedDict):messages:Annotated[list,operator.add]next_agent:strtask:dictartifacts:Annotated[dict,lambdaa,b:{**a,**b}]handoff_count:int2. 定义 Worker 节点每个 Worker 内部可以有自己的 tool call 循环defresearch_agent(state:State):# 内部调用 LLM 工具返回结构化结果result{research:华东区上季度销售额为 1.2 亿}return{artifacts:result,messages:[{role:assistant,content:研究完成}],next_agent:supervisor}defsql_agent(state:State):result{sql_result:[{region:华东,sales:120000000}]}return{artifacts:result,messages:[{role:assistant,content:查询完成}],next_agent:supervisor}defwriter_agent(state:State):result{final_report:根据数据华东区上季度销售额为 1.2 亿。}return{artifacts:result,messages:[{role:assistant,content:报告完成}],next_agent:supervisor}3. Supervisor 节点决定下一个 Agent 或结束defsupervisor(state:State):# 这里调用 LLM根据 task、artifacts 和 agent 描述决定 next_agent# 简化示例按顺序调度history[m[content]forminstate[messages]ifm[role]assistant]if研究完成notinhistory:next_agentresearch_agentelif查询完成notinhistory:next_agentsql_agentelif报告完成notinhistory:next_agentwriter_agentelse:next_agentFINISHreturn{next_agent:next_agent,handoff_count:state.get(handoff_count,0)1}4. 构建图builderStateGraph(State)builder.add_node(supervisor,supervisor)builder.add_node(research_agent,research_agent)builder.add_node(sql_agent,sql_agent)builder.add_node(writer_agent,writer_agent)builder.add_edge(START,supervisor)Supervisor 条件路由builder.add_conditional_edges(supervisor,lambdastate:state[next_agent],{research_agent:research_agent,sql_agent:sql_agent,writer_agent:writer_agent,FINISH:END})每个 Worker 完成后回到 Supervisorbuilder.add_edge(research_agent,supervisor)builder.add_edge(sql_agent,supervisor)builder.add_edge(writer_agent,supervisor)5. 编译注入检查点graph builder.compile(checkpointerMemorySaver())6. 执行config{configurable:{thread_id:task-123}}resultgraph.invoke({messages:[{role:user,content:分析华东区上季度销售并写报告}],task:{goal:销售分析报告},artifacts:{},handoff_count:0},configconfig)这个骨架的关键点Supervisor 是唯一控制流入口所有 Worker 完成后必须回到 Supervisor。Worker 不直接调用其他 Worker避免网状依赖。全局状态只保留结构化产物和消息摘要Worker 内部细节不外泄。通过 handoff_count 和 agent_history 可以检测循环、限制轮次。检查点保存器让整个多 Agent 流程可恢复、可中断。六、生产级多 Agent 的关键工程问题1. 循环与踢皮球多个 Agent 互相交接时容易出现“A 交给 BB 又交回 A”的死循环。必须设置max_handoffs全局最大交接次数。agent_history滑动窗口检测重复模式。Supervisor 路由 Prompt 中明确禁止无进展的重复交接。超过阈值后强制转人工或返回错误。2. 上下文隔离与共享的平衡共享任务目标、约束、全局产物索引、最终答案。隔离每个 Agent 的内部推理、工具原始返回、临时草稿。传递通过结构化任务包而不是把整个对话历史塞给下一个 Agent。3. 权限与安全每个 Worker 持有独立的最小权限凭证。Supervisor 只负责调度不应持有业务工具的写权限。高影响动作支付、删除、外发必须经过独立策略网关或人工确认。外部内容标记为不可信防止提示注入跨 Agent 传播。4. 错误处理与降级Worker 失败时返回结构化错误给 Supervisor而不是抛出异常中断全局。Supervisor 根据错误类型决定重试、换 Agent、降级、转人工。每个 Worker 有独立超时和重试上限避免拖垮全局。使用检查点失败后可从最后一个成功超步恢复。5. 可观测性为每个 Agent 创建独立的 trace span但共享同一个 thread_id。记录 Supervisor 的路由决策依据为什么选择这个 Agent。记录每次 Handoff 的输入任务包和输出产物。监控指标每个 Agent 的调用次数、成功率、Token 消耗、延迟、循环次数。6. 成本控制按 Agent 分配模型档位路由用轻量模型推理用大模型。限制每个 Agent 的内部工具循环次数。对 Worker 返回结果做结构化裁剪只把必要字段写入全局状态。监控“每成功任务成本”而不是只看单次调用成本。七、演进路径从单 Agent 到多 Agent不要一步到位。推荐路径单 Agent Tool Call 循环能跑通基本任务。单 Agent 状态图把循环显式化加入条件边、检查点、中断。Supervisor 2~3 个 Worker按领域拆分Supervisor 负责路由。Hierarchical 或并行 Worker任务复杂后引入层级或并行执行。事件驱动 消息队列高并发、长任务时解耦。每一步都要有评测集验证任务完成率、平均轮次、成本、延迟、人工介入率。只有当前架构在这些指标上遇到瓶颈时才引入下一层复杂度。一句话总结多 Agent 设计的本质是职责拆分 受控编排。用 Supervisor 或层级结构把控制流集中起来让每个 Agent 只做自己擅长的事通过结构化任务包通信共享必要的全局状态隔离私有上下文并配上检查点、权限网关、循环检测和可观测性。这样多 Agent 才是生产级的协作系统而不是一群互相踢皮球的聊天机器人。
返回列表