LLM之Agent（六十一）｜拆解 Coding Agent 的 harness：从零构建你的第一个 AI 编程助手

1. 为什么你的 Coding Agent 总在第三步崩掉：harness 层缺失的典型症状

很多人第一次写 AI 编程助手，代码大概长这样：一个 while 循环，把用户输入丢给模型，模型返回 tool_call 就执行，执行完把结果塞回 messages，再循环。跑 demo 没问题，一旦让它改一个真实仓库里的文件，问题就来了——它会在第三步或第四步开始重复读同一个文件、忘记前面已经改过什么、把cd之后的路径当成永久生效、甚至在等你确认的时候把整个上下文烧光。

这些症状看起来像"模型不够聪明"，实际上几乎全部出在 harness 层。所谓 harness，中文可以理解成"挽具"或"编排脚夫"，它是包在模型外面那一圈基础设施：状态机、上下文管理、权限门、执行隔离、事件流。模型权重和核心 API 行为是固定的，你能工程化的部分几乎全在 harness 里。

我拆过 Claude Code、OpenCode、Pi 这类主流 coding agent 的实现，得出一个反直觉的结论：真正让 agent 可用的，不是那个 ReAct 循环，而是循环外面的东西。一个 bare agent loop 大概 20 行就能写完，但一个能跑真实项目的 harness 需要处理 phase machine、steering queue、permission gate、sandbox、context compaction、memory 注入、可观测性这一整套。

这篇文章面向想理解 Agent 如何调度工具与上下文的开发者。我会带你把 harness 拆成可复制的配置片段，最后用一个端到端请求验证整条链路真的通了。你不需要先读完所有源码，跟着配置走一遍，黑箱就透明了。

先明确边界。Agent 层负责"想"：模型评估状态、选择 action 或 tool call、接收 observation、迭代。Harness 层负责"活"：它驱动 turn 的执行、管理输入队列、拦截危险操作、隔离命令执行、压缩上下文、把事件流分发给界面。Interface 层负责"看"：TUI 或 headless 远程执行。三层分离之后，你换模型、换界面、换沙箱，harness 逻辑都不用重写。

下面这张对照表帮你快速定位自己卡在哪一层：

看清这张表，你就知道接下来该配什么。

2. TaoToken 前置：给 harness 一个稳定的模型入口

harness 要跑起来，第一件事是让模型调用这条链路稳定。我试过把模型入口写死在代码里，结果每次换模型都要改源码、重跑测试，非常痛苦。正确做法是把 Base URL、Key、Model ID 三件套抽成配置，harness 只读配置不关心供应商。

TaoToken 在这里的角色是提供统一的模型调用入口。官网地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 端点是 https://taotoken.net/api 。注意 API 地址不带 UTM 参数，配置里写干净的 endpoint 就行。

你需要准备三样东西，缺一不可：

Base URL：https://taotoken.net/api，这是所有请求的前缀，harness 里的 model client 指向它。

API Key：在控制台创建，形如sk-开头的一串。这个 Key 只放在环境变量或本地配置文件里，绝对不要提交到 git。我见过有人把 Key 写进settings.py然后推到公开仓库，十分钟内就被扫走了。

Model ID：具体调用哪个模型。harness 的配置里要显式声明，不要依赖默认值，否则换环境时行为会漂移。

获取 Key 的入口在控制台的 API Keys 页面，地址是 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。创建之后复制一次，页面刷新就看不到了，先存到本地.env。

如果你只是想先验证模型能不能通，不想写代码，可以用模型对话页面直接发一条消息，地址是 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。这一步能帮你排除"是 Key 错了还是 harness 写错了"的干扰。

对于长期跑编码任务或 Agent 工作流的场景，Coding Plan 更合适，入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。它的计费方式对高频 tool call 更友好，因为 coding agent 一个 turn 可能触发十几次模型请求，按次计费会很难受。

接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，里面有各语言的调用示例。我建议你先照着文档跑通一个最小请求，再把它塞进 harness。

这里有个容易踩的坑：harness 里的 model client 通常需要兼容 OpenAI 风格的/chat/completions或 Anthropic 风格的/messages。TaoToken 的 API 端点支持标准协议，你在配置里把 base_url 指对，剩下的交给 SDK。不要自己手写 HTTP 拼接，容易在 header 和 body 格式上出错。

环境变量建议这样组织，harness 启动时统一读取：

# .env 本地文件，不要提交 TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_API_KEY=sk-你的key TAOTOKEN_MODEL_ID=你的模型ID

然后在代码里用os.environ或pydantic-settings读取。这样你的 harness 代码里不会出现任何硬编码的 Key，换环境只改.env。

