15.5 Skills脚本执行机制
第 15 章 · Skills的工作原理 · 第5部分:技能 Skills
说明:本章中的代码示例是为了帮助理解技术原理而提供的。实际创建 Skills 时,您只需要编写简单的脚本,系统会自动处理安全执行。
脚本在 Skills 中的角色#
脚本执行的必要性#
虽然 AI 模型具备强大的自然语言理解和生成能力,但在某些场景下,传统代码执行具有不可替代的优势:
确定性保证
- 数学计算的精确性
- 数据处理的可靠性
- 复杂算法的正确性
性能优势
- 大规模数据处理的高效性
- 重复操作的性能优化
- 系统集成的低延迟
功能扩展
- 访问本地文件系统
- 调用外部 API 和服务
- 执行系统级操作
脚本 vs 纯 AI 生成#
| 特性 | 脚本执行 | AI 生成 |
|---|---|---|
| 确定性 | 高 | 中 |
| 性能 | 高 | 中 |
| 可维护性 | 高 | 低 |
| 灵活性 | 中 | 高 |
| 复杂度处理 | 高 | 中 |
脚本集成架构#
脚本发现机制#
自动脚本扫描
系统会自动检查 Skills 目录中的脚本文件:
- 查找脚本目录:在 Skills 文件夹中寻找
scripts子目录 - 识别脚本文件:检查文件是否是可执行的脚本(.py、.sh等)
- 收集信息:记录每个脚本的基本属性
- 建立索引:创建脚本名称到信息的映射
这样系统就知道这个 Skills 有哪些可用的脚本工具。
脚本元数据提取
对于每个发现的脚本,系统会收集详细信息:
- 文件路径:脚本在磁盘上的位置
- 编程语言:Python、Shell、JavaScript等
- 文件大小:脚本占用的存储空间
- 修改时间:最后更新的时间
- 执行权限:是否有执行权限
- 依赖关系:脚本需要的外部库或工具
脚本调用接口#
统一调用接口
python复制classScriptExecutor:def__init__(self): self.sandbox = ScriptSandbox() self.cache = ScriptCache() asyncdefexecute_script(self, script_ref, arguments, context):""" 统一的脚本执行接口 """# 解析脚本引用 script_path = self.resolve_script_path(script_ref, context.skill_path) # 检查缓存 cache_key = self.generate_cache_key(script_path, arguments)if cache_key in self.cache:return self.cache[cache_key] # 执行脚本 result =await self.sandbox.execute_script(script_path, arguments) # 处理结果 processed_result = self.process_execution_result(result) # 缓存结果if self.should_cache(result): self.cache[cache_key]= processed_result return processed_result
参数传递机制
系统需要为脚本准备正确的输入参数,就像给函数传递参数:
- 获取变量:从 Skills 执行上下文中提取可用的变量值
- 处理模板:识别参数中的变量引用(如 $variable_name)
- 替换变量:将变量引用替换为实际值
- 保留常量:直接使用非变量的参数值
- 构建参数列表:按照正确顺序组织所有参数
这样脚本就能获得它需要的所有输入信息。
脚本安全沙箱#
沙箱设计原则#
最小权限原则
- 只授予必要的文件系统权限
- 限制网络访问范围
- 控制子进程创建
资源限制
- CPU 时间限制
- 内存使用限制
- 磁盘 I/O 限制
沙箱实现#
容器化执行
为了安全执行脚本,系统使用容器技术创建隔离环境:
- 选择容器镜像:使用安全的 Python 基础镜像
- 限制资源:设置内存和CPU使用限制
- 隔离网络:默认禁用网络访问(可配置)
- 只读挂载:脚本文件以只读方式挂载到容器
- 监控执行:捕获输出并验证执行结果
这样脚本在安全的沙箱环境中运行,不会影响主机系统。
系统级隔离
python复制classSystemSandbox:def__init__(self): self.chroot_dir ='/tmp/skill_sandbox' self.max_runtime =30# seconds asyncdefexecute_isolated(self, script_path, args):""" 使用系统级隔离执行脚本 """# 创建临时环境with tempfile.TemporaryDirectory()as temp_dir:# 复制脚本到隔离环境 isolated_script = os.path.join(temp_dir,'script.py') shutil.copy2(script_path, isolated_script) # 设置执行环境 env = os.environ.copy() env['PATH']='/usr/local/bin:/usr/bin' # 执行脚本 process =await asyncio.create_subprocess_exec('python', isolated_script,*args, cwd=temp_dir, env=env, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, timeout=self.max_runtime ) try: stdout, stderr =await process.communicate()return{'returncode': process.returncode,'stdout': stdout.decode(),'stderr': stderr.decode()}except asyncio.TimeoutError: process.kill()raise ScriptTimeoutError("Script execution timed out")
脚本结果处理#
输出格式标准化#
结构化输出解析
python复制defparse_structured_output(script_output):""" 解析脚本的结构化输出 """try:# 尝试 JSON 解析return json.loads(script_output)except json.JSONDecodeError:pass try:# 尝试 YAML 解析return yaml.safe_load(script_output)except yaml.YAMLError:pass # 回退到文本处理return parse_text_output(script_output)
多格式支持
- JSON:结构化数据交换
- YAML:配置和数据序列化
- CSV:表格数据处理
- 纯文本:简单消息传递
结果验证和转换#
输出验证
python复制defvalidate_script_output(output, expected_schema):""" 验证脚本输出是否符合预期格式 """ validator = SchemaValidator(expected_schema) try: validator.validate(output)returnTrue,Noneexcept ValidationError as e:returnFalse,str(e)
结果转换
python复制deftransform_output_for_context(output, target_format):""" 将脚本输出转换为适合 AI 上下文的格式 """ transformers ={'summary': summarize_output,'table': format_as_table,'code': format_as_code_block,'json': json.dumps } transformer = transformers.get(target_format,lambda x:str(x))return transformer(output)
脚本缓存和优化#
执行结果缓存#
缓存策略
python复制classScriptCache:def__init__(self, max_size=100): self.cache ={} self.max_size = max_size self.access_order =[] defget(self, key):"""获取缓存的执行结果"""if key in self.cache:# 更新访问顺序 self.access_order.remove(key) self.access_order.append(key)return self.cache[key]returnNone defput(self, key, value):"""缓存执行结果"""if key in self.cache:# 更新现有条目 self.access_order.remove(key)eliflen(self.cache)>= self.max_size:# 移除最少使用的条目 oldest_key = self.access_order.pop(0)del self.cache[oldest_key] self.cache[key]= value self.access_order.append(key)
缓存键生成
python复制defgenerate_cache_key(script_path, args, context):""" 生成脚本执行的缓存键 """# 包含脚本内容哈希、参数和相关上下文 script_hash = hashlib.md5(open(script_path,'rb').read()).hexdigest() args_hash = hashlib.md5(str(args).encode()).hexdigest() context_hash = hashlib.md5(str(context).encode()).hexdigest() returnf"{script_hash}:{args_hash}:{context_hash}"
性能优化#
并行执行
python复制asyncdefexecute_scripts_parallel(script_calls):""" 并行执行多个脚本调用 """ tasks =[]for call in script_calls: task = asyncio.create_task(execute_single_script(call)) tasks.append(task) # 等待所有任务完成 results =await asyncio.gather(*tasks, return_exceptions=True) # 处理结果和异常 processed_results =[]for result in results:ifisinstance(result, Exception): processed_results.append({'error':str(result)})else: processed_results.append(result) return processed_results
预编译优化
- 字节码缓存
- AST 预解析
- 依赖预加载
错误处理和调试#
脚本执行错误分类#
1. 语法错误
- Python 语法错误
- 导入错误
- 缩进错误
2. 运行时错误
- 类型错误
- 索引错误
- 文件操作错误
3. 逻辑错误
- 算法错误
- 数据处理错误
- 边界条件错误
调试支持#
详细错误信息
python复制defformat_script_error(error, script_path, args):""" 格式化脚本执行错误信息 """return{'error_type':type(error).__name__,'error_message':str(error),'script_path': script_path,'arguments': args,'stack_trace': traceback.format_exc(),'suggestions': generate_error_suggestions(error)}
调试模式
python复制classScriptDebugger:def__init__(self): self.breakpoints =set() self.watch_variables =set() asyncdefdebug_execute(self, script_path, args):""" 调试模式执行脚本 """# 设置调试环境 debug_env = self.setup_debug_environment() # 执行脚本 result =await self.execute_with_debugging(script_path, args, debug_env) # 收集调试信息 debug_info = self.collect_debug_info() return result, debug_info
总结#
脚本执行机制是 Skills 功能扩展的关键,它通过安全沙箱、标准化接口和智能缓存等技术,实现了 AI 智能与传统代码执行的完美结合。这种混合架构既保持了 AI 的灵活性,又确保了执行的可靠性和性能。
技术说明:本章中的代码示例展示了系统内部的工作机制,旨在帮助您理解原理。实际创建和使用 Skills 时,您不需要编写或修改这些代码,系统会自动处理所有技术细节。