生产环境中的 AI Agent，往往不是因为模型“说错一句话”而失控，而是因为模型生成的意图被系统直接变成了不可逆操作：覆盖文件、发送邮件、修改工单、部署服务，甚至发起付款。Vinoth Govindarajan 的分享将关注点从模型本身移到了 Agent Harness——包围模型、工具与业务状态的控制层。
一个可靠的 Harness 不负责让模型变得绝对正确。它要做的是：即使模型犯错、请求并发到达、审批过期或工具参数被篡改，系统仍然守住明确的不变量。
Agent Harness 是控制平面，不是提示词包装器
可以把生产 Agent 拆成三个部分：
- 模型负责提出意图：分析上下文，选择工具，生成参数。
- Harness 负责执行决策：检查状态、权限、审批与策略。
- 工具负责产生副作用：写文件、调用 API、更新数据库或发送消息。
这里最危险的设计，是让模型输出直接进入工具：
用户请求 -> LLM -> tool.execute(arguments)
更稳妥的路径应当加入一道确定性的控制平面：
用户请求
-> LLM 生成候选动作
-> Harness 规范化参数
-> 检查状态版本与执行权限
-> 必要时请求人工审批
-> 在副作用发生前再次校验
-> 执行并记录审计事件
提示词中的“不要访问工作目录之外的文件”不是安全边界。模型可能误解，也可能受到提示注入影响。真正的边界必须由普通程序代码、操作系统权限、沙箱和下游服务策略共同实施。
四个决定可靠性的系统不变量
1. 状态必须只有一个明确所有者
会话阶段、计划版本、已获批动作和任务结果不能由模型、工具回调与 Web 请求分别修改。Harness 应当成为状态的唯一写入者，其他组件只能提交事件或变更请求。
这使系统能够回答几个关键问题：
- 当前状态由谁修改？
- 修改依据的是哪个版本？
- 哪个操作导致了状态跃迁？
- 失败后能否安全重试？
在工程上，可以采用单写者事件循环、Actor、数据库事务，或者带版本号的状态机。具体实现可以不同，但所有权必须清晰。
2. 并发状态变更必须串行化
一个 Agent 可能同时收到模型回调、工具完成通知、用户取消请求和审批结果。如果这些事件并发写入同一任务，常见后果包括重复发送、取消后继续执行，以及旧计划覆盖新计划。
需要同时使用两类机制：
- 用锁、队列或事务串行化写入；
- 用版本号或比较交换拒绝基于旧状态生成的动作。
锁只能避免“同时写”，版本检查则可以避免“拿着旧结论晚到一步”。对于外部 API，还应增加幂等键，以防进程崩溃后的重试重复产生副作用。
3. 执行权限应按能力收缩
不要因为 Agent 需要读取仓库，就给它整个主机的 Shell；不要因为它能创建草稿，就默认允许发送邮件。
权限可以拆成更小的能力：
agent_policy:
tools:
- read_file
- write_file
filesystem:
readable_roots:
- /srv/project
writable_roots:
- /srv/project/generated
network:
allowed_hosts: []
approval_required:
- write_file
- send_email
- create_payment
这是一份可改造的策略示例，并非特定产品的配置格式。生产系统还应在容器、服务账号、文件权限和网络策略层重复实施这些限制，避免 Harness 中的一处缺陷变成完整权限逃逸。
4. 在用户可见的边缘再次验证
计划生成时通过校验，不代表执行时仍然安全。状态可能已改变，审批可能已撤销，路径可能经过规范化后指向另一个位置，或者工具参数在中间环节被重写。
因此，最后一次检查必须紧贴真实副作用：
- 在发送邮件前检查最终收件人、正文和附件；
- 在付款前检查最终金额、币种、收款方与审批摘要；
- 在写文件前解析真实路径并检查允许目录；
- 在部署前检查最终制品摘要、环境和变更单。
审批也不应只是一个布尔值。它应绑定到动作的规范化内容，例如工具名、参数、目标对象、状态版本和有效期。任何关键字段变化，都应使原审批失效。
一个可运行的最小 Harness
下面的 Python 示例演示四件事：单一状态所有者、串行化变更、目录级权限，以及绑定具体动作内容的审批。保存为 agent_harness.py 后可直接运行；示例只会写入临时目录。
from __future__ import annotations
import hashlib
import json
import os
import tempfile
import threading
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class Action:
tool: str
arguments: dict
expected_version: int
def digest(self) -> str:
payload = {
"tool": self.tool,
"arguments": self.arguments,
"expected_version": self.expected_version,
}
encoded = json.dumps(
payload, sort_keys=True, separators=(",", ":")
).encode()
return hashlib.sha256(encoded).hexdigest()
class AgentHarness:
def __init__(self, writable_root: Path):
self._root = writable_root.resolve()
self._lock = threading.RLock()
self._version = 0
self._approved_digests: set[str] = set()
def propose_write(self, relative_path: str, content: str) -> Action:
with self._lock:
return Action(
tool="write_file",
arguments={"path": relative_path, "content": content},
expected_version=self._version,
)
def approve(self, action: Action) -> None:
# 生产环境应记录审批人、时间、有效期和审批理由。
with self._lock:
self._approved_digests.add(action.digest())
def execute(self, action: Action) -> Path:
# 锁覆盖状态检查、副作用与状态提交，防止并发交错。
with self._lock:
if action.expected_version != self._version:
raise RuntimeError("拒绝执行：动作基于过期状态")
if action.tool != "write_file":
raise PermissionError("拒绝执行：工具不在允许列表中")
digest = action.digest()
if digest not in self._approved_digests:
raise PermissionError("拒绝执行：动作尚未获得精确审批")
# 在真正写入前重新解析并检查目标路径。
target = (self._root / action.arguments["path"]).resolve()
if target == self._root or self._root not in target.parents:
raise PermissionError("拒绝执行：目标超出可写目录")
target.parent.mkdir(parents=True, exist_ok=True)
temporary = target.with_suffix(target.suffix + ".tmp")
temporary.write_text(action.arguments["content"], encoding="utf-8")
os.replace(temporary, target)
self._approved_digests.remove(digest) # 一次性审批
self._version += 1
return target
if __name__ == "__main__":
with tempfile.TemporaryDirectory() as directory:
harness = AgentHarness(Path(directory))
action = harness.propose_write("reports/result.txt", "approved output\n")
harness.approve(action)
written = harness.execute(action)
print("written:", written)
print(written.read_text(encoding="utf-8"), end="")
# 同一个动作不能再次执行：版本已变化，审批也已消费。
try:
harness.execute(action)
except Exception as error:
print("second execution blocked:", error)
运行命令：
python3 agent_harness.py
这个示例适合解释设计原则，但不是完整沙箱。真实文件系统中还要处理符号链接竞态、挂载点变化与进程级权限；高风险执行器更适合放进隔离容器，并使用 openat、O_NOFOLLOW 等机制或专门的安全文件服务。
对于邮件、付款等远程操作，还应给下游请求附加幂等键，并把动作摘要写入不可变审计日志。否则 Harness 在请求成功后、状态提交前崩溃，重试仍可能导致重复操作。
审批边界应该展示“将要发生什么”
审批界面不应只显示“Agent 请求调用工具”。审批人需要看到最终效果，而不是模型的自然语言解释。例如付款审批至少应展示：
{
"operation": "create_payment",
"recipient": "vendor-4821",
"amount": "1250.00",
"currency": "USD",
"invoice": "INV-2025-1042",
"state_version": 37,
"expires_at": "2025-06-01T12:00:00Z"
}
Harness 可以对规范化后的对象计算摘要，并让审批记录绑定该摘要。若收款方或金额发生变化，就必须重新审批。这样，人工审批才是执行授权，而不只是流程装饰。
上线前的检查清单
评审一个生产 Agent 时，可以从以下问题入手：
- 是否存在唯一、明确的状态写入者？
- 并发事件是否通过队列、锁或事务串行化？
- 每个动作是否绑定状态版本，并能拒绝过期计划？
- 工具权限是否按目录、主机、资源和操作类型收缩？
- 审批是否绑定规范化参数，而非绑定一段模糊描述？
- 副作用发生前是否执行最后一次确定性校验？
- 外部写操作是否具有幂等键、超时与重试策略？
- 是否能从审计日志还原提议、审批、执行和结果？
- Harness 失效时，操作系统或下游服务是否仍有第二道边界？
OpenClaw 等现实案例的价值，在于提醒开发者：Agent 故障通常是模型、并发、权限和系统状态共同作用的结果。可靠性不能只靠更长的提示词。把 Harness 当成控制平面，并把关键不变量落实为可测试的代码，才是将 Agent 从演示环境带入生产系统的关键一步。