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

---
name: write-blog-post
description: >-
  Write blog posts for the Hugo + PaperMod Agent engineering series.
  Use when the user asks to write, draft, or create a blog article.
triggers:
  - 写文章
  - 写博客
  - 写一篇
  - draft
  - blog post
  - 文章
---

# 博客文章写作规范

## Front Matter 格式（必须严格遵循）

使用 TOML 格式（+++ 包裹），不要用 YAML（---）：

\```toml
+++
title = "文章标题"
date = 2026-03-28T10:00:00+08:00
draft = false
author = "Hex4C59"
description = "一句话描述文章核心内容"
summary = "2-3 句话的摘要"
tags = ["Agent", "相关标签"]
categories = ["Agent"]
series = ["agent-engineering"]
series_order = 下一个序号
difficulty = "beginner | intermediate | advanced"
article_type = "concept | tutorial"
ShowToc = true
+++
\```

注意事项：
- 时区必须是 +08:00
- series_order 递增，查看现有文章确定下一个数字
- article_type 只有两种：概念文章用 concept，实战教程用 tutorial

## 文章结构（必须包含以下章节）

1. **开头引用**：引用系列中之前的文章作为上下文铺垫
2. **先给结论**：用编号列表给出 5 条核心结论，每条 1-2 句话
3. **正文章节**：由浅入深展开，每节有清晰的小标题
4. **代码示例**：使用 Python + OpenAI SDK，代码必须可独立运行
5. **总结**：回顾核心要点，预告下一篇

## 写作风格

- 语气直接，避免学术化表达，像在和同事对话
- 多用类比解释抽象概念（如"MCP 之于 Agent 就像 USB 之于外设"）
- 每个代码块后面跟一段解释，说明"为什么这样设计"
- 用表格对比不同方案的优劣
- 图文并茂：每篇至少 2 张 SVG 配图

## 代码风格

- 使用 async/await 风格
- 函数和类有中文注释说明设计意图
- 关键变量用有意义的名称，不用 a b c
- 代码示例保持模块化，一个代码块只做一件事

## 不要做的事情

- 不要用 YAML front matter
- 不要用 JavaScript 代码示例（除非主题要求）
- 不要在结论之前放长篇背景介绍
- 不要使用"本文将介绍..."这种空洞的开场
- 不要漏掉系列文章的交叉引用

## 质量检查清单

- [ ] front matter 使用 TOML 格式
- [ ] 包含"先给结论"章节
- [ ] 结论是 5 条编号列表
- [ ] 代码使用 Python + OpenAI SDK
- [ ] 有至少 2 处对其他文章的引用
- [ ] 有至少 1 张 SVG 配图的说明
- [ ] 有"总结"章节
- [ ] 时区是 +08:00

---
name: create-svg-diagram
description: >-
  Create SVG diagrams for Agent engineering blog posts.
  Use when the user asks to create a diagram, chart, or illustration.
triggers:
  - SVG
  - 画图
  - 配图
  - 图表
  - diagram
---

# SVG 配图规范

## 全局字体

\```
font-family="Inter, -apple-system, BlinkMacSystemFont, system-ui, sans-serif"
\```

## 配色方案

| 用途 | 背景 | 描边 | 文字 |
|------|------|------|------|
| Agent/推理 | rgb(238, 237, 254) | rgb(83, 74, 183) | rgb(60, 52, 137) |
| 工具/外部 | rgb(250, 238, 218) | rgb(133, 79, 11) | rgb(99, 56, 6) |
| 成功/输出 | rgb(225, 245, 238) | rgb(15, 110, 86) | rgb(8, 80, 65) |
| 错误/警告 | rgb(252, 230, 227) | rgb(201, 77, 61) | rgb(140, 41, 29) |
| 中性/容器 | rgb(241, 239, 232) | rgb(95, 94, 90) | rgb(68, 68, 65) |

## SVG 模板

\```xml
<svg width="100%" viewBox="0 0 680 400"
     xmlns="http://www.w3.org/2000/svg"
     role="img" aria-labelledby="title desc">
  <title id="title">图表标题</title>
  <desc id="desc">图表描述</desc>
  <!-- 内容 -->
</svg>
\```

## 规则

- viewBox 宽度固定 680
- 圆角矩形 rx="10"（容器）、rx="6"（内部元素）
- 箭头使用 marker-end 定义
- 标题文字 14px font-weight 500
- 正文文字 12px
- 小号文字 11px

# agent/skill_loader.py
import os
import re
from pathlib import Path
from dataclasses import dataclass, field


@dataclass
class Skill:
    """一个 Skill 的完整信息。"""
    name: str
    description: str
    triggers: list[str]
    content: str            # Markdown 正文（不含 front matter）
    file_path: str


class SkillLoader:
    """
    Skill 发现、匹配和加载。
    支持三种匹配方式：手动指定、关键词匹配、全部加载。
    """

    def __init__(self, skills_dir: str):
        self.skills_dir = Path(skills_dir)
        self.skills: list[Skill] = []
        self._load_all_skills()

    def _load_all_skills(self):
        """扫描 skills 目录，解析所有 Skill 文件。"""
        if not self.skills_dir.is_dir():
            print(f"⚠️  Skill 目录不存在：{self.skills_dir}")
            return

        for path in sorted(self.skills_dir.glob("*.md")):
            try:
                skill = self._parse_skill_file(path)
                self.skills.append(skill)
            except Exception as e:
                print(f"⚠️  解析 Skill 失败 {path.name}: {e}")

        print(f"📚 加载了 {len(self.skills)} 个 Skill:")
        for s in self.skills:
            print(f"   - {s.name}: {s.description[:60]}")

    def _parse_skill_file(self, path: Path) -> Skill:
        """解析一个 Skill 文件——YAML front matter + Markdown 正文。"""
        text = path.read_text(encoding="utf-8")

        # 提取 YAML front matter
        fm_match = re.match(r'^---\s*\n(.*?)\n---\s*\n', text, re.DOTALL)
        if not fm_match:
            raise ValueError("缺少 YAML front matter")

        fm_text = fm_match.group(1)
        content = text[fm_match.end():]

        # 简易 YAML 解析（避免引入 pyyaml 依赖）
        name = self._extract_field(fm_text, "name") or path.stem
        description = self._extract_field(fm_text, "description") or ""
        triggers = self._extract_list(fm_text, "triggers")

        return Skill(
            name=name,
            description=description,
            triggers=triggers,
            content=content.strip(),
            file_path=str(path),
        )

    def get_by_name(self, name: str) -> Skill | None:
        """按名称精确获取 Skill。"""
        for s in self.skills:
            if s.name == name:
                return s
        return None

    def match_by_task(self, task: str) -> list[Skill]:
        """
        根据用户任务描述自动匹配 Skill。
        匹配策略：检查任务文本中是否包含 Skill 的 triggers 关键词。
        """
        matched = []
        task_lower = task.lower()

        for skill in self.skills:
            for trigger in skill.triggers:
                if trigger.lower() in task_lower:
                    matched.append(skill)
                    break  # 一个 trigger 命中就够了

        return matched

    def build_skill_context(self, skills: list[Skill]) -> str:
        """
        把多个 Skill 的内容合并成一个上下文块，
        准备注入到 System Prompt 中。
        """
        if not skills:
            return ""

        parts = [
            "# 领域知识（以下规范来自 Skill 文件，请严格遵循）\n"
        ]

        for skill in skills:
            parts.append(f"## Skill: {skill.name}\n")
            parts.append(skill.content)
            parts.append("\n---\n")

        return "\n".join(parts)

    def _extract_field(self, text: str, field: str) -> str | None:
        """从 YAML 文本中提取单值字段。"""
        # 处理多行值（>- 语法）
        pattern = rf'^{field}:\s*>-?\s*\n((?:\s+.*\n)*)'
        match = re.search(pattern, text, re.MULTILINE)
        if match:
            lines = match.group(1).strip().splitlines()
            return " ".join(line.strip() for line in lines)
        # 处理单行值
        match = re.search(rf'^{field}:\s*(.+)$', text, re.MULTILINE)
        if match:
            return match.group(1).strip().strip('"').strip("'")
        return None

    def _extract_list(self, text: str, field: str) -> list[str]:
        """从 YAML 文本中提取列表字段。"""
        pattern = rf'^{field}:\s*\n((?:\s+-\s+.*\n)*)'
        match = re.search(pattern, text, re.MULTILINE)
        if not match:
            return []
        items = []
        for line in match.group(1).strip().splitlines():
            item = line.strip().lstrip("- ").strip()
            if item:
                items.append(item)
        return items

# agent/core.py
import json
import openai
from agent.skill_loader import SkillLoader
from agent.prompts import BASE_SYSTEM_PROMPT

client = openai.AsyncOpenAI()
MODEL = "gpt-4o"
MAX_ITERATIONS = 15

TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取文件内容。用于查看现有文章的结构和风格作为参考。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"}
                },
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "创建或覆盖文件。用于写入生成的文章内容。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"},
                    "content": {"type": "string", "description": "文件内容"},
                },
                "required": ["path", "content"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "list_directory",
            "description": "列出目录内容。用于了解现有文章的目录结构。",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "目录路径"}
                },
                "required": ["path"],
            },
        },
    },
]


def _read_file(path: str) -> dict:
    try:
        with open(path, "r", encoding="utf-8") as f:
            return {"success": True, "content": f.read()}
    except Exception as e:
        return {"success": False, "error": str(e)}


def _write_file(path: str, content: str) -> dict:
    try:
        from pathlib import Path
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        return {"success": True, "path": path, "bytes": len(content.encode())}
    except Exception as e:
        return {"success": False, "error": str(e)}


def _list_directory(path: str) -> dict:
    try:
        from pathlib import Path
        p = Path(path)
        items = [
            {"name": item.name, "type": "dir" if item.is_dir() else "file"}
            for item in sorted(p.iterdir())
            if not item.name.startswith(".")
        ]
        return {"success": True, "items": items}
    except Exception as e:
        return {"success": False, "error": str(e)}


TOOL_REGISTRY = {
    "read_file": _read_file,
    "write_file": _write_file,
    "list_directory": _list_directory,
}


async def run_agent(
    user_message: str,
    skill_loader: SkillLoader,
    conversation_history: list[dict],
    skill_names: list[str] | None = None,
):
    """
    运行 Skill 驱动的 Agent。

    Skill 的加载有两种模式：
    1. 手动指定：传入 skill_names
    2. 自动匹配：根据 user_message 自动匹配 triggers
    """

    # ===== Skill 加载 =====
    if skill_names:
        # 手动模式：按名称加载
        skills = [
            s for name in skill_names
            if (s := skill_loader.get_by_name(name))
        ]
    else:
        # 自动模式：关键词匹配
        skills = skill_loader.match_by_task(user_message)

    if skills:
        print(f"📎 加载 Skill: {[s.name for s in skills]}")
    else:
        print("📎 未匹配到 Skill，使用通用模式")

    # ===== 组装上下文 =====
    skill_context = skill_loader.build_skill_context(skills)

    # Skill 注入位置：System Prompt 之后，作为独立的 system 消息
    system_messages = [
        {"role": "system", "content": BASE_SYSTEM_PROMPT},
    ]
    if skill_context:
        system_messages.append(
            {"role": "system", "content": skill_context}
        )

    conversation_history.append({"role": "user", "content": user_message})

    # ===== ReAct 循环 =====
    for iteration in range(MAX_ITERATIONS):
        print(f"\n[第 {iteration + 1} 轮]")

        messages = system_messages + conversation_history[-20:]

        response = await client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=TOOL_SCHEMAS,
        )

        choice = response.choices[0]
        message = choice.message

        assistant_msg = {"role": "assistant", "content": message.content}
        if message.tool_calls:
            assistant_msg["tool_calls"] = [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {
                        "name": tc.function.name,
                        "arguments": tc.function.arguments,
                    },
                }
                for tc in message.tool_calls
            ]
        conversation_history.append(assistant_msg)

        if message.content:
            print(f"\n{message.content[:300]}")

        if choice.finish_reason == "stop":
            # ===== Skill 质量检查 =====
            if skills:
                await _skill_quality_check(
                    message.content, skills, conversation_history
                )
            print("\n[✓ 任务完成]")
            break

        if choice.finish_reason == "tool_calls":
            for tc in message.tool_calls:
                tool_name = tc.function.name
                tool_args = json.loads(tc.function.arguments)

                print(f"  [工具] {tool_name}({list(tool_args.keys())})")

                result = TOOL_REGISTRY[tool_name](**tool_args)
                result_str = json.dumps(result, ensure_ascii=False)
                if len(result_str) > 3000:
                    result_str = result_str[:3000] + "\n[截断]"

                conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result_str,
                })


async def _skill_quality_check(
    output: str,
    skills: list,
    conversation_history: list[dict],
):
    """
    用 Skill 中的 Checklist 验证 Agent 的输出质量。
    这是 Reflection 的一种落地形式——用 Skill 标准替代泛用的自我检查。
    """
    checklists = []
    for skill in skills:
        if "质量检查清单" in skill.content or "checklist" in skill.content.lower():
            checklists.append(f"[{skill.name}] 的检查项")

    if not checklists:
        return

    check_response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "system",
            "content": (
                "你是一个质量审查员。根据以下 Skill 规范中的检查清单，"
                "验证 Agent 的输出是否符合要求。"
                "对每个检查项标注 ✓ 或 ✗，并简要说明原因。"
            ),
        }, {
            "role": "user",
            "content": f"Skill 规范：\n{[s.content for s in skills]}\n\nAgent 输出：\n{output[:3000]}",
        }],
    )

    print(f"\n📋 质量检查结果：")
    print(check_response.choices[0].message.content)

# agent/prompts.py

BASE_SYSTEM_PROMPT = """你是一个写作助手，帮助用户创建高质量的技术博客文章。

## 工作流程
1. 先查看现有文章的目录结构和风格（用 list_directory 和 read_file）
2. 参考现有文章的结构和规范
3. 按照规范撰写新文章的完整内容
4. 用 write_file 工具写入文件

## 重要原则
- 如果加载了 Skill 规范，**必须严格遵循其中的每一条规则**
- Skill 规范优先于你的默认习惯——如果 Skill 说用 TOML，哪怕你习惯 YAML，也用 TOML
- 写作前先阅读至少 2 篇现有文章，理解风格后再动笔

## 你不需要做的事
- 不需要自己决定文章风格——Skill 会告诉你
- 不需要猜测 front matter 格式——Skill 会给出模板
"""

📚 加载了 3 个 Skill:
   - write-blog-post: Write blog posts for the Hugo + PaperMod Agent...
   - create-svg-diagram: Create SVG diagrams for Agent engineering...
   - review-code: Review code changes following team standards...

> 帮我写一篇关于 Prompt Caching 的博客文章

📎 加载 Skill: ['write-blog-post']

[第 1 轮]
好的，我先看看现有文章的结构。
  [工具] list_directory(['content/agent'])
  [工具] read_file(['content/agent/prompt-design-agent/index.md'])

[第 2 轮]
好的，我已经了解了文章结构。根据 Skill 规范：
- 使用 TOML front matter
- series_order = 23（当前最大是 22）
- 必须包含"先给结论"章节

让我再读一篇最近的文章，确认代码风格。
  [工具] read_file(['content/agent/build-mcp-agent/index.md'])

[第 3 轮]
已经理解了代码风格。现在开始撰写文章。
  [工具] write_file({
    path: 'content/agent/prompt-caching/index.md',
    content: '+++\ntitle = "Prompt Caching：...'
  })

[第 4 轮]
文章已写入 content/agent/prompt-caching/index.md。

📋 质量检查结果：
  ✓ front matter 使用 TOML 格式
  ✓ 包含"先给结论"章节
  ✓ 结论是 5 条编号列表
  ✓ 代码使用 Python + OpenAI SDK
  ✓ 有 3 处对其他文章的引用
  ✗ 缺少 SVG 配图说明（文中提到了图但未标注图号）
  ✓ 有"总结"章节
  ✓ 时区是 +08:00

1 项未通过——建议补充 SVG 图表的图号标注。

[✓ 任务完成]

# 用户说：帮我写一篇文章并配图
matched_skills = skill_loader.match_by_task("帮我写一篇关于 X 的文章，画几张图")
# 命中：['write-blog-post', 'create-svg-diagram']

CREATE_SKILL_SCHEMA = {
    "type": "function",
    "function": {
        "name": "create_skill",
        "description": (
            "创建一个新的 Skill 文件。当你在任务中发现了用户"
            "的特定规范或偏好（用户纠正了你的输出、告诉了你某个约定），"
            "把它记录为 Skill 供以后使用。"
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "name": {"type": "string", "description": "Skill 名称"},
                "description": {"type": "string", "description": "适用场景"},
                "triggers": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "触发关键词",
                },
                "content": {"type": "string", "description": "Skill 正文"},
            },
            "required": ["name", "description", "triggers", "content"],
        },
    },
}