HARNESS 113

Build Your Own Harness

"抄起键盘，从零搭建 LLM Agent 外壳"

s01-s12 拆解了 Harness 的每一层。s13 把它们拼回去——
跟着本节，你将从空文件开始，一步步写出一个能用的 Agent 外壳。

1

什么是 Harness？— 一张图看懂全部 12 层

Harness = 套在 LLM 外面的壳。裸的 LLM API 只能"一问一答"——你发一条消息，它回一条消息。Harness 让它变成一个能干活、能记东西、能派任务、能管团队的 Agent。

s12 · Worktree Isolation 每个任务独立目录
s11 · Autonomous Agents IDLE → scan → claim → WORK
s10 · Team Protocols request_id + FSM
s09 · Agent Teams inbox/outbox 邮箱通信
s08 · Background Tasks 线程执行 + 通知注入
s07 · Task System .tasks/ 文件持久化
s06 · Context Compact 三层压缩策略
s05 · Skill Loading tool_result 注入知识
s04 · Subagent 上下文隔离的子 Agent
s03 · TodoWrite 外部记忆 + Nag 提醒
s02 · Tool Use dispatch map 模式
s01 · Agent Loop while True + LLM call

🔑 Harness 的两个核心作用

① 给 LLM 装上手和脚：LLM 本身是一颗大脑。它只能"想"。Harness 给它装上工具（读文件、写代码、跑命令、发消息），把"想"变成"做"。

② 给 LLM 装上记忆和纪律：LLM 没有长期记忆——每次 API 调用都是全新的。Harness 给它外部记忆（TodoWrite）、持久化状态（.tasks/）、通信协议（s10）、执行约束（compaction、超时、shutdown）。

下面我们层层搭建。每一层都是可运行的代码——复制粘贴就能跑。

本节代码使用 Anthropic Python SDK：pip install anthropic。你需要一个 API Key（设置 ANTHROPIC_API_KEY 环境变量）。所有代码基于 Claude 模型，但模式适用于任何 LLM。
2

Layer 1-2：一个能干活的最小 Agent（~40 行）

这是整个 Harness 的起点——while True + LLM 调用 + 工具 dispatch。不到 50 行，但已经是一个能用的 Agent 了。

Step 1: 定义工具（函数声明 → JSON Schema）

import anthropic, subprocess, os, json client = anthropic.Anthropic() # 定义工具：用 JSON Schema 描述你的函数 TOOLS = [ { "name": "bash", "description": "执行 shell 命令并返回输出", "input_schema": { "type": "object", "properties": { "command": {"type": "string"} }, "required": ["command"] } }, { "name": "read", "description": "读取文件内容", "input_schema": { "type": "object", "properties": { "file_path": {"type": "string"} }, "required": ["file_path"] } }, { "name": "write", "description": "写入文件", "input_schema": { "type": "object", "properties": { "file_path": {"type": "string"}, "content": {"type": "string"} }, "required": ["file_path", "content"] } } ]

Step 2: 工具执行器（Dispatch Map — s02 模式）

# 工具 → 执行函数的映射字典（不是 if/elif 链！） HANDLERS = { "bash": lambda args: subprocess.run(args["command"], shell=True, capture_output=True, text=True).stdout, "read": lambda args: open(args["file_path"]).read(), "write": lambda args: (open(args["file_path"], "w").write(args["content"]), "ok")[1], } def execute_tool(tool_name, tool_input): if tool_name in HANDLERS: return HANDLERS[tool_name](tool_input) return f"未知工具: {tool_name}"

Step 3: Agent Loop（s01 的 while True 双层循环）

