1. 从单体到协同为什么我们需要多 Agent 路由如果你最近在折腾 AI Agent 开发尤其是基于 Claude Code 这类框架大概率会遇到一个瓶颈单个 Agent 的能力边界。一开始我们可能只构建一个“万能”Agent让它处理所有事情——从解析用户意图、调用工具、到生成最终答案。这在小规模、需求简单的场景下没问题。但随着任务复杂度提升这个“全能选手”会变得臃肿、低效且容易出错。比如一个需要先进行数据分析、再生成图表、最后撰写报告的复杂任务让一个 Agent 从头干到尾不仅响应慢而且每个环节的专业度都可能打折扣。这时多 Agent 架构就成了自然而然的选择。我们可以设计一个“数据分析专家”Agent、一个“图表生成”Agent 和一个“报告撰写”Agent。但随之而来的核心问题就是谁来协调它们用户的一个请求过来应该先交给谁一个 Agent 处理完一部分工作后结果又该传递给谁这就是“路由”机制要解决的核心问题。它本质上是一个决策中心负责理解任务并将其分发给最合适的“专家”Agent 去执行。Claude Code 作为一个新兴的、备受关注的 Agent 开发框架其源码中关于多 Agent 路由与定义的设计为我们提供了一个非常值得研究的范本。它没有采用过于复杂、沉重的编排引擎而是通过一套相对清晰、基于 TypeScript 的约定和机制实现了轻量而有效的多 Agent 协作。理解这套机制不仅能帮你用好 Claude Code更能让你深刻理解现代 AI Agent 系统中“分工与协作”的设计哲学。2. 解剖 Claude Code 的 Agent 定义从“个体”到“角色”在讨论路由之前我们必须先搞清楚在 Claude Code 中一个 Agent 是如何被定义的。这不仅仅是给它起个名字那么简单而是为其赋予清晰的“角色”、“能力”和“边界”。2.1 核心定义接口AgentDefinition在 Claude Code 的源码结构中通常位于src/agent/或类似目录下你会发现一个或多个用于定义 Agent 的核心接口或类。我们暂且将其称为AgentDefinition。一个典型的定义可能包含以下关键属性interface AgentDefinition { id: string; // Agent 的唯一标识如 data_analyzer, chart_generator name: string; // Agent 的友好名称用于日志和界面显示 description: string; // **至关重要**描述该 Agent 的职责和能力范围 entryPoint: string; // 该 Agent 主逻辑处理函数的路径或标识 capabilities: string[]; // 该 Agent 所具备的能力标签如 [data_processing, python] inputSchema?: JSONSchema; // 期望的输入数据格式 outputSchema?: JSONSchema; // 承诺的输出数据格式 }这里最需要关注的是description和capabilities。在后续的路由决策中路由器Router很大程度上就是通过比对用户查询或任务描述与各个 Agent 的description和capabilities的语义相似度来决定将任务派发给谁。因此写好 Agent 的描述是有效路由的第一步。一个模糊的描述如“处理数据”远不如“专门使用 Python pandas 和 numpy 进行数据清洗、聚合与统计分析”来得有效。2.2 技能Skill与工具Tool的装配一个 Agent 的能力最终体现在它能调用哪些工具Tools或技能Skills。在 Claude Code 中这通常通过依赖注入或配置绑定的方式实现。// 示例定义一个图表生成 Agent class ChartGeneratorAgent implements Agent { private chartTool: ChartGenerationTool; private dataValidator: DataValidationTool; constructor(tools: ToolRegistry) { this.chartTool tools.get(advanced_chart); this.dataValidator tools.get(validate_dataset); } async execute(task: TaskContext): PromiseAgentResult { // 1. 使用 dataValidator 检查输入数据 const isValid await this.dataValidator.validate(task.input.data); if (!isValid) { /* 处理错误 */ } // 2. 使用 chartTool 生成图表 const chartUrl await this.chartTool.generate({ data: task.input.data, type: task.input.chartType || line }); // 3. 返回结果 return { success: true, output: { chartUrl }, nextAgent: report_writer }; } }关键点在于Agent 的定义与其技能是解耦的。技能库ToolRegistry是一个中心化的管理单元。这种设计的好处是你可以像搭积木一样为不同的 Agent 组合不同的技能而无需修改 Agent 的核心逻辑。这也为路由决策提供了另一个维度路由器不仅看 Agent 的“自我介绍”description还可以查看它实际装配了哪些“武器”tools从而做出更精准的判断。2.3 状态与上下文隔离在多 Agent 系统中确保每个 Agent 在处理任务时拥有独立、干净的上下文至关重要避免数据污染和副作用。Claude Code 通常通过TaskContext或Session对象来实现这一点。每个任务请求都会生成一个唯一的上下文其中包含了taskId: 任务唯一标识。input: 原始输入或上一个 Agent 的输出。metadata: 任务元数据如用户偏好、语言等。conversationHistory: 当前会话的历史记录可选。privateState: 供当前执行 Agent 使用的临时状态存储。当一个 Agent 执行完毕其privateState通常会被清理或序列化只有明确放置在output中并传递给下一个 Agent 的数据才会在流水线中持续存在。这种设计保证了 Agent 的“无状态性”和可替换性是构建稳健路由系统的基础。3. 路由机制详解任务如何找到“对的人”定义了多个 Agent 之后路由机制便是系统的中枢神经。Claude Code 的路由设计在我看来是一种“基于策略的匹配路由”它可能包含以下几个层次。3.1 静态路由与显式指派这是最简单直接的方式。你可以在任务发起时或在上一个 Agent 的返回结果中显式指定下一个要执行的 Agent 的 ID。// 在任务创建时指定 const task { input: “分析这份销售数据并生成趋势图”, initialAgentId: ‘data_analyzer’ // 明确告诉系统从谁开始 }; // 在 Agent 执行完毕后指定后继者 class DataAnalyzerAgent { async execute(task: TaskContext): PromiseAgentResult { // ... 分析逻辑 ... return { success: true, output: analysisReport, nextAgent: ‘chart_generator’ // 分析完交给画图的 }; } }这种方式完全由开发者控制流程清晰但缺乏灵活性。任何流程的变更都需要修改代码。3.2 动态路由基于语义匹配的决策这是更智能、也是 Claude Code 路由系统的核心亮点。一个典型的动态路由器AgentRouter的工作流程如下接收任务获取用户查询或上游 Agent 的输出。候选 Agent 检索从已注册的AgentRegistry中获取所有可用的 Agent 定义。匹配度计算这是路由算法的核心。通常采用以下一种或多种策略语义相似度匹配使用文本嵌入模型如 OpenAI 的text-embedding-ada-002或本地 SentenceTransformer将用户查询和每个 Agent 的description转换为向量然后计算余弦相似度。得分最高者胜出。关键词/能力标签匹配从查询中提取关键词与 Agent 的capabilities数组进行匹配。匹配数量越多权重越高。输入/输出模式匹配检查任务数据是否符合 Agent 定义的inputSchema。这更像是一种“能力过滤”确保 Agent 能处理给定的数据格式。决策与分发根据匹配得分选择最佳 Agent。有时还会设置阈值低于阈值则触发“未找到合适 Agent”的回退处理如交给一个默认的通用 Agent或直接向用户澄清。上下文传递将任务上下文可能经过适当格式化分发给选中的 Agent。class SemanticAgentRouter implements AgentRouter { constructor(private agentRegistry: AgentRegistry, private embeddingService: EmbeddingService) {} async route(taskContext: TaskContext): PromiseAgentDefinition { const query taskContext.input.query || JSON.stringify(taskContext.input); const queryEmbedding await this.embeddingService.embed(query); const candidates this.agentRegistry.getAll(); let bestMatch: AgentDefinition | null null; let highestScore -1; for (const agent of candidates) { // 计算语义相似度 const agentEmbedding await this.getCachedEmbedding(agent.description); const similarity this.cosineSimilarity(queryEmbedding, agentEmbedding); // 可选叠加能力标签匹配分 const keywordScore this.calculateKeywordOverlap(query, agent.capabilities); const totalScore similarity * 0.7 keywordScore * 0.3; // 加权总分 if (totalScore highestScore) { highestScore totalScore; bestMatch agent; } } if (bestMatch highestScore ROUTING_THRESHOLD) { return bestMatch; } // 回退到默认 Agent return this.agentRegistry.get(‘fallback_generalist’); } }3.3 路由链与工作流编排单一的路由决策往往不够。复杂任务需要多个 Agent 按特定顺序协作形成一个路由链或工作流。Claude Code 可能通过以下方式支持隐式链式路由如前所述每个 Agent 在执行完毕后可以在返回结果中指定nextAgent。路由器会尊重这个指示将上下文传递给下一个指定的 Agent形成一条链。显式工作流定义提供一种 DSL领域特定语言或配置方式预先定义好一个复杂任务的执行流程图。workflow: “数据分析报告流程” steps: - agent: “data_analyzer” condition: “input.type ‘dataset’” - agent: “chart_generator” dependsOn: [“data_analyzer”] - agent: “report_writer” dependsOn: [“data_analyzer”, “chart_generator”]路由器或一个独立的 Workflow Orchestrator会解析这个定义并按依赖关系顺序调用 Agent管理数据流向。这里有一个非常重要的实践经验在动态路由和预定义工作流之间存在一个“控制权”的权衡。动态路由灵活但不可预测预定义工作流可控但僵化。一个优秀的系统往往会结合两者。例如用一个“主协调”Agent 来动态解析超复杂任务将其分解为子任务然后这些子任务再通过预定义的小型工作流或动态路由去执行。这类似于人类项目经理的工作方式。4. 深入源码TypeScript 实现中的精妙设计要真正理解 Claude Code 的路由与定义机制免不了要窥探其 TypeScript 源码。我们可以从几个关键的设计模式入手。4.1 依赖注入与控制反转Claude Code 的架构很可能大量运用了依赖注入DI来管理 Agent、工具和路由器之间的复杂依赖关系。这带来了极佳的模块化和可测试性。// 一个简化的容器配置示例 class AppContainer { register() { // 注册工具 this.bindChartGenerationTool(‘chart_tool’).to(AdvancedChartTool); this.bindDataValidationTool(‘validation_tool’).to(DataValidator); // 注册 Agent this.bindAgent(‘data_analyzer’).to(DataAnalyzerAgent); this.bindAgent(‘chart_generator’).to(ChartGeneratorAgent).inject(‘chart_tool’, ‘validation_tool’); // 注册路由器 this.bindAgentRouter(‘router’).to(SemanticAgentRouter).inject(‘agent_registry’, ‘embedding_service’); this.bindAgentRegistry(‘agent_registry’).to(SimpleAgentRegistry); } }通过 DI 容器ChartGeneratorAgent不需要知道ChartGenerationTool的具体实现类只需要声明它需要这个依赖。容器负责在运行时将正确的实例注入进去。这使得替换路由算法比如从基于关键词的换成基于语义的或升级某个工具变得非常容易只需修改容器配置即可无需触动业务代码。4.2 事件驱动与状态管理在多 Agent 异步执行时如何跟踪任务状态、处理错误和实现超时控制事件总线Event Bus是一个常见的选择。// Agent 在执行关键动作时发出事件 class DataAnalyzerAgent { async execute(task: TaskContext): PromiseAgentResult { this.eventBus.emit(‘agent.started’, { agentId: this.id, taskId: task.taskId }); try { // ... 业务逻辑 ... this.eventBus.emit(‘agent.completed’, { agentId: this.id, taskId: task.taskId, output: result }); return result; } catch (error) { this.eventBus.emit(‘agent.failed’, { agentId: this.id, taskId: task.taskId, error }); throw error; } } } // 一个独立的监控服务监听这些事件 class TaskMonitorService { constructor(eventBus: EventBus) { eventBus.on(‘agent.failed’, (event) { console.error(Agent ${event.agentId} failed on task ${event.taskId}:, event.error); // 可能触发重试、通知用户或切换到备用 Agent }); eventBus.on(‘agent.completed’, (event) { // 更新任务状态并可能触发路由器寻找下一个 Agent this.taskManager.updateTask(event.taskId, { currentAgent: event.agentId, output: event.output }); this.router.routeNext(event.taskId); }); } }这种松耦合的设计让核心的业务逻辑Agent 执行与系统的协调、监控逻辑分离大大提升了系统的可扩展性和可维护性。4.3 配置化与约定优于配置从网络热词如“约定式路由”可以看出Claude Code 可能借鉴了前端框架如 Nuxt.js, Next.js的理念采用“约定优于配置”的原则来简化 Agent 和路由的定义。例如可能约定在src/agents/目录下的每个.ts文件默认导出一个类它就会被自动注册为一个 Agent。文件名即 Agent ID如DataAnalyzer.ts对应data_analyzer。在src/agents/index.ts或一个配置文件中可以通过简单的装饰器或元数据来定义路由规则。// src/agents/ChartGenerator.ts Agent({ name: ‘Chart Generator’, description: ‘Generates various charts (line, bar, pie) from structured data.’, capabilities: [‘visualization’, ‘chart’] }) export default class ChartGeneratorAgent implements Agent { // ... 实现 ... } // 路由规则可能在一个中心化的配置文件里 // config/routes.ts export const agentRoutes [ { pattern: ‘.*(chart|graph|plot).*’, agentId: ‘chart_generator’ }, { pattern: ‘.*analyze data.*’, agentId: ‘data_analyzer’ }, // 更复杂的规则可以是一个函数 { match: (input) input.dataType ‘csv’, agentId: ‘data_analyzer’ } ];这种方式极大地减少了样板代码让开发者能更专注于 Agent 的核心能力实现。5. 实战构建一个简单的多 Agent 路由系统理论说得再多不如动手实践。让我们抛开 Claude Code 的具体实现用 TypeScript 从头构思一个极简但核心的多 Agent 路由系统。这将帮助你彻底理解其中的关节。5.1 第一步定义 Agent 注册表这是系统的基石负责管理所有 Agent 的定义。// types.ts export interface AgentDefinition { id: string; name: string; description: string; capabilities: string[]; factory: () PromiseAgent; // 用于创建 Agent 实例的工厂函数 } export interface Agent { execute(context: TaskContext): PromiseAgentResult; } export interface AgentResult { success: boolean; output: any; nextAgentId?: string; // 用于显式指定后继 } export interface TaskContext { taskId: string; input: any; session?: Mapstring, any; } // AgentRegistry.ts export class SimpleAgentRegistry { private agents new Mapstring, AgentDefinition(); register(def: AgentDefinition): void { if (this.agents.has(def.id)) { throw new Error(Agent with id ${def.id} already registered.); } this.agents.set(def.id, def); } get(id: string): AgentDefinition | undefined { return this.agents.get(id); } getAll(): AgentDefinition[] { return Array.from(this.agents.values()); } async createAgent(id: string): PromiseAgent { const def this.get(id); if (!def) { throw new Error(Agent ${id} not found.); } return await def.factory(); } }5.2 第二步实现一个基于关键词的路由器我们先实现一个简单的路由器它基于关键词匹配。// KeywordRouter.ts export class KeywordRouter { constructor(private registry: SimpleAgentRegistry) {} async route(taskContext: TaskContext): PromiseAgentDefinition { const query typeof taskContext.input ‘string’ ? taskContext.input : JSON.stringify(taskContext.input); const queryWords this.extractKeywords(query.toLowerCase()); const candidates this.registry.getAll(); let bestScore 0; let bestAgent: AgentDefinition | null null; for (const agent of candidates) { let score 0; // 检查描述中的关键词 score this.countKeywordMatches(queryWords, agent.description.toLowerCase()); // 检查能力标签 for (const cap of agent.capabilities) { if (queryWords.has(cap.toLowerCase())) { score 2; // 能力标签匹配给予更高权重 } } if (score bestScore) { bestScore score; bestAgent agent; } } // 如果匹配度太低返回一个默认的通用 Agent if (bestScore 1 || !bestAgent) { const fallback this.registry.get(‘general_agent’); if (!fallback) { throw new Error(‘No suitable agent found and no fallback agent available.’); } return fallback; } return bestAgent; } private extractKeywords(text: string): Setstring { // 简单的分词和停用词过滤实际项目中可用更专业的 NLP 库 const stopWords new Set([‘the’, ‘a’, ‘an’, ‘and’, ‘or’, ‘but’, ‘in’, ‘on’, ‘at’, ‘to’, ‘for’]); const words text.split(/[\s\W]/).filter(word word.length 2 !stopWords.has(word)); return new Set(words); } private countKeywordMatches(queryWords: Setstring, text: string): number { let count 0; for (const word of queryWords) { if (text.includes(word)) { count; } } return count; } }5.3 第三步构建任务执行引擎这个引擎负责驱动整个流程接收任务 - 路由 - 执行 Agent - 处理结果并可能继续路由。// TaskExecutionEngine.ts export class TaskExecutionEngine { constructor( private registry: SimpleAgentRegistry, private router: KeywordRouter // 可以替换为其他路由器 ) {} async executeTask(initialInput: any, initialTaskId?: string): Promiseany { const taskId initialTaskId || task_${Date.now()}; const context: TaskContext { taskId, input: initialInput, session: new Map() }; let currentAgentDef: AgentDefinition | null null; let finalOutput: any; // 循环执行直到某个 Agent 不指定下一个 Agent或者执行失败 while (true) { // 决定当前由哪个 Agent 执行 if (!currentAgentDef) { // 第一次由路由器决定 currentAgentDef await this.router.route(context); } else if (context.input.nextAgentId) { // 如果上一个 Agent 指定了下一个则直接使用 const nextId context.input.nextAgentId; delete context.input.nextAgentId; // 清理避免循环引用 currentAgentDef this.registry.get(nextId); if (!currentAgentDef) { throw new Error(Specified next agent ${nextId} not found.); } } else { // 没有指定下一个任务结束 finalOutput context.input; break; } console.log([${taskId}] Routing to agent: ${currentAgentDef.name}); // 创建 Agent 实例并执行 const agent await this.registry.createAgent(currentAgentDef.id); const result await agent.execute(context); if (!result.success) { throw new Error(Agent ${currentAgentDef.id} execution failed.); } // 更新上下文为下一次循环或结束做准备 context.input result.output; // 如果 result 中指定了 nextAgentId它会被下一轮循环读取 if (result.nextAgentId) { context.input.nextAgentId result.nextAgentId; } } console.log([${taskId}] Task completed successfully.); return finalOutput; } }5.4 第四步组装并运行最后我们把所有部分组装起来并模拟一个运行场景。// main.ts import { SimpleAgentRegistry } from ‘./AgentRegistry’; import { KeywordRouter } from ‘./KeywordRouter’; import { TaskExecutionEngine } from ‘./TaskExecutionEngine’; // 1. 定义几个简单的 Agent const generalAgentDef { id: ‘general_agent’, name: ‘General Assistant’, description: ‘A general-purpose assistant that can handle various queries.’, capabilities: [‘general’, ‘chat’], factory: async () ({ execute: async (ctx) ({ success: true, output: I received: ${JSON.stringify(ctx.input)} }) }) }; const mathAgentDef { id: ‘math_agent’, name: ‘Math Expert’, description: ‘Specializes in mathematical calculations and problem solving.’, capabilities: [‘math’, ‘calculation’, ‘numbers’], factory: async () ({ execute: async (ctx) { const query ctx.input; // 极其简单的数学检测 if (query.includes(‘’)) { const [a, b] query.split(‘’).map(Number); return { success: true, output: The sum is ${a b} }; } return { success: true, output: I’m a math agent, but I couldn’t parse the math in: ${query} }; } }) }; const weatherAgentDef { id: ‘weather_agent’, name: ‘Weather Specialist’, description: ‘Provides weather information and forecasts. (Simulated)’, capabilities: [‘weather’, ‘forecast’, ‘temperature’], factory: async () ({ execute: async (ctx) ({ success: true, output: The weather is sunny and 22°C. (This is a simulation) }) }) }; // 2. 创建注册表并注册 Agent const registry new SimpleAgentRegistry(); registry.register(generalAgentDef); registry.register(mathAgentDef); registry.register(weatherAgentDef); // 3. 创建路由器和执行引擎 const router new KeywordRouter(registry); const engine new TaskExecutionEngine(registry, router); // 4. 运行测试 async function runTests() { console.log(‘Test 1: Math query’); const result1 await engine.executeTask(‘What is 15 27?’); console.log(‘Result:’, result1); // 应该路由到 math_agent console.log(‘\nTest 2: Weather query’); const result2 await engine.executeTask(‘What’s the weather like today?’); console.log(‘Result:’, result2); // 应该路由到 weather_agent console.log(‘\nTest 3: General query’); const result3 await engine.executeTask(‘Tell me a joke.’); console.log(‘Result:’, result3); // 应该回退到 general_agent } runTests().catch(console.error);运行这个简单的系统你会看到不同的查询被路由到了不同的“专家”Agent。这虽然简陋但已经包含了多 Agent 路由系统的核心骨架注册、匹配、执行、传递。6. 避坑指南与进阶思考在设计和实现自己的多 Agent 路由系统时有几个坑是几乎一定会遇到的。坑一路由的“冷启动”和“描述质量”问题。一个新 Agent 加入系统或者 Agent 的描述写得过于笼统会导致路由精度极差。解决方案除了优化描述可以引入“反馈学习”机制。当用户对某个路由结果进行纠正例如在聊天界面选择“这不是我想要的”系统可以记录这次“查询-正确 Agent”的配对用于微调路由模型或作为新的匹配规则。坑二Agent 间的通信协议不统一。如果数据分析 Agent 输出一个复杂的 JSON 对象而图表生成 Agent 期望一个数组协作就会失败。解决方案严格定义并校验每个 Agent 的inputSchema和outputSchema使用 JSON Schema 等。路由器或一个专门的“适配器”层可以在传递上下文前进行数据格式转换。在 Claude Code 的实践中可能要求每个 Agent 都遵循一个统一的AgentResult格式并在内部处理数据适配。坑三循环路由与死锁。Agent A 执行完把任务交给 BB 执行完又指回给 A形成死循环。解决方案在TaskContext中维护一个已执行 Agent 的 ID 列表。路由器在决策时检查这个列表如果发现循环则中断流程并报错或者强制路由给一个冲突解决 Agent。坑四性能与延迟。每次路由都计算一次语义相似度尤其是调用远程嵌入模型会带来显著延迟。解决方案缓存对固定的 Agent 描述进行向量化并缓存结果。分级路由先用快速的关键词匹配过滤掉明显不相关的 Agent只对少数候选进行精确的语义匹配。异步与流式对于长任务不要让用户等待所有 Agent 完成。可以设计成每个 Agent 完成后就流式返回部分结果。进阶思考从“路由”到“编排”当系统非常复杂时简单的“下一个 Agent”模式会力不从心。你需要一个更强大的工作流编排引擎。这个引擎可以支持并行执行多个 Agent。处理条件分支if-else。实现循环for-loop。具备错误处理和重试机制。提供可视化界面来设计和监控工作流。这时你可以考虑集成或借鉴像 Temporal、Camunda 这样的工作流引擎或者基于状态机如 XState自己实现一个轻量级的编排层。Claude Code 的未来版本也很有可能向这个方向演进。7. 总结与个人体会剖析 Claude Code 这类框架的多 Agent 路由与定义机制其价值远不止于学会使用一个工具。它更像是一份设计蓝图揭示了构建复杂 AI 应用时如何管理“复杂性”的核心思想通过清晰的角色定义、松耦合的组件通信和智能的任务分发将一个大问题分解为一系列小问题并由专门的“小脑”去解决。从我自己的实践来看在项目初期就花时间设计好 Agent 的职责边界和通信契约比事后修补要省力十倍。路由策略不必一开始就追求完美的 AI 语义匹配一个简单的基于关键词或规则的路由器往往能解决 80% 的问题并且更可控、更易调试。随着业务复杂度的增长再逐步引入更智能的路由机制。另外不要忽视可观测性。为你的路由决策加上详细的日志为什么选择 A 而不是 B匹配得分是多少为每个 Agent 的执行过程加上监控和指标。当系统出问题时这些日志是你定位问题的唯一救命稻草。多 Agent 系统调试起来比单体应用复杂得多清晰的日志流就像系统的“心电图”能让你快速找到阻塞的“血管”。最后保持架构的开放性。今天你用 Claude Code 的这套模式明天可能就需要接入 LangGraph 来编排更复杂的流程或者需要让你的 Agent 能够调用外部 SaaS 的 API。让路由层、Agent 定义层与底层的具体执行引擎无论是 Claude Code、LangChain 还是自定义的解耦你的系统才能拥有长久的生命力。
