Agent 07：一个简单实现（5）Skills

pdf-skill/
├── SKILL.md            # Skill 元数据和核心指令（必需）
├── forms.md            # 表单填写指南（可选）
├── reference.md        # 详细 API 参考（可选）
└── scripts/            # 实用脚本
    └── extract_fields.py

---
name: pdf-processing
description: Extract text and tables from PDF files, fill forms, merge documents. Use when working with PDF files or when the user mentions PDFs, forms, or document extraction.
---

# PDF Processing

## Quick start

Use pdfplumber to extract text from PDFs...

For advanced form filling, see [forms.md](forms.md).

@dataclass
class SkillDef:
    name: str
    description: str
    triggers: list[str]          # ["/commit", "commit changes"]
    tools: list[str]             # ["Bash", "Read"]  (allowed-tools)
    prompt: str                  # full prompt body after frontmatter
    file_path: str
    # Enhanced fields
    when_to_use: str = ""        # when Claude should auto-invoke this skill
    arguments: list[str] = field(default_factory=list)  # named arg names
    model: str = ""              # model override
    user_invocable: bool = True  # appears in /skills list
    context: str = "inline"      # "inline" or "fork" (fork = sub-agent)
    source: str = "user"         # "user", "project", "builtin"

def load_skills(include_builtins: bool = True) -> list[SkillDef]:
    """Return skills from disk + builtins, deduplicated (project > user > builtin)."""
    seen: dict[str, SkillDef] = {}

    # Builtins go in first (lowest priority)
    if include_builtins:
        for sk in _BUILTIN_SKILLS:
            seen[sk.name] = sk

    # User-level next, project-level last (highest priority)
    skill_paths = _get_skill_paths()
    for i, skill_dir in enumerate(reversed(skill_paths)):
        src = "user" if i == 0 else "project"
        if not skill_dir.is_dir():
            continue
        for md_file in sorted(skill_dir.glob("*.md")):
            skill = _parse_skill_file(md_file, source=src)
            if skill:
                seen[skill.name] = skill

    return list(seen.values())

_SKILL_SCHEMA = {
    "name": "Skill",
    "description": (
        "Invoke a named skill (reusable prompt template). "
        "<available_skills>..."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "name": {
                "type": "string",
                "description": "Skill name (e.g. 'commit', 'review')",
            },
            "args": {
                "type": "string",
                "description": "Arguments to pass to the skill (replaces $ARGUMENTS)",
                "default": "",
            },
        },
        "required": ["name"],
    },
}

{
  "type": "tool_use",
  "id": "toolu_123abc",
  "name": "Skill",
  "input": {
    "name": "pdf"
  }
}

def substitute_arguments(prompt: str, args: str, arg_names: list[str]) -> str:
    """Replace $ARGUMENTS (whole args string) and $ARG_NAME placeholders.

    Named args are positional: first word → first name, etc.
    """
    # Always substitute $ARGUMENTS
    result = prompt.replace("$ARGUMENTS", args)

    # Named args: split by whitespace
    arg_values = args.split()
    for i, arg_name in enumerate(arg_names):
        placeholder = f"${arg_name.upper()}"
        value = arg_values[i] if i < len(arg_values) else ""
        result = result.replace(placeholder, value)

    return result

def execute_skill(
    skill: SkillDef,
    args: str,
    state,
    config: dict,
    system_prompt: str,
) -> Generator:
    """Execute a skill.

    If skill.context == "fork", runs as an isolated sub-agent and yields its events.
    Otherwise (inline), injects the rendered prompt into the current agent loop.

    Args:
        skill: SkillDef to execute
        args: raw argument string from user (after the trigger word)
        state: AgentState
        config: config dict (may contain _depth, model, etc.)
        system_prompt: current system prompt string
    Yields:
        agent events (TextChunk, ToolStart, ToolEnd, TurnDone, …)
    """
    # 加载 SKILL.md，替换参数
    rendered = substitute_arguments(skill.prompt, args, skill.arguments)
    message = f"[Skill: {skill.name}]\n\n{rendered}"

    if skill.context == "fork":
        yield from _execute_forked(skill, message, config, system_prompt)
    else:
        yield from _execute_inline(message, state, config, system_prompt)

def _execute_inline(message: str, state, config: dict, system_prompt: str) -> Generator:
    """Run skill prompt inline in the current conversation."""
    import agent as _agent
    yield from _agent.run(message, state, config, system_prompt)

