Part 1: Make your AI agents boring: the determinism layer

from typing import Protocol, Literal
from dataclasses import dataclass

@dataclass(frozen=True)
class Proposal:
    decision_id: str
    capability: str
    action: dict                  # 结构化建议变更 —— 尚未执行
    confidence: float
    routing: Literal["auto", "hitl_recommended", "hitl_required", "reject"]
    reasoning: list[str]
    evidence: list[dict]

class Agent(Protocol):
    def propose(self, ctx: "Context") -> Proposal: ...     # 纯函数：不改变业务状态

class Substrate(Protocol):
    def apply(self, proposal: Proposal, approval: "Approval") -> "Effect": ...  # 唯一的修改器

def test_propose_is_pure():
    agent = ClassifyAgent(gateway=FakeGateway(scripted))
    assert agent.propose(ctx) == agent.propose(ctx)   # 相同上下文 → 相同建议；未修改现实世界

entry → load context → reason (LLM) → output guardrail → verify →
        judge (sampled) → compose confidence → route → prepare proposal → record → exit

自由形式 ReAct 循环     固定图结构（此方案）
--------------  -----------------------  ---------------------------------
控制流          模型决定下一步        提前已知
可测试性        困难（路径变化）      每个节点独立测试
延迟/成本       无上限                有界且可预测
审计            从轨迹重建            每次都是统一记录
适用场景        开放式探索            必须保证的决策

from typing import TypedDict, Literal, Optional

class GraphState(TypedDict):
    decision_id: str
    tenant_id: str
    identity: dict                      # 身份/权限来源（在入口时设置）
    inputs: dict                        # 验证后的请求
    context: dict                       # 加载的内存/引用切片
    model_output: Optional[dict]        # LLM 节点输出的原始结构化数据
    guardrail: dict                     # 应用的阻止/删除操作
    verification: dict                  # 确定性检查结果
    judge: Optional[dict]               # 第二意见结果（若采样）
    confidence: Optional[float]         # 综合评分
    routing: Optional[Literal["auto", "hitl_recommended", "hitl_required", "reject"]]
    proposal: Optional[dict]            # 最终成型的提案
    ledger_entry_id: Optional[str]

from typing import Protocol

class Node(Protocol):
    name: str
    def run(self, state: GraphState, deps: "Deps") -> GraphState: ...

1  entry              mint decision_id, bind tenant + identity              [shared]
2  pre_check          validate inputs, resolve refs, cheap early-outs       [capability]
3  context_load       fetch only the context this decision needs            [shared]
4  llm_decision       the reasoning step — structured in, structured out    [capability]
5  output_guardrail   PII / policy scrub on the model output                [shared]
6  verification       deterministic, capability-specific correctness checks [capability]
7  judge              sampled second-model review (high-stakes)             [shared]
8  confidence_compose composed score from the signals                       [shared]
9  routing            auto vs hitl vs reject                                [shared]
10 prepare_proposal   shape the final proposal                              [capability]
11 memory_write       write the agent's own audit memory (not business state)[shared]
12 exit               append the immutable ledger entry                     [shared]

def run(self, state, deps):
    out = state["model_output"]
    checks = {
        "in_enum":   out["decision"] in ALLOWED_DECISIONS,        # 不能返回列表外的操作
        "schema_ok": matches_schema(out, DECISION_SCHEMA),
        "rules_ok":  deps.rules.check(out, state["inputs"]),      # 业务不变量
    }
    state["verification"] = {"checks": checks, "score": sum(checks.values()) / len(checks)}
    return state

方法                                         如何                             保证                             使用场景
---------------------------------------------  ------------------------------  ------------------------------------  --------------------------------------
基于模式（JSON Schema / response_format）  要求生成符合特定模式的JSON   强，由提供方强制执行             大多数场景
工具/函数调用                              模型生成带类型的工具调用     强；适合"使用参数执行X"的场景   决策对应具体操作
语法约束解码                               限制生成的token符合语法    严格保证（无法生成无效内容）    严格/监管格式，本地模型

# 以类型定义契约 —— 下游消费的是类型化对象，而非自然语言
from pydantic import BaseModel
from typing import Literal

class Decision(BaseModel):
    decision: Literal["approve", "escalate", "reject"]   # 枚举本身即是安全护栏
    confidence: float
    reasons: list[str]

from pydantic import ValidationError
class NonConformingOutput(Exception): ...

def decide(base_prompt, model, max_retries=2) -> Decision:
    prompt = base_prompt
    for _ in range(max_retries + 1):
        raw = model.generate(prompt, schema=Decision.model_json_schema())
        try:
            return Decision.model_validate_json(raw)            # 成功：获得类型化对象
        except ValidationError as e:
            # 从基础提示重新构建 —— 不要追加到已存在的提示（会累积）
            prompt = f"{base_prompt}\n\nYour previous output was invalid: {e}. Return JSON only."
    raise NonConformingOutput()      # 失败封闭 —— 永远不向下传递猜测结果

