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

Agent 能力注册中心把工具和技能当作微服务治理一、Agent 的工具管理为什么需要注册中心单体 Agent 时代工具是硬编码在tools数组里的。当你有 10 个 Agent、50 个工具时问题爆发了工具发现一个新工具上线哪些 Agent 应该看到它版本管理工具 API 签名变了所有调用方都要同步更新——经典的分布式契约问题权限控制Agent A 可以调用发送邮件Agent B 不可以——权限怎么管理可用性工具服务宕机了Agent 调用超时 30 秒还不断——需要熔断这些问题的本质Agent 的工具调用就是微服务间的 RPC 调用。微服务治理的所有经验都可以平移过来。flowchart TD A[Agent 实例] -- B{能力注册中心} B -- C[工具注册] B -- D[工具发现] B -- E[健康检查] B -- F[权限控制] C -- C1[API Schema 定义] C -- C2[版本管理] C -- C3[服务端点] D -- D1[按标签过滤] D -- D2[能力路由] E -- E1[心跳检测] E -- E2[自动摘除] F -- F1[RBAC 模型] F -- F2[调用审计] style A fill:#4a90d9 style B fill:#e67e22二、能力注册中心的设计要点工具 Service把每个工具视为一个微服务。它的 API Schema、端点地址、健康状态、版本号都注册在中心。Agent 启动时拉取授权列表运行时按需发现。Schema 优先工具不是注册一个名字就完事必须带完整的 JSON Schema 描述——入参类型、必填字段、返回值结构。这不是给 Agent 看的是给 LLM 的 function calling 用的。Schema 写得好坏直接决定 Agent 调用工具的准确率。版本语义化工具的 Schema 版本用 SemVer。Agent 声明兼容的工具版本范围如 1.0,2.0。工具发布 2.0 breaking change 时1.x 版本仍然保留 30 天——给调用方迁移窗口。三、注册中心服务端实现import asyncio import json from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum from typing import Optional import redis.asyncio as redis class ToolStatus(Enum): ACTIVE active DEPRECATED deprecated OFFLINE offline dataclass class ToolSchema: 工具的 API Schema。 openai_schema 直接对应 OpenAI function calling 的参数格式。 这个设计避免中间转换——注册时存的格式就是 Agent 调用时的格式。 name: str version: str # SemVer description: str openai_schema: dict # 符合 OpenAI function definition 的 JSON Schema endpoint: str # 服务的 gRPC/HTTP 地址 status: ToolStatus ToolStatus.ACTIVE tags: list[str] field(default_factorylist) owner: str ttl_seconds: int 30 # 心跳超时 dataclass class AgentIdentity: agent_id: str role: str permissions: list[str] field(default_factorylist) class CapabilityRegistry: Agent 能力注册中心。 架构选择Redis 做热存储PostgreSQL 做冷存储。 热存储用于高频的发现和健康检查 1ms。 冷存储用于审计、版本回溯秒级即可。 def __init__(self, redis_url: str): self._redis redis.from_url(redis_url) async def register_tool(self, tool: ToolSchema) - bool: 注册或更新工具。 使用 HSET 而非 SET因为需要单独更新 status 字段而不覆盖其他数据。 key ftool:{tool.name} tool_data { name: tool.name, version: tool.version, description: tool.description, schema: json.dumps(tool.openai_schema), endpoint: tool.endpoint, status: tool.status.value, tags: json.dumps(tool.tags), owner: tool.owner, last_heartbeat: datetime.now().isoformat(), } await self._redis.hset(key, mappingtool_data) # 心跳 TTL超过 TTL 未更新工具自动标记离线 await self._redis.expire(key, tool.ttl_seconds * 2) return True async def heartbeat(self, tool_name: str) - None: 工具心跳更新。 设计决策心跳只更新 expire 时间不修改任何数据字段。 这保证心跳操作是 O(1) 的原子操作。 key ftool:{tool_name} exists await self._redis.exists(key) if not exists: return await self._redis.hset(key, last_heartbeat, datetime.now().isoformat()) await self._redis.expire(key, 60) # 续期 60s async def discover_tools( self, agent: AgentIdentity, tags: Optional[list[str]] None ) - list[dict]: Agent 发现可用工具。 过滤条件 1. status ACTIVE排除已下线/废弃的 2. 工具标签匹配 agent 的需求 3. Agent 有该工具的调用权限 result [] # 获取所有工具——实际生产环境应该用 SCAN 分批以避免阻塞 cursor 0 while True: cursor, keys await self._redis.scan(cursor, matchtool:*, count100) for key in keys: tool_data await self._redis.hgetall(key) tool_data {k.decode(): v.decode() for k, v in tool_data.items()} # 状态过滤 if tool_data.get(status) ! ToolStatus.ACTIVE.value: continue tool_name tool_data[name] tool_tags json.loads(tool_data.get(tags, [])) # 权限检查 if not self._has_permission(agent, tool_name): continue # 标签过滤 if tags and not any(t in tool_tags for t in tags): continue result.append({ name: tool_name, schema: json.loads(tool_data[schema]), version: tool_data[version], endpoint: tool_data.get(endpoint, ), }) if cursor 0: break return result def _has_permission(self, agent: AgentIdentity, tool_name: str) - bool: 权限验证。 支持通配符模式agent 拥有 email:* 权限 则可以调用 email.smtp.send、email.template.render 等。 for perm in agent.permissions: if perm.endswith(:*): prefix perm[:-1] if tool_name.startswith(prefix): return True elif perm tool_name: return True return False async def get_tool_status(self, tool_name: str) - Optional[dict]: 获取单个工具的详细信息。 key ftool:{tool_name} data await self._redis.hgetall(key) if not data: return None return {k.decode(): v.decode() for k, v in data.items()} class ToolHealthChecker: 工具健康检查器。 后台协程定期检查工具心跳是否超时自动将离线工具标记为 OFFLINE。 def __init__(self, registry: CapabilityRegistry, check_interval: int 10): self._registry registry self._interval check_interval async def run(self): while True: # 实际实现需遍历所有工具检查 last_heartbeat 是否超过 TTL await asyncio.sleep(self._interval)四、架构权衡与适用场景集中式 vs 去中心化这里选择集中式注册中心因为Agent 和工具的数量在百级以内Redis 单实例足够集中式更容易做审计和权限控制去中心化gossip 协议在这种规模下引入了不必要的复杂度性能边界Redis 单实例支持 10 万 QPS工具发现操作足够如果工具数量 10 万需要引入分层——按标签/域拆注册中心不适合的场景单体 Agent 3 个工具注册中心是过度设计离线/边缘 Agent无法访问中心注册中心的场景五、总结Agent 能力注册中心是把微服务治理的经验平移到 AI Agent 领域。核心产出是三件事服务发现Agent 知道有哪些工具可用、契约管理工具 API Schema、健康保障自动摘除故障工具。有了注册中心Agent 的工具管理从硬编码升级为动态发现——这是 Agent 从玩具走向生产的第一步。