1. 项目概述与设计思路拆解
1.1 到底在解决什么问题
接触过AI Agent开发的读者应该都有体会:让大模型写一段对话、生成一篇文案,这很容易;但如果你希望它操作某个系统、调用某个API、完成一个多步骤的工作流,事情就开始变得棘手。传统的做法是把每个功能封装成一个函数,在prompt里塞一堆“你能调用这些工具”的说明,然后靠模型自行决定什么时候调用哪个。小规模场景下面勉强能跑通,但一旦技能数量超过20个、参数规则稍微复杂一点,你就开始频繁踩坑——模型要么在错误的时机调用了错误的技能,要么干脆对着参数说明瞎编了一个JSON。
“agent-skills”这个项目解决的核心问题就在于此:如何让Agent的技能具备统一的描述规范、清晰的调用机制、可控的执行流程。它做的不是某一个具体的业务技能,而是一套技能管理的骨架。你可以把这套骨架理解为给Agent装了一个“标准接口的插线板”——不同的技能就像不同的电器插头,只要遵循同样的接口规范,插上就能用,换一个Agent环境也能复用。
这听起来像是对“Function Calling”的一种变相封装,但实际上它走得更远,它把“技能描述”“参数Schema”“执行逻辑”“技能间的依赖关系”整个生命周期都统一纳入了管理。我最初接触这个项目时,最大的感受是它把很多从业者平时靠经验“硬编码”在代码里的那些约定,抽象成了显式的配置与路由机制,这让Agent技能体系变得可维护、可扩展,也更容易团队成员协作。
1.2 面向的核心用户与实际场景
这个项目最直接的受众是两类人:一类是在做企业级Agent应用开发的工程师,他们手里的技能列表已经膨胀到难以用if-else管理;另一类是AI产品经理或解决方案架构师,他们需要评估“如何把公司现有系统的能力合理拆解并开放给AI调用”。
在具体场景上,我试过三种比较典型的方向:
- 企业内部知识库问答Agent:把“查询文档”“拉取工单”“汇总周报”等操作统一注册为技能,让模型按需组合调用。
- 自动化运维助手:将服务器状态检查、日志分析、告警确认等运维操作封装成技能,通过自然语言指令触发。
- 个人助手类应用:把日程管理、邮件起草、待办整理等轻量功能做成技能集,跨应用复用。
在这些场景中,技能数量越多,“agent-skills”这类分层设计带来的收益就越明显。而且项目本身不是某种商业产品的附属模块,它是一套设计模式加参考实现,你可以把它嵌入自己的项目,不必绑定特定的大模型供应商。
2. 核心架构与关键概念拆解
2.1 技能的定义本身就是一个协议
在动手之前,我最先关注的是“一个技能到底长什么样”。这个项目里,技能不再是一个裸函数,而是三个部分的集合:
- 描述层:技能的用途说明、适用场景、调用限制,以及触发条件的自然语言描述。
- 协议层:输入参数的JSON Schema定义、输出结果的结构约定、错误码定义。
- 执行层:实际完成业务逻辑的代码,可能是本地函数、远程API,甚至是一个子工作流。
如果你熟悉OpenAPI规范,会发现这个分层颇有几分相似。但它更贴合Agent场景,因为描述层不仅仅是为了人类阅读,本质上是在给大模型“读”——模型通过阅读技能描述来决定是否调用、何时调用。所以描述怎么写得准确,往往是决定整个Agent效果的关键变量。
从这一点延伸出去,项目还规定了一套“技能注册”的机制。每个技能在运行时会注册到一个统一的仓库里,并生成一个全局唯一标识符。这一步看起来多此一举,但实际意义非常深远:有了这个仓库,你才可能实现后续的动态加载、热更新、权限隔离,这些都是在真实项目中绕不开的能力。
2.2 编排调度是技能的“大脑”
光有一堆定义良好的技能还不行,真正的Agent能力取决于“怎么把它们串起来”。这个项目在编排层用了一种我非常认可的思路:把“技能调用”从模型直接生成JSON的不可控状态,改为“槽位填充”式的半受控状态。
什么叫“槽位填充”?比如你要实现一个“查询天气并安排提醒”的需求,系统内部会定义一个工作流模板,包含两个槽位(查询城市的参数、提醒时间的参数),模型的任务不是决定“要不要调用两个独立的技能”,而是根据用户的话把这个模板里的槽位填好,然后由编排引擎按顺序执行。这种设计大幅降低了多技能协同时的出错率,因为模型面对的任务复杂度从“全文生成”降到了“信息提取”,这在业界已经是被验证非常有效的稳定性方案。
好,废话不多说,直接进入实操环节。下面的内容覆盖环境搭建、核心代码写作、路由逻辑、避坑经验四块,每一块我都会把这里面的“所以然”讲清楚。
3. 环境准备与项目初始化
3.1 依赖选型与版本建议
如果你要自己从零搭建一套类似“agent-skills”的体系,不需要随仓库的依赖走,我自己实验时用的是下面这套组合,稳定性和可复现性都在可控范围内:
| 组件 | 版本建议 | 作用说明 |
|---|---|---|
| Python | 3.10+ | 目前主流LLM框架均已支持,类型标注更完善 |
| FastAPI | 0.104+ | 用于技能API服务化封装 |
| Pydantic | 2.x | 参数Schema定义与校验 |
| OpenAI SDK / Anthropic SDK | 最新稳定版 | 模型调用层,负责意图识别与槽位填充 |
| PostgreSQL + pgvector | 14+ / 0.5+ | 技能描述向量化检索,作为动态技能发现的基础 |
| Redis | 7.x | 技能调用缓存、分布式锁 |
这里有一点必须提醒:技能仓库中的描述文本如果很长,每次系统启动都加载全部文本会非常慢。把技能描述embedding成向量存到pgvector里,运行时根据用户输入做相似度检索召回候选技能,开销会降一个数量级,是规模化后的必选项。
3.2 初始化操作步骤
我推荐用uv来管理依赖,它的解析速度和缓存机制比pip好很多:
# 创建虚拟环境并安装依赖 uv venv .venv source .venv/bin/activate uv pip install fastapi pydantic openai anthropic sqlalchemy "pgvector" redis "uvicorn[standard]"接下来创建项目的基础目录结构。这一步很多人会随意为之,但目录结构的清晰程度,直接决定技能多了以后你还能不能管得住:
agent_skills/ ├── skills/ # 技能代码目录,每个技能一个子目录 ├── registry/ # 技能注册中心代码 ├── orchestrator/ # 编排引擎代码 ├── schemas/ # 公共数据结构定义 └── runtime/ # 运行时环境与主入口4. 技能定义与注册机制实现
4.1 技能基类的设计取舍
我把“agent-skills”的核心理念落实成一个最小的技能基类。这个基类不需要很复杂,但要把“描述、协议、执行”三件事固化下来:
# schemas/skill_base.py from abc import ABC, abstractmethod from typing import Any, Dict, Optional from pydantic import BaseModel, Field class SkillParameter(BaseModel): """技能参数协议层定义""" name: str type: str # string, integer, boolean, object description: str required: bool = True enum: Optional[list] = None default: Optional[Any] = None class SkillSpec(BaseModel): """技能描述层定义""" skill_id: str = Field(description="全局唯一技能ID") name: str = Field(description="技能名称") description: str = Field(description="给LLM看的触发条件与用途说明") version: str = "1.0.0" parameters: list[SkillParameter] = [] timeout_seconds: int = 30 tags: list[str] = [] enabled: bool = True class BaseSkill(ABC): """所有技能必须继承的基类""" spec: SkillSpec def __init__(self): if not self.spec: raise ValueError(f"{self.__class__.__name__} 必须定义 spec") @abstractmethod async def execute(self, **kwargs) -> Dict[str, Any]: """真正的执行逻辑,由子类实现""" pass def validate_params(self, params: dict) -> Dict[str, Any]: """基于协议层做输入参数校验""" validated = {} for p in self.spec.parameters: if p.name not in params: if p.required: raise ValueError(f"缺少必填参数: {p.name}") else: validated[p.name] = p.default else: validated[p.name] = params[p.name] return validated这套基类的设计核心在于:子类只关心execute里怎么写业务逻辑;协议和参数校验被彻底抽离。这带来一个实际好处——子类代码里不会出现if "city" not in input_params这种人人都写过又想删的代码。
4.2 实现一个具体技能:查天气
用一个最常见的“天气查询”技能来演示完整写法,顺便让大家直观理解描述层的重要性:
# skills/weather.py import httpx from schemas.skill_base import BaseSkill, SkillSpec class WeatherSkill(BaseSkill): spec = SkillSpec( skill_id="weather_query", name="天气查询", description=( "当用户询问某个城市当前天气、温度、风力或降雨概率时使用。" "注意:如果用户表达的是“明天/后天”等未来时间,则不要使用本技能," "应该改用 weather_forecast 技能。" ), parameters=[ {"name": "city", "type": "string", "description": "用户询问天气的城市名称", "required": True}, {"name": "unit", "type": "string", "description": "温度单位,celsius或fahrenheit", "required": False, "enum": ["celsius", "fahrenheit"], "default": "celsius"}, ], ) async def execute(self, **kwargs): params = self.validate_params(kwargs) # 这里接入的是mock数据源,实际项目换成你所在平台的天气API即可 async with httpx.AsyncClient() as client: resp = await client.get( "https://api.example-weather-service.com/v1/current", params={"city": params["city"]}, timeout=10 ) resp.raise_for_status() data = resp.json() return { "city": params["city"], "temperature": data["temperature"], "wind_level": data.get("wind_level", "未知"), "condition": data.get("condition", "未知"), }想特别强调的是description的写法。我见过非常多新手写的技能描述就是一句话“查询天气”,这对大模型来说信息量严重不足。比较靠谱的写法应该包含四类信息:
- 触发条件:什么情况下该用这个技能。
- 排除条件:什么情况下不该用它(这是一个反直觉但极其有效的写法)。
- 参数抽取提示:从用户哪类话术里抽取参数。
- 边界说明:时间跨度、地点范围等限制。
4.3 注册中心与生命周期管理
有了技能类和具体实现后,需要一个地方把全部技能管起来。我实现了一个轻量的注册中心:
# registry/skill_registry.py import importlib import pkgutil import inspect from typing import Dict, Type from schemas.skill_base import BaseSkill class SkillRegistry: def __init__(self): self._skills: Dict[str, BaseSkill] = {} self._skill_versions: Dict[str, str] = {} def register(self, skill: BaseSkill): skill_id = skill.spec.skill_id self._skills[skill_id] = skill self._skill_versions[skill_id] = skill.spec.version print(f"[Registry] 技能已注册: {skill_id}@{skill.spec.version}") def unregister(self, skill_id: str): self._skills.pop(skill_id, None) self._skill_versions.pop(skill_id, None) def get(self, skill_id: str) -> BaseSkill: return self._skills[skill_id] def discover_all(self): """ 自动扫描skills目录下所有模块,并注册其中的BaseSkill子类 """ import skills for module_info in pkgutil.iter_modules(skills.__path__): module = importlib.import_module(f"skills.{module_info.name}") for _, obj in inspect.getmembers(module, inspect.isclass): if ( issubclass(obj, BaseSkill) and obj is not BaseSkill and obj.__module__ == module.__name__ ): self.register(obj()) registry = SkillRegistry()这段代码里有几个容易被忽略的点:
inspect.getmembers配合issubclass判断是自动注册的核心,但必须加一个obj is not BaseSkill的排除,避免把抽象基类也注册进去。- 同名技能重复注册会静默覆盖,这个问题在真实项目里很容易引起线上故障,建议在register里做版本冲突检测,同skill_id不同version时明确抛异常。
5. 编排引擎与智能路由实现
5.1 LLM调用的消息协议设计
技能注册好了,接下来最关键的环节是:怎么让大模型“知道”有哪些技能,并且“学会”正确使用它们。这里我采用“系统提示词 + 技能清单 + 强制输出格式”的三段式方法:
# orchestrator/prompt_builder.py def build_skill_prompt(skills: list) -> str: lines = [] lines.append("你是一个技能编排助手,你的任务是根据用户需求,从以下技能中选择合适的技能并填好参数。") lines.append("你必须严格遵守以下规则:") lines.append("1. 只能调用给出的技能,不能虚构技能名。") lines.append("2. 如果用户需求与任何技能都不匹配,返回空数组。") lines.append("3. 不要解释,只输出JSON格式。") lines.append("") lines.append("可用技能列表:") for skill in skills: spec = skill.spec lines.append(f"- 技能ID: {spec.skill_id}") lines.append(f" 名称: {spec.name}") lines.append(f" 描述: {spec.description}") lines.append(f" 参数定义: {spec.parameters}") lines.append("") lines.append("输出格式:") lines.append("{\"tool_calls\": [{\"skill_id\": \"技能ID\", \"params\": {\"参数名\": 参数值}}]}") return "\n".join(lines)这里有一个关键细节:不要把全部技能都塞进prompt。当技能库膨胀到几十上百个时,token开销和模型的注意力分散会成为大问题。我的做法是分两级——先根据用户输入做一次向量检索,召回top 5候选技能,再把候选清单塞进prompt。这套路线上文提到过,用pgvector存储描述embedding,查询时走余弦距离度量。
5.2 路由解析与参数校验闭环
大模型输出JSON之后,要经过严格的协议校验才能进入执行阶段。我为这个环节单独写了一个模块:
# orchestrator/router.py import json from jsonschema import validate from jsonschema.exceptions import ValidationError class SkillRouter: def __init__(self, registry): self.registry = registry def parse_llm_output(self, raw_output: str) -> list: """解析模型输出,兼容markdown代码块包裹的情况""" text = raw_output.strip() if text.startswith("```"): text = text.split("\n", 1)[1].rsplit("```", 1)[0] return json.loads(text).get("tool_calls", []) def route(self, tool_calls: list) -> list: results = [] for call in tool_calls: skill_id = call.get("skill_id", "") skill = self.registry.get(skill_id) if not skill: results.append({"skill_id": skill_id, "status": "error", "error": "技能不存在"}) continue try: params = skill.validate_params(call.get("params", {})) result = skill.execute(**params) results.append({"skill_id": skill_id, "status": "ok", "result": result}) except Exception as e: results.append({"skill_id": skill_id, "status": "error", "error": str(e)}) return resultsparse_llm_output里的markdown兼容处理,看起来是个微不足道的细节,实际作用巨大。因为很多模型即使你强调“只输出JSON”,它仍然会给你包上\``json`代码块,不处理就会直接json.loads失败。这几个字符的兼容成本极低,收益却极高。
5.3 带校验的编排执行示例
我把编排引擎做成一个异步管道模式。管道的好处是每个阶段的输入输出都明确,方便以后插入“人工审核”“日志审计”等中间环节:
# orchestrator/pipeline.py import asyncio from orchestrator.prompt_builder import build_skill_prompt from orchestrator.router import SkillRouter from vector_store.skill_search import SkillSearch class SkillPipeline: def __init__(self, llm_client, registry, skill_search: SkillSearch): self.llm = llm_client self.router = SkillRouter(registry) self.search = skill_search async def run(self, user_message: str) -> dict: # 1. 向量检索召回候选技能 candidate_skills = self.search.retrieve(user_message, top_k=5) # 2. 构建提示词并调用LLM prompt = build_skill_prompt(candidate_skills) response = await self.llm.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": prompt}, {"role": "user", "content": user_message}, ], temperature=0, max_tokens=1000, ) raw_output = response.choices[0].message.content # 3. 解析并校验 tool_calls = self.router.parse_llm_output(raw_output) if not tool_calls: return {"status": "no_skill", "message": "没有匹配到任何技能"} # 4. 执行技能(并发执行) results = await asyncio.gather( *[self._execute_one(call) for call in tool_calls] ) return {"status": "ok", "results": results} async def _execute_one(self, call): skill_id = call.get("skill_id") skill = self.registry.get(skill_id) if not skill: return {"skill_id": skill_id, "status": "error", "error": "技能不存在"} try: params = skill.validate_params(call.get("params", {})) result = await skill.execute(**params) return {"skill_id": skill_id, "status": "ok", "result": result} except Exception as e: return {"skill_id": skill_id, "status": "error", "error": str(e)}这里的temperature=0是刻意设置。编排环节不需要创造性,只需要确定性——哪怕牺牲一点表达多样性,也要确保同一个用户输入在相同上下文下产生一致的动作序列。
6. 常见问题与排查技巧实录
6.1 模型总是选错技能怎么办
这个问题出现的频率最高。我排查时一般按下面这个顺序走:
- 先看技能的description是否清晰区分了场景边界。比如你有“查天气”和“查历史天气”两个技能,如果描述都写着“查询天气”,模型当然分不清。必须显式写明“当前天气用A,历史或预报数据用B”。
- 再看参数命名是否直观。参数名不要用缩写,尽量用完整业务词汇。
city_name比ct好得多。 - 最后看召回排序是否正确。如果向量检索召回的top技能里根本没有正确的那一个,那问题不在模型的调用能力,而在embedding与检索环节——检查一下技能的描述embedding是否与索引版本同步。
有一次我排查了很久,最后发现是技能描述改过但embedding没有重建,模型压根看不到新描述。这个坑极其隐蔽,建议在CI流程里加一个“描述变更自动重建embedding”的检查。
6.2 技能执行报错的熔断与降级
生产环境不可能每个技能都稳定返回。我一开始做的是“异常throw出来,由上层统一兜底”,后来发现这样做的结果是:一个技能的失败会导致整个Agent任务失败甚至卡死。
改成了更细粒度的失败策略:
| 失败类型 | 处理策略 | 说明 |
|---|---|---|
| 参数校验失败(缺少必填参数) | 返回给模型重新生成 | 话术里带上“需要补充参数XXX” |
| 技能执行超时 | 重试1次,仍失败则标记失败 | 超时时间从技能声明里读取 |
| 技能内部异常 | 返回错误信息,不重试 | 这类错误通常重试无意义 |
| 模型输出非合法JSON | 重新请求模型最多2次 | 并适当增强“只输出JSON”的提示 |
6.3 并发调用时的资源泄漏问题
当多个技能同时被触发时,如果技能内部使用了数据库连接、HTTP连接池,必须确保这些资源在技能对象内部是共享的,不能在execute里每次new一个连接。我见过有人在execute里频繁创建httpx.Client,在并发20路以上的时候直接打满文件描述符。
正确做法是让执行类在__init__阶段就初始化好线程池或连接池,execute只负责使用。如果技能涉及有状态操作(比如计数、限流),还要考虑加锁或用Redis原子操作,避免多协程交叉执行时产生脏数据。
7. 进阶扩展:让技能体系走向生产级
7.1 多租户隔离
如果你的Agent服务要支撑多个业务团队,技能权限隔离迟早是要面对的。比较轻量的方案是给每个技能加一个allowed_roles字段,在validate_params之前做一次角色判定:
def check_permission(skill_spec, user_roles): allowed = set(skill_spec.allowed_roles) if "admin" in user_roles: return True return bool(allowed & set(user_roles))权限失败时不要让模型“换一个技能”继续尝试,直接返回权限拒绝的错误,避免被恶意用户用prompt注入绕过。
7.2 技能调试的可观测性
技能执行链路通常跨“用户输入→模型召回→JSON解析→参数校验→业务执行→结果返回”五个环节,任何一个环节出问题都很难肉眼定位。我给代码埋了结构化日志,每个环节打一条记录:
[SKILL_TRACE] req_id=abc123 stage=llm_output cost=420ms raw="{\"tool_calls\":[...]}" [SKILL_TRACE] req_id=abc123 stage=validation skills=weather_query status=pass [SKILL_TRACE] req_id=abc123 stage=execution skill=weather_query status=ok cost=86ms排查问题时直接按req_id聚合,效率会高非常多。
7.3 大规模技能的动态加载策略
当技能数量超过几个量级之后,启动时全量扫描注册会变得不可接受。可以把“注册中心”改成“注册中心 + 懒加载”的模式:启动时只加载技能元数据(spec和描述),不加载执行类;只有在真正被路由命中的时候,才动态import对应的模块。这样既保持了全量技能的索引能力,又把启动时间控制在毫秒级别。
8. 一点经验之外的话
前后踩了几个项目迭代的坑之后,最大的感受是这个项目的设计哲学值得单独拎出来说。很多人做Agent时直觉是“让模型自由发挥”,但真正可上线的Agent恰恰需要的是约束和边界。技能定义处理解层、协议层、执行层三层拆开之后,每一层都可以独立迭代、独立测试,这个分层价值在多技能场景下才会真正显现,但等你发现问题再回头重构,成本起码是开始就做设计的五倍。
如果你打算在这个方向上继续探索,我建议下一步把精力放在“技能评测”上。准备一组用户问题的黄金测试集,每次技能描述或参数结构变更后自动跑一遍回归,对比编排输出的正确率。有了这套评测机制,后续的优化才能有理有据。
这个内容后续还可以扩展的方向包括:技能间的依赖编排、多Agent环境下的技能共享、技能质量自动评估工具链。我自己正在做的是把搜索召回的部分换成Rerank,让候选技能排序更准,之后有结论再单独写一篇和大家聊。