Agent Harness工程指南：2025权威排行榜与精选推荐
一、什么是 Agent Harness
简单来讲，Harness 是在 LLM 之上搭建的一层“智能体外壳”。它让模型不再只是回答问题，而是能感知环境、调用工具、自主决策，并输出可回溯的执行轨迹。| LLM API | Agent Harness | |
|---|---|---|
| 输入 | messages | 用户意图 + 工作环境 |
| 输出 | 一次回复 | 多步工具调用 + 最终结果 |
| 状态 | 无 | 有（任务、历史、记忆） |
| 扩展 | 无 | 工具、hooks、子 agent |
具体而言，一个优秀的 Harness 需要做到以下四件事：
1. 感知环境（文件、Shell、网页等）
2. 通过工具改变环境
3. 自主决策下一步
4. 输出可观察、可控、可回溯的执行轨迹
1.1 优秀 harness 的关键能力
- **工具齐全 + 工具描述清晰**：模型不知道有什么工具可用，一切都无从谈起 - **上下文管理**：长对话不爆窗口，这是工程核心 - **失败恢复**：单次工具失败不能导致整个 loop 崩溃 - **可控性**：用户能中断、回滚、二次确认 - **可观测**：每一步的 tokens、耗时、决策都一目了然1.2 主要案例
| 产品 | 形态 | 特色 |
|---|---|---|
| Claude Code | CLI/IDE | 工具丰富、hooks、permission 模式 |
| Cursor | IDE | 上下文检索、编辑流 |
| Cline | VSCode 插件 | 计划模式、自主执行 |
| Aider | CLI | git-based 编辑 |
| Devin / OpenHands | 浏览器 + 沙箱 VM | 长任务、多模态 |
二、Harness、上下文、提示词的边界
这三个概念经常被混用，但层次完全不同。理清边界，才知道某个问题应该在哪一层解决。2.1 一句话区分
| 概念 | 关注的核心问题 |
|---|---|
| 提示词工程（Prompt Engineering） | 怎么写好这一次的指令，让模型一次性输出更准确的结果 |
| 上下文工程（Context Engineering） | 窗口里该放什么，让模型在每一步都拿到刚好够用的信息 |
| Agent Harness 工程 | 怎么造一个能持续运行的智能体外壳，把循环、工具、安全、可观测性都做出来 |
2.2 三层关系图
``` ┌──────────────────────────────────────────────────────┐ │ Agent Harness 工程 │ │ 循环、工具、权限、Hook、子 agent、UI、可观测、评测 │ │ ┌────────────────────────────────────────────────┐ │ │ │ 上下文工程 │ │ │ │ 窗口预算、压缩、检索、记忆、文件状态、工具结果 │ │ │ │ ┌──────────────────────────────────────────┐ │ │ │ │ │ 提示词工程 │ │ │ │ │ │ 角色、示例、结构、CoT、约束输出、缓存 │ │ │ │ │ └──────────────────────────────────────────┘ │ │ │ └────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────┘ ```2.3 三层各自做什么
**提示词工程：管“一次调用”** 操心的对象是一次 LLM 请求里的 messages： - system prompt 的角色、规则、输出格式 - few-shot 示例 - CoT、结构化输出、function calling - 标点、分隔符、tag 包裹（`2.4 一个具体例子串起来
任务：「帮我修 utils.py 的 bug 并跑测试」| 层 | 在做什么 |
|---|---|
| 提示词工程 | 写 system：“你是软件工程助手，破坏性命令前先确认……” + Edit 工具的 description “替换字符串前必须先 Read” |
| 上下文工程 | 决定：Read 返回 cat -n 行号；测试输出 5000 行 → 截首尾各 15k；第 8 步上下文到 80% → 触发压缩，保留首条 + 最近 4 条 + 摘要 |
| Harness 工程 | 主循环跑 ReAct；Edit 触发 PreToolUse hook 校验路径；测试失败 → 工具返回 is_error=true 但循环不崩；用户按 ESC → 中断信号；写 NDJSON 日志供回放 |
2.5 边界其实在哪
实践中三者是一个连续谱，没有硬边界。同一个决策可以从不同的角度去解释和优化：| 决策 | 归类 |
|---|---|
| “把工具结果裁到 30k” | 上下文工程 |
| “工具描述里写明：输出会被裁剪” | 提示词工程 |
| “工具实现里做的裁剪逻辑” | Harness 工程 |
| “Read 之后才能 Edit” 的规则写在 description | 提示词工程 |
| “Read 之后才能 Edit” 的状态校验在工具里 | Harness 工程 |
| “Read 之后才能 Edit” 的失效检测（文件 hash 变了） | 上下文工程 + Harness 工程 |
2.6 三层的常见误区
| 层 | 误区 |
|---|---|
| 提示词工程 | 以为写得越细越好 → prompt 50KB、模型反而抓不住重点 |
| 上下文工程 | 把所有相关文件全塞进去 → token 爆 + 噪声拉低准确率 |
| Harness 工程 | 一上来就堆 30 个工具、5 个子 agent → 调试不动、成本失控 |
2.7 何时往上走一层
遇到问题时，先尝试在最低层解决，解决不了再上层： - prompt 改两版还不稳 → 是不是上下文里缺关键信息？（上下文工程） - 上下文已经塞满相关信息还不行 → 是不是工具能力不够 / 没有重试机制？（harness 工程） - harness 已经很完善但效果还差 → 回头看 prompt 是不是有冲突指令？（提示词工程）2.8 团队分工建议
- **应用开发者**：80% 时间在提示词工程 + 上下文工程，少量 harness 选型 - **平台/基建团队**：核心做 harness，给上层提供“安全的 agent runtime” - **研究者**：哪一层都可能。但模型能力进步会同时减少提示词工程的工作量、增加上下文/harness 的工作量——模型越聪明，越值得给它复杂的环境三、整体架构
``` ┌─────────────────────────────────────────────────────┐ │ User Interface │ │ (CLI / IDE plugin / Web / Desktop) │ └────────────────────────┬────────────────────────────┘ │ JSON-RPC / IPC / WebSocket ┌────────────────────────▼────────────────────────────┐ │ Harness Core │ │ ┌────────────────────────────────────────────────┐ │ │ │ Agent Loop │ │ │ │ ┌──────┐ ┌────────┐ ┌──────────┐ │ │ │ │ │ Plan │ │Execute │ │ Reflect │ │ │ │ │ └──────┘ └────────┘ └──────────┘ │ │ │ └────────────────────────────────────────────────┘ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐│ │ │ Prompt │ │ Tools │ │ Context │ │ Memory ││ │ │ Builder │ │ Registry │ │ Manager │ │ Store ││ │ └──────────┘ └──────────┘ └──────────┘ └────────┘│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐│ │ │Permission│ │ Hooks │ │Sub-Agent │ │Telemetry││ │ │ Engine │ │ System │ │ Manager │ │ ││ │ └──────────┘ └──────────┘ └──────────┘ └────────┘│ └────────────────────────┬────────────────────────────┘ │ ┌────────────────────────▼──────────────────────────┐ │ LLM Provider │ │ (Anthropic/OpenAI) │ └────────────────────────────────────────────────────┘ │ ┌────────────────────────▼──────────────────────────┐ │ Sandbox / FS / Net │ └────────────────────────────────────────────────────┘ ```四、Agent 主循环
3.1 经典 ReAct 循环
``` while not done: response = llm(messages, tools) if response.has_text(): emit_to_user(response.text) if response.has_tool_calls(): for call in response.tool_calls: result = execute_tool(call) messages.append(tool_result(call, result)) else: done = True ```3.2 Python 最小实现
```python import anthropic from dataclasses import dataclass, field client = anthropic.Anthropic() @dataclass class AgentState: messages: list[dict] = field(default_factory=list) tools: list[dict] = field(default_factory=list) tool_handlers: dict = field(default_factory=dict) max_steps: int = 50 step: int = 0 def run_agent(state: AgentState, user_input: str, system: str): state.messages.append({"role": "user", "content": user_input}) while state.step < state.max_steps: state.step += 1 resp = client.messages.create( model="claude-sonnet-4-6", system=system, tools=state.tools, messages=state.messages, max_tokens=4096, ) # 收集 assistant content state.messages.append({"role": "assistant", "content": resp.content}) if resp.stop_reason == "end_turn": return resp if resp.stop_reason == "tool_use": tool_results = [] for block in resp.content: if block.type == "tool_use": handler = state.tool_handlers[block.name] try: out = handler(**block.input) tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": str(out), }) except Exception as e: tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": f"Error: {e}", "is_error": True, }) state.messages.append({"role": "user", "content": tool_results}) raise RuntimeError("max steps reached") ```3.3 TypeScript 实现
```typescript import Anthropic from "@anthropic-ai/sdk"; interface ToolHandler { (input: any): Promise3.4 中断与恢复
```python import signal class InterruptError(Exception): ... def install_handler(): signal.signal(signal.SIGINT, lambda *_: (_ for _ in ()).throw(InterruptError())) # 主循环捕获中断，保留 messages，等待用户决定继续/重写/取消 ```五、系统提示词设计
4.1 结构
``` [身份] 你是 Claude Code，Anthropic 的官方 CLI 助手... [行为准则] 短而直接、避免油嘴滑舌... [工具使用规范] 优先用 Edit 而非 Write、并行调用独立工具... [执行边界] 破坏性操作前确认、不修改 git config... [输出格式] 默认无 markdown 包裹、文件引用用 file:line... [环境信息] CWD、平台、模型、工作目录是否 git 仓库... ```4.2 编写原则
1. **明确角色与边界**：是助手还是自治 agent？能不能改 git？能不能 push？ 2. **行为示例 > 抽象原则**：示例多 1 个，错误率降一截 3. **环境注入**：CWD、OS、工具列表、当前文件等动态拼入 4. **避免冲突**：不同段落规则相左时，模型会随机选一种4.3 提示词分层
```python def build_system_prompt(env: Env, user_settings: dict) -> str: parts = [ IDENTITY_BLOCK, BEHA VIOR_BLOCK, TOOL_USAGE_BLOCK, SAFETY_BLOCK, f"\n# Environment\n{render_env(env)}", ] if user_settings.get("memory_enabled"): parts.append(MEMORY_INSTRUCTIONS) if user_settings.get("project_md"): parts.append(f"\n# Project Instructions\n{user_settings['project_md']}") return "\n\n".join(parts) ```4.4 Prompt Caching
Anthropic API 支持显式缓存系统提示，TTL 5 分钟： ```python client.messages.create( system=[{ "type": "text", "text": LARGE_SYSTEM, "cache_control": {"type": "ephemeral"}, }], messages=..., ) ``` 收益：长 system prompt 重复使用时延迟降 50%+，成本降 90%。Harness 必须开。六、工具系统
5.1 工具定义
```typescript interface Tool { name: string; description: string; // 让模型决定何时调用 input_schema: JSONSchema; // 强约束输入 // 可选：行为提示 annotations?: { readOnly?: boolean; destructive?: boolean; idempotent?: boolean; }; } ```5.2 设计原则
1. **原子化**：一个工具做一件事，不要造“瑞士军刀” 2. **描述写给模型**：包含用途、限制、何时不该用 3. **Schema 严格**：用 enum/pattern 约束，避免模型瞎传 4. **失败可读**：错误信息直接告诉模型如何修正 5. **副作用透明**：annotations 帮 harness 决定是否需要确认5.3 文件编辑：Edit vs Write
``` Write：覆盖整文件 → 大文件成本高、容易丢内容 Edit ：基于 diff → 必须先 Read，模型出错率低 ``` 实际系统通常 Edit 为主，Write 作 escape hatch。5.4 Edit 工具实现
```python def edit_tool(file_path: str, old_string: str, new_string: str, replace_all: bool = False): if file_path not in read_log: raise ValueError("must Read file before Edit") text = open(file_path).read() count = text.count(old_string) if count == 0: raise ValueError(f"old_string not found in {file_path}") if count > 1 and not replace_all: raise ValueError(f"old_string appears {count} times; pass replace_all or expand context") new_text = text.replace(old_string, new_string) open(file_path, "w").write(new_text) return f"edited {file_path}" ```5.5 Read 工具
```python def read_tool(file_path: str, offset: int = 0, limit: int = 2000) -> str: with open(file_path) as f: lines = f.readlines() chunk = lines[offset : offset + limit] read_log[file_path] = True # cat -n 风格行号便于模型引用 return "".join(f"{offset+i+1:6}\t{l}" for i, l in enumerate(chunk)) ```5.6 Bash 工具与超时
```python import subprocess, threading def bash_tool(command: str, timeout_ms: int = 120_000) -> str: proc = subprocess.Popen( command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True ) timer = threading.Timer(timeout_ms / 1000, proc.kill) timer.start() out, _ = proc.communicate() timer.cancel() if proc.returncode != 0: raise RuntimeError(f"exit {proc.returncode}\n{out}") return out[-30000:] # 限制返回大小 ```5.7 工具结果裁剪
工具输出可能成百上千行 → 进入上下文撑爆 → 必须裁剪： ```python def truncate(s: str, max_chars: int = 30000) -> str: if len(s) <= max_chars: return s head = s[: max_chars // 2] tail = s[-max_chars // 2:] return f"{head}\n... [truncated {len(s) - max_chars} chars] ...\n{tail}" ```七、上下文管理
6.1 Token 预算
```python context_limit = model.max_context_tokens # e.g. 200k output_reserve = max_output_tokens # e.g. 8k input_budget = context_limit - output_reserve - safety_margin ```6.2 何时压缩
- 接近 input budget 90% → 触发压缩 - 工具结果过长 → 单步立即裁剪 - 用户主动 `/compact` / `/clear`6.3 压缩策略
```python async def compact(messages: list, llm) -> list: summary_prompt = ( "对以下对话做压缩摘要，保留：用户目标、关键决策、关键文件路径与代码片段、" "未完成事项。删除无关探索过程。\n\n" + serialize(messages) ) summary = await llm.complete(summary_prompt) # 保留首条 system + 最近 N 条原始消息 + 摘要 return [ messages[0], {"role": "user", "content": f"[Compacted history]\n{summary}"}, *messages[-4:], ] ```6.4 工具结果引用
不直接塞大输出到上下文，而是写入文件，模型按需重读： ```python def big_search_tool(query: str) -> str: matches = search(query) if len(matches) > 50: path = f"/tmp/search-{uuid4()}.txt" sa ve(path, matches) return f"Found {len(matches)} matches. Sa ved to {path}. Use Read to view." return format(matches) ```6.5 文件状态追踪
记录文件读过的版本，避免基于过期内容编辑： ```python class FileStateTracker: def __init__(self): self.last_seen: dict[str, str] = {} # path -> hash def on_read(self, path): self.last_seen[path] = file_hash(path) def check_before_edit(self, path): if path not in self.last_seen: raise ValueError("must Read before Edit") if self.last_seen[path] != file_hash(path): raise ValueError("file changed since last Read; please Read again") ```八、记忆系统
7.1 三层记忆
``` [Session] 本次对话的上下文 ← messages[] [Project] 当前项目的 CLAUDE.md / .rules ← 启动时注入 [Persistent] 跨项目的用户偏好 ← memory/*.md ```7.2 持久记忆设计
``` memory/ ├── MEMORY.md ← 索引，自动注入 system ├── user_role.md ← 用户身份 ├── feedback_testing.md ← 历次反馈 ├── project_team_alpha.md ← 项目背景 └── reference_dashboards.md ← 资源指针 ``` 每个文件 frontmatter： ``` --- name: 用户角色 description: 用户身份与领域 type: user --- 用户是后端工程师，主要写 Go，对数据库性能敏感... ```7.3 记忆写入触发
- 用户显式说“记住” / “以后...” - 用户纠正后 → 写 feedback 类型 - 用户透露背景信息 → 写 user 类型7.4 反索引
`MEMORY.md` 始终在 system 中，过长会浪费 token。约束： - 索引行 < 150 字符 - 总行数 < 200 - 内容真正放在外部 md 文件7.5 时效性校验
记忆可能过期。在使用前： ```python def use_memory(memory: Memory, current_repo_state): if memory.references_file: if not exists(memory.references_file): mark_stale(memory) return None return memory.content ```九、权限与沙箱
8.1 权限模型（Claude Code 思路）
``` permissions: allow: - Bash(npm test:*) ← 允许 npm test、npm test xxx - Bash(git status) - Edit(src/**/*.ts) deny: - Bash(rm -rf:*) - Bash(sudo:*) - Read(.env*) ask: - Bash(git push:*) ← 默认询问 ```8.2 模式
| 模式 | 行为 |
|---|---|
default | 按规则匹配，匹配不到 → 询问 |
acceptEdits | 文件编辑自动通过 |
plan | 只读模式，不能写 |
bypassPermissions | 全部放行（自负风险） |
8.3 实现
```python class PermissionEngine: def __init__(self, rules: list[Rule]): self.rules = rules async def check(self, tool: str, input: dict) -> "Decision": for r in self.rules: if r.matches(tool, input): return r.decision # ALLOW / DENY / ASK return Decision.ASK async def gate(self, tool, input): d = await self.check(tool, input) if d == Decision.DENY: raise PermissionError(f"denied: {tool}") if d == Decision.ASK: ans = await prompt_user(f"Allow {tool}({input})?") if ans not in ("yes", "always"): raise PermissionError("user denied") if ans == "always": self.add_allow(tool, input) ```8.4 沙箱执行
防止工具误删用户文件 / 联网泄露：| 沙箱 | 强度 | 适用 |
|---|---|---|
| chroot / 工作目录限制 | 弱 | 本地开发 |
| Linux namespace + seccomp | 中 | 容器内 |
| Docker / Podman | 中-强 | 服务端 agent |
| Firecracker microVM | 强 | 多租户 SaaS（Devin） |
| gVisor | 强 | 兼容性好 |
8.5 路径白名单
```python ALLOWED_ROOTS = ["/workspace"] def safe_path(p: str) -> str: abs_p = os.path.realpath(p) if not any(abs_p.startswith(r) for r in ALLOWED_ROOTS): raise PermissionError(f"path outside workspace: {abs_p}") return abs_p ```十、Hooks 与扩展点
9.1 钩子事件
| 事件 | 触发时机 | 用途 |
|---|---|---|
PreToolUse | 工具执行前 | 校验、阻止、改输入 |
PostToolUse | 工具执行后 | 日志、二次处理 |
UserPromptSubmit | 用户输入后 | 注入额外上下文 |
Stop | 一轮 agent 结束 | 触发副作用（lint/test） |
SessionStart | 进入 session | 初始化 |
Notification | 系统通知 | 桌面通知、Slack |
9.2 Hook 实现（外部进程）
```json { "hooks": { "PostToolUse": [ { "matcher": "Edit|Write", "hooks": [ { "type": "command", "command": "prettier --write $CLAUDE_FILE_PATH" } ] } ] } } ```9.3 Hook 协议
Hooks 通过 stdin 收 JSON、通过 stdout 返回结果： ```python import json, sys event = json.load(sys.stdin) # event = {"event":"PreToolUse", "tool":"Bash", "input":{"command":"rm ..."}} if event["tool"] == "Bash" and "rm -rf" in event["input"]["command"]: print(json.dumps({"action": "block", "reason": "destructive command"})) sys.exit(0) print(json.dumps({"action": "allow"})) ```十一、子 Agent 与并行
10.1 用途
- **隔离上下文**：探索性查找放进子 agent，结果不污染主上下文 - **并行加速**：独立任务并发跑 - **专家化**：用不同 system prompt 扮演 reviewer / planner / coder10.2 实现
```python async def spawn_subagent(prompt: str, tools: list, system: str) -> str: state = AgentState(tools=tools, tool_handlers=...) result = await run_agent_async(state, prompt, system) return extract_final_text(result) # 主 agent 内的工具 async def task_tool(description: str, prompt: str, agent_type: str = "general"): return await spawn_subagent(prompt, tools_for(agent_type), system_for(agent_type)) ```10.3 并行调用
LLM 一次返回多个 `tool_use` 块时，工具执行可以并发： ```python async def execute_calls(blocks): tasks = [run_one(b) for b in blocks if b.type == "tool_use"] return await asyncio.gather(*tasks, return_exceptions=True) ```10.4 子 agent 限制
- **不要嵌太深**：子 agent 再开子 agent 难以调试 - **隔离工具集**：子 agent 通常不需要“删除 PR”这种危险工具 - **超时**：子 agent 可能卡死，必须设置 max_steps + wall-clock timeout十二、Slash Commands / Skills
11.1 Slash Commands
用户主动选择的提示词模板： ``` /commit → 帮我做 git commit /review → 审查当前 diff ``` 实现： ```python @command("/commit") async def commit_cmd(args: str, ctx: AgentContext): diff = run("git diff --staged") prompt = f"基于以下 diff 写 conventional commit message：\n{diff}\n\n额外说明：{args}" return await ctx.run_inline(prompt) ```11.2 Skills（按需加载的模板包）
每个 skill 是独立目录： ``` skills/ ├── code-review/ │ ├── skill.md ← 触发条件、说明 │ └── templates/ └── deploy/ └── skill.md ``` `skill.md` frontmatter： ``` --- name: code-review description: 触发条件：用户请求代码审查或检查改动 --- 执行步骤： 1. 列出改动文件 2. 逐文件审阅... ``` 启动时把所有 skill 的 `name` + `description` 注入 system，模型决定何时调用 `Skill(name)` 工具加载完整内容。11.3 收益
- 系统提示不爆炸（仅注入名称+描述） - 用户可扩展（不修改主 harness） - 模型自决策 vs 用户主动 `/skill-name`十三、流式 UI 与 IPC
12.1 LLM 流式
```python with client.messages.stream( model="claude-sonnet-4-6", messages=messages, tools=tools, max_tokens=4096, ) as stream: for event in stream: if event.type == "content_block_delta": if event.delta.type == "text_delta": emit("text", event.delta.text) elif event.delta.type == "input_json_delta": emit("tool_input", event.delta.partial_json) ```12.2 事件类型
```typescript type HarnessEvent = | { type: "assistant_text"; delta: string } | { type: "tool_call_start"; id: string; name: string } | { type: "tool_call_input"; id: string; deltaJson: string } | { type: "tool_result"; id: string; output: string; isError: boolean } | { type: "step_finished"; tokensIn: number; tokensOut: number } | { type: "permission_request"; tool: string; input: any } | { type: "error"; message: string }; ```12.3 IPC 协议（CLI ↔ UI）
NDJSON over stdout 是最朴素可靠的方式： ```json {"type":"assistant_text","delta":"我先读一下文件..."} {"type":"tool_call_start","id":"t1","name":"Read"} {"type":"tool_call_input","id":"t1","deltaJson":"{\"file_path\":\""}... ``` UI 端流式渲染： ```typescript import readline from "readline"; const rl = readline.createInterface({ input: agentProc.stdout }); for await (const line of rl) { const e = JSON.parse(line); ui.dispatch(e); } ```12.4 中断协议
UI 端按 ESC： ```json {"type":"interrupt"} ← UI 写入 agent 的 stdin ``` agent 端： ```python def listen_stdin(): for line in sys.stdin: if json.loads(line)["type"] == "interrupt": interrupt_event.set() ``` 主循环检查 `interrupt_event.is_set()` 并优雅退出。十四、Token / 成本控制
13.1 度量
每一步记录： ```python @dataclass class StepMetrics: input_tokens: int cache_creation_input_tokens: int cache_read_input_tokens: int output_tokens: int latency_ms: int model: str ``` ```python def cost(m: StepMetrics, prices) -> float: p = prices[m.model] return ( m.input_tokens * p.input + m.cache_creation_input_tokens * p.cache_write + m.cache_read_input_tokens * p.cache_read + m.output_tokens * p.output ) ```13.2 控制手段
- Prompt cache：开启系统提示缓存 - 小模型路由：简单子任务用 Haiku - 限制 max_tokens：避免冗长输出 - 裁剪工具结果：单工具 30k 字符上限 - 压缩历史：到阈值触发 - 并行 → 串行：成本稳定但更慢13.3 路由示例
```python def pick_model(task_type: str) -> str: return { "simple_qa": "claude-haiku-4-5", "code_search": "claude-haiku-4-5", "code_edit": "claude-sonnet-4-6", "architect": "claude-opus-4-7", }.get(task_type, "claude-sonnet-4-6") ```十五、可观测性与调试
14.1 日志结构
```json { "ts": "2026-05-09T10:23:45Z", "session": "abc", "step": 7, "event": "tool_use", "tool": "Edit", "input": {...}, "output_size": 1234, "latency_ms": 312, "tokens_in": 4523, "tokens_out": 215 } ``` 每个 session 写一个 NDJSON 文件，方便回放。14.2 OpenTelemetry
```python from opentelemetry import trace tracer = trace.get_tracer("harness") with tracer.start_as_current_span("agent.step") as span: span.set_attribute("step", state.step) span.set_attribute("tokens.in", resp.usage.input_tokens) ... ```14.3 回放
记录完整 messages 历史 + 工具输入输出，可“录像”重放： ```python def replay(session_log: Path, until_step: int): events = [json.loads(l) for l in open(session_log)] for e in events[: until_step]: ui.dispatch(e) ```14.4 调试场景
| 现象 | 看哪 |
|---|---|
| Agent 卡死 | 查最后一条工具输出，是否超时/挂起 |
| 输出乱、工具用错 | system prompt + tool description |
| 上下文爆 | 看每步 tokens，是哪一步突然涨 |
| 成本失控 | 按工具/模型分组聚合 metrics |
| 编辑出错 | 查 read_log + Edit 输入 |
十六、评测与回归
15.1 评测维度
- **任务完成率**：能否完成给定任务 - **效率**：步数、token、耗时 - **正确性**：通过测试 / 人工评分 - **安全**：是否触碰危险操作 - **稳健性**：网络抖动、工具失败时的恢复15.2 离线评测
```python @dataclass class Task: id: str prompt: str setup: Callable # 准备 workspace verify: Callable # 跑后判定通过 results = [] for t in tasks: workspace = t.setup() metrics = run_harness(workspace, t.prompt) passed = t.verify(workspace) results.append((t.id, passed, metrics)) ``` 公开基准：SWE-bench、HumanEval、TerminalBench、WebArena。15.3 真实流量回归
抽样真实 session 重放： ```python for s in sampled_sessions: replay = run_harness_with_recorded_inputs(s) diff = compare(s.outputs, replay.outputs) if diff.regression: report(s.id, diff) ```15.4 A/B
prompt / 工具 / 模型变更上线前，按用户分桶 A/B： ``` A 桶 (control): 旧版本 B 桶 (treat): 新版本 比较：完成率、用户满意度、成本 ```十七、生产部署
16.1 部署形态
| 形态 | 例子 | 部署 |
|---|---|---|
| 本地 CLI | Aider / Claude Code | npm/pypi 包 + 用户机器 |
| 本地 GUI | Cursor / VSCode 插件 | Electron / VSIX |
| 云端 SaaS | Devin / OpenHands | K8s + 沙箱 VM |
| 混合 | UI 在云、工具在本地 | LSP 风格双向 IPC |
16.2 沙箱平台
云端 agent 必须每个 session 一个隔离环境： ``` session 创建 → 起 microVM/容器 → 拷贝项目 → 启动 harness → 流式回 UI session 结束 → 提交结果 → 销毁环境 ``` 参考实现： - **E2B**：托管的 Firecracker microVM，秒级启动 - **Modal**：Python serverless，可挂卷 - **Daytona / Coder**：开发环境平台 - **自建**：Firecracker / gVisor + K8s16.3 密钥管理
agent 经常需要用户的 API key（GitHub、AWS）。绝不能直接拿用户原始 token： - OAuth → 短期 token - 或袋里：用户的请求经 harness 后端，附加凭证后转发 - KMS 加密存储；运行时注入16.4 速率与配额
- 单用户 QPM/QPD - 单 session token 上限 - 工具调用次数上限 - 沙箱 CPU/内存/磁盘配额 - 出口流量限速 + 域名白名单16.5 灾备
- 上下文写检查点：每 N 步持久化 - 进程崩溃 → 恢复时从最后一次 checkpoint 加载 - 工具调用幂等：写之前先 read，避免重复执行十八、完整最小实现
下面是一个能跑的 ~200 行 Python harness，含工具、权限、流式输出、压缩。 ```python # harness.py import os, json, asyncio, subprocess, hashlib, anthropic from dataclasses import dataclass, field from pathlib import Path client = anthropic.Anthropic() MODEL = "claude-sonnet-4-6" MAX_TOKENS = 4096 COMPACT_THRESHOLD = 150_000 @dataclass class Harness: workspace: Path messages: list = field(default_factory=list) read_log: dict = field(default_factory=dict) # path -> hash permissions: dict = field(default_factory=lambda: { "allow": [], "deny": ["rm -rf /"] }) # ---------- 工具 ---------- def _safe(self, p: str) -> Path: p = (self.workspace / p).resolve() if not str(p).startswith(str(self.workspace.resolve())): raise PermissionError("path outside workspace") return p def tool_read(self, file_path: str, offset: int = 0, limit: int = 2000) -> str: p = self._safe(file_path) lines = p.read_text().splitlines() chunk = lines[offset : offset + limit] self.read_log[str(p)] = hashlib.md5(p.read_bytes()).hexdigest() return "\n".join( f"{offset+i+1:6}\t{l}" for i, l in enumerate(chunk) ) def tool_edit(self, file_path: str, old_string: str, new_string: str, replace_all: bool = False) -> str: p = self._safe(file_path) if str(p) not in self.read_log: raise ValueError("must Read before Edit") if hashlib.md5(p.read_bytes()).hexdigest() != self.read_log[str(p)]: raise ValueError("file changed since last Read") text = p.read_text() n = text.count(old_string) if n == 0: raise ValueError("old_string not found") if n > 1 and not replace_all: raise ValueError(f"old_string appears {n} times") p.write_text(text.replace(old_string, new_string)) self.read_log[str(p)] = hashlib.md5(p.read_bytes()).hexdigest() return f"edited {file_path}" def tool_bash(self, command: str, timeout_ms: int = 120_000) -> str: for d in self.permissions["deny"]: if d in command: raise PermissionError(f"denied: {d}") r = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=timeout_ms/1000, cwd=self.workspace ) out = (r.stdout + r.stderr)[-30000:] if r.returncode != 0: return f"[exit {r.returncode}]\n{out}" return out TOOLS = [ {"name": "Read", "description": "读取文件，offset/limit 可选", "input_schema": {"type":"object", "properties":{ "file_path":{"type":"string"}, "offset":{"type":"integer"}, "limit":{"type":"integer"} }, "required":["file_path"]}}, {"name": "Edit", "description": "替换文件中的字符串。必须先 Read", "input_schema": {"type":"object","properties":{ "file_path":{"type":"string"}, "old_string":{"type":"string"}, "new_string":{"type":"string"}, "replace_all":{"type":"boolean"} }, "required":["file_path","old_string","new_string"]}}, {"name": "Bash", "description": "执行 shell 命令", "input_schema": {"type":"object","properties":{ "command":{"type":"string"}, "timeout_ms":{"type":"integer"} }, "required":["command"]}}, ] def dispatch(self, name: str, args: dict) -> str: try: return getattr(self, f"tool_{name.lower()}")(**args) except Exception as e: return f"Error: {e}" # ---------- 上下文管理 ---------- def maybe_compact(self): approx = sum(len(json.dumps(m)) for m in self.messages) // 4 if approx < COMPACT_THRESHOLD: return head, tail = self.messages[:1], self.messages[-4:] text = json.dumps(self.messages[1:-4])[:80_000] summary = client.messages.create( model=MODEL, max_tokens=2000, messages=[{"role":"user", "content":f"压缩对话摘要：\n{text}"}], ).content[0].text self.messages = head + [ {"role":"user","content":f"[Compacted]\n{summary}"} ] + tail # ---------- 主循环 ---------- SYSTEM = ( "你是一个软件工程助手，可使用 Read/Edit/Bash 工具完成任务。" "回答简洁、行动直接。破坏性命令前先确认。" ) def run(self, user_input: str, on_event=lambda e: print(json.dumps(e, ensure_ascii=False))): self.messages.append({"role":"user","content":user_input}) for step in range(50): self.maybe_compact() with client.messages.stream( model=MODEL, max_tokens=MAX_TOKENS, system=[{ "type":"text","text":self.SYSTEM, "cache_control":{"type":"ephemeral"} }], tools=self.TOOLS, messages=self.messages, ) as stream: for event in stream: if (event.type == "content_block_delta" and event.delta.type == "text_delta"): on_event({"type":"text", "delta": event.delta.text}) resp = stream.get_final_message() self.messages.append({ "role":"assistant", "content":[b.model_dump() for b in resp.content] }) if resp.stop_reason == "end_turn": on_event({"type":"done"}) return resp results = [] for block in resp.content: if block.type == "tool_use": on_event({"type":"tool_call","name":block.name, "input":block.input}) out = self.dispatch(block.name, block.input) is_err = out.startswith("Error:") on_event({"type":"tool_result","name":block.name, "output":out[:200],"is_error":is_err}) results.append({ "type":"tool_result","tool_use_id":block.id, "content":out, "is_error": is_err }) self.messages.append({"role":"user","content":results}) raise RuntimeError("max steps reached") if __name__ == "__main__": import sys h = Harness(workspace=Path(os.getcwd())) while True: try: q = input("> ") except EOFError: break if not q.strip(): continue h.run(q) ``` 运行： ```bash export ANTHROPIC_API_KEY=... python harness.py > 帮我修复 utils.py 的 bug ``` 可在此基础上扩展： - ✅ Hooks（PreToolUse / PostToolUse） - ✅ 子 agent - ✅ Slash commands / Skills - ✅ NDJSON IPC + Web UI - ✅ Docker 沙箱 - ✅ 持久记忆 - ✅ 路由不同模型 - ✅ OpenTelemetry附录：速查
A.1 关键设计决策
| 维度 | 选项 |
|---|---|
| Loop | ReAct（流行）/ Plan-Act / Reflexion |
| 工具粒度 | 原子（Read/Edit）/ 复合（Search+Edit） |
| 上下文 | 无限增长 → 自动压缩；或滑窗 |
| 记忆 | 文件式（Claude Code）/ 向量库 / 知识图 |
| 沙箱 | 信任本机 / Docker / microVM |
| UI | CLI / TUI / IDE 插件 / Web |