15.5 Skills脚本执行机制

第 15 章 · Skills的工作原理 · 第5部分:技能 Skills

说明:本章中的代码示例是为了帮助理解技术原理而提供的。实际创建 Skills 时,您只需要编写简单的脚本,系统会自动处理安全执行。

脚本在 Skills 中的角色#

脚本执行的必要性#

虽然 AI 模型具备强大的自然语言理解和生成能力,但在某些场景下,传统代码执行具有不可替代的优势:

确定性保证

性能优势

功能扩展

脚本 vs 纯 AI 生成#

特性脚本执行AI 生成
确定性高中
性能高中
可维护性高低
灵活性中高
复杂度处理高中

脚本集成架构#

脚本发现机制#

自动脚本扫描

系统会自动检查 Skills 目录中的脚本文件:

  1. 查找脚本目录:在 Skills 文件夹中寻找 scripts 子目录
  2. 识别脚本文件:检查文件是否是可执行的脚本(.py、.sh等)
  3. 收集信息:记录每个脚本的基本属性
  4. 建立索引:创建脚本名称到信息的映射

这样系统就知道这个 Skills 有哪些可用的脚本工具。

脚本元数据提取

对于每个发现的脚本,系统会收集详细信息:

脚本调用接口#

统一调用接口

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

参数传递机制

系统需要为脚本准备正确的输入参数,就像给函数传递参数:

  1. 获取变量:从 Skills 执行上下文中提取可用的变量值
  2. 处理模板:识别参数中的变量引用(如 $variable_name)
  3. 替换变量:将变量引用替换为实际值
  4. 保留常量:直接使用非变量的参数值
  5. 构建参数列表:按照正确顺序组织所有参数

这样脚本就能获得它需要的所有输入信息。

脚本安全沙箱#

沙箱设计原则#

最小权限原则

资源限制

沙箱实现#

容器化执行

为了安全执行脚本,系统使用容器技术创建隔离环境:

  1. 选择容器镜像:使用安全的 Python 基础镜像
  2. 限制资源:设置内存和CPU使用限制
  3. 隔离网络:默认禁用网络访问(可配置)
  4. 只读挂载:脚本文件以只读方式挂载到容器
  5. 监控执行:捕获输出并验证执行结果

这样脚本在安全的沙箱环境中运行,不会影响主机系统。

系统级隔离

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)

多格式支持

结果验证和转换#

输出验证

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

预编译优化

错误处理和调试#

脚本执行错误分类#

1. 语法错误

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 时,您不需要编写或修改这些代码,系统会自动处理所有技术细节。