3. 可复制配置：把 harness 三件套写进 settings

这一节给你可以直接抄的配置片段。harness 的配置分三块：模型入口、权限门、沙箱。我按文件路径组织，你照着建目录就行。

先建项目结构：

my-agent/ config/ settings.toml permissions.json src/ harness/ runner.py queue.py gate.py agent/ loop.py

模型入口配置写在config/settings.toml。TOML 比 JSON 更适合写配置，因为支持注释：

# config/settings.toml [llm] base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" # 从环境变量读，不写明文 model_id = "你的模型ID" timeout_s = 120 max_retries = 3 [harness] # 上下文窗口预算，compaction 阈值基于它计算 context_window_tokens = 200000 # 保留最近多少 token 不压缩 keep_recent_tokens = 40000 # 触发 microcompaction 的预留比例 microcompaction_reserve_fraction = 0.15 # 触发完整 compaction 的预留比例 compaction_reserve_fraction = 0.20 [sandbox] # none | docker | modal mode = "docker" image = "python:3.12-slim" exec_timeout_s = 60

权限门配置写在config/permissions.json。规则顺序很重要：先走 deny，再走 allow，最后落到 mode 默认行为：

{ "mode": "default", "rules": [ { "match": { "tool": "bash", "command_regex": "rm\\s+-rf\\s+/" }, "decision": "deny", "reason": "禁止删除根目录" }, { "match": { "tool": "bash", "command_regex": "git\\s+push" }, "decision": "ask", "reason": "推送需要人工确认" }, { "match": { "tool": "read" }, "decision": "allow" }, { "match": { "tool": "glob" }, "decision": "allow" }, { "match": { "tool": "grep" }, "decision": "allow" }, { "match": { "tool": "write" }, "decision": "ask" }, { "match": { "tool": "edit" }, "decision": "ask" } ] }

注意mode字段。default模式下，只读工具自动放行，写文件和 bash 需要确认；edit模式下文件编辑自动放行，bash 仍然要问；bypass模式全部放行，只用于 headless 自动化，绝不能在有真实凭证的环境里开。

harness 读取配置的代码长这样，用 pydantic 做校验，字段缺失直接报错而不是静默用默认值：

# src/harness/config.py from pathlib import Path import json import tomllib from pydantic import BaseModel, Field class LLMConfig(BaseModel): base_url: str api_key_env: str model_id: str timeout_s: int = 120 max_retries: int = 3 class HarnessConfig(BaseModel): context_window_tokens: int = 200_000 keep_recent_tokens: int = 40_000 microcompaction_reserve_fraction: float = 0.15 compaction_reserve_fraction: float = 0.20 class SandboxConfig(BaseModel): mode: str = "none" image: str = "python:3.12-slim" exec_timeout_s: int = 60 class Settings(BaseModel): llm: LLMConfig harness: HarnessConfig sandbox: SandboxConfig def load_settings(root: Path) -> Settings: with open(root / "config" / "settings.toml", "rb") as f: raw = tomllib.load(f) return Settings(**raw) def load_permissions(root: Path) -> dict: with open(root / "config" / "permissions.json", "r", encoding="utf-8") as f: return json.load(f)

如果你用的是 Claude Code 或 Cline 这类现成工具，配置位置不一样，但三件套逻辑相同。Claude Code 的 settings 里要写ANTHROPIC_BASE_URL、ANTHROPIC_API_KEY、ANTHROPIC_MODEL；Cline 的 MCP 配置里要写baseUrl、apiKey、model。不管哪个工具，Base URL、Key、Model ID 三个字段一个都不能少，少一个就会在第一次请求时报 401 或 model not found。

CC Switch 这类多配置切换工具也遵循同样结构。它的配置文件里每个 profile 就是一组三件套，切换 profile 等于换模型入口。如果你同时用多个模型做对比，这个结构能省很多事。

配置写完，先别急着跑 agent。用一段最小代码验证模型入口通不通：

# scripts/check_llm.py import os from openai import OpenAI from pathlib import Path from src.harness.config import load_settings settings = load_settings(Path(".")) client = OpenAI( base_url=settings.llm.base_url, api_key=os.environ[settings.llm.api_key_env], ) resp = client.chat.completions.create( model=settings.llm.model_id, messages=[{"role": "user", "content": "只回复两个字：通了"}], timeout=settings.llm.timeout_s, ) print(resp.choices[0].message.content)

跑python scripts/check_llm.py，输出"通了"就说明模型入口没问题。这一步能帮你把模型问题和 harness 问题彻底分开。

4. 端到端验证：一次请求看清执行链路

