Agent 循环前必须锁定工具契约

一个假设一切的 Agent 并不算聪明。它只是一个无类型的 RPC 客户端。你应该在循环启动前锁定每一个工具。额外的 prompt 文本替代不了 schema。

The model is the last variable

你失败不是因为模型看起来蠢。你失败是因为适配器接受了虚构内容。Agent 捏造了一个字段，你的 mutate 路径随后执行了它。

钱可以转出去。工单可以关闭。文件可以消失。别再要求模型小心了。让工具拒绝非法的结构。

然后给循环一个硬性的副作用预算。这不是品味之争。这是基本的生产卫生规范。

What assuming looks like in logs

你会在日志中看到同样的三种谎言。它们出现在任何花哨的推理链之前。

必填键缺失，于是模型伪造了一个。

重试毫无羞耻地重复一次付费的变更操作。

字符串枚举被忽略，于是越界操作落地了。

这些都不是 prompt 问题。它们是契约问题。如果你不能命名允许的键，你就无法调试调用。如果你不能命名预算，你就无法停止循环。

先读原始的工具 JSON。别先读助手的散文。散文是表演。载荷才是事故本身。

A contract is a schema plus brakes

让契约刻意保持枯燥。枯燥意味着更容易测试。

Schema：名称、类型、枚举、必填字段

幂等键：一次变更操作对应一个键

副作用预算：每次运行最大付费调用次数

超时：墙上时钟，而不是"直到感觉做完了"

你把这些用代码锁定，而不是用系统 prompt 锁定。Prompt 是一个建议。围栏是一道门。

Artifact: a tool fence you can run tonight

下面的代码是一个标记好的提案。把它当作本地测试工具。等你加上传输层再调用 vendor。

# tool_fence.py
from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Any, Callable, Mapping

ALLOWED_ACTIONS = {"create_ticket", "add_comment", "noop"}

TICKET_SCHEMA = {
    "type": "object",
    "required": ["action", "ticket_id", "idempotency_key"],
    "additionalProperties": False,
    "properties": {
        "action": {"type": "string", "enum": list(ALLOWED_ACTIONS)},
        "ticket_id": {"type": "string", "minLength": 1, "maxLength": 64},
        "body": {"type": "string", "maxLength": 2000},
        "idempotency_key": {"type": "string", "minLength": 8, "maxLength": 128},
    },
}

class ContractError(ValueError):
    pass

def validate_shape(payload: Mapping[str, Any], schema: dict) -> None:
    if not isinstance(payload, dict):
        raise ContractError("payload must be an object")
    for key in schema.get("required", []):
        if key not in payload:
            raise ContractError(f"missing key: {key}")
    props = schema["properties"]
    if schema.get("additionalProperties") is False:
        extra = set(payload) - set(props)
        if extra:
            raise ContractError(f"unknown keys: {sorted(extra)}")
    action = payload.get("action")
    if "action" in payload and action not in props["action"]["enum"]:
        raise ContractError(f"illegal action: {action}")
    ticket_id = str(payload.get("ticket_id", ""))
    if "ticket_id" in payload and not (1 <= len(ticket_id) <= 64):
        raise ContractError("ticket_id length out of range")
    ident = str(payload.get("idempotency_key", ""))
    if "idempotency_key" in payload and not (8 <= len(ident) <= 128):
        raise ContractError("idempotency_key length out of range")

@dataclass
class SideEffectBudget:
    max_mutations: int
    used: int = 0

    def consume(self, action: str) -> None:
        if action == "noop":
            return
        if self.used >= self.max_mutations:
            raise ContractError("side-effect budget exhausted")
        self.used += 1

@dataclass
class ToolFence:
    schema: dict
    budget: SideEffectBudget
    seen_keys: set[str] = field(default_factory=set)
    audit: list[dict] = field(default_factory=list)
    clock: Callable[[], float] = time.time

    def run(self, payload: Mapping[str, Any], mutate: Callable[[dict], dict]) -> dict:
        started = self.clock()
        validate_shape(payload, self.schema)
        action = str(payload["action"])
        ident = str(payload["idempotency_key"])
        if ident in self.seen_keys:
            self.audit.append({"status": "duplicate", "key": ident})
            return {"ok": True, "duplicate": True}
        self.budget.consume(action)
        self.seen_keys.add(ident)
        result = mutate(dict(payload))
        self.audit.append(
            {
                "status": "applied",
                "action": action,
                "key": ident,
                "ms": int((self.clock() - started) * 1000),
            }
        )
        return result

