多Agent系统路由与定义机制：从概念到TypeScript工程实践

1. 项目概述：从“单兵作战”到“团队协作”的Agent进化

最近在折腾AI应用开发，尤其是Agent（智能体）框架时，发现一个挺普遍的问题：单个Agent能力再强，面对复杂任务也容易捉襟见肘。这就好比一个全栈工程师，既要写前端又要搞后端还要管运维，迟早会碰到瓶颈。于是，多Agent协作系统成了必然的进化方向。但问题也随之而来——多个Agent之间，任务怎么分配？消息如何传递？谁该在什么时候、以什么方式被调用？

这恰恰是“Claude Code 源码：Agent 工具 — 多 Agent 的路由与定义机制”这个项目要解决的核心问题。它不是一个从零开始的框架，而是聚焦于构建多Agent系统的“中枢神经系统”：路由与定义。简单说，它提供了一套机制，让你能像定义公司里的不同部门（Agent）一样，明确每个部门的职责（定义），并建立一套高效的内部协作流程（路由），确保任务能精准、高效地分派给最合适的“员工”去处理。

我花了不少时间研究相关的实现，特别是结合TypeScript这类强类型语言在构建复杂、可维护系统上的优势。你会发现，一个清晰的路由机制，不仅能避免Agent之间的混乱和冲突，更是实现复杂工作流自动化、提升系统可靠性和可扩展性的基石。无论你是想搭建一个自动化的内容创作流水线，还是一个能处理多步骤客服咨询的智能系统，理解并设计好多Agent的路由与定义，都是绕不开的关键一步。

2. 核心设计思路：如何构建一个“智能调度中心”

2.1 从“硬编码”到“声明式”的Agent定义转变

在早期的多Agent实验中，很多人的做法是“硬编码”。比如，在代码里直接写死：如果用户问题包含“翻译”，就调用TranslationAgent；如果包含“总结”，就调用SummaryAgent。这种方法在小规模原型阶段很快，但弊端显而易见：每增加一个新Agent，就要去修改核心的分发逻辑；Agent之间的依赖关系混乱；难以进行统一的技能描述和能力管理。

Claude Code相关项目所倡导的，是一种声明式的Agent定义机制。它的核心思想是：将每个Agent看作一个独立的、功能完备的模块，并为其附加丰富的“元数据”（Metadata）来描述自己。这套元数据通常包括：

1. 身份标识（ID/Name）：Agent的唯一名称，如translator,summarizer,code_reviewer。
2. 能力描述（Description/Capabilities）：用自然语言清晰描述这个Agent能做什么、擅长什么。例如：“一个专业的英译中翻译Agent，擅长处理技术文档和文学性文本。”
3. 输入/输出规格（Input/Output Schema）：定义这个Agent接受什么格式的数据，返回什么格式的数据。这在TypeScript中通常用接口（Interface）或Zod这类运行时类型校验库来定义，确保数据流动的类型安全。
4. 触发条件或路由键（Routing Keys/Triggers）：一组关键词、意图分类或规则，用于路由系统进行匹配。例如，translatorAgent的路由键可以是[“translate”, “translation”, “中文翻译”]。
5. 配置参数：如调用的底层模型（GPT-4, Claude等）、温度参数、上下文长度等。

通过这种声明式的定义，每个Agent都成了一个自描述的“插件”。路由系统不再需要关心Agent内部如何实现，它只需要读取这些元数据，就能知道该把任务派给谁。这极大地提升了系统的模块化和可扩展性。

注意：能力描述（Description）至关重要。它不仅是给人看的，更是给路由算法（尤其是基于语义相似度的路由）用的。清晰、准确、包含关键动词和领域名词的描述，能显著提升路由的准确率。

2.2 路由机制的三层设计：规则、语义与策略

路由机制是多Agent系统的“大脑”。一个健壮的路由系统通常不是单一策略，而是多层决策的叠加。我将其归纳为三个层次：

第一层：基于规则的路由（Rule-based Routing）这是最直接、确定性最高的一层。它直接匹配用户输入中的关键词、命令或结构化意图。

