第 1 部分:让你的 AI 智能体变得"无聊"——确定性层
想在高风险系统里用上大语言模型(LLM)智能体,诀窍不在于换个更聪明的模型,而在于把模型圈进单个节点,让系统其余部分都是普通、可测试的代码。下面讲具体的结构性做法,并给出实现所需的契约和类型。
做演示时,大家都爱那种会自己循环、调用工具、自己"想办法搞定"的自主智能体。可到了生产环境,没人待见它们。只要智能体的行为取决于模型今天恰好走了哪条路,你就没法测试它、没法审计它,更不能让它碰任何要紧的东西。在受监管或后果严重的系统里——资金流转、医疗、基础设施——"一般都能跑通"是绝对过不了关的。
这是一个用于生产环境 LLM 系统的六级成熟度模型,这里是第 1 级:确定性层。在你能做评估(evals)、置信度路由,或者任何更高层的功能之前,得先让底下这一层稳住不动。这一层的全部要点只有一个想法——把模型收进单个节点,让周围一切都是普通、可测试的代码——再通过几个设计动作来表达。模型的智能保留,但几乎全部的不确定性都被去掉。这篇文章给出的是契约,不只是概念。
智能体只提议,不动手
最重要的一条规则:智能体是一个纯函数,从上下文映射到一个待定的决策——所谓"纯",是相对业务状态而言;只要给定它的网关(gateway),输出就是确定的(网关里那次不确定的调用,也就是模型,是注入进来的,测试时可以换成假的)。它无权改变现实世界。真正动手的是一个单独、不聪明、测试覆盖得很足的组件——底层执行层(substrate)。由它来应用提议,而且必须在拿到所需审批之后。
from typing import Protocol, Literal
from dataclasses import dataclass
@dataclass(frozen=True)
class Proposal:
decision_id: str
capability: str
action: dict # the structured, proposed change — NOT yet applied
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: ... # pure: no side effects on business state
class Substrate(Protocol):
def apply(self, proposal: Proposal, approval: "Approval") -> "Effect": ... # the ONLY mutator
因为 propose() 是纯函数,它最核心的性质可以用一行断言概括:
def test_propose_is_pure():
agent = ClassifyAgent(gateway=FakeGateway(scripted))
assert agent.propose(ctx) == agent.propose(ctx) # same context → same proposal; no world touched
这道边界带来三个好处:
- 可测试。
propose()是纯函数——同样的上下文,产出同样的建议。要测逻辑,不用去 mock 整个外部世界。 - 安全。 就算 agent 被越狱或出了 bug,它产出的也只是一个坏建议,而不是一个坏操作。影响范围止步于「护栏或人说了一句不行」。
- 可组合。 Agent 之间从不互相调用。任务通过底座流转(比如一张 cases 表加一个调度器),所以不存在隐藏的副作用调用链,也就没有什么难以理清的东西。
就这一条约束,把「AI 干了件我们说不清的事」变成了「AI 提了个建议,而且有据可查是谁批准了它」。
每个能力对应一张固定图
自由形态的 ReAct 循环适合探索,但完全无法提供保证。把每个能力建模成一串固定的节点,只在需要判断的地方才动用模型,其余部分全是普通代码。下面这个箭头示意图为便于阅读做了简化;完整的节点列表(含 pre_check 和 memory_write)放在下一节。
entry → load context → reason (LLM) → output guardrail → verify →
judge (sampled) → compose confidence → route → prepare proposal → record → exit
| 自由形态 ReAct 循环 | 固定图(本文方案) | |
|---|---|---|
| 控制流 | 模型决定下一步 | 提前已知 |
| 可测试性 | 难(路径会变) | 每个节点可独立测试 |
| 延迟 / 成本 | 无界 | 有界、可预测 |
| 审计 | 从 trace 里反推 | 每次都留下统一的行 |
| 适合 | 开放式探索 | 必须保证结果的决策 |
你牺牲了一些「聪明」,换回来的是对系统行为做出推理的能力。真正需要聪明的部分——在混乱输入上做判断——仍然留在模型最擅长的那个节点里,周围全是你可以直接读懂、读透的代码。
一次请求的解剖
「让 agent 变得确定」这话很难落地,除非你先看清一次请求长什么样。所以我们来把单次请求沿着这张图走一遍,一个节点一个节点地看——也就是上面那张示意图背后的实现。
状态对象。 每个节点都通过图读、写同一个带类型的 state 值。把这一点做明确,是这场仗打赢的一半——正是它让节点可以独立测试:构造一个 state,断言结果即可。
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]
节点契约。每个节点形态都一样:state -> state。除了那唯一的模型节点,其余全是确定性的。正因为形态统一,你才能对每个节点单独做单元测试,也能从整体上推断这条路走不走得通。
from typing import Protocol
class Node(Protocol):
name: str
def run(self, state: GraphState, deps: "Deps") -> GraphState: ...
图。这 8 个节点是所有能力共用的,只有少数几个跟具体能力绑定。加一项新能力 = 实现约 4 个节点,其余 8 个直接继承。
1 entry 生成 decision_id,绑定 tenant + identity [共用]
2 pre_check 校验输入、解析引用、低成本提前返回 [能力相关]
3 context_load 只取这次决策需要的上下文 [共用]
4 llm_decision 推理环节 —— 结构化输入,结构化输出 [能力相关]
5 output_guardrail 对模型输出做 PII / 策略清洗 [共用]
6 verification 确定性的、跟能力相关的正确性校验 [能力相关]
7 judge 抽样让第二个模型复审(高风险时) [共用]
8 confidence_compose 把各路信号合成一个分数 [共用]
9 routing auto / hitl / reject 三选一 [共用]
10 prepare_proposal 把最终方案定形 [能力相关]
11 memory_write 写入 agent 自己的审计记忆(不是业务状态) [共用]
12 exit 追加不可篡改的账本条目 [共用]
真正承重的两个节点是 llm_decision 和 verification。前者是唯一的非确定性节点——结构化输入、按 schema 约束输出、对不上就重试——而它周围的一切都把它的输出当作不可信,直到校验通过为止(下一节细说)。后者是确定性的、跟具体能力绑定的代码,整个「把模型关进笼子」的思路就架在它上面:
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
固定形态的意义在于:控制流一眼可读(路径本身就是这张图),模型被限制在一个节点里,每条决策的形态都一致——每次形态一致,就意味着每次写出的审计记录一致,也意味着加校验、加指标、加护栏都只有一个地方可改。
用结构化输出代替自由文本
节点 4 值得单独展开讲,因为相当一部分 LLM 的「脆」都来自同一个选择:让模型返回自由文本,然后再去解析它。散文式表达本身有歧义,格式在两次调用之间会漂移,你的下游代码离崩溃只差一句意料之外的措辞。「当然!大概是通过吧,不过也可能要升级处理」——你于是背上了一个 NLU 问题,得从里面抠出「通过」,而明天换个说法就把你的正则搞挂了。你等于把系统绑死在了模型的文字风格上,而那是它最不稳定的东西。
解法不花哨,但很管用:约束模型只能吐出一个经过校验的结构,其他一律当作调用失败去重试。约束方式有三种,按需要的保证强度来选:
方式 做法 保证强度 适用场景
--------------------------------------------- ------------------------------ ------------------------------------ --------------------------------------
Schema 引导(JSON Schema / response_format) 要求返回符合 schema 的 JSON 强,由服务方强制 大多数场景
工具 / 函数调用 模型输出一次带类型的工具调用 强;对「带参数执行 X」很自然 决策对应某个动作时
语法约束解码 把 token 约束在一套语法内 硬保证(不可能输出非法内容) 严格/受监管的格式、本地模型
# 把契约写成类型 —— 下游消费的是带类型的对象,而不是散文
from pydantic import BaseModel
from typing import Literal
class Decision(BaseModel):
decision: Literal["approve", "escalate", "reject"] # 枚举本身即一道护栏
confidence: float
reasons: list[str]
约束式生成能减少畸形输出,但不能根除。所以要闭环处理:每条响应都校验,不匹配就重试——把校验报错喂回去。限制重试次数,失败就关闭(fail closed)。
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:
# 从 base_prompt 重建 —— 不要在已经附加过的 prompt 上继续附加(会不断累积)
prompt = f"{base_prompt}\n\nYour previous output was invalid: {e}. Return JSON only."
raise NonConformingOutput() # 失败即关闭 —— 绝不把猜测结果交给下游
在边界处做校验,意味着系统其余部分只会看到格式良好的决策;「模型到底有没有守规矩」这个麻烦问题被限制在这一个函数里。你把一个模糊的自然语言理解(NLU)问题换成了一个清晰的校验问题——不需要解析层,换模型或换提示词都稳,自带护栏(固定的枚举值不可能返回列表之外的选项),而且有一份带类型的契约可供黄金测试(golden test)断言。
有一点必须大声说明:结构化约束的是形式,不是正确性。一个完全合法的 {"decision":"approve","confidence":0.99} 可能是彻底错的。结构化输出消除的是解析失败,不是判断失败——所以它是生产系统的底座,上面还要叠评测、验证和置信度,而且不能被这些东西替代。
粘合层:置信度 → 路由
置信度(模型信号 + 验证 + 采样裁判)在置信度节点合成;路由节点用单一阈值把它变成四条路径之一,之后 prepare_proposal 再把结果盖到提案上:
def route(confidence: float, verified: bool, T: float = 0.85) -> str: # T is per-capability config, not a constant
if not verified: return "reject"
if confidence >= T: return "auto"
if confidence >= T - 0.2: return "hitl_recommended" # close: pre-fill the proposal for a human
return "hitl_required" # low: a human decides from scratch
verified 来自校验节点(verified = all(checks.values())),不是 agent 自己给自己打的分;四条路由状态,路由器都会输出。注意顺序:路由发生在提案成型之前(节点 9 → 节点 10),所以它拿到的是裸的 confidence 和 verified 标志,而不是一个 Proposal 对象。T 一开始要保守些——全部交给人工——只有数据证明某个切片安全了,才针对该切片往下调。人的注意力是最贵的资源,要花在系统真正拿不准的地方。
账本:只追加的权威记录
每一个被提出的决策都会变成一行不可变的记录,作为决策流程的最后一个节点无条件写入。它不是一行普通日志,而是「发生了什么、为什么这么发生」的权威记录。
CREATE TABLE decision_ledger (
decision_id TEXT PRIMARY KEY,
ts TIMESTAMPTZ NOT NULL,
tenant_id TEXT NOT NULL,
capability TEXT NOT NULL,
inputs_hash TEXT NOT NULL, -- hash, not the raw sensitive payload
model_version TEXT NOT NULL,
prompt_version TEXT NOT NULL,
decision JSONB NOT NULL,
confidence REAL NOT NULL,
routing TEXT NOT NULL, -- auto | hitl_* | reject
outcome TEXT, -- recorded later as a NEW superseding row, never an in-place UPDATE
supersedes TEXT REFERENCES decision_ledger(decision_id),
prev_hash TEXT, -- optional hash-chain for tamper-evidence
entry_hash TEXT
);
-- append-only: no UPDATE/DELETE grants; corrections are new rows that set `supersedes`.
把输入做哈希,不要把输入囤起来——要可验证,不要背责任。做成只追加(append-only):修正用新记录覆盖旧结论,绝不原地改写。如果你能 UPDATE 账本,那它就不是审计踪迹;把这条权限收回。高负载下也绝不能跳过写入——它是记录的记录,不是可以随手丢掉的遥测数据。
有界 ReAct:唯一该用循环的地方
如果你把上面这些都落实了——固定图、模型只用在一个节点里——早晚会撞上一类不匹配的问题:一件开放式的事,模型必须去查点东西,对查到的结果推理一下,可能再查一些,然后才做决定。这正是 ReAct 式工具循环的用武之地。错不在循环,而在无界循环。要把自主性当成一种刻意保留、由护栏守着的例外,除此之外哪儿都别给。
四条护栏把它控制住:(1) 硬性迭代上限——用 for,绝不用 while(not done),也就是不让模型自己决定何时停下;(2) 每个能力的工具白名单——只能调用显式为该能力划定的那份清单里的工具;(3) 每轮都留下完整轨迹——每一步都记录,让决策可复现;(4) 出口和其他一切保持一致——循环的输出照样要过护栏、验证、置信度和路由,而且仍然只是提议,绝不直接执行。
class DisallowedTool(Exception): ... # 循环伸手去够白名单外的工具时抛出
ALLOWED_TOOLS = { # 护栏 2:按能力划的白名单,而不是“所有工具”
"enrich_request": {"search_kb", "fetch_record", "lookup_reference"},
}
@dataclass
class Step: # 护栏 3:每轮一行轨迹
i: int; action: str; args: dict; result_digest: str
def bounded_react(state, deps, capability, MAX_STEPS=6) -> Proposal:
allow = ALLOWED_TOOLS[capability]
trace: list[Step] = []
for i in range(MAX_STEPS): # 护栏 1:硬性上限
action = deps.model.next_action(state, tools=sorted(allow))
if action.is_final: # 先检查这个——最终答案不带工具调用
return finalize(action.proposal, trace) # 护栏 4:返回的仍是 PROPOSAL
if action.tool not in allow: # 纵深防御
raise DisallowedTool(action.tool)
result = deps.tools[action.tool](**action.args)
trace.append(Step(i, action.tool, action.args, digest(result)))
state = state.with_observation(result) # 循环局部状态类型,不是固定图的 GraphState
return escalate("hit step cap", trace) # 有界:触顶 → 返回一个标记为 “hitl_required” 的 Proposal
轨迹会写入账本条目,所以基于循环的决策和固定图决策一样,都能完整重建。有两条铁轨各值得写一个测试——它一定会终止,而且它碰不到白名单之外的工具:
def test_always_terminates():
deps = fake_deps(model=never_final) # 一个永远不返回 is_final 的模型
p = bounded_react(state, deps, "enrich_request", MAX_STEPS=3)
assert p.routing == "hitl_required" # 撞到上限 → 转人工,没有卡死
def test_disallowed_tool_refused():
deps = fake_deps(model=calls("danger_tool")) # 一个不在白名单里的工具
with pytest.raises(DisallowedTool):
bounded_react(state, deps, "enrich_request")
至于到底要不要用这套东西,判断规则很简单:得到这个决策,所需的步骤数量和顺序会依赖过程中发现的东西吗?不会 → 用固定图(大多数能力都属此类)。会 → 用有界循环、四条铁轨,并作为例外记录下来。如果你伸手去用循环是「为了保险」或「为了灵活」,停手——那通常只是穿了件马甲的固定图场景。你并不需要的灵活性,就是你会凌晨两点爬起来调的「不确定性」。
反模式
-
智能体「就这一次」地写业务状态。 这样一来它就不再纯粹、无法测试,一个 bug 从「提案不佳」升级成「生产事故」。让底层存储成为唯一的改动入口。
-
隐式状态:步骤之间用临时的元组/字典传递——你就丧失了单独测试某个节点的能力。把状态做成有类型的对象。
-
提示词里写「返回 JSON」,却不强制 schema——你收到的还是散文、代码围栏或者尾巴上的废话。用 schema/工具/语法约束来强制,然后校验并重试;约束 ≠ 保证。
-
从模型那里拿来的「置信度」。 校准是歪的;应该用独立信号自己拼出来。
-
可改或可跳过的审计。 如果你能
UPDATE账本,它就不是审计轨迹;如果你在高负载下丢掉写入,你就没有记录可依。 -
「为了灵活」的无界循环。 默认用固定图。真要循环,就带上
for上限和按能力划分的白名单——绝不要while not done或者「所有工具都可用」。
小结
第一层级说到底就是一条纪律,始终如一地执行:把模型限制在一个节点里,让智能体变成一个只负责提建议的纯函数,把这个节点约束到经过验证的 schema 上,用独立信号拼出置信度,把拿不准的情况交给人工,审批之后只让一个简单的底层组件执行实际修改,再把每个决策写入只能追加的账本。如果某项能力确实需要自主性,就给它设预算——步数上限、工具白名单、追踪记录,以及和其他所有环节一样的输出检查。
模型仍然做它最擅长的事——对杂乱的输入做出判断——但它周围的系统是确定性的、可测试的、可审计的。无聊,但这是最好的那种无聊:可预测到能测试,可控到能信任,站得住脚到能上线。这就是整个系列其他内容所依赖的地基。
系列:在生产环境运行 LLM 系统——6 个层级中的第 1 层:确定性。
本文采用知识共享署名-相同方式共享 4.0 国际许可协议授权。