15.4 Skills执行流程

说明:本章中的代码示例是为了帮助理解技术原理而提供的。实际使用 Skills 时,您不需要编写或理解这些代码,它们只是系统内部的工作机制。

Skills激活机制

激活触发条件

Skills的激活基于用户请求与Skills描述的匹配度:

1. 显式激活

# 直接指定Skills名称
claude --skill pdf-processing

# 通过斜杠命令
/pdf-processing

2. 隐式激活

AI 代理根据请求内容自动判断:

  • 分析用户查询的关键词
  • 匹配Skills描述中的触发条件
  • 计算语义相似度
  • 选择最相关的Skills

激活决策流程

系统通过多维度分析来决定是否激活某个 Skills:

  1. 关键词匹配:检查用户请求和 Skills 描述中共同出现的关键词
  2. 语义理解:分析请求和 Skills 功能描述的含义相似度
  3. 上下文相关性:考虑当前的对话背景和工作环境
  4. 综合评分:将以上因素加权计算,决定是否激活

只有得分超过阈值的 Skills 才会被激活使用。

指令加载和解析

SKILL.md 完整加载

文件读取流程

加载 Skills 指令就像阅读一本使用说明书:

  1. 找到文件:定位 SKILL.md 文件的位置
  2. 读取内容:完整读取文件的所有内容
  3. 分离部分:区分配置信息和使用说明
  4. 验证格式:确保指令格式正确且完整
  5. 打包返回:将所有信息整理后返回给执行引擎

指令解析和结构化

系统需要理解 Skills 的详细说明,就像分析一篇文章的结构:

  1. 转换格式:将 Markdown 格式转换为便于处理的内部格式
  2. 提取章节:识别文档中的各个部分(如概述、使用方法等)
  3. 收集示例:找到所有使用示例和代码片段
  4. 整理引用:记录相关的外部资源和文档链接
  5. 保留原文:保存原始 Markdown 内容以备不时之需

这样系统就能准确理解 Skills 的功能和使用方法。

上下文注入策略

指令内容注入

将 Skills 指令添加到 AI 的工作记忆中,就像给厨师提供菜谱:

  1. 格式化指令:将指令转换为 AI 容易理解的格式
  2. 选择位置:将指令插入到合适的位置(系统提示之后,用户查询之前)
  3. 空间管理:确保指令不会占用太多工作记忆空间
  4. 更新状态:调整记忆窗口的大小和使用情况

这样 AI 就能记住如何正确使用这个 Skills。

位置优化

  • 指令放置在系统提示之后,用户查询之前
  • 确保指令内容不被截断
  • 保留足够空间给用户交互

任务执行引擎

执行状态管理

执行状态机

Skills 执行过程就像一个状态机,有明确的阶段转换:

  • 空闲状态:等待被调用
  • 加载状态:准备执行环境和资源
  • 执行状态:正在运行 Skills 的步骤
  • 完成状态:成功执行完毕
  • 失败状态:执行过程中出现错误

系统只允许合理的状态转换(如不能从"失败"直接跳到"执行"),确保执行过程的稳定性和可预测性。

步骤执行机制

顺序执行模型

async def execute_skill_steps(skill_data, user_context):
    """
    按顺序执行Skills步骤
    """
    execution_state = SkillExecutionState()

    try:
        # 步骤1: 环境准备
        await prepare_execution_environment(skill_data)

        # 步骤2: 变量初始化
        variables = initialize_variables(skill_data, user_context)

        # 步骤3: 逐步执行
        for step in skill_data['steps']:
            await execute_step(step, variables, execution_state)

            # 检查执行状态
            if execution_state.state == 'failed':
                break

        # 步骤4: 结果整理
        result = await finalize_execution(execution_state)

        return result

    except Exception as e:
        execution_state.transition_to('failed')
        raise SkillExecutionError(f"Execution failed: {e}")

条件分支执行

Skills 可以根据情况做出不同的选择,就像菜谱中的"如果没有鸡蛋,就用...":

  1. 评估条件:检查某个条件是否满足(如"文件存在吗?")
  2. 选择路径:如果条件为真,执行一个步骤;如果为假,执行另一个步骤
  3. 灵活处理:根据实际情况动态调整执行流程