现在加上故意会失败的测试。你要的是先有红色测试。

# test_tool_fence.py
from tool_fence import ContractError, SideEffectBudget, TICKET_SCHEMA, ToolFence

def apply(payload):
    return {"ok": True, "action": payload["action"]}

def make_fence(n=1):
    return ToolFence(TICKET_SCHEMA, SideEffectBudget(max_mutations=n))

def test_rejects_forged_action():
    fence = make_fence()
    bad = {
        "action": "delete_everything",
        "ticket_id": "T-1",
        "idempotency_key": "k-12345678",
    }
    try:
        fence.run(bad, apply)
        raise AssertionError("forged action passed")
    except ContractError as exc:
        assert "illegal action" in str(exc)

def test_retry_does_not_double_mutate():
    fence = make_fence(n=2)
    payload = {
        "action": "add_comment",
        "ticket_id": "T-1",
        "body": "ship after contract",
        "idempotency_key": "k-retry-01",
    }
    first = fence.run(payload, apply)
    second = fence.run(payload, apply)
    assert first["ok"] is True
    assert second["duplicate"] is True
    assert fence.budget.used == 1

def test_budget_kills_the_loop():
    fence = make_fence(n=1)
    first = {
        "action": "create_ticket",
        "ticket_id": "T-1",
        "idempotency_key": "k-aaaaaaa1",
    }
    second = {
        "action": "create_ticket",
        "ticket_id": "T-2",
        "idempotency_key": "k-aaaaaaa2",
    }
    fence.run(first, apply)
    try:
        fence.run(second, apply)
        raise AssertionError("budget did not fire")
    except ContractError as exc:
        assert "budget" in str(exc)

用 pytest 运行这个文件。

python -m pip install pytest
python -m pytest test_tool_fence.py -q

你现在有了一个 kill switch。Agent 仍然可以对话。适配器仍然可以拒绝。

Why the tests are the product

一个绿色的聊天演示几乎什么都证明不了。一个红色的契约测试证明围栏有效。把非法操作测试永久保留在 CI 中。把重复键测试永久保留在 CI 中。

如果任何一个测试消失了，循环就没有保险了。把删除的围栏测试当作删除的认证测试来处理。

Map model output before the fence

很多技术栈把工具包装成 JSON 字符串。你应该解析一次，拒绝两次。

import json
from tool_fence import ContractError

def strip_fence(raw: str) -> str:
    text = raw.strip()
    if text.startswith("```"):
        lines = text.splitlines()
        if lines and lines[-1].strip() == "```":
            lines = lines[1:-1]
        return "\n".join(lines)
    return text

def parse_tool_json(raw: str) -> dict:
    text = strip_fence(raw)
    if not text:
        raise ContractError("empty tool payload")
    try:
        data = json.loads(text)
    except json.JSONDecodeError as exc:
        raise ContractError(f"invalid json: {exc}") from exc
    if not isinstance(data, dict):
        raise ContractError("tool payload must be an object")
    return data

永远不要把列表传给 mutate。永远不要传递残留的 markdown 围栏。永远不要强制转换一个近似匹配的枚举。最接近的动词是数据被删除的方式。

Decision table: block, retry, or escalate

不要在热循环里即兴发挥。提前决定每个分支。

把这张表放在 on-call 文档旁边。如果一行缺失，你就是在压力下猜测。

场景	操作
非法 action	block
重复 idempotency_key	block（已处理）
预算耗尽	block
语义错误（在合法结构内）	escalate
Authz 错误	escalate
Schema 合法但指向错误目标	escalate

Soak the fence, not the chatbot

免费的 token 诱惑你聊更久。这不是 Agent 的正确训练方向。你应该把 token 花在非法载荷上。

构建一个生成器。故意喂垃圾数据。

# fuzz_tools.py
import json
import random
from tool_fence import ContractError, SideEffectBudget, TICKET_SCHEMA, ToolFence

ACTIONS = ["create_ticket", "add_comment", "noop", "drop_table", "", None]
IDS = ["T-1", "", "T" * 80, 12]

