Building Dynamic Workflows: A New Agentic System
Read this story for free: link
Most multi-agent systems work in one of two ways. A lead agent calls helpers one at a time and waits on every report, or an engineer hard-codes a pipeline that runs the same steps for every job. A newer approach, dynamic workflow introduced by Anthropic, lets the lead write the whole plan once as a short program, and plain code runs it with many workers in parallel.
Here is the architecture we are going to build, one component at a time:
- A benchmark: 72 planted bugs and a scorer that checks findings against a hidden key.
- An answer ledger: caching, replay and crash recovery.
- A runtime: workers that run in parallel phases under one budget.
- An audit kit: scan workers, and skeptics that vote on what they find.
- A plan language: restricted Python, checked before it can spend.
- The lead: it reads a 500-token fact sheet instead of the code.
- An evaluation: five designs and an ablation.
By the end, a lead that reads only that sheet finds 0.45 of the planted bugs on average, against 0.30 for a hand-written plan and 0.02 for one agent alone. The full code is in my GitHub repository, dynamic-workflow-agents.
These are the files we are going to write:
dynamic-workflows/
├── build/
│ ├── config.py # model roles, prices, paths
│ ├── bench.py # the hidden key and the scorer
│ ├── shards.py # code packed into shards
│ ├── llm.py # replies, ledger, front door, budget
│ ├── runtime.py # threads, parallel, background runs
│ ├── kit.py # scan, check, vote and report
│ ├── plans.py # whitelist, dry run, approval gate
│ ├── lead.py # fact sheet in, program out
│ └── prompts.json # the instructions each role reads
└── data/algorithms/ # the library, the bugs and the keydynamic-workflows/
├── build/
│ ├── config.py # model roles, prices, paths
│ ├── bench.py # the hidden key and the scorer
│ ├── shards.py # code packed into shards
│ ├── llm.py # replies, ledger, front door, budget
│ ├── runtime.py # threads, parallel, background runs
│ ├── kit.py # scan, check, vote and report
│ ├── plans.py # whitelist, dry run, approval gate
│ ├── lead.py # fact sheet in, program out
│ └── prompts.json # the instructions each role reads
└── data/algorithms/ # the library, the bugs and the keyWhy Multi-Agent Systems Often Fail
Most failures come from two limits. One agent can hold only so much text, and on a large job it forgets what it read or stops early. A lead that hands out tasks one by one becomes the bottleneck, and its context grows with every report.
Our audit shows both: 849 Python files, about 1.35 million tokens, with 72 bugs I planted myself. One agent alone found 0, 1 and 4 of them in three runs and stopped on its own each time. A lead with helpers covered all 14 shards but found only 12 bugs.
Designing Dynamic Workflows
Every job goes through six steps:
- Scout: plain code reads the job and writes a one-page fact sheet.
- Plan: the lead writes a short Python program from that sheet.
- Check: a whitelist refuses whatever a plan may not do.
- Rehearse: a free stand-in model runs the plan once to price it.
- Approve: a gate lets the plan go or holds it for a person.
- Run: the runtime executes the plan in the background under one budget.
Before the comparison runs I wrote down eleven predictions. The evaluation checks five of them:
- H1: the workflow beats one agent alone.
- H1b: it also beats one agent given the same money.
- H5: a second round of scanning adds bugs.
- H4: skeptics raise precision by at least 0.15 and cost at most 0.05 recall.
- H6: running a finding’s failing call makes the skeptic more accurate.
Environment Setup
We start from an empty project with one package, jsonschema, and any OpenAI-compatible endpoint.
uv init dynamic-workflows --python 3.12 && cd dynamic-workflows
uv add jsonschema
export LLM_BASE_URL="https://your-endpoint/v1" # any OpenAI-compatible server
export LLM_API_KEY="your key" # read once, never printeduv init dynamic-workflows --python 3.12 && cd dynamic-workflows
uv add jsonschema
export LLM_BASE_URL="https://your-endpoint/v1" # any OpenAI-compatible server
export LLM_API_KEY="your key" # read once, never printedMODELS gives each role its model, and PRICE` is the cost of one call in credits.
"""config.py: the settings every other file reads."""
import json, os
from pathlib import Path
CAP_CREDITS = 1500.0 # the cap of a front door given no budget of its own
BASE_URL = os.environ.get("LLM_BASE_URL", "http://127.0.0.1:8000/v1") # any OpenAI-compatible endpoint
API_KEY = os.environ.get("LLM_API_KEY", "") # read once from the environment, never printed
MODELS = {"lead": "Qwen/Qwen3.8-Max", "worker": "Qwen/Qwen3.8-27B", # the code asks for a role, not a model
"judge": "Qwen/Qwen3.7-Plus", "fast": "Qwen/Qwen3.8-Flash-Next"}
PRICE = {"Qwen/Qwen3.8-Max": 0.5, "Qwen/Qwen3.8-27B": 0.2, "Qwen/Qwen3.7-Plus": 0.2, "Qwen/Qwen3.8-Flash-Next": 0.2} # credits per call, whatever its length
TEMPERATURE, TOP_P = 0.6, 0.95 # reasoning models loop under greedy decoding, so we sample
SHARD_TOKENS, CHARS_PER_TOKEN = 100_000, 2.54542 # code per worker call, and characters per token (both measured later)
MAX_AGENTS_PER_RUN = 1000 # a backstop against a plan that loops forever
DATA = Path("data/algorithms") # the library, its planted bugs and the answer key
PROMPTS = json.loads((Path(__file__).parent / "prompts.json").read_text(encoding="utf-8")) # the instructions, one text per role"""config.py: the settings every other file reads."""
import json, os
from pathlib import Path
CAP_CREDITS = 1500.0 # the cap of a front door given no budget of its own
BASE_URL = os.environ.get("LLM_BASE_URL", "http://127.0.0.1:8000/v1") # any OpenAI-compatible endpoint
API_KEY = os.environ.get("LLM_API_KEY", "") # read once from the environment, never printed
MODELS = {"lead": "Qwen/Qwen3.8-Max", "worker": "Qwen/Qwen3.8-27B", # the code asks for a role, not a model
"judge": "Qwen/Qwen3.7-Plus", "fast": "Qwen/Qwen3.8-Flash-Next"}
PRICE = {"Qwen/Qwen3.8-Max": 0.5, "Qwen/Qwen3.8-27B": 0.2, "Qwen/Qwen3.7-Plus": 0.2, "Qwen/Qwen3.8-Flash-Next": 0.2} # credits per call, whatever its length
TEMPERATURE, TOP_P = 0.6, 0.95 # reasoning models loop under greedy decoding, so we sample
SHARD_TOKENS, CHARS_PER_TOKEN = 100_000, 2.54542 # code per worker call, and characters per token (both measured later)
MAX_AGENTS_PER_RUN = 1000 # a backstop against a plan that loops forever
DATA = Path("data/algorithms") # the library, its planted bugs and the answer key
PROMPTS = json.loads((Path(__file__).parent / "prompts.json").read_text(encoding="utf-8")) # the instructions, one text per rolePlanted-Bug Benchmark
I took a public library of 849 Python files and planted one bug in each of 72 files. Each bug makes at least one of the file’s doctest examples fail, and never more than 40% of them. The copy that agents read has those examples removed.
left out:the twelve kinds of one-line edit, the doctest proof and the leak test (the repository’s notebook has them)
"""bench.py: the files, the hidden key and the scorer."""
import ast, functools, json, re
from config import DATA
AUDITED = DATA / "audited" # the copy of the library that agents read
WORK = sorted(p.relative_to(AUDITED).as_posix() for p in AUDITED.rglob("*.py")) # every file an audit must read
KEY = json.loads((DATA / "hidden" / "answer_key.json").read_text(encoding="utf-8")) # the 72 planted bugs, in a folder no agent tool opens"""bench.py: the files, the hidden key and the scorer."""
import ast, functools, json, re
from config import DATA
AUDITED = DATA / "audited" # the copy of the library that agents read
WORK = sorted(p.relative_to(AUDITED).as_posix() for p in AUDITED.rglob("*.py")) # every file an audit must read
KEY = json.loads((DATA / "hidden" / "answer_key.json").read_text(encoding="utf-8")) # the 72 planted bugs, in a folder no agent tool opensThe scorer must know which function a line belongs to.
<button data-code="# bench.py
@functools.lru_cache(maxsize=None)
def audited_tree(path):
# parsed once, then reused
return ast.parse((AUDITED / path).read_text(encoding="utf-8"))
def function_at(path, line):
"""(name, first line, last line) of the innermost function around a line, or None."""
best = None
def walk(node, prefix):
nonlocal best
for child in ast.iter_child_nodes(node):
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
# the innermost function holding the line wins
if child.lineno <= line = best[1]):
best = (prefix + child.name, child.lineno, child.end_lineno)
walk(child, prefix + child.name + ".")
else: # methods get their class name
walk(child, prefix + child.name + "." if isinstance(child, ast.ClassDef) else prefix)
if path in WORK: # a path a model made up is never opened
walk(audited_tree(path), "")
return best” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
# bench.py
@functools.lru_cache(maxsize=None)
def audited_tree(path):
# parsed once, then reused
return ast.parse((AUDITED / path).read_text(encoding="utf-8"))
def function_at(path, line):
"""(name, first line, last line) of the innermost function around a line, or None."""
best = None
def walk(node, prefix):
nonlocal best
for child in ast.iter_child_nodes(node):
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
# the innermost function holding the line wins
if child.lineno <= line <= child.end_lineno and (best is None or child.lineno >= best[1]):
best = (prefix + child.name, child.lineno, child.end_lineno)
walk(child, prefix + child.name + ".")
else: # methods get their class name
walk(child, prefix + child.name + "." if isinstance(child, ast.ClassDef) else prefix)
if path in WORK: # a path a model made up is never opened
walk(audited_tree(path), "")
return best# bench.py
@functools.lru_cache(maxsize=None)
def audited_tree(path):
# parsed once, then reused
return ast.parse((AUDITED / path).read_text(encoding="utf-8"))
def function_at(path, line):
"""(name, first line, last line) of the innermost function around a line, or None."""
best = None
def walk(node, prefix):
nonlocal best
for child in ast.iter_child_nodes(node):
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
# the innermost function holding the line wins
if child.lineno <= line <= child.end_lineno and (best is None or child.lineno >= best[1]):
best = (prefix + child.name, child.lineno, child.end_lineno)
walk(child, prefix + child.name + ".")
else: # methods get their class name
walk(child, prefix + child.name + "." if isinstance(child, ast.ClassDef) else prefix)
if path in WORK: # a path a model made up is never opened
walk(audited_tree(path), "")
return bestA finding counts when it names the right file and function and lands within five lines of the bug:
match applies it to a list of findings.
0 else None
said = f.get("function", "").split(".")[-1] # the name the model gave, as a second chance
hits = [b for b in key if b["path"] == path and abs(line – b["line"])
# bench.py
def normalise_path(path):
# ./x, /x and .../audited/x all mean x
return re.sub(r"^(./|/|.*audited/)+", "", re.sub(r"/+", "/", str(path).strip().replace("\", "/")))
def match(findings, key=KEY, window=5):
"""A finding counts when it names the same file and function as a planted bug, within `window` lines."""
found, wrong = {}, set()
for f in findings:
path, line = normalise_path(f["file"]), int(f.get("line") or 0)
span = function_at(path, line) if line > 0 else None
said = f.get("function", "").split(".")[-1] # the name the model gave, as a second chance
hits = [b for b in key if b["path"] == path and abs(line - b["line"]) <= window
and ((span and span[0] == b["function"]) or (said and said == b["function"].split(".")[-1]))]
if hits:
found.setdefault(min(hits, key=lambda b: abs(line - b["line"]))["id"], f) # the nearest bug wins, and saying it twice counts once
else:
wrong.add((path, span[0] if span else f"line {line // 10}")) # one wrong idea per place, however often repeated
# recall over the key, precision over what was reported
tp, fp = len(found), len(wrong)
return {"found": sorted(found), "recall": tp / len(key), "precision": tp / (tp + fp) if tp + fp else None, "tp": tp, "fp": fp}# bench.py
def normalise_path(path):
# ./x, /x and .../audited/x all mean x
return re.sub(r"^(./|/|.*audited/)+", "", re.sub(r"/+", "/", str(path).strip().replace("\", "/")))
def match(findings, key=KEY, window=5):
"""A finding counts when it names the same file and function as a planted bug, within `window` lines."""
found, wrong = {}, set()
for f in findings:
path, line = normalise_path(f["file"]), int(f.get("line") or 0)
span = function_at(path, line) if line > 0 else None
said = f.get("function", "").split(".")[-1] # the name the model gave, as a second chance
hits = [b for b in key if b["path"] == path and abs(line - b["line"]) <= window
and ((span and span[0] == b["function"]) or (said and said == b["function"].split(".")[-1]))]
if hits:
found.setdefault(min(hits, key=lambda b: abs(line - b["line"]))["id"], f) # the nearest bug wins, and saying it twice counts once
else:
wrong.add((path, span[0] if span else f"line {line // 10}")) # one wrong idea per place, however often repeated
# recall over the key, precision over what was reported
tp, fp = len(found), len(wrong)
return {"found": sorted(found), "recall": tp / len(key), "precision": tp / (tp + fp) if tp + fp else None, "tp": tp, "fp": fp}Answer Ledger
I rerun model calls constantly. I store each reply under a fingerprint of its question and never pay for the same question twice.
A Reply holds one answer.
"""llm.py: replies, the ledger, one front door and one budget."""
import hashlib, json, re, sqlite3, threading, time, urllib.request, zlib
from dataclasses import asdict, dataclass
from jsonschema import Draft202012Validator
from config import API_KEY, BASE_URL, CAP_CREDITS, PRICE
@dataclass
class Reply:
text: str = "" # the answer
finish: str = None # stop, length or error
prompt_tokens: int = 0 # as the server counted them
completion_tokens: int = 0
seconds: float = 0.0
model: str = ""
cached: bool = False # True when the ledger answered instead of the network
class CacheMiss(RuntimeError):
"""Replay met a question the ledger has never seen."""
class SpendRefused(RuntimeError):
"""A paid call was refused: there is no client to pay with.""""""llm.py: replies, the ledger, one front door and one budget."""
import hashlib, json, re, sqlite3, threading, time, urllib.request, zlib
from dataclasses import asdict, dataclass
from jsonschema import Draft202012Validator
from config import API_KEY, BASE_URL, CAP_CREDITS, PRICE
@dataclass
class Reply:
text: str = "" # the answer
finish: str = None # stop, length or error
prompt_tokens: int = 0 # as the server counted them
completion_tokens: int = 0
seconds: float = 0.0
model: str = ""
cached: bool = False # True when the ledger answered instead of the network
class CacheMiss(RuntimeError):
"""Replay met a question the ledger has never seen."""
class SpendRefused(RuntimeError):
"""A paid call was refused: there is no client to pay with."""http_client is the only code that touches the network.
# llm.py
def http_client(model, messages, **params):
"""One chat completion over plain HTTP. A failed call raises, so it is never stored."""
body = json.dumps({"model": model, "messages": messages, **params}).encode() # the question in chat completions form
head = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"} # the key travels in a header only
start = time.perf_counter()
# one POST, ten minutes at most
with urllib.request.urlopen(urllib.request.Request(BASE_URL + "/chat/completions", body, head), timeout=600) as r:
data = json.load(r)
choice, use = data["choices"][0], data["usage"] # the answer and the server's token counts
return Reply(choice["message"].get("content") or "", choice["finish_reason"], use["prompt_tokens"], use["completion_tokens"],
time.perf_counter() - start, model)# llm.py
def http_client(model, messages, **params):
"""One chat completion over plain HTTP. A failed call raises, so it is never stored."""
body = json.dumps({"model": model, "messages": messages, **params}).encode() # the question in chat completions form
head = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"} # the key travels in a header only
start = time.perf_counter()
# one POST, ten minutes at most
with urllib.request.urlopen(urllib.request.Request(BASE_URL + "/chat/completions", body, head), timeout=600) as r:
data = json.load(r)
choice, use = data["choices"][0], data["usage"] # the answer and the server's token counts
return Reply(choice["message"].get("content") or "", choice["finish_reason"], use["prompt_tokens"], use["completion_tokens"],
time.perf_counter() - start, model)The fingerprint includes a sample index: samples 0 and 1 of one question get independent answers.
# llm.py
class Ledger:
"""Every answer, under a fingerprint of its question."""
def __init__(self, path):
# one row per answer, keyed by fingerprint
self.db, self.lock = sqlite3.connect(str(path), check_same_thread=False), threading.Lock()
self.db.execute("CREATE TABLE IF NOT EXISTS calls (key TEXT PRIMARY KEY, role, reply BLOB, seconds REAL, credits REAL, created REAL)")
@staticmethod
def key(model, messages, params, sample=0):
blob = json.dumps([model, messages, params, sample], sort_keys=True) # the question, exactly as it was asked
return hashlib.sha256(blob.encode()).hexdigest()
def get(self, key):
with self.lock:
row = self.db.execute("SELECT reply FROM calls WHERE key = ?", (key,)).fetchone()
# stored replies come back marked cached
return Reply(**{**json.loads(zlib.decompress(row[0])), "cached": True}) if row else None
def put(self, key, reply, role="", credits=0.0):
with self.lock:
self.db.execute("INSERT OR REPLACE INTO calls VALUES (?,?,?,?,?,?)",
(key, role, zlib.compress(json.dumps(asdict(reply)).encode()), reply.seconds, credits, time.time()))
self.db.commit() # a crash loses at most the calls still in flight# llm.py
class Ledger:
"""Every answer, under a fingerprint of its question."""
def __init__(self, path):
# one row per answer, keyed by fingerprint
self.db, self.lock = sqlite3.connect(str(path), check_same_thread=False), threading.Lock()
self.db.execute("CREATE TABLE IF NOT EXISTS calls (key TEXT PRIMARY KEY, role, reply BLOB, seconds REAL, credits REAL, created REAL)")
@staticmethod
def key(model, messages, params, sample=0):
blob = json.dumps([model, messages, params, sample], sort_keys=True) # the question, exactly as it was asked
return hashlib.sha256(blob.encode()).hexdigest()
def get(self, key):
with self.lock:
row = self.db.execute("SELECT reply FROM calls WHERE key = ?", (key,)).fetchone()
# stored replies come back marked cached
return Reply(**{**json.loads(zlib.decompress(row[0])), "cached": True}) if row else None
def put(self, key, reply, role="", credits=0.0):
with self.lock:
self.db.execute("INSERT OR REPLACE INTO calls VALUES (?,?,?,?,?,?)",
(key, role, zlib.compress(json.dumps(asdict(reply)).encode()), reply.seconds, credits, time.time()))
self.db.commit() # a crash loses at most the calls still in flightEvery call goes through one front door, LLM.
# llm.py
class LLM:
def __init__(self, ledger, mode="collect", client=None, scripted=None, budget=None):
self.ledger, self.mode, self.client, self.scripted = ledger, mode, client, scripted
self.budget = budget or Budget(CAP_CREDITS)# llm.py
class LLM:
def __init__(self, ledger, mode="collect", client=None, scripted=None, budget=None):
self.ledger, self.mode, self.client, self.scripted = ledger, mode, client, scripted
self.budget = budget or Budget(CAP_CREDITS)chat tries, in order: the ledger, replay refusal, a free stand-in, the budget, and only then the network.
# llm.py
def chat(self, model, messages, *, role="", sample=0, **params):
key = Ledger.key(model, messages, params, sample) # the same question always gets the same key
if hit := self.ledger.get(key):
return hit # free, instant and identical to the first answer
if self.mode == "replay":
raise CacheMiss(f"{role or model}: no stored answer, run in collect mode first")
if self.scripted: # a stand-in answers for free and leaves no trace
return self.scripted(model, messages, dict(params, sample=sample))
if self.client is None: # no client means nothing can be paid for
raise SpendRefused("no client, so nothing can be collected")
price = PRICE.get(model, 0.2)
self.budget.admit(price) # may wait here until somebody raises the cap
try:
reply = self.client(model, messages, **params)
except BaseException:
self.budget.settle(price, 0.0) # a failed call is not billed and not stored
raise
self.budget.settle(price, price)
self.ledger.put(key, reply, role, price)
return reply# llm.py
def chat(self, model, messages, *, role="", sample=0, **params):
key = Ledger.key(model, messages, params, sample) # the same question always gets the same key
if hit := self.ledger.get(key):
return hit # free, instant and identical to the first answer
if self.mode == "replay":
raise CacheMiss(f"{role or model}: no stored answer, run in collect mode first")
if self.scripted: # a stand-in answers for free and leaves no trace
return self.scripted(model, messages, dict(params, sample=sample))
if self.client is None: # no client means nothing can be paid for
raise SpendRefused("no client, so nothing can be collected")
price = PRICE.get(model, 0.2)
self.budget.admit(price) # may wait here until somebody raises the cap
try:
reply = self.client(model, messages, **params)
except BaseException:
self.budget.settle(price, 0.0) # a failed call is not billed and not stored
raise
self.budget.settle(price, price)
self.ledger.put(key, reply, role, price)
return replyWe test it with a fake client that counts calls.
<button data-code="import time
from config import MODELS
from llm import LLM, Ledger, Reply
paid = [] # one entry per call that reaches the "network"
def fake_client(model, messages, **params): # a stand-in for the network: it answers after a pause
paid.append(1)
time.sleep(0.05)
return Reply('{"answer": "ok"}', "stop", 10, 5, 0.05, model)
llm = LLM(Ledger(":memory:"), client=fake_client)
ask = lambda door, q: door.chat(MODELS["worker"], [{"role": "user", "content": q}], role="demo")
for q in ("what is 2+2?", "name a sorting algorithm", "what is 2+2?"):
print(f"{q:
import time
from config import MODELS
from llm import LLM, Ledger, Reply
paid = [] # one entry per call that reaches the "network"
def fake_client(model, messages, **params): # a stand-in for the network: it answers after a pause
paid.append(1)
time.sleep(0.05)
return Reply('{"answer": "ok"}', "stop", 10, 5, 0.05, model)
llm = LLM(Ledger(":memory:"), client=fake_client)
ask = lambda door, q: door.chat(MODELS["worker"], [{"role": "user", "content": q}], role="demo")
for q in ("what is 2+2?", "name a sorting algorithm", "what is 2+2?"):
print(f"{q:<26} from the ledger: {ask(llm, q).cached}")
try:
ask(LLM(llm.ledger, mode="replay", client=fake_client), "a new question")
except Exception as e:
print(f"{type(e).__name__}: {e}")
#### OUTPUT ####
what is 2+2? from the ledger: False
name a sorting algorithm from the ledger: False
what is 2+2? from the ledger: True
CacheMiss: demo: no stored answer, run in collect mode firstimport time
from config import MODELS
from llm import LLM, Ledger, Reply
paid = [] # one entry per call that reaches the "network"
def fake_client(model, messages, **params): # a stand-in for the network: it answers after a pause
paid.append(1)
time.sleep(0.05)
return Reply('{"answer": "ok"}', "stop", 10, 5, 0.05, model)
llm = LLM(Ledger(":memory:"), client=fake_client)
ask = lambda door, q: door.chat(MODELS["worker"], [{"role": "user", "content": q}], role="demo")
for q in ("what is 2+2?", "name a sorting algorithm", "what is 2+2?"):
print(f"{q:<26} from the ledger: {ask(llm, q).cached}")
try:
ask(LLM(llm.ledger, mode="replay", client=fake_client), "a new question")
except Exception as e:
print(f"{type(e).__name__}: {e}")
#### OUTPUT ####
what is 2+2? from the ledger: False
name a sorting algorithm from the ledger: False
what is 2+2? from the ledger: True
CacheMiss: demo: no stored answer, run in collect mode firstThe repeat came from the ledger for free, and replay raised CacheMiss without touching the network.
Shard Packer
Big shards look cheap because billing is per call, yet a model can miss details past the start of a long input. A shard is the code that one worker reads in a single call. numbered adds line numbers, and TOKENS sizes each file as a worker sees it.
5}| {ln}" for i, ln in enumerate(lines, 1))
TOKENS = {p: int(len(numbered(p)) / CHARS_PER_TOKEN) for p in WORK} # each file's size as a worker sees it” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
"""shards.py: numbered lines, packed into shards."""
from bench import AUDITED, WORK
from config import CHARS_PER_TOKEN
def numbered(path):
lines = (AUDITED / path).read_text(encoding="utf-8").split("n")
# a path header, then numbered lines
return f"### {path}n" + "n".join(f"{i:>5}| {ln}" for i, ln in enumerate(lines, 1))
TOKENS = {p: int(len(numbered(p)) / CHARS_PER_TOKEN) for p in WORK} # each file's size as a worker sees it"""shards.py: numbered lines, packed into shards."""
from bench import AUDITED, WORK
from config import CHARS_PER_TOKEN
def numbered(path):
lines = (AUDITED / path).read_text(encoding="utf-8").split("n")
# a path header, then numbered lines
return f"### {path}n" + "n".join(f"{i:>5}| {ln}" for i, ln in enumerate(lines, 1))
TOKENS = {p: int(len(numbered(p)) / CHARS_PER_TOKEN) for p in WORK} # each file's size as a worker sees itpack fills each shard with whole files until the next one would not fit.
<button data-code="# shards.py
def pack(paths, target):
"""Whole files packed into shards of about `target` tokens, first fit."""
shards, sizes = [], []
# folder by folder, biggest files first
for folder in sorted({p.split("/")[0] for p in paths}):
for p in sorted((q for q in paths if q.split("/")[0] == folder), key=lambda q: -TOKENS[q]):
# the first shard with room takes the file
for i, size in enumerate(sizes):
if size + TOKENS[p]
# shards.py
def pack(paths, target):
"""Whole files packed into shards of about `target` tokens, first fit."""
shards, sizes = [], []
# folder by folder, biggest files first
for folder in sorted({p.split("/")[0] for p in paths}):
for p in sorted((q for q in paths if q.split("/")[0] == folder), key=lambda q: -TOKENS[q]):
# the first shard with room takes the file
for i, size in enumerate(sizes):
if size + TOKENS[p] <= target:
shards[i].append(p)
sizes[i] += TOKENS[p]
break
else: # no room anywhere: start a new shard
shards.append([p])
sizes.append(TOKENS[p])
return shards# shards.py
def pack(paths, target):
"""Whole files packed into shards of about `target` tokens, first fit."""
shards, sizes = [], []
# folder by folder, biggest files first
for folder in sorted({p.split("/")[0] for p in paths}):
for p in sorted((q for q in paths if q.split("/")[0] == folder), key=lambda q: -TOKENS[q]):
# the first shard with room takes the file
for i, size in enumerate(sizes):
if size + TOKENS[p] <= target:
shards[i].append(p)
sizes[i] += TOKENS[p]
break
else: # no room anywhere: start a new shard
shards.append([p])
sizes.append(TOKENS[p])
return shardsMy rule, fixed in advance, takes the largest size within 0.05 of the best recall, and on 283 files with 24 planted bugs it picked 100,000 tokens. A later whole-system test disagreed: a slim plan found 0.38 of the bugs at 20,000-token shards and 0.18 at 100,000.
Every workflow compared below used 100,000-token shards.
Worker Threads and Event Stream
A run is one execution of a plan, and a plan is a Python function that receives the run.
"""runtime.py: workers, threads, events, phases and background runs."""
import collections, concurrent.futures as cf, threading, time
from dataclasses import dataclass, field
from config import MAX_AGENTS_PER_RUN, MODELS, PRICE, TEMPERATURE, TOP_P
from llm import CacheMiss, SpendRefused, extract_json, validate
class RunCancelled(RuntimeError):
"""The run was cancelled."""
ABORT = (RunCancelled, SpendRefused, CacheMiss) # only these end a run, other failures stay with the worker"""runtime.py: workers, threads, events, phases and background runs."""
import collections, concurrent.futures as cf, threading, time
from dataclasses import dataclass, field
from config import MAX_AGENTS_PER_RUN, MODELS, PRICE, TEMPERATURE, TOP_P
from llm import CacheMiss, SpendRefused, extract_json, validate
class RunCancelled(RuntimeError):
"""The run was cancelled."""
ABORT = (RunCancelled, SpendRefused, CacheMiss) # only these end a run, other failures stay with the workerEach worker leaves a thread, its whole conversation.
# runtime.py
@dataclass
class Thread:
"""One worker's whole conversation, kept so a person can read it afterwards."""
id: str
phase: str
label: str # what the plan called it, such as "scan:logic:maths#3"
model: str
role: str = "worker"
messages: list = field(default_factory=list) # everything the worker was sent
replies: list = field(default_factory=list) # every Reply it got back
status: str = "waiting" # waiting, running, done or failed
started: float = 0.0 # seconds since the run began
ended: float = 0.0
cost: float = 0.0 # credits charged for its paid calls
result: object = None
error: str = None# runtime.py
@dataclass
class Thread:
"""One worker's whole conversation, kept so a person can read it afterwards."""
id: str
phase: str
label: str # what the plan called it, such as "scan:logic:maths#3"
model: str
role: str = "worker"
messages: list = field(default_factory=list) # everything the worker was sent
replies: list = field(default_factory=list) # every Reply it got back
status: str = "waiting" # waiting, running, done or failed
started: float = 0.0 # seconds since the run began
ended: float = 0.0
cost: float = 0.0 # credits charged for its paid calls
result: object = None
error: str = NoneRun keeps everything a run leaves.
# runtime.py
class Run:
def __init__(self, title, goal="", *, llm, run_id=None, concurrency=16, max_agents=MAX_AGENTS_PER_RUN):
self.title, self.goal, self.llm, self.run_id = title, goal, llm, run_id or title.lower().replace(" ", "-")
self.t0, self.max_agents, self.lock = time.perf_counter(), max_agents, threading.Lock()
# slots cap concurrency, cancelled refuses new workers
self.slots, self.cancelled = threading.BoundedSemaphore(concurrency), threading.Event()
# what a person reads afterwards
self.threads, self.phases, self.events, self.dropped, self.current = [], [], [], [], None
self.status, self.result, self.error, self._worker = "created", None, None, None
def elapsed(self):
return time.perf_counter() - self.t0
def emit(self, kind, **data):
self.events.append({"t": round(self.elapsed(), 3), "type": kind, **data}) # the run's stream: every event, in order# runtime.py
class Run:
def __init__(self, title, goal="", *, llm, run_id=None, concurrency=16, max_agents=MAX_AGENTS_PER_RUN):
self.title, self.goal, self.llm, self.run_id = title, goal, llm, run_id or title.lower().replace(" ", "-")
self.t0, self.max_agents, self.lock = time.perf_counter(), max_agents, threading.Lock()
# slots cap concurrency, cancelled refuses new workers
self.slots, self.cancelled = threading.BoundedSemaphore(concurrency), threading.Event()
# what a person reads afterwards
self.threads, self.phases, self.events, self.dropped, self.current = [], [], [], [], None
self.status, self.result, self.error, self._worker = "created", None, None, None
def elapsed(self):
return time.perf_counter() - self.t0
def emit(self, kind, **data):
self.events.append({"t": round(self.elapsed(), 3), "type": kind, **data}) # the run's stream: every event, in orderA phase is a named stretch of the run, such as Scan or Verify.
# runtime.py
def phase(self, title):
with self.lock:
if self.current: # closing a phase records its workers and time
mine = [t for t in self.threads if t.phase == self.current["name"]]
self.emit("run.phase_ended", phase=self.current["name"], agents=len(mine), seconds=round(self.elapsed() - self.current["started"], 2))
self.current = {"name": title, "started": self.elapsed()} if title else None
if self.current:
self.phases.append(self.current)
self.emit("run.phase_started", phase=title, index=len(self.phases))
def log(self, message):
self.emit("run.log", message=str(message)[:500])# runtime.py
def phase(self, title):
with self.lock:
if self.current: # closing a phase records its workers and time
mine = [t for t in self.threads if t.phase == self.current["name"]]
self.emit("run.phase_ended", phase=self.current["name"], agents=len(mine), seconds=round(self.elapsed() - self.current["started"], 2))
self.current = {"name": title, "started": self.elapsed()} if title else None
if self.current:
self.phases.append(self.current)
self.emit("run.phase_started", phase=title, index=len(self.phases))
def log(self, message):
self.emit("run.log", message=str(message)[:500])_new_thread also holds the runaway backstop.
= self.max_agents: # over the limit: record what was refused
self.dropped.append(label)
self.emit("run.dropped", label=label, limit=self.max_agents)
if len(self.dropped) >= self.max_agents: # asking for twice the limit is treated as a runaway
self.log(f"the plan asked for {2 * self.max_agents} workers, twice its limit: cancelled as a runaway")
self.cancelled.set()
return None
t = Thread(f"{self.run_id}/t{len(self.threads) + 1:03d}", phase or (self.current["name"] if self.current else ""),
label or "worker", MODELS.get(model, model), role)
self.threads.append(t)
return t” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
# runtime.py
def _new_thread(self, label, model, phase, role):
"""The next worker's thread, or None when the worker limit is reached."""
with self.lock:
if len(self.threads) >= self.max_agents: # over the limit: record what was refused
self.dropped.append(label)
self.emit("run.dropped", label=label, limit=self.max_agents)
if len(self.dropped) >= self.max_agents: # asking for twice the limit is treated as a runaway
self.log(f"the plan asked for {2 * self.max_agents} workers, twice its limit: cancelled as a runaway")
self.cancelled.set()
return None
t = Thread(f"{self.run_id}/t{len(self.threads) + 1:03d}", phase or (self.current["name"] if self.current else ""),
label or "worker", MODELS.get(model, model), role)
self.threads.append(t)
return t# runtime.py
def _new_thread(self, label, model, phase, role):
"""The next worker's thread, or None when the worker limit is reached."""
with self.lock:
if len(self.threads) >= self.max_agents: # over the limit: record what was refused
self.dropped.append(label)
self.emit("run.dropped", label=label, limit=self.max_agents)
if len(self.dropped) >= self.max_agents: # asking for twice the limit is treated as a runaway
self.log(f"the plan asked for {2 * self.max_agents} workers, twice its limit: cancelled as a runaway")
self.cancelled.set()
return None
t = Thread(f"{self.run_id}/t{len(self.threads) + 1:03d}", phase or (self.current["name"] if self.current else ""),
label or "worker", MODELS.get(model, model), role)
self.threads.append(t)
return tagent() starts one worker and returns a checked value or None.
# runtime.py
def agent(self, prompt="", *, schema=None, model="worker", label="", system="", data=None, sample=0, role="worker", phase=None):
"""One worker: its checked JSON answer, or None. Only an exception that must end the whole run gets out."""
if self.cancelled.is_set():
raise RunCancelled(self.run_id)
t = self._new_thread(label, model, phase, role)
if t is None:
return None
self.emit("run.agent_started", thread=t.id, label=t.label, phase=t.phase)
with self.slots: # at most `concurrency` workers talk at the same time
t.started, t.status = self.elapsed(), "running"
try:
t.result = self._converse(t, system, (prompt + "nn" + (data or "")).strip(), schema, sample)
t.status = "done" if t.result is not None else "failed"
except ABORT:
t.status = "failed"
raise
except Exception as e: # any other failure stays the worker's own
t.status, t.error = "failed", f"{type(e).__name__}: {e}"[:300]
finally:
t.ended = self.elapsed()
self.emit("run.agent_ended", thread=t.id, label=t.label, phase=t.phase, ok=t.status == "done", credits=round(t.cost, 2),
seconds=round(t.ended - t.started, 2), cached=bool(t.replies) and all(r.cached for r in t.replies))
return t.result# runtime.py
def agent(self, prompt="", *, schema=None, model="worker", label="", system="", data=None, sample=0, role="worker", phase=None):
"""One worker: its checked JSON answer, or None. Only an exception that must end the whole run gets out."""
if self.cancelled.is_set():
raise RunCancelled(self.run_id)
t = self._new_thread(label, model, phase, role)
if t is None:
return None
self.emit("run.agent_started", thread=t.id, label=t.label, phase=t.phase)
with self.slots: # at most `concurrency` workers talk at the same time
t.started, t.status = self.elapsed(), "running"
try:
t.result = self._converse(t, system, (prompt + "nn" + (data or "")).strip(), schema, sample)
t.status = "done" if t.result is not None else "failed"
except ABORT:
t.status = "failed"
raise
except Exception as e: # any other failure stays the worker's own
t.status, t.error = "failed", f"{type(e).__name__}: {e}"[:300]
finally:
t.ended = self.elapsed()
self.emit("run.agent_ended", thread=t.id, label=t.label, phase=t.phase, ok=t.status == "done", credits=round(t.cost, 2),
seconds=round(t.ended - t.started, 2), cached=bool(t.replies) and all(r.cached for r in t.replies))
return t.resultFirst we pull the JSON out of a reply.
# llm.py
def extract_json(text):
"""The first JSON object in a reply, even inside a code fence."""
t = re.sub(r"^`{3}[a-zA-Z]*s*|s*`{3}$", "", text.strip()) # `{3} is three backticks
for start in [i for i, c in enumerate(t) if c == "{"]:
try:
return json.JSONDecoder().raw_decode(t[start:])[0]
except ValueError:
continue# llm.py
def extract_json(text):
"""The first JSON object in a reply, even inside a code fence."""
t = re.sub(r"^`{3}[a-zA-Z]*s*|s*`{3}$", "", text.strip()) # `{3} is three backticks
for start in [i for i, c in enumerate(t) if c == "{"]:
try:
return json.JSONDecoder().raw_decode(t[start:])[0]
except ValueError:
continuevalidate turns schema errors into short lines a model can fix.
# llm.py
def validate(obj, schema):
if obj is None:
return ["the reply is not a JSON object"]
# at most eight short problems the model can fix
errors = sorted(Draft202012Validator(schema).iter_errors(obj), key=lambda e: list(e.path))
return [f"{'/'.join(map(str, e.path)) or 'top level'}: {e.message[:120]}" for e in errors[:8]]# llm.py
def validate(obj, schema):
if obj is None:
return ["the reply is not a JSON object"]
# at most eight short problems the model can fix
errors = sorted(Draft202012Validator(schema).iter_errors(obj), key=lambda e: list(e.path))
return [f"{'/'.join(map(str, e.path)) or 'top level'}: {e.message[:120]}" for e in errors[:8]]A worker’s conversation has at most two turns.
Those turns live in _converse.
# runtime.py
def _converse(self, t, system, user, schema, sample):
params = {"temperature": TEMPERATURE, "top_p": TOP_P, "max_tokens": 16000}
if schema is not None:
params["response_format"] = {"type": "json_object"} # JSON mode when the plan wants a schema
if t.model == MODELS["worker"] or t.role == "scan":
params["enable_thinking"] = False # thinking doubled recall but took 50 times as long
t.messages = ([{"role": "system", "content": system}] if system else []) + [{"role": "user", "content": user}]
for turn in range(2): # the first answer, then at most one repair
r = self.llm.chat(t.model, t.messages, role=t.role, sample=sample, **params)
t.replies.append(r)
t.cost += 0 if r.cached else PRICE.get(t.model, 0.2)
obj = r.text if schema is None else extract_json(r.text)
errors = [] if schema is None else validate(obj, schema)
if not errors:
return obj
# the repair turn lists the problems
t.messages += [{"role": "assistant", "content": r.text[:6000]},
{"role": "user", "content": "Your answer does not match the schema:n- " + "n- ".join(errors)
+ "nSend the corrected JSON object only."}]
t.error = "schema: " + " | ".join(errors)[:240]# runtime.py
def _converse(self, t, system, user, schema, sample):
params = {"temperature": TEMPERATURE, "top_p": TOP_P, "max_tokens": 16000}
if schema is not None:
params["response_format"] = {"type": "json_object"} # JSON mode when the plan wants a schema
if t.model == MODELS["worker"] or t.role == "scan":
params["enable_thinking"] = False # thinking doubled recall but took 50 times as long
t.messages = ([{"role": "system", "content": system}] if system else []) + [{"role": "user", "content": user}]
for turn in range(2): # the first answer, then at most one repair
r = self.llm.chat(t.model, t.messages, role=t.role, sample=sample, **params)
t.replies.append(r)
t.cost += 0 if r.cached else PRICE.get(t.model, 0.2)
obj = r.text if schema is None else extract_json(r.text)
errors = [] if schema is None else validate(obj, schema)
if not errors:
return obj
# the repair turn lists the problems
t.messages += [{"role": "assistant", "content": r.text[:6000]},
{"role": "user", "content": "Your answer does not match the schema:n- " + "n- ".join(errors)
+ "nSend the corrected JSON object only."}]
t.error = "schema: " + " | ".join(errors)[:240]run_sync runs a whole plan.
# runtime.py
def run_sync(self, plan):
self.status = "running"
self.emit("run.created", title=self.title)
try:
self.result = plan(self)
self.status = "completed"
except RunCancelled:
self.status = "cancelled"
except Exception as e: # includes the aborts that end a run
self.status, self.error = "failed", f"{type(e).__name__}: {e}"[:300]
finally: # always close the last phase
self.phase(None)
self.emit("run.ended", status=self.status, seconds=round(self.elapsed(), 2), agents=len(self.threads),
credits=round(sum(t.cost for t in self.threads), 2))
return self.result# runtime.py
def run_sync(self, plan):
self.status = "running"
self.emit("run.created", title=self.title)
try:
self.result = plan(self)
self.status = "completed"
except RunCancelled:
self.status = "cancelled"
except Exception as e: # includes the aborts that end a run
self.status, self.error = "failed", f"{type(e).__name__}: {e}"[:300]
finally: # always close the last phase
self.phase(None)
self.emit("run.ended", status=self.status, seconds=round(self.elapsed(), 2), agents=len(self.threads),
credits=round(sum(t.cost for t in self.threads), 2))
return self.resultparallel runs thunks, functions with no arguments, all at once. It is a barrier: it returns when the slowest is done.
In one fixed-plan scan phase, finished workers waited 37 worker-minutes for the slowest.
# runtime.py
def parallel(self, thunks):
thunks = list(thunks)
out, aborts = [None] * len(thunks), []
# one task per thunk, up to 64 threads
with cf.ThreadPoolExecutor(max(1, min(len(thunks), 64))) as pool:
futures = {pool.submit(fn): i for i, fn in enumerate(thunks)}
for f in cf.as_completed(futures):
try:
out[futures[f]] = f.result() # results keep the order of the thunks
except ABORT as e:
aborts.append(e)
self.cancelled.set() # stop the work that has not started yet
except Exception as e:
self.log(f"a parallel task failed: {type(e).__name__}: {e}")
if aborts: # the first abort ends the phase for everyone
raise aborts[0]
return out# runtime.py
def parallel(self, thunks):
thunks = list(thunks)
out, aborts = [None] * len(thunks), []
# one task per thunk, up to 64 threads
with cf.ThreadPoolExecutor(max(1, min(len(thunks), 64))) as pool:
futures = {pool.submit(fn): i for i, fn in enumerate(thunks)}
for f in cf.as_completed(futures):
try:
out[futures[f]] = f.result() # results keep the order of the thunks
except ABORT as e:
aborts.append(e)
self.cancelled.set() # stop the work that has not started yet
except Exception as e:
self.log(f"a parallel task failed: {type(e).__name__}: {e}")
if aborts: # the first abort ends the phase for everyone
raise aborts[0]
return outOne worker on the fake client shows the events a run records:
from runtime import Run
SCHEMA = {"type": "object", "required": ["answer"]}
run = Run("one worker", llm=llm)
run.run_sync(lambda r: (r.phase("Scan"), r.agent("hello", schema=SCHEMA, label="scan#0"))[1])
t = run.threads[0]
print([e["type"][4:] for e in run.events])
print(f"{t.id} | {t.phase} | {t.label} | {t.status} | {t.cost} credits | {len(t.replies)} reply")
#### OUTPUT ####
['created', 'phase_started', 'agent_started', 'agent_ended', 'phase_ended', 'ended']
one-worker/t001 | Scan | scan#0 | done | 0.2 credits | 1 replyfrom runtime import Run
SCHEMA = {"type": "object", "required": ["answer"]}
run = Run("one worker", llm=llm)
run.run_sync(lambda r: (r.phase("Scan"), r.agent("hello", schema=SCHEMA, label="scan#0"))[1])
t = run.threads[0]
print([e["type"][4:] for e in run.events])
print(f"{t.id} | {t.phase} | {t.label} | {t.status} | {t.cost} credits | {len(t.replies)} reply")
#### OUTPUT ####
['created', 'phase_started', 'agent_started', 'agent_ended', 'phase_ended', 'ended']
one-worker/t001 | Scan | scan#0 | done | 0.2 credits | 1 replyA real worker on the library read 100,911 tokens and took 18.7 seconds.
Session Budget and Recovery
All workers draw on one meter. At the cap a worker waits until a person raises it.
admit reserves a call’s price first.
self.cap + 1e-9:
self.paused = True
self.cond.wait(timeout=1.0)
self.paused = False
self.reserved += cost” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
# llm.py
class Budget:
def __init__(self, cap):
# spent is billed, reserved is held for calls in flight
self.cap, self.spent, self.reserved, self.paused = float(cap), 0.0, 0.0, False
self.cond = threading.Condition()
def admit(self, cost):
with self.cond: # reserve the price, or wait here until the cap is raised
while self.spent + self.reserved + cost > self.cap + 1e-9:
self.paused = True
self.cond.wait(timeout=1.0)
self.paused = False
self.reserved += cost# llm.py
class Budget:
def __init__(self, cap):
# spent is billed, reserved is held for calls in flight
self.cap, self.spent, self.reserved, self.paused = float(cap), 0.0, 0.0, False
self.cond = threading.Condition()
def admit(self, cost):
with self.cond: # reserve the price, or wait here until the cap is raised
while self.spent + self.reserved + cost > self.cap + 1e-9:
self.paused = True
self.cond.wait(timeout=1.0)
self.paused = False
self.reserved += costsettle bills it, and raise_to lifts the cap.
# llm.py
def settle(self, reserved, actual):
with self.cond: # a reservation becomes spending, or is released if the call was not billed
self.reserved -= reserved
self.spent += actual
self.cond.notify_all()
def raise_to(self, cap):
with self.cond: # raising the cap wakes every waiting worker
self.cap, self.paused = float(cap), False
self.cond.notify_all()# llm.py
def settle(self, reserved, actual):
with self.cond: # a reservation becomes spending, or is released if the call was not billed
self.reserved -= reserved
self.spent += actual
self.cond.notify_all()
def raise_to(self, cap):
with self.cond: # raising the cap wakes every waiting worker
self.cap, self.paused = float(cap), False
self.cond.notify_all()start runs a long plan on its own thread.
# runtime.py
def start(self, plan):
# the plan runs on its own thread
self._worker = threading.Thread(target=self.run_sync, args=(plan,), daemon=True)
self._worker.start()
return self
def wait(self):
self._worker.join() # block until the plan has finished
return self.result
def cancel(self):
self.cancelled.set() # workers already talking finish, new ones are refused# runtime.py
def start(self, plan):
# the plan runs on its own thread
self._worker = threading.Thread(target=self.run_sync, args=(plan,), daemon=True)
self._worker.start()
return self
def wait(self):
self._worker.join() # block until the plan has finished
return self.result
def cancel(self):
self.cancelled.set() # workers already talking finish, new ones are refusedstatusline answers “how is it going?” from any thread.
# runtime.py
def statusline(self):
"""One line answering 'how is it going?' from any thread."""
n = collections.Counter(t.status for t in self.threads)
return (f"{self.run_id} | {self.status} | phase {len(self.phases)}: {self.current['name'] if self.current else '-'} | "
f"{n['running']} running, {n['done']} done, {n['failed']} failed | {sum(t.cost for t in self.threads):.1f} credits | "
f"{self.elapsed():.1f}s" + (" | PAUSED at the budget" if self.llm.budget.paused else ""))# runtime.py
def statusline(self):
"""One line answering 'how is it going?' from any thread."""
n = collections.Counter(t.status for t in self.threads)
return (f"{self.run_id} | {self.status} | phase {len(self.phases)}: {self.current['name'] if self.current else '-'} | "
f"{n['running']} running, {n['done']} done, {n['failed']} failed | {sum(t.cost for t in self.threads):.1f} credits | "
f"{self.elapsed():.1f}s" + (" | PAUSED at the budget" if self.llm.budget.paused else ""))Forty workers get a cap of 5 credits, enough for 25 calls at 0.2 each.
from llm import Budget
paid.clear()
capped = LLM(Ledger(":memory:"), client=fake_client, budget=Budget(5.0))
run = Run("forty workers", llm=capped, concurrency=10)
run.start(lambda r: r.parallel([lambda i=i: r.agent(f"question {i}", schema=SCHEMA) for i in range(40)]))
time.sleep(1.2)
print(f"requests sent {len(paid)}n{run.statusline()}")
capped.budget.raise_to(10.0)
run.wait()
print(f"cap raised to {capped.budget.cap}: {run.status}, spent {capped.budget.spent:.1f}, requests sent {len(paid)}")
#### OUTPUT ####
requests sent 25
forty-workers | running | phase 0: - | 10 running, 25 done, 0 failed | 5.0 credits | 1.2s | PAUSED at the budget
cap raised to 10.0: completed, spent 8.0, requests sent 40from llm import Budget
paid.clear()
capped = LLM(Ledger(":memory:"), client=fake_client, budget=Budget(5.0))
run = Run("forty workers", llm=capped, concurrency=10)
run.start(lambda r: r.parallel([lambda i=i: r.agent(f"question {i}", schema=SCHEMA) for i in range(40)]))
time.sleep(1.2)
print(f"requests sent {len(paid)}n{run.statusline()}")
capped.budget.raise_to(10.0)
run.wait()
print(f"cap raised to {capped.budget.cap}: {run.status}, spent {capped.budget.spent:.1f}, requests sent {len(paid)}")
#### OUTPUT ####
requests sent 25
forty-workers | running | phase 0: - | 10 running, 25 done, 0 failed | 5.0 credits | 1.2s | PAUSED at the budget
cap raised to 10.0: completed, spent 8.0, requests sent 40The run stopped at exactly 25 requests, and after the cap went up it finished with forty. Next I cancel a two-phase run about halfway, then run the same plan again on the same ledger.
<button data-code="plan = lambda r: (r.parallel([lambda i=i: r.agent(f"scan {i}", schema=SCHEMA, phase="Scan") for i in range(30)]),
r.parallel([lambda i=i: r.agent(f"verify {i}", schema=SCHEMA, model="judge", phase="Verify") for i in range(30)]))
paid.clear()
ledger = Ledger(":memory:")
first = Run("crash", llm=LLM(ledger, client=fake_client), concurrency=8).start(plan)
while len(paid)
plan = lambda r: (r.parallel([lambda i=i: r.agent(f"scan {i}", schema=SCHEMA, phase="Scan") for i in range(30)]),
r.parallel([lambda i=i: r.agent(f"verify {i}", schema=SCHEMA, model="judge", phase="Verify") for i in range(30)]))
paid.clear()
ledger = Ledger(":memory:")
first = Run("crash", llm=LLM(ledger, client=fake_client), concurrency=8).start(plan)
while len(paid) < 25: # let it get about halfway, then pull the plug
time.sleep(0.01)
first.cancel()
first.wait()
before = len(paid)
second = Run("crash", llm=LLM(ledger, client=fake_client), concurrency=8)
second.run_sync(plan)
print(f"first run {first.status} after {before} paid calls, then the second run {second.status} with {len(paid) - before} new calls, {len(paid)} in all")
#### OUTPUT ####
first run cancelled after 30 paid calls, then the second run completed with 30 new calls, 60 in allplan = lambda r: (r.parallel([lambda i=i: r.agent(f"scan {i}", schema=SCHEMA, phase="Scan") for i in range(30)]),
r.parallel([lambda i=i: r.agent(f"verify {i}", schema=SCHEMA, model="judge", phase="Verify") for i in range(30)]))
paid.clear()
ledger = Ledger(":memory:")
first = Run("crash", llm=LLM(ledger, client=fake_client), concurrency=8).start(plan)
while len(paid) < 25: # let it get about halfway, then pull the plug
time.sleep(0.01)
first.cancel()
first.wait()
before = len(paid)
second = Run("crash", llm=LLM(ledger, client=fake_client), concurrency=8)
second.run_sync(plan)
print(f"first run {first.status} after {before} paid calls, then the second run {second.status} with {len(paid) - before} new calls, {len(paid)} in all")
#### OUTPUT ####
first run cancelled after 30 paid calls, then the second run completed with 30 new calls, 60 in allThe plan needs sixty calls whether or not it is interrupted, and no call was paid for twice.
Audit Task Kit
Three kit steps call the model: scan gives a worker one shard and one lens, check gives a skeptic a batch, and report` writes the list.
The skeptic is told that most reports are false alarms and to answer real: false when unsure.
"""kit.py: the words of an audit: scan, check, vote and report."""
import collections
from bench import WORK, function_at, normalise_path
from config import MODELS, PROMPTS, SHARD_TOKENS
from shards import numbered, pack
LENSES = ["boundaries", "logic", "state"] # a lens points a worker at one family of mistakes, "general" at none
FINDINGS = {"type": "object", "required": ["findings"], "properties": {"findings": {"type": "array", "items": {
"type": "object", "required": ["file", "line", "claim"], "properties": {"line": {"type": "integer"}}}}}}
VERDICTS = {"type": "object", "required": ["verdicts"], "properties": {"verdicts": {"type": "array", "items": {
"type": "object", "required": ["report", "real"], "properties": {"real": {"type": "boolean"}}}}}}
REPORT = {"type": "object", "required": ["bugs"], "properties": {"bugs": {"type": "array", "items": {"type": "object", "required": ["ids"]}}}}"""kit.py: the words of an audit: scan, check, vote and report."""
import collections
from bench import WORK, function_at, normalise_path
from config import MODELS, PROMPTS, SHARD_TOKENS
from shards import numbered, pack
LENSES = ["boundaries", "logic", "state"] # a lens points a worker at one family of mistakes, "general" at none
FINDINGS = {"type": "object", "required": ["findings"], "properties": {"findings": {"type": "array", "items": {
"type": "object", "required": ["file", "line", "claim"], "properties": {"line": {"type": "integer"}}}}}}
VERDICTS = {"type": "object", "required": ["verdicts"], "properties": {"verdicts": {"type": "array", "items": {
"type": "object", "required": ["report", "real"], "properties": {"real": {"type": "boolean"}}}}}}
REPORT = {"type": "object", "required": ["bugs"], "properties": {"bugs": {"type": "array", "items": {"type": "object", "required": ["ids"]}}}}batches cuts a list into groups, and flatten` joins lists back into one.
# kit.py
def batches(items, size):
items, size = list(items), max(1, int(size))
return [items[i:i + size] for i in range(0, len(items), size)]
def flatten(lists):
return [x for xs in lists if xs for x in xs] # one list from a list of lists, skipping workers that failed (None)# kit.py
def batches(items, size):
items, size = list(items), max(1, int(size))
return [items[i:i + size] for i in range(0, len(items), size)]
def flatten(lists):
return [x for xs in lists if xs for x in xs] # one list from a list of lists, skipping workers that failed (None)finding builds one finding, and key_of names its place.
# kit.py
def finding(obj, source=""):
path, line, conf = normalise_path(obj.get("file", "")), int(obj.get("line") or 0), obj.get("confidence")
span = function_at(path, line) # models often name the wrong function
# texts are cut to keep prompts small
return {"file": path, "line": line, "function": span[0] if span else str(obj.get("function") or ""), "source": source,
"claim": str(obj.get("claim") or "")[:400], "failing_call": str(obj.get("failing_call") or "")[:400],
"expected": str(obj.get("expected") or "")[:200], "confidence": min(1.0, max(0.0, float(conf))) if isinstance(conf, (int, float)) else 0.5}
def key_of(f):
"""The place a finding is about: its file and function, or its file and ten-line block."""
return f"{f['file']}::{f['function']}" if f["function"] else f"{f['file']}::line {f['line'] // 10 * 10}"# kit.py
def finding(obj, source=""):
path, line, conf = normalise_path(obj.get("file", "")), int(obj.get("line") or 0), obj.get("confidence")
span = function_at(path, line) # models often name the wrong function
# texts are cut to keep prompts small
return {"file": path, "line": line, "function": span[0] if span else str(obj.get("function") or ""), "source": source,
"claim": str(obj.get("claim") or "")[:400], "failing_call": str(obj.get("failing_call") or "")[:400],
"expected": str(obj.get("expected") or "")[:200], "confidence": min(1.0, max(0.0, float(conf))) if isinstance(conf, (int, float)) else 0.5}
def key_of(f):
"""The place a finding is about: its file and function, or its file and ten-line block."""
return f"{f['file']}::{f['function']}" if f["function"] else f"{f['file']}::line {f['line'] // 10 * 10}"dedupe merges findings about one place and counts the workers behind them.
# kit.py
def dedupe(findings):
# group by place, keep the most confident
groups = collections.defaultdict(list)
for f in findings:
groups[key_of(f)].append(f)
return [dict(max(g, key=lambda f: (f["confidence"], -f["line"], f["claim"], f["failing_call"], f["source"])), # ties are broken by
reports=sum(f.get("reports", 1) for f in g)) for _, g in sorted(groups.items())] # content, so every run keeps the same one# kit.py
def dedupe(findings):
# group by place, keep the most confident
groups = collections.defaultdict(list)
for f in findings:
groups[key_of(f)].append(f)
return [dict(max(g, key=lambda f: (f["confidence"], -f["line"], f["claim"], f["failing_call"], f["source"])), # ties are broken by
reports=sum(f.get("reports", 1) for f in g)) for _, g in sorted(groups.items())] # content, so every run keeps the same oneA place survives only when enough skeptics call it real, which is what vote checks.
= need]” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
# kit.py
def vote(candidates, verdicts, need=2):
# keep places that at least `need` skeptics called real
real = collections.Counter(v["key"] for v in verdicts if v and v.get("real") is True)
return [dict(c, votes=real[key_of(c)]) for c in candidates if real[key_of(c)] >= need]# kit.py
def vote(candidates, verdicts, need=2):
# keep places that at least `need` skeptics called real
real = collections.Counter(v["key"] for v in verdicts if v and v.get("real") is True)
return [dict(c, votes=real[key_of(c)]) for c in candidates if real[key_of(c)] >= need]What a skeptic reads comes from reports, one block per finding.
60: # a very long function: a window around the line instead
first, last = f["line"] – 15, f["line"] + 15
code = "n".join(numbered(f["file"]).split("n")[max(1, first):last + 1]) if f["file"] in WORK else "(no such file)"
# one block per finding, with its code
blocks.append(f"— report {i} —nfile: {f['file']}nfunction: {f['function'] or '(none)'}nline: {f['line']}nclaim: {f['claim']}n"
f"failing call: {f['failing_call'] or '(none given)'}nexpected by the finder: {f['expected'] or '(not given)'}ncode:n{code}")
return "nn".join(blocks)” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
# kit.py
def reports(batch):
blocks = []
for i, f in enumerate(batch, 1):
name, first, last = function_at(f["file"], f["line"]) or ("", f["line"] - 15, f["line"] + 15)
if last - first > 60: # a very long function: a window around the line instead
first, last = f["line"] - 15, f["line"] + 15
code = "n".join(numbered(f["file"]).split("n")[max(1, first):last + 1]) if f["file"] in WORK else "(no such file)"
# one block per finding, with its code
blocks.append(f"--- report {i} ---nfile: {f['file']}nfunction: {f['function'] or '(none)'}nline: {f['line']}nclaim: {f['claim']}n"
f"failing call: {f['failing_call'] or '(none given)'}nexpected by the finder: {f['expected'] or '(not given)'}ncode:n{code}")
return "nn".join(blocks)# kit.py
def reports(batch):
blocks = []
for i, f in enumerate(batch, 1):
name, first, last = function_at(f["file"], f["line"]) or ("", f["line"] - 15, f["line"] + 15)
if last - first > 60: # a very long function: a window around the line instead
first, last = f["line"] - 15, f["line"] + 15
code = "n".join(numbered(f["file"]).split("n")[max(1, first):last + 1]) if f["file"] in WORK else "(no such file)"
# one block per finding, with its code
blocks.append(f"--- report {i} ---nfile: {f['file']}nfunction: {f['function'] or '(none)'}nline: {f['line']}nclaim: {f['claim']}n"
f"failing call: {f['failing_call'] or '(none given)'}nexpected by the finder: {f['expected'] or '(not given)'}ncode:n{code}")
return "nn".join(blocks)AuditKit binds the steps to one run, and run r offsets each sample index by 100 * r so three runs sample independently.
# kit.py
class AuditKit:
def __init__(self, run, number=0):
self.run, self.offset = run, 100 * number # run r uses sample indexes 100 r, 100 r + 1, ...
def scan(self, shard, lens="general", sample=0, model="worker"):
# a lens adds a paragraph, the shard is numbered code
label = f"scan:{lens}:{shard[0].split('/')[0]}#{sample}"
obj = self.run.agent(system=PROMPTS["scan"] + (f"nn{PROMPTS['lens_' + lens]}" if lens in LENSES else ""), schema=FINDINGS,
data="nn".join(map(numbered, shard)), model=model, role="scan", label=label, sample=self.offset + sample)
return [finding(f, label) for f in (obj or {}).get("findings", []) if isinstance(f, dict)]
def check(self, batch, sample=0, model="judge"):
batch = [f for f in batch if f][:8] # a skeptic reads at most eight findings at once
obj = batch and self.run.agent(system=PROMPTS["skeptic"], data=reports(batch), schema=VERDICTS, model=model, role="skeptic",
label=f"check:{batch[0]['file'].split('/')[0]}#{sample}", sample=self.offset + sample)
# match verdicts to findings by number
said = {v.get("report"): v for v in (obj or {}).get("verdicts", []) if isinstance(v, dict)}
return [{"key": key_of(f), "real": said.get(i, {}).get("real"), "reason": str(said.get(i, {}).get("reason", ""))[:200]}
for i, f in enumerate(batch, 1)]# kit.py
class AuditKit:
def __init__(self, run, number=0):
self.run, self.offset = run, 100 * number # run r uses sample indexes 100 r, 100 r + 1, ...
def scan(self, shard, lens="general", sample=0, model="worker"):
# a lens adds a paragraph, the shard is numbered code
label = f"scan:{lens}:{shard[0].split('/')[0]}#{sample}"
obj = self.run.agent(system=PROMPTS["scan"] + (f"nn{PROMPTS['lens_' + lens]}" if lens in LENSES else ""), schema=FINDINGS,
data="nn".join(map(numbered, shard)), model=model, role="scan", label=label, sample=self.offset + sample)
return [finding(f, label) for f in (obj or {}).get("findings", []) if isinstance(f, dict)]
def check(self, batch, sample=0, model="judge"):
batch = [f for f in batch if f][:8] # a skeptic reads at most eight findings at once
obj = batch and self.run.agent(system=PROMPTS["skeptic"], data=reports(batch), schema=VERDICTS, model=model, role="skeptic",
label=f"check:{batch[0]['file'].split('/')[0]}#{sample}", sample=self.offset + sample)
# match verdicts to findings by number
said = {v.get("report"): v for v in (obj or {}).get("verdicts", []) if isinstance(v, dict)}
return [{"key": key_of(f), "real": said.get(i, {}).get("real"), "reason": str(said.get(i, {}).get("reason", ""))[:200]}
for i, f in enumerate(batch, 1)]names lists what a plan may call.
# kit.py
def report(self, confirmed, model="lead"):
# one line per confirmed finding
rows = "n".join(f"C{i} | {f['file']}:{f['line']} | {f['function'] or '-'} | {f['votes']} votes | {f['claim'][:160]}"
for i, f in enumerate(confirmed, 1))
obj = (confirmed and self.run.agent(system=PROMPTS["report"], data=rows, schema=REPORT, model=model, role="report", label="report",
sample=self.offset)) or {}
rank = {i: g.get("severity", "unranked") for g in obj.get("bugs", []) for i in g["ids"]} # the severity the report gave each id
bugs = [dict(f, severity=rank.get(f"C{i}", "unranked")) for i, f in enumerate(confirmed, 1)] # a finding the writer forgot is added back
return {"bugs": bugs, "summary": str(obj.get("summary", ""))[:600], "left_out": sum(f"C{i}" not in rank for i in range(1, len(confirmed) + 1))}
def names(self):
run, budget = self.run, self.run.llm.budget # everything a plan may call
return {"agent": lambda prompt="", **kw: run.agent(prompt, **dict(kw, sample=self.offset + kw.get("sample", 0))),
"parallel": run.parallel, "phase": run.phase, "log": run.log, "budget": lambda: round(budget.cap - budget.spent - budget.reserved, 2),
"work": list(WORK), "shards": lambda tokens=SHARD_TOKENS: pack(WORK, int(tokens)), "batches": batches, "flatten": flatten,
"render": lambda shard: "nn".join(map(numbered, shard)), "dedupe": dedupe, "key_of": key_of, "scan": self.scan,
"check": self.check, "vote": vote, "report": self.report, "LENSES": list(LENSES), "MODELS": sorted(MODELS),
"SCHEMAS": {"findings": FINDINGS, "verdicts": VERDICTS}}# kit.py
def report(self, confirmed, model="lead"):
# one line per confirmed finding
rows = "n".join(f"C{i} | {f['file']}:{f['line']} | {f['function'] or '-'} | {f['votes']} votes | {f['claim'][:160]}"
for i, f in enumerate(confirmed, 1))
obj = (confirmed and self.run.agent(system=PROMPTS["report"], data=rows, schema=REPORT, model=model, role="report", label="report",
sample=self.offset)) or {}
rank = {i: g.get("severity", "unranked") for g in obj.get("bugs", []) for i in g["ids"]} # the severity the report gave each id
bugs = [dict(f, severity=rank.get(f"C{i}", "unranked")) for i, f in enumerate(confirmed, 1)] # a finding the writer forgot is added back
return {"bugs": bugs, "summary": str(obj.get("summary", ""))[:600], "left_out": sum(f"C{i}" not in rank for i in range(1, len(confirmed) + 1))}
def names(self):
run, budget = self.run, self.run.llm.budget # everything a plan may call
return {"agent": lambda prompt="", **kw: run.agent(prompt, **dict(kw, sample=self.offset + kw.get("sample", 0))),
"parallel": run.parallel, "phase": run.phase, "log": run.log, "budget": lambda: round(budget.cap - budget.spent - budget.reserved, 2),
"work": list(WORK), "shards": lambda tokens=SHARD_TOKENS: pack(WORK, int(tokens)), "batches": batches, "flatten": flatten,
"render": lambda shard: "nn".join(map(numbered, shard)), "dedupe": dedupe, "key_of": key_of, "scan": self.scan,
"check": self.check, "vote": vote, "report": self.report, "LENSES": list(LENSES), "MODELS": sorted(MODELS),
"SCHEMAS": {"findings": FINDINGS, "verdicts": VERDICTS}}Fixed baseline
I need a plan to beat first: scan every shard through all three lenses, check each finding with three skeptics, keep what two confirm.
"""plans.py: a fixed plan, the whitelist, the dry run and the gate."""
import ast, builtins, collections, hashlib, json, re, threading, zlib
from config import PRICE
from kit import AuditKit
from llm import LLM, Budget, Ledger, Reply
from runtime import Run
FIXED_PROGRAM = """phase("Scan")
found = flatten(parallel([lambda s=s, lens=lens: scan(s, lens) for s in shards(100000) for lens in LENSES]))
candidates = dedupe(found)
phase("Verify")
verdicts = flatten(parallel([lambda b=b, k=k: check(b, sample=k) for b in batches(candidates, 6) for k in range(3)]))
confirmed = vote(candidates, verdicts, need=2)
phase("Report")
result = report(confirmed)""""""plans.py: a fixed plan, the whitelist, the dry run and the gate."""
import ast, builtins, collections, hashlib, json, re, threading, zlib
from config import PRICE
from kit import AuditKit
from llm import LLM, Budget, Ledger, Reply
from runtime import Run
FIXED_PROGRAM = """phase("Scan")
found = flatten(parallel([lambda s=s, lens=lens: scan(s, lens) for s in shards(100000) for lens in LENSES]))
candidates = dedupe(found)
phase("Verify")
verdicts = flatten(parallel([lambda b=b, k=k: check(b, sample=k) for b in batches(candidates, 6) for k in range(3)]))
confirmed = vote(candidates, verdicts, need=2)
phase("Report")
result = report(confirmed)"""Across three runs the plan found 20, 23 and 22 of the 72 bugs, a recall of 0.28, 0.32 and 0.31, for about 14 credits a run.
Restricted Plan Language
Our fixed plan never changes, yet in its own records rounds two and three found 9 planted bugs that round one missed. A better plan scans until a round adds little and shrinks when budget() is low. A list of settings cannot express that, which is why the lead writes Python.
It is a restricted Python that can be checked before it runs and reaches only the names it is given. The restriction is a whitelist of rules that each forbid one thing. It is not a security boundary, because the plans come from our own lead.
# plans.py
ALLOWED = {getattr(ast, n) for n in ( # the only kinds of syntax a plan may use
"Module Expr Assign AugAssign For If Break Continue Return FunctionDef Lambda ListComp DictComp SetComp GeneratorExp comprehension "
"Call keyword Name Attribute Subscript Slice Constant List Tuple Dict Set BoolOp BinOp UnaryOp Compare IfExp JoinedStr FormattedValue "
"arguments arg Starred Load Store And Or Add Sub Mult Div FloorDiv Mod Not USub UAdd Eq NotEq Lt LtE Gt GtE In NotIn Is IsNot").split()}
WHY_NOT = {ast.While: "use a for loop over range(n), so the loop states its bound", ast.Import: "a plan has nothing to import",
ast.Try: "a plan must not hide errors, and failed workers already return None"}
METHODS = {"append", "extend", "get", "items", "keys", "values", "update", "add", "setdefault", "pop", "sort", "split", "strip", "lower", "upper",
"join", "startswith", "endswith", "count", "index", "replace"} # the methods a plan may call on data, with format left out on purpose
BUILTINS = ["len", "range", "enumerate", "zip", "sorted", "min", "max", "sum", "any", "all", "list", "dict", "set", "tuple", "str", "int",
"float", "round", "abs"]
class PlanRejected(ValueError):
"""A plan program failed the whitelist check or the dry run."""# plans.py
ALLOWED = {getattr(ast, n) for n in ( # the only kinds of syntax a plan may use
"Module Expr Assign AugAssign For If Break Continue Return FunctionDef Lambda ListComp DictComp SetComp GeneratorExp comprehension "
"Call keyword Name Attribute Subscript Slice Constant List Tuple Dict Set BoolOp BinOp UnaryOp Compare IfExp JoinedStr FormattedValue "
"arguments arg Starred Load Store And Or Add Sub Mult Div FloorDiv Mod Not USub UAdd Eq NotEq Lt LtE Gt GtE In NotIn Is IsNot").split()}
WHY_NOT = {ast.While: "use a for loop over range(n), so the loop states its bound", ast.Import: "a plan has nothing to import",
ast.Try: "a plan must not hide errors, and failed workers already return None"}
METHODS = {"append", "extend", "get", "items", "keys", "values", "update", "add", "setdefault", "pop", "sort", "split", "strip", "lower", "upper",
"join", "startswith", "endswith", "count", "index", "replace"} # the methods a plan may call on data, with format left out on purpose
BUILTINS = ["len", "range", "enumerate", "zip", "sorted", "min", "max", "sum", "any", "all", "list", "dict", "set", "tuple", "str", "int",
"float", "round", "abs"]
class PlanRejected(ValueError):
"""A plan program failed the whitelist check or the dry run."""Rules 2 and 4 check the syntax.
1_000_000:
out.append(f"line {line}: the number {n.value} is larger than any plan needs")
# rule 4: only allowed method calls
elif isinstance(n, ast.Attribute) and (n.attr.startswith("_") or id(n) not in called or n.attr not in METHODS):
out.append(f"line {line}: '.{n.attr}' is not allowed: a plan may only call {', '.join(sorted(METHODS))} on data")
return out” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
# plans.py
def rule_syntax(tree):
"""Rules 2 and 4: only the allowed syntax, no number above a million, and attributes only as calls of a few methods."""
# called attributes, so a bare one stands out
called, out = {id(n.func) for n in ast.walk(tree) if isinstance(n, ast.Call)}, []
for n in ast.walk(tree):
line = getattr(n, "lineno", 1)
if type(n) not in ALLOWED: # rule 2: syntax outside the list
out.append(f"line {line}: {type(n).__name__} is not allowed: {WHY_NOT.get(type(n), 'it is not part of the plan language')}")
# rule 2 again: a number no plan needs
elif isinstance(n, ast.Constant) and type(n.value) is int and abs(n.value) > 1_000_000:
out.append(f"line {line}: the number {n.value} is larger than any plan needs")
# rule 4: only allowed method calls
elif isinstance(n, ast.Attribute) and (n.attr.startswith("_") or id(n) not in called or n.attr not in METHODS):
out.append(f"line {line}: '.{n.attr}' is not allowed: a plan may only call {', '.join(sorted(METHODS))} on data")
return out# plans.py
def rule_syntax(tree):
"""Rules 2 and 4: only the allowed syntax, no number above a million, and attributes only as calls of a few methods."""
# called attributes, so a bare one stands out
called, out = {id(n.func) for n in ast.walk(tree) if isinstance(n, ast.Call)}, []
for n in ast.walk(tree):
line = getattr(n, "lineno", 1)
if type(n) not in ALLOWED: # rule 2: syntax outside the list
out.append(f"line {line}: {type(n).__name__} is not allowed: {WHY_NOT.get(type(n), 'it is not part of the plan language')}")
# rule 2 again: a number no plan needs
elif isinstance(n, ast.Constant) and type(n.value) is int and abs(n.value) > 1_000_000:
out.append(f"line {line}: the number {n.value} is larger than any plan needs")
# rule 4: only allowed method calls
elif isinstance(n, ast.Attribute) and (n.attr.startswith("_") or id(n) not in called or n.attr not in METHODS):
out.append(f"line {line}: '.{n.attr}' is not allowed: a plan may only call {', '.join(sorted(METHODS))} on data")
return outRule 3 checks every name.
# plans.py
def rule_names(tree, names):
"""Rule 3: only the given names. None starts with an underscore, and the language itself cannot be reassigned."""
# names the plan binds itself
bound = {n.id for n in ast.walk(tree) if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Store)}
bound |= {n.name for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)} | {n.arg for n in ast.walk(tree) if isinstance(n, ast.arg)}
out = []
for n in ast.walk(tree):
name = n.id if isinstance(n, ast.Name) else n.name if isinstance(n, ast.FunctionDef) else n.arg if isinstance(n, ast.arg) else None
stored = not isinstance(n, ast.Name) or isinstance(n.ctx, ast.Store)
if name is None:
continue
if name.startswith("_") and name != "_": # a lone _ is the usual name for an unused value
out.append(f"line {n.lineno}: the name '{name}' starts with an underscore, which is not allowed")
elif name in names and stored:
out.append(f"line {n.lineno}: '{name}' is part of the plan language and cannot be reassigned")
# a name it reads must be defined
elif not stored and name not in names | set(BUILTINS) | bound:
out.append(f"line {n.lineno}: the name '{name}' is not defined: the plan language has no such name")
return sorted(set(out), key=lambda s: int(s.split(":")[0][5:]))# plans.py
def rule_names(tree, names):
"""Rule 3: only the given names. None starts with an underscore, and the language itself cannot be reassigned."""
# names the plan binds itself
bound = {n.id for n in ast.walk(tree) if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Store)}
bound |= {n.name for n in ast.walk(tree) if isinstance(n, ast.FunctionDef)} | {n.arg for n in ast.walk(tree) if isinstance(n, ast.arg)}
out = []
for n in ast.walk(tree):
name = n.id if isinstance(n, ast.Name) else n.name if isinstance(n, ast.FunctionDef) else n.arg if isinstance(n, ast.arg) else None
stored = not isinstance(n, ast.Name) or isinstance(n.ctx, ast.Store)
if name is None:
continue
if name.startswith("_") and name != "_": # a lone _ is the usual name for an unused value
out.append(f"line {n.lineno}: the name '{name}' starts with an underscore, which is not allowed")
elif name in names and stored:
out.append(f"line {n.lineno}: '{name}' is part of the plan language and cannot be reassigned")
# a name it reads must be defined
elif not stored and name not in names | set(BUILTINS) | bound:
out.append(f"line {n.lineno}: the name '{name}' is not defined: the plan language has no such name")
return sorted(set(out), key=lambda s: int(s.split(":")[0][5:]))Rule 1 keeps a plan short and rule 5 asks it to assign result.
6000 or len(src.splitlines()) > 80: # rule 1: short enough to read in one sitting
return [f"line 1: the program has {len(src.splitlines())} lines and {len(src)} characters, and the limit is 80 and 6,000"]
try:
tree = ast.parse(src)
except SyntaxError as e:
return [f"line {e.lineno}: this is not valid Python ({e.msg})"]
has_result = any(isinstance(n, ast.Name) and n.id == "result" and isinstance(n.ctx, ast.Store) for n in ast.walk(tree)) # rule 5: the plan sets result
# rules 2, 3 and 4 walk the syntax tree
return rule_syntax(tree) + rule_names(tree, set(names)) + ([] if has_result else ["line 1: the program never assigns `result`"])” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
# plans.py
def check_program(src, names):
"""Every problem with a plan program, by line. An empty list means it may go on."""
if len(src) > 6000 or len(src.splitlines()) > 80: # rule 1: short enough to read in one sitting
return [f"line 1: the program has {len(src.splitlines())} lines and {len(src)} characters, and the limit is 80 and 6,000"]
try:
tree = ast.parse(src)
except SyntaxError as e:
return [f"line {e.lineno}: this is not valid Python ({e.msg})"]
has_result = any(isinstance(n, ast.Name) and n.id == "result" and isinstance(n.ctx, ast.Store) for n in ast.walk(tree)) # rule 5: the plan sets result
# rules 2, 3 and 4 walk the syntax tree
return rule_syntax(tree) + rule_names(tree, set(names)) + ([] if has_result else ["line 1: the program never assigns `result`"])# plans.py
def check_program(src, names):
"""Every problem with a plan program, by line. An empty list means it may go on."""
if len(src) > 6000 or len(src.splitlines()) > 80: # rule 1: short enough to read in one sitting
return [f"line 1: the program has {len(src.splitlines())} lines and {len(src)} characters, and the limit is 80 and 6,000"]
try:
tree = ast.parse(src)
except SyntaxError as e:
return [f"line {e.lineno}: this is not valid Python ({e.msg})"]
has_result = any(isinstance(n, ast.Name) and n.id == "result" and isinstance(n.ctx, ast.Store) for n in ast.walk(tree)) # rule 5: the plan sets result
# rules 2, 3 and 4 walk the syntax tree
return rule_syntax(tree) + rule_names(tree, set(names)) + ([] if has_result else ["line 1: the program never assigns `result`"])run_program runs a plan only after it passes, with nothing but the kit and a few built-ins in reach.
<button data-code="# plans.py
def run_program(src, kit):
problems = check_program(src, kit.names())
if problems:
raise PlanRejected(" | ".join(problems[:5]))
# only the kit's names and safe built-ins
env = dict(kit.names(), __builtins__={n: getattr(builtins, n) for n in BUILTINS})
exec(compile(src, f"", "exec"), env)
return env.get("result")” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
# plans.py
def run_program(src, kit):
problems = check_program(src, kit.names())
if problems:
raise PlanRejected(" | ".join(problems[:5]))
# only the kit's names and safe built-ins
env = dict(kit.names(), __builtins__={n: getattr(builtins, n) for n in BUILTINS})
exec(compile(src, f"<plan {kit.run.run_id}>", "exec"), env)
return env.get("result")# plans.py
def run_program(src, kit):
problems = check_program(src, kit.names())
if problems:
raise PlanRejected(" | ".join(problems[:5]))
# only the kit's names and safe built-ins
env = dict(kit.names(), __builtins__={n: getattr(builtins, n) for n in BUILTINS})
exec(compile(src, f"<plan {kit.run.run_id}>", "exec"), env)
return env.get("result")The checker refused all fourteen hostile programs I wrote, each with a reason a model can act on.
Dry Runs and the Approval Gate
A plan can pass every rule and still be costly, since how many workers it starts depends on what earlier workers send back. A dry run runs the plan as written against a stand-in model that answers at once and costs nothing.
> (j % 16)) % 60, "claim": "a made-up finding",
"confidence": 0.5} for j in range(seed % 3)]} # none, one or two findings per call
elif system.startswith("You check"):
seed = zlib.crc32((text[:200] + str(params.get("sample", 0))).encode())
answer = {"verdicts": [{"report": i, "real": (seed >> i) % 5
# plans.py
def blind(model, messages, params):
system, text = messages[0]["content"], messages[-1]["content"]
# answers come from a hash of the question
seed = zlib.crc32((text[:300] + str(params.get("sample", 0))).encode())
if system.startswith("You are reviewing"):
paths = re.findall(r"^### (S+)$", text, re.M) or ["unknown.py"]
answer = {"findings": [{"file": paths[(seed + 7 * j) % len(paths)], "line": 5 + (seed >> (j % 16)) % 60, "claim": "a made-up finding",
"confidence": 0.5} for j in range(seed % 3)]} # none, one or two findings per call
elif system.startswith("You check"):
seed = zlib.crc32((text[:200] + str(params.get("sample", 0))).encode())
answer = {"verdicts": [{"report": i, "real": (seed >> i) % 5 < 2} for i in range(1, len(re.findall(r"^--- report d+ ---$", text, re.M)) + 1)]}
else: # the report: every finding in a group of its own
answer = {"bugs": [{"ids": [i], "severity": "medium"} for i in re.findall(r"^(Cd+) |", text, re.M)], "summary": "a made-up report"}
return Reply(json.dumps(answer), "stop", model=model)# plans.py
def blind(model, messages, params):
system, text = messages[0]["content"], messages[-1]["content"]
# answers come from a hash of the question
seed = zlib.crc32((text[:300] + str(params.get("sample", 0))).encode())
if system.startswith("You are reviewing"):
paths = re.findall(r"^### (S+)$", text, re.M) or ["unknown.py"]
answer = {"findings": [{"file": paths[(seed + 7 * j) % len(paths)], "line": 5 + (seed >> (j % 16)) % 60, "claim": "a made-up finding",
"confidence": 0.5} for j in range(seed % 3)]} # none, one or two findings per call
elif system.startswith("You check"):
seed = zlib.crc32((text[:200] + str(params.get("sample", 0))).encode())
answer = {"verdicts": [{"report": i, "real": (seed >> i) % 5 < 2} for i in range(1, len(re.findall(r"^--- report d+ ---$", text, re.M)) + 1)]}
else: # the report: every finding in a group of its own
answer = {"bugs": [{"ids": [i], "severity": "medium"} for i in re.findall(r"^(Cd+) |", text, re.M)], "summary": "a made-up report"}
return Reply(json.dumps(answer), "stop", model=model)dry_run prices what the program did at the real prices, and a time limit covers its running time.
# plans.py
def dry_run(src, budget=150, timeout=60):
llm = LLM(Ledger(":memory:"), scripted=blind, budget=Budget(budget)) # budget() starts where the real run starts
run = Run("dry run", llm=llm, run_id="dry-" + hashlib.sha256(src.encode()).hexdigest()[:8], concurrency=64)
kit = AuditKit(run)
problems = check_program(src, kit.names())
if problems:
return None, problems
worker = threading.Thread(target=lambda: run.run_sync(lambda r: run_program(src, kit)), daemon=True)
worker.start()
worker.join(timeout)
if worker.is_alive(): # a clock bounds the running time, which no rule can check
run.cancel()
return None, [f"line 1: the dry run did not finish within {timeout} s, so the plan probably loops without end"]
if run.status != "completed":
return None, [f"line 1: the dry run ended {run.status}: {run.error or f'{len(run.dropped)} workers refused by the agent limit'}"]
# price the calls at the real prices
calls = collections.Counter(t.model for t in run.threads for _ in t.replies)
return {"agents": len(run.threads), "calls": sum(calls.values()), "credits": round(sum(PRICE.get(m, 0.2) * n for m, n in calls.items()), 1),
"phases": {p["name"]: sum(t.phase == p["name"] for t in run.threads) for p in run.phases}}, []# plans.py
def dry_run(src, budget=150, timeout=60):
llm = LLM(Ledger(":memory:"), scripted=blind, budget=Budget(budget)) # budget() starts where the real run starts
run = Run("dry run", llm=llm, run_id="dry-" + hashlib.sha256(src.encode()).hexdigest()[:8], concurrency=64)
kit = AuditKit(run)
problems = check_program(src, kit.names())
if problems:
return None, problems
worker = threading.Thread(target=lambda: run.run_sync(lambda r: run_program(src, kit)), daemon=True)
worker.start()
worker.join(timeout)
if worker.is_alive(): # a clock bounds the running time, which no rule can check
run.cancel()
return None, [f"line 1: the dry run did not finish within {timeout} s, so the plan probably loops without end"]
if run.status != "completed":
return None, [f"line 1: the dry run ended {run.status}: {run.error or f'{len(run.dropped)} workers refused by the agent limit'}"]
# price the calls at the real prices
calls = collections.Counter(t.model for t in run.threads for _ in t.replies)
return {"agents": len(run.threads), "calls": sum(calls.values()), "credits": round(sum(PRICE.get(m, 0.2) * n for m, n in calls.items()), 1),
"phases": {p["name"]: sum(t.phase == p["name"] for t in run.threads) for p in run.phases}}, []The approval gate applies a policy written before any plan was priced: a plan of up to 40 credits and a third of the budget passes on its own.
budget: # nor one costing more than the budget
return "refused", f"estimated {est['credits']:.1f} credits is more than the whole budget of {budget:.0f}"
limit, cost = min(AUTO_APPROVE["max_credits"], AUTO_APPROVE["max_share"] * budget), f"estimated {est['credits']:.1f} credits"
if est["credits"]
# plans.py
AUTO_APPROVE = {"max_credits": 40.0, "max_share": 1 / 3} # the policy, written down before any plan is priced
def approve(est, budget, person=None):
if est is None: # a plan that failed its checks never runs
return "refused", "the plan failed its check or its dry run"
if est["credits"] > budget: # nor one costing more than the budget
return "refused", f"estimated {est['credits']:.1f} credits is more than the whole budget of {budget:.0f}"
limit, cost = min(AUTO_APPROVE["max_credits"], AUTO_APPROVE["max_share"] * budget), f"estimated {est['credits']:.1f} credits"
if est["credits"] <= limit:
return "approved", f"{cost} is under the automatic limit of {limit:.1f}"
if person is None: # above the limit a person decides, or the plan waits
return "held", f"{cost} is over the automatic limit of {limit:.1f}, so a person must approve it"
yes = bool(person(est))
return ("approved" if yes else "refused"), f"a person {'approved' if yes else 'refused'} it after reading the {cost}"# plans.py
AUTO_APPROVE = {"max_credits": 40.0, "max_share": 1 / 3} # the policy, written down before any plan is priced
def approve(est, budget, person=None):
if est is None: # a plan that failed its checks never runs
return "refused", "the plan failed its check or its dry run"
if est["credits"] > budget: # nor one costing more than the budget
return "refused", f"estimated {est['credits']:.1f} credits is more than the whole budget of {budget:.0f}"
limit, cost = min(AUTO_APPROVE["max_credits"], AUTO_APPROVE["max_share"] * budget), f"estimated {est['credits']:.1f} credits"
if est["credits"] <= limit:
return "approved", f"{cost} is under the automatic limit of {limit:.1f}"
if person is None: # above the limit a person decides, or the plan waits
return "held", f"{cost} is over the automatic limit of {limit:.1f}, so a person must approve it"
yes = bool(person(est))
return ("approved" if yes else "refused"), f"a person {'approved' if yes else 'refused'} it after reading the {cost}"I priced three plans: the fixed plan, one that shrinks when budget() is low, and a runaway that passes every rule yet asks for a million workers.
100 else ["general"]
skeptics = 3 if budget() > 100 else 1
found = dedupe(flatten(parallel([lambda s=s, l=l: scan(s, l) for s in shards(40000) for l in lenses])))
verdicts = flatten(parallel([lambda b=b, k=k: check(b, sample=k) for b in batches(found, 6) for k in range(skeptics)]))
result = report(vote(found, verdicts, need=skeptics // 2 + 1))"""
RUNAWAY = 'for i in range(1000000):n agent("look again", label=f"again#{i}")nresult = report([])'
estimates = {"fixed plan": dry_run(FIXED_PROGRAM)[0], "fit the budget, sees 150": dry_run(FIT, 150)[0], "fit the budget, sees 45": dry_run(FIT, 45)[0]}
for name, e in estimates.items():
print(f"{name:3} workers, {e['credits']:>4.1f} credits | gate at a budget of 45: {approve(e, 45)[0]}")
print("runaway passes the rules:", not check_program(RUNAWAY, names), "| dry run:", dry_run(RUNAWAY)[1][0])
#### OUTPUT ####
fixed plan 49 workers, 10.1 credits | gate at a budget of 45: approved
fit the budget, sees 150 118 workers, 23.9 credits | gate at a budget of 45: held
fit the budget, sees 45 40 workers, 8.3 credits | gate at a budget of 45: approved
runaway passes the rules: True | dry run: line 1: the dry run ended cancelled: 1000 workers refused by the agent limit” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
from kit import AuditKit
from plans import FIXED_PROGRAM, approve, check_program, dry_run
names = AuditKit(Run("api", llm=llm)).names()
FIT = """lenses = LENSES if budget() > 100 else ["general"]
skeptics = 3 if budget() > 100 else 1
found = dedupe(flatten(parallel([lambda s=s, l=l: scan(s, l) for s in shards(40000) for l in lenses])))
verdicts = flatten(parallel([lambda b=b, k=k: check(b, sample=k) for b in batches(found, 6) for k in range(skeptics)]))
result = report(vote(found, verdicts, need=skeptics // 2 + 1))"""
RUNAWAY = 'for i in range(1000000):n agent("look again", label=f"again#{i}")nresult = report([])'
estimates = {"fixed plan": dry_run(FIXED_PROGRAM)[0], "fit the budget, sees 150": dry_run(FIT, 150)[0], "fit the budget, sees 45": dry_run(FIT, 45)[0]}
for name, e in estimates.items():
print(f"{name:<24} {e['agents']:>3} workers, {e['credits']:>4.1f} credits | gate at a budget of 45: {approve(e, 45)[0]}")
print("runaway passes the rules:", not check_program(RUNAWAY, names), "| dry run:", dry_run(RUNAWAY)[1][0])
#### OUTPUT ####
fixed plan 49 workers, 10.1 credits | gate at a budget of 45: approved
fit the budget, sees 150 118 workers, 23.9 credits | gate at a budget of 45: held
fit the budget, sees 45 40 workers, 8.3 credits | gate at a budget of 45: approved
runaway passes the rules: True | dry run: line 1: the dry run ended cancelled: 1000 workers refused by the agent limitfrom kit import AuditKit
from plans import FIXED_PROGRAM, approve, check_program, dry_run
names = AuditKit(Run("api", llm=llm)).names()
FIT = """lenses = LENSES if budget() > 100 else ["general"]
skeptics = 3 if budget() > 100 else 1
found = dedupe(flatten(parallel([lambda s=s, l=l: scan(s, l) for s in shards(40000) for l in lenses])))
verdicts = flatten(parallel([lambda b=b, k=k: check(b, sample=k) for b in batches(found, 6) for k in range(skeptics)]))
result = report(vote(found, verdicts, need=skeptics // 2 + 1))"""
RUNAWAY = 'for i in range(1000000):n agent("look again", label=f"again#{i}")nresult = report([])'
estimates = {"fixed plan": dry_run(FIXED_PROGRAM)[0], "fit the budget, sees 150": dry_run(FIT, 150)[0], "fit the budget, sees 45": dry_run(FIT, 45)[0]}
for name, e in estimates.items():
print(f"{name:<24} {e['agents']:>3} workers, {e['credits']:>4.1f} credits | gate at a budget of 45: {approve(e, 45)[0]}")
print("runaway passes the rules:", not check_program(RUNAWAY, names), "| dry run:", dry_run(RUNAWAY)[1][0])
#### OUTPUT ####
fixed plan 49 workers, 10.1 credits | gate at a budget of 45: approved
fit the budget, sees 150 118 workers, 23.9 credits | gate at a budget of 45: held
fit the budget, sees 45 40 workers, 8.3 credits | gate at a budget of 45: approved
runaway passes the rules: True | dry run: line 1: the dry run ended cancelled: 1000 workers refused by the agent limitThe same program costs 23.9 credits when the rehearsal sees a budget of 150 and 8.3 when it sees 45.
So the rehearsal has to see the same budget as the real run. The fixed plan’s estimate of 10.1 credits was 27% below the 13.8 of its three real runs.
Lead Planner
Ordinary code writes the lead’s fact sheet, and the sheet costs nothing and never changes between runs. Here it is, abridged.
#### OUTPUT ####
THE WORK
- 849 Python files, 91,327 lines, about 1,348,181 tokens with line numbers. The largest file is 13,353 tokens.
MEASURED SO FAR
- The measured best shard size is 100k tokens.
PRICES AND LIMITS
- Price per call in credits: worker 0.2, fast 0.2, judge 0.2, lead 0.5. Billing is per call, not per token.
- At most 1000 workers per run, and about 32 run at the same time.
- Budget for this job: 150 credits.#### OUTPUT ####
THE WORK
- 849 Python files, 91,327 lines, about 1,348,181 tokens with line numbers. The largest file is 13,353 tokens.
MEASURED SO FAR
- The measured best shard size is 100k tokens.
PRICES AND LIMITS
- Price per call in credits: worker 0.2, fast 0.2, judge 0.2, lead 0.5. Billing is per call, not per token.
- At most 1000 workers per run, and about 32 run at the same time.
- Budget for this job: 150 credits.It holds file and token counts, a few measured rates and the prices, about 500 tokens for a job of 1,348,181.
left out:the code that builds the sheet, about twenty lines of string formatting
Lead talks to the model and keeps a tally of what it read.
"""lead.py: the lead reads a fact sheet and writes the plan."""
import re
from bench import WORK
from config import MODELS, PRICE, PROMPTS, TEMPERATURE, TOP_P
from kit import AuditKit
from llm import LLM, Budget
from plans import approve, dry_run, run_program
from runtime import Run
FENCE = "`" * 3 # three backticks, built so this file's own fence stays open
class Lead:
"""The lead: it reads the fact sheet and writes plan programs."""
def __init__(self, llm):
self.llm, self.read = llm, []
def ask(self, messages, sample=0):
r = self.llm.chat(MODELS["lead"], messages, role="lead", sample=sample, temperature=TEMPERATURE, top_p=TOP_P, max_tokens=8000)
self.read.append(r.prompt_tokens) # kept, to count what the lead had to read
return r"""lead.py: the lead reads a fact sheet and writes the plan."""
import re
from bench import WORK
from config import MODELS, PRICE, PROMPTS, TEMPERATURE, TOP_P
from kit import AuditKit
from llm import LLM, Budget
from plans import approve, dry_run, run_program
from runtime import Run
FENCE = "`" * 3 # three backticks, built so this file's own fence stays open
class Lead:
"""The lead: it reads the fact sheet and writes plan programs."""
def __init__(self, llm):
self.llm, self.read = llm, []
def ask(self, messages, sample=0):
r = self.llm.chat(MODELS["lead"], messages, role="lead", sample=sample, temperature=TEMPERATURE, top_p=TOP_P, max_tokens=8000)
self.read.append(r.prompt_tokens) # kept, to count what the lead had to read
return rwrite_programasks for a plan, checks it, rehearses it, and hands every problem back, for up to two repairs.
# lead.py
def write_program(self, job, page, budget=150, sample=0, repairs=2):
# the job and the fact sheet are all the lead reads
messages = [{"role": "system", "content": PROMPTS["lead"]}, {"role": "user", "content": f"THE JOBn{job}nn{page}"}]
attempts = []
for _ in range(1 + repairs):
r = self.ask(messages, sample)
# take out the python block, then check and rehearse it
m = re.search(FENCE + r"(?:python|py)?s*n(.*?)" + FENCE, r.text, re.S)
src = m.group(1).strip() if m else None
est, problems = dry_run(src, budget) if src else (None, [f"line 1: the reply has no {FENCE}python block"])
attempts.append({"src": src, "problems": problems, "estimate": est})
if not problems:
break
# hand every problem back in the checker's own words
messages += [{"role": "assistant", "content": r.text[:8000]}, {"role": "user", "content": "Your program has these problems:n- "
+ "n- ".join(problems[:8]) + f"nSend the whole corrected program in one {FENCE}python block."}]
return {"src": src, "valid": not problems, "tries": len(attempts), "attempts": attempts, "estimate": est}# lead.py
def write_program(self, job, page, budget=150, sample=0, repairs=2):
# the job and the fact sheet are all the lead reads
messages = [{"role": "system", "content": PROMPTS["lead"]}, {"role": "user", "content": f"THE JOBn{job}nn{page}"}]
attempts = []
for _ in range(1 + repairs):
r = self.ask(messages, sample)
# take out the python block, then check and rehearse it
m = re.search(FENCE + r"(?:python|py)?s*n(.*?)" + FENCE, r.text, re.S)
src = m.group(1).strip() if m else None
est, problems = dry_run(src, budget) if src else (None, [f"line 1: the reply has no {FENCE}python block"])
attempts.append({"src": src, "problems": problems, "estimate": est})
if not problems:
break
# hand every problem back in the checker's own words
messages += [{"role": "assistant", "content": r.text[:8000]}, {"role": "user", "content": "Your program has these problems:n- "
+ "n- ".join(problems[:8]) + f"nSend the whole corrected program in one {FENCE}python block."}]
return {"src": src, "valid": not problems, "tries": len(attempts), "attempts": attempts, "estimate": est}I asked the same lead for ten programs, one per sample index. From here on the model is real, and each output is the one the original run recorded. That run also retried failed calls and showed each skeptic a sandbox result, two steps this build leaves out.
5} {w['estimate']['agents']:>8} {w['estimate']['credits']:>8.1f}")
#### OUTPUT ####
tries workers credits
1 263 52.9
1 182 36.4
1 261 52.5
2 173 34.9
2 252 50.4
2 112 22.4
2 196 39.2
1 226 45.5
1 263 52.9
1 263 52.9″ data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
from pathlib import Path
from lead import Lead
from llm import http_client
JOB = "Find the behaviour bugs in a library of 849 Python files [...]. Report each bug with its file, its line and a call that shows it."
PAGE = Path("page.txt").read_text() # the fact sheet shown above
live = LLM(Ledger("ledger.sqlite"), client=http_client) # the real model
programs = [Lead(live).write_program(JOB, PAGE, budget=150, sample=k) for k in range(10)]
print("tries workers credits")
for w in programs:
print(f"{w['tries']:>5} {w['estimate']['agents']:>8} {w['estimate']['credits']:>8.1f}")
#### OUTPUT ####
tries workers credits
1 263 52.9
1 182 36.4
1 261 52.5
2 173 34.9
2 252 50.4
2 112 22.4
2 196 39.2
1 226 45.5
1 263 52.9
1 263 52.9from pathlib import Path
from lead import Lead
from llm import http_client
JOB = "Find the behaviour bugs in a library of 849 Python files [...]. Report each bug with its file, its line and a call that shows it."
PAGE = Path("page.txt").read_text() # the fact sheet shown above
live = LLM(Ledger("ledger.sqlite"), client=http_client) # the real model
programs = [Lead(live).write_program(JOB, PAGE, budget=150, sample=k) for k in range(10)]
print("tries workers credits")
for w in programs:
print(f"{w['tries']:>5} {w['estimate']['agents']:>8} {w['estimate']['credits']:>8.1f}")
#### OUTPUT ####
tries workers credits
1 263 52.9
1 182 36.4
1 261 52.5
2 173 34.9
2 252 50.4
2 112 22.4
2 196 39.2
1 226 45.5
1 263 52.9
1 263 52.9Six were valid at the first try. The four that needed a repair (tries 2) had all broken the same rule, the length limit, with 103, 176, 141 and 159 lines where 80 is the maximum. All ten looped and decided from their results, but only one read budget().
A Full Audit of the Real Library
audit runs every stage in turn, from the lead’s first draft to the background run.
= set(WORK),
"credits": sum(PRICE.get(t.model, 0.2) * len(t.replies) for t in run.threads)}
return {"decision": decision, "why": why, "plan": plan, "run": run, "result": result, "stats": stats, "lead_read": sum(lead.read)}” data-toggle-ms=”1200″ aria-label=”Copy code” data-nosnippet=”” class=”code-copy-btn absolute right-3 top-3 size-8 p-1.5 flex items-center justify-center bg-black/50 text-white rounded-md transition-colors duration-200 cursor-pointer hover:bg-black/70″>
# lead.py
def audit(llm, job, page, budget=150, sample=0, person=None):
lead = Lead(llm)
plan = lead.write_program(job, page, budget, sample)
decision, why = approve(plan["estimate"], budget, person)
if decision != "approved":
return {"decision": decision, "why": why, "plan": plan}
# the approved plan runs on a budget of its own
run = Run("the lead's own plan", job, llm=LLM(llm.ledger, llm.mode, llm.client, llm.scripted, Budget(budget)), run_id=f"d{sample}", concurrency=32)
result = run.start(lambda r: run_program(plan["src"], AuditKit(run, sample))).wait()
# which files the scan workers saw
seen = {p for t in run.threads if t.role == "scan" for p in re.findall(r"^### (S+)$", t.messages[1]["content"], re.M)}
stats = {"agents": len(run.threads), "calls": sum(len(t.replies) for t in run.threads), "covered": seen >= set(WORK),
"credits": sum(PRICE.get(t.model, 0.2) * len(t.replies) for t in run.threads)}
return {"decision": decision, "why": why, "plan": plan, "run": run, "result": result, "stats": stats, "lead_read": sum(lead.read)}# lead.py
def audit(llm, job, page, budget=150, sample=0, person=None):
lead = Lead(llm)
plan = lead.write_program(job, page, budget, sample)
decision, why = approve(plan["estimate"], budget, person)
if decision != "approved":
return {"decision": decision, "why": why, "plan": plan}
# the approved plan runs on a budget of its own
run = Run("the lead's own plan", job, llm=LLM(llm.ledger, llm.mode, llm.client, llm.scripted, Budget(budget)), run_id=f"d{sample}", concurrency=32)
result = run.start(lambda r: run_program(plan["src"], AuditKit(run, sample))).wait()
# which files the scan workers saw
seen = {p for t in run.threads if t.role == "scan" for p in re.findall(r"^### (S+)$", t.messages[1]["content"], re.M)}
stats = {"agents": len(run.threads), "calls": sum(len(t.replies) for t in run.threads), "covered": seen >= set(WORK),
"credits": sum(PRICE.get(t.model, 0.2) * len(t.replies) for t in run.threads)}
return {"decision": decision, "why": why, "plan": plan, "run": run, "result": result, "stats": stats, "lead_read": sum(lead.read)}On a budget of 150 credits, the lead’s first program was priced at 263 workers and 52.9 credits, above the automatic limit of 40. My standing rule, written before I saw any estimate, approved it at under half the budget. Here is the program, trimmed to its core.
# the lead's own program, trimmed to its core
phase("Scan")
shardlist = list(shards(100000))
roundlenses = ["general", "logic", "boundaries", "state"] * 3 + ["general", "logic"] # fourteen rounds
allfindings = []
for sampleid, lens in enumerate(roundlenses):
found = flatten(parallel([lambda s=s, l=lens, r=sampleid: scan(s, l, sample=r) for s in shardlist]))
allfindings.extend(found)
candidates = dedupe(allfindings)
phase("Verify")
verdicts = flatten(parallel([lambda b=b, k=k: check(b, sample=k) for b in batches(candidates, 8) for k in range(3)]))
confirmed = vote(candidates, verdicts, need=2)
phase("Report")
result = report(confirmed)# the lead's own program, trimmed to its core
phase("Scan")
shardlist = list(shards(100000))
roundlenses = ["general", "logic", "boundaries", "state"] * 3 + ["general", "logic"] # fourteen rounds
allfindings = []
for sampleid, lens in enumerate(roundlenses):
found = flatten(parallel([lambda s=s, l=lens, r=sampleid: scan(s, l, sample=r) for s in shardlist]))
allfindings.extend(found)
candidates = dedupe(allfindings)
phase("Verify")
verdicts = flatten(parallel([lambda b=b, k=k: check(b, sample=k) for b in batches(candidates, 8) for k in range(3)]))
confirmed = vote(candidates, verdicts, need=2)
phase("Report")
result = report(confirmed)The fourteen scan rounds, the rotating lenses and the three skeptics per batch were the lead’s own choices, sized to its instruction to stay under half the budget.
<button data-code="from bench import match
from lead import audit
out = audit(live, JOB, PAGE, budget=150, sample=0, person=lambda est: est["credits"]
from bench import match
from lead import audit
out = audit(live, JOB, PAGE, budget=150, sample=0, person=lambda est: est["credits"] <= 75) # my standing rule: half the job budget
est, score = out["plan"]["estimate"], match(out["result"]["bugs"])
print(f"estimate {est['agents']} workers and {est['credits']:.1f} credits, gate: {out['decision']}")
print(f"{out['run'].status}: {out['stats']['agents']} workers, {out['stats']['calls']} calls, {out['stats']['credits']:.1f} credits at full price")
print(f"{len(out['result']['bugs'])} bugs reported, scored against the hidden key: recall {score['recall']:.2f}, precision {score['precision']:.2f} "
f"({score['tp']} planted, {score['fp']} other places)")
print(f"the lead read {out['lead_read']:,} prompt tokens in all, and every file was read by a scan worker: {out['stats']['covered']}")
#### OUTPUT ####
estimate 263 workers and 52.9 credits, gate: approved
completed: 236 workers, 247 calls, 49.7 credits at full price
50 bugs reported, scored against the hidden key: recall 0.53, precision 0.76 (38 planted, 12 other places)
the lead read 1,780 prompt tokens in all, and every file was read by a scan worker: Truefrom bench import match
from lead import audit
out = audit(live, JOB, PAGE, budget=150, sample=0, person=lambda est: est["credits"] <= 75) # my standing rule: half the job budget
est, score = out["plan"]["estimate"], match(out["result"]["bugs"])
print(f"estimate {est['agents']} workers and {est['credits']:.1f} credits, gate: {out['decision']}")
print(f"{out['run'].status}: {out['stats']['agents']} workers, {out['stats']['calls']} calls, {out['stats']['credits']:.1f} credits at full price")
print(f"{len(out['result']['bugs'])} bugs reported, scored against the hidden key: recall {score['recall']:.2f}, precision {score['precision']:.2f} "
f"({score['tp']} planted, {score['fp']} other places)")
print(f"the lead read {out['lead_read']:,} prompt tokens in all, and every file was read by a scan worker: {out['stats']['covered']}")
#### OUTPUT ####
estimate 263 workers and 52.9 credits, gate: approved
completed: 236 workers, 247 calls, 49.7 credits at full price
50 bugs reported, scored against the hidden key: recall 0.53, precision 0.76 (38 planted, 12 other places)
the lead read 1,780 prompt tokens in all, and every file was read by a scan worker: TrueThe ledger timed every call:
In the main paid session the wall clock was 41 minutes for 120 minutes of calls, and verification was the slow phase: 193 scan calls took 14 minutes, while 39 skeptic calls took 23.
The lead’s one planning call read 1,780 prompt tokens. One agent alone carried up to 91,171 in a single prompt, and the scan workers read 18.4 million between them.
With 20 or 60 credits on its sheet instead of 150, the lead’s plans cost 8.3 and 18.1 and caught 28% and 39% of the bugs.
Five-Design Evaluation
One scorer and one hidden key judge all five:
- A: one agent working alone.
- A2: that agent with a dynamic run’s budget.
- B: the lead with helpers.
- C: the hand-written fixed plan.
- D: the dynamic workflow.
#### OUTPUT ####
design runs mean recall mean precision mean credits
------ ---- ----------- -------------- ------------
A 3 0.02 1.00 23.7
A2 2 0.04 0.78 46.4
B 1 0.17 0.67 10.3
C 3 0.30 0.82 13.8
D 3 0.45 0.71 48.4#### OUTPUT ####
design runs mean recall mean precision mean credits
------ ---- ----------- -------------- ------------
A 3 0.02 1.00 23.7
A2 2 0.04 0.78 46.4
B 1 0.17 0.67 10.3
C 3 0.30 0.82 13.8
D 3 0.45 0.71 48.4The intervals are bootstrap resamples of the runs:
#### OUTPUT ####
[boot] mean recall D - A: 95% interval +0.30 to +0.56
[boot] mean recall D - A2: 95% interval +0.28 to +0.54
[boot] mean recall D - C: 95% interval +0.06 to +0.25#### OUTPUT ####
[boot] mean recall D - A: 95% interval +0.30 to +0.56
[boot] mean recall D - A2: 95% interval +0.28 to +0.54
[boot] mean recall D - C: 95% interval +0.06 to +0.25Money alone does not explain the gap: the agent given 248 turns still quit on its own.
The lead’s plan beats the fixed plan by +0.06 to +0.25 at about 3.5 times the cost. I never ran a fixed plan told to scan fourteen times, so the gain cannot be split between more reading and better planning.
Both workflows read every file, so most missed bugs were read and never reported. H1, H1b and H5 held, with a second scan round adding 5, 6 and 2 bugs.
H4 missed both bars: skeptics raised precision by 0.25, but the interval’s low end, +0.13, fell short of +0.15, and recall may have dropped by up to 0.069. H6 did not hold: skeptics scored 0.84 with the failing call and 0.85 without.
Ablation Ledger
For each feature I measured what removing it does to the number it was built for.
A second round of scanning and verification by skeptics earned their calls, while the lenses, the sandbox and a cross-generation judge came out neutral. On the 40 places of one run, a single skeptic gave precision 0.08 higher than three without losing recall, though the interval starts at +0.003.
Known Limits
Three of my eleven predictions held.
The dynamic runs were no more consistent than one agent, perhaps because each run was a different plan. They cost 3.5 times the money of the fixed plan, and 0.71 precision against 0.82. Precision is a lower bound, because every place outside the key counts as a false alarm.
Each design had three runs or fewer, on one library and one model family, and the intervals are wide. The library is public, though models asked to write its 72 functions back matched them at a median similarity of only 0.17.
What the Study Cost
Over three days the runs made 2,616 calls and spent 557.4 credits, about 29 hours of model time.
If you build on this, calibrate the blind stand-in on one real run first, and give the lead a stopping rule for its loops. The code, the data and the full notebook are in the repository.
GitHub – FareedKhan-dev/dynamic-workflow-agents: Claude and Qwen editions of Dynamic Workflows from…
Claude and Qwen editions of Dynamic Workflows from Scratch – FareedKhan-dev/dynamic-workflow-agents
