当前位置：首页 > news >正文

Agent 能力注册中心：把工具和技能当作微服务治理（续篇）

news 2026/8/1 7:48:29

Agent 能力注册中心：把工具和技能当作微服务治理（续篇）

场景痛点

Agent需要调用"查询订单状态"工具。开发者写了个order_query函数注册到Agent。一个月后，订单服务API改了，order_query返回格式变了。Agent不知道，直接把旧格式的结果喂给下游步骤——整个工作流炸了。

更常见的场景：5个开发者各自写了"查询用户信息"的工具，名字分别是user_info、get_user、query_user_detail、fetch_user_profile、user_lookup。功能重叠，参数风格不同。Agent选择哪个？随机选？那结果不稳定。

核心矛盾：Agent的工具和能力缺乏统一的注册、发现、版本管理机制。开发者按自己的习惯命名和实现工具，没有标准化的描述格式，没有版本号，没有健康状态监控。这和微服务早期的"服务混乱"问题一模一样——解决方案也一模一样：注册中心。

底层机制与原理剖析

能力注册中心的本质是Agent能力的Service Registry。每个工具/技能注册时声明：名称、版本、输入输出schema、能力描述、依赖关系、健康状态。Agent调用工具前先查注册中心——获取最新版本、校验输入参数、检查工具健康状态。

关键机制：

Schema-driven描述。工具注册时必须声明JSON Schema格式的输入输出。Agent编排器根据schema校验参数、自动拼接步骤间的数据流。没有schema的工具不允许注册——就像没有API文档的微服务不允许上线。
语义名称规范。工具名称遵循domain.action.target三段式命名：order.query.status、user.create.profile、payment.process.refund。语义命名消除"5个开发者写5个名字"的混乱。注册中心检测名称冲突——同一语义已有注册工具时，拒绝重复注册。
版本管理。工具API变更时注册新版本（v2），旧版本（v1）标记为deprecated但仍可用。Agent编排器默认使用最新版本，但可以指定版本号。deprecated版本30天后自动下线——给Agent足够的迁移窗口。
健康检查。注册中心定时探测每个工具的可用性。HTTP工具发健康检查请求，本地工具执行空参数校验。健康状态为down的工具，Agent编排器不会选择——避免调用已故障的工具浪费时间和token。

生产级代码实现

CapabilityRegistry：能力注册中心核心

