用 Markdown 文件定义 LLM 工作流,通过 fork/wait 组合并行、串行、循环等任意拓扑。
entry.md
├─ fork(skill="analyse.md", input="...") ← 非阻塞,立即返回 handle
├─ fork(skill="validate.md", input="...") ← 再派生一个,两者并行运行
│ ↓ ↓
│ (running) (running)
│ └──────────┬──────────┘
└──────── wait([handle_a, handle_b]) ← 阻塞直到两者完成
↓
finish(...)
为什么是 Markdown? 工作流逻辑写在 .md 文件里,可以版本控制、diff、人工审阅,也可以直接给 LLM 看。相比直接写 Python,skill 文件是可读的"任务说明书",改 prompt 不需要改代码。
Bring your own LLM client. 无托管平台,无供应商锁定。
git clone https://github.com/AlexKaiqi/skill-book.git
cd skill-book
pip install -e .
skillbook首次运行自动进入配置向导——选择 LLM 提供商、输入 API Key、选择模型,一次配置永久生效:
skillbook — 首次配置
选择 LLM 提供商:
1. OpenRouter(推荐,支持所有主流模型)
2. OpenAI
3. Anthropic
4. 自定义 OpenAI 兼容端点(内网代理等)
> 1
输入 API Key(https://openrouter.ai/keys):
> sk-or-...
选择默认模型:
1. anthropic/claude-sonnet-4.6(推荐)
2. google/gemma-4-31b-it
3. 自定义
> 1
✓ 配置已保存到 ~/.skillbook/config.json
配置完成后自动进入交互引导:
╭─────────────────────────────────────╮
│ Skillbook · Create / Optimize / Q&A │
╰─────────────────────────────────────╯
我可以帮你:
1. 创建 skillbook — 描述你的需求,我帮你生成全套 skill 文件
2. 优化 skillbook — 指定目录,我检查 P1-P6 原则并给出修改建议
3. 了解 skillbook — 解答概念、用法、最佳实践
也可以直接用子命令:
skillbook create "代码审查:拉取 PR diff → 并行检查风格/安全 → 汇总报告"
skillbook optimize ./my-skillbook
skillbook init # 重新配置环境变量 SKILLBOOK_MODEL + 对应 API Key 仍然兼容,优先级高于配置文件。SKILLBOOK_MODEL 格式为 model-id@provider(如 gpt-4o-mini@openai)。
如果你想自己动手写 skill 文件而非让引导生成:
<!-- skills/analyse.md -->
# 分析师
分析给定问题并给出结论。
---
**Input**
- `input`: 待分析的问题描述
**Output**
- summary: 分析结论(一段话)import asyncio
from skillbook import SkillbookRunner, BranchInput
from skillbook.adapters import OpenAICompatibleClient
runner = SkillbookRunner(
OpenAICompatibleClient.from_env(),
workspace="./runs/001",
)
result = asyncio.run(runner.run(
BranchInput(skill="skills/analyse.md", input="分析 CPU 飙升问题")
))
print(result.summary) # 分析结论
print(result.confidence) # 0.0–1.0也可以传入自己实现了 chat() 的客户端(见 ChatClient 协议)。
---
model: gpt-4o-mini # 可选,覆盖全局 SKILLBOOK_MODEL
---
# 步骤名称
一句话描述(`help()` 展示给父 agent 用于路由决策)。
---
**Input**
- `input`: 任务描述
- `input/data.csv`: 输入数据文件(workspace 相对路径)
**Output**
- summary: 分析结论
- artifacts: `output/report.json` (application/json)
<!-- body -->
具体执行指令...help(name) 默认返回 description + input + output(接口契约);section="detail" 返回 <!-- body --> 之后的全部内容。YAML frontmatter 始终被剥离,LLM 不会看到它。
Runner 初始化时指定一个目录作为 workspace,语义类似进程的 cwd:
- agent 的文件工具(
read_file/write_file/list_files)以此为相对路径基准 - workspace 不是沙箱,是约定的数据交换区
runner = SkillbookRunner(
client,
workspace="./runs/001", # cwd 语义,非 jail;None = 禁用文件工具
)四种执行单元,各自有对应的自动发现方式(详见添加 Skill):
| Kind | 内部 LLM | 执行方式 |
|---|---|---|
| LLMSkill | ✅ | 新建一个 LLM agent,跑 tool-call 循环,驱动 skill 文件 |
| FunctionSkill | ❌ | 同步/异步 Python callable,直接调用 |
| CliCommand | ❌ | 直接执行 shell 命令 |
| ChannelSkill | ❌ | 异步人机交互通道(移动端、Slack、语音等) |
每种 skill 都有三个接口字段:
| 字段 | 说明 |
|---|---|
description |
一句话,search 结果展示、父 agent 路由决策用 |
input |
参数说明(Markdown 文本);父 agent 读取后据此构造 input 字段 |
output |
输出说明:summary 的内容格式 + artifacts 类型 |
对 LLMSkill,三个字段从 skill 文件中读取。对其余 skill,三个字段在装饰器或构造时传入。
ChannelSkill 将任意异步人机交互界面(移动端 chat、Slack、语音输入输出……)包装为 skill。Agent 可以像调用任何其他 skill 一样 fork 它,框架不感知底层通道实现。
核心设计考量是感知延迟:channel 的响应时间由人或下游系统决定,因此实现应为 async,内部可流式消费输入输出,不阻塞 event loop。
# skills/channels.py
from skillbook import registry, BranchInput, BranchResult
from skillbook.skills import ChannelSkill
async def slack_reviewer(inp: BranchInput) -> BranchResult:
message = f"请审核以下结论:\n{inp.input}"
reply = await slack_client.send_and_wait(channel="#oncall", text=message)
approved = "approve" in reply.lower()
return BranchResult(
branch_id=inp.branch_id,
input=inp.input,
summary="approved" if approved else "rejected",
confidence=1.0,
)
registry.register("slack_review", ChannelSkill(
slack_reviewer,
description="通过 Slack 向值班工程师请求人工审批。",
))放入 skills/ 目录后,load_python_skills("skills/") 导入时自动完成注册。多个 channel 并存,agent 根据 description 自主选择。
HumanSkill 是 ChannelSkill 的向后兼容别名,两者完全等价。
将相关 skill 文件组织为目录,entry.md 为入口:
incident-response/ # 顶层目录包
├── entry.md # 入口,fork 下面的子步骤
├── triage.md # 子步骤(单文件 skill)
├── notify.md
└── investigate/ # 子目录包(嵌套 skill)
├── entry.md
├── collect-metrics.md
└── analyse-logs.md
使用目录名即可(自动解析 entry.md):
# 以下两种写法等价
fork(skill="incident-response", input="...")
fork(skill="incident-response/entry.md", input="...")skillbook 支持四种 skill 类型,全部通过自动发现注册——把文件放进目录,重启即生效,无需修改任何注册代码。
用 @registry.skill 装饰器标注函数,放入 skills/ 目录下任意 .py 文件:
# skills/db.py
from skillbook import registry, BranchInput, BranchResult
@registry.skill(description="查询数据库并返回摘要。")
async def db_lookup(inp: BranchInput) -> BranchResult:
...ChannelSkill 将 Slack、移动端、语音等任意异步通道包装为 skill,与其他 skill 接口完全一致。在同一个 .py 文件里用 registry.register 注册即可被自动发现:
# skills/channels.py
from skillbook import registry, BranchInput, BranchResult
from skillbook.skills import ChannelSkill
async def slack_reviewer(inp: BranchInput) -> BranchResult:
...
registry.register("slack_review", ChannelSkill(
slack_reviewer,
description="通过 Slack 向值班工程师请求人工审批。",
))启动时调用一次 load_python_skills(),目录下所有 Python skill 和 ChannelSkill 自动注册:
from skillbook import load_python_skills, SkillbookRunner
load_python_skills("skills/") # 递归扫描 *.py,触发装饰器和 register 调用
runner = SkillbookRunner(client) # 所有发现的 skill 自动可用load_python_skills() 递归扫描目录下所有 *.py(_ 开头的跳过),重复调用是幂等的。
.md 文件无需任何代码注册,fork 时直接用路径即可(见目录包):
fork(skill="skills/analyse.md", input="...")
fork(skill="skills/incident-response", input="...") # 目录包,自动找 entry.md也可以用 registry.register 显式注册别名:
registry.register("analyse", LLMSkill("skills/analyse.md"))用 YAML 文件定义,load_cli_tools() 批量注册:
# cli-tools.yaml
tools:
- name: kubectl_get_pods
description: List pods in a Kubernetes namespace.
command: "kubectl get pods -n {namespace}"
input: 'JSON: `{"namespace": "<str>"}`'
output: "Pod list as plain text; no artifacts."
params:
- name: namespace
required: true
- name: grep_logs
description: Search error logs by keyword.
command: "grep -n '{pattern}' {log_file}"
input: 'JSON: `{"pattern": "<str>", "log_file": "<str>"}`'
output: "Matching lines as plain text; no artifacts."
params:
- name: pattern
required: true
- name: log_file
required: truefrom skillbook import load_cli_tools
load_cli_tools("cli-tools.yaml")
runner = SkillbookRunner(client) # 自动获得所有注册 skillskillbook 自带三个内置 skillbook,覆盖"入门→创建→优化"全流程。
helper 是上面「快速开始」里的交互式引导,把 scaffold 和 optimizer 串联为对话式工作流:
skillbook # 默认就是 helper
skillbook create "代码审查 skillbook" # 单次创建
skillbook optimize ./my-skillbook # 单次优化给一段需求描述,自动生成完整的 skill 文件集合:
from skillbook import SkillbookRunner, BranchInput
from skillbook.builtins import SCAFFOLD_ENTRY
runner = SkillbookRunner(client, max_depth=4, max_turns=30)
result = await runner.run(BranchInput(
skill=SCAFFOLD_ENTRY,
input="做一个代码审查 skillbook:自动拉取 PR diff → 并行检查风格/安全/性能 → 汇总报告",
))
print(result.summary) # 生成的文件列表和验证结果内部流水线:
gather_requirements → design_topology → [generate_entry, generate_sub_skills, generate_functions] → validate_design
生成的 skill 文件遵循六项设计原则(P1-P6),包含 action-first 指令和具体 fork 示例。
扫描现有 skillbook,检查六项设计原则,诊断问题并生成修复建议:
from skillbook import SkillbookRunner, BranchInput
from skillbook.builtins import OPTIMIZER_ENTRY
from skillbook.builtins.optimizer.function_skills import ALL_OPTIMIZER_SKILLS
runner = SkillbookRunner(client, max_depth=4, max_turns=30, skills=ALL_OPTIMIZER_SKILLS)
result = await runner.run(BranchInput(
skill=OPTIMIZER_ENTRY,
input='{"target_path": "./my-skillbook/"}',
))
print(result.summary) # 优化报告:原则检查 + 拓扑分析 + 修复建议也可以命令行运行:
python -m skillbook.builtins.optimizer.run ./my-skillbook/六项设计原则(P1-P6):
| 原则 | 要求 |
|---|---|
| P1 Action-First | 先行动再解释——第一个动作必须是 fork 调用 |
| P2 Concrete Examples | 展示具体的 fork JSON 示例,而非抽象描述 |
| P3 Data Flow | 明确写出每次 fork 的数据来源(如 <scan_result.skills>) |
| P4 Scorer Alignment | 评分器应从 turn_log/fork_groups 提取数据,而非只解析 summary |
| P5 Auto-Redirect | 工具调用应包装为 FunctionSkill fork,而非直接调用 |
| P6 Token Budget | 叶节点 < 500 token,编排器 < 3000 token |
每个 LLM-driven agent 获得五个核心工具:
| 工具 | 作用 |
|---|---|
search(query) |
按关键词搜索已注册的 skill,适合从大型库中发现可用能力 |
help(name, section?) |
读取某个 skill 的接口说明 |
fork(skill, input) |
派生一个子分支;非阻塞,立即返回 handle;多次调用即并行 |
wait(handles) |
阻塞直到指定 handles 完成,返回各分支结果 |
finish(summary, confidence, artifacts?) |
结束当前分支,返回结构化结果 |
workspace 存在时自动注入文件工具:
| 工具 | 作用 |
|---|---|
write_file(path, content) |
写文件到 workspace |
read_file(path) |
从 workspace 读文件 |
list_files(pattern?) |
列出 workspace 文件 |
典型执行模式:
search / help # 发现并了解可用 skill
fork(skill=A, input="...") # 派生子分支(非阻塞)
fork(skill=B, input="...") # 再派生一个,两者并行运行
wait([handle_a, handle_b]) # 等待结果
finish(...) # 返回结论
fork 后不调用 wait = 后台 fire-and-forget 任务。
runner = SkillbookRunner(
client, # 实现 chat() 的 LLM 客户端
workspace="./runs/001", # cwd 语义,非 jail;None = 禁用文件工具
max_depth=3, # fork 嵌套深度上限
max_turns=20, # 每分支 LLM 调用次数上限
run_dir="runs/001", # 可选:开启 checkpoint 断点续跑
skills={...}, # 注册 skill dict,优先级高于全局注册表
)
result = await runner.run(inp) # 传入 BranchInput,返回 BranchResult后台任务管理:
runner.background_tasks # frozenset,当前运行中的后台任务
await runner.join_background() # 等待所有后台任务完成
runner.cancel_background() # 取消所有后台任务@dataclass
class BranchInput:
skill: str # skill 文件路径或注册名
input: str # 任务描述,可包含 workspace 文件路径引用
branch_id: str # 分支标识,如 "root.0.1"(默认自动生成)
depth: int # 当前嵌套深度(默认 0)@dataclass
class BranchResult:
branch_id: str
input: str
summary: str # 结论摘要
confidence: float # 0.0–1.0
artifacts: list[dict] # 输出文件声明,每项 {name, path, type}
fork_groups: list[ForkGroup] # 并行 fork 结构
turn_log: list[TurnEntry] # 每个 LLM turn 一条
trace: ExecutionTrace # timing 和 token 统计实现 chat() 方法即可接入任意 LLM:
from skillbook.protocols import LLMResponse, ToolCallResult
class MyClient:
def chat(
self,
messages: list[dict],
*,
tools: list[dict] | None = None,
model: str | None = None,
**kwargs,
) -> LLMResponse:
return LLMResponse(
content=response_text,
tool_calls=[
ToolCallResult(id=tc.id, function_name=tc.name, arguments=tc.args)
for tc in response.tool_calls or []
],
prompt_tokens=usage.input_tokens,
completion_tokens=usage.output_tokens,
)- Skill 文件 frontmatter
model: gpt-4o-mini— 仅该分支生效 SKILLBOOK_MODEL环境变量 — 格式model-id@provider,全局默认
runner = SkillbookRunner(client, run_dir="runs/001")目录结构:
{run_dir}/
├── _meta.json # status: running → completed
├── _result.json # 序列化的 BranchResult
└── 0/ # 第一个 fork 子分支(递归同结构)
├── _meta.json
└── _result.json
进入 run() 时若 _result.json 存在且有效,直接返回缓存结果。
每个 BranchResult 携带完整执行记录:
result.turn_log # list[TurnEntry] — 每个 LLM turn 一条
result.trace # ExecutionTrace — timing 和 token 统计
result.fork_groups # list[ForkGroup] — 并行 fork 结构TurnEntry 字段:turn_idx、tool_name、tool_args_summary、assistant_snippet、prompt_tokens、completion_tokens。
Skill 文件即任务图。 每个 .md 文件是一个 skill。通过 fork/wait 组合出任意拓扑:扇出并行、串行链、迭代循环、有向无环图。
显式 fork/wait,而非隐式阻塞。 fork 非阻塞,立即返回 handle;wait(handles) 显式等待指定分支。这与 Unix 进程语义一致。每个并行分支一次 fork 调用,语义清晰。
Skill,而非 Monolith。 并非每个分支都需要 LLM。Python 函数、Shell 命令、人工审核、移动端通知都可以注册为具名 Skill,agent 通过 description 自主发现和路由。
Channel 是 Skill 的一种。 ChannelSkill 将任意人机交互通道(Slack、移动端、语音……)统一为 skill 接口。多个 channel 并存,agent 自主选择,系统不感知底层实现。延迟由通道特性决定,实现使用 async 以避免阻塞。
文件是唯一 IPC 通道。 Skill 之间的数据传递走 workspace 文件。finish 的 summary 控制 context 膨胀,数据本体以 artifacts 形式存文件,父 agent 按需读取。
Markdown 是事实来源。 Skill 文件可被版本控制、diff、人工阅读。YAML frontmatter 在 LLM 看到内容之前会被剥离。
src/skillbook/ library source
src/skillbook/builtins/ 内置 skillbook(scaffold / optimizer / helper)
examples/quickstart/ 五分钟上手示例
examples/personal_assistant/ 完整工作流示例(FunctionSkill + ChannelSkill + workspace)
docs/design/ 架构和设计文档
tests/ pytest 测试(mock LLM,无需 API key)
开发安装:
python3.11 -m venv .venv
.venv/bin/pip install -e ".[dev]"运行测试:
.venv/bin/python -m pytest tests/ -v