从单Agent问答到多Agent协同：任务分解与动态路由策略 从单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 PoolAgent池 │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ 问答Agent│ │ 数据Agent│ │ 计算Agent│ │ 报告Agent│ │ │ │ (RAG) │ │ (SQL) │ │ (Python) │ │ (生成) │ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ Aggregator聚合器 │ │ - 合并子任务结果 │ │ - 冲突解决与最终生成 │ └─────────────────────────────────────────────────────────┘四、关键实现任务分解任务分解模块是整个系统的入口。我们使用LLM配合结构化输出来生成执行计划。fromtypingimportList,Dict,AnyfrompydanticimportBaseModelimportjsonclassSubTask(BaseModel):task_id:strdescription:strtask_type:str# qa, data_query, calculation, generationdependencies:List[str]# 依赖的前置任务IDrequired_tools:List[str]expected_output:strclassExecutionPlan(BaseModel):tasks:List[SubTask]execution_order:List[str]# 拓扑排序后的执行顺序classTaskDecomposer:def__init__(self,llm_client):self.llmllm_clientdefdecompose(self,user_query:str,context:str)-ExecutionPlan:promptf 你是一个任务分解专家。请将以下用户问题拆解为可执行的子任务列表。 用户问题:{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] }} responseself.llm.chat(prompt)plan_dictjson.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.namename self.descriptiondescription self.task_typestask_types self.toolstools self.embeddingembeddingclassDynamicRouter:def__init__(self,agents:List[AgentProfile]):self.agentsagents self.encoderSentenceTransformer(all-MiniLM-L6-v2)# 预计算Agent的embeddingforagentinself.agents:ifagent.embeddingisNone:agent.embeddingself.encoder.encode(agent.description)defroute(self,subtask:SubTask)-AgentProfile:# 1. 规则匹配精确过滤candidatesself._rule_filter(subtask)ifnotcandidates:# fallback: 放宽规则candidatesself._rule_filter(subtask,strictFalse)# 2. 向量相似度排序task_embself.encoder.encode(subtask.description)scores[]foragentincandidates:simnp.dot(task_emb,agent.embedding)/(np.linalg.norm(task_emb)*np.linalg.norm(agent.embedding))scores.append((agent,sim))# 按相似度降序排列scores.sort(keylambdax:x[1],reverseTrue)# 如果最高分低于阈值触发fallback到通用Agentifscores[0][1]0.5:returnself._get_fallback_agent()returnscores[0][0]def_rule_filter(self,subtask:SubTask,strict:boolTrue)-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.decomposerdecomposer self.routerrouter self.executorsagent_executors# agent_name - executor functionself.aggregatoraggregatordefrun(self,query:str)-str:# Step 1: 任务分解planself.decomposer.decompose(query)# Step 2: 按执行顺序执行results{}fortask_idinplan.execution_order:# 找到对应的子任务tasknext(tfortinplan.tasksift.task_idtask_id)# 路由到合适的Agentagentself.router.route(task)# 执行注入前置任务结果context{dep:results[dep]fordepintask.dependencies}resultself.executors[agent.name](task,context)results[task_id]result# Step 3: 聚合结果final_answerself.aggregator.aggregate(query,plan,results)returnfinal_answer# ---------- 使用示例 ----------if__name____main__:# 模拟初始化fromunittest.mockimportMock mock_llmMock()mock_llm.chatlambdax:{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),]decomposerTaskDecomposer(mock_llm)routerDynamicRouter(agents)# 模拟执行器executors{data_agent:lambdatask,ctx:模拟数据结果,qa_agent:lambdatask,ctx:模拟问答结果,}aggregatorlambdaquery,plan,results:f最终答案基于:{results}systemMultiAgentSystem(decomposer,router,executors,aggregator)answersystem.run(对比A公司和B公司的营收)print(answer)七、进阶策略自适应路由与反馈闭环在实际生产中静态路由往往不够。我们加入两个增强机制7.1 路由置信度自适应classAdaptiveRouter(DynamicRouter):def__init__(self,agents,confidence_threshold0.6):super().__init__(agents)self.thresholdconfidence_threshold self.feedback_history{}# agent_name - [success_rate]defroute(self,subtask:SubTask)-AgentProfile:# 获取基础路由结果agentsuper().route(subtask)# 根据历史成功率调整ifagent.nameinself.feedback_history:success_ratenp.mean(self.feedback_history[agent.name])ifsuccess_rate0.5:# 如果该Agent历史表现差尝试第二选择second_bestself._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_workers4,**kwargs):super().__init__(*args,**kwargs)self.executorThreadPoolExecutor(max_workersmax_workers)defrun(self,query:str)-str:planself.decomposer.decompose(query)# 构建依赖图task_map{t.task_id:tfortinplan.tasks}completedset()results{}whilelen(completed)len(plan.tasks):# 找出所有依赖已满足的任务ready[tfortinplan.tasksift.task_idnotincompletedandall(dincompletedfordint.dependencies)]# 并行执行withThreadPoolExecutor(max_workerslen(ready))asexecutor:futures{}fortaskinready:agentself.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_idfutures[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解决对的问题”。任务分解是战略动态路由是战术两者配合才能让系统既聪明又高效。