• 实现方式：预定义的路由表或if-else/switch逻辑。
• 适用场景：处理明确的、指令式的请求。例如，用户输入“/translate 你好世界”，系统直接解析命令/translate并路由到翻译Agent。
• 优点：简单、快速、零歧义。
• 缺点：僵硬，无法处理自然语言中复杂、隐含的意图。

第二层：基于语义的路由（Semantic Routing）这是让系统显得“智能”的关键。它不依赖精确的关键词匹配，而是理解用户查询的意图，并将其与Agent的能力描述进行相似度计算。

• 实现方式：通常借助文本嵌入模型（Embedding Model），将用户查询和所有Agent的能力描述转换成向量（vector），然后计算余弦相似度，选择相似度最高的Agent。
• 适用场景：处理开放域、描述性的自然语言请求。例如，用户说“帮我把这段技术文档的核心意思用中文提炼一下”，语义路由能将其匹配到“总结Agent”，即使句中没有“总结”这个词。
• 优点：灵活，能处理丰富的自然语言表达。
• 缺点：依赖嵌入模型的质量，计算有开销，可能存在相似度接近时的“边界模糊”问题。

第三层：基于策略的路由（Policy-based Routing）这是最复杂、也最强大的一层，它引入了决策逻辑和状态管理。

• 会话状态（Session State）：考虑当前对话的上下文。例如，用户先问了“北京的天气”，接着问“那上海呢？”，系统需要知道“那上海呢？”指的是“上海的天气”，并路由到天气查询Agent，而不是开启一个新任务。
• 工作流编排（Workflow Orchestration）：某些复杂任务需要多个Agent按顺序或并行执行。策略路由需要管理这个流程。例如，“帮我写一份市场分析报告”可能触发一个工作流：先由ResearchAgent搜集资料，再由AnalysisAgent分析数据，最后由ReportWritingAgent成文。路由系统在这里扮演了“流程控制器”的角色。
• 负载均衡与熔断：在多个同类型Agent实例间分配任务，或在某个Agent持续失败时将其从路由表中暂时移除。
• 实现方式：可能需要一个独立的“编排器”（Orchestrator）或“监督Agent”（Supervisor Agent），它维护对话状态和工作流定义，做出高级路由决策。

在实际项目中，这三层往往是结合使用的。一个常见的流程是：先检查是否有明确的规则匹配；如果没有，则进入语义路由；对于复杂任务，则由策略路由层接管，进行工作流分解和状态管理。

2.3 TypeScript在实现中的核心优势

为什么用TypeScript来实现这样的系统？在深入代码后，我体会到了几个不可替代的优势：

1. 类型安全（Type Safety）：这是最大的优点。我们可以为Agent定义严格的接口（interface Agent），为每个Agent的输入输出定义类型（interface TranslationInput { text: string; targetLang: string; }）。这能在编译阶段就捕获大量错误，比如错误地传递了参数，或者错误地处理了返回值。在构建由多个松散耦合的Agent组成的系统时，类型是防止“接口腐化”的最佳契约。
2. 出色的IDE支持：得益于类型系统，VS Code等IDE能提供无与伦比的自动补全、跳转到定义和重构支持。当你修改一个Agent的接口时，所有调用它的地方都会立刻被标记出来，极大提升了开发效率。
3. 面向对象与函数式编程的融合：TS既支持类（Class）来封装Agent的状态和行为，也支持高阶函数、泛型等函数式特性来构建灵活的路由器和中间件。例如，可以轻松实现一个泛型路由函数：route<T extends Agent>(query: string, agents: T[]): Promise<T | null>。
4. 丰富的生态系统：有Zod用于运行时校验，有LangChain.js、Vercel AI SDK等成熟框架的部分理念可借鉴，有各种向量数据库客户端（如@pinecone-database/pinecone）用于语义路由，工具链非常完善。

3. 核心模块拆解与实现细节

3.1 Agent定义模块：打造自描述的智能体

让我们先看看一个Agent的核心定义应该包含什么。以下是一个高度简化的示例，展示了用TypeScript接口和类来定义Agent的骨架：