这样可以让 Skills 根据不同情况做出智能决策。

资源动态加载

按需资源加载

Skills 采用"按需取用"的资源管理策略:

  1. 检查缓存:先查看是否已经加载过这个资源
  2. 解析路径:根据引用找到实际的文件位置
  3. 异步加载:在后台加载资源,不阻塞其他操作
  4. 缓存存储:保存加载的资源以便下次使用
  5. 返回内容:将资源提供给执行步骤

这样可以避免重复加载,提高执行效率。

资源类型处理

  • 脚本文件:验证可执行性,准备运行环境
  • 文档文件:解析格式,提取结构化信息
  • 模板文件:验证完整性,准备实例化
  • 数据文件:验证格式,准备处理

脚本执行子系统

脚本安全沙箱

执行环境隔离

class ScriptSandbox:
    def __init__(self):
        self.allowed_modules = set()
        self.timeout_seconds = 30
        self.memory_limit_mb = 100

    async def execute_script(self, script_path, arguments):
        """
        在沙箱中执行脚本
        """
        # 创建隔离环境
        env = self.create_isolated_environment()

        # 设置资源限制
        env.set_limits(memory=self.memory_limit_mb)

        # 执行脚本
        process = await asyncio.create_subprocess_exec(
            'python', script_path, *arguments,
            env=env,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            timeout=self.timeout_seconds
        )

        # 收集结果
        stdout, stderr = await process.communicate()

        return {
            'returncode': process.returncode,
            'stdout': stdout.decode(),
            'stderr': stderr.decode()
        }

安全策略

  • 白名单模块限制
  • 文件系统访问控制
  • 网络访问限制
  • 执行时间限制

脚本结果处理

输出解析

def parse_script_output(script_result):
    """
    解析脚本执行结果
    """
    if script_result['returncode'] != 0:
        # 处理错误输出
        error_info = parse_error_output(script_result['stderr'])
        raise ScriptExecutionError(error_info)

    # 解析标准输出
    output_data = parse_output_data(script_result['stdout'])

    # 验证输出格式
    validate_output_format(output_data)

    return output_data

结果集成

  • 将脚本结果注入到 AI 上下文
  • 更新执行状态变量
  • 触发后续步骤执行

错误处理和恢复

执行错误分类

1. Skills配置错误

  • SKILL.md 格式错误
  • 资源文件缺失
  • 依赖关系问题

2. 执行环境错误

  • 脚本执行失败
  • 资源加载错误
  • 权限不足

3. 逻辑错误

  • 步骤执行失败
  • 条件判断错误
  • 变量处理错误

错误恢复策略

自动恢复

async def attempt_recovery(error, execution_state):
    """
    尝试自动恢复执行
    """
    recovery_strategies = {
        'resource_missing': retry_with_fallback,
        'script_timeout': reduce_complexity,
        'permission_denied': request_elevation
    }

    strategy = recovery_strategies.get(error.type)
    if strategy:
        return await strategy(error, execution_state)

    return False

手动干预

  • 提供详细错误信息
  • 建议修复步骤
  • 支持断点续执行

执行监控和日志

执行跟踪

class ExecutionTracer:
    def __init__(self):
        self.trace_log = []
        self.performance_metrics = {}

    def log_step_execution(self, step_name, start_time, end_time, result):
        """
        记录步骤执行信息
        """
        trace_entry = {
            'step': step_name,
            'duration': end_time - start_time,
            'result': result,
            'timestamp': datetime.now()
        }

        self.trace_log.append(trace_entry)
        self.update_metrics(trace_entry)

性能监控

  • 执行时间统计
  • 资源使用监控
  • 成功率跟踪
  • 瓶颈识别

总结

Skills执行流程是一个复杂但高度优化的系统,它整合了激活机制、指令解析、步骤执行、资源管理和错误处理等多个组件。通过精细的状态管理和监控,Skills 能够可靠地执行各种复杂的任务。

技术说明:本章中的代码示例展示了系统内部的工作机制,旨在帮助您理解原理。实际创建和使用 Skills 时,您不需要编写或修改这些代码,系统会自动处理所有技术细节。