先交代一个背景。我们团队最早做某自动化运维 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 跑进真实业务，它们就是决定成败的关键。
先交代一个背景。我们团队最早做某自动化运维 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 跑进真实业务，它们就是决定成败的关键。