多智能体协作框架把复杂任务拆给多个Agent分工执行,由编排层统一调度,任务效果和结果稳定性都优于单个模型包打全场。本文以LangGraph和AutoGen为例,说明多智能体协作框架的编排设计、通信机制与任务流落地方式,并给出可运行的代码示例。
多智能体协作框架的基本结构与拓扑选型
多智能体协作框架由三层组成:智能体层承载具体能力,比如检索、生成、代码执行、质量校验;通信层定义消息传递格式与路由;编排层决定执行顺序和分支,通常以状态机或图实现。任务先被拆成子任务,再分配给对应Agent,执行结果进入下一节点,直到达到终止条件。
常见拓扑有三种:
- 中心化编排:一个supervisor智能体拆解任务并分派给子智能体,适合子任务边界清晰、可独立验证的场景,排查问题也最直观。
- 流水线串联:前一个Agent的输出是后一个的输入,适合固定流程,比如检索→写作→审校。
- 环式协商:多个Agent轮流传话直至达成共识,适合评审和创意发散类任务,但token消耗和延迟明显偏高。
选型先看任务可拆分程度,不追求模型分数和拓扑复杂度。大多数生产场景用中心化或流水线就够用。
LangGraph状态机编排与任务流设计
LangGraph把Agent编排建模为状态图,节点是Agent能力或工具,节点间靠状态迁移。状态字段保存规划结果、中间答案和校验标记,调度器根据状态决定下一步,整个过程可回放、可中断、可恢复,适合做严谨的任务流。
下面是一个”规划→写作→审校”三节点状态机示例:
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
question: str
plan: list[str]
answer: str
checked: bool
def planner(state: AgentState) -> dict:
state["plan"] = ["检索素材", "起草回答", "核对事实"]
return {"plan": state["plan"]}
def writer(state: AgentState) -> dict:
state["answer"] = "基于检索结果生成的回答,包含来源编号。"
state["checked"] = False
return {"answer": state["answer"], "checked": state["checked"]}
def reviewer(state: AgentState) -> dict:
ok = "来源" in state["answer"] and len(state["answer"]) > 50
return {"checked": ok}
builder = StateGraph(AgentState)
builder.add_node("planner", planner)
builder.add_node("writer", writer)
builder.add_node("reviewer", reviewer)
builder.set_entry_point("planner")
builder.add_edge("planner", "writer")
builder.add_edge("writer", "reviewer")
builder.add_conditional_edges(
"reviewer",
lambda s: "writer" if not s["checked"] else END,
{"writer": "writer", END: END},
)
app = builder.compile()
审校节点返回的checked状态决定是否回写重新生成。这种”先规划、后执行、再校验”的循环,比一次性Prompt更能控制复杂任务的输出质量,也是多智能体编排被用于生产的原因。
AutoGen的群聊协作与组管理机制
AutoGen提供另一套范式:多个Agent以对话形式参与任务,由GroupChatManager管理者协调发言顺序与终止条件,适合过程需要可观测、结论需要多角色交叉确认的场景。注册工具时把函数名暴露给Agent,Agent会自行判断调用时机。
from autogen import ConversableAgent
planner = ConversableAgent(
name="planner",
system_message="你是任务规划者,把问题拆解为不超过3个子任务。",
llm_config={"config_list": [{"model": "gpt-4o", "api_key": "YOUR_KEY"}]},
)
coder = ConversableAgent(
name="coder",
system_message="你是Python开发者,输出可运行代码。",
llm_config={"config": [{"model": "gpt-4o", "api_key": "YOUR_KEY"}]},
)
result = planner.initiate_chat(coder, message="把一个接口压力测试拆解并实现。")
群聊模式下要显式控制发言顺序和终止条件,否则容易出现两个Agent反复确认的循环对话,拉高token成本。
多智能体协作的通信设计与上下文控制
通信设计上的三个关键点:消息格式统一、上下文裁剪、共享状态隔离。消息统一用结构化字段,避免自然语言散落;每次节点传递只带必要字段,超长历史先压缩再入上下文;工具产生的临时状态放在独立存储里,避免污染主对话。
多智能体协作的常见故障与排查方法
生产环境常见故障有四类:Agent循环死锁、上下文超限、工具调用参数错误、结果与事实不符。排查手段对应为:设置最大轮次上限并记录trace_id;对长文档做分段检索;对工具参数做schema校验;对事实型输出加引用溯源节点。日志里能看到每一步状态变迁,故障定位效率明显提升。
多智能体协作的适用边界与选型建议
不是所有任务都值得上多智能体。单轮问答、简单工具调用用单Agent就够;任务可拆分、需要多轮校验、工具种类多时,再引入编排层。成本方面按轮次估算:轮次数每增加一轮,token开销近似线性增长,上线前先拿小样本把平均轮次和成本测出来,再决定是否落地多智能体架构。
原创文章,作者:小编,如若转载,请注明出处:https://www.yunthe.com/duo-zhi-neng-ti-xie-zuo-kuang-jia-shi-zhan-bian-pai-she-ji/