配置就绪后，跑一次完整的 turn，观察事件流。harness 的价值在于把黑箱变成可观测的事件序列。我设计一个最小验证场景：让 agent 读一个文件、改一个文件、跑一条命令，全程打印事件。

先写事件定义。事件是 frozen 且 hashable 的，这样 TUI 和远程可观测性可以共用同一个真实来源：

# src/harness/events.py from dataclasses import dataclass from typing import Union @dataclass(frozen=True) class TurnStarted: turn_id: str @dataclass(frozen=True) class AssistantTextDelta: text: str @dataclass(frozen=True) class ToolCallStarted: tool: str args: dict @dataclass(frozen=True) class ToolResult: tool: str ok: bool preview: str @dataclass(frozen=True) class PermissionRequested: tool: str reason: str @dataclass(frozen=True) class ContextCompacted: before_tokens: int after_tokens: int @dataclass(frozen=True) class TurnFinished: turn_id: str stop_reason: str Event = Union[ TurnStarted, AssistantTextDelta, ToolCallStarted, ToolResult, PermissionRequested, ContextCompacted, TurnFinished, ]

然后是 runner 的 phase machine。单飞（single-flight）是关键：一个 turn 可能包含多个 leg（iter → deferred pause → resume → follow-up），phase 在第一个 await 之前同步设置，保证状态查询不会看到中间态：

# src/harness/runner.py import enum from dataclasses import dataclass, field class Phase(enum.Enum): IDLE = "idle" DISPATCHING = "dispatching" # 第一个 await 之前的同步窗口 RUNNING = "running" class Boundary(enum.Enum): MODEL_REQUEST = "model_request" # 下一次模型调用前 drain steering WOULD_STOP = "would_stop" # drain follow-up，空则回 idle @dataclass class Runner: phase: Phase = Phase.IDLE _abort_flag: bool = False def dispatch(self): # 同步设置，避免竞态 self.phase = Phase.DISPATCHING self._abort_flag = False def mark_running(self): self.phase = Phase.RUNNING def request_abort(self): # 协作式 abort：设置 flag，turn 在下一个 boundary 停止 self._abort_flag = True def should_abort(self) -> bool: return self._abort_flag

双队列交互模型是防止 mid-turn 破坏的核心。用户按 Enter 的消息进 steering 队列，在下一个 model-request 边界注入；按 Alt+Enter 的消息进 follow-up 队列，只在 WOULD_STOP 边界处理：

# src/harness/queue.py import asyncio from dataclasses import dataclass, field def _drain(q: asyncio.Queue) -> list[str]: out = [] while not q.empty(): out.append(q.get_nowait()) return out @dataclass class InteractionQueues: steering: asyncio.Queue = field(default_factory=asyncio.Queue) follow_up: asyncio.Queue = field(default_factory=asyncio.Queue) def drain_steering(self) -> list[str]: return _drain(self.steering) def drain_follow_up(self) -> list[str]: return _drain(self.follow_up)

现在写主循环，把事件打出来。这是验证 harness 是否工作的核心：

# src/harness/main.py import asyncio import os from pathlib import Path from openai import AsyncOpenAI from src.harness.config import load_settings, load_permissions from src.harness.runner import Runner, Phase, Boundary from src.harness.queue import InteractionQueues from src.harness.events import ( TurnStarted, AssistantTextDelta, ToolCallStarted, ToolResult, TurnFinished, ) async def run_turn(prompt: str, settings, queues, runner): client = AsyncOpenAI( base_url=settings.llm.base_url, api_key=os.environ[settings.llm.api_key_env], ) runner.dispatch() yield TurnStarted(turn_id="t1") messages = [{"role": "user", "content": prompt}] tools = [ {"type": "function", "function": { "name": "read", "description": "读文件", "parameters": {"type": "object", "properties": { "path": {"type": "string"}}, "required": ["path"]}}}, {"type": "function", "function": { "name": "bash", "description": "执行命令", "parameters": {"type": "object", "properties": { "command": {"type": "string"}}, "required": ["command"]}}}, ] for leg in range(8): # MODEL_REQUEST 边界：注入 steering for msg in queues.drain_steering(): messages.append({"role": "user", "content": msg}) runner.mark_running() resp = await client.chat.completions.create( model=settings.llm.model_id, messages=messages, tools=tools, timeout=settings.llm.timeout_s, ) choice = resp.choices[0].message if choice.content: yield AssistantTextDelta(text=choice.content) if not choice.tool_calls: # WOULD_STOP 边界：处理 follow-up follow = queues.drain_follow_up() if follow: for msg in follow: messages.append({"role": "user", "content": msg}) continue yield TurnFinished(turn_id="t1", stop_reason="completed") runner.phase = Phase.IDLE return messages.append(choice) for call in choice.tool_calls: yield ToolCallStarted(tool=call.function.name, args={}) # 这里接真实工具执行，示例用占位 result = f"[{call.function.name} 执行完成]" yield ToolResult(tool=call.function.name, ok=True, preview=result[:80]) messages.append({ "role": "tool", "tool_call_id": call.id, "content": result, }) yield TurnFinished(turn_id="t1", stop_reason="max_legs") async def main(): settings = load_settings(Path(".")) queues = InteractionQueues() runner = Runner() async for ev in run_turn("读一下 README.md 然后告诉我项目是做什么的", settings, queues, runner): print(f"[{type(ev).__name__}] {ev}") if __name__ == "__main__": asyncio.run(main())

