第 02 章 · 30 行手写一个最小 Harness

本篇目标：不依赖任何 Agent 框架，跑通一次完整的「模型 → 工具 → 回填 → 自停」。 写完这一章，后面 17 章都是在给这 30 行加东西。

2.1 先去掉神秘感

所有 Agent Harness 的内核都是同一个循环：

while 未完成:
    response = call_model(messages)        # 模型输出：文本 或 工具调用
    append(response)
    if response 没有工具调用: break         # 停止条件
    for each tool_call:
        result = execute(tool_call)         # 现实世界发生副作用
        append(tool_result)                 # 把结果回填给模型

DSPy、LangGraph、Claude Code、DSH 的差别，全在这五行之外：工具怎么注册、状态怎么持久化、上下文怎么压缩、执行怎么隔离、失败怎么恢复。内核没有第二种写法。

官方 DSH docs/architecture.md 的 turn flow 是同一结构的形式化版本：一个 step = 一次模型请求 + 它调用的工具；一个 turn = 零到多个 step，直到「nothing is owed」才关闭。

2.2 可运行的最小实现

零第三方依赖，只用 Python 标准库。设置 DEEPSEEK_API_KEY 后可直接运行。

# mini_harness.py —— 零依赖。DEEPSEEK_API_KEY=<key> python mini_harness.py
import json, os, subprocess, urllib.request
from pathlib import Path

ENDPOINT = "https://api.deepseek.com/chat/completions"
MODEL    = "deepseek-chat"
ROOT     = Path.cwd().resolve()          # 最小沙箱：所有文件操作限制在此目录
SYSTEM   = "You are a helpful software engineer assistant. 用工具完成任务，完成后直接给结论。"

TOOLS = [{
    "type": "function",
    "function": {
        "name": "read_file",
        "description": "读取工作区内一个文本文件的完整内容",
        "parameters": {"type": "object", "properties": {
            "path": {"type": "string", "description": "相对工作区的路径"}
        }, "required": ["path"]},
    },
}, {
    "type": "function",
    "function": {
        "name": "write_file",
        "description": "把内容写入工作区内一个文件（覆盖）",
        "parameters": {"type": "object", "properties": {
            "path": {"type": "string"}, "content": {"type": "string"}
        }, "required": ["path", "content"]},
    },
}]

def call_model(messages):
    payload = json.dumps({"model": MODEL, "messages": messages,
                          "tools": TOOLS, "tool_choice": "auto",
                          "temperature": 0}).encode()
    req = urllib.request.Request(ENDPOINT, payload, headers={
        "Authorization": f"Bearer {os.environ['DEEPSEEK_API_KEY']}",
        "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.load(r)

def safe_path(rel):
    p = (ROOT / rel).resolve()
    if not p.is_relative_to(ROOT):          # 最小但真实的边界，挡住 ../ 越界
        raise PermissionError(f"越界访问: {rel}")
    return p

def dispatch(name, args):
    if name == "read_file":
        return safe_path(args["path"]).read_text(encoding="utf-8")
    if name == "write_file":
        safe_path(args["path"]).write_text(args["content"], encoding="utf-8")
        return "OK"
    return f"未知工具: {name}"

def run(task, max_steps=12):
    messages = [{"role": "system", "content": SYSTEM},
                {"role": "user", "content": task}]
    for step in range(1, max_steps + 1):
        resp = call_model(messages)
        msg  = resp["choices"][0]["message"]
        usage = resp.get("usage", {})
        print(f"[step {step}] in={usage.get('prompt_tokens')} out={usage.get('completion_tokens')}")
        messages.append(msg)
        calls = msg.get("tool_calls") or []
        if not calls:                        # ← 唯一的停止条件
            return msg["content"]
        for c in calls:                      # 串行；只读调用本可并行，见 2.4
            fn = c["function"]
            try:
                out = dispatch(fn["name"], json.loads(fn["arguments"] or "{}"))
            except Exception as e:
                out = f"ERROR: {type(e).__name__}: {e}"   # 错误回填，让模型自纠
            messages.append({"role": "tool", "tool_call_id": c["id"], "content": out})
    return "达到 max_steps 上限（疑似死循环）"

if __name__ == "__main__":
    print(run("读 README.md，写一份 3 行摘要到 SUMMARY.md"))

2.3 逐行拆解：五行之外的六个工程决策

判据 4（强制 vs 建议）：在系统提示词里写「不要修改工作区外的文件」是建议；is_relative_to 抛异常是强制。凡是需要 100% 生效的约束，必须由代码执行，不能由模型自觉遵守。第 08 章把这条推到极致。

这也解释了 官方 SAFETY.md 为什么直说「沙箱、审批、权限控制能降低风险，但不保证隔离」。上面这个 safe_path 只防路径穿越，防不住符号链接、write_file 覆盖 .env、以及子进程联网。边界的强度取决于你实际实现了多少，不取决于你声明了多少。

2.4 逐步加料：从 30 行到生产级

按此顺序叠加，每一步都对应本系列一章。不要跳步——每加一层，都要能单独验证它修掉了哪一类失败。

2.5 与 DSH 的对照

官方 docs/architecture.md 描述的真实 turn 流程（缩写）：

turn/start
  claim next-step input + queued message
  snapshot model selection; assemble prompt sections + step-admitted tool schemas
  -> agent/pre-step                    # 可改写/拒绝本轮输入
  step/start
  agent/request -> prepareCall         # 绑定模型适配器，取消则不提交任何事
  stream -> llm/stream -> assistant/stream
  tool/call* -> tools/pre-execute -> tools/execute -> tools/post-execute -> tool/result*
  step/end
  tools owe another request, or input arrived -> next step
  -> agent/turn-stopping
turn/end

对照我们的 30 行：

这一章真正的结论：框架不提供新能力，框架提供接缝（seam）。DSH 把每个接缝都做成可替换插件，于是「加能力」不再需要改内核。第 11 章解释这在数学上为什么成立。

2.6 本章产出与验收

核心产出：mini_harness.py 跑通一次真实任务（读文件 → 写文件 → 自停），并保留一份 JSONL 运行日志。

验收点：

• 去掉 max_steps 后能人为造出一次死循环，并说清它烧了多少 token
• 把 safe_path 注释掉，用 ../ 让模型写出工作区外文件，理解「建议 ≠ 边界」
• 故意让一个工具返回错误，观察模型是否自纠；不自纠则思考错误信息该怎么写
• 能对着日志指出：哪几行是 durable 事实，哪几行只是进程内状态

延伸阅读