// 首先，定义Agent能力的元数据接口 interface AgentCapabilities { name: string; description: string; // 用于语义匹配的详细描述 keywords: string[]; // 用于规则匹配的关键词 inputSchema: any; // 可以使用zod、json-schema等具体定义 outputSchema: any; } // 定义基础的Agent接口，所有具体Agent都必须实现 interface IAgent { readonly capabilities: AgentCapabilities; invoke(input: any, context?: AgentContext): Promise<any>; } // 定义一个上下文，用于在Agent间传递会话状态、用户信息等 interface AgentContext { sessionId: string; conversationHistory: Array<{role: string, content: string}>; userPreferences?: any; } // 一个具体的翻译Agent实现示例 class TranslationAgent implements IAgent { capabilities: AgentCapabilities = { name: 'translator', description: '一个专业的翻译助手，可将英文技术文档、日常对话准确翻译成中文。也支持中译英。', keywords: ['翻译', 'translate', '英文', '中文', 'language'], inputSchema: z.object({ text: z.string(), sourceLang: z.string().optional(), targetLang: z.string() }), outputSchema: z.object({ translatedText: z.string() }) }; async invoke(input: { text: string; targetLang: string }, context?: AgentContext): Promise<{ translatedText: string }> { // 这里整合实际的AI模型调用，比如调用OpenAI或Claude的API const prompt = `请将以下${input.sourceLang || '英文'}文本翻译成${input.targetLang}：\n${input.text}`; // ... 调用AI模型 ... const result = await callAIModel(prompt); return { translatedText: result }; } }

实操心得：

• 描述（description）要具体：避免使用“处理文本”这种模糊描述。应写成“分析用户反馈情感倾向并提取关键问题点”，这样语义路由才能准确匹配。
• 输入输出模式（Schema）要强制校验：在invoke方法内部，一定要用inputSchema对输入进行校验。这能防止错误数据导致下游AI调用失败或产生不可预期的结果。Zod库在这方面非常好用，它能提供清晰的错误信息。
• 上下文（Context）的设计：AgentContext是串联多轮对话和跨Agent协作的生命线。除了会话历史，考虑加入本次任务的全局目标、已生成的中介结果等，方便后续Agent利用。

3.2 路由引擎模块：实现多层决策逻辑

路由引擎是系统的调度中心。下面我们实现一个结合了规则和语义的两层路由引擎。

// 一个简单的规则路由匹配器 class RuleBasedRouter { private agentKeywordMap: Map<string, IAgent> = new Map(); registerAgent(agent: IAgent) { agent.capabilities.keywords.forEach(keyword => { this.agentKeywordMap.set(keyword.toLowerCase(), agent); }); } route(query: string): IAgent | null { const lowerQuery = query.toLowerCase(); for (const [keyword, agent] of this.agentKeywordMap.entries()) { if (lowerQuery.includes(keyword)) { return agent; } } return null; } } // 一个基于向量相似度的语义路由匹配器（概念示例，需接入真实嵌入模型和向量数据库） import { Pinecone } from '@pinecone-database/pinecone'; // 示例向量数据库客户端 class SemanticRouter { private pc: Pinecone; private indexName: string; constructor(pineconeApiKey: string, indexName: string) { this.pc = new Pinecone({ apiKey: pineconeApiKey }); this.indexName = indexName; } async registerAgent(agent: IAgent) { const index = this.pc.index(this.indexName); // 将Agent的描述文本转换为向量 const embedding = await generateEmbedding(agent.capabilities.description); // 将向量和Agent的元数据存入向量数据库 await index.upsert([{ id: agent.capabilities.name, values: embedding, metadata: { agentName: agent.capabilities.name } }]); } async route(query: string): Promise<IAgent | null> { const index = this.pc.index(this.indexName); // 将用户查询转换为向量 const queryEmbedding = await generateEmbedding(query); // 在向量数据库中搜索最相似的Agent描述 const results = await index.query({ vector: queryEmbedding, topK: 3, // 返回最相似的3个 includeMetadata: true }); if (results.matches && results.matches.length > 0 && results.matches[0].score > 0.7) { // 设定一个相似度阈值 const topAgentName = results.matches[0].metadata?.agentName; // 这里需要根据agentName找到对应的IAgent实例，可能需要一个注册表 return getAgentByName(topAgentName); } return null; } } // 主路由器，组合两种策略 class MasterRouter { private ruleRouter: RuleBasedRouter; private semanticRouter: SemanticRouter; private agentRegistry: Map<string, IAgent> = new Map(); // Agent注册表 constructor(semanticRouter: SemanticRouter) { this.ruleRouter = new RuleBasedRouter(); this.semanticRouter = semanticRouter; } async registerAgent(agent: IAgent) { this.agentRegistry.set(agent.capabilities.name, agent); this.ruleRouter.registerAgent(agent); await this.semanticRouter.registerAgent(agent); } async route(query: string): Promise<IAgent | null> { // 第一优先级：规则匹配 const ruleBasedAgent = this.ruleRouter.route(query); if (ruleBasedAgent) { console.log(`[路由] 规则匹配到Agent: ${ruleBasedAgent.capabilities.name}`); return ruleBasedAgent; } // 第二优先级：语义匹配 const semanticAgent = await this.semanticRouter.route(query); if (semanticAgent) { console.log(`[路由] 语义匹配到Agent: ${semanticAgent.capabilities.name}`); return semanticAgent; } // 都未匹配到，可以返回一个默认Agent（如通用对话Agent）或抛出错误 console.log(`[路由] 未找到匹配的Agent`); return this.agentRegistry.get('default_chat_agent') || null; } }