def agent_loop(user_prompt, max_turns=50): messages = [{"role": "user", "content": user_prompt}] system = "你是编程助手。用工具完成任务。完成任务后不要调用工具，直接输出结果。" for turn in range(max_turns): # 安全上限 # ① 调用 LLM response = client.messages.create( model="claude-sonnet-4-6", max_tokens=8000, system=system, messages=messages, tools=TOOLS, ) # ② 如果 LLM 不调工具 → 任务完成，返回文本 if response.stop_reason != "tool_use": return response.content[0].text # ③ 如果 LLM 要调工具 → 执行工具 → 把结果注入 messages → 继续循环 for block in response.content: if block.type == "tool_use": result = execute_tool(block.name, block.input) messages.append({"role": "user", "content": [ {"type": "tool_result", "tool_use_id": block.id, "content": result} ]}) return "达到最大轮次"
✅ 试着跑一下：agent_loop("在当前目录创建一个 hello.py，打印 Hello World")。
这 40 行代码已经是一个能用的 Agent。它有了大脑（LLM）、手（bash/read/write）、循环（while True）。
3

Layer 3-4：外部记忆 + 子 Agent（+35 行）

TodoWrite：让 Agent 在多步任务中不迷路（s03）

LLM 没有长期记忆。10 步以后它就忘了"刚才做到哪了"。加一个 todo_write 工具——Agent 把计划写到文件，每完成一步就回来划掉。

# 新增工具：TodoWrite {"name": "todo_write", "description": "写入/更新任务列表。用 JSON 数组格式。", "input_schema": {"properties": {"todos": {"type": "string"}}, "required": ["todos"]}}, # Handler: 把任务列表写到文件 "todo_write": lambda args: ( open(".harness/todos.json", "w").write(args["todos"]), "已保存。注意：每完成一项就把它的 status 改为 completed。" )[1],

Subagent：外包脏活，上下文不污染（s04）

主 Agent 不需要自己搜 50 个文件。派一个子 Agent——它独立跑 agent_loop，干完活只返回摘要。

# 新增工具：task（启动子 Agent） {"name": "task", "description": "启动子 Agent 执行独立任务，返回摘要", "input_schema": {"properties": {"prompt": {"type": "string"}}, "required": ["prompt"]}}, # Handler：调用 agent_loop，但 messages 是全新的（上下文隔离） "task": lambda args: agent_loop( f"{args['prompt']}\n\n完成后直接输出结果，不要调工具问问题。", max_turns=25 # 子 Agent 的轮次限制更严格 ),
现在你的 Agent 能做到：「分析这个项目用了哪些第三方库」——子 Agent 搜所有 requirements.txt 和 package.json，汇总返回。主 Agent 的上下文不会被 50 个文件目录撑爆。
4

Layer 5-6：技能加载 + 上下文压缩（+30 行）

Skill Loading：按需注入领域知识（s05）

不要把整个手册塞进 system prompt。用 skill 工具——Agent 需要什么，自己去文件系统里加载。

# 新增工具：skill — 按需加载技能文件 {"name": "skill", "description": "加载指定技能的知识文件", "input_schema": {"properties": {"name": {"type": "string"}}, "required": ["name"]}}, # Handler：读取 .harness/skills/{name}.md，把内容作为 tool_result 返回 "skill": lambda args: open(f".harness/skills/{args['name']}.md").read()

用法：在 .harness/skills/ 下放 react.md、django.md 等知识文件。Agent 需要时自己调 skill("react") 加载。

Context Compaction：上下文满了就压缩（s06）

Agent 跑了 30 轮后 messages 有几百条。加一个压缩逻辑：当 messages 超过阈值，调 LLM 把历史总结成摘要。

def maybe_compact(messages, max_messages=30): if len(messages) < max_messages: return messages # 还没满，不压缩 # 调用 LLM 总结前半部分 to_summarize = messages[:len(messages)//2] summary = client.messages.create( model="claude-sonnet-4-6", max_tokens=1000, messages=[{"role": "user", "content": f"把这轮对话总结成一段摘要：{to_summarize}"}] ).content[0].text # 替换：摘要 + 后半部分 return [{"role": "user", "content": f"[之前的对话摘要] {summary}"}] + messages[-5:]

在 agent_loop 的 for 循环开头加一行：messages = maybe_compact(messages)。

5