def _execute_forked(
    skill: SkillDef,
    message: str,
    config: dict,
    system_prompt: str,
) -> Generator:
    """Run skill as an isolated sub-agent (separate conversation context)."""
    import agent as _agent
	
	# 修改执行上下文，覆盖模型和工具
    # Build a sub-agent config with depth tracking
    depth = config.get("_depth", 0) + 1
    sub_config = {**config, "_depth": depth, "_system_prompt": system_prompt}
    if skill.model:
        sub_config["model"] = skill.model
    # allowed-tools 表示这些工具已经预先批准，不需要询问用户
    if skill.tools:
        sub_config["_allowed_tools"].append(skill.tools)

    # Run in fresh state (no shared history)
    # 这里是简化实现，实际上 fork 语义下应该有更清晰的输出机制，现在把所有东西都 yield 出来，工具调用、错误和中间状态都会暴露；更简洁的方式是 forked agent 生成一个最终结果消息，并将其作为单个 tool_result 注入回来
    sub_state = _agent.AgentState()
    yield from _agent.run(message, sub_state, sub_config, system_prompt)

<command-message>The "pdf" skill is loading</command-message>
<command-name>pdf</command-name>
<command-args>report.pdf</command-args>

// Complete message array sent to API for Turn 1
{
  model: "claude-sonnet-4-5-20250929",
  messages: [
    {
      role: "user",
      content: "Extract text from report.pdf"
    },
    {
      role: "assistant",
      content: [
        {
          type: "tool_use",
          id: "toolu_123abc",
          name: "Skill",
          input: { command: "pdf" }
        }
      ]
    },
    {
      role: "user",
      content: "<command-message>The \"pdf\" skill is loading</command-message>\n<command-name>pdf</command-name>"
      // isMeta: false (default) - VISIBLE to user in UI
    },
    {
      role: "user",
      content: "You are a PDF processing specialist...\n\n## Process\n1. Validate PDF exists\n2. Run pdftotext...",
      isMeta: true  // HIDDEN from UI, sent to API
    },
    {
      role: "user",
      content: {
        type: "command_permissions",
        allowedTools: ["Bash(pdftotext:*)", "Read", "Write"],
        model: undefined
      }
    }
  ]
}

def repl(config: dict, initial_prompt: str = None):
	while True:
		# 捕捉用户输入
        try:
            cwd_short = Path.cwd().name
            prompt = clr(f"\n[{cwd_short}] ", "dim") + clr("❯ ", "cyan", "bold")
            user_input = input(prompt).strip()
        if not user_input:
            continue
		
		# 首先处理 "/" 开头的命令，包括内置斜杠命令和 skills
        result = handle_slash(user_input, state, config)
        # 如果识别到 skills，就注入
        if isinstance(result, tuple):
            skill, skill_args = result
            info(f"Running skill: {skill.name}" + (f" [{skill.context}]" if skill.context == "fork" else ""))
            try:
                from skill import substitute_arguments
                rendered = substitute_arguments(skill.prompt, skill_args, skill.arguments)
                run_query(f"[Skill: {skill.name}]\n\n{rendered}")
            except KeyboardInterrupt:
                print(clr("\n  (interrupted)", "yellow"))
            continue
        if result:
            continue
        ...

def handle_slash(line: str, state, config) -> Union[bool, tuple]:
    """Handle /command [args]. Returns True if handled, tuple (skill, args) for skill match."""
    if not line.startswith("/"):
        return False
    parts = line[1:].split(None, 1)
    if not parts:
        return False
    cmd = parts[0].lower()
    args = parts[1] if len(parts) > 1 else ""
    handler = COMMANDS.get(cmd)
    if handler:
        handler(args, state, config)
        return True

    # Fall through to skill lookup
    from skill import find_skill
    skill = find_skill(line)
    if skill:
        cmd_parts = line.strip().split(maxsplit=1)
        skill_args = cmd_parts[1] if len(cmd_parts) > 1 else ""
        # 返回 tuple，告诉 REPL 这是个 skill，要继续处理
        return (skill, skill_args)

    err(f"Unknown command: /{cmd}  (type /help for commands)")
    return True

def find_skill(query: str) -> Optional[SkillDef]:
    """Find a skill whose trigger matches the first word (or whole string) of query."""
    query = query.strip()
    if not query:
        return None

    first_word = query.split()[0]
    for skill in load_skills():
        for trigger in skill.triggers:
            if first_word == trigger:
                return skill
            if trigger.startswith(first_word + " "):
                return skill
    return None