// registry/capability-registry.ts import { Redis } from 'ioredis'; import { Ajv } from 'ajv'; type HealthStatus = 'healthy' | 'degraded' | 'down'; type VersionStatus = 'active' | 'deprecated' | 'offline'; interface CapabilityDescriptor { // 语义名称：domain.action.target name: string; version: string; // semver格式：1.0.0 versionStatus: VersionStatus; description: string; // 人类可读描述 inputSchema: object; // JSON Schema outputSchema: object; // JSON Schema runtimeType: 'http' | 'local' | 'grpc'; // 工具运行方式 endpoint?: string; // HTTP/GRPC工具的地址 handlerPath?: string; // 本地工具的函数路径 dependencies: string[]; // 依赖的其他能力名称 healthStatus: HealthStatus; healthCheckEndpoint?: string; registeredAt: string; deprecatedAt?: string; offlineAt?: string; // 计划下线时间 owner: string; // 负责人/团队 tags: string[]; // 搜索标签 } class CapabilityRegistry { private redis: Redis; private ajv: Ajv; private healthChecker: HealthChecker; constructor(redisUrl: string) { this.redis = new Redis(redisUrl); this.ajv = new Ajv({ strict: true }); this.healthChecker = new HealthChecker(this); } // 注册新能力 async register(descriptor: CapabilityDescriptor): Promise<RegisterResult> { // 1. 验证名称规范：必须是domain.action.target三段式 // 为什么强制命名规范：语义命名让Agent能根据意图自动匹配工具， // 自由命名导致Agent无法理解工具用途 const nameParts = descriptor.name.split('.'); if (nameParts.length !== 3) { return { success: false, error: `名称必须为domain.action.target三段式格式，当前: ${descriptor.name}` }; } // 2. 检查名称冲突：同一语义名称已有活跃版本 const existing = await this.findByName(descriptor.name); const activeVersions = existing.filter( d => d.versionStatus === 'active' && d.version !== descriptor.version ); if (activeVersions.length > 0) { // 同名称已有活跃版本，但版本号不同——允许（这是版本升级） // 同名称同版本——拒绝 const sameVersion = activeVersions.find(d => d.version === descriptor.version); if (sameVersion) { return { success: false, error: `名称${descriptor.name}版本${descriptor.version}已注册` }; } } // 3. 校验input/output schema合法性 try { this.ajv.compile(descriptor.inputSchema); this.ajv.compile(descriptor.outputSchema); } catch (e: any) { return { success: false, error: `Schema校验失败: ${e.message}` }; } // 4. 存储到RegistryDB const key = `capability:${descriptor.name}:${descriptor.version}`; descriptor.registeredAt = new Date().toISOString(); descriptor.versionStatus = 'active'; descriptor.healthStatus = 'healthy'; await this.redis.set(key, JSON.stringify(descriptor)); // 5. 更新名称索引（用于按名称搜索） await this.redis.sadd(`capability_index:${descriptor.name}`, descriptor.version); // 6. 启动健康检查 this.healthChecker.registerCheck(descriptor); return { success: true, descriptor }; } // 标记能力为deprecated async deprecate(name: string, version: string, offlineDate: string): Promise<void> { const descriptor = await this.loadDescriptor(name, version); if (!descriptor) throw new Error(`能力 ${name}:${version} 不存在`); descriptor.versionStatus = 'deprecated'; descriptor.deprecatedAt = new Date().toISOString(); descriptor.offlineAt = offlineDate; // 30天后下线 await this.redis.set( `capability:${name}:${version}`, JSON.stringify(descriptor) ); // 通知订阅者：工具版本变更 // 为什么通知而非让订阅者自己发现：变更可能影响Agent编排器的步骤参数， // 及时通知允许编排器提前适配新版本 await this.publishVersionChange(name, version, 'deprecated'); } // 能力发现：Agent根据意图查询匹配的工具 async discover( intent: string, requiredTags?: string[] ): Promise<CapabilityDescriptor[]> { // 1. 意图解析：从自然语言意图提取domain和action // 为什么需要意图解析：Agent说"我想查询订单状态"， // 注册中心需要映射到"order.query.status" const parsed = this.parseIntent(intent); // 2. 按domain+action搜索 const candidates = await this.findByDomainAction(parsed.domain, parsed.action); // 3. 过滤：排除down/deprecated（除非显式指定版本） // 为什么排除deprecated而非保留：Agent默认使用最新版本， // deprecated版本只是过渡期的兼容选项 const filtered = candidates.filter( d => d.healthStatus !== 'down' && d.versionStatus === 'active' ); // 4. 标签匹配 if (requiredTags) { const tagged = filtered.filter( d => requiredTags.every(tag => d.tags.includes(tag)) ); if (tagged.length > 0) return tagged; // 标签不匹配时返回全部候选——标签是可选筛选条件，不是硬性过滤 // 为什么不硬过滤：标签是辅助分类，核心匹配靠domain+action } // 5. 按版本排序：最新版本优先 filtered.sort((a, b) => b.version.localeCompare(a.version)); return filtered; } // 参数校验：调用工具前校验输入参数是否符合schema async validateInput( name: string, version: string, input: any ): Promise<ValidationResult> { const descriptor = await this.loadDescriptor(name, version); if (!descriptor) { return { valid: false, errors: [`能力 ${name}:${version} 不存在`] }; } const validate = this.ajv.compile(descriptor.inputSchema); const valid = validate(input); if (!valid) { return { valid: false, errors: validate.errors?.map(e => `${e.instancePath}: ${e.message}`) ?? [] }; } return { valid: true }; } // 按名称查找所有版本 private async findByName(name: string): Promise<CapabilityDescriptor[]> { const versions = await this.redis.smembers(`capability_index:${name}`); const descriptors: CapabilityDescriptor[] = []; for (const version of versions) { const descriptor = await this.loadDescriptor(name, version); if (descriptor) descriptors.push(descriptor); } return descriptors; } // 按domain+action查找（模糊匹配） private async findByDomainAction( domain: string, action: string ): Promise<CapabilityDescriptor[]> { // 遍历所有能力名称索引，匹配domain和action const allNames = await this.redis.keys('capability_index:*'); const matchedNames = allNames.filter(key => { const name = key.replace('capability_index:', ''); const parts = name.split('.'); return parts[0] === domain && parts[1] === action; }); const results: CapabilityDescriptor[] = []; for (const nameKey of matchedNames) { const name = nameKey.replace('capability_index:', ''); const descriptors = await this.findByName(name); results.push(...descriptors); } return results; } // 意图解析：从自然语言提取domain和action private parseIntent(intent: string): { domain: string; action: string } { // 关键词映射表 const domainMap: Record<string, string> = { '订单': 'order', '用户': 'user', '支付': 'payment', '商品': 'product', '库存': 'inventory', '通知': 'notification' }; const actionMap: Record<string, string> = { '查询': 'query', '创建': 'create', '更新': 'update', '删除': 'delete', '发送': 'send', '计算': 'compute' }; let domain = ''; let action = ''; for (const [cn, en] of Object.entries(domainMap)) { if (intent.includes(cn)) domain = en; } for (const [cn, en] of Object.entries(actionMap)) { if (intent.includes(cn)) action = en; } // 未匹配到时使用通配符 if (!domain) domain = '*'; if (!action) action = '*'; return { domain, action }; } private async loadDescriptor(name: string, version: string): Promise<CapabilityDescriptor | null> { const key = `capability:${name}:${version}`; const data = await this.redis.get(key); return data ? JSON.parse(data) : null; } private async publishVersionChange(name: string, version: string, changeType: string): void { await this.redis.publish( 'capability_changes', JSON.stringify({ name, version, changeType, timestamp: new Date().toISOString() }) ); } } interface RegisterResult { success: boolean; error?: string; descriptor?: CapabilityDescriptor; } interface ValidationResult { valid: boolean; errors?: string[]; }

