Skip to content

Latest commit

 

History

57 Commits

Folders and files

NameName
Last commit message
Last commit date
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

skillbook

用 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

如果你想自己动手写 skill 文件而非让引导生成:

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 协议)。


核心概念

Skill 文件格式

---
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 不会看到它。

Workspace — 工作目录

Runner 初始化时指定一个目录作为 workspace,语义类似进程的 cwd

  • agent 的文件工具(read_file / write_file / list_files)以此为相对路径基准
  • workspace 不是沙箱,是约定的数据交换区
runner = SkillbookRunner(
    client,
    workspace="./runs/001",   # cwd 语义,非 jail;None = 禁用文件工具
)

Skill — 执行单元

四种执行单元,各自有对应的自动发现方式(详见添加 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 — 人机交互通道

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 自主选择。

HumanSkillChannelSkill 的向后兼容别名,两者完全等价。

目录包(Directory Package)

将相关 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="...")

添加 Skill

skillbook 支持四种 skill 类型,全部通过自动发现注册——把文件放进目录,重启即生效,无需修改任何注册代码。

Python 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(人机交互通道)

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_ 开头的跳过),重复调用是幂等的。

Markdown Skill(LLM 驱动)

.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"))

CliCommand(Shell 命令)

用 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: true
from skillbook import load_cli_tools

load_cli_tools("cli-tools.yaml")

runner = SkillbookRunner(client)  # 自动获得所有注册 skill

内置 Skillbook

skillbook 自带三个内置 skillbook,覆盖"入门→创建→优化"全流程。

helper — 交互式引导(推荐入口)

helper 是上面「快速开始」里的交互式引导,把 scaffold 和 optimizer 串联为对话式工作流:

skillbook                            # 默认就是 helper
skillbook create "代码审查 skillbook"  # 单次创建
skillbook optimize ./my-skillbook     # 单次优化

scaffold — 脚手架:从自然语言生成 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 示例。

optimizer — 优化器:分析并改进现有 skillbook

扫描现有 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

Skill 文件中的工具

每个 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 任务。


API 参考

SkillbookRunner

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()      # 取消所有后台任务

BranchInput

@dataclass
class BranchInput:
    skill: str       # skill 文件路径或注册名
    input: str       # 任务描述,可包含 workspace 文件路径引用
    branch_id: str   # 分支标识,如 "root.0.1"(默认自动生成)
    depth: int       # 当前嵌套深度(默认 0)

BranchResult

@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 统计

ChatClient 协议

实现 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,
        )

模型选择

  1. Skill 文件 frontmatter model: gpt-4o-mini — 仅该分支生效
  2. SKILLBOOK_MODEL 环境变量 — 格式 model-id@provider,全局默认

Checkpoint(断点续跑)

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_idxtool_nametool_args_summaryassistant_snippetprompt_tokenscompletion_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 文件。finishsummary 控制 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

About

No description, website, or topics provided.

Resources

Stars

0 stars

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages