15.5 Skills脚本执行机制

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

脚本在 Skills 中的角色

脚本执行的必要性

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

确定性保证

数学计算的精确性
数据处理的可靠性
复杂算法的正确性

性能优势

大规模数据处理的高效性
重复操作的性能优化
系统集成的低延迟

功能扩展

访问本地文件系统
调用外部 API 和服务
执行系统级操作

脚本 vs 纯 AI 生成

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

脚本集成架构

脚本发现机制

自动脚本扫描

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

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

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

脚本元数据提取

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

文件路径：脚本在磁盘上的位置
编程语言：Python、Shell、JavaScript等
文件大小：脚本占用的存储空间
修改时间：最后更新的时间
执行权限：是否有执行权限
依赖关系：脚本需要的外部库或工具

脚本调用接口

统一调用接口

class ScriptExecutor:
    def __init__(self):
        self.sandbox = ScriptSandbox()
        self.cache = ScriptCache()

    async def execute_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使用限制
隔离网络：默认禁用网络访问（可配置）
只读挂载：脚本文件以只读方式挂载到容器
监控执行：捕获输出并验证执行结果

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

系统级隔离

class SystemSandbox:
    def __init__(self):
        self.chroot_dir = '/tmp/skill_sandbox'
        self.max_runtime = 30  # seconds

    async def execute_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")

脚本结果处理

输出格式标准化

结构化输出解析

def parse_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：表格数据处理
纯文本：简单消息传递

结果验证和转换

输出验证

def validate_script_output(output, expected_schema):
    """
    验证脚本输出是否符合预期格式
    """
    validator = SchemaValidator(expected_schema)

    try:
        validator.validate(output)
        return True, None
    except ValidationError as e:
        return False, str(e)

结果转换

def transform_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)

脚本缓存和优化

执行结果缓存

缓存策略

class ScriptCache:
    def __init__(self, max_size=100):
        self.cache = {}
        self.max_size = max_size
        self.access_order = []

    def get(self, key):
        """获取缓存的执行结果"""
        if key in self.cache:
            # 更新访问顺序
            self.access_order.remove(key)
            self.access_order.append(key)
            return self.cache[key]
        return None

    def put(self, key, value):
        """缓存执行结果"""
        if key in self.cache:
            # 更新现有条目
            self.access_order.remove(key)
        elif len(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)

缓存键生成

def generate_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()

    return f"{script_hash}:{args_hash}:{context_hash}"

性能优化

并行执行

async def execute_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:
        if isinstance(result, Exception):
            processed_results.append({'error': str(result)})
        else:
            processed_results.append(result)

    return processed_results

预编译优化

字节码缓存
AST 预解析
依赖预加载

错误处理和调试

脚本执行错误分类

1. 语法错误

Python 语法错误
导入错误
缩进错误

2. 运行时错误

类型错误
索引错误
文件操作错误

3. 逻辑错误

算法错误
数据处理错误
边界条件错误

调试支持

详细错误信息

def format_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)
    }

调试模式

class ScriptDebugger:
    def __init__(self):
        self.breakpoints = set()
        self.watch_variables = set()

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