HealthChecker：能力健康检查

// registry/health-checker.ts import { CapabilityRegistry, CapabilityDescriptor, HealthStatus } from './capability-registry'; class HealthChecker { private checks: Map<string, HealthCheckConfig> = new Map(); private intervalHandle: NodeJS.Timer | null = null; constructor(private registry: CapabilityRegistry) {} // 注册健康检查配置 registerCheck(descriptor: CapabilityDescriptor): void { const key = `${descriptor.name}:${descriptor.version}`; this.checks.set(key, { type: descriptor.runtimeType, endpoint: descriptor.healthCheckEndpoint ?? descriptor.endpoint, handlerPath: descriptor.handlerPath, intervalMs: 30000, // 每30秒检查一次 timeoutMs: 5000, // 单次检查超时5秒 consecutiveFailures: 0, maxConsecutiveFailures: 3 // 连续3次失败标记为down }); } // 启动定期健康检查 start(): void { // 为什么用定时器而非事件驱动：健康检查是周期性运维任务， // 不依赖外部事件触发 this.intervalHandle = setInterval(() => this.runAllChecks(), 30000); } stop(): void { if (this.intervalHandle) { clearInterval(this.intervalHandle); } } private async runAllChecks(): void { for (const [key, config] of this.checks) { const [name, version] = key.split(':'); try { const healthy = await this.checkOnce(config); if (healthy) { config.consecutiveFailures = 0; await this.registry.updateHealthStatus(name, version, 'healthy'); } else { config.consecutiveFailures++; if (config.consecutiveFailures >= config.maxConsecutiveFailures) { await this.registry.updateHealthStatus(name, version, 'down'); } else { await this.registry.updateHealthStatus(name, version, 'degraded'); } } } catch { // 健康检查自身失败（网络中断等），标记为degraded而非down // 为什么标记degraded：检查失败可能是检查器自身问题，不一定代表工具故障 config.consecutiveFailures++; await this.registry.updateHealthStatus(name, version, 'degraded'); } } } private async checkOnce(config: HealthCheckConfig): Promise<boolean> { switch (config.type) { case 'http': // HTTP工具：发健康检查请求 try { const response = await fetch(config.endpoint!, { method: 'GET', signal: AbortSignal.timeout(config.timeoutMs) }); return response.ok; } catch { return false; } case 'local': // 本地工具：执行空参数校验（不实际调用，只验证handler可用） try { // 检查handler函数是否可导入 // 为什么只校验导入而非实际执行：空参数执行可能产生副作用 const module = await import(config.handlerPath!); return typeof module.default === 'function'; } catch { return false; } case 'grpc': // GRPC工具：检查连接是否可用 try { // 简化版：检查endpoint是否可达 const response = await fetch(`http://${config.endpoint!.replace(':443', ':80')}/health`, { signal: AbortSignal.timeout(config.timeoutMs) } ); return response.ok; } catch { return false; } default: return false; } } } interface HealthCheckConfig { type: 'http' | 'local' | 'grpc'; endpoint?: string; handlerPath?: string; intervalMs: number; timeoutMs: number; consecutiveFailures: number; maxConsecutiveFailures: number; }

