Agent 框架中通常如何实现 Skill 的注册与发现机制？

面试官：“在搭建 Agent 框架时，Skill 的注册与发现机制通常是怎么实现的？”

候选人：

“Skill 的注册与发现本质上是解决一个问题：Agent 怎么知道有哪些 Skill 可用，并在需要时快速找到最合适的那个。我实现过几种方案，从简单到复杂，核心都是围绕‘元信息的管理和匹配’展开。

---

方案一：本地注册 + 配置驱动

这是最简单的方案，适合 Skill 数量少、单机运行的场景。

实现方式：

• 开发者通过配置文件、注解或装饰器声明 Skill。比如 Python 里用 @register_skill(name="send_email") 装饰一个 Skill 类。

• 系统启动时扫描指定目录或模块，收集所有带注解的 Skill 类。

• 提取每个 Skill 的元信息（名称、描述、参数 schema、触发示例等），加载到内存的一个注册表（通常是字典或 HashMap）。

• Agent 运行时直接查这张内存表做匹配。

# 示例：基于装饰器的注册
class SkillRegistry:
    _skills = {}

    @classmethod
    def register(cls, name, description, trigger_examples):
        def wrapper(skill_cls):
            cls._skills[name] = {
                "class": skill_cls,
                "description": description,
                "trigger_examples": trigger_examples
            }
            return skill_cls
        return wrapper

    @classmethod
    def find_best_match(cls, user_intent):
        # 用嵌入模型计算相似度，返回最匹配的 Skill
        pass

优点：零依赖，启动快，调试方便。

缺点：Skill 数量多了后内存压力大，不支持动态增减，也无法跨进程共享。

---

方案二：注册中心 + 服务发现

当 Skill 数量达到几十上百，或者需要跨多个 Agent 实例共享时，就需要引入独立的注册中心。

架构设计：

• Skill Provider：每个 Skill 作为一个独立服务部署，启动时向注册中心上报自己的元信息，并定期发送心跳维持租约。

• Registry Center：用 etcd、Consul、Nacos 或 Redis 来存储 Skill 的元信息。数据结构可以是一个 Hash，Key 是 Skill 名称，Value 是元信息 JSON。

• Agent 侧 Discovery：Agent 启动时从注册中心拉取全量 Skill 列表缓存到本地，同时订阅变更通知，Skill 上下线时实时更新本地缓存。

# 基于 Redis 的注册与发现简化实现
class RedisSkillRegistry:
    def register(self, skill_meta):
        redis.hset("skills", skill_meta["name"], json.dumps(skill_meta))
        redis.publish("skill:update", skill_meta["name"])  # 广播变更

    def discover_all(self):
        skills = redis.hgetall("skills")
        return [json.loads(v) for v in skills.values()]

    def watch_changes(self, callback):
        pubsub = redis.pubsub()
        pubsub.subscribe("skill:update")
        for msg in pubsub.listen():
            callback(msg)  # 更新本地缓存

优点：支持动态上下线，多实例共享，高可用。

缺点：引入外部依赖，增加运维复杂度，网络延迟可能影响匹配速度。

---

方案三：基于向量数据库的语义发现

当 Skill 数量更大（几百上千），或者触发条件非常依赖语义理解时，简单的元信息匹配不够用，需要上向量化检索。

实现方式：

• 每个 Skill 注册时，把它的描述和触发示例用嵌入模型（如 text-embedding-3-small）转成向量。

• 向量存储在 Milvus、Pinecone 或 pgvector 等向量数据库中。

• Agent 收到用户意图后，同样用嵌入模型把意图转成向量。

• 在向量数据库中做 ANN（近似最近邻）搜索，返回 Top-K 个语义最相似的 Skill。

• 再结合参数匹配、权限校验等规则做精排，选出最终要激活的 Skill。

# 基于向量检索的 Skill 匹配
class VectorSkillDiscovery:
    def register(self, skill_meta):
        embedding = embedder.encode(skill_meta["description"])
        vector_db.insert(
            name=skill_meta["name"],
            embedding=embedding,
            metadata=skill_meta
        )

    def discover(self, user_intent, top_k=5):
        intent_embedding = embedder.encode(user_intent)
        results = vector_db.search(intent_embedding, top_k=top_k)
        return self.rerank(results, user_intent)  # 结合规则精排

优点：语义匹配精准度高，支持大规模 Skill 库，可处理模糊意图。

缺点：依赖嵌入模型和向量数据库，增加系统复杂度和成本。

---

方案四：分层混合发现

实际生产环境往往是上面几种方案的组合，我画一个分层架构：

用户意图
   │
   ▼
第一层：标签/分类粗筛（内存缓存）
   └── 从几百个 Skill 快速缩小到十几个候选
   │
   ▼
第二层：语义向量匹配（向量数据库）
   └── 从十几个候选中根据语义相似度选出 Top-3
   │
   ▼
第三层：参数匹配 + 前置条件校验（规则引擎）
   └── 检查必填参数是否齐全、权限是否满足，选出最终 Skill

---