拓冰网站定制 
我最近在重构智能体项目时彻底受够了那种模型提示词工具的快餐式架构。一开始确实跑得欢但等技能一多——搜索、代码执行、文件处理、数据分析全堆在一起——代码就开始失控了。工具之间的边界模糊、上下文互相污染、权限管理形同虚设改一个功能得连带调试三个模块。后来我干脆推倒重来把整个体系按照一套技能标准重新设计了一遍也就是这个agent-skills项目。简单说它不是一个具体的业务工具而是一套给 Agent 做技能插拔的框架每个独立能力都封装成一个自包含的 Skill 模块统一输入输出、统一注册与发现机制、统一运行沙箱和权限策略Agent 通过路由层按需调用。这套设计对正在头疼工具越来越多、越来越乱的开发者特别有参考价值尤其是做 AI 应用落地、RAG 增强、自动化工作流的朋友。这篇博客我会从整体设计思路讲起然后拆解一个技能模块的核心结构再带你把一个代码执行 数据分析的技能完整实现一遍最后整理我在实际落地过程中踩过的一些坑和排查心得。整个过程尽量讲清楚为什么这么设计而不只是给一堆代码片段。1. 技能插拔背后的架构思路把工具调用升级为可治理的技能体系1.1 从工具到技能一次性解决上下文污染和权限失控很多人在给 Agent 加能力的时候第一反应就是往 system prompt 里塞定义或者在代码里堆一堆if function_name xxx的分支。这种方式的痛点在早期不明显但一旦技能数量超过 10 个问题就暴露得非常快Agent 根本分不清什么时候该用哪个工具工具间频繁互相调用导致上下文爆炸而且没有任何统一的权限控制手段——套一次 shell 命令就能让 Agent 做超出预期的事情。我在agent-skills里做的第一件关键设计是把工具这个概念彻底替换为技能。这两者的本质区别在于工具是函数的被动暴露而技能是一个自包含的主动性单元。什么意思呢一个技能模块不仅包含能做什么元信息、怎么做执行函数还包含什么时候用触发条件、允许多大权限权限声明、依赖什么资源资源清单甚至包括失败后怎么恢复。这种信息 self-contained 的设计让 Agent 在决策时拿到了足够多的上下文而不需要靠猜。举个实际例子。假设你要给 Agent 加一个运行 Python 代码的技能。如果只是简单暴露一个exec函数那 Agent 可能在任何场景下都试图执行代码哪怕是在只该做文本处理的任务里。但如果你定义了一个PythonExecutorSkill声明了when_to_use字段比如仅当需要计算数学表达式、处理数据文件、或生成可视化图表时使用Agent 的决策准确度会明显上升。同时permissions字段声明允许读取 /data 目录禁止访问网络这样即便 Agent 写了一段恶意的网络请求代码沙箱层也会直接拦截。1.2 技能注册与发现为什么路由表比手写判断更可靠整个技能系统的骨架是一个技能注册中心。所有技能在系统启动时通过装饰器或配置文件注册进去形成一个统一的技能路由表。这个路由表不仅仅是名字到函数的映射它内部维护了一份技能元数据索引技能名称、描述、参数 schema、触发关键词、期望的输出格式、权限级别一应俱全。我不建议在代码里手写if skill_name python_executor这类逻辑来做分发。原因是 Agent 在生成调用指令时往往带有自然语言的模糊性比如它可能会说帮我跑下这段代码看看结果而技能名称可能是python_executor。这时候你需要的是一个语义匹配层。我用了一个比较轻量的做法把每个技能的描述文本做向量化Agent 的自然语言请求也做向量化然后算相似度。当然如果你不想引入向量库也可以用关键词 规则做兜底——这个后面我会详细说。注册中心还承担了另一个重要职责技能可用性探测。有些技能依赖外部服务或模型比如某个技能需要调用一个 OCR 服务如果服务挂了还注册为可用Agent 就会反复尝试然后失败。我在注册中心做了健康检查机制每次 Agent 发起调用前先查一下技能状态不可用的直接不进候选列表。这个设计大大减少了 Agent 决策的无效尝试率。1.3 技能编排顺序执行、条件分支与并行调度单技能调用只是基本功真正让 Agent 能干复杂活的是技能的编排能力。我在agent-skills里定义了一种简单的技能编排 DSL支持三种基本组合方式顺序执行、条件分支和并行调度。顺序执行很好理解比如先读取文件再分析内容再生成摘要。条件分支则是根据上一步的输出决定下一步执行哪个技能比如如果文件是图片格式调用图像识别技能如果是文本格式调用文本分析技能。并行调度用于多个相对独立的技能同时执行比如同时搜索多个数据源再汇总可以显著缩短整体耗时。这里的核心难点在于技能间的数据传递。每个技能的输出不能只是简单的字符串需要一个结构化的数据块包含状态码、结构化结果、中间产物引用、错误信息等。这样下一个技能才能准确地定位到需要的数据。我设计了一个SkillResult类统一封装输出格式这在系统里是一个非常底层的约定。2. 核心细节解析定义一份技能模块的标准接口2.1 技能基类 Skill六大组成部分的职责划分关于技能模块的标准接口我最终沉淀出来六个核心组成部分缺一不可。在很多工具调用框架里开发者只看重函数签名但在我这个体系里函数签名只是其中一小部分。第一个是meta元信息。包含技能名称、版本、作者、描述、触发场景示例。描述信息我建议写得越详细越好因为这是 Agent 决定是否调用你的核心依据。第二个是parameters_schema参数声明用 JSON Schema 格式定义这个技能需要哪些输入Agent 在调用前会先按照 schema 生成参数不合法的会被前置校验拦截。第三个是permissions权限声明定义这个技能可以访问哪些文件路径、能否联网、能否执行系统命令等。第四个是execute方法这是技能的实际执行逻辑。第五个是validate_result结果校验方法确保技能输出符合预期的格式。第六个是recover异常恢复方法定义了当执行失败时应该尝试哪些兜底策略。这六个部分组合在一起就形成了一个完备的技能定义。下面是我项目中base_skill.py的核心代码删减了业务逻辑只保留骨架from abc import ABC, abstractmethod from enum import Enum from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class SkillPermission(BaseModel): # 允许访问的文件路径前缀例如 [/data, /tmp]空列表表示禁止文件访问 allowed_file_paths: List[str] Field(default_factorylist) # 是否允许网络访问 allow_network: bool False # 是否允许执行系统命令 allow_system_command: bool False # 资源限额最大执行时间秒、最大输出长度 max_execution_time: int 30 max_output_length: int 8192 class SkillMeta(BaseModel): name: str Field(..., description技能唯一名称) version: str Field(default0.1.0) description: str Field(..., description对技能能力的详细描述) when_to_use: str Field(..., description说明什么场景下使用该技能) when_not_to_use: str Field(default, description说明什么场景下不要使用该技能非常重要) tags: List[str] Field(default_factorylist) author: str Field(defaultunknown) class SkillResult(BaseModel): skill_name: str status: str success # success | failed | partial data: Any None message: str metadata: Dict[str, Any] Field(default_factorydict) # 如果执行过程中产出了中间文件把文件路径放在这里供依赖链上的后续技能使用 artifacts: List[str] Field(default_factorylist) duration_ms: int 0 error_type: str error_detail: str class BaseSkill(ABC): # 技能元信息 meta: SkillMeta # 参数声明遵循 JSON Schema 格式 parameters_schema: Dict[str, Any] {} # 权限声明 permissions: SkillPermission SkillPermission() # 依赖的技能名称列表 dependencies: List[str] [] def __init__(self, context: Optional[Dict[str, Any]] None): # context 是全局共享上下文比如知识库连接、日志器、配置项等 self.context context or {} abstractmethod def execute(self, params: Dict[str, Any]) - SkillResult: pass def validate_parameters(self, params: Dict[str, Any]) - List[str]: 参数校验返回错误信息列表空列表表示校验通过 errors [] # 这里用 jsonschema 做校验为了简洁省略具体实现 return errors def validate_result(self, result: SkillResult) - bool: # 校验结果是否符合预期用于编排引擎判断是否继续走 DAG return result.status success def recover(self, params: Dict[str, Any], error: Exception) - SkillResult: 技能执行失败后的兜底逻辑默认返回错误子类可覆盖 return SkillResult( skill_nameself.meta.name, statusfailed, messageunhandled error, error_typetype(error).__name__, error_detailstr(error), )关于这段代码我想特别强调几个容易被忽视的点。when_not_to_use这个字段必须有。它和when_to_use一样重要甚至更重要。因为 Agent 的误调用很多时候不是因为不知道该用什么而是因为不知道不该用什么。比如一个负责生成文案的 Agent如果不声明当用户只是闲聊时不要调用本技能它可能在任何对话场景下都触发文本生成技能导致回复内容极其生硬。SkillResult里我设计了artifacts字段专门存中间产物的引用。举个例子代码执行技能运行完一段 Python 后可能生成了一张图表chart.png这个文件的路径会被放进artifacts这样后续的文档生成技能就能直接引用它而不需要重新执行一遍代码。recover方法的设计来自一次惨痛教训。之前没有恢复机制时只要一个技能执行失败整个任务链就断了。后来我意识到很多失败其实是有固定模式的API 超时可以重试一次文件不存在可以自动切换到备用路径格式解析失败可以尝试另一种解析策略。把这些常见恢复逻辑放到recover里Agent 的任务成功率提升非常明显。2.2 技能编排引擎DAG 调度与中间产物管理有了标准接口之后接下来的核心问题就是怎么把多个技能组装成一个工作流。我在agent-skills里实现了一个轻量级的编排引擎基于有向无环图DAG来管理技能之间的依赖关系。为什么用 DAG 而不是简单顺序执行呢因为实际任务中技能之间往往存在部分并行的关系。比如一个调研任务需要同时调用网页搜索技能、知识库检索技能和舆情分析技能三个技能之间没有依赖可以并行但它们的输出都会汇总到最后的报告生成技能里。顺序执行会浪费大量时间而 DAG 调度器可以自动识别可并行的节点。编排引擎的基本工作流程是先把一个复杂任务分解为若干技能节点根据依赖关系构建 DAG然后按拓扑序遍历符合条件的并行节点同时执行。每个节点执行完会产出SkillResult引擎根据validate_result判断是否继续执行下游节点。如果遇到失败节点会先调用recover尝试修复修复不了就中止整个 DAG并向 Agent 返回失败的原因链。中间产物管理是这个引擎的另一大重头戏。我用一个内存对象存储维护所有技能节点的输出同时支持把大型中间结果落盘到临时目录。对于一份很大的数据文件不会直接塞进内存传递给下一个节点而是写入/tmp/skill_artifacts/目录然后在节点间传递文件路径。这个设计对长任务的稳定性非常关键——我之前就碰到过因为大对象占满内存导致整个 Agent 崩掉的情况改成落盘方案后彻底解决了。关于 DAG 调度有个小技巧值得分享在 Agent 生成技能编排图的时候不要让它自由发挥而是提供一个技能编排模板库。预先定义一些常见任务的固定流程比如数据报告生成流程、竞品调研流程、日志排查流程等Agent 只需要从模板里选择并填充参数。这样既保证了执行效率又减少了编排结构错误的风险。2.3 权限沙箱技能执行的安全边界到底怎么画技能系统如果要外部扩展安全就是第一优先级。我给不同级别的技能划分了四层沙箱策略可以根据技能的可信度灵活选择。第一层是纯逻辑层适用于不涉及外部资源操作的技能比如 JSON 格式化、文本处理、时间计算。这种技能不需要文件访问也不需要网络直接在进程内跑就行了。第二层是文件沙箱适用于需要读写文件但不能联网的技能。做法是把文件访问强制限制在声明的前缀路径下用os.path.realpath先做路径归一化再判断是否在允许范围内防目录穿越攻击。第三层是网络白名单适用于需要调用外部 API 的技能。这里我用了一个本地代理强制所有 HTTP 请求通过代理转发代理只放行预先配置的域名白名单。第四层是完整进程沙箱适用于执行不可信代码的场景比如运行用户提交的 Python 代码。这里我把执行放到一个独立的子进程里面并且设置了 CPU 时间限制、内存限制、文件系统只读挂载用resource模块监控资源使用情况。这里专门说一下 python 代码执行技能的子进程沙箱实现。完整方案是这样的父进程通过 subprocess 启动一个子 Python 进程子进程预先加载受限的builtins删除__import__或者替换为受限版本然后通过管道传入要执行的代码在子进程中用exec(compiled_code, restricted_globals)执行执行结果 queue 回传给父进程。import os import resource import subprocess import tempfile from typing import Dict, Any class CodeSandbox: def __init__(self, allowed_paths: list[str] | None None, memory_limit_mb: int 512, timeout_s: int 10): self.allowed_paths allowed_paths or [/tmp/skill_artifacts] self.memory_limit_mb memory_limit_mb self.timeout_s timeout_s def run(self, code: str) - Dict[str, Any]: 在子进程 资源限制下执行一段 Python 代码 # 临时目录作为工作目录避免污染主进程目录 workdir tempfile.mkdtemp(prefixsandbox_, dir/tmp) with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse, encodingutf-8) as f: f.write(self._wrap_code(code)) fpath f.name try: proc subprocess.run( [/usr/bin/python3, fpath], capture_outputTrue, textTrue, timeoutself.timeout_s, cwdworkdir, env{...}, # 精简环境变量 ) return { stdout: proc.stdout, stderr: proc.stderr, returncode: proc.returncode, } except subprocess.TimeoutExpired: return {stdout: , stderr: execution timeout, returncode: -1} finally: os.remove(fpath) # 清理临时目录可以按需保留用于调试 def _wrap_code(self, code: str) - str: return f import builtins # 移除危险的内置函数 _original_import builtins.__import__ def _restricted_import(name, *args, **kwargs): raise ImportError(fimport not allowed: {{name}}) builtins.__import__ _restricted_import # 这里可以进一步替换 open 等内置函数 import sys import resource # 设置内存限制 resource.setrlimit(resource.RLIMIT_AS, (self.memory_limit_mb * 1024 * 1024, self.memory_limit_mb * 1024 * 1024)) # 设置 CPU 限制 resource.setrlimit(resource.RLIMIT_CPU, (self.timeout_s, self.timeout_s)) _exec_globals {{}} try: exec(compile({code!r}, sandbox, exec), _exec_globals) except Exception as e: print(fEXEC_ERROR: {{type(e).__name__}}: {{e}}, filesys.stderr) sys.exit(1) 提示不要把沙箱当成绝对安全边界。对于高风险的不可信代码最稳妥的做法还是直接跑容器隔离或者调用云端的代码执行 API。子进程沙箱只是提高了攻击门槛适合内部工具这类中等威胁场景。这段实现里有两个容易被忽略的细节一是要自定义fpath写到/tmp而不是放在当前工作目录避免权限混乱二是resource.setrlimit(RLIMIT_CPU)必须放在子进程里设置如果在父进程设置会影响主程序自己。我第一次做的时候就是在父进程设了内存限制结果 Agent 主程序直接被 OOM 杀掉了非常尴尬。3. 实操过程从零实现一个代码执行 数据分析技能3.1 场景设计与参数声明让 Agent 清楚地知道该不该调你好现在进入最实际的部分。我带你完整实现一个DataAnalysisSkill它的作用是执行一段 Python 数据分析代码并且生成统计摘要和可视化图表。这个技能在日常业务里用得非常频繁非常适合作为参考案例。第一步是设计它的SkillMeta和参数声明。这个环节建议不要急躁因为你写得越细致Agent 的调用准确率就越高。下面是我在实际项目中用到的版本class DataAnalysisSkill(BaseSkill): meta SkillMeta( namedata_analysis, version1.0.0, description执行 Python 数据分析和可视化代码支持 pandas、numpy、matplotlib。可以读取已落盘的 CSV/JSON 数据文件输出统计结果和图表。, when_to_use当用户要求分析结构化数据、计算统计指标、绘制图表、处理 CSV/Excel/JSON 数据文件时使用。尤其是涉及数据集、报表、实验数据等场景。, when_not_to_use当用户只是问一般性问题、要求写非数据类的代码、或者数据尚未准备完成时不要使用本技能。, tags[data, analysis, visualization], ) parameters_schema { type: object, properties: { code: { type: string, description: 要执行的 Python 代码必须以纯代码片段形式提供不要包含 markdown 代码块标记。, }, data_files: { type: array, items: {type: string}, description: 需要读取的数据文件路径列表必须是沙箱允许访问的路径。, }, output_charts: { type: boolean, description: 是否生成可视化图表默认 true。, default: True, }, }, required: [code], } permissions SkillPermission( allowed_file_paths[/tmp/skill_artifacts, /data], allow_networkFalse, allow_system_commandFalse, max_execution_time60, max_output_length8192, ) dependencies [file_access]我特别要讲一下dependencies字段。DataAnalysisSkill依赖file_access技能但依赖不代表它直接调用文件技能而是它要求文件技能必须先执行完把文件落到指定的/tmp/skill_artifacts目录之后本技能才有东西可分析。这相当于隐性传递了一个数据契约上一个技能负责产出文件本技能负责消费文件。还有一点description里的措辞非常重要。我见过很多人写技能描述时过于笼统比如这个技能可以分析数据Agent 看了一脸懵不知道和别的技能有什么差异。尽量写得具体、带有典型场景关键词比如我上面写了统计指标CSV/Excel/JSON 数据文件这些词Agent 在语义匹配时的命中率会高很多。3.2 核心执行逻辑写一个既能跑又扛造的 execute 方法接下来是execute方法。这一步主要做几件事先校验参数然后调用代码沙箱执行用户传入的代码再解析执行结果主要是 stdout 和产生的图表文件最后封装成SkillResult返回。import json import os import re import time from typing import Dict, Any class DataAnalysisSkill(BaseSkill): def execute(self, params: Dict[str, Any]) - SkillResult: start_time time.time() # 1. 参数校验 errors self.validate_parameters(params) if errors: return SkillResult( skill_nameself.meta.name, statusfailed, message参数校验失败: ; .join(errors), error_typeParameterError, error_detailerrors, duration_msint((time.time() - start_time) * 1000), ) code params[code] data_files params.get(data_files, []) output_charts params.get(output_charts, True) # 2. 检查数据文件是否存在 for f in data_files: if not os.path.exists(f): return SkillResult(...statusfailed, messagef数据文件不存在: {f}, ...) # 3. 在代码前面注入一些公共配置中文字体支持、输出图表目录等 chart_dir /tmp/skill_artifacts/charts os.makedirs(chart_dir, exist_okTrue) injected_setup f import matplotlib matplotlib.use(Agg) import matplotlib.pyplot as plt import json, csv, os plt.rcParams[font.sans-serif] [SimHei, DejaVu Sans] plt.rcParams[axes.unicode_minus] False # 声明图表输出目录 CHART_DIR {chart_dir!r} os.makedirs(CHART_DIR, exist_okTrue) def save_chart(fig, name): path os.path.join(CHART_DIR, name) fig.savefig(path, dpi150, bbox_inchestight) plt.close(fig) print(fCHART_SAVED: {{path}}) return path full_code injected_setup \n code # 4. 执行代码沙箱 sandbox CodeSandbox(allowed_paths[/tmp/skill_artifacts, /data], timeout_s60) result sandbox.run(full_code) # 5. 解析执行结果 stdout result[stdout] stderr result[stderr] returncode result[returncode] if returncode ! 0: return SkillResult( skill_nameself.meta.name, statusfailed, messagef代码执行失败stderr: {stderr}, error_typeCodeExecutionError, error_detailstderr, duration_msint((time.time() - start_time) * 1000), ) # 6. 从 stdout 中捕获 print 输出和图表文件 print_output [] chart_paths [] for line in stdout.splitlines(): if line.startswith(CHART_SAVED:): chart_paths.append(line.replace(CHART_SAVED:, ).strip()) else: print_output.append(line) # 7. 尝试从 stdout 中解析 JSON 结果如果用户代码用 print_json 方法 structured_data None for line in print_output: if line.startswith(RESULT_JSON:): try: structured_data json.loads(line.replace(RESULT_JSON:, ).strip()) except json.JSONDecodeError: pass return SkillResult( skill_nameself.meta.name, statussuccess, data{ print_output: print_output, structured_data: structured_data, chart_paths: chart_paths, }, messagef数据分析和可视化完成生成 {len(chart_paths)} 个图表文件, artifactschart_paths, metadata{returncode: returncode, output_lines: len(print_output)}, duration_msint((time.time() - start_time) * 1000), )有几个处理细节我想单独拉出来讲。injected_setup的注入方式非常实用。很多用户提交的数据分析代码都会用到 matplotlib但如果不加matplotlib.use(Agg)在无显示环境里会直接报cannot connect to X server。我在执行前统一注入这些公共配置用户不需要关心底层环境细节只管写业务代码。还有中文字体配置如果不加生成的图表里中文会变成方块。捕获图表路径的方式是让用户代码调用我注入的save_chart函数它会打印一行特殊标记CHART_SAVED:然后我在 stdout 里按前缀解析。这种约定式的通信方式虽然简单但很可靠——你不知道用户代码里会 print 什么但你可以用特殊前缀把结构化信息和普通输出区分开。RESULT_JSON:前缀同理。用户如果希望返回结构化数据给下游技能用可以在代码里打印print(fRESULT_JSON: {json.dumps(result)})我这边就能自动解析成 Python 对象塞进structured_data字段。这样就打通了数据分析技能和下游报告生成技能之间的数据链路。3.3 注册与对话接口怎么让 Agent 真正用起来技能本身写好了还需要注册进技能中心并且给 Agent 一个接入层。class SkillRegistry: def __init__(self): self._skills: Dict[str, BaseSkill] {} self._skill_metas: Dict[str, SkillMeta] {} def register(self, skill: BaseSkill): self._skills[skill.meta.name] skill self._skill_metas[skill.meta.name] skill.meta def get_skill(self, name: str) - BaseSkill: skill self._skills.get(name) if skill is None: raise KeyError(fskill not found: {name}) return skill def list_available_skills(self) - str: 生成给 Agent 看的技能列表摘要 lines [] for name, meta in self._skill_metas.items(): if self.is_healthy(name): lines.append(f## {name}) lines.append(f描述: {meta.description}) lines.append(f何时使用: {meta.when_to_use}) lines.append(f何时不使用: {meta.when_not_to_use}) lines.append(f参数声明(JSON): {json.dumps(self._skills[name].parameters_schema, ensure_asciiFalse)}) lines.append() return \n.join(lines) def is_healthy(self, name: str) - bool: # 真实实现里面可以做健康检查比如 ping 一下依赖服务 return True # 初始化注册中心 registry SkillRegistry() registry.register(DataAnalysisSkill())这里最关键的是list_available_skills方法。它会生成一段格式化的技能列表文本直接作为 system prompt 的一部分拼进去。Agent 在读这段内容的时候就知道我有这些技能可用、各是什么、参数怎么填。这比手写函数列表要标准得多因为元数据全在技能类里面不需要额外维护一份文档。Agent 调用技能的对话接口一般是这样的当 LLM 决定调用某个技能时它以 JSON 形式输出一个工具调用指令包含skill_name和params字段。你的执行层解析这个 JSON从注册中心取出对应技能执行execute(params)然后把SkillResult转成文本反馈给 LLM。我建议在 Agent 决策前先把list_available_skills()的输出经过一次长度裁剪。如果技能特别多整段塞给 LLM 会浪费大量 token。可以按任务类型做粗筛只把和当前任务相关的技能描述发过去。这个技能粗筛我通常用一个轻量分类器或者关键词匹配实现效果不错。还有一个小建议当多个技能执行失败时反馈给 LLM 的信息要带着技能名称失败原因建议恢复动作这样 LLM 才能做出正确判断。如果你只返回一个failed字符串LLM 大概率不知道该重试、换技能还是求助用户。3.4 完整演示Agent 跑一个真实的销售数据分析任务为了让你更直观地理解整条链路我模拟一个完整的执行过程。假设用户说分析一下sales_data.csv这家公司的销售趋势按月汇总销售额并输出柱状图。第一步Agent 收到用户请求后先读取技能列表判断data_analysis技能合适于是生成调用指令{ skill_name: data_analysis, params: { code: import pandas as pd\n\ndf pd.read_csv(/data/sales_data.csv)\ndf[order_date] pd.to_datetime(df[order_date])\ndf[month] df[order_date].dt.to_period(M)\nmonthly df.groupby(month)[amount].sum().reset_index()\nmonthly[month] monthly[month].astype(str)\nprint(monthly.to_string(indexFalse))\n\nimport matplotlib.pyplot as plt\nfig, ax plt.subplots(figsize(10, 5))\nax.bar(monthly[month], monthly[amount], colorskyblue)\nax.set_title(月度销售额趋势)\nax.set_xlabel(月份)\nax.set_ylabel(销售额)\nplt.xticks(rotation45)\npath save_chart(fig, monthly_sales.png)\nprint(fRESULT_JSON: {{\monthly\: monthly.to_dict(records)}})\n, data_files: [/data/sales_data.csv], output_charts: true } }第二步执行层校验参数、注入公共配置、送入沙箱运行。沙箱运行结束后返回SkillResult( statussuccess, data{ print_output: [2024-01 52000, 2024-02 62800, 2024-03 71100, ...], structured_data: {monthly: [{month: 2024-01, amount: 52000}, ...]}, chart_paths: [/tmp/skill_artifacts/charts/monthly_sales.png], }, message数据分析和可视化完成生成 1 个图表文件, artifacts[/tmp/skill_artifacts/charts/monthly_sales.png], )第三步执行层把SkillResult转成文本反馈给 LLM。LLM 据此生成面向用户的自然语言回答2024年销售整体保持增长3月销售额最高达到71100元。以下是月度趋势图[图片路径]。第四步如果有下游技能要做进一步分析比如生成一份营销建议报告它会从structured_data里取monthly列表或者从artifacts里读图表文件路径直接复用不需要重跑数据分析。这样一个完整的数据分析技能就接入成功了。4. 常见问题与避坑实录这些坑我替你踩过了4.1 技能误调用还是那个什么时候不该用的老大难我在前面反复强调when_not_to_use的重要性这里用一个真实的翻车案例来说明。最开始做技能系统时我给一个网络搜索技能写描述只写了搜索互联网获取最新信息没写什么场景不该用。结果 Agent 在用户问你能做什么这种元问题上都触发了网络搜索白白浪费了几秒时间和大量 token还反馈回一堆无关的搜索片段。后来我把when_not_to_use写成当用户是在咨询本系统自身的功能和限制时不要使用搜索技能直接基于系统说明回答误调用率一下子降了 70% 以上。所以如果你发现 Agent 老是在不合适的场景调用某个技能先别急着调description赶紧补when_not_to_use字段用当……时的句式明确负面清单。另一个排查技能误调用的高效做法是记录调用日志。我为每个技能调用都加了一行结构化日志包含用户请求原文、调用参数、返回状态。当误调用发生可以从日志里反推 LLM 是看到了什么上下文才触发这个调用的然后针对性修改描述或增加负面条件而不是猜。4.2 上下文污染输出太长把模型窗口塞爆怎么优化如果说误调用是频率问题那上下文污染就是杀伤力问题。一次技能调用的输出可能非常大比如代码执行技能的 stdout 有几十 KB 甚至更大如果原封不动全部塞回给 LLM很快窗口就满了。我的处理策略是输出摘要化。SkillResult里的data字段如果太大就只保留前 N 行截断后的摘要对于完整的输出落盘到临时文件并把文件路径放进message或metadata让 LLM 知道完整结果在哪个文件里如果需要可以读取。这样 LLM 既不丢失信息也不需要在一轮对话里承受巨大 token。具体实现上max_output_length在权限声明里已经定义了。我在执行层对这行做了强制检查def truncate_output(text: str, max_len: int 4096) - str: if len(text) max_len: return text head text[:max_len // 2] tail text[-max_len // 4:] return head f\n...[中间省略 {len(text) - max_len * 3 // 4} 字完整输出见 /tmp/skill_artifacts/full_output.log]...\n tail如果你做数据类技能还有一点非常关键不要直接回传完整 DataFrame。我见过有人让技能返回一个几万行的 DataFrame 转 JSON 塞给 LLM结果完全不可用。应该先做聚合或抽样只回传摘要统计量比如 Top 10、分组汇总、分布直方图。LLM 不需要看到每一行数据也能给出有价值的分析。4.3 沙箱逃逸与资源耗尽安全性真的不能只看表面沙箱安全是个无止境的话题。子进程沙箱能拦多少说实话拦不住真正的攻击者。它最大的价值在于防止无意的越权和粗心的破坏比如代码里写了os.remove(/)、shutil.rmtree(/etc)、无限循环内存暴涨这类情况。在我自己的项目里还碰到过一次比较典型的半攻击——某个测试用户在代码里执行了subprocess.run([curl, http://internal-service/], shellTrue)。我的沙箱已经把__import__禁了但用户可以用__builtins__里的__import__绕过禁止或者直接用subprocess模块的旧缓存所以防 import 这种方案很脆弱。要真正限制子进程必须在进程文件系统层面做手脚比如在子进程里先用seccomp过滤系统调用或者干脆用容器。我个人给项目的定级是内部使用 有账号体系 非公网这个风险级别用子进程沙箱够了。如果你的服务要公开给用户别犹豫直接上容器隔离一个请求一个轻量容器用完即焚。如果要用容器但不想自建复杂编排系统可以用一些成熟的容器沙箱方案总之不建议在子进程沙箱上投入过多精力追求绝对安全投入产出比不划算。资源耗尽问题也要提前设防。即使有沙箱内存限制也要给CodeSandbox.run加上外层的执行超时保护防止极端情况下subprocess.run的 timeout 参数因为某种原因失效主进程被卡死。我一般在执行层外面再包一层concurrent.futures的超时控制双保险。4.4 编排状态丢失一条任务在半夜跑挂了第二天怎么恢复长任务场景下我踩过一个很大的坑DAG 编排引擎是纯内存态的如果进程重启所有正在进行的任务状态全部丢失。一开始我抱着反正开发阶段没人用的心态觉得没关系结果一上线就被打脸——一个跑了 20 分钟的数据抓取任务在第 19 分钟因为一次发布重启直接归零。后来我给编排引擎加了持久化每个技能节点的执行结果和 DAG 的当前进度以 JSON 格式定期写盘。每个任务有唯一 ID编排引擎重启后扫描任务目录恢复未完成节点的状态从断点继续跑。这其实就是一个简化版的任务队列。另外一个实际问题DAG 中某些节点是可以幂等重试的比如搜索、读取文件有些则不是比如发送邮件、支付操作。我在技能元信息里增加了一个is_idempotent字段编排引擎在恢复过程中只会自动重跑幂等节点非幂等节点会停下来请求人工确认。这个小设计帮你避免了很多灾难级事故。实现幂等恢复并不复杂关键是状态保存的意识。哪怕你只是把每个节点的SkillResult存到一个 SQLite 表里也比纯内存状态好得多。不要觉得持久化很麻烦用 SQLite 或者简单的文件日志半天就能接好。5. 技能生态的更多可能性从工具集合到可演进的能力网络一个可插拔的技能系统价值远不止于调用代码更方便。当技能定义变成标准化的、带语义描述的、可独立升级的单元之后整个 Agent 就从一个只能执行固定流程的程序变成了一个可以根据任务动态组装能力的工作平台。这份操作空间会引出很多有意思的东西。比如技能的市场化分发。内部不同团队可以各自维护技能包发布到统一的技能仓库主系统只要动态拉取新技能包、注册即可获得新能力完全不需要改主流程代码。这个模式比较像通过插件生态来扩展主应用只不过插件的使用者变成了 AI Agent 而不是人类用户。再比如技能的自动组合探索。编排引擎可以做多轮采样同一个任务尝试不同的技能组合和参数用执行结果反馈来学习哪条路径最优。这就是一个面向 Agent 的 AutoML 雏形。我目前只是在离线日志里做了简单的成功率统计但已经能明显看出不同技能描述对调用成功率的影响了。还有一个方向是技能质量监控。因为每个技能都有标准化的SkillResult你可以统计每个技能的被调用次数、成功率、平均耗时、最常出现的错误类型。这些指标能直接指导你优化整个 Agent 系统的瓶颈——是某个技能本身容易出错还是技能描述让 LLM 理解有误还是编排顺序不合理。这种可观测性是手写工具函数时完全不具备的。做技术选型时我一直有一个判断标准如果一个设计能让你在三个星期后不需要重写代码就值得付出多一天的架构成本。agent-skills这套体系对我来说是在这个标准下成立的。模式并不复杂核心就是一份标准接口、一个注册中心、一个编排引擎、一个沙箱层但组合起来之后给 Agent 扩容能力这件事突然变得很有条理了。我在实际迭代中的体会是技能系统的难点从来不在单一技能的实现而在于如何约束技能与技能之间的关系、如何让技能的意图被 Agent 准确理解、如何在扩展和稳定之间维持平衡。这些问题的解法没有标准答案但至少先把骨架搭好后面每一步改动都会轻松很多。 Agent技能编排实战：从工具调用到可插拔的企业级AI架构
我最近在重构智能体项目时彻底受够了那种模型提示词工具的快餐式架构。一开始确实跑得欢但等技能一多——搜索、代码执行、文件处理、数据分析全堆在一起——代码就开始失控了。工具之间的边界模糊、上下文互相污染、权限管理形同虚设改一个功能得连带调试三个模块。后来我干脆推倒重来把整个体系按照一套技能标准重新设计了一遍也就是这个agent-skills项目。简单说它不是一个具体的业务工具而是一套给 Agent 做技能插拔的框架每个独立能力都封装成一个自包含的 Skill 模块统一输入输出、统一注册与发现机制、统一运行沙箱和权限策略Agent 通过路由层按需调用。这套设计对正在头疼工具越来越多、越来越乱的开发者特别有参考价值尤其是做 AI 应用落地、RAG 增强、自动化工作流的朋友。这篇博客我会从整体设计思路讲起然后拆解一个技能模块的核心结构再带你把一个代码执行 数据分析的技能完整实现一遍最后整理我在实际落地过程中踩过的一些坑和排查心得。整个过程尽量讲清楚为什么这么设计而不只是给一堆代码片段。1. 技能插拔背后的架构思路把工具调用升级为可治理的技能体系1.1 从工具到技能一次性解决上下文污染和权限失控很多人在给 Agent 加能力的时候第一反应就是往 system prompt 里塞定义或者在代码里堆一堆if function_name xxx的分支。这种方式的痛点在早期不明显但一旦技能数量超过 10 个问题就暴露得非常快Agent 根本分不清什么时候该用哪个工具工具间频繁互相调用导致上下文爆炸而且没有任何统一的权限控制手段——套一次 shell 命令就能让 Agent 做超出预期的事情。我在agent-skills里做的第一件关键设计是把工具这个概念彻底替换为技能。这两者的本质区别在于工具是函数的被动暴露而技能是一个自包含的主动性单元。什么意思呢一个技能模块不仅包含能做什么元信息、怎么做执行函数还包含什么时候用触发条件、允许多大权限权限声明、依赖什么资源资源清单甚至包括失败后怎么恢复。这种信息 self-contained 的设计让 Agent 在决策时拿到了足够多的上下文而不需要靠猜。举个实际例子。假设你要给 Agent 加一个运行 Python 代码的技能。如果只是简单暴露一个exec函数那 Agent 可能在任何场景下都试图执行代码哪怕是在只该做文本处理的任务里。但如果你定义了一个PythonExecutorSkill声明了when_to_use字段比如仅当需要计算数学表达式、处理数据文件、或生成可视化图表时使用Agent 的决策准确度会明显上升。同时permissions字段声明允许读取 /data 目录禁止访问网络这样即便 Agent 写了一段恶意的网络请求代码沙箱层也会直接拦截。1.2 技能注册与发现为什么路由表比手写判断更可靠整个技能系统的骨架是一个技能注册中心。所有技能在系统启动时通过装饰器或配置文件注册进去形成一个统一的技能路由表。这个路由表不仅仅是名字到函数的映射它内部维护了一份技能元数据索引技能名称、描述、参数 schema、触发关键词、期望的输出格式、权限级别一应俱全。我不建议在代码里手写if skill_name python_executor这类逻辑来做分发。原因是 Agent 在生成调用指令时往往带有自然语言的模糊性比如它可能会说帮我跑下这段代码看看结果而技能名称可能是python_executor。这时候你需要的是一个语义匹配层。我用了一个比较轻量的做法把每个技能的描述文本做向量化Agent 的自然语言请求也做向量化然后算相似度。当然如果你不想引入向量库也可以用关键词 规则做兜底——这个后面我会详细说。注册中心还承担了另一个重要职责技能可用性探测。有些技能依赖外部服务或模型比如某个技能需要调用一个 OCR 服务如果服务挂了还注册为可用Agent 就会反复尝试然后失败。我在注册中心做了健康检查机制每次 Agent 发起调用前先查一下技能状态不可用的直接不进候选列表。这个设计大大减少了 Agent 决策的无效尝试率。1.3 技能编排顺序执行、条件分支与并行调度单技能调用只是基本功真正让 Agent 能干复杂活的是技能的编排能力。我在agent-skills里定义了一种简单的技能编排 DSL支持三种基本组合方式顺序执行、条件分支和并行调度。顺序执行很好理解比如先读取文件再分析内容再生成摘要。条件分支则是根据上一步的输出决定下一步执行哪个技能比如如果文件是图片格式调用图像识别技能如果是文本格式调用文本分析技能。并行调度用于多个相对独立的技能同时执行比如同时搜索多个数据源再汇总可以显著缩短整体耗时。这里的核心难点在于技能间的数据传递。每个技能的输出不能只是简单的字符串需要一个结构化的数据块包含状态码、结构化结果、中间产物引用、错误信息等。这样下一个技能才能准确地定位到需要的数据。我设计了一个SkillResult类统一封装输出格式这在系统里是一个非常底层的约定。2. 核心细节解析定义一份技能模块的标准接口2.1 技能基类 Skill六大组成部分的职责划分关于技能模块的标准接口我最终沉淀出来六个核心组成部分缺一不可。在很多工具调用框架里开发者只看重函数签名但在我这个体系里函数签名只是其中一小部分。第一个是meta元信息。包含技能名称、版本、作者、描述、触发场景示例。描述信息我建议写得越详细越好因为这是 Agent 决定是否调用你的核心依据。第二个是parameters_schema参数声明用 JSON Schema 格式定义这个技能需要哪些输入Agent 在调用前会先按照 schema 生成参数不合法的会被前置校验拦截。第三个是permissions权限声明定义这个技能可以访问哪些文件路径、能否联网、能否执行系统命令等。第四个是execute方法这是技能的实际执行逻辑。第五个是validate_result结果校验方法确保技能输出符合预期的格式。第六个是recover异常恢复方法定义了当执行失败时应该尝试哪些兜底策略。这六个部分组合在一起就形成了一个完备的技能定义。下面是我项目中base_skill.py的核心代码删减了业务逻辑只保留骨架from abc import ABC, abstractmethod from enum import Enum from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class SkillPermission(BaseModel): # 允许访问的文件路径前缀例如 [/data, /tmp]空列表表示禁止文件访问 allowed_file_paths: List[str] Field(default_factorylist) # 是否允许网络访问 allow_network: bool False # 是否允许执行系统命令 allow_system_command: bool False # 资源限额最大执行时间秒、最大输出长度 max_execution_time: int 30 max_output_length: int 8192 class SkillMeta(BaseModel): name: str Field(..., description技能唯一名称) version: str Field(default0.1.0) description: str Field(..., description对技能能力的详细描述) when_to_use: str Field(..., description说明什么场景下使用该技能) when_not_to_use: str Field(default, description说明什么场景下不要使用该技能非常重要) tags: List[str] Field(default_factorylist) author: str Field(defaultunknown) class SkillResult(BaseModel): skill_name: str status: str success # success | failed | partial data: Any None message: str metadata: Dict[str, Any] Field(default_factorydict) # 如果执行过程中产出了中间文件把文件路径放在这里供依赖链上的后续技能使用 artifacts: List[str] Field(default_factorylist) duration_ms: int 0 error_type: str error_detail: str class BaseSkill(ABC): # 技能元信息 meta: SkillMeta # 参数声明遵循 JSON Schema 格式 parameters_schema: Dict[str, Any] {} # 权限声明 permissions: SkillPermission SkillPermission() # 依赖的技能名称列表 dependencies: List[str] [] def __init__(self, context: Optional[Dict[str, Any]] None): # context 是全局共享上下文比如知识库连接、日志器、配置项等 self.context context or {} abstractmethod def execute(self, params: Dict[str, Any]) - SkillResult: pass def validate_parameters(self, params: Dict[str, Any]) - List[str]: 参数校验返回错误信息列表空列表表示校验通过 errors [] # 这里用 jsonschema 做校验为了简洁省略具体实现 return errors def validate_result(self, result: SkillResult) - bool: # 校验结果是否符合预期用于编排引擎判断是否继续走 DAG return result.status success def recover(self, params: Dict[str, Any], error: Exception) - SkillResult: 技能执行失败后的兜底逻辑默认返回错误子类可覆盖 return SkillResult( skill_nameself.meta.name, statusfailed, messageunhandled error, error_typetype(error).__name__, error_detailstr(error), )关于这段代码我想特别强调几个容易被忽视的点。when_not_to_use这个字段必须有。它和when_to_use一样重要甚至更重要。因为 Agent 的误调用很多时候不是因为不知道该用什么而是因为不知道不该用什么。比如一个负责生成文案的 Agent如果不声明当用户只是闲聊时不要调用本技能它可能在任何对话场景下都触发文本生成技能导致回复内容极其生硬。SkillResult里我设计了artifacts字段专门存中间产物的引用。举个例子代码执行技能运行完一段 Python 后可能生成了一张图表chart.png这个文件的路径会被放进artifacts这样后续的文档生成技能就能直接引用它而不需要重新执行一遍代码。recover方法的设计来自一次惨痛教训。之前没有恢复机制时只要一个技能执行失败整个任务链就断了。后来我意识到很多失败其实是有固定模式的API 超时可以重试一次文件不存在可以自动切换到备用路径格式解析失败可以尝试另一种解析策略。把这些常见恢复逻辑放到recover里Agent 的任务成功率提升非常明显。2.2 技能编排引擎DAG 调度与中间产物管理有了标准接口之后接下来的核心问题就是怎么把多个技能组装成一个工作流。我在agent-skills里实现了一个轻量级的编排引擎基于有向无环图DAG来管理技能之间的依赖关系。为什么用 DAG 而不是简单顺序执行呢因为实际任务中技能之间往往存在部分并行的关系。比如一个调研任务需要同时调用网页搜索技能、知识库检索技能和舆情分析技能三个技能之间没有依赖可以并行但它们的输出都会汇总到最后的报告生成技能里。顺序执行会浪费大量时间而 DAG 调度器可以自动识别可并行的节点。编排引擎的基本工作流程是先把一个复杂任务分解为若干技能节点根据依赖关系构建 DAG然后按拓扑序遍历符合条件的并行节点同时执行。每个节点执行完会产出SkillResult引擎根据validate_result判断是否继续执行下游节点。如果遇到失败节点会先调用recover尝试修复修复不了就中止整个 DAG并向 Agent 返回失败的原因链。中间产物管理是这个引擎的另一大重头戏。我用一个内存对象存储维护所有技能节点的输出同时支持把大型中间结果落盘到临时目录。对于一份很大的数据文件不会直接塞进内存传递给下一个节点而是写入/tmp/skill_artifacts/目录然后在节点间传递文件路径。这个设计对长任务的稳定性非常关键——我之前就碰到过因为大对象占满内存导致整个 Agent 崩掉的情况改成落盘方案后彻底解决了。关于 DAG 调度有个小技巧值得分享在 Agent 生成技能编排图的时候不要让它自由发挥而是提供一个技能编排模板库。预先定义一些常见任务的固定流程比如数据报告生成流程、竞品调研流程、日志排查流程等Agent 只需要从模板里选择并填充参数。这样既保证了执行效率又减少了编排结构错误的风险。2.3 权限沙箱技能执行的安全边界到底怎么画技能系统如果要外部扩展安全就是第一优先级。我给不同级别的技能划分了四层沙箱策略可以根据技能的可信度灵活选择。第一层是纯逻辑层适用于不涉及外部资源操作的技能比如 JSON 格式化、文本处理、时间计算。这种技能不需要文件访问也不需要网络直接在进程内跑就行了。第二层是文件沙箱适用于需要读写文件但不能联网的技能。做法是把文件访问强制限制在声明的前缀路径下用os.path.realpath先做路径归一化再判断是否在允许范围内防目录穿越攻击。第三层是网络白名单适用于需要调用外部 API 的技能。这里我用了一个本地代理强制所有 HTTP 请求通过代理转发代理只放行预先配置的域名白名单。第四层是完整进程沙箱适用于执行不可信代码的场景比如运行用户提交的 Python 代码。这里我把执行放到一个独立的子进程里面并且设置了 CPU 时间限制、内存限制、文件系统只读挂载用resource模块监控资源使用情况。这里专门说一下 python 代码执行技能的子进程沙箱实现。完整方案是这样的父进程通过 subprocess 启动一个子 Python 进程子进程预先加载受限的builtins删除__import__或者替换为受限版本然后通过管道传入要执行的代码在子进程中用exec(compiled_code, restricted_globals)执行执行结果 queue 回传给父进程。import os import resource import subprocess import tempfile from typing import Dict, Any class CodeSandbox: def __init__(self, allowed_paths: list[str] | None None, memory_limit_mb: int 512, timeout_s: int 10): self.allowed_paths allowed_paths or [/tmp/skill_artifacts] self.memory_limit_mb memory_limit_mb self.timeout_s timeout_s def run(self, code: str) - Dict[str, Any]: 在子进程 资源限制下执行一段 Python 代码 # 临时目录作为工作目录避免污染主进程目录 workdir tempfile.mkdtemp(prefixsandbox_, dir/tmp) with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse, encodingutf-8) as f: f.write(self._wrap_code(code)) fpath f.name try: proc subprocess.run( [/usr/bin/python3, fpath], capture_outputTrue, textTrue, timeoutself.timeout_s, cwdworkdir, env{...}, # 精简环境变量 ) return { stdout: proc.stdout, stderr: proc.stderr, returncode: proc.returncode, } except subprocess.TimeoutExpired: return {stdout: , stderr: execution timeout, returncode: -1} finally: os.remove(fpath) # 清理临时目录可以按需保留用于调试 def _wrap_code(self, code: str) - str: return f import builtins # 移除危险的内置函数 _original_import builtins.__import__ def _restricted_import(name, *args, **kwargs): raise ImportError(fimport not allowed: {{name}}) builtins.__import__ _restricted_import # 这里可以进一步替换 open 等内置函数 import sys import resource # 设置内存限制 resource.setrlimit(resource.RLIMIT_AS, (self.memory_limit_mb * 1024 * 1024, self.memory_limit_mb * 1024 * 1024)) # 设置 CPU 限制 resource.setrlimit(resource.RLIMIT_CPU, (self.timeout_s, self.timeout_s)) _exec_globals {{}} try: exec(compile({code!r}, sandbox, exec), _exec_globals) except Exception as e: print(fEXEC_ERROR: {{type(e).__name__}}: {{e}}, filesys.stderr) sys.exit(1) 提示不要把沙箱当成绝对安全边界。对于高风险的不可信代码最稳妥的做法还是直接跑容器隔离或者调用云端的代码执行 API。子进程沙箱只是提高了攻击门槛适合内部工具这类中等威胁场景。这段实现里有两个容易被忽略的细节一是要自定义fpath写到/tmp而不是放在当前工作目录避免权限混乱二是resource.setrlimit(RLIMIT_CPU)必须放在子进程里设置如果在父进程设置会影响主程序自己。我第一次做的时候就是在父进程设了内存限制结果 Agent 主程序直接被 OOM 杀掉了非常尴尬。3. 实操过程从零实现一个代码执行 数据分析技能3.1 场景设计与参数声明让 Agent 清楚地知道该不该调你好现在进入最实际的部分。我带你完整实现一个DataAnalysisSkill它的作用是执行一段 Python 数据分析代码并且生成统计摘要和可视化图表。这个技能在日常业务里用得非常频繁非常适合作为参考案例。第一步是设计它的SkillMeta和参数声明。这个环节建议不要急躁因为你写得越细致Agent 的调用准确率就越高。下面是我在实际项目中用到的版本class DataAnalysisSkill(BaseSkill): meta SkillMeta( namedata_analysis, version1.0.0, description执行 Python 数据分析和可视化代码支持 pandas、numpy、matplotlib。可以读取已落盘的 CSV/JSON 数据文件输出统计结果和图表。, when_to_use当用户要求分析结构化数据、计算统计指标、绘制图表、处理 CSV/Excel/JSON 数据文件时使用。尤其是涉及数据集、报表、实验数据等场景。, when_not_to_use当用户只是问一般性问题、要求写非数据类的代码、或者数据尚未准备完成时不要使用本技能。, tags[data, analysis, visualization], ) parameters_schema { type: object, properties: { code: { type: string, description: 要执行的 Python 代码必须以纯代码片段形式提供不要包含 markdown 代码块标记。, }, data_files: { type: array, items: {type: string}, description: 需要读取的数据文件路径列表必须是沙箱允许访问的路径。, }, output_charts: { type: boolean, description: 是否生成可视化图表默认 true。, default: True, }, }, required: [code], } permissions SkillPermission( allowed_file_paths[/tmp/skill_artifacts, /data], allow_networkFalse, allow_system_commandFalse, max_execution_time60, max_output_length8192, ) dependencies [file_access]我特别要讲一下dependencies字段。DataAnalysisSkill依赖file_access技能但依赖不代表它直接调用文件技能而是它要求文件技能必须先执行完把文件落到指定的/tmp/skill_artifacts目录之后本技能才有东西可分析。这相当于隐性传递了一个数据契约上一个技能负责产出文件本技能负责消费文件。还有一点description里的措辞非常重要。我见过很多人写技能描述时过于笼统比如这个技能可以分析数据Agent 看了一脸懵不知道和别的技能有什么差异。尽量写得具体、带有典型场景关键词比如我上面写了统计指标CSV/Excel/JSON 数据文件这些词Agent 在语义匹配时的命中率会高很多。3.2 核心执行逻辑写一个既能跑又扛造的 execute 方法接下来是execute方法。这一步主要做几件事先校验参数然后调用代码沙箱执行用户传入的代码再解析执行结果主要是 stdout 和产生的图表文件最后封装成SkillResult返回。import json import os import re import time from typing import Dict, Any class DataAnalysisSkill(BaseSkill): def execute(self, params: Dict[str, Any]) - SkillResult: start_time time.time() # 1. 参数校验 errors self.validate_parameters(params) if errors: return SkillResult( skill_nameself.meta.name, statusfailed, message参数校验失败: ; .join(errors), error_typeParameterError, error_detailerrors, duration_msint((time.time() - start_time) * 1000), ) code params[code] data_files params.get(data_files, []) output_charts params.get(output_charts, True) # 2. 检查数据文件是否存在 for f in data_files: if not os.path.exists(f): return SkillResult(...statusfailed, messagef数据文件不存在: {f}, ...) # 3. 在代码前面注入一些公共配置中文字体支持、输出图表目录等 chart_dir /tmp/skill_artifacts/charts os.makedirs(chart_dir, exist_okTrue) injected_setup f import matplotlib matplotlib.use(Agg) import matplotlib.pyplot as plt import json, csv, os plt.rcParams[font.sans-serif] [SimHei, DejaVu Sans] plt.rcParams[axes.unicode_minus] False # 声明图表输出目录 CHART_DIR {chart_dir!r} os.makedirs(CHART_DIR, exist_okTrue) def save_chart(fig, name): path os.path.join(CHART_DIR, name) fig.savefig(path, dpi150, bbox_inchestight) plt.close(fig) print(fCHART_SAVED: {{path}}) return path full_code injected_setup \n code # 4. 执行代码沙箱 sandbox CodeSandbox(allowed_paths[/tmp/skill_artifacts, /data], timeout_s60) result sandbox.run(full_code) # 5. 解析执行结果 stdout result[stdout] stderr result[stderr] returncode result[returncode] if returncode ! 0: return SkillResult( skill_nameself.meta.name, statusfailed, messagef代码执行失败stderr: {stderr}, error_typeCodeExecutionError, error_detailstderr, duration_msint((time.time() - start_time) * 1000), ) # 6. 从 stdout 中捕获 print 输出和图表文件 print_output [] chart_paths [] for line in stdout.splitlines(): if line.startswith(CHART_SAVED:): chart_paths.append(line.replace(CHART_SAVED:, ).strip()) else: print_output.append(line) # 7. 尝试从 stdout 中解析 JSON 结果如果用户代码用 print_json 方法 structured_data None for line in print_output: if line.startswith(RESULT_JSON:): try: structured_data json.loads(line.replace(RESULT_JSON:, ).strip()) except json.JSONDecodeError: pass return SkillResult( skill_nameself.meta.name, statussuccess, data{ print_output: print_output, structured_data: structured_data, chart_paths: chart_paths, }, messagef数据分析和可视化完成生成 {len(chart_paths)} 个图表文件, artifactschart_paths, metadata{returncode: returncode, output_lines: len(print_output)}, duration_msint((time.time() - start_time) * 1000), )有几个处理细节我想单独拉出来讲。injected_setup的注入方式非常实用。很多用户提交的数据分析代码都会用到 matplotlib但如果不加matplotlib.use(Agg)在无显示环境里会直接报cannot connect to X server。我在执行前统一注入这些公共配置用户不需要关心底层环境细节只管写业务代码。还有中文字体配置如果不加生成的图表里中文会变成方块。捕获图表路径的方式是让用户代码调用我注入的save_chart函数它会打印一行特殊标记CHART_SAVED:然后我在 stdout 里按前缀解析。这种约定式的通信方式虽然简单但很可靠——你不知道用户代码里会 print 什么但你可以用特殊前缀把结构化信息和普通输出区分开。RESULT_JSON:前缀同理。用户如果希望返回结构化数据给下游技能用可以在代码里打印print(fRESULT_JSON: {json.dumps(result)})我这边就能自动解析成 Python 对象塞进structured_data字段。这样就打通了数据分析技能和下游报告生成技能之间的数据链路。3.3 注册与对话接口怎么让 Agent 真正用起来技能本身写好了还需要注册进技能中心并且给 Agent 一个接入层。class SkillRegistry: def __init__(self): self._skills: Dict[str, BaseSkill] {} self._skill_metas: Dict[str, SkillMeta] {} def register(self, skill: BaseSkill): self._skills[skill.meta.name] skill self._skill_metas[skill.meta.name] skill.meta def get_skill(self, name: str) - BaseSkill: skill self._skills.get(name) if skill is None: raise KeyError(fskill not found: {name}) return skill def list_available_skills(self) - str: 生成给 Agent 看的技能列表摘要 lines [] for name, meta in self._skill_metas.items(): if self.is_healthy(name): lines.append(f## {name}) lines.append(f描述: {meta.description}) lines.append(f何时使用: {meta.when_to_use}) lines.append(f何时不使用: {meta.when_not_to_use}) lines.append(f参数声明(JSON): {json.dumps(self._skills[name].parameters_schema, ensure_asciiFalse)}) lines.append() return \n.join(lines) def is_healthy(self, name: str) - bool: # 真实实现里面可以做健康检查比如 ping 一下依赖服务 return True # 初始化注册中心 registry SkillRegistry() registry.register(DataAnalysisSkill())这里最关键的是list_available_skills方法。它会生成一段格式化的技能列表文本直接作为 system prompt 的一部分拼进去。Agent 在读这段内容的时候就知道我有这些技能可用、各是什么、参数怎么填。这比手写函数列表要标准得多因为元数据全在技能类里面不需要额外维护一份文档。Agent 调用技能的对话接口一般是这样的当 LLM 决定调用某个技能时它以 JSON 形式输出一个工具调用指令包含skill_name和params字段。你的执行层解析这个 JSON从注册中心取出对应技能执行execute(params)然后把SkillResult转成文本反馈给 LLM。我建议在 Agent 决策前先把list_available_skills()的输出经过一次长度裁剪。如果技能特别多整段塞给 LLM 会浪费大量 token。可以按任务类型做粗筛只把和当前任务相关的技能描述发过去。这个技能粗筛我通常用一个轻量分类器或者关键词匹配实现效果不错。还有一个小建议当多个技能执行失败时反馈给 LLM 的信息要带着技能名称失败原因建议恢复动作这样 LLM 才能做出正确判断。如果你只返回一个failed字符串LLM 大概率不知道该重试、换技能还是求助用户。3.4 完整演示Agent 跑一个真实的销售数据分析任务为了让你更直观地理解整条链路我模拟一个完整的执行过程。假设用户说分析一下sales_data.csv这家公司的销售趋势按月汇总销售额并输出柱状图。第一步Agent 收到用户请求后先读取技能列表判断data_analysis技能合适于是生成调用指令{ skill_name: data_analysis, params: { code: import pandas as pd\n\ndf pd.read_csv(/data/sales_data.csv)\ndf[order_date] pd.to_datetime(df[order_date])\ndf[month] df[order_date].dt.to_period(M)\nmonthly df.groupby(month)[amount].sum().reset_index()\nmonthly[month] monthly[month].astype(str)\nprint(monthly.to_string(indexFalse))\n\nimport matplotlib.pyplot as plt\nfig, ax plt.subplots(figsize(10, 5))\nax.bar(monthly[month], monthly[amount], colorskyblue)\nax.set_title(月度销售额趋势)\nax.set_xlabel(月份)\nax.set_ylabel(销售额)\nplt.xticks(rotation45)\npath save_chart(fig, monthly_sales.png)\nprint(fRESULT_JSON: {{\monthly\: monthly.to_dict(records)}})\n, data_files: [/data/sales_data.csv], output_charts: true } }第二步执行层校验参数、注入公共配置、送入沙箱运行。沙箱运行结束后返回SkillResult( statussuccess, data{ print_output: [2024-01 52000, 2024-02 62800, 2024-03 71100, ...], structured_data: {monthly: [{month: 2024-01, amount: 52000}, ...]}, chart_paths: [/tmp/skill_artifacts/charts/monthly_sales.png], }, message数据分析和可视化完成生成 1 个图表文件, artifacts[/tmp/skill_artifacts/charts/monthly_sales.png], )第三步执行层把SkillResult转成文本反馈给 LLM。LLM 据此生成面向用户的自然语言回答2024年销售整体保持增长3月销售额最高达到71100元。以下是月度趋势图[图片路径]。第四步如果有下游技能要做进一步分析比如生成一份营销建议报告它会从structured_data里取monthly列表或者从artifacts里读图表文件路径直接复用不需要重跑数据分析。这样一个完整的数据分析技能就接入成功了。4. 常见问题与避坑实录这些坑我替你踩过了4.1 技能误调用还是那个什么时候不该用的老大难我在前面反复强调when_not_to_use的重要性这里用一个真实的翻车案例来说明。最开始做技能系统时我给一个网络搜索技能写描述只写了搜索互联网获取最新信息没写什么场景不该用。结果 Agent 在用户问你能做什么这种元问题上都触发了网络搜索白白浪费了几秒时间和大量 token还反馈回一堆无关的搜索片段。后来我把when_not_to_use写成当用户是在咨询本系统自身的功能和限制时不要使用搜索技能直接基于系统说明回答误调用率一下子降了 70% 以上。所以如果你发现 Agent 老是在不合适的场景调用某个技能先别急着调description赶紧补when_not_to_use字段用当……时的句式明确负面清单。另一个排查技能误调用的高效做法是记录调用日志。我为每个技能调用都加了一行结构化日志包含用户请求原文、调用参数、返回状态。当误调用发生可以从日志里反推 LLM 是看到了什么上下文才触发这个调用的然后针对性修改描述或增加负面条件而不是猜。4.2 上下文污染输出太长把模型窗口塞爆怎么优化如果说误调用是频率问题那上下文污染就是杀伤力问题。一次技能调用的输出可能非常大比如代码执行技能的 stdout 有几十 KB 甚至更大如果原封不动全部塞回给 LLM很快窗口就满了。我的处理策略是输出摘要化。SkillResult里的data字段如果太大就只保留前 N 行截断后的摘要对于完整的输出落盘到临时文件并把文件路径放进message或metadata让 LLM 知道完整结果在哪个文件里如果需要可以读取。这样 LLM 既不丢失信息也不需要在一轮对话里承受巨大 token。具体实现上max_output_length在权限声明里已经定义了。我在执行层对这行做了强制检查def truncate_output(text: str, max_len: int 4096) - str: if len(text) max_len: return text head text[:max_len // 2] tail text[-max_len // 4:] return head f\n...[中间省略 {len(text) - max_len * 3 // 4} 字完整输出见 /tmp/skill_artifacts/full_output.log]...\n tail如果你做数据类技能还有一点非常关键不要直接回传完整 DataFrame。我见过有人让技能返回一个几万行的 DataFrame 转 JSON 塞给 LLM结果完全不可用。应该先做聚合或抽样只回传摘要统计量比如 Top 10、分组汇总、分布直方图。LLM 不需要看到每一行数据也能给出有价值的分析。4.3 沙箱逃逸与资源耗尽安全性真的不能只看表面沙箱安全是个无止境的话题。子进程沙箱能拦多少说实话拦不住真正的攻击者。它最大的价值在于防止无意的越权和粗心的破坏比如代码里写了os.remove(/)、shutil.rmtree(/etc)、无限循环内存暴涨这类情况。在我自己的项目里还碰到过一次比较典型的半攻击——某个测试用户在代码里执行了subprocess.run([curl, http://internal-service/], shellTrue)。我的沙箱已经把__import__禁了但用户可以用__builtins__里的__import__绕过禁止或者直接用subprocess模块的旧缓存所以防 import 这种方案很脆弱。要真正限制子进程必须在进程文件系统层面做手脚比如在子进程里先用seccomp过滤系统调用或者干脆用容器。我个人给项目的定级是内部使用 有账号体系 非公网这个风险级别用子进程沙箱够了。如果你的服务要公开给用户别犹豫直接上容器隔离一个请求一个轻量容器用完即焚。如果要用容器但不想自建复杂编排系统可以用一些成熟的容器沙箱方案总之不建议在子进程沙箱上投入过多精力追求绝对安全投入产出比不划算。资源耗尽问题也要提前设防。即使有沙箱内存限制也要给CodeSandbox.run加上外层的执行超时保护防止极端情况下subprocess.run的 timeout 参数因为某种原因失效主进程被卡死。我一般在执行层外面再包一层concurrent.futures的超时控制双保险。4.4 编排状态丢失一条任务在半夜跑挂了第二天怎么恢复长任务场景下我踩过一个很大的坑DAG 编排引擎是纯内存态的如果进程重启所有正在进行的任务状态全部丢失。一开始我抱着反正开发阶段没人用的心态觉得没关系结果一上线就被打脸——一个跑了 20 分钟的数据抓取任务在第 19 分钟因为一次发布重启直接归零。后来我给编排引擎加了持久化每个技能节点的执行结果和 DAG 的当前进度以 JSON 格式定期写盘。每个任务有唯一 ID编排引擎重启后扫描任务目录恢复未完成节点的状态从断点继续跑。这其实就是一个简化版的任务队列。另外一个实际问题DAG 中某些节点是可以幂等重试的比如搜索、读取文件有些则不是比如发送邮件、支付操作。我在技能元信息里增加了一个is_idempotent字段编排引擎在恢复过程中只会自动重跑幂等节点非幂等节点会停下来请求人工确认。这个小设计帮你避免了很多灾难级事故。实现幂等恢复并不复杂关键是状态保存的意识。哪怕你只是把每个节点的SkillResult存到一个 SQLite 表里也比纯内存状态好得多。不要觉得持久化很麻烦用 SQLite 或者简单的文件日志半天就能接好。5. 技能生态的更多可能性从工具集合到可演进的能力网络一个可插拔的技能系统价值远不止于调用代码更方便。当技能定义变成标准化的、带语义描述的、可独立升级的单元之后整个 Agent 就从一个只能执行固定流程的程序变成了一个可以根据任务动态组装能力的工作平台。这份操作空间会引出很多有意思的东西。比如技能的市场化分发。内部不同团队可以各自维护技能包发布到统一的技能仓库主系统只要动态拉取新技能包、注册即可获得新能力完全不需要改主流程代码。这个模式比较像通过插件生态来扩展主应用只不过插件的使用者变成了 AI Agent 而不是人类用户。再比如技能的自动组合探索。编排引擎可以做多轮采样同一个任务尝试不同的技能组合和参数用执行结果反馈来学习哪条路径最优。这就是一个面向 Agent 的 AutoML 雏形。我目前只是在离线日志里做了简单的成功率统计但已经能明显看出不同技能描述对调用成功率的影响了。还有一个方向是技能质量监控。因为每个技能都有标准化的SkillResult你可以统计每个技能的被调用次数、成功率、平均耗时、最常出现的错误类型。这些指标能直接指导你优化整个 Agent 系统的瓶颈——是某个技能本身容易出错还是技能描述让 LLM 理解有误还是编排顺序不合理。这种可观测性是手写工具函数时完全不具备的。做技术选型时我一直有一个判断标准如果一个设计能让你在三个星期后不需要重写代码就值得付出多一天的架构成本。agent-skills这套体系对我来说是在这个标准下成立的。模式并不复杂核心就是一份标准接口、一个注册中心、一个编排引擎、一个沙箱层但组合起来之后给 Agent 扩容能力这件事突然变得很有条理了。我在实际迭代中的体会是技能系统的难点从来不在单一技能的实现而在于如何约束技能与技能之间的关系、如何让技能的意图被 Agent 准确理解、如何在扩展和稳定之间维持平衡。这些问题的解法没有标准答案但至少先把骨架搭好后面每一步改动都会轻松很多。