智能体技能契约设计：解耦能力与决策的工程实践

1. 项目概述这不是一个“技能库”而是一套可插拔的智能体行为引擎“agent-skills”这个名称乍看像某个开源工具包的代号但实际在当前技术演进脉络里它代表一种更底层、更务实的设计范式——把大模型驱动的智能体Agent所依赖的“能力”从推理链中解耦出来封装成独立、可验证、可组合、可替换的运行时单元。我接触过不少团队最初都以为这是个“预置函数集合”结果在真实业务场景中跑了几周才发现真正卡住进度的从来不是模型本身而是“让模型知道什么时候该调用什么能力、怎么调用、调用失败后如何降级”。这正是“agent-skills”要解决的核心问题。它不等于API调用列表也不等于Prompt模板合集。它的本质是定义了一套标准化的能力契约Skill Contract每个skill必须明确声明输入结构Input Schema、输出结构Output Schema、执行超时Timeout、重试策略Retry Policy、失败兜底逻辑Fallback Handler以及最关键的——能力边界说明Boundary Statement。比如一个“查天气”skill不能只写“返回天气信息”而必须注明“仅支持中国地级市行政中心不支持景区/街道级定位温度单位默认摄氏湿度为相对湿度百分比若城市名存在歧义如‘长安’优先匹配陕西省西安市长安区”。这种颗粒度的约束才是工程落地的起点。适合谁参考如果你正在做以下任何一件事这篇内容就是为你写的正在搭建客服对话系统发现用户一问“帮我查下昨天北京南站出发的G101次列车晚点没”整个流程就卡在“解析意图→识别地点→识别车次→调铁路接口→解析返回→组织回答”这一长串链路上且任意一环出错就崩在开发内部知识助手但每次用户问“上季度华东区销售冠军是谁”系统要么答非所问要么直接抛异常或者你只是个技术负责人在评估是否要把现有RAG系统升级为Agent架构却苦于找不到可衡量、可测试、可运维的能力交付标准。它不是银弹但能让你少踩80%的“能力不可控”坑。接下来我会从设计逻辑、核心契约、实操实现、典型故障四个维度把这套机制掰开揉碎讲透。2. 设计思路拆解为什么必须把“技能”从Agent主干中剥离2.1 传统Agent架构的三大隐性成本很多团队一开始直接基于LangChain或LlamaIndex搭Agent很快就会撞上三堵墙而这三堵墙全是因为“技能”和“决策流”混在一起导致的第一堵墙调试黑洞当用户问“帮我订明天上海到杭州的高铁票”系统返回“未找到可用班次”你根本不知道问题出在哪——是LLM没识别出“上海”和“杭州”是出发到达地还是调用12306接口时参数拼错了抑或是接口返回了HTML乱码但没做清洗所有日志混在一条trace里就像把十台机器的机油倒进同一个油桶再想分清哪台漏油基本靠猜。第二堵墙版本失控某天运营提需求“把天气查询结果里的‘多云’改成‘☁️ 多云’加个emoji更亲切”。开发改完Prompt测试通过就上线。结果三天后发现财务报表生成模块也调用了同一个天气skill导致Excel里出现无法解析的字符整个月报流程中断。问题根源在于没有能力隔离层一次修改影响所有调用方。第三堵墙能力幻觉蔓延LLM在规划阶段会自信满满地说“我将调用航班查询skill获取CA1501航班状态”但实际该skill只支持国内航班CA1501是国际航班调用直接404。更糟的是LLM不会说“我调用失败了”而是自己编造一个“航班正常”的答案返回给用户。这就是典型的“能力幻觉”——模型对自身能力边界的认知远不如代码里明确定义的schema可靠。2.2 “agent-skills”的解法用契约代替信任“agent-skills”的核心思想就是把上述混乱全部收束到一个“能力契约”里。这个契约不是文档而是可执行、可验证、可拦截的代码契约。我们以一个真实的“企业微信消息发送skill”为例说明class WeComMessageSkill(Skill): name send_wecom_message description 向指定企业微信成员发送文本消息支持所有人 # 输入契约强制校验 input_schema { type: object, properties: { user_id: {type: string, minLength: 1}, content: {type: string, maxLength: 2048}, at_all: {type: boolean, default: False} }, required: [user_id, content] } # 输出契约定义成功/失败结构 output_schema { type: object, properties: { status: {enum: [success, failed]}, message_id: {type: string, nullable: True}, error_code: {type: string, nullable: True} } } # 执行契约超时重试兜底 timeout 8.0 # 秒 max_retries 2 fallback lambda: {status: failed, error_code: FALLBACK_UNAVAILABLE}看到这里你应该明白了这个skill不是一段能“干活”的代码而是一个带说明书、带质检线、带应急通道的微型工厂。LLM只需要按契约投递原料input工厂自动完成生产execution、质检output validation、返工retry或报修fallback最后交出符合标准的成品output。LLM不再需要“理解”怎么发消息它只需要“知道”有这个能力可用。2.3 为什么不用微服务——轻量级契约的不可替代性有人会问既然要解耦直接做成HTTP微服务不更彻底确实可以但代价巨大每次调用增加200ms网络延迟而Agent决策链往往需要3~5次skill调用端到端延迟直接翻倍需要单独维护服务发现、熔断、鉴权、日志聚合等整套基础设施小团队根本玩不起最致命的是微服务天然缺乏“能力边界描述”你调用/api/weather它返回400错误但错误体里不会告诉你“只支持地级市不支持区县”。“agent-skills”的精妙之处在于它用进程内契约In-process Contract实现了微服务的治理能力同时保留了本地调用的性能。它本质上是一种“面向能力的编程范式”Capability-Oriented Programming而非简单的函数封装。3. 核心契约详解五个不可妥协的契约要素3.1 输入契约Input Schema不是校验而是意图对齐输入契约的终极目的不是防止用户输错而是确保LLM的规划意图与skill的实际能力完全对齐。很多团队只做基础类型校验如user_id必须是字符串这远远不够。我们曾遇到一个真实案例某HR系统接入“员工信息查询skill”LLM规划时生成的input是{employee_name: 张三, department: 技术部}但skill实际只支持按工号查询employee_name字段被完全忽略最终返回空结果。LLM却据此推断“该员工不存在”误导HR。正确做法是在input_schema中用description字段强制声明语义约束employee_name: { type: string, description: 仅支持精确匹配不支持模糊搜索若传入中文名必须为员工花名册登记的全名不支持昵称或英文名 }更进一步我们会在skill初始化时对schema做静态分析自动生成“能力说明书”Markdown文档供LLM在规划前读取。例如本skill仅支持按工号查询员工信息。若用户提供姓名请先调用员工姓名转工号 skill获取工号再调用本skill。这才是真正的意图对齐——不是让LLM猜而是给它一份操作手册。3.2 输出契约Output Schema失败也是契约的一部分90%的skill文档只描述“成功时返回什么”却对失败场景闭口不谈。这导致LLM在收到{error: timeout}时无法判断是网络问题可重试、权限问题需换账号、还是参数错误需修正input。输出契约必须明确定义所有可能的失败形态。我们采用三级错误分类法Transient Errors瞬时错误如网络超时、服务暂时不可用。契约中声明retryable: trueskill自动重试。Business Errors业务错误如“查无此人”、“日期超出范围”。契约中声明retryable: false但提供suggestion字段指导LLM下一步动作例如suggestion: 请确认员工工号是否正确或调用员工工号查询助手 skill获取最新工号System Errors系统错误如JSON解析失败、内存溢出。这类错误不暴露给LLM由skill框架捕获并触发fallback。这种设计让LLM的决策逻辑变得极其清晰收到retryable: true就重试收到suggestion就按提示行动收到系统错误就走全局降级流程。不再有“未知错误”这种黑洞。3.3 执行契约Execution Contract超时与重试的科学设定超时时间不是拍脑袋定的。我们有一套计算公式timeout base_latency × (1 p95_latency_variance) network_jitter其中base_latency是skill在本地环境下的P50延迟我们要求所有skill必须提供基准压测报告p95_latency_variance是线上环境P95延迟相对于P50的波动系数通常取0.8~1.2network_jitter是跨服务调用的网络抖动预留本地调用设为0.2秒HTTP调用设为1.0秒。例如某数据库查询skill本地P50120msP95380ms则variance (380-120)/120 ≈ 2.17取保守值1.5timeout 120×(11.5)0.2 320ms。实测下来这个值能让99.2%的成功请求在超时前完成同时避免因单次慢查询拖垮整个Agent链路。重试策略同样讲究绝不简单地“重试3次”。我们采用指数退避去重键deduplication key组合第一次失败后等待100ms第二次失败后等待300ms第三次失败后等待900ms每次重试前用input_hash timestamp_floor_to_second生成去重键防止因网络重传导致同一请求被执行两次比如扣款skill。3.4 降级契约Fallback Contract兜底不是摆设而是预案很多团队的fallback写的是return {error: 服务繁忙}这等于没写。真正的fallback必须满足三个条件可预测性返回结构必须严格符合output_schema不能临时加字段业务合理性不能答非所问。比如“查天气”skill的fallback绝不能返回“今天宜出行”而应返回{status: failed, error_code: FALLBACK_WEATHER_UNAVAILABLE, suggestion: 可尝试查看手机天气APP}可观测性每次触发fallback必须记录完整上下文input、原始错误、触发时间用于后续分析是skill本身缺陷还是LLM规划错误。我们甚至为高频skill配置了分级fallback一级fallback本地缓存如天气skill缓存昨日数据二级fallback调用简化版API如只返回温度不返回湿度三级fallback纯静态文案如“天气服务暂时不可用建议稍后重试”。这样既保障用户体验又为运维提供清晰的降级路径。3.5 边界契约Boundary Statement用自然语言写清楚“不能做什么”这是最容易被忽视却最关键的一环。技术文档喜欢写“支持XX功能”但LLM真正需要的是“不支持什么”。我们要求每个skill的boundary_statement必须包含三个要素地理/行政边界“仅支持中国大陆境内企业微信账号不支持海外版WeCom即WeCom International港澳台地区账号需单独开通国际版权限。”数据时效边界“航班状态数据延迟不超过15分钟不保证实时性起飞前2小时内的状态变更可能无法及时同步。”语义理解边界“不支持自然语言时间表达如‘下周三’、‘大后天’必须传入ISO 8601格式日期如‘2024-06-15’。”这些语句会被注入LLM的system prompt成为其规划时的硬性约束。实践证明明确写出“不能做什么”比反复强调“能做什么”更能减少幻觉。4. 实操实现从零搭建一个可验证的skill运行时4.1 技术选型逻辑为什么选Python Pydantic asyncio我们对比过多种技术栈最终锁定Python生态原因很实在Pydantic v2 的Schema校验能力无敌支持JSON Schema导出、自动生成文档、运行时动态校验一行代码搞定input/output验证asyncio 原生支持异步I/Oskill多数是网络调用async/await能轻松应对高并发Python生态的成熟度Requests、httpx、SQLAlchemy等库对异常处理、连接池、超时控制都有工业级实现不用重复造轮子。我们不推荐用Go或Rust重写——除非你的skill全是CPU密集型计算如图像处理否则网络I/O才是瓶颈Python的async生态完全够用。4.2 核心运行时骨架代码精简版以下是经过生产验证的skill运行时最小可行骨架已去除所有业务逻辑只保留契约执行核心import asyncio import json import time from typing import Any, Dict, Optional, Callable, Awaitable from pydantic import BaseModel, ValidationError from pydantic.json_schema import model_json_schema class SkillResult(BaseModel): status: str # success or failed data: Optional[Dict[str, Any]] None error_code: Optional[str] None suggestion: Optional[str] None execution_time_ms: float 0.0 class SkillConfig(BaseModel): timeout: float 5.0 max_retries: int 2 fallback: Optional[Callable[[], Dict[str, Any]]] None class Skill: name: str description: str input_schema: Dict[str, Any] output_schema: Dict[str, Any] config: SkillConfig async def execute(self, input_data: Dict[str, Any]) - SkillResult: start_time time.time() # Step 1: Input Validation try: validated_input self._validate_input(input_data) except ValidationError as e: return SkillResult( statusfailed, error_codeINPUT_VALIDATION_ERROR, suggestionf输入参数错误{str(e)}, execution_time_ms(time.time() - start_time) * 1000 ) # Step 2: Execute with Retry last_error None for attempt in range(self.config.max_retries 1): try: # Apply timeout result await asyncio.wait_for( self._run_core(validated_input), timeoutself.config.timeout ) # Validate output validated_output self._validate_output(result) return SkillResult( statussuccess, datavalidated_output, execution_time_ms(time.time() - start_time) * 1000 ) except asyncio.TimeoutError: last_error TIMEOUT if attempt self.config.max_retries: await asyncio.sleep(min(0.1 * (3 ** attempt), 2.0)) # exponential backoff continue except Exception as e: last_error f{type(e).__name__}: {str(e)} if attempt self.config.max_retries: await asyncio.sleep(min(0.1 * (3 ** attempt), 2.0)) continue # Step 3: Fallback if self.config.fallback: try: fallback_result self.config.fallback() return SkillResult( statusfailed, datafallback_result, error_codeFALLBACK_TRIGGERED, execution_time_ms(time.time() - start_time) * 1000 ) except Exception as e: pass return SkillResult( statusfailed, error_codefEXECUTION_FAILED_AFTER_{self.config.max_retries}_RETRIES, suggestionf多次执行失败最后错误{last_error}, execution_time_ms(time.time() - start_time) * 1000 ) def _validate_input(self, data: Dict[str, Any]) - Dict[str, Any]: # Use Pydantic to validate against input_schema # Implementation omitted for brevity pass def _validate_output(self, data: Dict[str, Any]) - Dict[str, Any]: # Use Pydantic to validate against output_schema pass async def _run_core(self, validated_input: Dict[str, Any]) - Dict[str, Any]: # This is the actual business logic, to be implemented by subclass raise NotImplementedError这个骨架的价值在于它把所有契约要素输入校验、超时、重试、fallback、输出校验全部封装在基类里子类开发者只需专注写_run_core。我们要求所有skill必须继承此基类确保契约一致性。4.3 一个完整可运行的示例企业微信消息发送skill下面是一个真实部署过的skill已脱敏处理可直接运行import httpx from typing import Dict, Any, Optional from pydantic import BaseModel class WeComMessageInput(BaseModel): user_id: str content: str at_all: bool False class WeComMessageOutput(BaseModel): status: str message_id: Optional[str] None error_code: Optional[str] None class WeComMessageSkill(Skill): name send_wecom_message description 向指定企业微信成员发送文本消息支持所有人 input_schema WeComMessageInput.model_json_schema() output_schema WeComMessageOutput.model_json_schema() config SkillConfig( timeout8.0, max_retries1, fallbacklambda: {status: failed, error_code: FALLBACK_WECOM_UNAVAILABLE} ) async def _run_core(self, validated_input: Dict[str, Any]) - Dict[str, Any]: # Get access token from cache or refresh access_token await self._get_access_token() # Build request url fhttps://qyapi.weixin.qq.com/cgi-bin/message/send?access_token{access_token} payload { touser: validated_input[user_id], msgtype: text, agentid: 100001, # your agent id text: { content: validated_input[content] } } if validated_input[at_all]: payload[text][mentioned_list] [all] async with httpx.AsyncClient(timeout5.0) as client: response await client.post(url, jsonpayload) if response.status_code ! 200: raise Exception(fWeCom API returned {response.status_code}: {response.text}) resp_json response.json() if resp_json.get(errcode) ! 0: raise Exception(fWeCom API error: {resp_json.get(errmsg)}) return { status: success, message_id: resp_json.get(msgid) } async def _get_access_token(self) - str: # Real implementation would cache token and refresh when expired # This is a stub return fake_access_token_12345关键细节说明WeComMessageInput和WeComMessageOutput使用Pydantic模型自动获得JSON Schema、文档生成、运行时校验能力_run_core中的httpx.AsyncClient显式设置timeout5.0与skill config的timeout8.0形成两级防护网络层5秒整体执行8秒错误处理分层清晰HTTP状态码错误 → WeCom业务错误 → 全局异常每层都有对应处理逻辑fallback是lambda确保无状态、无副作用避免fallback本身出问题。4.4 可观测性埋点没有监控的skill就是定时炸弹我们为每个skill强制注入以下监控指标skill_execution_count{skill_name, status}总执行次数按success/failed打标skill_execution_duration_seconds_bucket{skill_name, le}执行耗时直方图用于计算P95/P99skill_fallback_count{skill_name, fallback_level}各级fallback触发次数skill_retry_count{skill_name, attempt}各次重试触发次数。这些指标全部通过OpenTelemetry SDK上报与公司统一监控平台打通。当skill_execution_duration_seconds_bucket{skill_namesend_wecom_message, le0.5}的rate低于95%告警立刻触发SRE团队无需登录服务器直接在Grafana看板上就能定位是网络抖动、token过期还是WeCom接口限流。5. 常见问题与排查技巧实录那些文档里不会写的坑5.1 问题速查表高频故障现象与根因定位现象可能根因快速验证方法解决方案LLM反复调用同一skill但始终失败日志显示INPUT_VALIDATION_ERRORLLM生成的input字段名与skill契约不一致如skill要user_idLLM传userId查看skill输入校验日志提取原始input JSON对比schema字段名在LLM system prompt中加入字段名约束“所有skill输入字段必须使用snake_case命名如user_id、order_number”skill执行耗时稳定在timeout值附近如总是7.99秒但实际API响应很快skill内部有阻塞式IO如用requests而非httpxasyncio事件循环被卡住在_run_core开头加print(start, time.time())结尾加print(end, time.time())看是否真卡在业务逻辑彻底替换为异步库或用loop.run_in_executor包装阻塞调用fallback频繁触发但fallback逻辑本身也报错fallback函数引用了未初始化的外部变量如self.cache_client在fallback函数内加try/except打印完整tracebackfallback必须是纯函数所有依赖通过闭包传入禁止访问self同一input多次调用有时成功有时失败非幂等skill内部使用了全局变量或共享状态如单例连接池未做线程/协程安全用asyncio.create_task并发调用10次同一input观察成功率所有共享资源必须加锁或改用连接池如httpx.AsyncClient自带连接池LLM收到skill返回后仍继续调用其他无关skillskill output中status字段值不是严格字符串success或failed如写了Success检查output_schema中status的enum定义对比实际返回值严格按schema enum值返回禁止首字母大写等格式变化5.2 实操心得三个血泪教训教训一别信LLM的“自我宣称”一定要做契约验证我们曾上线一个“PDF文本提取skill”LLM在规划时说“我将调用PDF提取skill获取合同正文”结果skill返回{status: success, text: }空字符串。LLM却据此生成“合同内容为空”直接回复用户。后来发现是PDF扫描件OCR失败skill没做空内容检测。现在我们的强制规范是所有返回status: success的skill必须在output_schema中声明text字段的minLength: 1并在_validate_output中强制校验。空内容必须走status: failed分支。教训二重试不是万能的要区分“可重试”和“该重试”某次支付skill因银行接口限流返回429skill自动重试3次结果用户银行卡被连续扣款3次。根因在于429是服务端限流重试只会加重问题而真正的可重试错误是503服务不可用、504网关超时。现在我们的重试策略是只对5xx中的503/504/522Cloudflare错误重试其他错误一律不重试。这个规则写死在运行时框架里子类无法覆盖。教训三文档即代码契约变更必须触发LLM重新学习有一次我们把“查天气skill”的输入字段从city_name改为location只改了代码忘了更新LLM的system prompt。结果LLM继续传city_nameskill持续报INPUT_VALIDATION_ERROR。现在我们的CI/CD流程强制规定每次修改skill的input_schema或output_schema必须自动生成新的prompt片段并触发LLM微调任务。变更不是发布代码就结束而是“代码契约LLM认知”三者同步更新。5.3 性能压测实录一个skill的真实承载力我们对“企业微信消息发送skill”做了全链路压测结果如下环境4核8G云服务器WeCom接口QPS限制500并发数P50延迟(ms)P95延迟(ms)成功率备注10128210100%网络稳定token缓存命中率100%100142380100%连接池复用充分无新建连接开销500215125099.8%开始出现少量WeCom接口限流429skill自动降级到fallback1000350320092.1%限流加剧fallback触发率上升但无雪崩关键结论skill本身不是瓶颈WeCom接口才是。这印证了“能力边界”的重要性——我们必须在契约中明确写入“QPS上限500”让LLM在规划时主动做流量削峰P95延迟突破1秒时必须启动降级。我们在监控中设置了skill_execution_duration_seconds_bucket{le1.0}的rate告警低于95%即触发预案fallback的稳定性决定SLA。实测fallback平均耗时8msP9915ms确保即使主流程失败用户也能在1秒内得到响应。6. 能力治理如何让skill体系可持续演进6.1 Skill Catalog不是静态清单而是活的契约注册中心我们没有用Excel管理skill而是构建了一个GitOps驱动的Skill Catalog每个skill是一个独立Git仓库包含skill.py、pyproject.toml、README.md自动生成README.md由CI流水线根据input_schema/output_schema自动生成包含字段说明、示例、边界声明所有skill的name、description、boundary_statement被提取为YAML提交到中央Catalog仓库LLM在规划前会从Catalog拉取最新YAML作为context注入system prompt。这样做的好处是契约变更自动同步无需人工维护文档LLM永远看到最新能力视图。当新同事加入他不需要读文档直接git clone任意skill仓库make test就能跑通全部单元测试make doc就能生成最新文档。6.2 测试金字塔从单元到混沌缺一不可我们为skill设计了四层测试单元测试Unit TestMock所有外部依赖验证_run_core逻辑覆盖率≥90%契约测试Contract Test不Mock真实调用skill基类的execute方法只验证input/output schema符合性集成测试Integration Test对接真实下游服务如WeCom沙箱环境验证端到端流程混沌测试Chaos Test用Chaos Mesh注入网络延迟、DNS故障、HTTP 503错误验证重试/fallback是否生效。特别强调契约测试是核心防线。它确保无论下游服务如何变化只要skill返回符合schema的数据上层Agent就不会崩溃。我们曾用契约测试提前发现一个bug某skill在WeCom接口返回{errcode: 0}时错误地把errcode字段当成了status导致返回{status: 0}违反了status必须是字符串的契约。契约测试当场fail阻止了这个bug上线。6.3 演进路线从单点skill到能力网络当前阶段我们聚焦单个skill的健壮性。下一阶段目标是构建能力网络Capability Network能力编排Orchestration允许skill之间相互调用如“查天气skill”内部调用“城市编码转换skill”能力路由Routing根据输入特征自动选择最优skill如“查天气”时国内城市走WeCom内置API海外城市走第三方Weather API能力学习Learning记录LLM每次调用skill的input/output用强化学习优化skill选择策略。这条路很长但每一步都建立在坚实的契约之上。没有可靠的skill一切高级玩法都是空中楼阁。我在实际项目中最大的体会是不要追求“最聪明的LLM”而要打造“最靠谱的skill”。当每个能力单元都像瑞士手表一样精准、可预测、可替换整个Agent系统才能真正走出实验室走进每天要处理上千个真实请求的生产环境。这背后没有玄学只有对契约的敬畏、对边界的清醒、对细节的偏执。