Layer 7-8：持久化任务 + 后台执行（+50 行）

Task System：任务图持久化到磁盘（s07）

Agent 的任务不能只活在 TodoWrite 里——重启就没了。把每个任务存为一个 JSON 文件，带 status、owner、blockedBy。

# 新建 .harness/tasks/ 目录 TASKS_DIR = Path(".harness/tasks") TASKS_DIR.mkdir(parents=True, exist_ok=True) def create_task(title, description="", blocked_by=None): task_id = len(list(TASKS_DIR.glob("task_*.json"))) + 1 task = { "id": task_id, "title": title, "description": description, "status": "pending", "owner": None, "blockedBy": blocked_by or [], "worktree": "" } json.dump(task, open(TASKS_DIR / f"task_{task_id}.json", "w")) return task

现在工具集里多了一个 task_create、task_list、task_update。任务状态从"LLM 的记忆里"变成了"磁盘上的文件"——重启不丢。

Background Tasks：慢操作丢后台（s08）

安装依赖要 2 分钟——Agent 不需要傻等。把命令丢进后台线程，完成时自动把结果注入 messages。

import threading, queue bg_notifications = queue.Queue() # 后台任务完成后结果放这里 def bg_run(command): def worker(): result = subprocess.run(command, shell=True, capture_output=True, text=True) bg_notifications.put(f"后台任务完成:\n{result.stdout}\n{result.stderr}") threading.Thread(target=worker, daemon=True).start() return "已在后台启动" # 在 agent_loop 的每轮开始：检查有没有后台结果 while not bg_notifications.empty(): messages.append({"role": "user", "content": bg_notifications.get()})
6

Layer 9-10：Agent 团队 + 通信协议（+60 行）

Agent Teams：文件邮箱通信（s09）

每人一个 JSONL 收件箱。Agent A 发消息给 B → 追加一行到 B 的 .jsonl 文件。B 每轮开始时读收件箱 → drain（读完清空）。

INBOX_DIR = Path(".harness/inbox") INBOX_DIR.mkdir(parents=True, exist_ok=True) def send_message(sender, to, content): with open(INBOX_DIR / f"{to}.jsonl", "a") as f: f.write(json.dumps({"from": sender, "content": content, "ts": time.time()}) + "\n") return f"已发送给 {to}" def read_inbox(name): path = INBOX_DIR / f"{name}.jsonl" if not path.exists(): return [] msgs = [json.loads(l) for l in path.read_text().splitlines() if l] path.write_text("") # drain return msgs

Team Protocols：request_id + FSM（s10）

让 Agent 之间的消息不再是"自由聊天"，而是可追踪的请求-响应。

import uuid _tracker = {} # {request_id: {status, target, ...}} _tracker_lock = threading.Lock() def shutdown_request(target): req_id = str(uuid.uuid4())[:8] with _tracker_lock: _tracker[req_id] = {"target": target, "status": "pending"} send_message("lead", target, json.dumps({ "type": "shutdown_request", "request_id": req_id })) return f"关机请求 {req_id} 已发送（状态：pending）" def shutdown_response(request_id, approve): with _tracker_lock: if request_id in _tracker: _tracker[request_id]["status"] = "approved" if approve else "rejected" return "ok" if approve else "rejected"

同一个模式覆盖关机、计划审批、代码审查——只要改消息类型名。request_id + pending→approved/rejected FSM 是通用的。

7

Layer 11-12：自主认领 + Worktree 隔离（+40 行）

Autonomous Claim：Agent 自己找活干（s11）

Agent 干完活别傻站着——扫描任务池，看到 unclaimed + unblocked 的任务就认领。

def scan_and_claim(agent_name): for f in sorted(TASKS_DIR.glob("task_*.json")): task = json.loads(f.read_text()) if (task["status"] == "pending" and not task.get("owner") and not task.get("blockedBy")): task["owner"] = agent_name task["status"] = "in_progress" json.dump(task, open(f, "w"), indent=2) return task # 返回第一个能做的 return None # IDLE 阶段（agent_loop 结束后）：每 5s 扫描一次 for _ in range(12): # 12 × 5s = 60s time.sleep(5) task = scan_and_claim(name) if task: agent_loop(f"执行任务: {task['title']}\n{task['description']}") break

