第5课：Skill Loading —— 按需加载知识

你将学到什么

• 什么是"技能（Skill）"机制，为什么 Agent 需要它
• 如何用 YAML frontmatter + Markdown 定义技能文件
• "两层加载"策略的设计思路：目录层 vs 内容层
• 如何让 Agent 自己决定什么时候加载哪个技能
• 最重要的： 省 token、省钱、还能让 Agent 拥有"无限知识"

核心概念：给 Agent 一个"技能书架"

问题：知识太多，塞不下

想象你是一个程序员，面前有100本技术手册。你会怎么做？

• 笨办法： 把100本书全部摊开放在桌上 —— 桌子放不下，而且你大部分时间只需要其中1-2本
• 聪明办法： 桌上只放一份"目录清单"，写着每本书的名字和简介。需要哪本，再从书架上拿下来翻开

这就是"两层加载"策略：

第一层（始终加载）：技能目录            第二层（按需加载）：技能全文
+---------------------------+         +---------------------------+
| Skills available:         |         | <skill name="git">       |
|   - git: Git操作指南 [vcs]|  -----> | 完整的 Git 使用教程...    |
|   - docker: 容器化部署    |         | 各种命令示例...           |
|   - testing: 测试最佳实践 |         | 最佳实践...               |
+---------------------------+         +---------------------------+
    几十个 token                          几千个 token
    每次对话都带着                          只在需要时才加载

为什么不全部塞进系统提示词？

三个字：太贵了。

• 系统提示词里的每个 token，每轮对话都要付费
• 假设你有10个技能，每个5000 token，那就是 50000 token
• 对话10轮，光技能知识就消耗 500000 个 input token
• 用"两层加载"，目录可能只有 200 token，按需加载平均只用 1-2 个技能

省了几十倍的钱！

技能文件长什么样？

每个技能就是一个 Markdown 文件，放在 skills/ 目录下，用 YAML frontmatter 描述元信息：

text复制
skills/
  git/
    SKILL.md          <-- 技能文件
  docker/
    SKILL.md
  testing/
    SKILL.md

一个 SKILL.md 的内容：

markdown复制
---
name: git
description: Git版本控制操作指南
tags: vcs, version-control
---

# Git 操作指南

## 常用命令
- `git status` - 查看状态
- `git add .` - 暂存所有改动
...（完整教程内容）

--- 之间的部分是 YAML 元信息（名字、描述、标签），下面是技能的完整内容。

和上一课的对比：到底改了什么？

上一课的结构：                        本课的结构：

+-- 初始化配置                        +-- 初始化配置
+-- System Prompt（固定内容）          +-- SkillLoader 类（新增！）
                                     |     _load_all: 扫描技能文件
                                     |     _parse_frontmatter: 解析元信息
                                     |     get_descriptions: 生成目录
                                     |     get_content: 加载完整内容
                                     +-- System Prompt（动态拼接技能目录！）
+-- 工具函数们                        +-- 工具函数们（和之前一样）
+-- TOOL_HANDLERS                    +-- TOOL_HANDLERS（多了 load_skill）
+-- TOOLS                            +-- TOOLS（多了 load_skill 工具定义）
+-- agent_loop                       +-- agent_loop（没变！）
+-- main                             +-- main（没变！）

核心变化：

1. 新增了 SkillLoader 类 —— 负责扫描、解析、加载技能文件
2. 系统提示词从"写死的"变成"动态拼接的" —— 自动包含技能目录
3. 新增了 load_skill 工具 —— 让 Agent 可以按需加载技能内容

不变的： agent_loop 一行没改！加工具不需要动循环逻辑。

ASCII 流程图

用户: "帮我用 Git 创建一个新分支"
   |
   v
+---------------------------+
| System Prompt 里有:       |
| Skills available:         |
|   - git: Git操作指南      |    <-- 第一层：只有目录，很便宜
|   - docker: 容器化部署    |
+---------------------------+
   |
   v