跑起来你会看到类似这样的输出：

[TurnStarted] TurnStarted(turn_id='t1') [ToolCallStarted] ToolCallStarted(tool='read', args={}) [ToolResult] ToolResult(tool='read', ok=True, preview='[read 执行完成]') [AssistantTextDelta] AssistantTextDelta(text='这个项目是一个...') [TurnFinished] TurnFinished(turn_id='t1', stop_reason='completed')

这条事件序列就是 harness 的"心电图"。你能清楚看到：turn 开始、工具被调用、结果返回、模型生成文本、turn 结束。如果中间某一步缺失，问题就定位到了具体环节。

验证成功的标志有三个：事件按顺序出现、stop_reason是completed而不是max_legs、工具结果被正确回填到 messages。三个都满足，说明你的 harness 主链路通了。

5. 本篇常见错排查：401、local proxy failed、reading choices、OAuth

配置和验证过程中，报错几乎都集中在这几类。我按真实报错信息给你对照排查。

401 Unauthorized / invalid api key

最常见。原因通常是 Key 没读到、Key 写错、或者 base_url 和 Key 不匹配。先确认环境变量真的加载了：

python -c "import os; print(os.environ.get('TAOTOKEN_API_KEY', 'NOT SET')[:8])"

如果输出NOT SET，说明.env没被加载。Python 不会自动读.env，你需要python-dotenv或手动 export。如果输出了前 8 位但请求还是 401，检查 base_url 是否写成了带路径的形式，比如https://taotoken.net/api/v1，有些 SDK 会自己拼/v1，重复拼接就会 404 或 401。正确写法是只写到https://taotoken.net/api。

local proxy failed / connection refused

这个报错说明请求根本没发出去，卡在本地网络层。检查三件事：base_url 是否拼错、本机是否有残留的代理环境变量、DNS 是否能解析。用 curl 直接测：

curl -sS -o /dev/null -w "%{http_code}\n" https://taotoken.net/api

返回 200 或 401 都说明网络通，返回 000 说明连接失败。如果本机有HTTP_PROXY之类的环境变量，先 unset 再试。注意不要在代码里硬编码任何代理地址，harness 应该直连配置的 base_url。

reading 'choices' of undefined / KeyError: 'choices'

这个报错几乎都是响应结构不符合预期。可能原因：模型 ID 写错导致返回了错误对象、SDK 版本和 API 协议不匹配、或者请求体格式不对。先打印完整响应：

resp = await client.chat.completions.create(...) print(resp.model_dump_json(indent=2))

如果返回体里是{"error": {...}}而不是{"choices": [...]}，那就是请求本身被拒了，去看 error 字段的具体信息。常见的是 model not found，说明 Model ID 和账号可用模型不匹配，去控制台确认一下。

OAuth / authentication failed / token expired

如果你用的是 Claude Code 这类带 OAuth 流程的工具，报错可能来自它的登录态而不是你的 API Key。这类工具通常有两套认证：一套是工具自身的账号登录，一套是模型 API 的 Key。两者不能混。检查工具的配置文件里，模型入口是否指向了正确的 base_url 和 Key。Claude Code 的配置在~/.claude/settings.json，Cline 的在 VS Code 设置里，Codex 的在~/.codex/auth.json。

以 Codex 的auth.json为例，三件套要写全：

{ "base_url": "https://taotoken.net/api", "api_key": "sk-你的key", "model": "你的模型ID" }

少任何一个字段，工具都会回退到默认认证流程，然后报 OAuth 相关错误。Cline 的 MCP 配置同理，baseUrl、apiKey、model三个字段缺一不可。CC Switch 切换 profile 时，如果某个 profile 只填了两个字段，切过去就会认证失败。

上下文超限 / context length exceeded