def route(confidence: float, verified: bool, T: float = 0.85) -> str:  # T是按能力配置的参数，非固定值
    if not verified:           return "reject"
    if confidence >= T:        return "auto"
    if confidence >= T - 0.2:  return "hitl_recommended"   # 接近阈值：预填充提案供人工确认
    return "hitl_required"                                 # 低置信度：人工从零开始决策

verified 是由验证节点推导得出（verified = all(checks.values())），而非代理自身设置的字段，路由器会发出所有四种路由状态。注意顺序：路由在提案成形前执行（节点 9 → 节点 10），因此它接收的是原始置信度和 verified 标志——而非 Proposal 对象。从保守的 T 开始——所有内容都交给人类处理——仅当数据证明其安全性时，再按切片逐步降低阈值。人类的注意力——这一昂贵的资源——只会被用在系统不确定的地方。

### 账本：只追加的记录

每个提议的决策都会成为账本中不可更改的一行——作为每个决策的最后一个节点，无条件写入。这不是日志条目；而是事件发生及其原因的权威记录。

对输入进行哈希处理，而非存储它们——实现可验证性的同时避免责任风险。确保账本只追加：更正通过新行超继旧记录，永不覆盖。如果允许更新账本，它就不是审计轨迹；撤销该权限。并且永远不要在高负载时跳过写入——这是权威记录，而非可丢弃的遥测数据。

### 有界 ReAct：循环唯一合法存在的地方

如果你已遵循上述所有规则——固定图结构、模型仅限于单个节点——最终会遇到一个无法适配的问题：需要模型查找信息、分析结果、可能进一步查找，最后做出决策的开放性场景。这就是 ReAct 风格工具循环的用途。错误不在于循环本身，而在于无界循环。将自主性作为经过深思熟虑、受严格限制的例外处理——且仅限于此场景。

四个限制条件确保其可控性：(1) 严格迭代次数上限——禁止使用 while(not done) 这类无限制循环，模型无法自行决定何时停止；(2) 按能力划分的工具白名单——仅能调用明确列出的、与当前能力绑定的工具；(3) 每次迭代的完整追踪记录——每一步都被记录，确保决策可回放；(4) 与其他流程相同的退出机制——循环的输出仍需经过防护网、验证、置信度评估和路由处理，最终仍以提案形式输出，永不直接执行。

是否需要使用此机制的判断规则：做出决策是否需要依赖那些步骤数量和顺序取决于沿途发现内容的步骤？不需要 → 固定图（大多数能力）。需要 → 有界循环，四条规则，作为例外情况记录。如果你为了“安全”或“灵活性”而使用循环，请停止 —— 这通常是固定图场景的伪装。不需要的灵活性就是凌晨2点要调试的非确定性。

### 反模式

- 代理仅在此处编写业务状态。现在它不再是纯函数，无法测试，错误会变成事件而非糟糕的提案。保持基质层为唯一修改者。

- 隐式状态以临时元组/字典在步骤间传递 —— 你将失去单独测试节点的能力。将状态设为类型化对象。

- 提示中要求“返回JSON”但没有模式约束 —— 你仍会得到散文、围栏或尾注。使用模式/工具/语法约束，然后验证并重试；约束 ≠ 保证。

- 从模型中提取“置信度”。校准错误；改用独立信号组合而成。

- 可变或跳过的审计。如果你能更新账本，那不是审计轨迹；如果在负载下丢弃写入，你将没有记录。

- 为“灵活性”使用无界循环。默认使用固定图。使用循环时，采用for循环上限和按能力的允许列表 —— 永远不要使用“while not done”或“所有工具可用”。

### 总结

Level 1 是一项被严格应用的准则：将模型限制在单一节点，使代理成为仅提出建议的纯函数，将该节点约束到经过验证的模式，通过独立信号合成置信度以将临界情况路由给人类，在获得批准后让一个简单的底层系统作为唯一修改者，并将每个决策记录到仅追加的账本中。当某种能力确实需要自主性时，需进行预算控制——设置步骤上限、工具白名单、进行追踪，并与其他所有内容执行相同的输出检查。

模型仍然专注于其擅长的领域——对复杂输入进行判断——但围绕它的系统是确定性的、可测试的、可审计的。以最理想的方式：足够可预测以进行测试，足够受控以建立信任，足够可辩护以投入生产。本系列后续所有内容都是建立在这个基础之上的。

系列文章：在生产环境中运行 LLM 系统 —— 第 1 级：确定性

]

这些文章采用知识共享 署名-相同方式共享 4.0 国际许可协议授权。

creativecommons.org/licenses/by-sa/4.0/deed.en