+---------------------------+
| AI 判断：这个问题和 git   |
| 有关，我需要加载 git 技能 |
+---------------------------+
   |
   v  调用 load_skill(name="git")
+---------------------------+
| SkillLoader.get_content() |
| 返回完整 Git 技能内容     |    <-- 第二层：按需加载全文
+---------------------------+
   |
   v
+---------------------------+
| AI 拿到知识后，结合知识   |
| 执行 bash 命令创建分支    |
+---------------------------+
   |
   v
"已创建分支 feature/xxx"

关键点： 如果用户问的是 Docker 相关问题，git 技能就永远不会被加载，省了一大笔 token！

完整代码

python复制
#!/usr/bin/env python3
"""s05_skill_loading.py - Skills
按需加载知识的 Agent：用"两层加载"策略管理技能，
第一层只在系统提示词里放技能名和简介，
第二层在 Agent 需要时通过 load_skill 工具加载完整内容。
"""

import os
import re
import subprocess
import yaml                   # 用来解析 YAML 格式的技能元信息
from pathlib import Path

from anthropic import Anthropic
from dotenv import load_dotenv

# ── 环境初始化 ──────────────────────────────────────────────
load_dotenv(override=True)

# 兼容自定义 API 地址（比如用代理或自部署）
if os.getenv("ANTHROPIC_BASE_URL"):
    os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)

WORKDIR = Path.cwd()                                    # 工作目录，所有文件操作都限制在这里面
client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
MODEL = os.environ["MODEL_ID"]                           # 模型 ID，从环境变量读取
SKILLS_DIR = WORKDIR / "skills"                          # 技能文件存放目录


# ── 技能加载器 ──────────────────────────────────────────────
# 这是本课的核心！负责扫描、解析、管理所有技能文件
class SkillLoader:
    def __init__(self, skills_dir: Path):
        self.skills_dir = skills_dir
        self.skills = {}          # 存储所有技能：{名字: {meta, body, path}}
        self._load_all()          # 初始化时就扫描一遍

    def _load_all(self):
        """扫描 skills 目录下所有 SKILL.md 文件，解析并存储"""
        if not self.skills_dir.exists():
            return                # 目录不存在就跳过，不报错
        # rglob 会递归搜索所有子目录，找到所有 SKILL.md
        for f in sorted(self.skills_dir.rglob("SKILL.md")):
            text = f.read_text()
            meta, body = self._parse_frontmatter(text)   # 拆分元信息和正文
            name = meta.get("name", f.parent.name)        # 优先用 YAML 里的 name，否则用目录名
            self.skills[name] = {"meta": meta, "body": body, "path": str(f)}

    def _parse_frontmatter(self, text: str) -> tuple:
        """
        解析 YAML frontmatter。
        文件格式：
            ---
            name: xxx
            description: yyy
            ---
            正文内容...

        返回 (元信息字典, 正文字符串)
        """
        # 用正则匹配 --- 包裹的 YAML 块
        match = re.match(r"^---\n(.*?)\n---\n(.*)", text, re.DOTALL)
        if not match:
            return {}, text       # 没有 frontmatter，整个文件都是正文
        try:
            meta = yaml.safe_load(match.group(1)) or {}   # 解析 YAML
        except yaml.YAMLError:
            meta = {}             # YAML 格式错误，当作空的处理
        return meta, match.group(2).strip()

    def get_descriptions(self) -> str:
        """
        生成技能目录（第一层）。
        这个字符串会被塞进系统提示词，所以要尽量简短！
        每个技能只占一行：名字 + 描述 + 标签
        """
        if not self.skills:
            return "(no skills available)"
        lines = []
        for name, skill in self.skills.items():
            desc = skill["meta"].get("description", "No description")
            tags = skill["meta"].get("tags", "")
            line = f"  - {name}: {desc}"
            if tags:
                line += f" [{tags}]"       # 标签帮助 AI 更好地判断要不要加载
            lines.append(line)
        return "\n".join(lines)

    def get_content(self, name: str) -> str:
        """
        加载技能的完整内容（第二层）。
        只在 Agent 调用 load_skill 工具时才执行。
        用 XML 标签包裹，方便 AI 识别技能内容的边界。
        """
        skill = self.skills.get(name)
        if not skill:
            # 技能不存在，返回错误信息和可用技能列表，帮助 AI 自我纠正
            return f"Error: Unknown skill '{name}'. Available: {', '.join(self.skills.keys())}"
        return f"<skill name=\"{name}\">\n{skill['body']}\n</skill>"


