引言

最近 GitHub 上一个名为 agent-skills 的开源项目火了——4600+ 颗星,last 30 days trending #1。它不是又一个 LLM 封装框架,而是一套AI Agent 技能系统的完整设计范式

简单说:它让 AI Agent 不再是"只会对话"的聊天机器人,而是能像游戏角色一样学习、组合、执行各种技能的智能体。本文从源码层面拆解它的核心设计。

一、整体架构:Skills Registry 模式

项目核心是一个 SkillRegistry,相当于 Agent 的"技能书"。所有技能统一注册、统一调度:

from agent_skills import SkillRegistry, BaseSkill

registry = SkillRegistry()

@registry.register("web_search")
class WebSearchSkill(BaseSkill):
    """网页搜索技能"""
    pass

源码行不多,核心只有 3 个模块:

| 模块 | 职责 |

|------|------|

| registry.py | 技能注册中心、依赖注入 |

| executor.py | 技能执行引擎、错误处理 |

| skills/ | 内置技能集合 |

关键设计决策:技能 = 代码,不是 prompt。每个技能是一个 Python 类,有 can_handle() 判断触发条件和 execute() 执行逻辑,比纯 prompt 方案更可控。

二、SkillRegistry 源码拆解

class SkillRegistry:
    def __init__(self):
        self._skills: dict[str, type[BaseSkill]] = {}
    
    def register(self, name=None):
        """装饰器:注册技能"""
        def decorator(cls):
            skill_name = name or cls.__name__.lower()
            self._skills[skill_name] = cls
            return cls
        return decorator
    
    def get_skill(self, name: str) -> BaseSkill:
        cls = self._skills.get(name)
        if not cls:
            raise SkillNotFoundError(f"技能 '{name}' 未注册")
        return cls()
    
    def search(self, query: str, top_k: int = 3) -> list[BaseSkill]:
        """根据语义搜索匹配技能(基于描述文本的简单关键词匹配)"""
        scored = []
        for name, cls in self._skills.items():
            desc = getattr(cls, 'description', '') or cls.__doc__ or ''
            score = self._similarity(query, desc)
            scored.append((score, name, cls))
        scored.sort(reverse=True)
        return [cls() for _, _, cls in scored[:top_k]]

这个模式的好处是:技能与调用解耦。Agent 收到任务后,用 search() 自动匹配技能,不需要硬编码 if-else 调用链。

三、BaseSkill 的设计哲学

class BaseSkill:
    name: str = ""
    description: str = ""
    requires: list[str] = []  # 前置技能依赖
    timeout: int = 30

    def can_handle(self, task: str) -> bool:
        """判断能否处理该任务"""
        return False
    
    async def execute(self, context: dict, **kwargs) -> dict:
        """执行技能,返回结果"""
        raise NotImplementedError

设计亮点:

  • **`requires` 依赖注入**——技能可以依赖其他技能,Registry 自动解析 DAG
  • **统一的 `execute` 签名**——所有技能收到同一个 context 字典,包含当前会话状态、历史、工具
  • **异步原生**——IO 密集型技能(搜索、爬虫)不阻塞主流程

实际内置技能示例(skills/web_search.py):

@registry.register("web_search")
class WebSearchSkill(BaseSkill):
    name = "web_search"
    description = "执行网络搜索,返回前N条结果"
    requires = ["http_client"]
    
    async def execute(self, context, query: str = "", max_results: int = 5):
        client = context["tools"]["http_client"]
        results = await client.search(query, max_results)
        return {"results": results, "count": len(results)}

四、Executor 引擎:技能编排

真正有意思的是 executor.py。它不只是一个"调函数"的工具,而是带状态管理的执行引擎:

class SkillExecutor:
    def __init__(self, registry: SkillRegistry):
        self.registry = registry
        self._history: list[dict] = []
    
    async def execute_plan(self, plan: list[dict], context: dict) -> list[dict]:
        """按计划依次执行技能"""
        results = []
        for step in plan:
            skill_name = step.get("skill")
            params = step.get("params", {})
            try:
                skill = self.registry.get_skill(skill_name)
                # 解析依赖(前置技能)
                for dep in skill.requires:
                    dep_result = context.get("_deps", {}).get(dep)
                    if not dep_result:
                        dep_skill = self.registry.get_skill(dep)
                        dep_result = await dep_skill.execute(context)
                        context.setdefault("_deps", {})[dep] = dep_result
                
                result = await skill.execute(context, **params)
                self._history.append({"skill": skill_name, "result": result})
                results.append(result)
            except Exception as e:
                results.append({"error": str(e), "skill": skill_name})
                # 非关键技能失败不阻断整体流程
                if step.get("critical", False):
                    raise
        return results

execute_plan 实现了有序技能链:一个任务可以被拆解为多步计划,每步调用不同技能,上一步输出自动注入下一步 context。这是 Agent 从"问答"走向"任务执行"的关键一步。

五、生产环境实战建议

如果你想把 agent-skills 用在真实项目中,这几点值得注意:

5.1 技能粒度控制

每个技能做一件事,做精。示例项目中 web_search 只负责搜索,内容提取由 page_reader 技能处理。拆分让技能可复用。

5.2 超时和退避

async def execute(self, context, **kwargs):
    for attempt in range(3):
        try:
            return await asyncio.wait_for(
                self._do_search(kwargs["query"]), timeout=10
            )
        except asyncio.TimeoutError:
            if attempt == 2:
                return {"error": "search timeout after 3 retries"}
            await asyncio.sleep(2 ** attempt)

5.3 技能缓存

高频技能(如 get_current_timelist_files)可以加内存缓存,避免重复执行:

class CachedSkill(BaseSkill):
    _cache: dict = {}
    cache_ttl: int = 60
    
    async def execute(self, context, **kwargs):
        key = f"{self.name}:{hash(frozenset(kwargs.items()))}"
        if key in self._cache:
            cached_at, value = self._cache[key]
            if time.time() - cached_at < self.cache_ttl:
                return value
        result = await self._execute_impl(context, **kwargs)
        self._cache[key] = (time.time(), result)
        return result

总结

agent-skills 4600+ 星不是白来的。它用清晰的注册模式、统一的执行接口、解耦的技能依赖设计,解决了 AI Agent 落地中最实际的"技能管理"问题。

与其说它是一个框架,不如说是一种设计范式——把"能力"建模为可注册、可搜索、可组合的技能单元。对于正在做 AI Agent 开发的团队来说,这个 repo 值得仔细读一遍源码,一共不到 2000 行,精华全在模式上。

源码地址: https://github.com/nicherahul/agent-skills(建议 star 后读代码,比读文档管用)

如果觉得有用,欢迎点赞收藏!

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