这个不是认证问题，是 harness 的 compaction 没生效。检查context_window_tokens是否和实际模型窗口一致，keep_recent_tokens是否设得太大。如果keep_recent_tokens接近context_window_tokens，compaction 永远触发不了，因为保留区就占满了。经验值是保留区占窗口的 20% 到 30%，触发阈值设在 80% 左右，给模型响应和后续 tool output 留 headroom。

工具执行卡死 / 等待确认无响应

这是队列设计问题。如果你只有一个队列，等待用户确认时会阻塞整个循环。正确做法是 permission gate 返回 ASK 时，tool 抛出 ApprovalRequired，循环暂停并返回 deferred 状态，通过独立的 decision channel 等待用户输入。用户输入 y/n/a 后 resolve future，循环恢复。这样等待期间不占用计算资源，也不会死锁。

排查完这几类，你的 harness 基本就稳了。每次遇到新报错，先看它属于哪一层：认证层、网络层、协议层、还是 harness 逻辑层。分层之后，排查范围立刻缩小。

6. 把 harness 用起来：从验证到长期编码

主链路通了之后，你可以按需扩展。harness 的每个组件都是可插拔的，不用一次全上。

先加 memory 注入。在项目根目录放AGENTS.md，harness 启动时读取并注入到 system prompt。这样 agent 每次都知道项目约定，不用你重复交代：

# src/harness/memory.py from pathlib import Path def assemble_memory(cwd: Path) -> str: blocks = [] for path in discover_memory_files(cwd): content = path.read_text(encoding="utf-8", errors="ignore") if path.name == "MEMORY.md": content = "\n".join(content.splitlines()[:200]) blocks.append(f"# From {path}\n{content}") return "\n\n".join(blocks) def discover_memory_files(cwd: Path): # 从 cwd 向上遍历到文件系统根，收集 AGENTS.md 和 MEMORY.md current = cwd.resolve() found = [] while True: for name in ("AGENTS.md", "MEMORY.md"): candidate = current / name if candidate.exists(): found.append(candidate) if current.parent == current: break current = current.parent return list(reversed(found))

再加 context compaction。两级级联：microcompaction 不调模型，只把旧的 tool 输出体替换成占位符；完整 compaction 调一次便宜的模型，把老历史总结成固定骨架。触发阈值基于 token 预算：

# src/harness/compaction.py import enum class CompactOutcome(enum.Enum): COMPACTED = "compacted" NOTHING_TO_COMPACT = "nothing_to_compact" SUMMARIZER_FAILED = "summarizer_failed" def split_tail(messages, *, keep_recent_tokens: int) -> int: """从尾部累积 token，snap 到 compaction boundary， 保证 tool-call/result 对不被拆开。""" total = 0 for i in range(len(messages) - 1, -1, -1): total += estimate_tokens(messages[i]) if total >= keep_recent_tokens: return snap_to_boundary(messages, i) return 0 def microcompact(messages, *, keep_recent_tokens: int): """无 LLM 层：把旧 tool 输出体清空。""" boundary = split_tail(messages, keep_recent_tokens=keep_recent_tokens) for msg in messages[:boundary]: if msg.get("role") == "tool": msg["content"] = "[已压缩]" return messages

沙箱层按需开启。本地开发用mode = "none"，跑真实命令时切docker。fresh-exec 模式下，每条命令作为全新进程运行，cd和export不会跨调用持久化。这个设计看起来反直觉，但它让本地和远程行为字节级一致，避免"本地能跑远程不能跑"的问题：

# src/harness/sandbox.py class SandboxExecutor: """Fresh-exec：cd/export 不持久。""" def __init__(self, backend, workspace): self._backend = backend self._workspace = workspace self._created = False async def run(self, command: str, *, timeout_s: float): if not self._created: await self._backend.create(self._workspace) self._created = True return await self._backend.exec("bash", "-lc", command, timeout_s=timeout_s)

如果你要跑长期任务或 Agent 工作流，建议用 Coding Plan，入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。高频 tool call 场景下，稳定的计费和额度比单次便宜更重要，因为 harness 一个 turn 可能触发十几次模型请求。

接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，里面有各语言和各工具的完整配置示例。API Keys 管理在 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite ，建议给不同项目建不同的 Key，方便按项目排查和吊销。

最后说一个我踩过的坑：不要一上来就把所有组件都打开。先跑通模型入口，再加事件流，再加权限门，最后加沙箱和 compaction。每加一层就跑一次端到端验证，确认事件序列没变。这样出问题时你永远知道是哪一层引入的。harness 的复杂度是必要的，但引入复杂度必须可控。