# ── 初始化技能加载器 ─────────────────────────────────────────
SKILL_LOADER = SkillLoader(SKILLS_DIR)

# ── 系统提示词（动态拼接！） ──────────────────────────────────
# 注意这里用 f-string 把技能目录嵌入进去了
# 这样每次启动时，系统提示词会自动包含最新的技能列表
SYSTEM = f"""You are a coding agent at {WORKDIR}.
Use load_skill to access specialized knowledge before tackling unfamiliar topics.

Skills available:
{SKILL_LOADER.get_descriptions()}"""
# ↑ 只有名字和简介，不会很长，每轮对话都带着也不贵


# ── 路径安全函数 ─────────────────────────────────────────────
def safe_path(p: str) -> Path:
    """
    确保路径不会逃逸到工作目录之外。
    比如传入 "../../etc/passwd" 会被拦截。
    """
    path = (WORKDIR / p).resolve()                       # 转为绝对路径
    if not path.is_relative_to(WORKDIR):                 # 检查是否在工作目录内
        raise ValueError(f"Path escapes workspace: {p}")
    return path


# ── 工具执行函数 ─────────────────────────────────────────────
def run_bash(command: str) -> str:
    """执行 shell 命令，有危险命令过滤和超时保护"""
    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
    if any(d in command for d in dangerous):
        return "Error: Dangerous command blocked"
    try:
        r = subprocess.run(command, shell=True, cwd=WORKDIR,
                           capture_output=True, text=True, timeout=120)
        out = (r.stdout + r.stderr).strip()
        return out[:50000] if out else "(no output)"     # 限制输出长度，防止爆 token
    except subprocess.TimeoutExpired:
        return "Error: Timeout (120s)"

def run_read(path: str, limit: int = None) -> str:
    """读取文件内容，可选限制行数"""
    try:
        lines = safe_path(path).read_text().splitlines()
        if limit and limit < len(lines):
            lines = lines[:limit] + [f"... ({len(lines) - limit} more)"]
        return "\n".join(lines)[:50000]
    except Exception as e:
        return f"Error: {e}"

def run_write(path: str, content: str) -> str:
    """写入文件内容，自动创建父目录"""
    try:
        fp = safe_path(path)
        fp.parent.mkdir(parents=True, exist_ok=True)     # 确保目录存在
        fp.write_text(content)
        return f"Wrote {len(content)} bytes"
    except Exception as e:
        return f"Error: {e}"

def run_edit(path: str, old_text: str, new_text: str) -> str:
    """精确替换文件中的指定文本（只替换第一处匹配）"""
    try:
        fp = safe_path(path)
        content = fp.read_text()
        if old_text not in content:
            return f"Error: Text not found in {path}"
        fp.write_text(content.replace(old_text, new_text, 1))
        return f"Edited {path}"
    except Exception as e:
        return f"Error: {e}"


# ── 工具分发表 ──────────────────────────────────────────────
# 把工具名映射到执行函数，agent_loop 查表调用即可
TOOL_HANDLERS = {
    "bash":       lambda **kw: run_bash(kw["command"]),
    "read_file":  lambda **kw: run_read(kw["path"], kw.get("limit")),
    "write_file": lambda **kw: run_write(kw["path"], kw["content"]),
    "edit_file":  lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
    "load_skill": lambda **kw: SKILL_LOADER.get_content(kw["name"]),   # 新增！加载技能
}