Worktree Isolation：每个任务独立目录（s12）

认领任务后，创建一个 git worktree——在隔离目录里干活，不跟别人冲突。

WT_DIR = Path(".harness/worktrees") def create_worktree(task_id): name = f"task-{task_id}" wt_path = WT_DIR / name subprocess.run(["git", "worktree", "add", str(wt_path), "-b", f"wt/{name}", "HEAD"]) os.chdir(wt_path) # 切换工作目录 return wt_path def cleanup_worktree(task_id, complete=True): name = f"task-{task_id}" wt_path = WT_DIR / name if complete: subprocess.run(["git", "-C", str(wt_path), "add", "."]) subprocess.run(["git", "-C", str(wt_path), "commit", "-m", f"Task #{task_id} done"]) subprocess.run(["git", "worktree", "remove", str(wt_path), "--force"]) os.chdir(PROJECT_ROOT) # 切回主目录
8

完整的 harness.py 文件结构

把所有层组装在一起——这是最终的文件树和代码骨架。

# 你的项目目录 my-agent-project/ ├── harness.py ← 主文件（~400 行） ├── .harness/ │ ├── config.json ← 模型、工具配置 │ ├── skills/ │ │ ├── react.md ← Agent 可加载的知识文件 │ │ ├── python.md │ │ └── testing.md │ ├── tasks/ │ │ ├── task_1.json ← 持久化任务图 │ │ ├── task_2.json │ │ └── ... │ ├── inbox/ │ │ ├── lead.jsonl ← Agent 收件箱 │ │ ├── alice.jsonl │ │ └── bob.jsonl │ ├── worktrees/ │ │ └── index.json ← worktree registry │ └── events.jsonl ← 生命周期事件流 ├── .claude/ ← (可选) Claude Code 风格 │ ├── settings.json │ └── CLAUDE.md └── src/ ← 你的项目代码（Agent 在这里干活）

harness.py 主结构

#!/usr/bin/env python3 """Harness — LLM Agent 外壳。从零搭建，跟着 s01-s12 层层叠出来。""" # ═══ 基础层：Agent Loop + Tool Dispatch (s01-s02) ═══ import anthropic, subprocess, os, json, threading, uuid, time, queue from pathlib import Path client = anthropic.Anthropic() PROJECT_ROOT = Path.cwd() HARNESS_DIR = PROJECT_ROOT / ".harness" # ── 工具定义 ── TOOLS = [...] # JSON Schema 工具列表 HANDLERS = {...} # 工具名 → 执行函数 # ── Agent Loop ── def agent_loop(prompt, max_turns=50, name="main"): messages = [{"role": "user", "content": prompt}] for turn in range(max_turns): check_inbox(name, messages) # s09 check_bg_notifications(messages) # s08 messages = maybe_compact(messages) # s06 response = client.messages.create(...) if response.stop_reason != "tool_use": return response.content[0].text for block in response.content: if block.type == "tool_use": result = HANDLERS[block.name](block.input) append_tool_result(messages, block, result) # ═══ 记忆层：TodoWrite + Subagent (s03-s04) ═══ def todo_write(todos): ... def subagent(prompt): return agent_loop(prompt, max_turns=25) # ═══ 知识层：Skill + Compaction (s05-s06) ═══ def load_skill(name): ... def maybe_compact(messages): ... # ═══ 任务层：Task System + Background (s07-s08) ═══ def create_task(title, **kwargs): ... def bg_run(command): ... # ═══ 团队层：Teams + Protocols (s09-s10) ═══ def send_message(sender, to, content): ... def read_inbox(name): ... def shutdown_request(target): ... # ═══ 自主与隔离层：Autonomous + Worktree (s11-s12) ═══ def scan_and_claim(agent_name): ... def create_worktree(task_id): ... def cleanup_worktree(task_id): ... # ═══ 启动 ═══ if __name__ == "__main__": agent_loop(input("> "))
9

