Agent技能层实战:从注册中心到调度器的完整实现

发布时间:2026/10/8 5:16:05
Agent技能层实战:从注册中心到调度器的完整实现 最近在做Agent应用落地时我越来越强烈地感受到一个瓶颈模型能力再强如果它手上没有趁手的工具很多任务根本执行不下去。这也是我把大量精力投入到agent-skills这个方向上的原因。简单来说它是一套让Agent具备可插拔、可复用、可管理的执行技能的系统化方案。这篇文章我会从项目设计思路、技能注册机制、核心代码实现到常见踩坑记录把整个搭建过程完整拆解一遍希望能给正在做Agent应用的同学一些实质性的参考。1. 项目概述与设计动机1.1 为什么Agent需要技能层先说你遇到的最典型场景你让一个Agent帮你查询数据库并生成报表它如果只会调用一个写死的函数那它就是一个死板的工具如果它需要什么功能都靠你在代码里临时加逻辑那它就是一座永远维护不完的屎山。我之前踩过很深的坑就是在大模型里塞了大量的system prompt去约束行为结果模型输出的工具调用格式稍微偏一点整个任务就崩了。后来想通了。Agent真正需要的不是一个一个零散的工具函数而是一层结构化的技能——每个技能都有自己的名称、描述、输入参数定义、执行逻辑Agent可以根据任务描述自主判断该用哪个技能技能层负责把模型输出的抽象意图翻译成具体可执行的代码。这层抽象有几个非常明显的好处技能与Agent解耦新增一个能力不需要改动Agent核心逻辑只需要注册一个新技能。模型调用更稳定模型只需从技能清单中挑选而不是从一大段自然语言提示词里猜。组合演化成为可能复杂任务可以通过多个基础技能的组合完成技能本身可以互相调用。1.2 这个方案能解决什么问题agent-skills这套体系我把它定位成一个轻量级的Agent技能编排框架。它解决的痛点非常具体当你需要让多个Agent共享一套能力库同时又想保持各Agent行为差异时你需要一个独立的技能层来解决这个问题。举个例子你有一个售前客服Agent和一个售后技术支持Agent它们都需要查订单、查库存、看物流状态但话术风格完全不同。如果把这些能力都写在各个Agent的prompt里改一处要同步好几处如果在技能层统一实现两个Agent只要各自声明我使用订单查询技能、库存查询技能剩下的事情交给技能层就行。这套方法尤其适合以下人群正在用LangChain、LlamaIndex或自建Prompt Pipeline做Agent应用的开发者想让多个Agent共享工具能力、但不想忍受重复代码的后端工程师对模型工具调用稳定性不满意、希望用工程手段兜底的产品技术负责人2. 核心设计思路技能注册与调度的取舍2.1 注册中心模式让Agent看见技能在设计上我没有做特别复杂的东西核心就是一套技能注册中心Skill Registry。每个技能本质上是一个简单粗暴的接口。我们用Python里的dataclass来定义技能元信息结构清晰且方便序列化。每个技能需要给Agent提供三类信息name技能的唯一标识Agent在输出工具调用时靠它来指定目标description用自然语言描述技能的作用模型靠它来匹配任务意图parametersJSON Schema格式的参数定义告诉模型调用时需要传哪些字段这部分如果你用Pydantic的话会更顺手因为它能直接生成JSON Schema。我当时没有引入额外重量级依赖用dataclass加手写Schema也完全够用逻辑更透明。这个阶段最重要的工作是把技能描述写清楚因为它决定了模型能不能在正确的时候把技能找出来。2.2 技能执行器的统一抽象注册中心解决的是有什么技能的问题接下来任务就是解决技能怎么执行的问题。我统一规定了一个BaseSkill接口所有技能必须实现execute方法。from abc import ABC, abstractmethod from typing import Any, Dict class BaseSkill(ABC): name: str description: str parameters: Dict[str, Any] {} abstractmethod def execute(self, **kwargs) - Any: 执行技能的具体逻辑 pass这层抽象有个好处就是执行细节完全隐藏在技能内部。比如天气查询技能内部可能是请求一个HTTP API也可能是读一份静态文件对Agent来说这些都是透明的它只需要知道输入城市名称输出天气信息就够了。2.3 自动感知调度真正让模型决定用什么技能整个设计里最关键的环节就是技能调度器。它的职责简单说就一句话把一组技能的系统提示词打包发给模型等模型返回结构化的工具调用再找出对应的技能执行器去跑。为了让模型稳定地输出调用意图我做了一个非常有效的尝试——把技能列表精简后塞进工具列表def build_system_skills_prompt(skills): lines [你是一个智能助手在合适的时候使用以下技能完成任务, ] for skill in skills: lines.append(f- {skill.name}: {skill.description}) return \n.join(lines)注意这里千万不要一股脑把所有技能都塞进去。我实测过当技能数量超过15个时模型混淆同名参数的情况会明显增加。这个阈值关系到后面的技能检索层。3. 功能模块拆解与实操实现3.1 从工具升级为技能的关键三步把一个普通函数包装成Agent能识别和调用的技能我在实际操作中总结出三个标准步骤定义Schema - 编写执行逻辑 - 注册进中心。以订单查询为例你可以这样设计class QueryOrderSkill(BaseSkill): name query_order description 根据订单号查询订单状态和物流信息输入参数为order_id parameters { type: object, properties: { order_id: { type: string, description: 订单编号格式如 SO20250101 } }, required: [order_id] } def execute(self, order_id: str) - Dict[str, Any]: # 模拟查询逻辑实际项目应调用订单服务 if not order_id.startswith(SO): raise ValueError(无效的订单号格式) return { order_id: order_id, status: paid, logistics: 已出库预计3天后送达 }这里有个细节值得我们注意参数里的description字段很多初学者会忽略但它恰恰是影响模型调用质量的重要因素。模型不是程序员它不会去看函数内部实现它只能通过这段描述来理解该传什么。比如你在order_id的描述里写清楚格式规范模型就会倾向于生成符合格式的值而不是凭空造一个。3.2 技能注册中心的完整实现注册中心我用了Python里非常常见的类装饰器模式让它用起来像Flask注册路由一样简洁顺手。核心代码如下class SkillRegistry: def __init__(self): self._skills: Dict[str, BaseSkill] {} def register(self, skill_cls): skill skill_cls() self._skills[skill.name] skill return skill_cls def get(self, name: str) - BaseSkill: return self._skills.get(name) def all(self): return list(self._skills.values()) skill_registry SkillRegistry() skill_registry.register class WeatherSkill(BaseSkill): name get_weather description 根据城市名称获取当前的天气状况 parameters { type: object, properties: { city: {type: string, description: 城市名如北京、上海} }, required: [city] } def execute(self, city: str): return {city: city, weather: 晴, temperature: 23}使用装饰器注册的好处是逻辑集中且可扩展在想复用场景时特别顺手。比如你有两个不同的服务都需要天气信息直接把WeatherSkill注册到各自的Registry里即可。3.3 调度器的连接逻辑让模型真正跑起来调度器是串联大模型与技能中心的枢纽环节。我用一段极简的伪代码来说明它的完整链路def run_agent(user_message, available_skills): # 1. 构建给模型的技能提示词 skills_prompt build_system_skills_prompt(available_skills) # 2. 调用模型这里是伪代码示例请根据实际接入的模型SDK来写 response llm_chat( systemskills_prompt, useruser_message ) # 3. 判断模型是否要求调用技能 if response.get(tool_calls): call_info response[tool_calls][0] skill skill_registry.get(call_info[name]) if skill is None: return {error: fUnknown skill: {call_info[name]}} try: result skill.execute(**call_info[arguments]) return result except Exception as e: return {error: str(e)} # 4. 如果模型给出的是普通文本回答直接返回 return {answer: response[content]}这一整段逻辑看似简单但实际执行起来却会出现各种奇怪的问题尤其是模型幻觉性调用——它明明没有拿到参数却编造了一个参数出来。我在后面章会专门讲这个问题以及它的解法。3.4 技能参数校验与执行结果统一化在实际交付中参数校验这步不能省。模型生成的JSON看起来合理但可能包含多余字段、错误类型甚至在要求传int时传了字符串。我在每个技能执行前增加了统一的参数校验逻辑from jsonschema import validate, ValidationError def safe_execute(skill: BaseSkill, arguments: dict): try: validate(instancearguments, schemaskill.parameters) except ValidationError as e: return {error: f参数校验失败: {e.message}} try: result skill.execute(**arguments) # 统一包装结果方便模型二次理解 return {skill: skill.name, result: result} except Exception as e: return {skill: skill.name, error: str(e)}这一步强烈建议做。如果不做执行期报错往往等到任务的最后一步才发现排查成本很高。而且做了校验之后如果模型参数不对你可以把错误信息反馈给模型让它自我纠错重新生成一次调用这在体验上会有非常好的效果。4. 技能编排与组合从单一技能到复合任务4.1 用技能调技能的能力单一技能能解决的问题很有限。比如用户说北京今天适合跑步吗你需要先查天气再做一个运动建议判断。在agent-skills的方案里我通过技能内部调用注册中心来实现这一点——每个技能持有共享Registry的引用class RunAdviceSkill(BaseSkill): name run_advice description 根据城市天气情况给出是否适合跑步的建议 parameters { type: object, properties: { city: {type: string, description: 城市名称} }, required: [city] } def __init__(self, registry): self._registry registry def execute(self, city: str): weather_skill self._registry.get(get_weather) weather weather_skill.execute(citycity) temp weather[temperature] if temp and temp 30: return {city: city, advice: 温度偏高建议清晨或傍晚跑步} return {city: city, advice: 天气适宜可以正常跑步}这个设计最大的价值是避免在系统提示词层写死复杂业务规则。规则属于代码它应该是可控、可测试、可迭代的而不是让模型每次靠概率来猜。复合技能的执行链路是可追踪的哪个环节出了问题直接看日志定位即可。4.2 技能的动态加载与懒实例化随着项目推进技能数量会快速膨胀。如果启动时把所有技能全部实例化项目资源和时间开销都不小。我后期做了一版延迟实例化优化注册中心不直接存对象改存类在get()时按需创建。class LazySkillRegistry: def __init__(self): self._skill_classes {} def register(self, skill_cls): self._skill_classes[skill_cls.name] skill_cls return skill_cls def get(self, name: str) - BaseSkill: skill_cls self._skill_classes.get(name) if skill_cls is None: return None return skill_cls() # 按需创建实例这个优化真正做到物尽其用。对一套支撑多Agent的场景来说这个收益非常可观——比如系统里注册了90个技能但某个客服Agent实际上只用到10个延迟实例化帮它省掉了大量不必要的初始化开销。4.3 面向多Agent的技能的隔离与共享再深入一步当一套技能库要给多个Agent用时你需要考虑好哪些技能共享、哪些技能隔离。我是这样做的基础能力查天气、算日期、HTTP请求全部共享放公共Registry。业务能力查订单、改订单、退换货按域拆分每个域维护一个Registry。Agent启动时把公共Registry与业务Registry合并成这个Agent的可用技能视图互不干扰。这样一来客服Agent能改订单但数据分析Agent完全没有这个权限。技能权限这块建议在入口处统一控制不要让技能内部自己去判断谁能调我而是由上层代理决定我能看到并调用哪些技能。5. 常见问题与排查技巧实录5.1 模型反复调用同一个错误技能这是典型的描述不清问题。你定义了一个query_order技能但模型总是跑去调用get_order而后者根本不存在。我的排查步骤是先打开模型实际收到的技能列表看name和description是否准确。再检查两个技能名的相似程度——query_order和get_order在语义上太接近模型极易混淆。解决办法也很直接要么合并两个技能要么在描述的措辞上做区分标记比如用于售前查询的订单信息接口勿用于售后场景。5.2 模型编造参数而不是正确取值经常出现的情况是模型调用query_order时order_id参数传了一个12345而真实订单号是SO20250101。究其根本是因为系统提示词里没有说明参数来源——用户提到的信息需要转译成参数。我的解法是在技能描述里明确写上参数必须从用户对话中提取不得自行编造如果用户未提供订单号请主动询问用户。加上这句提示后幻觉参数的情况大幅度减少。5.3 参数类型对不上导致的执行错误模型返回的JSON里{age: 25}和{age: 25}对Python来说含义完全不同。jsonschema校验能帮你挡住大部分问题但更好的方案是宽容解析——在safe_execute之前做一次类型强制转换比如数字字符串转成int再交给技能执行器。这层容错设计能显著降低模型输出的随机性影响。5.4 执行超时与外部依赖故障技能执行不可能永远快外部API可能宕机网络可能超时。我在每个技能执行器外层统一加了超时控制用的Python自带concurrent.futures非常轻量from concurrent.futures import ThreadPoolExecutor, TimeoutError def execute_with_timeout(skill, arguments, timeout10): with ThreadPoolExecutor(max_workers1) as executor: future executor.submit(skill.execute, **arguments) try: return future.result(timeouttimeout) except TimeoutError: return {error: f技能 {skill.name} 执行超时}这里超时时间要根据技能特性来设置别一刀切用全局默认值。调用慢日志的HTTP接口和本地文件读取两者的等待时间完全是两个量级建议每个技能在注册时自己带上timeout字段。5.5 工具使用效果不稳定时的降级策略就算你做好了以上所有步骤模型偶尔还是会乱来——这个时候不能直接崩建议设计一条降级链路。我的方案是在工具调用失败后把报错信息返回给模型让它重新选择技能或直接回答用户。比如第一次调用模型选错技能执行报错。系统把错误信息拼接进上下文技能query_order执行失败参数校验未通过。可选技能有...。模型重新生成一次调用这次选对了。这个链路在内部叫一次纠错机会实际效果非常好把工具调用的成功率从85%拉到了95%以上。记住不要无限循环重试超过2次就直接转入兜底话术避免死循环浪费时间。6. 经验总结与实践建议做到这一步agent-skills这个方向已经能给一套真实的Agent应用带来质的提升。我个人的感觉是它真正的价值不在于某一个技能写得多精妙而在于确立了一套Agent能力边界的工程规范。在这套规范下你可以很从容地面对这样的问题新增需求时我不用重新调整Agent的记忆和提示词只需要注册一个新的技能类排查故障时我直接看技能执行日志不需要靠猜模型在想什么。还有一个经验想分享给大家技能设计初期要克制盲目拆分的冲动别把一个发送邮件动作拆成获取收件人获取正文调用SMTP三个技能。模型更擅长使用粒度适中的工具而不是过度碎片化的微操作。我一般按一个技能能否独立解决一个用户可感知的子任务来判定拆分粒度。这轮实践做下来我最大的体会是想让Agent在真实业务里稳定可用与其反复琢磨Prompt话术不如把精力放在技能层的基础设施建设上。它是整个链路里最可控、最可优化、也最容易出成果的部分。后续我还在规划技能版本管理和技能效果看板这两个方向希望这套方案能帮助更多遇到同样瓶颈的同学少走一点弯路。