def junk():
    return {
        "action": random.choice(ACTIONS),
        "ticket_id": random.choice(IDS),
        "body": "x" * random.choice([0, 10, 5000]),
        "idempotency_key": random.choice(["short", "k-12345678", None]),
        "extra": "nope",
    }

def main(n=200):
    fence = ToolFence(TICKET_SCHEMA, SideEffectBudget(max_mutations=3))
    blocked = 0
    applied = 0
    for _ in range(n):
        payload = {k: v for k, v in junk().items() if v is not None}
        try:
            fence.run(payload, lambda p: {"ok": True})
            applied += 1
        except (ContractError, TypeError):
            blocked += 1
    print(json.dumps({
        "n": n,
        "blocked": blocked,
        "applied": applied,
        "used": fence.budget.used,
    }))

if __name__ == "__main__":
    main()

python fuzz_tools.py

你要 blocked 高，used 低。如果 used 快速增长，你的围栏就是做样子。

Read the JSON like an incident

不要欣赏 dump。去审问它。每次运行问三个问题。

• 哪个非法 action 差点通过了？
• 是否有重复的 key 消耗了预算？
• noop 时墙上时间是否爆炸了？

把答案写在 PR 里。如果你回答不了，你就没有测试过。一次静默的 fuzz 运行只是 CPU 发热。

Count mutations the way you count deploys

你已经对部署做了上限限制。用同样的方式限制 Agent 的写操作。

• 一次运行，三次变更，然后停止
• 一个工单，一次创建，然后只能评论
• 一个幂等键对应一个用户意图

把那些上限放在配置文件里，而不是聊天里。配置文件可以被审查。Prompt 不能。

{
  "max_mutations": 3,
  "max_wall_ms": 8000,
  "allowed_actions": ["create_ticket", "add_comment", "noop"]
}

在进程启动时加载这个文件。缺失就拒绝启动。缺失刹车不是运行时意外。是部署 bug。

Where a scratch model pool actually helps

你仍然需要在循环里放一个模型。模型生成 JSON，围栏判断那个 JSON。

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

MonkeyCode is an open source project. It offers free model access and a free server option. There is also a free 10 million token pool for this kind of drill. Use the pool to generate bad tool calls. Use the free server as a scratch box for the fence.

不要把这些选项当作生产 SLA。把循环指向 scratch box。保存每一个提出的载荷。通过 test_tool_fence.py 重放失败。

一个有用的命令序列是这样的：

mkdir -p artifacts
python fuzz_tools.py > artifacts/fuzz.json
python -m pytest test_tool_fence.py -q

如果模型无法输出 schema，不要增大 prompt。缩小工具反而。更小的枚举是特性。更大的故事不是。

What this opinion rejects

这个观点拒绝了三个常见习惯。

更多 system prompt。谨慎的词语不能对 JSON 做类型检查。

无限重试。没有 key 的重试会重复副作用。

Eval as vibes。"它看起来有帮助"不是一个关卡。

你可以保留 Agent。你不能保留开放的 mutate 路径。无类型的支付 RPC 永远不会上线。无类型的 Agent 也不应该上线。

这个围栏不能证明模型是聪明的。它只能证明适配器是严格的。

围栏能拦住	围栏拦不住
非法 action 结构	合法结构指向错误工单
缺失必填键	body 字符串内的语义谎言
重复 mutation	mutate 函数里的 Authz bug
Schema 注入	仍然符合 schema 的 prompt 注入
	重启后忘记内存中的 seen_keys

后续要把幂等键存到 SQLite 或 Redis 里。内存随进程消失。在添加更多工具前先持久化审计日志。

Who should not use this

在以下情况跳过这个方案。

• 你没有 mutate 路径，只有只读搜索
• 你无法列出有限的 action 枚举
• 你需要多租户 authz 但现在没有
• 你期望模型每天发明新的工具名

如果你不能列出工具，你还没准备好。没有目录的循环是即兴表演。在 notebook 里即兴没问题。对着线上工单即兴不行。

Ship the contract first

把 schema 放进 review。把预算放进 review。然后再让 Agent 说话。

没有围栏的循环就是一个慢脚本。你不会用那个脚本做支付上线。也不要把它用于工单。

如果你已经有了一个 scratch server 和免费的 token pool，在添加另一个 prompt 规则之前先在那里运行红色测试。