1. 项目整体设计与思路拆解1.1 为什么要给Agent设计“技能”如果你和我一样在去年到今年这段时间里密集玩过各种大模型应用大概率会遇到同一个问题大模型本身只会“说话”不会“办事”。你跟它说要查天气它能写得像模像样地回复你一段话但它不会真的去调用天气API更不会去读你服务器上的日志文件。这就是业界常说的“模型是大脑但不是手脚”。Agent-skill这套东西本质上就是给Agent装上一套“可插拔的手脚”。我在第一批做实验的时候也走过弯路一开始是写死一堆工具函数用一个巨大的Python字典把函数名和函数对象映射起来然后靠prompt告诉模型有哪些工具可用。试过一次之后我就明白这种方式只能跑Demo根本撑不住真实业务场景。原因很简单函数一多模型就懵了不知道什么时候该用哪个而且调用参数一复杂模型传参经常传错轻则报错重则拿错误参数去执行酿成大问题。所以我开始考虑把“技能”这个概念独立出来。每一个技能就是一个封装良好的、自描述的任务执行单元它自带描述、参数Schema、执行逻辑、重试策略和错误处理。模型只需要像调用工具那样根据描述和参数定义来完成选型和传参真正干活的代码全部封装在技能内部。这样大脑归大脑手脚归手脚各司其职整体可靠性一下就上来了。1.2 技能体系的核心构成一个完整的Agent技能体系个人觉得至少要包含这么几层第一层是技能定义层。这一层管的是“技能长什么样”包括技能的元信息、能力描述、参数Schema、返回结果Schema。我见过很多项目把技能定义放在一个JSON文件里但我更推荐用一个独立的类来承载原因后面实操部分会说。第二层是技能注册与发现层。Agent启动时需要知道当前环境里有哪些技能可用怎么加载、怎么卸载怎么处理重名冲突。这一层相当于技能的操作系统。第三层是技能调度与执行层。当模型决定调用某个技能后系统需要校验参数、执行技能代码、捕获异常、重试、返回结构化结果。这里还需要考虑技能执行的并发控制、超时控制、资源隔离。第四层是技能观测与治理层。技能调用的日志、耗时、成功率、Token消耗都要记录下来否则时间一长系统就变成黑盒出了问题根本无从排查。这四个层对应到代码模块就很好规划了skill_definition、skill_registry、skill_executor、skill_observer。下面的内容我会逐步拆开来讲。1.3 技术选型的基本原则选型这件事我的原则是“不追求框架先用最朴素的方式跑通核心路径”。网上有很多现成的Agent框架但我在实际项目里发现框架越重定制就越痛苦。特别是技能这种高度个性化的部分往往需要和公司内部的API、服务、数据模型深度绑定用通用框架反而要去改框架的源码。所以我的建议是用Python 3.10以上版本做基础环境Pydantic做数据校验用标准库里的importlib做技能加载再加一个简单的装饰器模式做注册。如果后面并发上来了再考虑引入Celery或RPC框架。大模型调用层建议直接对接OpenAI兼容的接口协议这样后续切换模型供应商的成本会低很多。这套选型方案的好处是每一层都能看得见、摸得着出了问题直接用pdb去调试不需要在框架内部绕来绕去。坏处是你得自己处理一些边界情况但作为开发者熟悉这些边界反而是进步最快的方式。2. Agent技能的核心细节解析与实操要点2.1 技能定义Schema的编写技巧技能定义是整个体系的地基Schema写得不好后面模型选型、参数生成、结果解析都会出问题。我先放一个典型的结构再解释每个字段怎么取舍。from pydantic import BaseModel, Field from typing import Any, Optional class SkillDefinition(BaseModel): name: str Field(description技能名称必须全局唯一使用snake_case) description: str Field(description技能的功能描述用于给模型做选型) description_for_model: Optional[str] Field( defaultNone, description专门给大模型读的长描述可以包含使用场景、前置条件、典型示例, ) parameters: dict Field( description参数Schema遵循JSON Schema格式用于约束模型生成的参数 ) required_scopes: list[str] Field( default_factorylist, description该技能运行所需的作用域如读取文件、调用网络等 ) timeout_seconds: float Field(default10.0, description技能运行超时时间) max_retries: int Field(default0, description技能失败后最大重试次数) version: str Field(default1.0.0, description技能版本号) enabled: bool Field(defaultTrue, description技能是否启用)很多人会问description和description_for_model区别在哪。我的经验是description给开发者看讲究简洁准确写“查询天气”就够了description_for_model是给模型看的需要把什么时候用、怎么用、有什么坑都写清楚。比如查询天气你可以这样写当用户询问当前天气情况时使用此技能。必须先通过用户提供的位置信息解析经纬度 再调用本技能。如果用户只提供了城市名在调用前需要先用geocode技能进行坐标解析。 返回结果包含温度、天气现象、风向风速和穿衣建议。 如果参数中的经纬度不在中国境内请在返回结果中说明“该地点暂不支持”。这段描述既告诉模型技能的能力边界又给出了前置条件和典型用法模型对何时调用、传什么参数的判断就会准确很多。我实际测试过只写一句description的选型准确率大约在70%左右补全了description_for_model后能提高到90%以上。2.2 技能注册与发现机制的实现思路注册机制说白了就是“让系统知道有哪些技能”。我采用的方式是目录约定加装饰器自动注册。项目里建一个skills目录每个子目录放一个技能模块模块里通过skill.register装饰器把技能实例注册到全局注册表里。这里有一个取舍直接用importlib遍历目录加载所有模块比手动在配置里写技能列表要灵活得多。新增技能只需要丢一个文件夹进去重启服务就能自动发现省去了维护配置文件的过程。这个方案在技能数量少于100个时完全够用再往上就需要考虑模块热加载和注册表的并发安全问题。注册表本身不复杂我用了Thread-safe的单例模式import threading from typing import Dict class SkillRegistry: _instance None _lock threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance super().__new__(cls) cls._instance._skills {} cls._instance._lock threading.RLock() return cls._instance def register(self, name, skill_cls, overrideFalse): with self._lock: if name in self._skills and not override: raise KeyError(f技能 {name} 已存在如需覆盖请设置 overrideTrue) self._skills[name] skill_cls def get(self, name): with self._lock: return self._skills.get(name) def list_skills(self): with self._lock: return { name: skill_cls.get_definition().model_dump() for name, skill_cls in self._skills.items() }注册表加了锁之后并发环境下也不会出现重复注册或读脏数据的问题。技能加载器则负责扫描目录逐个加载模块import importlib import inspect import pkgutil import skills def load_skills_from_package(package_nameskills): registry SkillRegistry() package importlib.import_module(package_name) for _, module_name, _ in pkgutil.iter_modules(package.__path__): module importlib.import_module(f{package_name}.{module_name}) # 这里不直接调用任何东西因为装饰器在模块导入时就会执行注册 return registry用pkgutil.iter_modules扫描的好处是不需要手动维护模块列表只要skills包下每个子模块在导入时执行了注册这里的for循环就足够把所有技能都加载起来。2.3 技能上下文与状态传递的经验教训技能不像普通函数那样“无状态”执行很多时候它需要感知对话的上下文、用户的身份、会话的历史记录。我在最初设计时直接给每个技能函数加了session_id和user_id两个参数结果代码里到处穿这些参数又丑又容易漏。后来我改成上下文对象模式每个技能执行时接收一个SkillContext对象这个对象由框架统一构造里面包含当前会话ID、用户ID、历史消息摘要、外部环境的配置项比如API Key、数据库连接池等等。技能代码只需要从context里取自己关心的部分不需要关心这些数据是从哪来的。dataclass class SkillContext: session_id: str user_id: str history: list[dict] config: dict这里有个容易踩的坑千万不要直接在技能里修改context的数据。多个技能可能共用一个上下文如果某个技能把context里的user_id改成别的值后续所有技能都会受影响。我的做法是把context设计成只读的要在技能里保存中间结果就使用技能自己的返回值或者用专门的共享存储模块比如Redis不要依赖context。3. 实操过程与核心环节实现3.1 搭建一个最小可运行的技能框架接下来我会带你从零搭一套能跑起来的技能框架。示例中我会实现一个“记住用户偏好”的技能和一个“获取服务器CPU负载”的技能这两个技能分别是“纯内部状态”和“读外部数据”两种典型代表。先搭目录结构agent-skills-demo/ ├── main.py # Agent入口负责加载技能并调用LLM ├── skill_definition.py # 技能定义与装饰器 ├── skill_registry.py # 技能注册表 ├── skill_loader.py # 技能加载器 ├── skill_executor.py # 技能执行器 └── skills/ ├── __init__.py ├── remember_preference.py └── get_cpu_load.py首先是skill_definition.py核心是定义skill装饰器。装饰器做的事情是接收技能名称、描述、参数Schema等元信息将类包装成统一的技能实例并注册到全局注册表中。from functools import wraps from skill_registry import SkillRegistry class BaseSkill: name description parameters {} context_class None def run(self, **kwargs) - str: raise NotImplementedError def skill( name: str, description: str, parameters: dict, description_for_model: str , timeout_seconds: float 10.0, max_retries: int 0, ): def decorator(cls): original_run cls.run wraps(original_run) def wrapper(self, *args, **kwargs): # 执行前可以加日志、校验等逻辑 result original_run(self, *args, **kwargs) return result cls.run wrapper cls.name name cls.description description cls.parameters parameters cls.description_for_model description_for_model or description cls.timeout_seconds timeout_seconds cls.max_retries max_retries SkillRegistry().register(name, cls) return cls return decorator这里有个细节我用了functools.wraps是为了让装饰后的run方法保留原始函数的元信息方便后续做技能调用链追踪。虽然装饰器本身只做了一层轻包装但日志输出和问题排查时函数名和文档字符串是否保留直接影响调试体验。安装依赖并创建虚拟环境这些基础操作就不啰嗦了切换到项目目录后直接python -m venv venv source venv/bin/activate pip install pydantic openai python-dotenv3.2 实现一个“记住用户偏好”的技能这个技能不需要访问外部系统它只是把用户偏好保存到内存字典中。虽然很简单但它能体现一个技能类的基本写法。注意技能类的run方法不接收context对象而是接收参数校验后的具体参数这样技能与自己需要的参数耦合度最低。from skill_definition import skill # 内存存储实际项目可以替换为数据库或Redis _preferences {} skill( nameremember_preference, description记录用户长期偏好, description_for_model( 当用户表达出对某个话题、语气、时间、地点或饮食等方面的偏好时 使用此技能将偏好记录下来。即使当前对话不需要立即用到该偏好 只要用户明确提出或强烈暗示了长期偏好就应主动记录。 参数preference_content是偏好的具体描述scope是偏好适用范围。 ), parameters{ type: object, properties: { preference_content: { type: string, description: 用户的偏好内容例如喜欢简洁回复、偏好夜间工作 }, scope: { type: string, enum: [global, chat], description: global代表长期全局有效chat代表仅当前对话有效 } }, required: [preference_content, scope] }, timeout_seconds2.0, ) class RememberPreferenceSkill: def run(self, preference_content: str, scope: str global) - str: key (scope, preference_content) _preferences[key] True return f已记住偏好{preference_content}范围{scope}写这个技能时有几个经验第一描述里一定要告诉模型“什么时候不用调用”也很重要比如用户只是开玩笑就不要记录否则长期运行的Agent会积累一堆垃圾偏好。第二参数枚举值要尽量明确比如scope用enum定义后模型基本不会生成超出范围的值比直接写字符串省心得多。3.3 实现一个“获取服务器CPU负载”的技能记住偏好的技能是纯内部状态现在来写一个真正干活的技能读取服务器CPU负载。为了演示的完整性我直接用psutil库这在日常运维类Agent里非常常见。pip install psutil技能代码import json import psutil from skill_definition import skill skill( nameget_cpu_load, description获取当前服务器CPU核心数量及各核使用率, description_for_model( 当用户询问服务器CPU负载情况时使用此技能例如“看看CPU现在忙不忙”、 “最近一分钟的负载是多少”。该技能不需要额外参数直接调用即可。 返回结果包含逻辑核心数、每个核心的使用率以及最近1/5/15分钟的平均负载。 如果使用率超过85%建议在返回中附加一句“当前CPU负载较高建议排查进程情况”。 ), parameters{ type: object, properties: {}, required: [] }, timeout_seconds3.0, ) class GetCpuLoadSkill: def run(self) - str: cores psutil.cpu_count(logicalTrue) per_core psutil.cpu_percent(intervalNone, percpuTrue) load_avg psutil.getloadavg() data { cores: cores, per_core_percent: per_core, load_avg_1min: round(load_avg[0], 2), load_avg_5min: round(load_avg[1], 2), load_avg_15min: round(load_avg[2], 2), } return json.dumps(data, ensure_asciiFalse)注意这里我把intervalNone意思是获取当前瞬时CPU使用率而不是一段时间的平均值避免技能执行时阻塞太多时间。如果你希望更准确的负载数据可以在技能里加一个调参方法但默认不要给模型太多选择否则它可能会传一个不合理的interval时长。3.4 技能执行器的参数校验与调度逻辑技能定义好了之后关键是如何被模型调用。执行器的职责是接收模型返回的工具调用消息解析出技能名称和参数然后去注册表找到技能做参数校验最后执行技能并返回结果。import json import time import traceback from typing import Any from skill_definition import BaseSkill from skill_registry import SkillRegistry from skill_executor import SkillExecutor这里SkillExecutor我直接写在同一个模块里了。核心的execute方法是这样class SkillExecutor: def __init__(self): self.registry SkillRegistry() def execute(self, skill_name: str, raw_params: dict) - dict: skill_cls self.registry.get(skill_name) if skill_cls is None: return {status: error, message: f未找到技能{skill_name}} skill_instance skill_cls() start_time time.time() retries 0 max_retries skill_cls.max_retries while True: try: result skill_instance.run(**raw_params) elapsed time.time() - start_time return { status: success, skill_name: skill_name, result: result, elapsed_ms: round(elapsed * 1000, 2), } except TypeError as e: # 参数不匹配时直接返回不重试因为重试大概率还是同样的错误 return { status: error, skill_name: skill_name, error: f参数错误{e}, elapsed_ms: round((time.time() - start_time) * 1000, 2), } except Exception as e: if retries max_retries: return { status: error, skill_name: skill_name, error: str(e), elapsed_ms: round((time.time() - start_time) * 1000, 2), } retries 1 time.sleep(0.5 * retries) # 简单的退避策略 def extract_and_execute(self, message: Any) - dict: # 兼容OpenAI工具调用格式 if hasattr(message, tool_calls) and message.tool_calls: tool_call message.tool_calls[0] skill_name tool_call.function.name raw_params json.loads(tool_call.function.arguments or {}) return self.execute(skill_name, raw_params) return {status: skipped, message: 没有需要执行的技能}重试策略这里有一个容易被忽略的点TypeError不要重试。模型偶尔会生成缺参数或多参数的情况这种错误无论重试多少次都会失败直接返回错误信息给模型让模型根据错误修正参数再发起下一次调用反而更高效。而网络抖动、外部API 500之类的错误重试才有价值。3.5 接入大模型的完整调用链现在把整个链路串起来。我使用OpenAI接口格式模型会被传入“技能描述列表”也就是把注册表里的每个技能定义转换成工具列表然后正常对话。当模型返回工具调用时由执行器完成调用并再次发送给模型。import os from openai import OpenAI def build_tool_schema(registry): tools [] for name, skill_cls in registry.list_skills().items(): definition skill_cls.get_definition() tools.append({ type: function, function: { name: definition.name, description: definition.description_for_model, parameters: definition.parameters, } }) return tools def run_agent(user_input: str): registry SkillRegistry() tools build_tool_schema(registry) client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1), ) messages [{role: user, content: user_input}] response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, tool_choiceauto, ) message response.choices[0].message if message.tool_calls: messages.append(message) executor SkillExecutor() for tool_call in message.tool_calls: execution_result executor.execute( tool_call.function.name, json.loads(tool_call.function.arguments or {}), ) messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(execution_result, ensure_asciiFalse), }) second_response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, tool_choiceauto, ) return second_response.choices[0].message.content else: return message.content注意这里我把技能执行结果用JSON字符串作为tool消息回传给模型模型的后续回答会基于这个结果组织语言。这种做法的好处是“让模型做摘要而不是让它背数据”例如CPU负载是“55%”还是“0.55”模型知道怎么组织语言不会生硬地把JSON读出来。4. 常见问题与排查技巧实录4.1 技能命名冲突两种不同语义的技能撞了名技能注册表遇到重名会直接抛异常这是正确的保护机制。但在实际项目里重名的原因往往不是两个人取了相同的名字而是同一个技能在不同版本里改了语义。比如我在一个项目中A组定义了send_email技能用于发送邮件通知B组也定义了send_email但用途是向用户发送营销活动邮件。两个技能名字一模一样参数结构却完全不同。模型本来想调用A组的结果工具列表里只有B组参数传成了A组的结构执行就直接报错。解决办法是建立“命名空间”概念。在技能名前加上团队或业务前缀比如notification.send_email和marketing.send_email。模型在选型时可以根据前缀更精确地判断。另一种办法是把技能归属作为元数据的一部分注册表的key仍然用全名但是在展示给模型时把归属信息放进描述里。4.2 模型参数生成错误怎么让模型传参更准确我统计过Agent技能调用失败的原因里有近一半都出在参数生成环节。模型生成的参数要么缺字段、要么类型不对、要么枚举值超纲。纯靠prompt约束是不够的必须从Schema层面做校验。用Pydantic做校验是第一步但更重要的是要在错误信息里告诉模型“错在哪、应该怎么改”。我记得有一次系统返回给模型的错误是简单的“skill execute error”模型完全不知道要怎么修正连续尝试了三次都在原地打转。后来我把错误信息改成一长串提示比如参数校验失败字段city为必填项但未提供。正确的调用方式为 {city: 用户所在的城市名称, unit: celsius|fahrenheit}模型看到这样的提示后二次调用的成功率几乎达到了百分百。这个经验也可以推广到其他Agent系统错误信息要面向模型的修复行为来设计而不只是给开发者看。4.3 技能执行超时外部依赖导致整个对话卡住如果你的技能里调用了一个不稳定的第三方API超时问题几乎一定会遇到。有一次我在技能里调用内部搜索服务那个服务偶尔需要好几秒才返回结果整个Agent对话就像卡死了一样用户体验非常差。我推荐两个方案并用一是给技能统一设置超时控制用concurrent.futures的Future加超时参数二是让技能支持“快速失败”模式比如设计一个默认阈值超过阈值就返回“该服务当前响应较慢请稍后再试”而不是一直等待。超时时间长短要按技能类型区分。纯内存操作2秒足够数据库查询可以放宽到5秒外部HTTP调用默认10秒。你要在技能定义里写清楚不要所有技能都用同一个超时否则有些技能会频繁超时有些则会导致整个链路无限挂起。4.4 技能调试技巧日志、回放与技能沙箱调试Agent技能比调试普通函数要麻烦因为它多了“模型选型”这一步。光看技能内部日志不够你还要知道模型到底为什么要选这个技能、传了什么参数。我常用的调试手段有三种第一全量日志。把每次模型请求的完整内容、模型返回的工具调用、技能执行的输入和输出都记录到结构化日志里。这个日志量会比较大所以按session_id做了拆分方便单独查看某一次对话的完整链路。第二录制回放。把线上请求的参数和模型响应录制下来在本地用测试脚本重新执行技能这样技能改动后能在完全相同的输入下做回归测试。我现在每改一个技能必做一次回放比对避免改了一个技能弄坏了另一个技能的调用。第三技能沙箱。如果是给多用户使用的Agent技能的执行环境必须做隔离。我最初用的是线程池但有些技能会写文件、改环境变量线程隔离根本兜不住。后来换成子进程隔离每次执行技能都会在一个独立的子进程里运行用multiprocessing的Queue拿结果这样即使技能内部把环境搞乱了也不会影响主进程和其他技能。4.5 技能编排多个技能如何串联单技能调用跑通之后很快会遇到多技能组合的场景。比如用户说“帮我查一下明天北京的天气然后帮我订一个8点去机场的出租车”这明显是两个技能前后依赖的调用。我踩过的坑是把编排逻辑写死在Agent的代码里模型一旦没有按预想顺序执行流程就断了。更好的方式是把编排能力交给模型让模型在回答中同时调用多个工具然后根据每个工具的返回结果决定下一步动作。OpenAI的parallel tool call天然支持一次返回多个tool_call但要注意语义依赖如果第二个技能依赖第一个技能的返回结果并行调用就不适用了得让模型先调用第一个、看到结果后再发起第二个调用。还有一种方式是自己实现一个小小的状态机定义技能之间的依赖关系。我一般只在技能链路非常固定时才这么做比如“先解析地址再查询天气最后生成出行建议”这种固定流程。对于开放式的多技能组合让模型自由调度才是更灵活的方案。5. 技能治理与可观测性建设技能多了以后最头疼的不是写技能而是不知道怎么保证它们稳定、可信。我把这块归到“治理”说几个实战中特别有用的点。第一是技能调用统计。每个技能的执行次数、平均耗时、失败率、Token消耗至少要按天做一次汇总。我遇到过某个月某个技能的调用量突然翻了十倍查下来原因是有个Prompt改动导致模型误用了这个技能。如果没有统计这种问题根本发现不了。第二是技能灰度发布。你可以用一个简单的流量配置表让新技能先只对10%的session生效观察调用准确率没有下降后再全量放开。这个套路在相关场景里我见过多次用起来非常管用。实现起来不复杂注册一个技能时带上灰度因子比如user_id的hash值取模命中灰度区间的才注册否则跳过。第三是技能版本管理。技能的改动不应该影响线上运行至少要保留上一个版本的备份。我的做法是每个技能类里带一个version字段升级时保留旧文件在注册表里记录version的激活状态。回滚时只需要改配置不需要重新部署代码。第四是技能质量评估体系。定期把一批历史对话输入回放给Agent看它在“是否需要调用技能”和“调用是否正确”两个维度上的表现。这个评估最好做成自动化流水线通过对比正确答案和模型回答的相似度给出一个技能使用质量的分数。我自己用这套方法把技能选型准确率从70%提到了93%效果非常明显。6. 一些实战外的体会这几个月把agent-skills项目从雏形做到基本可用最大的感受是技能化设计不是把函数包一层壳那么简单它实际上是在“模型的不确定性世界”和“代码的确定性世界”之间架一座桥。桥的质量取决于你对模型弱点的理解以及对工程细节的敬畏。我自己在写技能时反复提醒自己三件事一是技能描述一定要站在模型视角写而不是站在开发者视角写二是参数Schema要尽量严格但错误提示要尽量温和三是先考虑失败场景再考虑成功路径。你要永远假设模型有可能会传一个你没见过的参数组合技能代码要能优雅地兜住这些意外。如果你正在做一个Agent项目我的建议是不要急着引入重框架先从手写几个技能开始把注册、校验、执行、日志这条链路跑通你会在过程中发现很多框架帮你隐藏掉的坑。踩过这些坑之后你自然会更理解框架的设计意图也就能做出更适合自己业务的技能体系。后面我会继续分享技能评估、多Agent协作方面的实践欢迎一起交流。
