AI Agent技能编排实战：从Function Calling到统一注册与路由

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”的体系不需要随仓库的依赖走我自己实验时用的是下面这套组合稳定性和可复现性都在可控范围内组件版本建议作用说明Python3.10目前主流LLM框架均已支持类型标注更完善FastAPI0.104用于技能API服务化封装Pydantic2.x参数Schema定义与校验OpenAI SDK / Anthropic SDK最新稳定版模型调用层负责意图识别与槽位填充PostgreSQL pgvector14 / 0.5技能描述向量化检索作为动态技能发现的基础Redis7.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_idweather_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]}, timeout10 ) 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(fskills.{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_k5) # 2. 构建提示词并调用LLM prompt build_skill_prompt(candidate_skills) response await self.llm.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: prompt}, {role: user, content: user_message}, ], temperature0, max_tokens1000, ) 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)}这里的temperature0是刻意设置。编排环节不需要创造性只需要确定性——哪怕牺牲一点表达多样性也要确保同一个用户输入在相同上下文下产生一致的动作序列。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_idabc123 stagellm_output cost420ms raw{\tool_calls\:[...]} [SKILL_TRACE] req_idabc123 stagevalidation skillsweather_query statuspass [SKILL_TRACE] req_idabc123 stageexecution skillweather_query statusok cost86ms排查问题时直接按req_id聚合效率会高非常多。7.3 大规模技能的动态加载策略当技能数量超过几个量级之后启动时全量扫描注册会变得不可接受。可以把“注册中心”改成“注册中心 懒加载”的模式启动时只加载技能元数据spec和描述不加载执行类只有在真正被路由命中的时候才动态import对应的模块。这样既保持了全量技能的索引能力又把启动时间控制在毫秒级别。8. 一点经验之外的话前后踩了几个项目迭代的坑之后最大的感受是这个项目的设计哲学值得单独拎出来说。很多人做Agent时直觉是“让模型自由发挥”但真正可上线的Agent恰恰需要的是约束和边界。技能定义处理解层、协议层、执行层三层拆开之后每一层都可以独立迭代、独立测试这个分层价值在多技能场景下才会真正显现但等你发现问题再回头重构成本起码是开始就做设计的五倍。如果你打算在这个方向上继续探索我建议下一步把精力放在“技能评测”上。准备一组用户问题的黄金测试集每次技能描述或参数结构变更后自动跑一遍回归对比编排输出的正确率。有了这套评测机制后续的优化才能有理有据。这个内容后续还可以扩展的方向包括技能间的依赖编排、多Agent环境下的技能共享、技能质量自动评估工具链。我自己正在做的是把搜索召回的部分换成Rerank让候选技能排序更准之后有结论再单独写一篇和大家聊。