CapabilityVersionManager：版本生命周期管理

// registry/version-manager.ts class CapabilityVersionManager { private registry: CapabilityRegistry; constructor(registry: CapabilityRegistry) { this.registry = registry; } // 自动下线检查：每天扫描deprecated能力，到期自动下线 // 为什么自动化而非人工操作：人工下线容易遗漏，过期工具调用失败影响Agent稳定性 async autoOfflineCheck(): Promise<OfflineReport> { const report: OfflineReport = { offlined: [], pending: [], errors: [] }; // 扫描所有deprecated能力 const allIndexKeys = await this.registry.scanIndexKeys(); for (const key of allIndexKeys) { const name = key.replace('capability_index:', ''); const versions = await this.registry.findByName(name); const deprecated = versions.filter(v => v.versionStatus === 'deprecated'); for (const cap of deprecated) { if (!cap.offlineAt) continue; const offlineDate = new Date(cap.offlineAt); const now = new Date(); if (now >= offlineDate) { // 到期下线 try { await this.registry.updateVersionStatus( cap.name, cap.version, 'offline' ); report.offlined.push(`${cap.name}:${cap.version}`); } catch (e: any) { report.errors.push(`${cap.name}:${cap.version} - ${e.message}`); } } else { const daysLeft = Math.ceil( (offlineDate.getTime() - now.getTime()) / 86400000 ); report.pending.push(`${cap.name}:${cap.version} - ${daysLeft}天后下线`); } } } return report; } // 版本升级辅助：创建新版本并标记旧版本deprecated async upgradeVersion( name: string, oldVersion: string, newDescriptor: CapabilityDescriptor, transitionDays: number = 30 ): Promise<void> { // 注册新版本 newDescriptor.name = name; await this.registry.register(newDescriptor); // 标记旧版本deprecated const offlineDate = new Date( Date.now() + transitionDays * 86400000 ).toISOString().split('T')[0]; await this.registry.deprecate(name, oldVersion, offlineDate); } } interface OfflineReport { offlined: string[]; pending: string[]; errors: string[]; }