先交代一个背景。我们团队最早做某自动化运维 Agent 时，第一版代码就是一个while True循环里反复调用大模型、执行工具、再喂回模型。前三天跑得挺顺，第四天它突然开始疯狂调用同一个内部接口，把下游服务打到限流。查了半天才发现，模型在某一步判断失误后，整个循环没有任何护栏，错误被一遍遍自我强化，直到资源耗尽。那一刻我意识到：Agent 能不能干活靠模型，但能不能稳定干活，靠的是外层那套工程设施。这个设施，就是 Harness——从 Agent 循环到上下文管理、沙箱隔离、结果验证，再到长时运行的所有支撑逻辑。这篇文章就来拆解我们踩过坑之后重新搭建的 Harness 工程，包含完整思路和可直接抄的代码骨架。
1. Harness 到底在解决什么问题：从裸循环到工程化
1.1 一个让所有人崩溃的裸循环
先看我们最早那版代码，几乎是所有 Agent Demo 的原型：
while True: response = llm.chat(messages) action = parse_action(response) if action.type == "finish": break result = execute(action) messages.append({"role": "user", "content": f"观察：{result}"})这段代码的问题不是逻辑错，而是完全没有防御。没有任何步数上限、没有上下文窗口管理、没有并发控制、没有执行沙箱，更谈不上验证和恢复。它假设大模型每次输出都是可解析的、可执行的、可信任的，但现实中模型会输出格式错误、会编造不存在的工具、会在一个错误决策上反复循环，甚至可能被注入恶意提示词后执行危险操作。
我们那次事故的直接原因，是模型在一次输出中将某个删除接口当成了幂等重置接口，连续调用了 47 次。循环本身没有任何"这个操作是否合理"的判断机制，所以错误被无限放大。事后复盘时我们才真正理解一句话：Agent 的价值在于模型，但 Agent 的可靠性完全取决于 Harness。
1.2 Harness 的定义与责任边界
Harness 这个词直译是"背带"或"线束"，在工程语境里，它指的是包裹在模型周围的那层管理设施。它不是某个框架的专有名词，而是一整套设计理念。
我们内部把 Harness 拆成五个核心模块：调度循环（Agent Loop）、上下文管理器（Context Manager）、执行沙箱（Sandbox）、验证器（Verifier）和持久化运行时（Runtime）。这五个模块正好对应标题里的五个关键词。
职责边界必须划清楚：Harness 不负责"思考"，所有思考交给模型；Harness 只负责"让思考能安全、可控、可复现地转化为行动"。这条边界很关键，一旦 Harness 开始越权干预模型的输出逻辑，系统就会变得难以调试。
1.3 为什么 Demo 能跑、生产会崩
很多团队都有这个困惑：同样的循环，本地跑几个 Case 没问题，一上生产就各种崩溃。原因很简单，Demo 环境里任务的复杂度低、模型偶尔犯错可以被手动纠正、上下文不会超长、工具调用频率也不高。但生产环境是另一回事：
| 场景 | Demo 环境 | 生产环境 |
|---|---|---|
| 任务复杂度 | 单步决策 | 多步决策、依赖链长 |
| 模型容错 | 允许失败重试 | 失败可能导致资损 |
| 上下文长度 | 可控 | 可能超长、信息冲突 |
| 工具调用 | 少量、安全 | 高频、权限敏感 |
| 运行时长 | 秒级 | 分钟到小时级甚至更长 |
| 并发 | 单实例 | 多实例并发执行 |
每一项差异，都需要 Harness 提供对应的工程能力。所以不要问"Harness 是不是过度设计"，要问"我的 Agent 是不是已经跑到了需要 Harness 的阶段"。以我们经验，只要 Agent 要处理真实业务数据、要自动执行有副作用的操作、要连续运行超过 10 分钟，就必须开始考虑这层设计。
2. Agent 主循环的演进：从同步塞消息到事件驱动
2.1 第一版循环的问题
回到刚才那段while True代码。它最大的问题不只是没有防护，而是把所有环节串成了一个同步阻塞链：模型响应慢，整个系统等着；工具执行慢，整个系统等着。更糟糕的是，如果某一步抛出异常，整个循环就死了，之前所有中间状态全部丢失。
在生产环境里，我们需要的是可中断、可恢复、可观测的循环。我们把循环从"同步调用"改成了"事件驱动"模型。核心思想是：整个 Agent 的运行过程被抽象成一系列事件（事件流），每个事件由循环处理器统一调度。
class AgentRuntime: def __init__(self): self.event_queue = deque() self.state = AgentState() self.hooks = [] def submit(self, event: Event): self.event_queue.append(event) def run(self): while self.event_queue: event = self.event_queue.popleft() self.dispatch(event) def dispatch(self, event: Event): for hook in self.hooks: event = hook(event) if event is None: return if isinstance(event, ModelRequested): self.handle_model_request(event) elif isinstance(event, ToolInvoked): self.handle_tool_invoke(event) elif isinstance(event, TaskCompleted): self.handle_task_complete(event) else: raise UnknownEventError(event)这段代码的精髓在于：任何环节（模型调用、工具执行、验证）完成之后，不是直接拿到结果继续走，而是向事件队列提交一个新事件。这样每个环节都解耦了，任意一步都可以被单独暂停、重试、记录，甚至替换成人工处理。
2.2 状态机设计
事件驱动的循环需要一个明确的状态机来约束 Agent 的行为。我们用的是五态模型：
Pending（等待调度）→Planning（模型分析中）→ToolExecuting（工具执行中）→Verifying（结果验证中）→Completed / Failed（完成 / 失败）
class AgentState(Enum): PENDING = "pending" PLANNING = "planning" TOOL_EXECUTING = "tool_executing" VERIFYING = "verifying" COMPLETED = "completed" FAILED = "failed"状态机不是形式主义，它有几个实打实的作用。第一，它是所有防御逻辑的挂载点——比如在TOOL_EXECUTING状态设置超时，超过 30 秒强制中断；在PLANNING状态设置最大重试次数，防止模型反复输出不可解析的结果。第二，它是观测的基础，每个状态的停留时间、迁移次数都是重要的监控指标。第三，它是持久化的关键，我们后面讲长时运行时会看到，状态可以序列化存储，进程重启后从最后一个状态恢复。
2.3 循环终止条件的硬约束
无约束的循环是生产事故的温床。我们给循环加了三个硬性终止条件：
最大步数限制：每一步（模型调用 + 工具执行）计数，默认上限 25 步，可配置。这样即使模型陷入死循环，系统也能强制退出。无进展检测：如果连续 5 步的观察结果与之前完全相同，判定为循环内耗，强制停止。这一步非常有效，专门对付模型"原地打转"的症状。人类介入触发：当验证器返回低置信度结果且任务不可自动回退时，状态切换到PENDING_HUMAN_REVIEW，挂起等待人工确认。
关于终止条件的优先级，我们踩过一个坑：把"最大步数"放在最后判断，结果前面两个条件都没触发，模型一口气跑了 60 多步。后来调整了判断顺序，每一步执行的末尾先查步数，再查重复度，最后查用户中断标记，才把问题堵住。
3. 上下文的搭建、管理与保鲜：Agent 的记忆中枢
3.1 上下文不是把消息全塞进去
很多人以为上下文就是把所有对话历史和工具结果都堆给模型，窗口不够就截断。这是我们早期犯过的最贵的错误。上下文（Context）是 Agent 的"工作记忆"，它决定了模型在当前决策点能看到什么，"能看到什么"直接决定了"会怎么决策"。
我们后来把上下文系统拆成三层结构：
核心层（Core Context）：系统提示词、用户原始指令、任务目标、当前状态。这一层几乎不被压缩，是模型运行的"宪法"。工作层（Working Context）：当前任务的中间结果、最近的工具观察、验证反馈。这一层是动态变化最频繁的，需要精细管理。记忆层（Memory Context）：历史任务摘要、长期偏好、领域知识。这一层通过检索按需注入，而不是全量加载。
def build_prompt(state: AgentState, memory_store: MemoryStore) -> list[dict]: messages = [] # 核心层 messages.append({"role": "system", "content": state.system_prompt}) messages.append({"role": "user", "content": state.user_instruction}) # 记忆层：按相关度检索 if memory_store: relevant_memories = memory_store.search(state.current_goal, top_k=5) messages.append({"role": "system", "content": f"相关经验：{relevant_memories}"}) # 工作层：最近的对话与观察 messages.extend(state.working_messages[-20:]) return messages3.2 Token 预算分配：谁该占多少
上下文空间是有限的，分配策略直接决定模型表现。我们内部有一个固定的预算分配经验值，供参考：
| 内容类型 | 预算占比 | 说明 |
|---|---|---|
| 系统提示与工具声明 | 20% | 必须字字珍贵，避免冗长的 JSON Schema |
| 用户任务描述 | 10% | 精准提炼目标，不要原文粘贴长文档 |
| 工作层（最近消息/观察） | 50% | 这是模型推理的主要依据 |
| 记忆层（检索结果） | 10% | 按相关性注入，最多 5 条 |
| 预留缓冲 | 10% | 防止模型输出超长导致窗口溢出 |
这里想强调一个经验：工具描述要尽量精简。我们最初给每个工具写了一大段 Markdown 说明和用法示例，结果上下文里光是工具说明就占了 40%，压得推理空间很小。后来改成一句话描述 + 参数表格，效果反而更好——模型并不需要看大段解释，它只需要知道工具叫什么、接受什么参数、返回什么。
3.3 压缩策略：截断是最差的选择
上下文一旦超限，最简单的办法是截断最早的对话。但这样会让 Agent 忘掉任务的起因。我们采用分层次压缩：
轻度压缩：移除中间地带的冗余观察，保留每轮工具调用的"结论摘要"，丢弃原始返回中的大块文本。中度压缩：把早于 N 轮的对话整体生成一个摘要节点，替换进上下文。深度压缩：把当前任务拆解出的关键决策点和约束提炼成结构化记录（比如 JSON），存入长期记忆，然后用一句话概括旧上下文。
def summarize_messages(messages: list[dict], llm) -> str: # 把早期对话先分块摘要，再合并摘要 summaries = [] for block in chunk(messages, size=10): summary = llm.summarize(block) summaries.append(summary) return llm.summarize(summaries, target=200)注意：摘要本身是有损的，不能无限制地层层摘要。我们内部规定最多做两级摘要，再多就触发"重新规划"机制——让模型基于长期记忆从当前状态重新推导下一步，而不是继续在烂上下文中挣扎。
3.4 上下文污染的实战坑
上下文管理里最容易忽略的是污染。典型场景：某次工具调用返回了错误信息，模型把错误的中间结果当成了下一步推理的依据，并且这个错误信息被后续多轮引用，直到最后给出一个完全偏离目标的答案。
我们的对策是引入来源标注（Provenance）。每一条工作层消息都附带它的来源类型——tool、model、system、verifier——模型在推理时可以看到这条消息是否已经被验证器标记为 "suspicious"。一旦验证失败，相关上下文会被打上标签，并在下一次模型调用前重新构造工作层消息，移除被污染的部分。这个机制救了我们好几次，值得优先做。
4. 沙箱：工具执行的可控边界
4.1 一个差点酿成事故的教训
机制讲到这里，我们经历了第一次真正意义上的"事故"。在金融数据项目中，Agent 被分配了读取数据库的权限。某次任务中模型误触发了DROP TABLE操作。虽然数据库配置了自动备份，但整个团队还是惊出一身冷汗。如果我们当时没有执行沙箱，只是裸调数据库驱动，结果就是生产数据直接消失。
这个事故让我们彻底明白：Agent 的工具调用必须经过沙箱层，这是非讨论项，不是可选项。
4.2 沙箱分级设计
不是所有工具都需要同等强度的隔离，一概而论会导致效率和安全的双重损失。我们按风险分成三级：
零信任沙箱（高风险：文件删除、代码执行、数据写入）：运行在隔离容器中，网络默认断开，文件系统只读，只对外开放少数白名单接口。受限沙箱（中风险：只读数据库查询、HTTP 请求）：进程级隔离，设置 CPU/内存配额，网络允许访问白名单域名，禁止写操作。无沙箱直连（低风险：纯计算、字符串处理、格式化）：在宿主进程内执行，仅做参数校验和超时熔断。
4.3 沙箱的工程实现
我们最常用的沙箱实现是容器技术。给 Agent 的代码执行搭了一个基础的 Docker 沙箱：
import docker import tempfile client = docker.from_env() def run_in_sandbox(code: str, memory_limit_mb: int = 256) -> SandboxResult: with tempfile.TemporaryDirectory() as tmp_dir: # 写入待执行代码 code_path = Path(tmp_dir) / "task.py" code_path.write_text(code) # 挂载一个只读系统目录 + 可写的临时工作目录 container = client.containers.run( "python:3.11-slim", f"python {code_path.name}", working_dir="/work", volumes={ tmp_dir: {"bind": "/work", "mode": "rw"}, "/usr/share/readonly_data": {"bind": "/data", "mode": "ro"}, }, mem_limit=f"{memory_limit_mb}m", network_disabled=True, # 禁止网络访问 detach=True, ) try: result = container.wait(timeout=30) logs = container.logs().decode() return SandboxResult(exit_code=result["StatusCode"], output=logs) finally: container.remove(force=True)这套方案有几个关键点：网络必须默认禁用，需要联网的工具单独开白名单端口；外部数据目录以只读方式挂载，从物理上杜绝 Agent 修改基础数据；设置资源配额，防止执行死循环时把宿主机拖垮；超时强制强杀容器，而不是等它自然结束。
4.4 沙箱结果的回收
沙箱执行完，还要解决一个问题：结果怎么安全地拿回 Agent 循环。我们的做法是只允许沙箱向宿主机暴露两种信息：标准输出（stdout） 中的文本结果，和写入指定输出目录的产物文件（必须是格式化数据）。其他的一切都视为未授权。更关键的是，沙箱的输出在被送进上下文之前，需要经过一层数据脱敏与清洗——去掉路径信息、环境变量、敏感字符串，防止模型因为看到宿主机的内部细节而产生错误推理。
5. 验证体系：没有评估的 Agent 就是闭着眼睛开车
5.1 从一次发布事故到验证器的诞生
我们曾经让 Agent 自动修改一份配置文件并发布到测试环境。它在修改配置后，又额外提交了一个包含生产环境密钥的调试文件。当时循环检测到"任务已完成"就直接结束了，没有任何人发现这个多出来的文件。直到安全巡检才暴露问题。从那时起，"验证必须内嵌在循环里"就成了铁律。
5.2 三层验证架构
我们的验证体系分三层，每一层解决不同的问题：
工具层验证（Tool-level Validation）：在工具执行前校验参数合法性，执行后校验返回结果格式。比如 SQL 查询执行前，用语法解析器检查是否有DROP、DELETE等高危操作在非授权表上执行。过程层验证（Process-level Verification）：在每一个关键步骤之后，验证该步骤是否与预期目标一致。例如，模型要读取 A 表的数据，验证器会检查查询结果是否包含预期的 schema，如果返回值与目标无关，就标记这个步骤为可疑。端到端验收（End-to-end Acceptance）：任务执行完成后，验证器根据预定义的验收规则（或独立的评审模型）对最终产物进行整体判断。只有通过了这一层，循环才会向用户返回"完成"。
def verify_step(state: AgentState, action: Action, result: ToolResult) -> Verdict: checks = [] # 语法/结构校验 checks.append(validate_against_schema(action, result)) # 断言校验 for assertion in state.current_plan.assertions: checks.append(assertion.matches(action, result)) # 目标一致性校验 checks.append(check_alignment(result, state.current_goal)) if all(check.passed for check in checks): return Verdict.PASS return Verdict.FAIL5.3 验证失败后的自我修正
验证器返回失败，不意味着直接终止任务。我们把失败分成三类，分别处理：
可重试错误（比如格式不对、瞬时的限流）：自动重试，最多三次，每次重试前更新上下文告诉模型"刚才的方式为什么不行"。可修正错误（比如参数错误、步骤顺序错误）：让模型基于验证反馈重新规划下一步，加入"已失败一次"的提示。不可恢复错误（比如安全保障、越权操作）：强制终止，转入人工处理。
这个机制的奇妙之处在于，它让 Agent 具备了初步的反思能力。很多失败的步骤，换个参数或换个顺序就能成功。如果没有验证反馈就直接重来，纯靠运气，成功率会低很多。
5.4 离线评测集：验证体系本身也必须被验证
循环内的验证器管的是"当下这个任务对不对"，但验证器自身的规则是否合理、覆盖是否全面，需要离线评测集来保证。我们维护了一个包含几十个典型任务的评测集，每个任务都有标准答案和预期步骤序列。每次更新验证器或升级模型，都要先跑一遍评测集，确认没有行为回退。这样我们才敢把验证器放到生产环境里去拦 Agent 的操作。
这里补充一个低成本的起步方式：先把最终的规则抽出来做成一个纯函数，输入是"任务目标 + Agent 步骤轨迹 + 输出"，输出是 Pass/Fail，后面再慢慢加验证维度。不要一上来就上大模型评审，成本和延迟都不可控。
6. 长时运行的工程难题：断点、恢复、心跳与循环失控
6.1 长时运行与"一走一下午"的任务
有的 Agent 任务不是秒级，而是分钟级、小时级，比如批量处理一批文件、爬一个大型网站的公开页面、或者跑一个复杂的数据清洗流程。这种任务的特殊之处在于：单次进程的生命周期不够用，进程可能重启、容器可能迁移、网络可能中断。如果 Harness 不处理长时运行，那 Agent 只适合跑短任务。
6.2 任务持久化与事件溯源
我们的做法是把整个运行状态做成可持久化的。核心思路是事件溯源：每次事件（模型请求、工具调用、验证结果、状态迁移）都记录到存储中，而 Agent 的当前状态则是这些事件的重放结果。
class PersistableState: def __init__(self, task_id: str): self.task_id = task_id self.state = AgentState.PENDING self.step_count = 0 self.working_messages = [] self.plan = [] def append_event(self, event: dict): # 先本地更新 self.apply(event) # 再写入持久化存储 storage.append_event(self.task_id, event) @classmethod def from_task_id(cls, task_id: str): events = storage.read_events(task_id) state = cls(task_id) for event in events: state.apply(event) return state这里每个事件都是不可变的数据记录，比如 "model_response: {...}"、 "tool_result: {...}"、 "verdict: fail"——这样即使进程崩溃，重启后拉取事件列表，就能把 Agent 恢复到崩溃前的那个决策点，而不是从零开始。
6.3 心跳、看门狗与循环失控防护
长时运行最怕的不是慢，而是失控。我们引入了心跳上报机制：Agent 每执行一步就上报一次心跳，心跳里包含当前状态、步数、最近一次活动时间。外界由一个看门狗（Watcher）进程监控所有运行中的任务：
僵尸任务检测：心跳超过某阈值（比如 5 分钟）未更新，判定为挂死，强制重置该任务的沙箱和网络连接，然后从最后一个事件恢复。步数爆炸检测：步数超过预期值（比如预计 10 步，实际 200 步），触发警告并暂停。这通常是上下文污染导致模型在死胡同里乱绕。资源消耗监控：内存、CPU、磁盘占用的持续监控，防止某个任务把资源耗尽影响同机的其他任务。
6.4 人类介入的时机设计
长时运行还会遇到一个尴尬场景：模型在执行到第 30 步时遇到了一个二选一的决策，两个选项都无法通过验证器判断优劣。这时候最合理的设计是挂起并等待人工确认，而不是让模型自己猜。
我们的实现是设置一个human_in_the_loop事件类型。当验证器置信度低于阈值且该决策影响不可逆时，任务状态切换为AWAITING_HUMAN，同时把当前的关键上下文、候选方案、各自风险整理成一份简报推送给运维人员。人工确认后，结果作为一个事件写回，Agent 从该决策点继续运行。这条链路我们统计过，在早期阶段大约有 15% 的任务需要人工介入，随着验证器规则越来越完善，现在降到了 5% 以下。
关于人类介入，最需要注意的点是：不要让这个机制变成频繁打断的"人工客服"。如果同一个人工决策点出现多次，说明是验证规则缺失，应该回去补验证器，而不是靠人硬扛。
结合我们自己的重构经验，如果再重新做一遍，我一定会从第一天就把"沙箱隔离"和"验证器环内反馈"这两层先落地，再逐步补上下文精修和长时运行。这些工程细节一开始做只是多花两天时间，但一旦 Agent 跑进真实业务，它们就是决定成败的关键。
