从单Agent问答到多Agent协同：任务分解与动态路由策略
当一个问题需要多种能力协同解决时，单Agent的“全能”幻想往往败给现实。本文从实践出发，拆解多Agent协同的核心命题——任务分解与动态路由，并附可运行的代码示例。
一、单Agent的边界
过去两年，基于大语言模型的单Agent问答系统发展迅速。给定一个用户问题，检索相关上下文，让LLM生成答案——这套RAG流程已经成为标配。
但随着业务场景变复杂，单Agent的局限开始暴露：
- 知识冲突：一个模型很难同时精通法律、医疗、编程等多个垂直领域
- 上下文过载：把多个子任务塞进一次推理，既浪费token，也降低准确率
- 工具调用混乱：当Agent同时拥有计算器、数据库查询、外部API等多种工具时，模型经常选错或漏调
举个例子，用户问：“对比A公司和B公司过去三年的营收增长率，并结合行业平均值给出投资建议。”
这个问题天然包含多个子任务：
- 查询A公司财报数据
- 查询B公司财报数据
- 查询行业平均增长率
- 计算复合增长率
- 综合对比并生成投资建议
单Agent的做法是一次性把所有信息塞给LLM。但财报数据可能来自不同的内部数据库，行业数据需要外部API，计算增长率又需要精确的数值运算。把这些全部交给一个模型决策，出错概率极高。
二、多Agent协同的核心命题
多Agent系统不是简单地“多开几个模型实例”。它的设计围绕两个核心命题展开：
命题一：任务分解（Task Decomposition）
如何将复杂问题拆解为可独立执行的子任务？
好的分解需要满足：
- 独立性：子任务之间耦合度低，可并行或顺序执行
- 可分配性：每个子任务能匹配到擅长该领域的Agent
- 可组合性：子任务的结果能合并为最终答案
命题二：动态路由（Dynamic Routing）
如何将每个子任务调度到最合适的Agent？
路由策略需要考虑：
- Agent能力画像：每个Agent擅长什么领域、支持什么工具
- 任务特征匹配：当前子任务属于什么类型（问答、计算、检索、生成等）
- 负载与成本：避免某些Agent过载，控制API调用成本
这两个命题相互依赖——好的分解能降低路由难度，好的路由能弥补分解的不足。
三、系统架构设计
我们设计一个轻量级的多Agent协同框架，核心组件如下：
┌─────────────────────────────────────────────────────────┐ │ 用户输入 │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ Planner（规划器） │ │ - 任务分解（LLM + Few-shot） │ │ - 生成DAG格式的执行计划 │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ Router（路由器） │ │ - 基于向量相似度 + 规则匹配 选择Agent │ │ - 支持fallback机制 │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ Agent Pool（Agent池） │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ 问答Agent│ │ 数据Agent│ │ 计算Agent│ │ 报告Agent│ │ │ │ (RAG) │ │ (SQL) │ │ (Python) │ │ (生成) │ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ Aggregator（聚合器） │ │ - 合并子任务结果 │ │ - 冲突解决与最终生成 │ └─────────────────────────────────────────────────────────┘四、关键实现：任务分解
任务分解模块是整个系统的入口。我们使用LLM配合结构化输出来生成执行计划。
fromtypingimportList,Dict,AnyfrompydanticimportBaseModelimportjsonclassSubTask(BaseModel):task_id:strdescription:strtask_type:str# "qa", "data_query", "calculation", "generation"dependencies:List[str]# 依赖的前置任务IDrequired_tools:List[str]expected_output:strclassExecutionPlan(BaseModel):tasks:List[SubTask]execution_order:List[str]# 拓扑排序后的执行顺序classTaskDecomposer:def__init__(self,llm_client):self.llm=llm_clientdefdecompose(self,user_query:str,context:str="")->ExecutionPlan:prompt=f""" 你是一个任务分解专家。请将以下用户问题拆解为可执行的子任务列表。 用户问题:{user_query}上下文:{context}要求: 1. 每个子任务必须独立、可分配 2. 明确标注任务类型: qa(问答), data_query(数据查询), calculation(计算), generation(生成) 3. 标注依赖关系 4. 输出为JSON格式 输出格式: {{ "tasks": [ {{ "task_id": "task_1", "description": "查询A公司2023-2025年营收", "task_type": "data_query", "dependencies": [], "required_tools": ["sql_db"], "expected_output": "营收数据列表" }} ], "execution_order": ["task_1", "task_2"] }} """response=self.llm.chat(prompt)plan_dict=json.loads(response)returnExecutionPlan(**plan_dict)五、关键实现：动态路由
路由器根据子任务的特征和Agent的能力进行匹配。我们采用混合路由策略：向量检索做初筛，规则引擎做精排。
fromsentence_transformersimportSentenceTransformerimportnumpyasnpfromtypingimportOptionalclassAgentProfile:def__init__(self,name:str,description:str,task_types:List[str],tools:List[str],embedding:np.ndarray):self.name=name self.description=description self.task_types=task_types self.tools=tools self.embedding=embeddingclassDynamicRouter:def__init__(self,agents:List[AgentProfile]):self.agents=agents self.encoder=SentenceTransformer('all-MiniLM-L6-v2')# 预计算Agent的embeddingforagentinself.agents:ifagent.embeddingisNone:agent.embedding=self.encoder.encode(agent.description)defroute(self,subtask:SubTask)->AgentProfile:# 1. 规则匹配（精确过滤）candidates=self._rule_filter(subtask)ifnotcandidates:# fallback: 放宽规则candidates=self._rule_filter(subtask,strict=False)# 2. 向量相似度排序task_emb=self.encoder.encode(subtask.description)scores=[]foragentincandidates:sim=np.dot(task_emb,agent.embedding)/(np.linalg.norm(task_emb)*np.linalg.norm(agent.embedding))scores.append((agent,sim))# 按相似度降序排列scores.sort(key=lambdax:x[1],reverse=True)# 如果最高分低于阈值，触发fallback到通用Agentifscores[0][1]<0.5:returnself._get_fallback_agent()returnscores[0][0]def_rule_filter(self,subtask:SubTask,strict:bool=True)->List[AgentProfile]:candidates=[]foragentinself.agents:# 任务类型必须匹配ifsubtask.task_typenotinagent.task_types:continue# 工具必须包含（严格模式）ifstrictandnotall(tinagent.toolsfortinsubtask.required_tools):continuecandidates.append(agent)returncandidatesdef_get_fallback_agent(self)->AgentProfile:# 返回一个通用的LLM问答Agentreturnself.agents[0]六、完整执行流程
将上述模块组合成一个完整的协同系统：
classMultiAgentSystem:def__init__(self,decomposer:TaskDecomposer,router:DynamicRouter,agent_executors:Dict[str,callable],aggregator):self.decomposer=decomposer self.router=router self.executors=agent_executors# agent_name -> executor functionself.aggregator=aggregatordefrun(self,query:str)->str:# Step 1: 任务分解plan=self.decomposer.decompose(query)# Step 2: 按执行顺序执行results={}fortask_idinplan.execution_order:# 找到对应的子任务task=next(tfortinplan.tasksift.task_id==task_id)# 路由到合适的Agentagent=self.router.route(task)# 执行（注入前置任务结果）context={dep:results[dep]fordepintask.dependencies}result=self.executors[agent.name](task,context)results[task_id]=result# Step 3: 聚合结果final_answer=self.aggregator.aggregate(query,plan,results)returnfinal_answer# ---------- 使用示例 ----------if__name__=="__main__":# 模拟初始化fromunittest.mockimportMock mock_llm=Mock()mock_llm.chat=lambdax:'{"tasks": [{"task_id": "t1", "description": "查询数据", "task_type": "data_query", "dependencies": [], "required_tools": ["sql"], "expected_output": "data"}], "execution_order": ["t1"]}'# 创建Agent画像agents=[AgentProfile("data_agent","擅长SQL数据查询",["data_query"],["sql"],None),AgentProfile("qa_agent","擅长通用问答和文本理解",["qa","generation"],["llm"],None),]decomposer=TaskDecomposer(mock_llm)router=DynamicRouter(agents)# 模拟执行器executors={"data_agent":lambdatask,ctx:"模拟数据结果","qa_agent":lambdatask,ctx:"模拟问答结果",}aggregator=lambdaquery,plan,results:f"最终答案基于:{results}"system=MultiAgentSystem(decomposer,router,executors,aggregator)answer=system.run("对比A公司和B公司的营收")print(answer)七、进阶策略：自适应路由与反馈闭环
在实际生产中，静态路由往往不够。我们加入两个增强机制：
7.1 路由置信度自适应
classAdaptiveRouter(DynamicRouter):def__init__(self,agents,confidence_threshold=0.6):super().__init__(agents)self.threshold=confidence_threshold self.feedback_history={}# agent_name -> [success_rate]defroute(self,subtask:SubTask)->AgentProfile:# 获取基础路由结果agent=super().route(subtask)# 根据历史成功率调整ifagent.nameinself.feedback_history:success_rate=np.mean(self.feedback_history[agent.name])ifsuccess_rate<0.5:# 如果该Agent历史表现差，尝试第二选择second_best=self._get_second_choice(subtask)ifsecond_best:returnsecond_bestreturnagentdefrecord_feedback(self,agent_name:str,success:bool):ifagent_namenotinself.feedback_history:self.feedback_history[agent_name]=[]self.feedback_history[agent_name].append(1.0ifsuccesselse0.0)# 只保留最近100次iflen(self.feedback_history[agent_name])>100:self.feedback_history[agent_name].pop(0)7.2 并行执行优化
对于无依赖关系的子任务，可以并行执行以降低延迟：
importasynciofromconcurrent.futuresimportThreadPoolExecutorclassParallelMultiAgentSystem(MultiAgentSystem):def__init__(self,*args,max_workers=4,**kwargs):super().__init__(*args,**kwargs)self.executor=ThreadPoolExecutor(max_workers=max_workers)defrun(self,query:str)->str:plan=self.decomposer.decompose(query)# 构建依赖图task_map={t.task_id:tfortinplan.tasks}completed=set()results={}whilelen(completed)<len(plan.tasks):# 找出所有依赖已满足的任务ready=[tfortinplan.tasksift.task_idnotincompletedandall(dincompletedfordint.dependencies)]# 并行执行withThreadPoolExecutor(max_workers=len(ready))asexecutor:futures={}fortaskinready:agent=self.router.route(task)ctx={dep:results[dep]fordepintask.dependencies}futures[executor.submit(self.executors[agent.name],task,ctx)]=task.task_idforfutureinasyncio.as_completed(futures):task_id=futures[future]results[task_id]=future.result()completed.add(task_id)returnself.aggregator.aggregate(query,plan,results)八、总结与最佳实践
多Agent协同系统不是银弹，但在复杂任务场景下，它提供了比单Agent更可控、更可扩展的架构。以下是我们从实践中总结的几点经验：
分解粒度要适中：子任务太小会导致路由开销过大，太大会失去多Agent的优势。一般以“一个子任务调用1-2个工具”为参考。
路由需要可观测：在生产环境中记录每次路由决策和结果，便于排查问题和优化策略。
Agent能力要正交：避免多个Agent能力重叠，否则路由会变得混乱。每个Agent应该有明确的职责边界。
始终保留fallback：无论路由策略多完善，都要有一个兜底的通用Agent，确保系统不会完全失效。
成本与性能的权衡：多Agent意味着多次LLM调用。对于简单问题，可以直接走单Agent快速通道，避免过度设计。
多Agent协同的终极目标不是“用更多Agent解决问题”，而是“用对的Agent解决对的问题”。任务分解是战略，动态路由是战术，两者配合才能让系统既聪明又高效。