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_factory=list) owner: str = "" ttl_seconds: int = 30 # 心跳超时 @dataclass class AgentIdentity: agent_id: str role: str permissions: list[str] = field(default_factory=list) 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 = f"tool:{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, mapping=tool_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 = f"tool:{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, match="tool:*", count=100) 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 = f"tool:{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 从玩具走向生产的第一步。
