拓冰网站定制 
说实话我最早对“agent 工具”的理解相当朴素写几个函数给 LLM 配上 JSON Schema模型能选对就行。直到我维护的一个对话代理在工具数量逼近 30 个之后突然开始“退化”——选错工具、参数乱填、同一个错误在一个会话里反复出现。我花了一段时间调查后明白问题根本不在于“这个工具写得不好”而在于“工具”这个抽象层级太小了。这个项目我给的代号是agent-skills最终沉淀下来的不是一堆更好的工具函数而是一套围绕“技能”的组织方式技能注册、技能索引、执行期生命周期、状态上下文、复合编排以及一套适用于技能而不是简单函数调用的可靠性治理。如果你正在做 function calling、Agent 编排或者手头出现“工具数量太多、质量参差、模型选择不准”的症状这篇文章应该能提供一个相对完整的解决方案思路。我会把核心模型、注册表、执行器的关键代码贴出来也会讲清楚每一步的取舍以及我在实战过程中踩过的坑。1. 工具数量一多代理就开始“退化”真正的根源在哪里先描述一个典型场景。项目早期只有 5 个工具接入 OpenAI function calling 或者 Anthropic tool use 都很顺模型基本能选对参数也基本规范。可等工具增加到 20 个以上肉眼可见的问题就冒出来了。第一个症状是 token 成本陡增。20 多个工具的完整 JSON Schema 一次全塞进系统提示词每次请求光工具定义就要烧掉两三千 token。这还不算完模型会在无关工具之间犹豫甚至为了“完成任务”强行调用一个不相干的工具。第二个症状是格式和语义不统一。有人把搜索参数命名为query又有人叫keywords底层其实是同一个搜索服务有的工具返回纯文本有的返回结构化 JSON代理后处理层根本没法统一。这种不一致会导致模型经常猜测参数的含义。第三个症状是错误恢复几乎为零。工具抛异常对话就断层了。没有重试没有降级更没有 fallback。用户问一句“帮我查一下昨天订单”数据库查询超时整个会话就废掉了。第四个症状是状态泄漏严重。多个工具共享同一个数据库连接和缓存却没人统一维护 session。请求一多就开始串数据A 用户的上下文跑到 B 用户那边。这四个症状指向同一个根因我把“工具调用”当成接口设计问题但它本质上是组织问题。一个函数只要“能被调用”就行一个技能则需要考虑描述、校验、上下文、失败路径、依赖关系和可观测性。我在 agent-skills 里给“技能”下了个明确的定义技能是可复用、可组合、可观测的代理能力单元。它包含输入输出契约、运行时上下文、生命周期钩子和失败策略而不是一个单纯的函数指针。从工程实现角度看“工具”和“技能”的差异体现在下面这张表里维度工具函数技能抽象层级单函数能力域 方法 策略输入校验散落在各函数内部统一 Schema 校验运行上下文无或函数自己管理独立 SkillContext依赖关系隐式靠开发者自觉显式声明 dependencies失败处理try/except 各写各的生命周期钩子统一治理可观测性日志记录结构化追踪 耗时 元数据组合方式基本没有Pipeline / Composite所以 agent-skills 里所有能力在注册表里暴露为技能。代理层只面对注册表的查询接口和执行接口不直接操作任何函数。这个抽象带来的直接好处是底层逻辑可以任意重构代理侧接入点几乎不变。2. 技能模型与注册表agent-skills 的地基怎么打技能系统最核心的不是执行逻辑而是“模型结构”和“注册机制”。这两件事没设计好后面全是补丁。2.1 AgentSkill 类技能的唯一契约边界我最初把技能设计成普通 dataclass然后发现不行——因为技能需要有执行函数、校验逻辑、生命周期钩子单一的 dataclass 会让使用方很难扩展。最后采用了“基类 可选执行函数 钩子”的组合方式。我把核心模型简化成这个样子# core.py from __future__ import annotations import time from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Optional class Hook(str, Enum): PRE_VALIDATE pre_validate PRE_EXECUTE pre_execute POST_EXECUTE post_execute ON_ERROR on_error dataclass class SkillResult: skill_name: str ok: bool output: Any None error: Optional[str] None duration_ms: float 0.0 meta: dict[str, Any] field(default_factorydict) def to_dict(self) - dict: return { skill_name: self.skill_name, ok: self.ok, output: self.output, error: self.error, duration_ms: self.duration_ms, meta: self.meta, } class AgentSkill: def __init__( self, name: str, description: str, input_schema: dict, tags: list[str] | None None, dependencies: list[str] | None None, execute_fn: Callable[[dict, SkillContext], Any] | None None, hooks: dict[Hook, Callable] | None None, ): self.name name self.description description self.input_schema input_schema self.tags tags or [] self.dependencies dependencies or [] self.execute_fn execute_fn self.hooks hooks or {}这个设计有一个关键点执行函数execute_fn接收两个参数第一个是入参 payload第二个是SkillContext。很多人写工具函数时只接收用户传来的参数完全没有办法感知会话级状态导致技能内部无法访问用户标识、缓存、配置等运行时信息。把SkillContext作为显式入参传进去是我觉得 agent-skills 最值得参考的一处设计。SkillResult则是唯一返回类型。不管技能内部发生什么执行器返回的一定是这个结构体代理层只需要判断ok字段再决定继续还是终止。同时duration_ms和meta天然支持了性能观测。2.2 SkillRegistry三种索引与技能发现注册表在 agent-skills 里承担的是“服务的注册与发现”职责。除了把技能对象存到一个字典里我额外维护了标签索引和域名索引方便运行时按照不同维度做候选集筛选。# registry.py class DuplicateSkillError(Exception): pass class SkillNotFoundError(Exception): pass class SkillRegistry: def __init__(self): self._skills: dict[str, AgentSkill] {} self._tag_index: dict[str, set[str]] {} self._domain_index: dict[str, set[str]] {} def register(self, skill: AgentSkill, domain: str default) - AgentSkill: if skill.name in self._skills: raise DuplicateSkillError(skill.name) self._skills[skill.name] skill self._domain_index.setdefault(domain, set()).add(skill.name) for tag in skill.tags: self._tag_index.setdefault(tag, set()).add(skill.name) return skill def unregister(self, name: str) - None: skill self._skills.pop(name, None) if skill is None: raise SkillNotFoundError(name) for tag in skill.tags: self._tag_index[tag].discard(name) def get(self, name: str) - AgentSkill: try: return self._skills[name] except KeyError: raise SkillNotFoundError(name) from None def search(self, query: str , tags: list[str] | None None) - list[AgentSkill]: query_l query.lower() scored [] for skill in self._skills.values(): if tags and not set(tags) set(skill.tags): continue pool f{skill.name} {skill.description} { .join(skill.tags)}.lower() score 0 for word in query_l.split(): if word.strip(): score int(word in pool) scored.append((score, skill)) scored.sort(keylambda x: (-x[0], x[1].name)) return [skill for score, skill in scored if score 0]这里search用的是最简单的关键词打分。它在真实项目里的效果已经足够好因为技能描述通常很短词频不高简单的包含匹配就能把错误候选过滤掉。如果你有更复杂的语义检索需求可以在这个函数里接入嵌入模型把pool向量化之后算余弦相似度接口保持不变。2.3 让模型“看到”技能的摘要层这里要解决一个很现实的问题技能很多但每次把全部技能定义丢给 LLM 是不可接受的。我在 agent-skills 里引入了“摘要层”的概念。什么是摘要层就是一个简化版的技能名录每条只有三样东西name、description、tags。代理在系统提示词里只看到这张名录而不是完整的 JSON Schema。当模型决定某个技能可能有用再通过一次工具调用把该技能的完整 schema 拉出来注入下一轮上下文。{ skill_candidates: [ { name: db_safe_query, description: 执行只读 SQL 查询适合获取数据库中的统计信息, tags: [db, readonly, statistics] }, { name: web_fetch_clean, description: 抓取网页并把正文转换为 Markdown适合调研和资料整理, tags: [web, search, markdown] } ] }这个“先选技能再填充 schema”的两段式调用能把 system prompt 里的工具定义占用的 token 压缩掉 70% 左右。模型注意力集中在“是否要用这个技能”上参数合法性由执行器兜底而不是让模型凭空猜测。3. 执行生命周期与 SkillContext技能不是裸函数技能与普通函数最大的区别是它有一套完整的生命周期。我把一次技能调用拆成六个阶段预处理输入校验依赖解析执行后处理错误兜底3.1 六段式生命周期设计对应到运行时大概是这样一个流程# runtime.py class SkillRuntime: def __init__( self, registry: SkillRegistry, context_provider: Callable[[], SkillContext], ): self.registry registry self.fetch_context context_provider def run(self, skill_name: str, payload: dict) - SkillResult: try: skill self.registry.get(skill_name) except SkillNotFoundError: return SkillResult(skill_nameskill_name, okFalse, errorskill_not_found) ctx self.fetch_context() start time.time() try: # 1. 预处理钩子 pre skill.hooks.get(Hook.PRE_VALIDATE) if pre: pre(payload, ctx) # 2. 输入校验 self._validate(skill, payload) # 3. 依赖解析 resolved self._resolve_dependencies(skill, payload, ctx) # 4. 执行 result skill.execute_fn(resolved, ctx) # 5. 后处理钩子 post skill.hooks.get(Hook.POST_EXECUTE) if post: result post(result, ctx) return SkillResult( skill_nameskill_name, okTrue, outputresult, duration_ms(time.time() - start) * 1000, ) except Exception as exc: # 6. 错误兜底 error_handler skill.hooks.get(Hook.ON_ERROR) if error_handler: output error_handler(exc, ctx) return SkillResult( skill_nameskill_name, okTrue, outputoutput, duration_ms(time.time() - start) * 1000, meta{recovered_from: str(exc)}, ) return SkillResult( skill_nameskill_name, okFalse, errorstr(exc), duration_ms(time.time() - start) * 1000, ) def _validate(self, skill: AgentSkill, payload: dict) - None: # 这里用 jsonschema 库做严格校验 from jsonschema import validate as json_validate json_validate(instancepayload, schemaskill.input_schema) def _resolve_dependencies(self, skill: AgentSkill, payload: dict, ctx: SkillContext) - dict: resolved dict(payload) for dep in skill.dependencies: if dep in payload: continue if dep in ctx.memory: resolved[dep] ctx.memory[dep] else: raise SkillDependencyError(skill.name, dep) return resolved这么设计有几个直接好处输入校验前置。如果 LLM 传了个非法参数在进执行函数之前就被拦住了不会出现“技能执行到一半才发现字段不存在”的尴尬。依赖解析单独拎出来。有些技能执行前需要一些公共变量比如当前用户 ID、数据库连接池、API Key。它们不在 LLM 的入参里而要从上下文记忆里取。这一步如果写在技能内部每个技能都要重复一次“从上下文里取参数”的逻辑很容易漏。错误兜底让技能有机会“再生产”。不是报了错就得把整个对话打断。比如数据库连接失败时ON_ERROR钩子可以自动切换只读缓存并把元数据标记成recovered_from让代理知道这次结果经过了降级处理不要盲目信任。3.2 SkillContext 与状态传递SkillContext在 agent-skills 里相当于 Web 开发中的 Request Context。每个技能执行时都会拿到当前会话的上下文但它本身不允许技能随意修改全局状态只能读除非显式写入 memory。# context.py import time class SkillContext: def __init__(self, scope: str, config: dict, memory: dict | None None): self.scope scope self.config config self.memory memory if memory is not None else {} self.started_at time.time() def set_memory(self, key: str, value) - None: self.memory[key] value def get_memory(self, key: str, defaultNone): return self.memory.get(key, default)scope用来区分上下文属于哪个会话。config存全局配置比如 API 端点、超时时间。memory是会话级缓存技能之间共享但生命周期由执行器统一管理。我遇到过不少人在做工具层时图省事直接把状态写在模块级全局变量里。这在单用户测试没问题多用户并发一上来就串数据。SkillContext强制把状态隔离在 scope 之内基本避免了这个问题。4. 组合与编排从原子技能到流水线任务单个技能很好写难的是组合。代理真正干活的场景往往是多步任务先查数据再处理格式最后生成图表。如果靠 LLM 一次工具调用把三件事全部做完不仅模型负担重出错的概率也会指数级上升。agent-skills 里我把技能分成两类原子技能和编排技能。原子技能只做一件事编排技能负责把多个技能串成流水线。4.1 PipelineSkill 的设计一个典型的流水线技能是这样定义的# pipeline.py class PipelineSkill(AgentSkill): def __init__( self, name: str, description: str, input_schema: dict, steps: list[dict], runtime: SkillRuntime, ): super().__init__( namename, descriptiondescription, input_schemainput_schema, tags[composite], ) self.steps steps self.runtime runtime def execute_fn(self, payload: dict, ctx: SkillContext) - Any: carry {} for step in self.steps: step_name step[skill] step_payload self._build_step_payload(step, payload, carry) result self.runtime.run(step_name, step_payload) if not result.ok: raise SkillPipelineError(step_name, result.error) carry[step_name] result.output return carrysteps是流水线的描述数组。每个 step 里包含技能名和参数映射规则参数可以从原始入参payload里取也可以从上一个技能的输出carry里取。4.2 编排时的数据映射数据映射是流程编排里最容易被忽略的环节。如果每个 step 都是“把上一步的整个输出塞给下一步”接口之间会形成强耦合。我在 agent-skills 里定义了一个简单的路径取值语法$step_name.field表示取名为step_name的技能输出中的field字段。def _build_step_payload(self, step: dict, payload: dict, carry: dict) - dict: raw_inputs step.get(inputs, {}) built {} for param_name, expr in raw_inputs.items(): if expr.startswith($): path expr[1:].split(.) source carry.get(path[0], {}) value source.get(path[1]) if len(path) 1 else source built[param_name] value else: built[param_name] expr return built举例来说生成一份业务分析快照可以组织成这样的流水线{ name: business_snapshot, steps: [ { skill: db_safe_query, inputs: {sql: SELECT * FROM orders WHERE created_at now() - interval 7 days, limit: 100} }, { skill: transform_for_chart, inputs: {raw_rows: $db_safe_query.rows} }, { skill: render_chart_plot, inputs: {series: $transform_for_chart.series, format: png} } ] }这种方案的好处是流水线定义完全数据化不绑定具体代码。你可以把流水线配置放到配置文件甚至数据库里运营同学也能通过后台调整步骤顺序而不用动代码。坏处自然也有配置化带来的调试困难。所以我后来加了一个dry_run模式可以在不真正调用外部服务的情况下把每步的输入输出打印出来方便排查问题。5. 可靠性治理技能要能报错也要能恢复在真实生产环境里我把每个技能当成一个“微服务”来对待。不是说要用微服务架构而是要有同样的治理意识输入校验、超时控制、重试策略、灰度发布、可观测性。这一节挑几个我在 agent-skills 里真正落地的东西来说。5.1 输入校验的三层防线第一层是 JSON Schema 严格校验。测试时我发现一个问题LLM 经常喜欢传多余的字段。比如技能定义里只接受query和limit模型非要塞一个filters。如果additionalProperties不设为false多余的字段就会静默透传导致技能内部函数签名出错。所以我所有技能 Schema 的additionalProperties都强制为false。第二层是字段值域校验。字符串类型不等于可以为空整数类型不等于可以随意大。比如数据库查询技能里的limit我一般限制在 1 到 100 之间超出直接拒绝防止模型一次倒腾出 10 万行数据。第三层是业务规则校验。这一层没法用通用 Schema 表达必须在 PRE_VALIDATE 钩子里写。比如“查询日期不得早于系统上线日期”“用户名不能包含特殊字符”。这些规则如果只依赖 Schema很快就会捅娄子。5.2 重试与降级策略我在 agent-skills 里给技能执行器加了重试策略表。每个技能可以声明自己的重试配置retry_policy { max_retries: 2, backoff_sec: 0.3, retry_on: [TimeoutError, DatabaseConnectionError], fallback_skill: db_safe_query_cached, }执行器在捕获到retry_on中的异常时会按照指数退避重试。如果重试仍然失败再看有没有fallback_skill有的话自动切换。这个降级逻辑非常有用。有一次我把主查询服务升级导致部分查询超时代理一秒切到了缓存版技能用户无感只有 meta 里多了一条recovered_from。5.3 技能回放测试普通单元测试只能验证技能函数本身但没法验证技能在“代理上下文”里是否好用。我在 agent-skills 里建立了一个“案例回放”机制每次真实调用结束后把入参、技能名、耗时、结果摘要落到一个测试用例目录里。代码一旦变更自动回放这些历史案例对比结果是否仍然符合预期。一个回放用例大概是这样的{ skill: db_safe_query, payload: {sql: SELECT count(*) FROM users WHERE statusactive, limit: 10}, assert: {ok: true, latency_ms_lte: 500} }回放不追求线上完全一致主要是捕捉两类问题技能是否突然报错、性能是否明显劣化。这个机制在技能系统持续迭代时非常重要因为修改一个底层技能很可能会影响好几个上层编排流程。没有回放测试回归只能靠人工去聊天框里不停试效率太低。6. 实战中踩过的几个大坑最后分享几个我在做 agent-skills 过程中真实踩过的坑每一个都让我在夜里调了很久。坑一技能命名随意导致选择混乱。早期我有个技能叫do_stuff描述写“做点相关的事”模型基本靠猜。后来我把技能命名改成“动词 对象”的模式例如db_safe_query、web_fetch_clean、chart_render_plot描述也改成“当用户需要……时使用本技能”。这个调整让技能选择准确率肉眼可见地上升。坑二描述里塞了太多限制条件。我一开始把“不使用该接口查询超过一年的数据”这种限制直接写进技能描述。后来发现模型经常因为描述太啰嗦而忽略真正的调用条件反而在生成参数时犹豫。更好的做法是描述里只写“做什么”限制条件放进 Schema 校验和业务规则校验里。坑三依赖解析做得太隐式。最初依赖是技能内部悄悄从全局变量里取的导致一个问题技能单独测试没问题放到代理编排里就各种拿不到值。后来我强制把依赖声明在dependencies字段里执行器统一解析问题就消失了。坑四技能输出没有一个标准结构。有的技能返回字符串有的返回 JSON代理层洗数据洗到崩溃。后来我规定技能输出必须是 JSON 序列化对象且至少包含data和meta两个字段。meta里放耗时、来源、置信度等辅助信息data放业务结果。代理层只读data观察层只读meta。坑五早期没有做 token 预算控制。即使有摘要层代理在处理长会话时依然可能把大量技能输出塞进历史消息里。后来我加了一个简单的 token 记账器在 POST_EXECUTE 钩子里累计每次输出的估算 token超出会话预算就直接压缩历史消息。这个机制保证了长会话可用性。如果你现在正在设计自己的代理工具层我的建议是别等到工具数量失控才重构从第一天就按“技能”来建模。保持技能内部高内聚外部通过注册表和生命周期统一管理组合任务交给流水线配置。这一套做下来代理的可维护性和稳定性会有质的提升。最后再分享一个我在实际迭代中的体会技能系统不是越复杂越好它要解决的真正问题只有一个——让代理能力的边界清晰、可管理、可组合、可观测。把这几件事做到位项目规模再大也不会在工具层烂成一锅粥。 从工具到技能：Agent 工具数量失控时的架构重构指南
说实话我最早对“agent 工具”的理解相当朴素写几个函数给 LLM 配上 JSON Schema模型能选对就行。直到我维护的一个对话代理在工具数量逼近 30 个之后突然开始“退化”——选错工具、参数乱填、同一个错误在一个会话里反复出现。我花了一段时间调查后明白问题根本不在于“这个工具写得不好”而在于“工具”这个抽象层级太小了。这个项目我给的代号是agent-skills最终沉淀下来的不是一堆更好的工具函数而是一套围绕“技能”的组织方式技能注册、技能索引、执行期生命周期、状态上下文、复合编排以及一套适用于技能而不是简单函数调用的可靠性治理。如果你正在做 function calling、Agent 编排或者手头出现“工具数量太多、质量参差、模型选择不准”的症状这篇文章应该能提供一个相对完整的解决方案思路。我会把核心模型、注册表、执行器的关键代码贴出来也会讲清楚每一步的取舍以及我在实战过程中踩过的坑。1. 工具数量一多代理就开始“退化”真正的根源在哪里先描述一个典型场景。项目早期只有 5 个工具接入 OpenAI function calling 或者 Anthropic tool use 都很顺模型基本能选对参数也基本规范。可等工具增加到 20 个以上肉眼可见的问题就冒出来了。第一个症状是 token 成本陡增。20 多个工具的完整 JSON Schema 一次全塞进系统提示词每次请求光工具定义就要烧掉两三千 token。这还不算完模型会在无关工具之间犹豫甚至为了“完成任务”强行调用一个不相干的工具。第二个症状是格式和语义不统一。有人把搜索参数命名为query又有人叫keywords底层其实是同一个搜索服务有的工具返回纯文本有的返回结构化 JSON代理后处理层根本没法统一。这种不一致会导致模型经常猜测参数的含义。第三个症状是错误恢复几乎为零。工具抛异常对话就断层了。没有重试没有降级更没有 fallback。用户问一句“帮我查一下昨天订单”数据库查询超时整个会话就废掉了。第四个症状是状态泄漏严重。多个工具共享同一个数据库连接和缓存却没人统一维护 session。请求一多就开始串数据A 用户的上下文跑到 B 用户那边。这四个症状指向同一个根因我把“工具调用”当成接口设计问题但它本质上是组织问题。一个函数只要“能被调用”就行一个技能则需要考虑描述、校验、上下文、失败路径、依赖关系和可观测性。我在 agent-skills 里给“技能”下了个明确的定义技能是可复用、可组合、可观测的代理能力单元。它包含输入输出契约、运行时上下文、生命周期钩子和失败策略而不是一个单纯的函数指针。从工程实现角度看“工具”和“技能”的差异体现在下面这张表里维度工具函数技能抽象层级单函数能力域 方法 策略输入校验散落在各函数内部统一 Schema 校验运行上下文无或函数自己管理独立 SkillContext依赖关系隐式靠开发者自觉显式声明 dependencies失败处理try/except 各写各的生命周期钩子统一治理可观测性日志记录结构化追踪 耗时 元数据组合方式基本没有Pipeline / Composite所以 agent-skills 里所有能力在注册表里暴露为技能。代理层只面对注册表的查询接口和执行接口不直接操作任何函数。这个抽象带来的直接好处是底层逻辑可以任意重构代理侧接入点几乎不变。2. 技能模型与注册表agent-skills 的地基怎么打技能系统最核心的不是执行逻辑而是“模型结构”和“注册机制”。这两件事没设计好后面全是补丁。2.1 AgentSkill 类技能的唯一契约边界我最初把技能设计成普通 dataclass然后发现不行——因为技能需要有执行函数、校验逻辑、生命周期钩子单一的 dataclass 会让使用方很难扩展。最后采用了“基类 可选执行函数 钩子”的组合方式。我把核心模型简化成这个样子# core.py from __future__ import annotations import time from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Optional class Hook(str, Enum): PRE_VALIDATE pre_validate PRE_EXECUTE pre_execute POST_EXECUTE post_execute ON_ERROR on_error dataclass class SkillResult: skill_name: str ok: bool output: Any None error: Optional[str] None duration_ms: float 0.0 meta: dict[str, Any] field(default_factorydict) def to_dict(self) - dict: return { skill_name: self.skill_name, ok: self.ok, output: self.output, error: self.error, duration_ms: self.duration_ms, meta: self.meta, } class AgentSkill: def __init__( self, name: str, description: str, input_schema: dict, tags: list[str] | None None, dependencies: list[str] | None None, execute_fn: Callable[[dict, SkillContext], Any] | None None, hooks: dict[Hook, Callable] | None None, ): self.name name self.description description self.input_schema input_schema self.tags tags or [] self.dependencies dependencies or [] self.execute_fn execute_fn self.hooks hooks or {}这个设计有一个关键点执行函数execute_fn接收两个参数第一个是入参 payload第二个是SkillContext。很多人写工具函数时只接收用户传来的参数完全没有办法感知会话级状态导致技能内部无法访问用户标识、缓存、配置等运行时信息。把SkillContext作为显式入参传进去是我觉得 agent-skills 最值得参考的一处设计。SkillResult则是唯一返回类型。不管技能内部发生什么执行器返回的一定是这个结构体代理层只需要判断ok字段再决定继续还是终止。同时duration_ms和meta天然支持了性能观测。2.2 SkillRegistry三种索引与技能发现注册表在 agent-skills 里承担的是“服务的注册与发现”职责。除了把技能对象存到一个字典里我额外维护了标签索引和域名索引方便运行时按照不同维度做候选集筛选。# registry.py class DuplicateSkillError(Exception): pass class SkillNotFoundError(Exception): pass class SkillRegistry: def __init__(self): self._skills: dict[str, AgentSkill] {} self._tag_index: dict[str, set[str]] {} self._domain_index: dict[str, set[str]] {} def register(self, skill: AgentSkill, domain: str default) - AgentSkill: if skill.name in self._skills: raise DuplicateSkillError(skill.name) self._skills[skill.name] skill self._domain_index.setdefault(domain, set()).add(skill.name) for tag in skill.tags: self._tag_index.setdefault(tag, set()).add(skill.name) return skill def unregister(self, name: str) - None: skill self._skills.pop(name, None) if skill is None: raise SkillNotFoundError(name) for tag in skill.tags: self._tag_index[tag].discard(name) def get(self, name: str) - AgentSkill: try: return self._skills[name] except KeyError: raise SkillNotFoundError(name) from None def search(self, query: str , tags: list[str] | None None) - list[AgentSkill]: query_l query.lower() scored [] for skill in self._skills.values(): if tags and not set(tags) set(skill.tags): continue pool f{skill.name} {skill.description} { .join(skill.tags)}.lower() score 0 for word in query_l.split(): if word.strip(): score int(word in pool) scored.append((score, skill)) scored.sort(keylambda x: (-x[0], x[1].name)) return [skill for score, skill in scored if score 0]这里search用的是最简单的关键词打分。它在真实项目里的效果已经足够好因为技能描述通常很短词频不高简单的包含匹配就能把错误候选过滤掉。如果你有更复杂的语义检索需求可以在这个函数里接入嵌入模型把pool向量化之后算余弦相似度接口保持不变。2.3 让模型“看到”技能的摘要层这里要解决一个很现实的问题技能很多但每次把全部技能定义丢给 LLM 是不可接受的。我在 agent-skills 里引入了“摘要层”的概念。什么是摘要层就是一个简化版的技能名录每条只有三样东西name、description、tags。代理在系统提示词里只看到这张名录而不是完整的 JSON Schema。当模型决定某个技能可能有用再通过一次工具调用把该技能的完整 schema 拉出来注入下一轮上下文。{ skill_candidates: [ { name: db_safe_query, description: 执行只读 SQL 查询适合获取数据库中的统计信息, tags: [db, readonly, statistics] }, { name: web_fetch_clean, description: 抓取网页并把正文转换为 Markdown适合调研和资料整理, tags: [web, search, markdown] } ] }这个“先选技能再填充 schema”的两段式调用能把 system prompt 里的工具定义占用的 token 压缩掉 70% 左右。模型注意力集中在“是否要用这个技能”上参数合法性由执行器兜底而不是让模型凭空猜测。3. 执行生命周期与 SkillContext技能不是裸函数技能与普通函数最大的区别是它有一套完整的生命周期。我把一次技能调用拆成六个阶段预处理输入校验依赖解析执行后处理错误兜底3.1 六段式生命周期设计对应到运行时大概是这样一个流程# runtime.py class SkillRuntime: def __init__( self, registry: SkillRegistry, context_provider: Callable[[], SkillContext], ): self.registry registry self.fetch_context context_provider def run(self, skill_name: str, payload: dict) - SkillResult: try: skill self.registry.get(skill_name) except SkillNotFoundError: return SkillResult(skill_nameskill_name, okFalse, errorskill_not_found) ctx self.fetch_context() start time.time() try: # 1. 预处理钩子 pre skill.hooks.get(Hook.PRE_VALIDATE) if pre: pre(payload, ctx) # 2. 输入校验 self._validate(skill, payload) # 3. 依赖解析 resolved self._resolve_dependencies(skill, payload, ctx) # 4. 执行 result skill.execute_fn(resolved, ctx) # 5. 后处理钩子 post skill.hooks.get(Hook.POST_EXECUTE) if post: result post(result, ctx) return SkillResult( skill_nameskill_name, okTrue, outputresult, duration_ms(time.time() - start) * 1000, ) except Exception as exc: # 6. 错误兜底 error_handler skill.hooks.get(Hook.ON_ERROR) if error_handler: output error_handler(exc, ctx) return SkillResult( skill_nameskill_name, okTrue, outputoutput, duration_ms(time.time() - start) * 1000, meta{recovered_from: str(exc)}, ) return SkillResult( skill_nameskill_name, okFalse, errorstr(exc), duration_ms(time.time() - start) * 1000, ) def _validate(self, skill: AgentSkill, payload: dict) - None: # 这里用 jsonschema 库做严格校验 from jsonschema import validate as json_validate json_validate(instancepayload, schemaskill.input_schema) def _resolve_dependencies(self, skill: AgentSkill, payload: dict, ctx: SkillContext) - dict: resolved dict(payload) for dep in skill.dependencies: if dep in payload: continue if dep in ctx.memory: resolved[dep] ctx.memory[dep] else: raise SkillDependencyError(skill.name, dep) return resolved这么设计有几个直接好处输入校验前置。如果 LLM 传了个非法参数在进执行函数之前就被拦住了不会出现“技能执行到一半才发现字段不存在”的尴尬。依赖解析单独拎出来。有些技能执行前需要一些公共变量比如当前用户 ID、数据库连接池、API Key。它们不在 LLM 的入参里而要从上下文记忆里取。这一步如果写在技能内部每个技能都要重复一次“从上下文里取参数”的逻辑很容易漏。错误兜底让技能有机会“再生产”。不是报了错就得把整个对话打断。比如数据库连接失败时ON_ERROR钩子可以自动切换只读缓存并把元数据标记成recovered_from让代理知道这次结果经过了降级处理不要盲目信任。3.2 SkillContext 与状态传递SkillContext在 agent-skills 里相当于 Web 开发中的 Request Context。每个技能执行时都会拿到当前会话的上下文但它本身不允许技能随意修改全局状态只能读除非显式写入 memory。# context.py import time class SkillContext: def __init__(self, scope: str, config: dict, memory: dict | None None): self.scope scope self.config config self.memory memory if memory is not None else {} self.started_at time.time() def set_memory(self, key: str, value) - None: self.memory[key] value def get_memory(self, key: str, defaultNone): return self.memory.get(key, default)scope用来区分上下文属于哪个会话。config存全局配置比如 API 端点、超时时间。memory是会话级缓存技能之间共享但生命周期由执行器统一管理。我遇到过不少人在做工具层时图省事直接把状态写在模块级全局变量里。这在单用户测试没问题多用户并发一上来就串数据。SkillContext强制把状态隔离在 scope 之内基本避免了这个问题。4. 组合与编排从原子技能到流水线任务单个技能很好写难的是组合。代理真正干活的场景往往是多步任务先查数据再处理格式最后生成图表。如果靠 LLM 一次工具调用把三件事全部做完不仅模型负担重出错的概率也会指数级上升。agent-skills 里我把技能分成两类原子技能和编排技能。原子技能只做一件事编排技能负责把多个技能串成流水线。4.1 PipelineSkill 的设计一个典型的流水线技能是这样定义的# pipeline.py class PipelineSkill(AgentSkill): def __init__( self, name: str, description: str, input_schema: dict, steps: list[dict], runtime: SkillRuntime, ): super().__init__( namename, descriptiondescription, input_schemainput_schema, tags[composite], ) self.steps steps self.runtime runtime def execute_fn(self, payload: dict, ctx: SkillContext) - Any: carry {} for step in self.steps: step_name step[skill] step_payload self._build_step_payload(step, payload, carry) result self.runtime.run(step_name, step_payload) if not result.ok: raise SkillPipelineError(step_name, result.error) carry[step_name] result.output return carrysteps是流水线的描述数组。每个 step 里包含技能名和参数映射规则参数可以从原始入参payload里取也可以从上一个技能的输出carry里取。4.2 编排时的数据映射数据映射是流程编排里最容易被忽略的环节。如果每个 step 都是“把上一步的整个输出塞给下一步”接口之间会形成强耦合。我在 agent-skills 里定义了一个简单的路径取值语法$step_name.field表示取名为step_name的技能输出中的field字段。def _build_step_payload(self, step: dict, payload: dict, carry: dict) - dict: raw_inputs step.get(inputs, {}) built {} for param_name, expr in raw_inputs.items(): if expr.startswith($): path expr[1:].split(.) source carry.get(path[0], {}) value source.get(path[1]) if len(path) 1 else source built[param_name] value else: built[param_name] expr return built举例来说生成一份业务分析快照可以组织成这样的流水线{ name: business_snapshot, steps: [ { skill: db_safe_query, inputs: {sql: SELECT * FROM orders WHERE created_at now() - interval 7 days, limit: 100} }, { skill: transform_for_chart, inputs: {raw_rows: $db_safe_query.rows} }, { skill: render_chart_plot, inputs: {series: $transform_for_chart.series, format: png} } ] }这种方案的好处是流水线定义完全数据化不绑定具体代码。你可以把流水线配置放到配置文件甚至数据库里运营同学也能通过后台调整步骤顺序而不用动代码。坏处自然也有配置化带来的调试困难。所以我后来加了一个dry_run模式可以在不真正调用外部服务的情况下把每步的输入输出打印出来方便排查问题。5. 可靠性治理技能要能报错也要能恢复在真实生产环境里我把每个技能当成一个“微服务”来对待。不是说要用微服务架构而是要有同样的治理意识输入校验、超时控制、重试策略、灰度发布、可观测性。这一节挑几个我在 agent-skills 里真正落地的东西来说。5.1 输入校验的三层防线第一层是 JSON Schema 严格校验。测试时我发现一个问题LLM 经常喜欢传多余的字段。比如技能定义里只接受query和limit模型非要塞一个filters。如果additionalProperties不设为false多余的字段就会静默透传导致技能内部函数签名出错。所以我所有技能 Schema 的additionalProperties都强制为false。第二层是字段值域校验。字符串类型不等于可以为空整数类型不等于可以随意大。比如数据库查询技能里的limit我一般限制在 1 到 100 之间超出直接拒绝防止模型一次倒腾出 10 万行数据。第三层是业务规则校验。这一层没法用通用 Schema 表达必须在 PRE_VALIDATE 钩子里写。比如“查询日期不得早于系统上线日期”“用户名不能包含特殊字符”。这些规则如果只依赖 Schema很快就会捅娄子。5.2 重试与降级策略我在 agent-skills 里给技能执行器加了重试策略表。每个技能可以声明自己的重试配置retry_policy { max_retries: 2, backoff_sec: 0.3, retry_on: [TimeoutError, DatabaseConnectionError], fallback_skill: db_safe_query_cached, }执行器在捕获到retry_on中的异常时会按照指数退避重试。如果重试仍然失败再看有没有fallback_skill有的话自动切换。这个降级逻辑非常有用。有一次我把主查询服务升级导致部分查询超时代理一秒切到了缓存版技能用户无感只有 meta 里多了一条recovered_from。5.3 技能回放测试普通单元测试只能验证技能函数本身但没法验证技能在“代理上下文”里是否好用。我在 agent-skills 里建立了一个“案例回放”机制每次真实调用结束后把入参、技能名、耗时、结果摘要落到一个测试用例目录里。代码一旦变更自动回放这些历史案例对比结果是否仍然符合预期。一个回放用例大概是这样的{ skill: db_safe_query, payload: {sql: SELECT count(*) FROM users WHERE statusactive, limit: 10}, assert: {ok: true, latency_ms_lte: 500} }回放不追求线上完全一致主要是捕捉两类问题技能是否突然报错、性能是否明显劣化。这个机制在技能系统持续迭代时非常重要因为修改一个底层技能很可能会影响好几个上层编排流程。没有回放测试回归只能靠人工去聊天框里不停试效率太低。6. 实战中踩过的几个大坑最后分享几个我在做 agent-skills 过程中真实踩过的坑每一个都让我在夜里调了很久。坑一技能命名随意导致选择混乱。早期我有个技能叫do_stuff描述写“做点相关的事”模型基本靠猜。后来我把技能命名改成“动词 对象”的模式例如db_safe_query、web_fetch_clean、chart_render_plot描述也改成“当用户需要……时使用本技能”。这个调整让技能选择准确率肉眼可见地上升。坑二描述里塞了太多限制条件。我一开始把“不使用该接口查询超过一年的数据”这种限制直接写进技能描述。后来发现模型经常因为描述太啰嗦而忽略真正的调用条件反而在生成参数时犹豫。更好的做法是描述里只写“做什么”限制条件放进 Schema 校验和业务规则校验里。坑三依赖解析做得太隐式。最初依赖是技能内部悄悄从全局变量里取的导致一个问题技能单独测试没问题放到代理编排里就各种拿不到值。后来我强制把依赖声明在dependencies字段里执行器统一解析问题就消失了。坑四技能输出没有一个标准结构。有的技能返回字符串有的返回 JSON代理层洗数据洗到崩溃。后来我规定技能输出必须是 JSON 序列化对象且至少包含data和meta两个字段。meta里放耗时、来源、置信度等辅助信息data放业务结果。代理层只读data观察层只读meta。坑五早期没有做 token 预算控制。即使有摘要层代理在处理长会话时依然可能把大量技能输出塞进历史消息里。后来我加了一个简单的 token 记账器在 POST_EXECUTE 钩子里累计每次输出的估算 token超出会话预算就直接压缩历史消息。这个机制保证了长会话可用性。如果你现在正在设计自己的代理工具层我的建议是别等到工具数量失控才重构从第一天就按“技能”来建模。保持技能内部高内聚外部通过注册表和生命周期统一管理组合任务交给流水线配置。这一套做下来代理的可维护性和稳定性会有质的提升。最后再分享一个我在实际迭代中的体会技能系统不是越复杂越好它要解决的真正问题只有一个——让代理能力的边界清晰、可管理、可组合、可观测。把这几件事做到位项目规模再大也不会在工具层烂成一锅粥。