从零跑到你的第一个 Agent

Step-by-step：跟着做

  • ① 创建项目目录：mkdir my-agent && cd my-agent && git init
  • ② 安装依赖：pip install anthropic
  • ③ 设置 API Key：export ANTHROPIC_API_KEY="sk-ant-..."
  • ④ 复制上面的代码：从 Layer 1 开始，粘贴到 harness.py
  • ⑤ 逐层添加：每加一层就测一下——python harness.py
  • ⑥ 创建 .harness/skills/：放几个 .md 知识文件
  • ⑦ 让 Agent 干活：agent_loop("分析当前目录的代码结构，总结到 ARCHITECTURE.md")

🧪 手工验证清单：确保每层都工作

Layer怎么验证预期结果
1-2: Loop+Tools「创建一个 test.txt 写 hello」文件创建成功，内容正确
3: TodoWrite「帮我规划一个登录页面，列出 3 个步骤」.harness/todos.json 有 3 项，完成一条变 completed
4: Subagent「搜一下项目里有没有用 React」子 Agent 启动→搜索→返回结果，主 Agent 上下文不膨胀
5: SkillAgent 调 skill("react")tool_result 返回 react.md 的内容
6: Compact让 Agent 做 40 轮操作messages 被压缩到摘要 + 最后几条
7: Tasks「创建 3 个任务，列出所有」.harness/tasks/ 下有 3 个 JSON 文件
8: Background「在后台安装 requests 包然后告诉我」Agent 继续干活，安装完成后自动通知
9: TeamsAlice 发消息给 Bobbob.jsonl 有新行，Bob 循环读到后回复
10: ProtocolsLead 发 shutdown_request 给 Alice_tracker 状态 pending→approved，Alice 优雅退出
11: Autonomous创建 2 个 pending 任务，Agent idleAgent 每 5s 扫描，自动 claim 并执行
12: WorktreeAgent claim 任务自动创建 worktree，cwd 切换到隔离目录，完成后 commit + cleanup

🔑 最关键的 3 个设计原则（记住这些就够了）

① 文件即状态：不要依赖 LLM 的记忆。TodoWrite、.tasks/、inbox/、events.jsonl——所有状态都在磁盘上。崩溃不怕。

② 隔离代替协调：subagent（上下文隔离）、worktree（文件系统隔离）、inbox（通信隔离）。能隔离的就不要协调。

③ 渐进式复杂度：不要一次性写 400 行。从 40 行 agent_loop 开始，跑通了再加一层。每加一层都要能独立验证。

📖 从 Harness 到生产系统：后续方向

你手上这个 harness.py 是一个教学级的 Agent 外壳。如果要用于生产，下一步是：

① 安全沙箱：用 Docker/虚拟机包裹 bash 工具——Agent 不能 rm -rf /。

② 流式输出：client.messages.create(stream=True)——让用户看到 Agent 的实时思考过程。

③ 权限系统：s10 的 plan_approval 可以升级为多级审批——高风险操作需要多层确认。

④ 多模型支持：HANDLERS 不变，只换 client 和 tools JSON Schema——适配 OpenAI、Gemini、本地模型。

⑤ 监控与遥测：events.jsonl 接入 Grafana/Prometheus——看 Token 消耗、任务成功率、Agent 响应时间。

但核心骨架不会变。while True + dispatch map + 持久化状态 + 隔离执行——这四个东西是所有 Agent 系统的共同底座。

🎉 13 节课全部完成。
s01-s12 拆解了 Claude Code 的每一层。
s13 教会你从零搭一个自己的。

工程不是学完的，是改完的。动手吧。

s13 是最终的实战篇章。全部 13 节课已完成。
返回首页，回顾完整体系。

🏠 返回课程首页