实现要点：

1. 注册表模式：MasterRouter维护一个agentRegistry，这是通过Agent名称快速查找Agent实例的映射。语义路由返回的是Agent名称，需要通过这个注册表拿到实例。
2. 路由优先级：这里采用了“规则优先”的策略，因为规则匹配更快、更确定。也可以根据业务场景调整，比如对于某些复杂查询，即使有关键词也优先走语义路由。
3. 阈值设置：语义路由中的score > 0.7是一个重要参数。阈值太高，会导致很多查询无法匹配；阈值太低，则可能产生错误匹配。这个值需要根据你的嵌入模型和具体任务进行校准。
4. 向量数据库的选择与优化：对于Agent数量不多（比如几十个）的场景，甚至可以在内存中计算相似度，无需引入外部向量数据库。但对于成百上千个Agent，使用Pinecone、Weaviate或Qdrant等专业向量数据库是必要的。注意为Agent描述向量创建独立的索引（indexName）。

3.3 工作流编排模块：处理复杂多步任务

当单个Agent无法完成任务时，就需要工作流编排。这通常由一个特殊的“监督Agent”或“编排器”来完成。

interface WorkflowStep { agentName: string; input: (context: WorkflowContext) => any; // 动态生成输入 condition?: (context: WorkflowContext) => boolean; // 执行条件 } interface WorkflowContext { originalQuery: string; sessionId: string; stepsOutput: Map<string, any>; // 存储每一步的输出 currentStep: number; } class WorkflowOrchestrator { private workflows: Map<string, WorkflowStep[]> = new Map(); // 预定义工作流，例如“撰写报告” registerWorkflow(name: string, steps: WorkflowStep[]) { this.workflows.set(name, steps); } async execute(workflowName: string, initialContext: Partial<WorkflowContext>): Promise<any> { const steps = this.workflows.get(workflowName); if (!steps) throw new Error(`工作流 ${workflowName} 未定义`); const context: WorkflowContext = { originalQuery: '', sessionId: `wf_${Date.now()}`, stepsOutput: new Map(), currentStep: 0, ...initialContext }; for (let i = 0; i < steps.length; i++) { const step = steps[i]; context.currentStep = i; // 检查执行条件 if (step.condition && !step.condition(context)) { console.log(`步骤 ${i} (${step.agentName}) 条件不满足，跳过`); continue; } // 准备输入 const input = step.input(context); // 获取Agent实例（假设有一个全局的agent管理器） const agent = getAgentByName(step.agentName); if (!agent) throw new Error(`Agent ${step.agentName} 未找到`); console.log(`[工作流] 执行步骤 ${i}: ${step.agentName}`); // 执行Agent const output = await agent.invoke(input, { sessionId: context.sessionId }); // 存储输出，供后续步骤使用 context.stepsOutput.set(step.agentName, output); } // 返回最终结果，可能是最后一步的输出，也可能是所有结果的聚合 return this.aggregateResult(context); } private aggregateResult(context: WorkflowContext): any { // 简单实现：返回最后一步的输出 const lastStep = this.workflows.get(workflowName)?.[context.currentStep]; if (lastStep) { return context.stepsOutput.get(lastStep.agentName); } return null; } } // 使用示例：定义一个“市场分析报告”工作流 const orchestrator = new WorkflowOrchestrator(); orchestrator.registerWorkflow('market_analysis_report', [ { agentName: 'research_agent', input: (ctx) => ({ topic: ctx.originalQuery }), // 从初始上下文中获取主题 }, { agentName: 'data_analysis_agent', input: (ctx) => ({ rawData: ctx.stepsOutput.get('research_agent')?.data }), // 使用上一步的结果 }, { agentName: 'report_writing_agent', input: (ctx) => ({ analysis: ctx.stepsOutput.get('data_analysis_agent')?.insights, format: 'markdown' }), } ]); // 当主路由器识别到复杂任务时，触发工作流 // if (isComplexTask(query)) { // const result = await orchestrator.execute('market_analysis_report', { originalQuery: query }); // }

编排逻辑的核心：

• 动态输入：每个步骤的输入是一个函数，它能访问整个工作流上下文，从而可以基于之前步骤的结果来动态构建输入。这是实现Agent间数据传递的关键。
• 条件执行：condition字段允许实现分支逻辑。例如，如果研究Agent返回的数据不足，可以跳过分析步骤，直接路由到“数据不足提示Agent”。
• 错误处理与回退：生产环境中，必须在每个步骤加入重试、超时和失败处理逻辑。一个步骤失败，整个工作流是终止、回退还是进入补偿流程，都需要仔细设计。

4. 实战配置与高级技巧

4.1 配置管理：让系统灵活可调

一个实用的多Agent系统，其配置项会很多。硬编码在代码里是灾难。推荐使用环境变量加配置文件的方式。

// config/agents.ts - Agent能力定义配置 export const agentConfigs = [ { name: 'translator', description: '专业翻译，擅长中英互译，特别是技术文档。', keywords: ['翻译', 'translate', '英文', '中文'], model: 'gpt-4', // 指定使用的模型 temperature: 0.2, // ... 其他配置 }, { name: 'summarizer', description: '文本总结专家，能提取长篇文章、会议记录的核心要点。', keywords: ['总结', '概括', '摘要', 'summarize'], model: 'claude-3-sonnet', maxTokens: 500, }, // ... 更多Agent配置 ]; // config/routing.ts - 路由策略配置 export const routingConfig = { ruleBased: { enabled: true, priority: 'high', // 优先级高于语义路由 }, semantic: { enabled: true, embeddingModel: 'text-embedding-3-small', // 使用的嵌入模型 vectorIndex: 'agent_capabilities_index', similarityThreshold: 0.72, // 相似度阈值，可调 topK: 5, }, fallbackAgent: 'default_chat_agent', // 兜底Agent }; // 系统初始化时，根据配置动态创建Agent实例和路由器 import { agentConfigs } from './config/agents'; import { routingConfig } from './config/routing'; class AgentSystem { async initialize() { const agents: IAgent[] = []; for (const config of agentConfigs) { // 根据config.type等字段，动态实例化不同的Agent类 const agent = AgentFactory.createAgent(config); await this.masterRouter.registerAgent(agent); agents.push(agent); } // 根据routingConfig配置路由器的行为 this.masterRouter.setThreshold(routingConfig.semantic.similarityThreshold); // ... } }

配置化的好处：你可以不修改代码，仅通过调整配置文件，就能增加新的Agent、修改Agent描述以优化路由、调整路由阈值、切换底层模型等。这为A/B测试和线上调优提供了极大便利。

4.2 性能优化与缓存策略

随着Agent数量增加和查询量上升，性能成为关键。

1. 向量嵌入缓存：用户查询的向量化计算和Agent描述的向量查询是主要开销。

