大多数 multi-agent 系统的工作方式都属于以下两种之一：主 agent 逐个调用辅助 agent，并等待每一份报告；或者由工程师硬编码一条 pipeline，让每个任务都执行相同的步骤。一种较新的方法，即 Anthropic 引入的 dynamic workflow https://x.com/ClaudeDevs/status/2108591328732856655 ，允许主 agent 一次性将完整计划写成一个简短程序，再由普通代码使用多个 worker 并行执行。
下面是我们将逐个组件构建的架构：
一个 benchmark：包含 72 个植入的 bug，以及一个将发现结果与隐藏答案键进行比对的 scorer。
一个 answer ledger：提供缓存、回放和崩溃恢复。
一个 runtime：在同一个预算下，于多个并行阶段运行 worker。
一个 audit kit：包含扫描 worker，以及对其发现结果进行投票的 skeptic。
一种 plan language：受限 Python，在产生费用之前完成检查。
主 agent：读取一份 500-token 的 fact sheet，而不是直接读取代码。
一次 evaluation：比较五种设计，并进行一次 ablation。
最终，只读取这份 fact sheet 的主 agent，平均能够发现 0.45 个植入 bug，而手写计划为 0.30，单个 agent 独立工作时仅为 0.02。完整代码位于我的 GitHub repository：dynamic-workflow-agents。
下面是我们将要编写的文件：
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 key
为什么 Multi-Agent 系统经常失败
大多数失败源于两个限制。一个 agent 能够容纳的文本有限；面对大型任务时，它会忘记已经读过的内容，或者提前停止。逐个分发任务的主 agent 会成为瓶颈，并且它的上下文会随着每份报告的到来而不断增长。
我们的 audit 体现了这两个问题：849 个 Python 文件、约 135 万个 token，以及我亲自植入的 72 个 bug。单个 agent 独立运行三次，分别发现了其中的 0、1 和 4 个，并且每次都自行停止。带有辅助 agent 的主 agent 覆盖了全部 14 个 shard，但只发现了 12 个 bug。
设计 Dynamic Workflow
每个任务都要经过六个步骤：
Scout：普通代码读取任务，并写出一页 fact sheet。
Plan：主 agent 根据该 fact sheet 编写一个简短的 Python 程序。
Check：whitelist 拒绝计划不允许执行的操作。
Rehearse：免费的替代模型运行一次计划，以估算其成本。
Approve：approval gate 决定放行计划，或将其留给人工处理。
Run：runtime 在后台、使用一个统一预算执行计划。
在运行比较之前，我写下了十一项预测。本次 evaluation 检验其中五项：
H1：该 workflow 优于单个 agent 独立工作。
H1b：在使用相同资金的情况下，它也优于单个 agent。
H5：第二轮扫描能够增加发现的 bug 数量。
H4：skeptic 至少将 precision 提高 0.15，同时最多损失 0.05 的 recall。
H6：运行某个 finding 的 failing call 能够提高 skeptic 的准确率。
环境设置
我们从一个空项目开始，其中只有一个 package：jsonschema，以及任意 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 printed
MODELS 为每个角色指定其 model，而 PRICE 表示一次调用的 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
植入 Bug 的 Benchmark
我使用了一个包含 849 个 Python 文件的公开 library，并在其中 72 个文件里各植入了一个 bug。每个 bug 至少会导致该文件的一个 doctest 示例失败，并且失败示例从不会超过全部示例的 40%。agent 读取的副本移除了这些示例。
left out:十二种单行编辑、doctest 证明和泄漏测试（repository 的 notebook 中包含这些内容）
"""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
scorer 必须知道某一行属于哪个 function。
# 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
当 finding 指向正确的文件和 function，并且与 bug 的距离不超过五行时，该 finding 计数：
match 将其应用于 finding 列表。
# 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
我会不断重复运行 model call。我会将每条回复保存到其问题的 fingerprint 下，并且不会为同一个问题支付两次费用。
一个 Reply 保存一条答案。
"""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 是唯一接触 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)
fingerprint 包含一个 sample index：同一个问题的 sample 0 和 sample 1 会得到彼此独立的答案。
# 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
每次调用都会经过一个 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)
chat 依次尝试：ledger、replay refusal、plus stand-in、budget，最后才访问 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
我们用一个能够统计调用次数的 fake client 对其进行测试。
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 first
重复的问题由 ledger 免费处理，replay 则在未接触 network 的情况下抛出 CacheMiss。
Shard Packer
大型 shard 看起来很便宜，因为计费按调用次数进行；但对于较长输入开头之后的细节，model 可能会遗漏。shard 是一个 worker 在单次调用中读取的代码。numbered 添加行号，而 TOKENS 按 worker 实际看到的大小计算每个文件。
"""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
pack 使用完整文件填充每个 shard，直到下一个文件无法放入。
# 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
我预先固定了一条规则：选择在最佳 recall 的 0.05 范围内具有最大 size 的设置。在包含 24 个植入 bug 的 283 个文件上，该规则选择了 100,000 个 token。但之后进行的 whole-system 测试得出了不同结果：在 20,000-token shard 下，较精简的计划发现了 0.38 的 bug；在 100,000-token shard 下则为 0.18。
下面所有 workflow 的比较都使用了 100,000-token shard。
Worker Thread 与 Event Stream
一个 run 是计划的一次执行，而计划是一个接收该 run 的 Python function。
"""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
每个 worker 都会留下一个 thread，其中包含它的完整 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
Run 保存一个 run 留下的所有内容。
# 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
phase 是 run 中一个有名称的阶段，例如 Scan 或 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])
_new_thread 还负责防止 runaway。
# 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
agent() 启动一个 worker，并返回经过检查的值或 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 + "\n\n" + (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
首先从 reply 中提取 JSON。
# 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
validate 将 schema 错误转换为 model 能够修复的简短行。
# 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]]
一个 worker 的 conversation 最多包含两个 turn。
这些 turn 位于 _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]
run_sync 执行整个计划。
# 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
parallel 执行 thunks，即无参数函数，并且一次性全部执行。它是一个 barrier：直到最慢的任务完成后才返回。
在一次固定计划的 scan 阶段中，已完成的 worker 为最慢的 worker 等待了 37 个 worker-minutes。
# 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
fake client 上的一个 worker 展示了 run 记录的事件：
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 reply
一个真实 worker 读取了 100,911 个 token，并耗时 18.7 秒。
Session Budget 与恢复
所有 worker 共用同一个计量器。达到 cap 后，worker 会等待，直到人工提高 cap。
admit 会先预留一次调用的价格。
# 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
settle 对其进行计费，而 raise_to 提高 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()
start 在其独立 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
statusline 可以从任意 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 ""))
40 个 worker 获得 5 credits 的 cap，足够支付 25 次、每次 0.2 credits 的调用。
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 40
run 恰好在第 25 个 request 处停止；cap 提高后，它完成了全部 40 个 request。接下来，我在一个两阶段 run 执行到一半时取消它，然后在同一个 ledger 上再次运行相同计划。
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 all
无论是否被中断，该计划都需要 60 次调用，并且没有任何调用被重复付费。
Audit Task Kit
kit 中的三个步骤会调用 model：scan 为 worker 提供一个 shard 和一个 lens，check 为 skeptic 提供一批结果，而 report 负责写出列表。
系统会告知 skeptic：大多数报告都是误报；如果不确定，应回答 real: false。
"""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 将列表切分为多个 group，而 flatten 将列表重新合并。
# 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 构建一个 finding，而 key_of 标识其位置。
# 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 合并关于同一位置的 finding，并统计其背后的 worker 数量。
# 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
只有当足够多的 skeptic 将某个位置判定为真实时，该位置才会保留；这正是 vote 检查的内容。
# 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]
skeptic 阅读的内容来自 reports，每个 finding 对应一个 block。
# 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 "\n\n".join(blocks)
AuditKit 将这些步骤绑定到一个 run；run r 将每个 sample index 偏移 100 * r，因此三次 run 会独立采样。
# 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"\n\n{PROMPTS['lens_' + lens]}" if lens in LENSES else ""), schema=FINDINGS,
data="\n\n".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 列出计划可以调用的内容。
# 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: "\n\n".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
首先，我需要一个能够击败的计划：通过全部三个 lens 扫描每个 shard，使用三个 skeptic 检查每个 finding，并保留至少两个 skeptic 确认的结果。
"""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)"""
三次 run 分别发现了 20、23 和 22 个 bug；recall 分别为 0.28、0.32 和 0.31，每次运行约消耗 14 credits。
受限 Plan Language
我们的固定计划从不改变，但其自身记录显示，第二轮和第三轮一共发现了 9 个第一轮遗漏的植入 bug。更好的计划会持续扫描，直到某轮增加的结果很少，并在 budget() 较低时缩减规模。一个设置列表无法表达这种行为，因此由主 agent 编写 Python。
这是一种可以在执行前检查、并且只能访问所提供名称的受限 Python。其限制机制是由规则组成的 whitelist，每条规则禁止一种行为。它不是安全边界，因为这些计划来自我们自己的主 agent。
# 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."""
规则 2 和规则 4 检查 syntax。
# 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
规则 3 检查每个 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:]))
规则 1 限制计划长度，规则 5 要求其赋值给 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 只有在计划通过检查后才执行，并且其可访问范围中只有 kit 和少量 built-in。
# 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")
我编写了十四个 hostile program，checker 拒绝了全部程序，并且针对每个程序给出了 model 可以据此采取行动的原因。
Dry Run 与 Approval Gate
一个计划即使通过了全部规则，也可能成本高昂，因为它启动的 worker 数量取决于早期 worker 返回的内容。dry run 使用一个能够立即回答且不产生费用的 stand-in 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"^(C\d+) \|", text, re.M)], "summary": "a made-up report"}
return Reply(json.dumps(answer), "stop", model=model)
dry_run 按真实价格估算程序的成本，并以时间限制覆盖其运行时间。
# 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}}, []
approval gate 使用一个在任何计划定价之前就写好的 policy：成本不超过 40 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}"
我为三个计划估算了成本：固定计划、在 budget() 较低时缩小规模的计划，以及一个虽然通过所有规则却要求一百万个 worker 的 runaway 计划。
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 limit
当 rehearsal 看到的预算为 150 时，同一个程序成本为 23.9 credits；当预算为 45 时，成本为 8.3 credits。
因此，rehearsal 必须看到与真实 run 相同的预算。固定计划估算出的 10.1 credits 比其三次真实 run 的 13.8 credits 低 27%。
Lead Planner
普通代码负责编写主 agent 的 fact sheet；该 sheet 不产生费用，并且在不同 run 之间保持不变。下面是其缩略内容。
#### 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.
其中包含文件和 token 数量、一些已测量的 rate 以及价格；对于一个包含 1,348,181 个 token 的任务，整个 sheet 约为 500 个 token。
left out:构建该 sheet 的代码，约二十行字符串格式化代码
Lead 与 model 对话、保留它读取内容的计数。
"""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
write_program 请求一个计划、检查该计划、进行 rehearsal，并在最多两次修复中将所有问题反馈给 model。
# 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 JOB\n{job}\n\n{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}
我让同一个主 agent 为每个 sample index 编写十个程序。从这里开始使用真实 model，并且每个输出都是原始 run 记录的内容。该 run 还会重试失败的调用，并向每个 skeptic 展示 sandbox 结果；这两个步骤在本次构建中省略了。
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.9
其中六个在第一次尝试时就有效。需要修复的四个（tries 为 2）都违反了同一条规则，即长度限制：它们分别有 103、176、141 和 159 行，而最大限制为 80 行。十个程序都进行了循环，并根据循环结果做出决定，但只有一个读取了 budget()。
对真实 Library 进行完整 Audit
audit 依次运行每个阶段，从主 agent 的第一份草稿到后台 run。
# 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)}
在 150 credits 的预算下，主 agent 的第一个程序被估算为 263 个 worker、52.9 credits，超过了 40 credits 的自动限制。我在看到任何估算值之前就写下了自己的固定规则：只要低于预算的一半，就批准该计划。下面是经过裁剪、只保留核心部分的程序。
# 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)
十四轮 scan、轮换使用的 lens，以及每个 batch 的三个 skeptic，都是主 agent 自己的选择，并根据“保持在预算一半以内”的指令确定了规模。
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: True
ledger 为每次调用记录了耗时：
在主要的付费 session 中，wall clock 时间为 41 分钟，而调用总计耗时 120 分钟；verification 是较慢的阶段：193 次 scan 调用耗时 14 分钟，39 次 skeptic 调用耗时 23 分钟。
主 agent 的一次 planning call 读取了 1,780 个 prompt token。单个 agent 在一次 prompt 中最多携带 91,171 个 token，而 scan worker 合计读取了 1,840 万个 token。
如果 fact sheet 中的预算不是 150，而是 20 或 60 credits，主 agent 生成的计划成本分别为 8.3 和 18.1 credits，并发现了 28% 和 39% 的 bug。
五种设计的 Evaluation
一个 scorer 和一个 hidden key 对五种设计进行统一评判：
A：单个 agent 独立工作。
A2：该 agent 使用 dynamic run 的预算。
B：带有辅助 agent 的主 agent。
C：手写的固定计划。
D：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
以下区间是对各次 run 进行 bootstrap resampling 得出的结果：
#### 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
仅凭资金无法解释这一差距：即使为 agent 提供 248 个 turn，它仍然会自行停止。
主 agent 的计划以大约 3.5 倍的成本击败了固定计划，优势为 +0.06 至 +0.25。我从未运行过一个被告知要扫描十四次的固定计划，因此无法将这项收益拆分为“更多读取”与“更好的计划”分别带来的贡献。
两种 workflow 都读取了每个文件，因此大多数遗漏的 bug 都被读取过，只是没有被报告。H1、H1b 和 H5 均成立：第二轮 scan 分别增加了 5、6 和 2 个 bug。
H4 未同时满足两个要求：skeptic 将 precision 提高了 0.25，但区间下界为 +0.13，低于 +0.15；recall 可能最多下降 0.069。H6 不成立：带 failing call 时，skeptic 的得分为 0.84；不带 failing call 时则为 0.85。
Ablation Ledger
针对每项 feature，我测量了移除它对其目标指标的影响。
第二轮 scan 和由 skeptic 执行的 verification 证明了它们各自的调用成本是值得的；而 lens、sandbox 和跨 generation judge 的效果则近似中性。在一次 run 的 40 个位置上，一个 skeptic 相比三个 skeptic 在不损失 recall 的情况下，将 precision 提高了 0.08，但区间下界仅为 +0.003。
已知限制
我的十一项预测中只有三项成立。
dynamic run 并不比单个 agent 更稳定，这或许是因为每次 run 使用的计划都不同。它们的成本是固定计划的 3.5 倍，而 precision 为 0.71，固定计划为 0.82。precision 是一个下界，因为 hidden key 之外的每个位置都会被计为误报。
每种设计最多只有三次 run，并且都基于同一个 library 和同一个 model family；因此区间很宽。该 library 是公开的，但当要求 model 将其中的 72 个 function 写回去时，中位相似度仅为 0.17。
这项研究的成本
在三天内，这些 run 共进行了 2,616 次调用，消耗 557.4 credits，约等于 29 小时的 model 时间。
如果你要基于此构建系统，请先在一次真实 run 上校准 blind stand-in，并为主 agent 的循环提供停止规则。代码、数据和完整 notebook 都位于 repository 中。
https://github.com/FareedKhan-dev/dynamic-workflow-agents