GitHub 4600+星!agent-skills源码级拆解:AI编码Agent的技能系统到底怎么设计的?
引言
最近 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_time、list_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 后读代码,比读文档管用)
如果觉得有用,欢迎点赞收藏!
更多推荐


所有评论(0)