引子:两个极端,和我想走的第三条路
想搞懂 Agent 到底怎么跑起来,网上给的路一般是两个极端。
一个是"30 行代码手写 Agent",你照着抄完,它能跑,但你不知道它为什么能跑——没有日志、没有并发、没有权限,一旦模型行为怪一点,你就只能靠 print 猜。
另一个是直接读一个成熟框架。比如 DeepSeek Harness(下称 DSH)开源的这个:Cordis 插件内核、20 多个工具、会话日志、沙箱、子 agent、PTY、遥测……读完你会觉得"太强了",然后更不敢自己写了。
我想要的是第三条:一条能自己解释清楚的最小 agent 循环。
读输入 → 拼请求 → 流式收 → 装配 tool_call → 执行工具 → 结果进历史 → 再发一次
加上两件东西:一份事件日志作为唯一真源,一道权限/沙箱作为闸门。就这些。
判断做没做成的标准也只有一条:任意一次对话,我都能从日志文件里重放出模型当时看到的每一个字节。
下面这份东西,是我通读 DSH 源码(MIT 协议,可以直接 clone)之后给自己列的施工图。所有"DSH 是怎么做的",都逐条核对过源码并标了 path:line,行号基于我读的那次 master 快照,仅供参考。
1. 先划边界,这一步比写代码重要
1.1 要做的(MVP 必备)
- 单进程 REPL,流式输出,Ctrl-C 能取消当前这一轮
- 一个 OpenAI 兼容的模型适配层(换个 baseURL 就换 provider)
- 5 个工具:read / write / edit / bash / glob+grep
- 追加式 JSONL 会话日志 + 从日志推导模型历史 + --resume
- 三档权限模式 + 危险操作单次审批(fail-closed)
- 输出截断 / 单条工具结果裁剪 / token 估算
1.2 明确不做的(写进 README,防止自己手痒)
1.3 一个反直觉的发现:最小 ≠ 安全
DSH 官方有个 sdk-minimal profile,是唯一不叠加 base 的完整树,只有 37 行。但它干了件很有意思的事:
它的 tool-fs 把 read/write/edit 全关掉了(只留 readImage),文件操作用一个持久 bash 代劳;权限是 danger-full-access。
(packages/bundle/sdk-minimal/cordis.patch.yml:44-50, :73, :144)
也就是说,官方的"最小"是用安全性换来的——它砍到只剩一个 shell,然后在 shell 里裸奔。
这给了我第一个明确判断:别照抄。 我的版本应该是"工具多两个、权限多一道",代价大约是 150 行代码。
2. 七个决策,以及它们的反悔成本
做小项目最怕的不是选错,是选了一个三周后改不动的选项。所以每个决策我都标了"反悔成本"。
D3 值得单独说:为什么不能拿 messages 数组当状态
所有人写 Agent 的第一版都长这样:内存里维护一个 messages: list,每轮 append,退出时 dump 成 JSON。这个写法三天内绝对没问题,三周后你会同时遇到三个 bug:
- --resume 之后模型看到的上下文和上次退出时不一样
- 工具结果和 call_id 对不上,provider 直接 400
- 想看"当时到底发了什么",只能去翻终端 scrollback
DSH 用一整节讲这句话(docs/architecture.md:128):
Model-visible means logged. 任何进入请求的东西,必须先是一个日志事件。
我的做法是把日志当数据库、把历史当视图:每次请求前,从事件流投影出 messages,投影完就扔,不缓存。多花 O(n) 的时间,换掉一整个 bug 类别。
3. 架构:10 个模块,约 1000 行
┌─────────────┐
stdin ──────────►│ main.py │ REPL / argparse / SIGINT
└──────┬──────┘
│ user 输入
┌──────▼──────────────────────────────┐
│ agent.py turn/step 状态机 │
│ ┌─────────────────────────────────┐ │
│ │ while: 组装→请求→装配→执行→记日志 │ │
│ └─────────────────────────────────┘ │
└──┬──────────┬──────────┬────────────┘
│ │ │
┌────────▼──┐ ┌────▼─────┐ ┌──▼─────────┐
│ prompt.py │ │ llm.py │ │ tools.py │
│ 系统提示 │ │ SSE+装配 │ │ 注册表+管线 │
└───────────┘ └────┬─────┘ └──┬─────────┘
│ │
┌─────▼──────┐ ┌─▼────────┐
│ (网络) │ │ batch.py │ 独占 barrier + 顺序提交
└────────────┘ │ policy.py│ 权限判定
└─┬────────┘
│
┌─────────▼─────────┐
│ base_tools.py │ read/write/edit/bash/glob
└───────────────────┘
┌──────────────────────────────────────────────┐
│ events.py 唯一真源:append / derive_messages │
│ / JSONL 读写 / 崩溃尾部修复 │
└──────────────────────────────────────────────┘
compact.py(M5 才加)
依赖方向必须单向:
main → agent → {llm, tools, prompt} → {batch, policy} → base_tools
所有模块 → events(events 不 import 任何东西)
3.1 事件日志长什么样
@dataclass
class SessionEvent:
type: str # turn/start | user | assistant | tool/call | tool/result | turn/end
seq: int
time: float
data: dict
class SessionLog:
def __init__(self, path):
self.path, self.seq = path, 0
def append(self, type_: str, **data) -> dict:
self.seq += 1
ev = {"type": type_, "seq": self.seq, "time": time.time(), "data": data}
with open(self.path, "a", encoding="utf-8") as f:
f.write(json.dumps(ev, ensure_ascii=False) + "\n")
return ev # 校验 + seq 连续 + flush,都在这里做
一行一事件,追加写,人眼可读。出事的时候 tail -f 就是调试器。
3.2 投影:历史是日志的视图
def derive_messages(events: list[dict]) -> list[dict]:
out = []
for ev in events:
t, d = ev["type"], ev["data"]
if t == "user":
out.append({"role": "user", "content": d["text"]})
elif t == "assistant":
if not d["blocks"]: # 不变量 10:空 content 不投影
continue
out.append({"role": "assistant", "content": d["text"],
"tool_calls": d["tool_calls"]})
elif t == "tool/result":
out.append({"role": "tool", "tool_call_id": d["call_id"],
"content": d["content"]})
return out
注意这段代码里没有排序——因为日志里 tool/result 本来就已经按模型顺序落盘了(见不变量 3)。"提交串行"这条约束的价值,在这里直接体现成一个更简单的投影函数。
4. 必须做对的 12 条不变量
这一节是整篇文章我最想写给自己的部分。下面每一条,都是 DSH 里被真实踩出来的坑,不是风格偏好。违反了就是 bug。
4.1 最容易写错的是 5/6/7/8,都在流式那一层
装配器的核心就三件事:按 index 分组、参数拼接而不是赋值、关闭只认第一次。
class BlockAssembler:
def __init__(self):
self.blocks: dict[int, dict] = {}
self.closed: set[int] = set()
def feed(self, chunk):
i = chunk["index"]
if i in self.closed: # 不变量 5:block-end 先到者胜
return
b = self.blocks.setdefault(i, {"type": chunk["type"], "text": "", "args": ""})
if chunk["kind"] == "text-delta":
b["text"] += chunk["delta"]
elif chunk["kind"] == "tool-call-delta":
b["args"] += chunk["arguments_delta"] # 拼接,不是 =
elif chunk["kind"] == "block-end":
self.closed.add(i)
然后在 finish 那一步做丢弃:
tool_calls = assembled_tool_calls()
if finish.kind == "max-tokens":
tool_calls = [] # 不变量 6:半截参数一律不执行
if finish.kind == "stop" and not blocks:
raise LLMError("EMPTY_RESPONSE") # 不变量 8:空响应当错误,不当成功
第 6 条我是从源码里"读到"的,但真正理解它是在自己想明白一件事之后:模型被截断时给的 tool_call,语法上可能完全合法。{"path": "/etc/hos 是明显残缺的,但 {"path": "/etc"} 是完整 JSON——它只是不是你想要的。所以判据不能是"JSON 能不能解析",只能是"这一轮是不是被 max-tokens 掐断的"。
4.2 两条文档没写、但一定会咬人的
- timeoutMs 写在工具定义里只是声明,管线不强制(packages/core/tools/README.md:205)。→ 必须自己在 dispatch 层套 subprocess.run(timeout=...),否则一个 tail -f 就能挂死整个 agent。
- turn 没有内置预算,死循环的唯一出口是你在 turn_stopping 处计数取消(packages/core/agent-loop/README.md:220)。→ 自己加 max_steps_per_turn。模型会非常有耐心地"再读一次同一个文件",直到你的钱包替它转圈。
5. 分阶段:每个里程碑都能独立跑起来
这是我认为最该学的一条工程习惯:不要按模块做,按"能跑起来的东西"做。
M0 · 骨架(约 2 小时)
一个能读 stdin、能打印、Ctrl-C 能退的壳。任务:main.py 的 REPL + argparse、events.py 的 append / JSONL 落盘、一个假的"回声模型"替身。
验收:python -m miniagent -p "hi" 打印 hi,并且 .miniagent/sessions/<id>.jsonl 里有 4 条事件(turn/start、user、assistant、turn/end)。
M1 · 真实模型 + 流式(1~2 晚)
单轮问答跑通,流式打印,无工具。任务:llm.py 的 SSE 客户端 + BlockAssembler(先只处理 text-delta 和 finish)、错误码归一(429/RATE_LIMIT、CONTEXT_WINDOW_EXCEEDED、EMPTY_RESPONSE)、prompt.py 生成 system 文本并落成节点 0。
验收:-p "用三句话解释什么是 SSE" 逐字流式输出;日志里 request/header 记下了 provider / model / contextWindow。
这一阶段最容易翻车:不变量 5、7、8。
M2 · 工具循环闭环(2~3 晚,核心)
模型能真的读文件、跑命令、改文件。顺序很重要:
- tools.py 注册表 + schema 渲染进请求
- base_tools.py:先只做 read + bash(最常用、最容易验证)
- assembler 补 tool-call-delta 装配
- batch.py:先纯串行实现(不上池),但保留按顺序提交的骨架
- agent.py:step 循环 + 六类终止出口 + 失败补 result
- 再补 write / edit / glob
验收(全过才算完成):
- -p "读一下 README.md,告诉我项目是干什么的" → 真的调了 read
- -p "把 foo.py 里的 bar 改成 baz" → 调了 read 然后调了 edit
- -p "运行 false 然后告诉我结果" → 错误作为 tool_result 回到模型,模型能继续,turn 不崩
- 拔网线 / 杀进程后 --resume:历史完整,无 dangling tool_result
M3 · 自保(1~2 晚,最容易跳过但最该做)
目标很朴素:它不会在我睡着的时候删库。
任务:policy.py 三档 mode(read-only / workspace-write / danger-full-access)+ 路径前缀检查;ask 单次审批(stdin 一次问答,超时算 unavailable → deny);输出截断(单条工具结果 > 8192 字符裁成 head 4096 + tail 1024 + 截断提示);max_steps_per_turn;Ctrl-C → cancel(signal) 并给未派发调用补 ABORTED_BEFORE_DISPATCH。
验收:--mode workspace-write 下执行 -p "删除 ~/Documents 下所有文件" → 被 deny 且 turn 正常结束;--mode read-only 下 write 一律拒。
参照的 DSH 数值:pruner 的 thresholdChars: 8192 / headChars: 4096 / tailChars: 1024(packages/bundle/base/cordis.patch.yml:421-425);read 工具上限 2000 字符/行、50KB/次(packages/fs/tool-fs/src/read-render.ts:11,14)。
M4 · 体验(1 晚)
渲染(工具调用折叠成一行、diff 高亮)、/compact /new /model 斜杠命令、每轮结束打印 token 用量与累计花费、会话标题、--json 事件流(给脚本用)。
验收:日常不需要看原始日志,也能读懂它在干嘛。
M5 · 可选增强
compaction 摘要(阈值 0.8 × contextWindow,保留尾部 0.16)、重复调用软提醒(3/5/8 次)、子 agent、skill 目录、MCP。
6. 高频陷阱 Top 10(按"坑掉的时间"排序)
- 把 messages 数组当状态。写三周后发现 resume、取消、失败恢复全要改。→ 事件日志才是唯一真源。
- 工具调用参数拼接错。argumentsDelta 是累加,不是赋值;content_block_delta 里的 input_json_delta 同理。
- tool_call 没有配对 result。异常路径忘了写 result,下一轮直接 400。→ 用 try/finally 保证每个 call 都落地。
- 并发默认开着。write 和 edit 同时跑。→ 默认 exclusive。
- 结果按完成顺序 append。模型把 A 的结果当成 B 的。→ 池子并发,提交串行。
- max-tokens 截断了还去执行工具。参数是 {"path": "/etc/hos。
- usage 放在 finish 之后发。账算不准。
- 把空响应当成功。HTTP 200 但内容为空,用户看到"完成",实际什么都没变。
- 审批时"没人回答"当 allow。等于静默全放行。→ 只认 allowed-once。
- 没加 max_steps_per_turn。模型卡在"再读一次同一个文件",你的钱包替它转圈。
7. 验收:怎么算"能自用了"
全打勾再宣布完工:
- -p "..." 单次任务能跑完,退出码正确(0 完成 / 1 失败)
- 流式输出逐字可见,不是等完整响应才打印
- 能读、能改、能执行命令,且三个工具能在一次任务里串起来
- 工具报错时模型能看到错误并自己纠正一次
- Ctrl-C 立刻停,历史里留下 interrupted 标记,后续请求不炸
- 杀进程 → --resume → 继续对话,模型看到的历史与中断前一致
- 越界写(workdir 外)被拒绝,且拒绝原因写进日志
- 单条超长输出被截断到阈值内,模型能感知"这里有截断"
- 一次会话的日志文件能独立读懂(人眼可读的事件流)
- 单轮步数上限触发时,明确告诉模型 / 用户,而不是静默卡死
8. 附录 A:DSH 的 minimal profile 到底挂了什么
那 37 行归纳成 12 类:应用外壳(2) / 官方遥测侧载(3) / 模型路由(1) / 模型运行时(2) / 会话日志(4) / agent 内核(3) / 工具注册表与执行器(2) / 提示词装配(1) / 沙箱与策略(2) / 文件附件工作目录(5) / 子进程与持久 shell(4) / 上下文与后台任务(6)。
- 该抄的:会话日志四件套(事件日志 + 投影 + JSONL + checkpoint)、agent 内核三件套(agent / agent-loop / timer)、工具注册表与执行器的分离。
- 不必抄的:JSON-RPC 外壳、遥测侧载、MCP resources、image offload、session-title、jobs、持久 PTY(普通 subprocess 起一次 shell 就够)。
- 反着抄的:danger-full-access 默认值、tool-fs 全关、没有审批行。
DSH 有而我没有(并且我要清楚自己没有):compaction、spill(超过 12500 token 转存文件)、repeat-tool-reminder、timeout-policy、checkpoint barrier、OS 级沙箱(bwrap / Landlock / Seatbelt)。
前四个可以用 30 行粗糙替代,最后一个不能。因为"路径前缀检查"根本挡不住 bash -c "cd / && rm -rf ~"。所以 workspace-write 模式下对 bash 应当始终 ask,而不是靠解析命令去猜意图。
这是我从这次读源码里得到的最重要的一条认知:你没有 OS 级隔离的时候,不要在应用层假装自己有。
9. 附录 B:一条 90 分钟的最小验证路径
如果不想一上来就写正式代码,先用一天里的 90 分钟写个一次性的 150 行脚本:
读 stdin → 拼一个 system + user → POST /v1/chat/completions (stream=true)
→ 逐块 print → 结束
跑通它你就知道:网络层没问题、模型能连上、SSE 到底长什么样。
然后把它扔掉,按 §3 的模块重写。
不要在那个 150 行脚本上迭代——它注定会变成一个没有日志、没有工具、没有权限、不可维护的东西。我见过太多"再改改就能用"的脚本死在这个阶段。
最后
写这份东西的过程里,最有价值的不是那 27 条任务,而是那 12 条不变量。任务清单会随着实现变化,而不变量是"你一旦搞错就得重写"的那一层。
如果只能带走一句话:
Agent 的复杂度不在循环里,在循环之外那些"模型看到的东西必须等于日志里的东西"的约束里。
(DSH 是 MIT 协议开源项目,本文所有源码引用都可以自己 clone 下来核对;行号基于我阅读时的 master 快照,版本变动后可能偏移。)
本文由 AI图鉴自动抓取并翻译,版权归原作者所有;如需查看原文请点下方链接。