Codex 技能系统：如何让 Agent 学会新能力

┌──────────────────────────────────┐
│          Codex Agent             │
│                                  │
│  ┌────────────────────────────┐  │
│  │      技能管理器             │  │
│  │  ┌─────┐ ┌─────┐ ┌─────┐  │  │
│  │  │ 技能│ │ 技能│ │ 技能│  │  │
│  │  │  A  │ │  B  │ │  C  │  │  │
│  │  └─────┘ └─────┘ └─────┘  │  │
│  └────────────────────────────┘  │
│           ↓ 按需加载              │
│  ┌────────────────────────────┐  │
│  │      Agent Loop             │  │
│  └────────────────────────────┘  │
└──────────────────────────────────┘

my-skill/
├── SKILL.md           # 技能定义文件（必需）
├── tools/             # 工具脚本
│   ├── clean.py
│   └── validate.py
├── templates/         # 模板文件
│   └── report.html
├── examples/          # 示例
│   └── example_usage.md
└── README.md          # 人类可读文档

---
name: data-pipeline
version: 1.2.0
description: ETL 数据管道，支持 CSV/JSON/Excel 读取、清洗、转换、输出
author: community
triggers:
  - "清洗数据"
  - "数据管道"
  - "ETL"
  - "data pipeline"
---

## 能力

### 读取
- CSV (支持大文件流式读取)
- JSON (支持嵌套和数组)
- Excel (.xlsx/.xls)
- Parquet

### 清洗
- 缺失值处理 (填充/删除/插值)
- 异常值检测 (IQR/Z-score/自定义)
- 数据类型转换
- 去重

### 转换
- 列映射重命名
- 透视/逆透视
- 分组聚合
- 时序重采样

### 输出
- CSV/JSON/Excel
- 数据库写入 (PostgreSQL/MySQL/SQLite)
- API 推送

## 执行流程

1. 读取输入文件，推断 schema
2. 执行清洗规则
3. 执行转换规则
4. 验证输出数据质量
5. 写入目标

## 工具

### clean_csv
- 用途: 清洗 CSV 文件
- 参数: input_path, output_path, rules
- 示例: `python tools/clean.py input.csv output.csv --fill-mean --drop-duplicates`

### validate
- 用途: 验证数据质量
- 参数: file_path, schema
- 输出: 验证报告 (通过率、异常数、建议)

# 伪代码：技能加载
class SkillManager:
    def __init__(self, skills_dir: str):
        self.skills = {}
        self.load_skills(skills_dir)
    
    def load_skills(self, dir: str):
        for skill_dir in glob(f"{dir}/*/SKILL.md"):
            skill = parse_skill_md(skill_dir)
            self.skills[skill.name] = skill
    
    def match_skill(self, user_input: str) -> Skill | None:
        for skill in self.skills.values():
            for trigger in skill.triggers:
                if trigger in user_input.lower():
                    return skill
        return None
    
    def hot_reload(self, skill_name: str):
        """热加载单个技能，无需重启"""
        skill_dir = f"{self.skills_dir}/{skill_name}"
        if exists(skill_dir):
            self.skills[skill_name] = parse_skill_md(skill_dir)

# 搜索技能
codex skill search "数据清洗"
# 输出:
# data-pipeline      v1.2.0   ETL 数据管道         ⭐ 1.2k
# csv-cleaner        v0.8.3   CSV 专用清洗工具     ⭐ 856
# excel-processor    v2.1.0   Excel 高级处理       ⭐ 2.1k

# 安装技能
codex skill install data-pipeline

# 安装特定版本
codex skill install data-pipeline@1.2.0

# 列出已安装技能
codex skill list
# data-pipeline      v1.2.0   [active]
# web-scraper        v3.0.1   [active]
# report-generator   v1.5.2   [active]

# 更新技能
codex skill update data-pipeline

# 卸载技能
codex skill uninstall data-pipeline

# 使用 CLI 创建
codex skill create my-data-tool
# 生成结构:
# my-data-tool/
# ├── SKILL.md
# ├── tools/
# │   └── main.py
# ├── templates/
# └── README.md

---
name: report-generator
version: 1.0.0
description: 从数据源生成格式化报告，支持 Markdown/HTML/PDF
author: your-name
triggers:
  - "生成报告"
  - "create report"
  - "数据报告"
  - "月报"
  - "周报"
---

## 能力

从多种数据源（CSV/JSON/数据库/API）读取数据，
执行分析，生成格式化报告。

支持输出格式：
- Markdown (默认)
- HTML (含 CSS 样式)
- PDF (需安装 wkhtmltopdf)
- Excel (含图表)

## 执行流程

1. 解析用户输入，确定数据源和报告类型
2. 读取数据源
3. 执行分析（汇总/趋势/对比/异常检测）
4. 选择模板
5. 生成报告
6. 保存到指定路径

## 工具

### analyze_data
- 用途: 分析数据并返回统计结果
- 参数: data_path, analysis_type
- analysis_type: summary | trend | comparison | anomaly

### render_report
- 用途: 将分析结果渲染为报告
- 参数: analysis_result, template, output_format
- 模板: templates/report_md.html, templates/report_pdf.html

## 示例

用户: "用 sales.csv 生成月度销售报告，输出 PDF"

