AI图鉴

AI图鉴首页 › AI 资讯 › 读完 DeepSeek Harness 源码,我给自己画了一张「最小 CLI A

读完 DeepSeek Harness 源码,我给自己画了一张「最小 CLI Agent」施工图

来源:掘金 · 发布时间:11-04 · 由 AI图鉴每小时自动聚合

通读 DSH 源码之后,我给自己列了份施工清单:最小 CLI Agent 该做哪几件事、7 个关键决策怎么选、12 条不变量违反了会怎样、5 个能独立跑起来的里程碑,每条都标了源码出处。

引子:两个极端,和我想走的第三条路

想搞懂 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图鉴自动抓取并翻译,版权归原作者所有;如需查看原文请点下方链接。

相关 AI 工具

相关免费羊毛

同源资讯

阅读原文(juejin.cn)

标题、摘要与正文均来自该信源的公开页面,版权归原作者所有;本站自动聚合与翻译,仅作出处跳转。

← 返回 AI 资讯列表AI工具图鉴首页

返回 AI图鉴首页 · 提交工具 · 商务合作 · 关于本站 · 609 款 AI 工具分类导航 · 每日 AI 热门项目与动态