# ── 工具定义（告诉 AI 有哪些工具可用） ────────────────────────
TOOLS = [
    {"name": "bash", "description": "Run a shell command.",
     "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
    {"name": "read_file", "description": "Read file contents.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},
    {"name": "write_file", "description": "Write content to file.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
    {"name": "edit_file", "description": "Replace exact text in file.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
    # 新增的 load_skill 工具 —— 这是本课的关键！
    {"name": "load_skill", "description": "Load specialized knowledge by name.",
     "input_schema": {"type": "object", "properties": {"name": {"type": "string", "description": "Skill name to load"}}, "required": ["name"]}},
]


# ── Agent 循环（和之前一模一样！） ────────────────────────────
def agent_loop(messages: list):
    """
    核心 Agent 循环：发消息 -> 收回复 -> 如果要用工具就执行 -> 继续循环
    注意：这个函数从第1课到现在，核心逻辑没变过！
    """
    while True:
        # 1. 把对话历史发给 AI
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        # 2. 把 AI 的回复加入历史
        messages.append({"role": "assistant", "content": response.content})

        # 3. 如果 AI 不想用工具了，说明它已经有最终答案，退出循环
        if response.stop_reason != "tool_use":
            return

        # 4. 遍历所有工具调用，执行并收集结果
        results = []
        for block in response.content:
            if block.type == "tool_use":
                handler = TOOL_HANDLERS.get(block.name)    # 查表找执行函数
                try:
                    output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
                except Exception as e:
                    output = f"Error: {e}"
                print(f"> {block.name}:")                   # 打印工具调用日志
                print(str(output)[:200])                    # 只显示前200字符
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)})

        # 5. 把工具结果作为 user 消息加入历史，继续循环
        messages.append({"role": "user", "content": results})


# ── 主函数 ──────────────────────────────────────────────────
if __name__ == "__main__":
    history = []
    while True:
        try:
            query = input("\033[36ms05 >> \033[0m")        # 青色提示符
        except (EOFError, KeyboardInterrupt):
            break
        if query.strip().lower() in ("q", "exit", ""):
            break
        history.append({"role": "user", "content": query})
        agent_loop(history)
        # 打印 AI 的文本回复
        response_content = history[-1]["content"]
        if isinstance(response_content, list):
            for block in response_content:
                if hasattr(block, "text"):
                    print(block.text)
        print()

代码逐行拆解

模块一：SkillLoader 类 —— 技能管理的核心

这是本课最重要的新增内容。它做三件事：扫描技能文件、生成目录、按需加载。

1. 初始化和扫描

python复制
class SkillLoader:
    def __init__(self, skills_dir: Path):
        self.skills_dir = skills_dir
        self.skills = {}          # 存储所有技能
        self._load_all()          # 启动时扫描一遍

初始化时立刻扫描所有技能文件。这是个"一次性"操作，程序启动时做一次就行。

python复制
    def _load_all(self):
        if not self.skills_dir.exists():
            return
        for f in sorted(self.skills_dir.rglob("SKILL.md")):
            text = f.read_text()
            meta, body = self._parse_frontmatter(text)
            name = meta.get("name", f.parent.name)
            self.skills[name] = {"meta": meta, "body": body, "path": str(f)}

rglob("SKILL.md") 递归搜索所有子目录。每找到一个 SKILL.md，就解析它的 YAML 头部和正文，存到字典里。

2. 解析 YAML frontmatter

python复制
    def _parse_frontmatter(self, text: str) -> tuple:
        match = re.match(r"^---\n(.*?)\n---\n(.*)", text, re.DOTALL)
        if not match:
            return {}, text
        try:
            meta = yaml.safe_load(match.group(1)) or {}
        except yaml.YAMLError:
            meta = {}
        return meta, match.group(2).strip()

这个函数把一个 Markdown 文件拆成两部分：

• --- 之间的 YAML 元信息（名字、描述、标签）
• 剩下的正文内容

如果文件没有 --- 开头，整个文件就当正文处理。

3. 两个关键方法：目录 vs 内容

python复制
    def get_descriptions(self) -> str:    # 第一层：生成目录
        # 每个技能只输出一行：名字 + 描述 + 标签
        # 这个结果会被塞进系统提示词

    def get_content(self, name: str) -> str:    # 第二层：加载全文
        # 返回用 XML 标签包裹的完整技能内容
        # 只在 Agent 调用 load_skill 时才执行

这就是"两层加载"的实现！ get_descriptions() 在程序启动时调用一次，结果嵌入系统提示词。get_content() 在 Agent 需要时通过工具调用触发。

模块二：动态系统提示词

python复制
SKILL_LOADER = SkillLoader(SKILLS_DIR)

SYSTEM = f"""You are a coding agent at {WORKDIR}.
Use load_skill to access specialized knowledge before tackling unfamiliar topics.

Skills available:
{SKILL_LOADER.get_descriptions()}"""

重点看这个 f-string：{SKILL_LOADER.get_descriptions()} 会被替换成技能目录。比如：

text复制
Skills available:
  - git: Git版本控制操作指南 [vcs, version-control]
  - docker: 容器化部署指南 [container, deploy]
  - testing: 测试最佳实践 [test, qa]

只有名字和简介，不到100个 token。但 Agent 看到这个目录后，就知道自己"有这些技能可以用"。

模块三：load_skill 工具

python复制
TOOL_HANDLERS = {
    ...
    "load_skill": lambda **kw: SKILL_LOADER.get_content(kw["name"]),
}

在分发表里多加一行就行了。当 AI 调用 load_skill(name="git") 时，get_content("git") 被执行，返回完整的 Git 技能文档。这个文档作为工具结果回传给 AI，AI 就"学到了"这个知识。

模块四：其他部分

safe_path、run_bash、run_read、run_write、run_edit、agent_loop、main 这些和之前的课程完全一样，没有任何改动。这再次证明了 Agent Loop 架构的扩展性 —— 加新功能不需要改核心循环。

运行效果

准备：创建一个技能文件

先创建一个 Git 技能：

bash复制
mkdir -p skills/git
cat > skills/git/SKILL.md << 'EOF'
---
name: git
description: Git版本控制操作指南
tags: vcs, version-control
---

# Git 操作指南

## 创建新分支
git checkout -b <branch-name>

## 查看所有分支
git branch -a

## 合并分支
git merge <branch-name>

## 撤销最近一次提交（保留修改）
git reset --soft HEAD~1
EOF

示例对话

text复制
s05 >> 帮我创建一个新的 git 分支叫 feature/login

> load_skill:
<skill name="git">
# Git 操作指南
## 创建新分支
git checkout -b <branch-name>
...
</skill>

> bash:
Switched to a new branch 'feature/login'

已经帮你创建并切换到了 feature/login 分支。

注意 Agent 的行为：

1. 看到问题和 git 有关 -> 先调用 load_skill(name="git") 加载知识
2. 拿到知识后 -> 调用 bash 执行 git checkout -b feature/login
3. 最后给出文字回复

如果问一个和技能无关的问题

text复制
s05 >> 当前目录下有哪些文件？

> bash:
skills/
README.md
...

当前目录下有以下文件：...

Agent 没有调用 load_skill！ 因为这个问题不需要任何专业知识，直接用 bash 就能解决。这就是"按需"的意思。

关键收获

1. "两层加载"是省钱利器 —— 系统提示词里只放目录（便宜），完整内容按需加载（节省）
2. 技能文件用 YAML + Markdown —— 元信息和内容分离，结构清晰，方便管理
3. Agent 自己决定加载哪个技能 —— 你不需要硬编码"如果问 git 问题就加载 git 技能"，AI 自己会判断
4. 扩展性极好 —— 想加新技能？在 skills/ 目录下放个 SKILL.md 就行，代码一行不用改
5. agent_loop 依然没变 —— 加工具就是加一行分发表，核心循环永远不需要动

下一课预告