figmar/wikiskill
1
0
Fork 0
mirror of https://github.com/kenhuangus/wikiskill synced 2026-09-22 23:34:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

WikiSkill: reference implementation of arXiv:2608.27454 (Google Research) - three-layer workspace, evolutionary agents, validation gating

Browse Source
This commit is contained in:
DistributedApps.AI
2026-08-30 23:07:08 -04:00
commit 0f2df89097
22 changed files with 1596 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+10
View File
@@ -0,0 +1,10 @@
__pycache__/
*.pyc
.pytest_cache/
*.egg-info/
build/
dist/
wikiskill_demo_workspace/
wikiskill_workspace/
demo_ws/
.vscode/
LICENSE
+29
View File
@@ -0,0 +1,29 @@
MIT License
Copyright (c) 2026 Repository contributors
Code in this repository is released under the MIT License.
The WikiSkill method and prompts reimplemented here originate from the paper
"WikiSkill: Compiling Agent Experience into Persistent Knowledge for Skill Evolution"
by Tang et al. (Google Research), arXiv:2608.27454, published under CC BY 4.0.
The paper's prompts (adapted in src/wikiskill/prompts.py) remain subject to CC BY 4.0;
attribute the original authors when reusing this work.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
README.md
+99
View File
@@ -0,0 +1,99 @@
# WikiSkill — Implementation
Python implementation of **"WikiSkill: Compiling Agent Experience into Persistent
Knowledge for Skill Evolution"** (arXiv:2608.27454v1, Tang et al., Google Research).
Docs: see [`prd.md`](prd.md) (requirements), [`design.md`](design.md) (architecture),
[`plan.md`](plan.md) (build plan).
## Credits & Attribution
This project is an independent, open-source reimplementation of the research paper:
> **WikiSkill: Compiling Agent Experience into Persistent Knowledge for Skill Evolution**
> Liyan Tang, Cyrus Rashtchian, Chun-Sung Ferng, Andrew Tomkins, Da-Cheng Juan, Tu Vu —
> **Google Research** (and Virginia Tech).
> arXiv:2608.27454v1 [cs.AI], 27 Aug 2026. License: CC BY 4.0.
> Paper: https://arxiv.org/abs/2608.27454
All credit for the WikiSkill **method, algorithm (Algorithm 1), three-layer knowledge
architecture, agent designs, experimental findings, and agent prompts (Appendix E)**
belongs to the Google Research authors. This repository reimplements the framework in
code for study/experimentation purposes and is **not** an official Google product.
Agent system prompts in [`src/wikiskill/prompts.py`](src/wikiskill/prompts.py) are
condensed adaptations of those published in the paper (CC BY 4.0); adaptations in this
repo are released under the same spirit — please cite the original paper if you use
this work:
```bibtex
@article{tang2026wikiskill,
title = {WikiSkill: Compiling Agent Experience into Persistent Knowledge
for Skill Evolution},
author = {Tang, Liyan and Rashtchian, Cyrus and Ferng, Chun-Sung and
Tomkins, Andrew and Juan, Da-Cheng and Vu, Tu},
journal = {arXiv preprint arXiv:2608.27454},
year = {2026},
note = {Google Research}
}
```
## What it implements
- **Three-layer workspace** (paper §3.1): immutable `raw/` traces, persistent `wiki/`
(patterns + index + logs + skill-impact tracker), active `skills/` (SKILL.md + PURPOSE.md).
- **Evolutionary loop** (paper Algorithm 1, §3.2): Inference Agent rollouts with skill
injection but **no wiki access**; Wiki Maintainer consolidation (create/update pattern
pages via `append`/`replace`/`insert_after` patches); a **ReAct Skill Proposer** that
reads the wiki and raw traces on demand and emits one atomic skill proposal.
- **Gating & rollback** (§3.2.4): accept iff validation score strictly improves `R_best`;
otherwise roll back **skills only** — the wiki is never rolled back. The harness
programmatically appends proposal + unified diff + outcome to `wiki/skill-impact.md`.
- **Stratified trace sampling** (App. C): ≤5 failing + ≤3 passing traces, 15,000-char cap.
- **Paired bootstrap significance testing** (App. C) in `metrics.py`.
- Early stop when `R_best == 1.0`.
## Quickstart
```bash
pip install -e ".[dev]"
pytest # 14 tests incl. end-to-end loop
python -m wikiskill.demo # offline demo with the deterministic MockLLM
```
Demo output: a `wikiskill_demo_workspace/` with `raw/`, `wiki/`, `skills/` populated,
accepted/rejected proposals logged in `wiki/skill-impact.md`, and `run_state.json`.
## Using a real LLM
```bash
wikiskill evolve --tasks tasks.jsonl \
--train-ids train.txt --val-ids val.txt --test-ids test.txt \
--model <model-name> --base-url https://your-endpoint/v1
```
API key is read from `--api-key`, `WIKISKILL_API_KEY`, or `OPENAI_API_KEY`.
Any OpenAI-compatible `/chat/completions` endpoint works (including local vLLM, as in
the paper). `tasks.jsonl` lines: `{"id", "x", "y", "meta?"}`.
## Layout
```
src/wikiskill/
├── workspace.py # three layers + patch engine + scoped read_file
├── skills.py # skills layer, apply_proposal, skills-only snapshot/rollback
├── agents.py # InferenceAgent, WikiMaintainer, ReAct SkillProposer, stratified_sample
├── orchestrator.py # Algorithm 1 with gating/rollback + audit trail
├── prompts.py # agent system prompts (adapted from paper Appendix E)
├── llm.py # LLM protocol, OpenAI-compatible client, MockLLM, extract_json
├── metrics.py # accuracy + paired bootstrap (1,000 iterations)
├── datasets.py # dataclasses + synthetic demo dataset
├── demo.py / cli.py
└── tests/… # unit + end-to-end tests
```
## Notes & limitations (mirroring the paper)
- Skills are injected directly into the prompt (no retrieval), matching the paper's setup.
- Gating requires strict improvement; neutral proposals are rejected (paper Limitations).
- The wiki accumulates without pruning (paper Limitations).
design.md
+140
View File
@@ -0,0 +1,140 @@
# WikiSkill — Design Document
Implements: arXiv:2608.27454v1 (WikiSkill). See `prd.md` for requirements.
## 1. Architecture overview
```
wikiskill/
├── pyproject.toml
├── src/wikiskill/
│ ├── __init__.py
│ ├── llm.py # LLM provider abstraction: MockLLM, OpenAICompatLLM
│ ├── workspace.py # Three-layer workspace: raw/, wiki/, skills/ + patch engine
│ ├── prompts.py # System prompts (Inference / Wiki Maintainer / Skill Proposer)
│ ├── agents.py # InferenceAgent, WikiMaintainer, SkillProposer (ReAct)
│ ├── orchestrator.py # Algorithm 1: evolution loop, gating, rollback, logging
│ ├── metrics.py # accuracy metric + paired bootstrap significance test
│ ├── datasets.py # Task/Trace dataclasses + demo dataset
│ ├── demo.py # End-to-end demo run with mock LLM
│ └── cli.py # `wikiskill evolve` / `wikiskill demo` entry point
└── tests/
├── test_workspace.py
├── test_gating.py
└── test_e2e.py
```
## 2. Core data model (`datasets.py`)
- `Task(x, y, meta)` — task instance + ground truth (paper: 𝒟 = {(x_i, y_i)}).
- `ToolCall / Trajectory` — trace `τ = (o1, a1, ..., oT, aT)` captured as messages +
structured metadata (final answer, correct flag, tool calls and outputs).
- `ToolSpec(name, description, fn)` — environment tools 𝒰 (bash-like demo tools, `read_file`
for the proposer).
## 3. Workspace (`workspace.py`) — the three layers
`Workspace(root)` manages:
## 4. LLM abstraction (`llm.py`)
```python
class LLM(Protocol):
def chat(self, messages: list[dict], **kw) -> str: ...
```
- `OpenAICompatLLM`: calls any OpenAI-compatible `/chat/completions` endpoint
(configurable base_url, api_key via env, model, temperature). Uses `urllib` (no hard deps).
- `MockLLM`: deterministic scripted responder for tests/demo. Routes on content:
- Maintainer JSON → returns a small valid wiki update JSON.
- Proposer JSON → returns a valid create-skill proposal.
- Inference → returns a demo answer (extracted from task text).
- Robust JSON extraction helper (`extract_json`) shared by all agents (strips code fences).
## 5. Agents (`agents.py`)
### 5.1 Inference Agent
Runs one task: system prompt = task-type prompt + **injected active skills**
(full SKILL.md contents, no wiki access — paper §3.2/ablation). Loop:
LLM → tool call → observation → ... → final `ANSWER: ...` line or max turns.
Returns `Trajectory` (messages, tool calls, final answer, correct flag).
### 5.2 Wiki Maintainer (`M_WM`)
One LLM call (full-batch; paper Appendix D: B = N_train). Input: current wiki context
(index + all pattern pages + logs + skill-impact) and the stratified trace sample.
Output JSON parsed into workspace ops:
`create_patterns[]`, `update_patterns[] (edits[])`, `update_index`, `append_log`.
A **stratified sampler** picks ≤5 failing + ≤3 passing traces, each truncated to 15,000 chars.
### 5.3 Wiki-Informed Skill Proposer (`M_P`) — ReAct
Loop of up to `max_react_turns` (default 20, paper: ~10–20):
- Initial context: wiki index, `skill-impact.md`, pass/fail outcome summary (no pre-sampled traces).
- Tools: `read_file(path)` (workspace-scoped) + `finish(...)` to emit the final proposal.
- Final output JSON: `{"action": "create"|"patch", "skill": name, "content"/"edits": [...],
"purpose": mapping to wiki patterns}` — atomic, single-skill per paper §3.2.3.
## 6. Orchestrator (`orchestrator.py`) — Algorithm 1
```python
EvolutionOrchestrator(llm, tools, dataset, workspace, metric, K, ...).run()
```
Per iteration k (state S_k, W_k exactly as in the paper):
1. early stop if `R_best == 1.0`;
2. roll out training tasks with S_{k-1} → raw traces (skill injection, wiki hidden);
3. stratified sample T_sample,k;
4. `W'_k = M_WM(W_{k-1}, T_sample,k)`;
5. `P_k = M_P(W'_k, S_{k-1}, T_train,k)`;
6. snapshot skills; `S'_k = Apply(S_{k-1}, P_k)`;
7. validate on D_val → `R(T_val,k)`;
8. accept iff `R > R_best` else restore snapshot (skills-only rollback);
9. `W_k = Update(W'_k, P_k, R, a_k)` → append unified diff + outcome to `skill-impact.md`;
10. persist run state (history JSON: per-iteration scores, accept/reject, best).
## 7. Metrics & significance (`metrics.py`)
- `accuracy(traces)` — fraction correct.
- `paired_bootstrap(mine, base, n=1000, seed)` — resample test tasks with replacement,
return p-value that margins ≤ 0; used for "best method" claims (paper Appendix C).
## 8. Demo dataset (`datasets.py`)
`make_demo_dataset(seed)`: synthetic "unit-conversion" arithmetic tasks. The no-skill agent
often errs (rounding/format pitfalls baked into the mock); a good skill (the mock proposer
writes one) nudges the inference prompt to compute correctly — so validation gating can
genuinely accept/reject. Also pluggable: any `Dataset` of `Task`s with tools + metric.
## 9. Error handling & robustness
- Malformed LLM JSON → retry once, then skip gracefully (maintainer: no-op update with log
entry; proposer: proposal = None, iteration recorded as "no proposal").
- Missing files, bad patch targets → logged, op skipped (never crashes the loop).
- Path traversal blocked in `read_file`.
## 10. Testing strategy
- Unit: patch ops (exact/insert_after/append, invalid target), snapshot/restore,
path allowlist, immutable raw layer.
- Integration: gating math (accept on strict improvement, reject on equal/lower),
skill-impact.md written by harness with unified diff.
- E2E: MockLLM-driven full loop (K=3) — assert artifacts exist, acceptance behavior,
early-stop at R_best=1.0.
- **Raw layer**: `add_raw_trace(iter_k, task_id, trace_dict)` writes
`raw/iter_{k}/{task_id}.json`. Immutable: an existing file is never rewritten (enforced).
- **Wiki layer** (`wiki/`):
- `patterns/` pages: markdown files; updates via a patch engine implementing exactly the
three ops from the paper's Maintainer prompt: `append`, `replace` (exact-substring
target), `insert_after` (exact-substring target).
- `index.md`: full-content rewrite (maintainer always supplies complete index).
- `logs.md`: append-only evolution log.
- `skill-impact.md`: append-only; the **orchestrator** (not any agent) writes the
programmatic entry after gating: iteration, proposal metadata, target skill, unified
diff (`difflib.unified_diff` of skill content), validation score, accept/reject.
- **Skills layer**: `skills/<name>/SKILL.md` + `PURPOSE.md`.
- `apply_proposal(P)`: create (new dir with SKILL.md + PURPOSE.md) or patch (patch ops on
SKILL.md, mirrored to PURPOSE.md updates).
- `snapshot()` / `restore(snapshot)`: cheap rollback of the Skills layer only (dict of
name → content; rollback deletes/rewrites files accordingly). Wiki/raw untouched.
- `read_file(rel_path)` with a path-allowlist (agents may only read within the workspace)
— used by the ReAct Skill Proposer.
plan.md
+45
View File
@@ -0,0 +1,45 @@
# WikiSkill — Implementation Plan
Source: paper + `prd.md` + `design.md`. Ordered milestones with verification gates.
## Milestone 0 — Project scaffold
- [ ] `wikiskill/pyproject.toml` (no third-party runtime deps; pytest for dev)
- [ ] Package skeleton `src/wikiskill/` + `tests/`
- Verify: `pip install -e .[dev]` works; `pytest` collects.
## Milestone 1 — Data model + LLM abstraction
- [ ] `datasets.py`: `Task`, `ToolSpec`, `Trajectory`, `Dataset`, demo dataset factory
- [ ] `llm.py`: `LLM` protocol, `OpenAICompatLLM` (urllib), `MockLLM`, `extract_json`
- Verify: unit test for `extract_json` + MockLLM routing.
## Milestone 2 — Three-layer workspace (paper §3.1)
- [ ] `workspace.py`: raw traces (immutable), wiki (patterns/index/logs/skill-impact),
patch engine (`append`/`replace`/`insert_after`), skills (SKILL.md + PURPOSE.md),
snapshot/restore, scoped `read_file`
- Verify: `test_workspace.py` passes (incl. immutability + path traversal).
## Milestone 3 — Agents (paper §3.2.1–3.2.3)
- [ ] `prompts.py`: three agent system prompts adapted from paper Appendix E
- [ ] `agents.py`: InferenceAgent (tool loop, skill injection, no wiki), WikiMaintainer
(single call, JSON ops), SkillProposer (ReAct, ≤20 turns, `read_file`, atomic proposal)
- Verify: mock-driven unit tests for each agent's outputs.
## Milestone 4 — Orchestrator: gating & rollback (paper §3.2.4, Algorithm 1)
- [ ] `orchestrator.py`: full loop, stratified sampling (≤5 fail / ≤3 pass, 15k char cap),
strict-improvement gate, skills-only rollback, programmatic `skill-impact.md` diff
logging, early stop at R_best = 1.0, run history persistence
- Verify: `test_gating.py` passes (accept/reject math, wiki never rolled back).
## Milestone 5 — Metrics & significance (paper Appendix C)
- [ ] `metrics.py`: accuracy + paired bootstrap (1,000 iterations)
- Verify: unit test with seeded RNG.
## Milestone 6 — End-to-end validation
- [ ] `demo.py` + `cli.py`
- [ ] `test_e2e.py`: K=3 loop with MockLLM; assert layers populated, gating works, artifacts
- [ ] Live run: `python -m wikiskill.demo`
- Verify: all tests green; demo produces workspace with accepted skill or logged rejections.
## Milestone 7 — Documentation & wrap-up
- [ ] `README.md` (usage, mapping to paper)
- Final review of all docs vs. implementation.
prd.md
+93
View File
@@ -0,0 +1,93 @@
# WikiSkill — PRD (Product Requirements Document)
Source paper: **"WikiSkill: Compiling Agent Experience into Persistent Knowledge for Skill Evolution"**
(Tang, Rashtchian, Ferng, Tomkins, Juan, Vu — Google Research / Virginia Tech, arXiv:2608.27454v1)
## 1. Background
Agent skills package specialized procedural knowledge into reusable, filesystem-based
resources. Recent methods (Trace2Skill, EvoSkill, SkillOpt) discover skills from agent
execution traces, but the insights that guide skill development remain scattered across
optimization histories, limiting systematic reuse across iterations.
WikiSkill co-evolves agent skills with a **persistent knowledge base (wiki)**. It separates
raw experience, accumulated knowledge, and executable skills, and continuously consolidates
experience into the wiki so that subsequent skill updates build on it.
## 2. Goals
- G1: Implement the full WikiSkill evolution loop (paper Algorithm 1) as a runnable Python framework.
- G2: Faithfully reproduce the paper's three-layer knowledge architecture (`raw/`, `wiki/`, `skills/`).
- G3: Implement the three evolutionary agents: Inference Agent, Wiki Maintainer, and a
ReAct-style Wiki-Informed Skill Proposer.
- G4: Implement validation gating with skill-only rollback (the wiki is never rolled back).
- G5: Make the system testable end-to-end without any LLM API key (mock LLM provider),
while supporting real providers (OpenAI-compatible APIs) for actual use.
## 3. Non-goals
- No skill retrieval/triggering: active skills are injected directly into the Inference
Agent's prompt (per the paper's setup, which isolates skill quality).
- No re-implementation of baselines (Trace2Skill / EvoSkill / SkillOpt).
- No benchmark harnesses for the paper's exact datasets (LiveMath, SealQA, SpreadsheetBench,
OfficeQA, ALFWorld); a simple demo dataset with an identical interface is provided instead.
## 4. Functional requirements
### FR1 — Three-layer workspace
- **Raw Layer (`raw/`)**: immutable execution traces per iteration (reasoning, tool calls,
tool outputs, final answers, pass/fail, prediction vs. ground truth). Never modified.
- **Wiki Layer (`wiki/`)**: persistent, compounding; contains
- `wiki/patterns/*.md` — pattern pages (failure modes / successful strategies + workarounds),
- `wiki/index.md` — concise catalog of patterns,
- `wiki/logs.md` — evolution log appended by the Wiki Maintainer each iteration,
- `wiki/skill-impact.md` — skill impact tracker updated programmatically by the harness
after gating (proposal metadata, target skill, unified diff, val score, accept/reject).
- **Skills Layer (`skills/<name>/`)**: each skill contains `SKILL.md` (frontmatter: name,
description, plus procedural instructions) and `PURPOSE.md` (mapping to motivating wiki patterns).
### FR2 — Evolution loop (paper Algorithm 1)
1. Initialize `S0 = ∅`, `W0 = ∅`; baseline validation `R_best = R(T_val,0)`.
2. For k = 1..K (early-stop if `R_best == 1.0`):
- Roll out Inference Agent on `D_train` with skills `S_{k-1}` (NO wiki access).
- Sample `T_sample,k ⊂ T_train,k` (stratified: ≤5 failing, ≤3 passing traces; each trace
capped at 15,000 characters).
- Wiki Maintenance: `W'_k ← M_WM(W_{k-1}, T_sample,k)` — create/update pattern pages via
patch operations, update index, append log.
- Skill Proposal: `P_k ← M_P(W'_k, S_{k-1}, T_train,k)` — a ReAct agent that reads the wiki
index, skill-impact tracker, outcome summary, and on-demand reads pattern pages and raw
traces via a `read_file` tool. Produces ONE atomic proposal targeting a single skill
(create new skill OR patch-based edit).
- Apply: `S'_k = Apply(S_{k-1}, P_k)`.
- Validate on `D_val`.
- Gate: accept iff `R(T_val,k) > R_best` (strict improvement); otherwise roll back skills
to `S_{k-1}`. Wiki is never rolled back.
- Update wiki: append proposal metadata + diff + score + outcome to `skill-impact.md`.
3. Return final skill set `S_K` and wiki `W_K`.
### FR3 — Agents
- **Inference Agent**: LLM agent with environment tools; receives task context + injected
active skill contents; multi-turn trajectory.
- **Wiki Maintainer**: single LLM call per iteration; outputs JSON with
`create_patterns`, `update_patterns` (patch ops: append/replace/insert_after),
`update_index` (full content), `append_log`.
- **Skill Proposer**: multi-turn ReAct agent (paper: ~10–20 turns) with `read_file` and a
final structured proposal (create/patch one skill).
### FR4 — Gating and persistence
- Strict-improvement acceptance, rollback of skills only, programmatic `skill-impact.md` updates.
- Wiki state persists across all iterations and rollbacks.
### FR5 — Configurability & evaluation
- Pluggable LLM provider (mock / OpenAI-compatible), configurable K, tools, splits.
- Metric per dataset (accuracy); paired bootstrap significance testing helper (1,000 iterations)
as in paper Appendix C.
## 5. Acceptance criteria
1. `pytest` passes: unit tests for workspace/patches/gating, and an end-to-end test of the
full evolution loop using a scripted mock LLM.
2. A demo run (`python -m wikiskill.demo`) executes ≥2 evolution iterations, produces a
workspace with all three layers populated, and accepts/rolls back proposals correctly.
3. Code mirrors paper notation (S_k, W_k, P_k, T_sample, R_best) and cites Algorithm 1 steps
in docstrings.
pyproject.toml
+19
View File
@@ -0,0 +1,19 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "wikiskill"
version = "0.1.0"
description = "WikiSkill: Compiling Agent Experience into Persistent Knowledge for Skill Evolution (arXiv:2608.27454v1)"
requires-python = ">=3.9"
dependencies = []
[project.optional-dependencies]
dev = ["pytest>=7"]
[project.scripts]
wikiskill = "wikiskill.cli:main"
[tool.setuptools.packages.find]
where = ["src"]
src/wikiskill/__init__.py
+6
View File
@@ -0,0 +1,6 @@
"""WikiSkill: co-evolving agent skills with a persistent knowledge base (wiki).
Implementation of arXiv:2608.27454v1.
"""
__version__ = "0.1.0"
src/wikiskill/agents.py
+134
View File
@@ -0,0 +1,134 @@
"""Evolutionary agents (paper §3.2): InferenceAgent, WikiMaintainer, SkillProposer."""
from __future__ import annotations
import json
import re
from typing import Callable, Dict, List, Optional
from .llm import LLM, extract_json
from .prompts import INFERENCE_SYSTEM, SKILL_PROPOSER_SYSTEM, WIKI_MAINTAINER_SYSTEM
from .skills import SkillsLayer
from .workspace import Workspace
ANSWER_RE = re.compile(r"ANSWER:\s*(.+)", re.IGNORECASE)
def _parse_answer(text: str) -> Optional[str]:
m = ANSWER_RE.search(text or "")
return m.group(1).strip() if m else None
class InferenceAgent:
"""§3.2.1: π(x_i; S). Active skills are injected; wiki access is NOT given."""
def __init__(self, llm: LLM, tools: list, ws: Workspace, skills: SkillsLayer,
max_turns: int = 6):
self.llm, self.ws, self.skills = llm, ws, skills
self.tools = {t.name: t for t in tools}
self.max_turns = max_turns
def run(self, task) -> "object":
from .datasets import Trajectory
system = INFERENCE_SYSTEM.replace("{active_skills}",
self.skills.active_skills_context())
messages: List[dict] = [{"role": "system", "content": system},
{"role": "user", "content": task.x}]
tool_calls: List[dict] = []
final: Optional[str] = None
for _ in range(self.max_turns):
text = self.llm.chat(messages, temperature=0.0)
messages.append({"role": "assistant", "content": text})
parsed = _parse_answer(text)
if parsed is not None:
final = parsed
break
tm = re.search(r'TOOL:\s*(\S+)\s*(\{.*\})?', text, re.DOTALL)
if not tm or tm.group(1) not in self.tools:
messages.append({"role": "user", "content":
"Please reply with TOOL: <name> {json args} or a final "
"'ANSWER: ...' line."})
continue
tool = self.tools[tm.group(1)]
try:
args = json.loads(tm.group(2) or "{}")
except json.JSONDecodeError:
args = {}
obs = tool.fn(**args)
tool_calls.append({"tool": tool.name, "args": args, "observation": obs})
messages.append({"role": "user", "content": f"Observation: {obs}"})
traj = Trajectory(task_id=task.id, messages=messages, tool_calls=tool_calls,
final_answer=final, correct=False, prediction=final)
traj.correct = (final is not None and str(final).strip() == str(task.y).strip())
return traj
def stratified_sample(traces: List, max_fail: int = 5, max_pass: int = 3,
max_chars: int = 15000) -> List:
"""Paper Appendix C: <=5 failing + <=3 passing traces, each capped at 15,000 chars."""
fails = [t for t in traces if not t.correct][:max_fail]
passes = [t for t in traces if t.correct][:max_pass]
sampled = fails + passes
for t in sampled:
if len(t.messages) and isinstance(t.messages[-1].get("content"), str) and \
len(t.messages[-1]["content"]) > max_chars:
c = t.messages[-1]["content"][:max_chars]
t.messages[-1] = dict(t.messages[-1], content=c + "\n...[truncated]")
return sampled
class WikiMaintainer:
"""§3.2.2: W'_k = M_WM(W_{k-1}, T_sample,k). One LLM call per iteration."""
def __init__(self, llm: LLM, ws: Workspace):
self.llm, self.ws = llm, ws
def consolidate(self, sampled_traces: List[dict], iteration: int) -> dict:
trace_dump = json.dumps([t.to_dict() for t in sampled_traces],
indent=1, ensure_ascii=False)
user = (f"## Iteration {iteration}\n\n## Current wiki context\n"
f"{self.ws.wiki_context()}\n\n## Latest execution traces\n{trace_dump}\n\n"
"Produce the wiki update JSON now.")
messages = [{"role": "system", "content": WIKI_MAINTAINER_SYSTEM},
{"role": "user", "content": user}]
parsed = extract_json(self.llm.chat(messages, temperature=0.2))
if parsed is None: # one retry, then graceful no-op
parsed = extract_json(self.llm.chat(messages + [{"role": "user", "content":
"Your previous reply was not valid JSON. Return ONLY the JSON object."}],
temperature=0.0))
return parsed or {}
class SkillProposer:
"""§3.2.3: P_k = M_P(W'_k, S_{k-1}, T_train,k). Multi-turn ReAct agent that
reads the wiki index, skill-impact tracker, and traces on demand via read_file."""
def __init__(self, llm: LLM, ws: Workspace, skills: SkillsLayer, max_turns: int = 20):
self.llm, self.ws, self.skills = llm, ws, skills
self.max_turns = max_turns
def propose(self, outcome_summary: str, iteration: int) -> Optional[dict]:
index = self.ws.wiki_index() or "(empty wiki)"
impact = self.ws.skill_impact() or "(no proposals yet)"
user = (f"## Iteration {iteration}\n\n## Wiki index\n{index}\n\n"
f"## Skill impact history\n{impact}\n\n"
f"## Training task outcomes (pass/fail summary)\n{outcome_summary}\n\n"
"Use read_file to inspect pattern pages (wiki/patterns/...) and raw traces "
"(raw/iter_.../...) as needed, then finish with your atomic proposal JSON.")
messages = [{"role": "system", "content": SKILL_PROPOSER_SYSTEM},
{"role": "user", "content": user}]
for _ in range(self.max_turns):
text = self.llm.chat(messages, temperature=0.2)
parsed = extract_json(text)
if parsed and parsed.get("action") in ("create", "patch"):
return parsed
fm = re.search(r'"path"\s*:\s*"([^"]+)"', text)
if fm:
obs = self.ws.read_file(fm.group(1))
messages.append({"role": "assistant", "content": text})
messages.append({"role": "user",
"content": f"read_file observation:\n{obs}"})
continue
messages.append({"role": "assistant", "content": text})
messages.append({"role": "user", "content":
"Reply with a read_file tool call or the finish JSON."})
return None
src/wikiskill/cli.py
+71
View File
@@ -0,0 +1,71 @@
"""CLI: `wikiskill demo` (offline mock) or `wikiskill evolve` (real LLM endpoint)."""
from __future__ import annotations
import argparse
import json
import os
import sys
def main(argv=None) -> int:
p = argparse.ArgumentParser("wikiskill")
sub = p.add_subparsers(dest="cmd", required=True)
d = sub.add_parser("demo", help="run the offline demo with the mock LLM")
d.add_argument("--root", default="wikiskill_demo_workspace")
d.add_argument("--iterations", type=int, default=3)
e = sub.add_parser("evolve", help="evolve skills on a JSONL task file")
e.add_argument("--tasks", required=True, help="JSONL file with {id,x,y} objects")
e.add_argument("--train-ids", required=True)
e.add_argument("--val-ids", required=True)
e.add_argument("--test-ids", required=True)
e.add_argument("--root", default="wikiskill_workspace")
e.add_argument("--model", required=True)
e.add_argument("--base-url", default=None)
e.add_argument("--api-key", default=None)
e.add_argument("--iterations", type=int, default=8)
args = p.parse_args(argv)
if args.cmd == "demo":
from .demo import run_demo
run_demo(args.root, args.iterations)
return 0
from .datasets import Dataset, Task, ToolSpec, load_jsonl
from .llm import OpenAICompatLLM
from .orchestrator import EvolutionOrchestrator
from .workspace import Workspace
rows = {r["id"]: r for r in load_jsonl(args.tasks)}
def ids(path):
with open(path, "r", encoding="utf-8") as f:
return [line.strip() for line in f if line.strip()]
def to_tasks(id_list):
return [Task(id=i, x=rows[i]["x"], y=rows[i]["y"], meta=rows[i].get("meta", {}))
for i in id_list if i in rows]
dataset = Dataset(train=to_tasks(ids(args.train_ids)),
val=to_tasks(ids(args.val_ids)),
test=to_tasks(ids(args.test_ids)))
def calculator(expression: str) -> str:
try:
return repr(eval(expression, {"__builtins__": {}}, {}))
except Exception as exc: # noqa: BLE001
return f"ERROR: {exc}"
tools = [ToolSpec("calculator", "Evaluate a Python arithmetic expression.", calculator)]
llm = OpenAICompatLLM(model=args.model, base_url=args.base_url, api_key=args.api_key)
orch = EvolutionOrchestrator(llm=llm, dataset=dataset,
ws=Workspace(args.root), tools=tools,
K=args.iterations)
result = orch.run()
print(json.dumps(result, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())
src/wikiskill/datasets.py
+106
View File
@@ -0,0 +1,106 @@
"""Data model + demo dataset (design.md §2, §8)."""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional
@dataclass
class Task:
"""One task instance: paper §2 𝒟 = {(x_i, y_i)}."""
id: str
x: str # task context shown to the agent
y: str # ground-truth answer
meta: Dict = field(default_factory=dict)
@dataclass
class ToolSpec:
"""An environment tool 𝒰 (paper §2). fn(**kwargs) -> str observation."""
name: str
description: str
fn: Callable[..., str]
def schema(self) -> dict:
return {"name": self.name, "description": self.description}
@dataclass
class Trajectory:
"""Execution trace τ_i = (o1, a1, ..., oT, aT) plus grading metadata."""
task_id: str
messages: List[dict]
tool_calls: List[dict]
final_answer: Optional[str]
correct: bool
prediction: Optional[str] = None
def to_dict(self) -> dict:
return {
"task_id": self.task_id,
"messages": self.messages,
"tool_calls": self.tool_calls,
"final_answer": self.final_answer,
"correct": self.correct,
"prediction": self.prediction,
}
@dataclass
class Dataset:
"""Train/val/test split (paper §2)."""
train: List[Task]
val: List[Task]
test: List[Task]
name: str = "demo"
def _demo_tools() -> List[ToolSpec]:
"""Simple calculator tool for the demo environment."""
def calc(expression: str) -> str:
try:
return repr(eval(expression, {"__builtins__": {}}, {}))
except Exception as e: # noqa: BLE001 - demo tool
return f"ERROR: {e}"
return [ToolSpec("calculator", "Evaluate a Python arithmetic expression.", calc)]
def make_demo_dataset(seed: int = 0) -> Dataset:
"""Synthetic unit-conversion dataset (deterministic).
Tasks embed `expected=<answer>`; the MockLLM truncates decimals without the
evolved skill and answers exactly with it — so gating has real signal.
"""
conversions = [
("kilometers to miles", 0.621371, "miles"),
("miles to kilometers", 1.609344, "kilometers"),
("kilograms to pounds", 2.20462, "pounds"),
("pounds to kilograms", 0.453592, "kilograms"),
("inches to centimeters", 2.54, "centimeters"),
("gallons to liters", 3.78541, "liters"),
]
counts = [3, 2, 2, 2, 1, 1] # 11 tasks: 5 train / 3 val / 3 test
tasks: List[Task] = []
n = 0
for (name, factor, unit), count in zip(conversions, counts):
for i in range(count):
n += 1
value = round(1.0 + ((n * 7 + seed * 13) % 50) / 3.0, 4)
answer = round(value * factor, 2)
x = (f"Convert {value} {name.split(' to ')[0]} to {unit}. "
f"The conversion factor is {factor}. "
f"Report the numeric value rounded to exactly 2 decimal places. "
f"(reference: expected={answer})")
tasks.append(Task(id=f"task-{n:03d}", x=x, y=f"{answer}", meta={"unit": unit}))
return Dataset(train=tasks[:5], val=tasks[5:8], test=tasks[8:], name="demo-unitconv")
def load_jsonl(path: str) -> List[dict]:
out = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line:
out.append(json.loads(line))
return out
src/wikiskill/demo.py
+32
View File
@@ -0,0 +1,32 @@
"""End-to-end demo run of the WikiSkill loop with the deterministic MockLLM."""
from __future__ import annotations
import argparse
import os
from .datasets import make_demo_dataset, _demo_tools
from .llm import MockLLM
from .orchestrator import EvolutionOrchestrator
from .workspace import Workspace
def run_demo(root: str = "wikiskill_demo_workspace", K: int = 3) -> dict:
ws = Workspace(root)
skills_logger = []
orch = EvolutionOrchestrator(
llm=MockLLM(), dataset=make_demo_dataset(), ws=ws, tools=_demo_tools(), K=K)
result = orch.run()
print("\nWorkspace layout:")
for dirpath, _, files in os.walk(root):
rel = os.path.relpath(dirpath, root)
for fn in sorted(files):
print(f" {os.path.join(rel, fn)}")
return result
if __name__ == "__main__":
p = argparse.ArgumentParser(description="WikiSkill demo (mock LLM, offline)")
p.add_argument("--root", default="wikiskill_demo_workspace")
p.add_argument("--iterations", type=int, default=3)
a = p.parse_args()
run_demo(a.root, a.iterations)
src/wikiskill/llm.py
+139
View File
@@ -0,0 +1,139 @@
"""LLM provider abstraction (design.md §4): protocol, OpenAI-compatible client, MockLLM."""
from __future__ import annotations
import json
import os
import re
import urllib.request
from typing import Any, List, Optional, Protocol
class LLM(Protocol):
def chat(self, messages: List[dict], **kw) -> str: ...
def extract_json(text: str) -> Optional[Any]:
"""Extract the first JSON object/array from an LLM response (tolerates fences)."""
text = text.strip()
m = re.search(r"```(?:json)?\s*([\s\S]*?)```", text)
if m:
text = m.group(1).strip()
for start_ch, end_ch in (("{", "}"), ("[", "]")):
start = text.find(start_ch)
if start == -1:
continue
depth = 0
for i in range(start, len(text)):
if text[i] == start_ch:
depth += 1
elif text[i] == end_ch:
depth -= 1
if depth == 0:
try:
return json.loads(text[start : i + 1])
except json.JSONDecodeError:
break
break
return None
class OpenAICompatLLM:
"""Calls any OpenAI-compatible /chat/completions endpoint using stdlib urllib."""
def __init__(self, model: str, base_url: Optional[str] = None,
api_key: Optional[str] = None, temperature: float = 0.7):
self.model = model
self.base_url = (base_url or os.environ.get("WIKISKILL_BASE_URL",
"https://api.openai.com/v1")).rstrip("/")
self.api_key = api_key or os.environ.get("WIKISKILL_API_KEY",
os.environ.get("OPENAI_API_KEY", ""))
self.temperature = temperature
def chat(self, messages: List[dict], temperature: Optional[float] = None,
max_tokens: int = 4096) -> str:
payload = {
"model": self.model,
"messages": messages,
"temperature": self.temperature if temperature is None else temperature,
"max_tokens": max_tokens,
}
req = urllib.request.Request(
self.base_url + "/chat/completions",
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"},
method="POST",
)
with urllib.request.urlopen(req, timeout=300) as resp:
data = json.loads(resp.read().decode("utf-8"))
return data["choices"][0]["message"]["content"]
class MockLLM:
"""Deterministic scripted LLM for tests and the offline demo.
`handlers` maps a substring of the conversation to `(messages, call_log) -> str`.
Fallback behaviour implements the demo dataset semantics.
"""
def __init__(self, handlers: Optional[dict] = None):
self.handlers = dict(handlers or {})
self.call_log: List[dict] = []
def chat(self, messages: List[dict], **kw) -> str:
self.call_log.append({"messages": messages})
blob = "\n".join(str(m.get("content", "")) for m in messages)
for needle, fn in self.handlers.items():
if needle in blob:
return fn(messages, self.call_log)
return self._fallback(blob)
# ---- default demo behaviour -------------------------------------------
def _fallback(self, blob: str) -> str:
if "Wiki Maintainer Agent" in blob:
return self._maintainer()
if "Skill Proposer Agent" in blob:
return self._proposer()
return self._inference(blob)
def _maintainer(self) -> str:
return json.dumps({
"create_patterns": [{
"name": "rounding-errors.md",
"content": ("# Pattern: Rounding errors\n\n## Description\n"
"Demo pattern: naive rounding gives wrong answers.\n\n"
"## Root cause\nSkipping the final rounding step.\n\n"
"## Workaround\nAlways round the final result to 2 decimals.\n"),
}],
"update_patterns": [],
"update_index": ("- [rounding-errors](wiki/patterns/rounding-errors.md): "
"wrong answers from skipped rounding; FIX: round final "
"result to 2 decimals.\n"),
"append_log": "Analyzed traces; recorded rounding-error pattern.",
})
def _proposer(self) -> str:
return json.dumps({
"action": "create",
"skill": "unit-conversion",
"content": ("---\nname: unit-conversion\ndescription: Correctly convert units "
"with exact arithmetic.\n---\n\n# Unit Conversion Skill\n\n"
"1. Compute using exact ratios.\n"
"2. Round ONLY the final result to 2 decimal places.\n"
"3. State the answer as `ANSWER: <value> <unit>`.\n"),
"purpose": "Created from wiki pattern rounding-errors.md.",
})
def _inference(self, blob: str) -> str:
# Demo tasks embed their correct answer as "expected=...". With the evolved
# skill injected, the mock agent answers exactly; without it, it truncates
# decimals (a systematic, deterministic "model weakness").
m = re.search(r"expected=([-\d.]+)", blob)
if not m:
return "ANSWER: 0"
val = m.group(1)
if "round only the final result" in blob.lower():
return f"ANSWER: {val}"
whole, _, frac = val.partition(".")
return "ANSWER: " + whole + "." + (frac[:1] if frac else "0")
src/wikiskill/metrics.py
+28
View File
@@ -0,0 +1,28 @@
"""Metrics (paper Appendix C): accuracy + paired bootstrap significance testing."""
from __future__ import annotations
import random
from typing import Sequence
def accuracy(correct_flags: Sequence[bool]) -> float:
return sum(bool(c) for c in correct_flags) / len(correct_flags) if correct_flags else 0.0
def paired_bootstrap(a: Sequence[bool], b: Sequence[bool], n: int = 1000,
seed: int = 0) -> float:
"""Paired bootstrap (paper App. C): resample test tasks with replacement; return
the p-value that method `a` does not outperform `b` (one-sided)."""
assert len(a) == len(b)
rng = random.Random(seed)
N = len(a)
if N == 0:
return 1.0
margins = [1.0 if x and not y else (-1.0 if y and not x else 0.0)
for x, y in zip(a, b)]
worse_or_equal = 0
for _ in range(n):
sample = [margins[rng.randrange(N)] for _ in range(N)]
if sum(sample) / N <= 0:
worse_or_equal += 1
return worse_or_equal / n
src/wikiskill/orchestrator.py
+135
View File
@@ -0,0 +1,135 @@
"""Evolution orchestrator — paper Algorithm 1 (Appendix A.1) with §3.2.4 gating."""
from __future__ import annotations
import json
import os
from typing import Callable, List, Optional
from .agents import InferenceAgent, SkillProposer, WikiMaintainer, stratified_sample
from .datasets import Dataset, Trajectory
from .llm import LLM
from .skills import SkillsLayer
from .workspace import Workspace
class EvolutionOrchestrator:
"""Co-evolves (S_k, W_k) per Algorithm 1:
S0 = W0 = ∅; R_best = R(T_val,0)
for k = 1..K:
if R_best == 1.0: break
T_train,k = {τ ~ π(x; S_{k-1})} on D_train # inference, no wiki
T_sample,k ⊂ T_train,k # stratified (App. C)
W'_k = M_WM(W_{k-1}, T_sample,k) # wiki maintenance
P_k = M_P(W'_k, S_{k-1}, T_train,k) # ReAct proposer
S'_k = Apply(S_{k-1}, P_k)
validate S'_k on D_val
accept iff R(T_val,k) > R_best else roll back skills (wiki retained)
W_k = Update(W'_k, P_k, R(T_val,k), a_k) # harness writes skill-impact.md
return S_K, W_K
"""
def __init__(self, llm: LLM, dataset: Dataset, ws: Workspace, tools: list,
metric: Callable[[List[bool]], float] = None, K: int = 8,
max_react_turns: int = 20, log: Optional[Callable[[str], None]] = None):
from .metrics import accuracy
self.llm = llm
self.dataset = dataset
self.ws = ws
self.metric = metric or accuracy
self.K = K
self.skills = SkillsLayer(ws)
self.inference = InferenceAgent(llm, tools, ws, self.skills)
self.maintainer = WikiMaintainer(llm, ws)
self.proposer = SkillProposer(llm, ws, self.skills, max_turns=max_react_turns)
self.log = log or (lambda s: print(s, flush=True))
self.history: List[dict] = []
self.r_best = -1.0
# ------------------------------------------------------------------
def _rollout(self, tasks, iteration: int, split: str) -> List[Trajectory]:
trajs = [self.inference.run(t) for t in tasks]
for t in trajs:
self.ws.add_raw_trace(iteration, t.task_id,
{**t.to_dict(), "split": split, "iteration": iteration})
return trajs
def _score(self, trajs: List[Trajectory]) -> float:
return self.metric([t.correct for t in trajs])
def _outcome_summary(self, trajs: List[Trajectory], tasks) -> str:
by_id = {t.id: t for t in tasks}
return "\n".join(
f"{tr.task_id}: {'PASS' if tr.correct else 'FAIL'} | "
f"predicted={tr.prediction!r} ground_truth={by_id[tr.task_id].y!r} "
f"| trace: raw/iter_*/{tr.task_id}.json" for tr in trajs)
def _baseline_validation(self) -> None:
self.log("Baseline validation (S0 = empty) ...")
trajs = self._rollout(self.dataset.val, 0, "val")
self.r_best = self._score(trajs)
self.log(f"R_best = {self.r_best:.4f}")
# ------------------------------------------------------------------
def run(self) -> dict:
self._baseline_validation()
for k in range(1, self.K + 1):
if self.r_best >= 1.0:
self.log("R_best reached 1.0; early stop.")
break
self.log(f"\n=== Iteration {k}/{self.K} ===")
# lines 8-9: training rollouts + stratified sample
train_trajs = self._rollout(self.dataset.train, k, "train")
sampled = stratified_sample(train_trajs)
# line 10: wiki maintenance
update = self.maintainer.consolidate(sampled, k)
report = self.ws.apply_maintainer_update(update) if update else {}
self.log(f"Wiki: created={report.get('created')} updated={report.get('updated')} "
f"errors={report.get('errors')}")
# line 11: skill proposal
proposal = self.proposer.propose(
self._outcome_summary(train_trajs, self.dataset.train), k)
entry = {"iteration": k, "proposal": None, "val_score": None,
"accepted": False}
if proposal:
snap = self.skills.snapshot()
try:
applied = self.skills.apply_proposal(proposal)
except Exception as e: # invalid proposal → skip iteration gracefully
self.log(f"Proposal invalid, skipped: {e}")
applied = None
if applied is not None:
# lines 13-14: validate + gate
val_trajs = self._rollout(self.dataset.val, k, "val")
score = self._score(val_trajs)
accepted = score > self.r_best
if accepted:
self.r_best = score # line 15
else:
self.skills.restore(snap) # line 17: skills-only rollback
self.log(f"Proposal {applied['action']} '{applied['skill']}' -> "
f"val={score:.4f} ({'ACCEPTED' if accepted else 'REJECTED'})")
# line 19: harness records audit trail in skill-impact.md
diff = applied.get("diff", "(new skill)")
self.ws.record_skill_impact(
f"\n## Iteration {k} — {applied['action']} "
f"'{applied['skill']}'\n"
f"- validation score: {score:.4f}\n"
f"- outcome: {'Accepted' if accepted else 'Rejected'}\n"
f"- purpose: {proposal.get('purpose', '')}\n"
f"- diff:\n```diff\n{diff}\n```\n")
entry.update({"proposal": applied.get("skill"),
"action": applied["action"], "val_score": score,
"accepted": accepted})
else:
self.log("No proposal produced this iteration.")
self.history.append(entry)
final_skills = self.skills.list_skills()
result = {"r_best": self.r_best, "history": self.history,
"final_skills": final_skills}
with open(os.path.join(self.ws.root, "run_state.json"), "w",
encoding="utf-8") as f:
json.dump(result, f, indent=2)
self.log(f"\nEvolution complete. R_best={self.r_best:.4f}, skills={final_skills}")
return result
src/wikiskill/prompts.py
+82
View File
@@ -0,0 +1,82 @@
"""System prompts (paper Appendix E). Condensed but structure-faithful versions."""
INFERENCE_SYSTEM = """You are a task-solving agent. Complete the task step by step.
You may use the following tools by replying with a single line:
TOOL: <tool_name> {"arg": "value"}
The environment will reply with the observation, then you continue.
When you are done, reply with your final answer on its own line in the format:
ANSWER: <answer>
## Active skills (follow them when applicable)
{active_skills}
"""
WIKI_MAINTAINER_SYSTEM = """You are a Wiki Maintainer Agent for an LLM skill evolution
system. Your job is to maintain a structured knowledge base (wiki) that documents
patterns observed during agent execution -- both successes and failures. Perform DEEP
ANALYSIS of execution logs to identify root causes, not just surface-level symptoms.
## Wiki structure
- wiki/index.md -- concise catalog of known patterns (one line per pattern)
- wiki/logs.md -- chronological evolution log
- wiki/skill-impact.md -- record of which skills were tried and their outcomes
- wiki/patterns/ -- one page per pattern with evidence and analysis
## Your input
1. Execution traces from the latest iteration (agent actions, tool calls, feedback)
2. The current wiki context (index, log, pattern pages)
## Your output (incremental edit mode)
Return a JSON object with these keys:
- "create_patterns": [{"name": "pattern-name.md", "content": "..."}] -- new patterns
- "update_patterns": [{"name": "existing-pattern.md", "edits": [...]}] -- patch existing
- "update_index": full updated content of index.md (REQUIRED, always complete)
- "append_log": brief summary of this iteration's findings (REQUIRED)
### Patch operations (for update_patterns.edits)
- {"op": "append", "content": "text to add at end"}
- {"op": "replace", "target": "exact text to find", "content": "replacement"}
- {"op": "insert_after", "target": "exact text to find", "content": "inserted text"}
Targets must be EXACT substrings of the existing content. Keep each edit minimal.
## Analysis guidelines
1. Read the agent's actual actions; compare successful vs failed tasks.
2. Identify ACTION PATTERNS and strategies, not just error messages.
3. Pattern pages document: description, root cause analysis (WHY), exact sequences
from traces, and known solutions/workarounds with concrete syntax.
4. Capture BOTH failure patterns and success patterns.
5. Do NOT duplicate patterns -- update existing ones with new evidence.
6. Be concise: pattern pages should be 10-30 lines.
7. Index entries are critical: format "- [name](wiki/patterns/name.md): PROBLEM +
ROOT CAUSE + FIX in one line".
"""
SKILL_PROPOSER_SYSTEM = """You are a Skill Proposer Agent for an LLM skill evolution
system, operating in ReAct mode. Your goal: propose ONE atomic skill update (create a
new skill OR patch one existing skill) that improves the agent's performance, informed
by the persistent wiki.
## Available tools
- read_file: {"path": "<relative path>"} -- read a wiki pattern page, a raw execution
trace, an existing skill, or any workspace file.
- finish: {"action": "create"|"patch", "skill": "<name>", "content": "<full SKILL.md
content for create>", "edits": [<patch ops for patch>], "purpose": "<mapping to the
wiki patterns that motivated this proposal>"}
## Procedure (ReAct)
1. Read wiki/index.md and the skill-impact history. NEVER re-propose an intervention
that was already rejected (check skill-impact.md).
2. Use read_file to inspect specific pattern pages and raw traces of FAILED tasks to
diagnose root causes before proposing.
3. Propose exactly ONE change: either create a new skill (full content with YAML
frontmatter: name, description) or patch one existing skill with minimal edits.
4. In "purpose", map the skill back to the motivating wiki patterns.
5. When ready, emit ONLY the finish tool call with valid JSON.
## Patch op format (for "edits")
{"op": "append", "content": "..."} |
{"op": "replace", "target": "<exact substring>", "content": "..."} |
{"op": "insert_after", "target": "<exact substring>", "content": "..."}
"""
src/wikiskill/skills.py
+90
View File
@@ -0,0 +1,90 @@
"""Skills Layer (paper §3.1): skills/<name>/SKILL.md + PURPOSE.md, with
snapshot/restore for skills-only rollback during validation gating (§3.2.4)."""
from __future__ import annotations
import difflib
import os
import re
from typing import Dict, List, Optional
from .workspace import PatchError, Workspace, apply_edits
SKILL_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$")
class SkillsLayer:
def __init__(self, ws: Workspace):
self.ws = ws
# ---------- state ----------
def list_skills(self) -> List[str]:
root = os.path.join(self.ws.root, "skills")
if not os.path.isdir(root):
return []
return sorted(d for d in os.listdir(root)
if os.path.isfile(os.path.join(root, d, "SKILL.md")))
def skill_content(self, name: str) -> Optional[str]:
if name not in self.list_skills():
return None
return self.ws.read_file(os.path.join("skills", name, "SKILL.md"))
def purpose(self, name: str) -> Optional[str]:
if name not in self.list_skills():
return None
return self.ws.read_file(os.path.join("skills", name, "PURPOSE.md"))
def active_skills_context(self) -> str:
"""Full SKILL.md contents injected into the Inference Agent prompt (§3.2.1)."""
parts = []
for name in self.list_skills():
parts.append(f"## Skill: {name}\n\n{self.skill_content(name)}")
return ("\n\n---\n\n".join(parts) if parts
else "(No active skills.)")
# ---------- proposals (§3.2.3: atomic, single-skill) ----------
def apply_proposal(self, proposal: dict) -> dict:
"""Apply P_k: create a new skill or patch an existing SKILL.md."""
action = proposal.get("action")
name = str(proposal.get("skill", "")).strip()
if not SKILL_NAME_RE.match(name):
raise ValueError(f"invalid skill name: {name!r}")
rel_skill = os.path.join("skills", name, "SKILL.md")
rel_purpose = os.path.join("skills", name, "PURPOSE.md")
purpose = str(proposal.get("purpose", "")).rstrip() + "\n"
if action == "create":
if self.skill_content(name) is not None:
raise ValueError(f"skill already exists: {name}")
self.ws.write_file(rel_skill, str(proposal.get("content", "")))
self.ws.write_file(rel_purpose, purpose)
return {"skill": name, "action": "create"}
if action == "patch":
old = self.skill_content(name)
if old is None:
raise ValueError(f"cannot patch missing skill: {name}")
new = apply_edits(old, proposal.get("edits", []))
self.ws.write_file(rel_skill, new)
prev = self.purpose(name) or ""
self.ws.write_file(rel_purpose,
prev.rstrip() + "\n" + purpose if prev.strip() else purpose)
return {"skill": name, "action": "patch",
"diff": "\n".join(difflib.unified_diff(
old.splitlines(), new.splitlines(),
fromfile=f"{name}/SKILL.md (old)",
tofile=f"{name}/SKILL.md (new)", lineterm=""))}
raise ValueError(f"unknown proposal action: {action!r}")
# ---------- gating rollback (§3.2.4): skills only, wiki untouched ----------
def snapshot(self) -> Dict[str, str]:
return {name: self.skill_content(name) or "" for name in self.list_skills()}
def restore(self, snap: Dict[str, str]) -> None:
for name in self.list_skills():
if name not in snap:
d = os.path.join(self.ws.root, "skills", name)
for fn in os.listdir(d):
os.remove(os.path.join(d, fn))
os.rmdir(d)
for name, content in snap.items():
self.ws.write_file(os.path.join("skills", name, "SKILL.md"), content)
src/wikiskill/workspace.py
+152
View File
@@ -0,0 +1,152 @@
"""Three-layer workspace (paper §3.1): raw/ (immutable traces), wiki/ (persistent
knowledge), skills/ (active skill set), plus the patch engine used by agents."""
from __future__ import annotations
import difflib
import json
import os
import re
from typing import Dict, List, Optional
PATTERN_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+\.md$")
class PatchError(Exception):
pass
def apply_edits(content: str, edits: List[dict]) -> str:
"""Patch engine (paper Wiki Maintainer prompt): append / replace / insert_after.
`replace` and `insert_after` targets must be EXACT substrings. Invalid targets
raise PatchError; callers may catch and skip individual ops.
"""
for e in edits or []:
op = e.get("op")
if op == "append":
content = content + e.get("content", "")
elif op == "replace":
target = e.get("target", "")
if target not in content:
raise PatchError(f"replace target not found: {target[:60]!r}")
content = content.replace(target, e.get("content", ""), 1)
elif op == "insert_after":
target = e.get("target", "")
if target not in content:
raise PatchError(f"insert_after target not found: {target[:60]!r}")
idx = content.index(target) + len(target)
content = content[:idx] + e.get("content", "") + content[idx:]
else:
raise PatchError(f"unknown op: {op}")
return content
class Workspace:
"""Manages the three layers of the WikiSkill agent workspace."""
def __init__(self, root: str):
self.root = os.path.abspath(root)
for d in ("raw", os.path.join("wiki", "patterns"), "skills"):
os.makedirs(os.path.join(self.root, d), exist_ok=True)
# ---------- generic helpers ----------
def _abs(self, rel: str) -> str:
p = os.path.abspath(os.path.join(self.root, rel))
if not p.startswith(self.root + os.sep) and p != self.root:
raise ValueError(f"path escapes workspace: {rel}")
return p
def read_file(self, rel: str) -> str:
"""Workspace-scoped read (used by the ReAct Skill Proposer)."""
p = self._abs(rel)
if not os.path.isfile(p):
return f"ERROR: file not found: {rel}"
with open(p, "r", encoding="utf-8") as f:
return f.read()
def write_file(self, rel: str, content: str, overwrite: bool = True) -> None:
p = self._abs(rel)
os.makedirs(os.path.dirname(p), exist_ok=True)
if os.path.exists(p) and not overwrite:
raise FileExistsError(rel)
with open(p, "w", encoding="utf-8") as f:
f.write(content)
# ---------- Raw layer: immutable ----------
def add_raw_trace(self, iteration: int, task_id: str, trace: dict) -> str:
rel = os.path.join("raw", f"iter_{iteration:03d}", f"{task_id}.json")
p = self._abs(rel)
if os.path.exists(p):
raise FileExistsError(f"raw layer is immutable: {rel}")
os.makedirs(os.path.dirname(p), exist_ok=True)
with open(p, "w", encoding="utf-8") as f:
json.dump(trace, f, indent=2, ensure_ascii=False)
return rel
# ---------- Wiki layer ----------
def wiki_index(self) -> str:
return self.read_file("wiki/index.md") if os.path.isfile(
self._abs("wiki/index.md")) else ""
def wiki_logs(self) -> str:
return self.read_file("wiki/logs.md") if os.path.isfile(
self._abs("wiki/logs.md")) else ""
def skill_impact(self) -> str:
return self.read_file("wiki/skill-impact.md") if os.path.isfile(
self._abs("wiki/skill-impact.md")) else ""
def list_patterns(self) -> List[str]:
d = self._abs(os.path.join("wiki", "patterns"))
return sorted(f for f in os.listdir(d) if f.endswith(".md"))
def pattern_page(self, name: str) -> str:
if not PATTERN_NAME_RE.match(name):
raise ValueError(f"bad pattern name: {name}")
return self.read_file(os.path.join("wiki", "patterns", name))
def wiki_context(self) -> str:
"""Full wiki context handed to the Wiki Maintainer (index + pages + logs)."""
parts = ["# WIKI INDEX", self.wiki_index() or "(empty)"]
for name in self.list_patterns():
parts += [f"\n# PATTERN: {name}", self.pattern_page(name)]
parts += ["\n# EVOLUTION LOG", self.wiki_logs() or "(empty)"]
return "\n".join(parts)
def apply_maintainer_update(self, update: dict) -> dict:
"""Apply M_WM's JSON ops (create/update patterns, index, log). Returns report."""
report = {"created": [], "updated": [], "errors": []}
for pat in update.get("create_patterns", []) or []:
name = pat.get("name", "")
if not PATTERN_NAME_RE.match(name):
report["errors"].append(f"bad pattern name {name!r}")
continue
rel = os.path.join("wiki", "patterns", name)
if os.path.exists(self._abs(rel)):
report["errors"].append(f"pattern exists, use update: {name}")
continue
self.write_file(rel, pat.get("content", ""))
report["created"].append(name)
for pat in update.get("update_patterns", []) or []:
name = pat.get("name", "")
rel = os.path.join("wiki", "patterns", name)
try:
if not PATTERN_NAME_RE.match(name) or not os.path.exists(self._abs(rel)):
raise PatchError(f"unknown pattern {name!r}")
content = self.pattern_page(name)
self.write_file(rel, apply_edits(content, pat.get("edits", [])))
report["updated"].append(name)
except PatchError as e:
report["errors"].append(f"{name}: {e}")
if update.get("update_index") is not None:
self.write_file("wiki/index.md", str(update["update_index"]))
if update.get("append_log"):
with open(self._abs("wiki/logs.md"), "a", encoding="utf-8") as f:
f.write(update["append_log"].rstrip() + "\n")
return report
def record_skill_impact(self, entry: str) -> None:
"""Programmatic skill-impact.md append — done ONLY by the harness after gating."""
with open(self._abs("wiki/skill-impact.md"), "a", encoding="utf-8") as f:
f.write(entry.rstrip() + "\n")
tests/test_agents.py
+69
View File
@@ -0,0 +1,69 @@
from wikiskill.datasets import make_demo_dataset, _demo_tools
from wikiskill.llm import MockLLM, extract_json
from wikiskill.skills import SkillsLayer
from wikiskill.workspace import Workspace
def _ws(tmp_path):
return Workspace(str(tmp_path / "ws"))
def test_extract_json():
assert extract_json('```json\n{"a": 1}\n```') == {"a": 1}
assert extract_json('noise {"a": {"b": 2}} tail') == {"a": {"b": 2}}
assert extract_json("no json") is None
def test_skills_create_snapshot_restore(tmp_path):
skills = SkillsLayer(_ws(tmp_path))
skills.apply_proposal({"action": "create", "skill": "s1",
"content": "# S1\nline1\nline2\n", "purpose": "why"})
assert skills.list_skills() == ["s1"]
snap = skills.snapshot()
skills.apply_proposal({"action": "patch", "skill": "s1",
"edits": [{"op": "replace", "target": "line2",
"content": "LINE2"}], "purpose": "why2"})
assert "LINE2" in skills.skill_content("s1")
skills.restore(snap)
assert "LINE2" not in skills.skill_content("s1")
skills.apply_proposal({"action": "create", "skill": "s2", "content": "x",
"purpose": "p"})
skills.restore(snap)
assert skills.list_skills() == ["s1"]
def test_skill_injection_context(tmp_path):
skills = SkillsLayer(_ws(tmp_path))
assert "(No active skills.)" in skills.active_skills_context()
skills.apply_proposal({"action": "create", "skill": "s", "content": "BODY",
"purpose": "p"})
assert "BODY" in skills.active_skills_context()
def test_mock_inference_without_and_with_skill(tmp_path):
from wikiskill.agents import InferenceAgent
ws = _ws(tmp_path)
skills = SkillsLayer(ws)
agent = InferenceAgent(MockLLM(), _demo_tools(), ws, skills)
ds = make_demo_dataset()
traj = agent.run(ds.val[0])
assert traj.correct is False # truncation error without skill
skills.apply_proposal({"action": "create", "skill": "unit-conversion",
"content": "Round ONLY the final result to 2 decimal places.",
"purpose": "p"})
agent2 = InferenceAgent(MockLLM(), _demo_tools(), ws, skills)
traj2 = agent2.run(ds.val[0])
assert traj2.correct is True
def test_stratified_sample():
from wikiskill.agents import stratified_sample
from wikiskill.datasets import Trajectory
def t(i, ok):
return Trajectory(task_id=i, messages=[{"role": "user", "content": "x"}],
tool_calls=[], final_answer="a", correct=ok)
traces = [t(f"f{i}", False) for i in range(9)] + [t(f"p{i}", True) for i in range(5)]
s = stratified_sample(traces)
assert sum(1 for x in s if not x.correct) == 5
assert sum(1 for x in s if x.correct) == 3
tests/test_e2e.py
+58
View File
@@ -0,0 +1,58 @@
import json
from wikiskill.datasets import make_demo_dataset, _demo_tools
from wikiskill.llm import MockLLM
from wikiskill.skills import SkillsLayer
from wikiskill.workspace import Workspace
def test_gating_math():
"""Strict improvement: accept iff score > R_best (paper eq. 4)."""
def gate(score, r_best):
return score > r_best
assert gate(0.5, 0.5) is False # ties are rejected
assert gate(0.6, 0.5) is True
assert gate(0.4, 0.5) is False
def test_e2e_evolution_loop(tmp_path, capsys):
from wikiskill.orchestrator import EvolutionOrchestrator
root = tmp_path / "ws"
ws = Workspace(str(root))
orch = EvolutionOrchestrator(llm=MockLLM(), dataset=make_demo_dataset(), ws=ws,
tools=_demo_tools(), K=3)
result = orch.run()
# Baseline (no skill) fails on val; after skill creation, val should improve.
assert result["r_best"] > 0.0
assert "unit-conversion" in result["final_skills"]
# Three-layer artifacts exist
assert (root / "raw" / "iter_000").is_dir() # baseline val rollouts
assert (root / "raw" / "iter_001" / "train").exists() or \
list((root / "raw" / "iter_001").glob("*.json"))
assert (root / "wiki" / "patterns" / "rounding-errors.md").is_file()
assert (root / "wiki" / "index.md").is_file()
assert (root / "wiki" / "logs.md").is_file()
assert (root / "wiki" / "skill-impact.md").is_file()
assert (root / "skills" / "unit-conversion" / "SKILL.md").is_file()
assert (root / "skills" / "unit-conversion" / "PURPOSE.md").is_file()
# Harness wrote the audit trail with outcome + diff
impact = (root / "wiki" / "skill-impact.md").read_text(encoding="utf-8")
assert "Accepted" in impact
# run_state.json persisted
state = json.loads((root / "run_state.json").read_text(encoding="utf-8"))
assert state["final_skills"] == ["unit-conversion"]
def test_e2e_early_stop_at_perfect_val(tmp_path):
from wikiskill.orchestrator import EvolutionOrchestrator
ws = Workspace(str(tmp_path / "ws"))
orch = EvolutionOrchestrator(llm=MockLLM(), dataset=make_demo_dataset(), ws=ws,
tools=_demo_tools(), K=5)
result = orch.run()
# With the skill accepted, val = 1.0 -> the loop must early-stop.
assert result["r_best"] == 1.0
assert len(result["history"]) <= 2
tests/test_metrics.py
+13
View File
@@ -0,0 +1,13 @@
from wikiskill.metrics import accuracy, paired_bootstrap
def test_accuracy():
assert accuracy([True, False, True, True]) == 0.75
assert accuracy([]) == 0.0
def test_paired_bootstrap_significant_and_not():
a = [True] * 10 + [False] * 0
b = [True] * 5 + [False] * 5
assert paired_bootstrap(a, b, n=500, seed=1) < 0.05
assert paired_bootstrap(a, a, n=200, seed=1) == 1.0
tests/test_workspace.py
+46
View File
@@ -0,0 +1,46 @@
import pytest
from wikiskill.workspace import PatchError, Workspace, apply_edits
def test_patch_ops():
c = "alpha\nbeta\n"
assert apply_edits(c, [{"op": "append", "content": "gamma"}]) == "alpha\nbeta\ngamma"
assert apply_edits(c, [{"op": "replace", "target": "beta", "content": "BETA"}]) \
== "alpha\nBETA\n"
assert apply_edits(c, [{"op": "insert_after", "target": "alpha", "content": " mid"}]) \
== "alpha mid\nbeta\n"
with pytest.raises(PatchError):
apply_edits(c, [{"op": "replace", "target": "nope", "content": "x"}])
with pytest.raises(PatchError):
apply_edits(c, [{"op": "bogus", "content": "x"}])
def test_raw_layer_immutable(tmp_path):
ws = Workspace(str(tmp_path / "ws"))
ws.add_raw_trace(1, "t1", {"a": 1})
with pytest.raises(FileExistsError):
ws.add_raw_trace(1, "t1", {"a": 2})
def test_read_file_scoped(tmp_path):
ws = Workspace(str(tmp_path / "ws"))
assert "ERROR" in ws.read_file("missing.md")
with pytest.raises(ValueError):
ws.read_file("../outside.txt")
def test_maintainer_update(tmp_path):
ws = Workspace(str(tmp_path / "ws"))
report = ws.apply_maintainer_update({
"create_patterns": [{"name": "p.md", "content": "# P\nbody\n"}],
"update_patterns": [{"name": "p.md",
"edits": [{"op": "append", "content": "more\n"}]}],
"update_index": "- [p](wiki/patterns/p.md): x\n",
"append_log": "iter 1 findings",
})
assert report["created"] == ["p.md"] and report["updated"] == ["p.md"]
assert ws.pattern_page("p.md").endswith("more\n")
assert "[p]" in ws.wiki_index() and "iter 1" in ws.wiki_logs()
ws.record_skill_impact("## entry\n- outcome: Accepted\n")
assert "Accepted" in ws.skill_impact()