   • 查询缓存：对相同的用户查询文本，可以直接缓存其向量和路由结果（设定一个较短的TTL，比如1分钟）。
   • Agent描述向量预加载：在系统启动时，将所有Agent的描述向量计算好并存入向量数据库，避免每次路由都重复计算。

2. Agent实例池：对于无状态的Agent（每次调用独立），可以创建实例池，避免频繁的实例创建和销毁开销。

3. 异步与并发处理：

   • 在语义路由查询向量数据库时，使用异步操作避免阻塞。
   • 如果工作流中有可以并行执行的步骤（例如，同时调用研究Agent和数据分析Agent），应使用Promise.all并发执行以缩短总耗时。

// 简单的查询缓存示例 import NodeCache from 'node-cache'; const queryCache = new NodeCache({ stdTTL: 60 }); // TTL 60秒 class CachedSemanticRouter extends SemanticRouter { async route(query: string): Promise<IAgent | null> { const cacheKey = `route:${query}`; const cachedAgentName = queryCache.get<string>(cacheKey); if (cachedAgentName) { return getAgentByName(cachedAgentName); } // 未命中缓存，执行实际的向量查询 const agent = await super.route(query); if (agent) { queryCache.set(cacheKey, agent.capabilities.name); } return agent; } }

4.3 可观测性与日志记录

一个黑盒的多Agent系统是难以维护的。必须建立完善的可观测性。

1. 结构化日志：记录每一次路由决策的详细信息。

