# Skill 实战：教 Agent 写你风格的博客

作者：Hex4C59 · 发布于 2026-03-29

原文：https://hex4c59.cc/agent/build-skill-expert-agent/


模型最后回复“文章已保存”，不能说明文章已经写出，更不能说明它符合规范。如果检查者只收到这句话，即使它认真列出检查清单，也没有看到需要检查的对象。

这篇教程把过程做完整：从 Markdown 文件选择写作规则，让 Agent 通过受限工具写出一篇草稿，再由 Python **重新打开那份文件**检查。用户允许长期保存一条规则时，才调用已经注册的 `create_skill` 工具，并立即重新扫描文件库。

读者需要能读写简单 Python，理解函数、JSON 和文件路径。这里的 Skill 是带元数据的操作说明；加载它会改变模型收到的文字，不会训练模型，也不会增加文件权限。示例默认运行本地固定响应，只有显式选择 `--live` 才调用配置的模型服务。

[Skill 概念篇](https://hex4c59.cc/agent/skill-teach-agent-how-to-work/)介绍了把领域规则放进独立文档的思路。本篇关注这套文档怎样被选中、怎样参与调用，以及写出文件以后究竟验证了什么。

## 先给结论

- 把规则放进 Skill 文件，可以独立维护说明；程序仍须实现扫描、匹配和读取，文件不会自己生效。
- 工具权限由 Python 检查。模型不能通过增加“已批准”参数来覆盖文件，也不能绕过 `create_skill` 直接改规则库。
- 文章验收读取磁盘产物。文件缺失、不是本轮写出、结构不符合要求，都会使命令退出失败。
- 本例只检查结构和少量文本条件。它不判定事实、代码运行结果或作者风格；fixture 通过也不是模型能力证据。

## 先运行完整示例

下载[Skill Agent 完整示例](/downloads/agent-examples/skills.zip)，解压后进入 `skills` 目录。需要 Python 3.11 或更高版本；下载包自带共享 runtime 和固定依赖清单。

以下命令适用于 macOS/Linux 的 shell：

```sh
unzip skills.zip -d skills
cd skills
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python app.py
python -m unittest -v test_app
```

Windows 可以用 `.venv\Scripts\python.exe` 执行相同的 Python 命令，无需使用上面的激活方式。本次实际运行环境是 macOS arm64、Python 3.11.16，没有在 Windows 或 Linux 上重跑本例。

默认命令会创建独立临时工作区，复制随包的示例资料和规则，完成后清理。它不把文章写进当前项目。若想查看并保留文件，可以明确提供数据目录：

```sh
work=$(mktemp -d)
python app.py --workspace "$work"
cat "$work/drafts/article.md"
```

默认运行的实际输出先标明：

```text
mode=fixture scenario=success (本地固定响应)
```

随后 JSON 中的 `status` 为 `complete`，`selected_skills` 为 `["write-blog"]`，`artifact_check.artifact_read` 为 `true`、`errors` 为空，`semantic_correctness` 为 `"not-assessed"`。这表示固定响应确实调用了工具，文件被读回并通过结构检查；没有调用模型 API。

## 文件和规则分别放在哪里

下载包中的 `app.py` 是实现来源，`test_app.py` 验证行为。共享的 `agent_runtime.py` 负责模型与工具之间的循环，按 JSON Schema 检查参数，再调用登记的 Python 函数。模型请求使用工具，与工具实际执行，是两项工作；这一分工也可对照 [OpenAI 的函数调用说明](https://developers.openai.com/api/docs/guides/function-calling)。

运行时的数据目录采用下面的布局：

```text
workspace/
├── inputs/source-notes.md
├── inputs/skill-proposal.json
├── drafts/article.md
└── skills/
    ├── write-blog/SKILL.md
    ├── write-blog/references/structure.md
    └── explain-code/SKILL.md
```

`source-notes.md` 是合成任务资料：只讨论一个进程中的 Python 字典缓存，明确没有性能测量。文章中的数据和条件按这个范围写，不编造耗时、命中率或费用。

两份 Skill 也是本教程自己的练习规则。它们不代表这个博客当前的 front matter、文章长度、系列编号或配图规范。启动时只补上缺失的随包示例，不覆盖工作区已有文件。

## Skill 怎样被加载和选中

先看随包的写作 Skill。下面直接引用同一份源文件：


```md
---
name: write-blog
description: >-
  Write a small Chinese teaching draft from the supplied notes. Use when the task
  explicitly requests a blog draft. These are this tutorial's sample rules.
triggers:
  - 写博客
  - 写一篇
  - blog draft
references:
  - references/structure.md
---

# 写作示例规则

为会写简单 Python 的读者解释任务指定的主题。先读取工作区里的
`inputs/source-notes.md`，区分资料给出的事实与为了演示而设置的条件。

1. 按 references/structure.md 组织草稿；写清对象、动作及至少一种失效条件。
2. 把文章写到本次任务指定的 `drafts/` Markdown 路径。现有文件须由外部授权覆盖。
3. 写入后说明文件路径。程序还会重新读回文件检查结构；保存说明不能代替文章。

这些规则只用于本教程的合成草稿，不是任何现有博客的发布规范。

```


文件开头 `---` 包住的是 YAML 元数据，后面是给 Agent 阅读的正文。这个加载器要求四项元数据：

| 字段 | 在本例中的用途 |
| --- | --- |
| `name` | 精确选择的名称，必须与所在目录名一致 |
| `description` | 告诉模型这份规则用于什么 |
| `triggers` | 自动匹配使用的连续文本片段 |
| `references` | 选中后还需读取的参考文件，相对于本 Skill 目录 |

`triggers`、`references` 是这个加载器自定义的字段，不是所有 Skill 平台都支持的接口。程序使用 PyYAML 的安全加载器解析 YAML，拒绝重复键、缺少字段、未知字段和不合要求的名称，不再用几条正则猜测 YAML 内容。

`SkillLoader.scan()` 按路径排序扫描 `skills/*/SKILL.md`，解析并缓存正文。只扫描这一层，不递归寻找任意深度的 Skill。每次扫描替换原有注册表，所以刷新不会把同一文件重复加入内存。

选择有三种明确用法：

```sh
python app.py
python app.py --skill write-blog
python app.py --no-skills
```

第一条按任务文字匹配：任务和触发词都经过 `casefold()`，只比较是否包含连续子串。例如 `BLOG DRAFT` 能命中 `blog draft`；“明天天气怎样”不会命中。它不理解近义词，也没有使用向量检索。

第二条精确选择名称，可重复 `--skill` 组合规则；名称不存在时直接报错。第三条明确选空列表，不退回自动匹配。没有匹配也不会加载全部 Skill。

选中后，`context()` 才读取 `references` 列出的文件，且路径只能留在该 Skill 的目录。写作规范及其参考内容，与基础说明组成传给公共循环的一条 system 消息。新规则是输入中的指导，仍受用户任务和工具权限约束；程序没有实现任意规则之间的语义冲突检测。

## 文件工具怎样限制读写

本例注册了 `read_file`、`write_file`、`match_skills` 和 `create_skill` 四个工具。名称、schema 和 handler 都在 `tools_for()` 中，工具不是只向模型展示一份参数说明。

`read_file` 只能读取工作区内的 UTF-8 普通文件，每次最多 64 KiB。`write_file` 只能写 `drafts/` 下的 Markdown 文件，大小限制相同。绝对路径、父目录跳转、符号链接都会被拒绝；写作工具不能修改 `skills/` 或 `inputs/`。因此，即使模型要求把一条新规则写入规则库，普通写工具也不能替它绕过持久化授权。

第一次写入新草稿不需要覆盖批准。若目标已经存在，先查看内容，再在下一次运行中明确授权：

```sh
cat "$work/drafts/article.md"
python app.py --workspace "$work" --approve-overwrite drafts/article.md
```

批准绑定启动时读到的文件摘要，只能使用一次。文件在批准后改变，写入会被拒绝。工具参数中没有供模型填写的 `approved` 或 `overwrite` 开关。

这些检查是可信本地用户工作区的边界，不是操作系统沙箱，也不防另一个恶意进程同时替换目录。临时文件加替换用于本例写入，不等于实现文件锁或任意崩溃恢复。

## 质量门禁必须看到实际文章

`write_file` 返回路径、字节数和 SHA-256。摘要标识这次写入的内容，不能代替文章本身。共享循环结束后，`execute()` 总会调用 `check_artifact()`，由它重新读取指定文件并检查：

1. 文件能读取，而且本轮确实通过写工具写入过。上一次留下的合格文章不能冒充本轮产物。
2. 读回内容的摘要与本轮最后一次成功写入记录一致；写入后文件被替换，检查不会继续采用内存里的旧正文。
3. 文章的 TOML front matter 能解析，`title`、`description` 是非空字符串，`draft = true`，`tags` 是非空字符串列表。
4. 正文出现任务指定的主题，并包含有内容的“先给结论”“示例”“参考资料”二级章节。
5. 代码围栏闭合，至少有一个标为 `python` 的非空代码块；Python 能解析出语句。这里不执行文章代码。
6. 参考资料包含 Markdown 链接。HTTPS 外链需要有主机名；本地引用必须指向工作区内存在的普通文件，不能通过符号链接越界。

这是针对本教程草稿格式写出的 Python 门禁，不会从任意 Skill 的文字清单自动生成。`--no-skills` 只改变模型收到的指导，不关闭门禁。这个入口验收的是一篇草稿，并不是任意任务都能使用的通用聊天入口。

门禁失败时，程序不会因为最后一句回复好听而报告完成。反过来，结构通过也不等于事实正确：一个主题词可能出现在错误解释里，一个链接可能指向不可靠的内容，能解析的 Python 也可能算错。结果因此明确保留 `semantic_correctness: "not-assessed"`，事实与风格仍需人工审阅。

## 用漏写和错误文件检验门禁

下面两个命令是刻意保留的负例，都应退出 `1`：

```sh
python app.py --fixture skip
python app.py --fixture invalid
```

`skip` 的固定响应只说“文章已保存”，没有调用写工具。公共循环会正常交回这句话，文章门禁却得到 `artifact_read: false`，因此整体是 `validation-failed`。工作区即使已有一篇合法文章，也会因“本轮没有写入”而失败。

`invalid` 确实写出一份缺少 front matter 和代码的文件。门禁能够读到它，`artifact_read` 为 `true`，但错误列表不为空，仍不能报告完成。

这组差异比给最后回复打分更直接：检查对象始终是磁盘文件；两次最后回复相同，结果却取决于文章是否存在、内容是什么。文件失败时会留下可检查的错误原因，显式工作区中的产物也保留供人查看，不自动修改或重新发布。

## 创建长期 Skill，需要批准完整提案

用户在一次对话里纠正措辞，不自动表示希望长期保留规则。把偏好写进 Skill 会影响之后的任务，因此本例将批准放在模型之外。

随包的 `sample-data/skill-proposal.json` 是可审阅的合成提案，包含名称、用途、触发词和完整正文。先查看，再明确批准：

```sh
cat sample-data/skill-proposal.json
python app.py --fixture learn --approve-skill-proposal inputs/skill-proposal.json
```

默认临时工作区复制的是完全相同的随包提案。若要跨运行保留新 Skill，使用显式工作区，并查看工作区中的提案后再批准；已有草稿也须另行批准覆盖：

```sh
cat "$work/inputs/skill-proposal.json"
python app.py --workspace "$work" --fixture learn \
  --approve-skill-proposal inputs/skill-proposal.json \
  --approve-overwrite drafts/article.md
```

`Workspace.approve_proposal()` 记录完整提案的摘要。`create_skill` 只接受与批准完全一致的名称、说明、触发词和正文，不接受模型自行声明的批准。它不会覆盖已有 Skill 目录；普通写工具也不能写入这个目录。

创建成功后，`SkillLoader.scan()` 立即重新读取规则库。`learn` 固定响应随后真正调用 `match_skills("缓存检查")`，能够取得新建的 `explain-cache` 正文。扫描次数从 `1` 变成 `2`，当前注册表包含新规则。已经发出的初始 system 消息不会被追溯修改；本轮通过匹配工具取得新内容，下一轮任务则从刷新后的库选择。

不带批准参数执行 `python app.py --fixture learn` 会拒绝创建，退出 `1`，也不会产生新 Skill。测试还让模型给工具增加 `approved: true`，确认 schema 会拒绝额外参数。这些规则由工具处理函数和参数验证执行，不只写在 prompt 里。

这与[记忆类型与设计](https://hex4c59.cc/agent/memory-types-and-design/)讨论的持久化有关，但本例保存的是人明确批准的操作说明，没有实现自动推断用户偏好或从对话自行学习。

## 完整实现与可选模型调用

下面引用下载包所使用的 `app.py`。规则扫描、工具注册、授权与文章门禁都以这份源码为准，无需把多个过时片段拼起来：


````py
"""A runnable, local Skill loader and artifact-reading writing agent (Python 3.11+)."""
from __future__ import annotations

import argparse
import ast
import asyncio
from dataclasses import asdict, dataclass
import hashlib
import json
from pathlib import Path
import re
import tempfile
import tomllib
from urllib.parse import urlsplit

import yaml
from agent_runtime import (ScriptedModel, Tool, atomic_write_text, confined_path,
                           model_from_env, run_agent)

MAX_BYTES = 64 * 1024
NAME = re.compile(r"[a-z][a-z0-9-]{0,47}")
REQUIRED_HEADINGS = ("先给结论", "示例", "参考资料")


class UniqueLoader(yaml.SafeLoader):
    """Reject duplicate YAML keys rather than silently replacing a rule."""


def unique_mapping(loader, node):
    result = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node)
        if not isinstance(key, str) or key in result:
            raise ValueError("Skill metadata keys must be unique strings")
        result[key] = loader.construct_object(value_node)
    return result


UniqueLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, unique_mapping)


def digest(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def read_text(path: Path) -> str:
    if not path.exists():
        raise FileNotFoundError("File does not exist")
    if not path.is_file():
        raise ValueError("Expected a regular file")
    with path.open("rb") as stream:
        data = stream.read(MAX_BYTES + 1)
    if len(data) > MAX_BYTES:
        raise ValueError("File exceeds the 64 KiB limit")
    return data.decode("utf-8")


def proposal_digest(proposal: dict) -> str:
    return digest(json.dumps(proposal, ensure_ascii=False, sort_keys=True))


def validate_proposal(proposal: dict) -> None:
    if set(proposal) != {"name", "description", "triggers", "content"}:
        raise ValueError("A proposal requires name, description, triggers and content")
    if not isinstance(proposal["name"], str) or not NAME.fullmatch(proposal["name"]):
        raise ValueError("Invalid Skill name")
    for field in ("description", "content"):
        if not isinstance(proposal[field], str) or not proposal[field].strip():
            raise ValueError("Skill description and content must not be empty")
    triggers = proposal["triggers"]
    if (not isinstance(triggers, list) or not 1 <= len(triggers) <= 12
            or any(not isinstance(t, str) or not t.strip() or len(t) > 80 for t in triggers)):
        raise ValueError("Provide 1–12 non-empty trigger strings")
    if not re.search(r"^# .+", proposal["content"], re.M):
        raise ValueError("Skill content requires a top-level heading")
    if len(json.dumps(proposal, ensure_ascii=False).encode("utf-8")) > MAX_BYTES // 2:
        raise ValueError("Skill proposal is too large")


@dataclass(frozen=True)
class Skill:
    name: str
    description: str
    triggers: tuple[str, ...]
    content: str
    path: str
    references: tuple[str, ...]


def parse_skill(text: str, relative: str) -> Skill:
    lines = text.splitlines()
    if not lines or lines[0] != "---":
        raise ValueError("Skill requires YAML front matter")
    try:
        end = lines.index("---", 1)
    except ValueError as error:
        raise ValueError("Unclosed Skill front matter") from error
    metadata = yaml.load("\n".join(lines[1:end]), Loader=UniqueLoader)
    if not isinstance(metadata, dict):
        raise ValueError("Skill metadata must be a mapping")
    if set(metadata) != {"name", "description", "triggers", "references"}:
        raise ValueError("Skill metadata requires name, description, triggers, references")
    content = "\n".join(lines[end + 1:]).strip()
    proposal = {key: metadata[key] for key in ("name", "description", "triggers")}
    proposal["content"] = content
    validate_proposal(proposal)
    references = metadata["references"]
    if (not isinstance(references, list) or len(references) > 4
            or any(not isinstance(p, str) or not p.strip() for p in references)):
        raise ValueError("references must contain at most four relative paths")
    return Skill(proposal["name"], proposal["description"], tuple(proposal["triggers"]),
                 content, relative, tuple(references))


class Workspace:
    """Local-user file boundary. This is not a hostile-process sandbox."""

    def __init__(self, root: Path):
        confined_path(root, ".", allow_root=True)
        self.root = root.resolve()
        self.overwrites: dict[str, str] = {}
        self.approved_proposals: set[str] = set()
        self.written: dict[str, str] = {}
        self.reads: list[str] = []

    def read(self, relative: str) -> str:
        text = read_text(confined_path(self.root, relative))
        self.reads.append(relative)
        return text

    def approve_overwrite(self, relative: str) -> None:
        target = confined_path(self.root, relative)
        if target.relative_to(self.root).parts[0] != "drafts" or target.suffix != ".md":
            raise PermissionError("Only draft Markdown paths can be approved for overwrite")
        self.overwrites[relative] = digest(self.read(relative))

    def approve_proposal(self, relative: str) -> None:
        proposal = json.loads(self.read(relative))
        validate_proposal(proposal)
        self.approved_proposals.add(proposal_digest(proposal))

    def write_file(self, path: str, content: str) -> dict:
        target = confined_path(self.root, path)
        if target.relative_to(self.root).parts[0] != "drafts" or target.suffix != ".md":
            raise PermissionError("write_file can write only Markdown under drafts/")
        if len(content.encode("utf-8")) > MAX_BYTES:
            raise ValueError("Draft exceeds the 64 KiB limit")
        if target.exists():
            current = digest(self.read(path))
            if self.overwrites.get(path) != current:
                raise PermissionError("Overwrite requires external approval for the current file")
            del self.overwrites[path]
        target.parent.mkdir(parents=True, exist_ok=True)
        target = confined_path(self.root, path)
        atomic_write_text(target, content)
        # Record actual bytes decoded from disk, including the platform's line endings.
        written_text = self.read(path)
        self.written[path] = digest(written_text)
        return {"path": path, "bytes": len(written_text.encode("utf-8")), "sha256": digest(written_text)}


class SkillLoader:
    def __init__(self, workspace: Workspace):
        self.workspace = workspace
        self.skills: dict[str, Skill] = {}
        self.scans = 0
        self.scan()

    def scan(self) -> None:
        directory = confined_path(self.workspace.root, "skills")
        if not directory.is_dir():
            raise ValueError("Initialize a skills/ directory first")
        fresh = {}
        for path in sorted(directory.glob("*/SKILL.md")):
            relative = path.relative_to(self.workspace.root).as_posix()
            skill = parse_skill(self.workspace.read(relative), relative)
            if skill.name != path.parent.name or skill.name in fresh:
                raise ValueError("Skill names must be unique and match directory names")
            fresh[skill.name] = skill
        self.skills = fresh
        self.scans += 1

    def select(self, task: str, names: list[str] | None = None) -> list[Skill]:
        if names is not None:
            missing = set(names) - self.skills.keys()
            if missing:
                raise ValueError("Unknown Skill: " + ", ".join(sorted(missing)))
            return [self.skills[name] for name in dict.fromkeys(names)]
        folded = task.casefold()
        return [skill for skill in self.skills.values()
                if any(trigger.casefold() in folded for trigger in skill.triggers)]

    def context(self, selected: list[Skill]) -> str:
        parts = []
        for skill in selected:
            parts.append(f"## Skill {skill.name}\n{skill.description}\n{skill.content}")
            skill_root = confined_path(self.workspace.root, str(Path(skill.path).parent))
            for reference in skill.references:
                target = confined_path(skill_root, reference)
                relative = target.relative_to(self.workspace.root).as_posix()
                parts.append(f"### Reference {reference}\n{self.workspace.read(relative)}")
        result = "\n\n".join(parts)
        if len(result.encode("utf-8")) > MAX_BYTES:
            raise ValueError("Selected Skill context exceeds 64 KiB")
        return result

    def create_skill(self, name: str, description: str, triggers: list[str], content: str) -> dict:
        proposal = dict(name=name, description=description, triggers=triggers, content=content)
        validate_proposal(proposal)
        approved = proposal_digest(proposal)
        if approved not in self.workspace.approved_proposals:
            raise PermissionError("Persisting a Skill requires explicit approval of this exact proposal")
        relative = f"skills/{name}/SKILL.md"
        target = confined_path(self.workspace.root, relative)
        if target.exists() or target.parent.exists():
            raise FileExistsError("create_skill never overwrites an existing Skill directory")
        metadata = dict(name=name, description=description, triggers=triggers, references=[])
        text = "---\n" + yaml.safe_dump(metadata, allow_unicode=True, sort_keys=False) + "---\n\n" + content
        parse_skill(text, relative)
        target.parent.mkdir()
        atomic_write_text(confined_path(self.workspace.root, relative), text)
        self.workspace.approved_proposals.remove(approved)
        self.scan()  # Replace the registry immediately; the next match_skills call sees it.
        return {"created": relative, "available": sorted(self.skills), "scan": self.scans}


def schema(properties: dict) -> dict:
    return {"type": "object", "properties": properties, "required": list(properties),
            "additionalProperties": False}


def tools_for(workspace: Workspace, loader: SkillLoader) -> list[Tool]:
    string = {"type": "string", "minLength": 1}
    def match_skills(task: str) -> dict:
        selected = loader.select(task)
        return {"names": [s.name for s in selected], "context": loader.context(selected)}
    return [
        Tool("read_file", "Read a UTF-8 file inside the workspace (64 KiB maximum).",
             schema({"path": string}), lambda path: {"content": workspace.read(path)}),
        Tool("write_file", "Write a draft Markdown file; host approval is required to overwrite.",
             schema({"path": string, "content": string}), workspace.write_file),
        Tool("match_skills", "Match the current registry and return selected Skill text and references.",
             schema({"task": string}), match_skills),
        Tool("create_skill", "Persist exactly a human-approved proposal; unapproved calls are rejected.",
             schema({"name": string, "description": string,
                     "triggers": {"type": "array", "items": string, "minItems": 1, "maxItems": 12},
                     "content": string}), loader.create_skill),
    ]


@dataclass
class ArtifactCheck:
    ok: bool
    path: str
    artifact_read: bool
    sha256: str | None
    errors: list[str]
    semantic_correctness: str = "not-assessed"


def check_artifact(workspace: Workspace, path: str, required_text: str) -> ArtifactCheck:
    """Always reread disk; never accept the assistant's final save message as evidence."""
    errors = []
    try:
        text = workspace.read(path)
    except FileNotFoundError:
        return ArtifactCheck(False, path, False, None, ["missing-artifact"])
    except (OSError, ValueError, UnicodeError) as error:
        return ArtifactCheck(False, path, False, None, ["artifact-unreadable: " + type(error).__name__])
    if path not in workspace.written:
        errors.append("artifact-not-written-this-run")
    elif digest(text) != workspace.written[path]:
        errors.append("artifact-changed-since-write")
    lines = text.splitlines()
    body = ""
    try:
        if not lines or lines[0] != "+++":
            raise ValueError("Expected TOML front matter")
        end = lines.index("+++", 1)
        metadata = tomllib.loads("\n".join(lines[1:end]))
        body = "\n".join(lines[end + 1:])
        if any(not isinstance(metadata.get(k), str) or not metadata[k].strip()
               for k in ("title", "description")):
            errors.append("frontmatter-title-or-description")
        if metadata.get("draft") is not True:
            errors.append("frontmatter-draft-must-be-true")
        tags = metadata.get("tags")
        if (not isinstance(tags, list) or not tags
                or any(not isinstance(t, str) or not t.strip() for t in tags)):
            errors.append("frontmatter-tags")
    except (ValueError, tomllib.TOMLDecodeError):
        errors.append("invalid-frontmatter")
    prose, blocks, opened, current = [], [], None, []
    language = ""
    for line in body.splitlines():
        fence = re.match(r"^ {0,3}(`{3,}|~{3,})(.*)$", line)
        if opened is None and fence:
            opened = fence.group(1)
            language = fence.group(2).strip()
            current = []
        elif opened is not None:
            if (fence and fence.group(1)[0] == opened[0]
                    and len(fence.group(1)) >= len(opened) and not fence.group(2).strip()):
                blocks.append((language, "\n".join(current)))
                opened = None
            else:
                current.append(line)
        else:
            prose.append(line)
    if opened is not None:
        errors.append("unclosed-code-fence")
    prose_text = "\n".join(prose)
    sections = re.split(r"^## (.+)\s*$", prose_text, flags=re.M)
    section_map = dict(zip(sections[1::2], sections[2::2]))
    for heading in REQUIRED_HEADINGS:
        if not section_map.get(heading, "").strip():
            errors.append("missing-section: " + heading)
    if required_text.casefold() not in prose_text.casefold():
        errors.append("missing-expected-content: " + required_text)
    python_blocks = [code for lang, code in blocks if lang == "python" and code.strip()]
    if not python_blocks:
        errors.append("missing-python-code")
    for code in python_blocks:
        try:
            if not ast.parse(code).body:
                errors.append("empty-python-program")
        except SyntaxError:
            errors.append("invalid-python-syntax")
    links = re.findall(r"\[([^\]\n]+)\]\(([^\s)]+)\)", section_map.get("参考资料", ""))
    if not links:
        errors.append("missing-reference")
    for _, target in links:
        try:
            url = urlsplit(target)
        except ValueError:
            errors.append("invalid-reference-target")
            continue
        if url.scheme:
            if url.scheme != "https" or not url.netloc:
                errors.append("invalid-reference-target")
        else:
            # References may use ../inputs from drafts/, but their normalized target must stay in root.
            try:
                candidate = (workspace.root / path).parent / url.path
                relative = candidate.resolve().relative_to(workspace.root).as_posix()
                # Check every lexical component too, so resolution cannot hide a symlink.
                for parent in [candidate, *candidate.parents]:
                    if parent == workspace.root:
                        break
                    if parent.is_symlink():
                        raise ValueError("Symlink reference")
                read_text(confined_path(workspace.root, relative))
            except (OSError, ValueError, UnicodeError):
                errors.append("missing-or-outside-reference")
    return ArtifactCheck(not errors, path, True, digest(text), errors)


async def execute(model, workspace: Workspace, loader: SkillLoader, *, task: str,
                  artifact: str = "drafts/article.md", topic: str = "缓存",
                  names: list[str] | None = None, timeout: float = 30.0) -> dict:
    workspace.written.clear()  # A previous run's valid file is not a new artifact.
    selected = loader.select(task, names)
    context = loader.context(selected)
    system = ("你是教学写作助手。用户任务和工具权限优先于下面可信的示例 Skill。"
              "先读 inputs/source-notes.md，再把完整草稿写到指定路径。"
              "不要把长期偏好当作已经获准保存，create_skill 的授权由程序检查。"
              "完成声明不能代替文件；稍后程序会从磁盘读取并检查。\n" + context)
    result = await run_agent(model, system, f"{task}\n产物路径：{artifact}\n正文主题：{topic}",
                             tools_for(workspace, loader), max_turns=8,
                             max_tool_calls=16, timeout=timeout)
    artifact_check = check_artifact(workspace, artifact, topic)
    tool_errors = [json.loads(m["content"]) for m in result.messages
                   if m.get("role") == "tool" and json.loads(m["content"]).get("ok") is False]
    status = ("complete" if result.status == "complete" and artifact_check.ok and not tool_errors
              else "validation-failed" if not artifact_check.ok else "incomplete")
    return {"status": status, "runtime_status": result.status,
            "selected_skills": [s.name for s in selected], "scans": loader.scans,
            "available_skills": sorted(loader.skills), "tool_errors": len(tool_errors),
            "artifact_check": asdict(artifact_check)}


def sample_article(topic: str = "缓存") -> str:
    return ('+++\ntitle = "字典缓存示例"\ndescription = "一个合成的缓存草稿"\n'
            'draft = true\ntags = ["Python"]\n+++\n\n## 先给结论\n'
            + topic + '先按键查已有值，缺失时再计算。以下是假设数据，不含性能实测。\n\n'
            '## 示例\n假设输入固定，同一进程中重复请求相同的键：\n\n'
            '```python\ncache = {}\nkey = "demo"\nif key not in cache:\n'
            '    cache[key] = 2 + 3\nprint(cache[key])\n```\n\n'
            '原始数据变化但不更新缓存时，旧结果仍可能被读取。\n\n## 参考资料\n'
            '[合成任务资料](../inputs/source-notes.md)\n')


def call(name: str, arguments: dict, number: int) -> dict:
    return {"role": "assistant", "content": None, "tool_calls": [{"id": f"call-{number}",
            "type": "function", "function": {"name": name,
            "arguments": json.dumps(arguments, ensure_ascii=False)}}]}


def fixture_model(workspace: Workspace, scenario: str, artifact: str, topic: str) -> ScriptedModel:
    if scenario == "skip":
        return ScriptedModel([{"role": "assistant", "content": "文章已保存。"}])
    responses = []
    if scenario == "learn":
        proposal = json.loads(workspace.read("inputs/skill-proposal.json"))
        responses.extend([call("create_skill", proposal, 1),
                          call("match_skills", {"task": "缓存检查"}, 2)])
    responses.append(call("read_file", {"path": "inputs/source-notes.md"}, 3))
    content = "缺少 front matter 和代码的文章" if scenario == "invalid" else sample_article(topic)
    responses.append(call("write_file", {"path": artifact, "content": content}, 4))
    responses.append({"role": "assistant", "content": "文章已保存。"})
    return ScriptedModel(responses)


def initialize_samples(root: Path) -> None:
    """Seed explicit built-in demo files, never overwrite existing user files."""
    source = Path(__file__).resolve().parent
    confined_path(root, ".", allow_root=True)
    for directory in ("drafts", "inputs", "skills"):
        confined_path(root, directory).mkdir(exist_ok=True)
    for package_dir, target_dir in (("sample-skills", "skills"), ("sample-data", "inputs")):
        for item in sorted((source / package_dir).rglob("*")):
            if not item.is_file():
                continue
            relative = (Path(target_dir) / item.relative_to(source / package_dir)).as_posix()
            target = confined_path(root, relative)
            if not target.exists():
                target.parent.mkdir(parents=True, exist_ok=True)
                atomic_write_text(confined_path(root, relative), read_text(item))


async def run_cli(args, root: Path) -> dict:
    initialize_samples(root)
    workspace = Workspace(root)
    for relative in args.approve_overwrite:
        workspace.approve_overwrite(relative)
    for relative in args.approve_skill_proposal:
        workspace.approve_proposal(relative)
    loader = SkillLoader(workspace)
    names = [] if args.no_skills else args.skill
    model = model_from_env() if args.live else fixture_model(workspace, args.fixture, args.output, args.topic)
    try:
        return await execute(model, workspace, loader, task=args.task, artifact=args.output,
                             topic=args.topic, names=names)
    finally:
        if args.live:
            await model.aclose()


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--workspace", type=Path, help="Explicit data directory; otherwise a temporary directory is cleaned up")
    parser.add_argument("--live", action="store_true", help="Use configured live API instead of fixtures")
    parser.add_argument("--fixture", choices=("success", "skip", "invalid", "learn"), default="success")
    parser.add_argument("--task", default="写一篇解释缓存的博客草稿，依据提供的合成资料。")
    parser.add_argument("--topic", default="缓存", help="Text that must appear in the article prose")
    parser.add_argument("--output", default="drafts/article.md")
    choose = parser.add_mutually_exclusive_group()
    choose.add_argument("--skill", action="append", help="Exact Skill name; repeat to combine")
    choose.add_argument("--no-skills", action="store_true")
    parser.add_argument("--approve-overwrite", action="append", default=[], metavar="PATH")
    parser.add_argument("--approve-skill-proposal", action="append", default=[], metavar="PATH")
    args = parser.parse_args(argv)
    print("mode=live" if args.live else f"mode=fixture scenario={args.fixture} (本地固定响应)")
    try:
        if args.workspace:
            args.workspace.mkdir(parents=True, exist_ok=True)
            result = asyncio.run(run_cli(args, args.workspace))
        else:
            with tempfile.TemporaryDirectory(prefix="skill-agent-") as temporary:
                result = asyncio.run(run_cli(args, Path(temporary)))
        print(json.dumps(result, ensure_ascii=False, sort_keys=True))
        return 0 if result["status"] == "complete" else 1
    except (OSError, ValueError, yaml.YAMLError) as error:
        print(json.dumps({"status": "error", "error": str(error)}, ensure_ascii=False))
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

````


默认 `ScriptedModel` 只重放规定的工具调用，不能替你生成任意主题。`--task`、`--topic` 可以改变请求和主题检查；固定响应仍是那段缓存练习，不因此成为真实生成模型。

若要接入真实模型，先在自己的进程环境配置 `OPENAI_API_KEY` 和 `OPENAI_MODEL`，兼容服务可另设 `OPENAI_BASE_URL`，然后运行：

```sh
python app.py --live
```

凭据缺失会明确报错，不会回退到 fixture。公共 `model_from_env()` 使用 Chat Completions 适配器，所选模型需要支持该 API 和函数调用。应用在 `finally` 中等待 `aclose()`，成功、失败或取消后都负责关闭。网络、权限、费用和模型实际输出另行取决于所选服务；本轮没有 API 凭据，没有进行真实请求或模型质量评测。

## 验证范围与后续使用

本次在 macOS arm64、Python 3.11.16 环境运行 `python -m unittest -v test_app`，27 项测试通过。测试使用标准库 unittest，模型部分为本地固定响应，实际检查文件、YAML、工具 schema、摘要和扫描结果。

重点覆盖正常写后重读、漏写、无效结构、写后文件变化、旧文件冒充、路径越界、符号链接、一次性覆盖批准、精确提案授权及创建后重新匹配。还检查超时和取消，以及 live 适配器的关闭路径；适配器测试使用替代模型，不产生网络请求。

当前规则库只做关键词和名称选择；没有语义检索、规则冲突裁决、自动修订、代码执行或发布功能。真实模型可能漏读规则，也可能提交结构正确但内容错误的文章。把事实资料、人工审阅与自动结构门禁分别安排，才能知道每一步究竟提供了什么证据。

[Research Agent](https://hex4c59.cc/agent/build-research-agent/)可以继续讨论资料收集，[文件管理 Agent](https://hex4c59.cc/agent/build-file-management-agent/)与[Coding Agent](https://hex4c59.cc/agent/build-cli-coding-agent/)讨论其他实际工具，[多 Agent 代码审查](https://hex4c59.cc/agent/build-multi-agent-code-review/)则把审阅职责分开。无论换哪种任务，传入一份规范都不能替代工具权限或产物验证。

*上一篇：[MCP 集成实战：让 Agent 连接真实服务](https://hex4c59.cc/agent/build-mcp-agent/)*

*下一篇：[持久记忆实战：让 Agent 跨会话记住用户](https://hex4c59.cc/agent/build-persistent-memory-agent/)*