执行:
1. analyze_data("sales.csv", "summary") → 统计结果
2. analyze_data("sales.csv", "trend") → 趋势分析
3. render_report(stats, "report_pdf", "pdf") → report.pdf

# tools/analyze.py
import pandas as pd
import numpy as np
from typing import Literal

def analyze_data(
    data_path: str, 
    analysis_type: Literal["summary", "trend", "comparison", "anomaly"]
) -> dict:
    """分析数据并返回统计结果"""
    df = pd.read_csv(data_path)
    
    if analysis_type == "summary":
        return {
            "total_rows": len(df),
            "columns": df.columns.tolist(),
            "dtypes": df.dtypes.astype(str).to_dict(),
            "missing": df.isnull().sum().to_dict(),
            "stats": df.describe().to_dict()
        }
    
    elif analysis_type == "trend":
        if "date" not in df.columns:
            raise ValueError("趋势分析需要 date 列")
        df["date"] = pd.to_datetime(df["date"])
        df = df.set_index("date").sort_index()
        return {
            "daily_mean": df.resample("D").mean().to_dict(),
            "weekly_mean": df.resample("W").mean().to_dict(),
            "monthly_mean": df.resample("M").mean().to_dict()
        }
    
    elif analysis_type == "anomaly":
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        anomalies = {}
        for col in numeric_cols:
            q1, q3 = df[col].quantile([0.25, 0.75])
            iqr = q3 - q1
            lower = q1 - 1.5 * iqr
            upper = q3 + 1.5 * iqr
            anomalies[col] = {
                "count": ((df[col] < lower) | (df[col] > upper)).sum(),
                "lower_bound": lower,
                "upper_bound": upper
            }
        return anomalies

# 1. 登录 ClawHub
codex skill login

# 2. 验证技能格式
codex skill validate my-report-generator/
# ✓ SKILL.md format valid
# ✓ Tools directory found
# ✓ Templates directory found
# ✓ All triggers are unique

# 3. 发布
codex skill publish my-report-generator/
# Publishing report-generator@1.0.0...
# ✓ Uploaded
# ✓ Indexed
# Live at https://clawhub.com/skills/report-generator

# 4. 版本更新
codex skill version bump --patch  # 1.0.0 → 1.0.1
codex skill publish my-report-generator/

# 安装技能后自动热加载
codex skill install excel-processor
# ✓ Installed excel-processor@2.1.0
# ✓ Hot-loaded into current session

# 手动重新加载
codex skill reload excel-processor

# 开发模式：文件变化自动重载
codex skill dev my-skill/
# Watching my-skill/ for changes...
# [10:00:05] SKILL.md changed → reloaded
# [10:00:12] tools/main.py changed → reloaded

# 伪代码
def hot_reload_skill(skill_name: str):
    # 1. 卸载旧版本
    old_skill = skill_manager.skills.pop(skill_name, None)
    if old_skill:
        cleanup_skill_resources(old_skill)
    
    # 2. 重新加载
    new_skill = parse_skill_md(f"~/.codex/skills/{skill_name}/SKILL.md")
    skill_manager.skills[skill_name] = new_skill
    
    # 3. 通知 Agent Loop
    agent_loop.notify_skill_changed(skill_name)
    
    # 4. 不影响正在执行的任务
    # 新技能在下一个任务循环中生效

// ~/.codex/skills/registry.json
{
  "skills": {
    "data-pipeline": {
      "installed_version": "1.2.0",
      "latest_version": "1.3.0",
      "pinned": false,
      "auto_update": true
    },
    "web-scraper": {
      "installed_version": "3.0.1",
      "latest_version": "3.0.1",
      "pinned": true,
      "auto_update": false
    }
  },
  "dependencies": {
    "data-pipeline": ["pandas>=2.0", "openpyxl>=3.1"],
    "web-scraper": ["playwright>=1.40"]
  }
}

# 锁定版本
codex skill pin web-scraper

# 检查更新
codex skill outdated
# data-pipeline  1.2.0 → 1.3.0  (minor)

# 更新所有未锁定技能
codex skill update --all

# 回滚到上一版本
codex skill rollback data-pipeline

# 用户请求: "抓取竞品网站数据，清洗后生成对比分析报告"

Codex 技能组合执行:
1. [web-scraper] 抓取 3 个竞品网站的产品数据
2. [data-pipeline] 清洗和标准化数据格式
3. [excel-processor] 生成对比分析 Excel
4. [report-generator] 生成 Markdown 分析报告
5. [email-skill] 发送报告到指定邮箱

# 伪代码：技能链式调用
def execute_skill_chain(task: str):
    skills = plan_skills(task)
    # skills = [web-scraper, data-pipeline, report-generator]
    
    context = {"input": task}
    for skill in skills:
        result = skill.execute(context)
        context["input"] = result.output  # 传递给下一个技能
    
    return context["input"]

# 安装前审查技能内容
codex skill audit web-scraper
# Checking web-scraper@3.0.1...
# ⚠ WARNING: requests network access
# ⚠ WARNING: writes to filesystem
# ✓ No shell execution
# ✓ No credential access
# Trust level: MEDIUM

## 性能提示 (写入 SKILL.md)

- 大文件处理使用流式读取，避免全量加载
- 网络请求添加超时和重试
- 耗时操作提供进度反馈
- 缓存中间结果避免重复计算