   logger.info('Routing decision made', { query, matchedBy: 'rule_based', // or 'semantic' matchedAgent: agent?.capabilities.name, semanticScore: score, // 语义匹配时的分数 timestamp: new Date().toISOString(), sessionId, });

2. 链路追踪（Tracing）：为每个用户请求生成一个唯一的traceId，并贯穿整个处理流程（路由、Agent调用、工作流步骤）。这样当出现问题时，可以快速定位是哪个环节、哪个Agent出的错。可以使用OpenTelemetry等标准。

3. 关键指标监控：

   • 路由延迟：规则路由和语义路由的平均耗时。
   • 路由命中率：规则路由、语义路由、兜底路由的占比。
   • Agent调用成功率与延迟：每个Agent的调用成功率和响应时间。
   • 工作流成功率：复杂工作流的整体成功率和各步骤成功率。

这些指标能帮你发现系统的瓶颈（比如某个语义路由太慢）或薄弱环节（比如某个Agent经常失败）。

5. 常见问题与排查实录

在实际开发和调试中，我遇到了不少典型问题。这里列出一个排查清单，希望能帮你少走弯路。

5.1 路由不准确或失败

5.2 Agent执行异常

5.3 系统性能瓶颈

5.4 扩展性与维护性挑战

随着Agent数量增长到几十上百个，手动管理配置和依赖关系变得困难。

解决方案：

• 配置中心：将Agent配置、路由配置移至独立的配置中心或数据库，支持动态更新，无需重启服务。
• 依赖注入容器：使用tsyringe、inversifyJS等IoC容器来管理Agent实例的创建和生命周期，自动解决Agent间的依赖（如果某个AnalysisAgent依赖一个DataFetcher服务）。
• Agent发现与自注册：设计一个注册中心，Agent服务在启动时主动向路由中心注册自己的能力和端点。这在微服务架构的多Agent系统中尤其有用。

构建一个健壮、高效的多Agent路由与协作系统，是一个持续迭代和优化的过程。从清晰的定义开始，实现一个多层、互补的路由策略，用TypeScript的类型系统来保障代码质量，再辅以完善的配置、监控和故障处理机制，你就能搭建起一个真正具备“团队协作”能力的AI应用骨架。这个骨架本身不直接产生最终答案，但它确保了正确的“专家”能在正确的“时间”处理正确的“问题”，这才是复杂AI应用价值倍增的关键。