概率内核,确定性外壳:五个「可」的伪代码实践

承接《Agent 应用工程师的六层》的分水岭判断,把「可观测、可恢复、可验证、可治理、可持续进化」逐个拆成最小工程实现:每个「可」一段伪代码,共享同一个设计原则——状态放内核外面,决策穿过内核,副作用经过关卡。

顶级 Agent 应用工程师的分水岭,不是更会写 Prompt、使用更多框架或更快接入新模型,而是能否把一个概率性的智能内核,包装成可观测、可恢复、可验证、可治理、可持续进化的生产系统。

上一篇写下这句话时,五个「可」还是判断标准。这一篇把它们变成代码——每个「可」给出一段最小可用的伪代码实现,说明它到底在防什么、最小要做到什么、最容易在哪里做错。

先给结论,五段伪代码共享同一个形状:

五个「可」不是五个独立的工程任务,而是同一条设计原则的五次应用——状态放在内核外面,决策穿过内核,副作用经过关卡。每个「可」,就是在这条数据通路上装一个确定性的控制回路。

这个原则可以叫它「概率内核,确定性外壳」——函数式编程里有个经典架构叫 “functional core, imperative shell”(纯逻辑在核心,副作用在外壳);LLM 系统恰好把它反了过来:最不可靠的东西在核心,所有可靠性都得由外壳提供

为什么是这五个「可」:每个都在修一条被打破的假设

传统生产系统的可靠性工程,建立在几条从不言明的假设上。response = llm(prompt) 这一行代码同时打破了它们全部:

「可」被打破的假设外壳的修复手段核心数据结构
可观测行为可以从代码推断出来以「决策」为单位的语义 trace轨迹(trajectory)
可恢复失败后重跑会得到同样结果事件溯源 + 幂等副作用追加写事件日志
可验证正确性可以用断言表达运行时护栏 + 统计评测门golden set + 分数分布
可治理权限在编码时静态确定按后果分级的行动关卡策略表 + 审计日志
可进化行为只随代码变化失败回流 + 版本化 + 灰度用例流水线 + 注册表
flowchart LR
  subgraph shell["确定性外壳"]
    S["状态<br/>(事件日志)"] --> P["构造 prompt"]
    P --> K
    K --> G["护栏<br/>(可验证)"]
    G --> A["行动关卡<br/>(可治理)"]
    A --> E["执行副作用<br/>(幂等键)"]
    E --> S
  end
  K["概率内核<br/>llm()"]
  shell -.->|"全程语义 trace(可观测)"| T["轨迹存储"]
  T -.->|"失败回流为用例(可进化)"| V["评测门 + 灰度"]
  V -.->|"新版本 prompt / 模型"| P

注意图里内核只有一个节点:llm() 是整个系统里唯一允许不确定的地方。下面五节,每节装一个回路。

一、可观测:观测的单位是决策,不是请求

要解决的根本问题:LLM 系统的行为无法从代码推断。传统服务出问题,读代码就能缩小范围;Agent 出问题,代码完全正确,问题在某一步的 prompt、上下文或模型采样里。所以必须把每一次决策的输入、输出和依据都记下来——不是为了画监控大盘,是为了事后能重建「它当时为什么这么想」。

def agent_step(task, state):
    with tracer.span("agent.step", task_id=task.id, step=state.step_n) as span:
        prompt = build_prompt(task, state)
        span.set("gen_ai.request.model", MODEL_ID)        # 模型 + 版本,行为的输入之一
        span.set("agent.prompt_version", PROMPT_VERSION)  # 进化回路要靠它归因
        span.set("gen_ai.input.messages", redact(prompt)) # 全文或摘要,按敏感度决定

        decision = llm(prompt)                            # ← 概率内核,只在这里

        span.set("gen_ai.usage.input_tokens", decision.usage.input)
        span.set("gen_ai.usage.output_tokens", decision.usage.output)
        span.set("agent.decision.kind", decision.kind)    # tool_call / answer / ask_user

        if decision.kind == "tool_call":
            with span.child("tool", name=decision.tool) as t:
                result = execute(decision)                # 副作用也在同一条轨迹里
                t.set("tool.result_digest", digest(result))
        return decision

三个要点:

  • 属性名不要自造。 gen_ai.* 这套字段来自 OpenTelemetry 的 GenAI 语义约定——OTel 从 2024 年起专门为 LLM 调用和 agent 操作定义了标准 span 属性(模型、token 用量、输入输出内容、工具调用),Datadog、New Relic 等后端原生支持。用标准字段,你的轨迹在任何厂商的工具里都可读。
  • 记录到「能重建现场」为止。 事后调试需要的最小集合:完整 prompt(或其哈希 + 可寻址存储)、模型标识与参数、prompt 版本号、决策内容、工具输入输出摘要。少一样,某类问题就永远查不出来。
  • 全文记录是有代价的开关。 prompt 里常有用户数据,全文入库牵扯隐私与存储成本。常见折中:默认记摘要 + 哈希,采样一小比例记全文,出错的轨迹强制记全文。

二、可恢复:恢复的是任务,不是进程

要解决的根本问题:Agent 任务是长链条(几十步、跨分钟甚至小时),任何一步都可能崩——而重跑不会得到同样的结果,因为内核是概率性的。所以不能靠「失败了从头再来」,必须让任务状态独立于进程存在。

def run_task(task):
    log = EventLog.open(task.id)      # 追加写,唯一事实来源
    state = replay(log)               # 崩溃后:从事件重建状态,而不是重新开始

    while not state.done:
        decision = agent_step(task, state)
        log.append("decision", state.step_n, decision)   # 先记录,再执行

        result = None
        if decision.kind == "tool_call":
            key = idempotency_key(task.id, state.step_n, decision)
            if log.has_effect(key):
                result = log.get_effect(key)   # 崩在「执行后、记录前」?重放结果,不重放副作用
            else:
                result = execute(decision, idempotency_key=key)  # 下游用 key 去重
                log.append("effect", key, result)

        state = apply(state, decision, result)  # 纯函数:事件进,新状态出
        log.checkpoint(state)                   # 快照只是加速,日志才是真相

三个要点:

  • 恢复时重放决策,而不是重新决策。 这是概率内核带来的特殊约束:崩溃前模型已经决定「调用退款工具」,恢复后如果重新问模型,它可能这次决定不退了——任务就精神分裂了。事件日志里已有的决策直接复用,只有日志尽头才允许再次调用内核。
  • 幂等键是副作用的安全带。 崩溃可能发生在「工具执行成功」和「结果写入日志」之间,恢复后无法区分执行过没有。解法是把幂等键传给下游:重复请求带同一个键,下游返回上次的结果而不是再退一次款。
  • 状态转移必须是纯函数。 apply(state, decision, result) 里不允许任何 I/O 和随机性,否则重放会漂移。这就是「状态放内核外面」的字面含义。

这一节展开可以写一整篇——事实上已经写过,讲中断恢复的完整设计空间。

三、可验证:护栏管单次,评测管分布

要解决的根本问题:正确性无法用断言表达——「回答得好」没有 assertEquals。实践里它拆成两层,便宜的确定性检查逐次跑,昂贵的统计度量批量跑,两者不可互相替代:

# 第一层:运行时护栏——每个输出都过,同步、便宜、确定性
def guard(decision, state):
    checks = [
        schema_valid(decision),                  # 结构:能否解析成合法动作
        cites_only_known_sources(decision),      # 幻觉的廉价代理:引用必须在检索结果里
        within_task_scope(decision, state.task), # 没有悄悄扩大任务范围
        no_pii_leak(decision),                   # 输出不含不该出现的敏感字段
    ]
    for c in checks:
        if not c.ok:
            return Reject(reason=c.name)   # 拒绝 → 带原因回内核重试,或降级到人工
    return Accept()

# 第二层:评测门——每次改动都过,异步、昂贵、统计
def eval_gate(candidate, baseline, golden_set, n_runs=5):
    sc = [score(run(candidate, case)) for case in golden_set for _ in range(n_runs)]
    sb = [score(run(baseline,  case)) for case in golden_set for _ in range(n_runs)]

    diff = mean(sc) - mean(sb)
    if diff < 0 and is_significant(sc, sb):      # 区分「回归」与「采样噪声」
        return Block(evidence=worst_cases(candidate, golden_set))
    return Pass(delta=diff)

三个要点:

  • 护栏是断言,评测是统计,别混。 护栏挡的是「这一次输出坏了」,必须确定性、毫秒级,因为它在请求路径上;评测答的是「这个版本整体变差了吗」,必须多次采样(n_runs),因为单次分数是从分布里抽的一个样本。用护栏做版本决策会漏掉分布性退化,用评测做单次拦截则太慢太贵。
  • 打分器自己就是被测对象。 打分若靠 LLM 裁判,裁判会错;靠规则断言,断言会有漏洞——对十个流行 agent 基准的审计发现打分缺陷可导致最高 100% 的相对误估。评测代码要用 checklist 审计,和业务代码同等对待。
  • 这一层就是「eval 是新的 PRD」的落点。 golden set 编码产品意图,评测门守住它——为什么这份「规格」和 TDD 的测试不是一回事、它自己会怎么失效,上一篇整篇都在讲。

四、可治理:按后果分级,不按工具分级

要解决的根本问题:Agent 的权限无法在编码时静态确定——同一个「发邮件」工具,发内部草稿和发给客户是完全不同的风险。治理的单位不是工具,是动作的后果

RISK_TIERS = {
    "read":         Tier.AUTO,         # 查订单、搜文档:随便跑
    "reversible":   Tier.AUTO_BUDGET,  # 存草稿、建内部工单:可撤销,限额内自动
    "irreversible": Tier.APPROVAL,     # 退款、给客户发信:过人工审批
    "forbidden":    Tier.DENY,         # 改权限、删数据:任何情况都不给
}

def gate(action, ctx):
    tier = classify(action)                          # 按「后果 + 参数」分级:
                                                     # refund(¥10) 和 refund(¥10000) 不同级
    audit.append(agent_id=ctx.agent_id, action=action,
                 tier=tier, task=ctx.task_id)        # 先审计,无论放行与否

    match tier:
        case Tier.DENY:
            return Deny()
        case Tier.APPROVAL:
            return await human_approval(action, ctx)  # 阻塞等人,超时即拒绝
        case Tier.AUTO_BUDGET:
            if not ctx.budget.try_consume(cost(action)):
                return Escalate("budget_exhausted")   # 限额耗尽 → 升级而非静默失败
            return Allow()
        case Tier.AUTO:
            return Allow()

三个要点:

  • 预算限的是爆炸半径,不是账单。 单任务的 token、工具调用次数、累计金额限额,防的是 Agent 在无人注意时高速执行一连串「单看都合理」的动作——这正是治理研究点名的 agent 特有风险:没有人在环中,多个连续动作可以在人注意到之前造成后果。Chan 等人的 Visibility into AI Agents(arXiv:2401.13138,FAccT 2024)给出的三件套——agent 标识符、实时监控、活动日志——对应上面代码里的 agent_idgate 本身和 audit
  • 治理和验证是两个独立旋钮。 验证答「这个行为好不好」(分数),治理答「这类行为允许自动化到什么程度」(风险等级)。eval 分数再高,不可逆动作照样要过关卡——Expedia 的实践是按 agent 风险等级配置强制关卡(toll gates),且明确不让 agent 直接订票付款。分数不能兑换权限。
  • 审批要设计成会被用的样子。 审批队列太长,人就会盲批,治理形同虚设。所以 APPROVAL 档要控制流量(该 AUTO 的别上审批),并且给审批人看的是决策上下文(第一节的轨迹),不是让他重查一遍。

五、可进化:让每个事故变成回归测试

要解决的根本问题:系统的行为随模型升级、数据漂移、用户变化而漂,不动代码也会退化。所以「改进」不能靠零星手动调 prompt,得是一条有版本、有门禁、有回滚的流水线:

def evolve():
    # 1. 失败回流:线上轨迹 → 评测用例
    for traj in sample_production_traces():
        verdict = llm_judge(traj)
        if verdict.uncertain:
            verdict = human_label(traj)              # 裁判不确定的交给人
        if verdict.failed:
            golden_set.add(to_eval_case(traj, verdict))  # 事故 → 永久回归测试
            maybe_revise_criteria(verdict)           # 有些失败暴露的是标准缺失,不是行为错误

    # 2. 一切行为输入都有版本
    candidate = registry.new_version(
        prompt=..., model_pin=..., tool_schemas=..., changelog=...)

    # 3. 回归门(上一节的 eval_gate)
    if eval_gate(candidate, registry.current, golden_set) is Block:
        return

    # 4. 金丝雀:离线过了 ≠ 线上分布也过
    registry.route(candidate, traffic=0.05)
    if online_metrics(candidate, window="48h").worse_than(registry.current):
        registry.rollback(candidate)
    else:
        registry.promote(candidate)

三个要点:

  • 这就是 EDDOps 的闭环。 arXiv:2411.13768 把评测定义为「贯穿生命周期的治理函数」:离线评测(第 3 步)+ 在线评测(第 1、4 步)互相喂养。上面的伪代码就是那篇论文流程模型的最小实现。
  • 注意 maybe_revise_criteria——golden set 不只是加用例,还要改标准。 Criteria drift 研究证明评测标准无法先验写全:很多线上失败暴露的不是「模型没达标」,而是「标准里根本没这一条」。只加用例不修标准,进化回路会越转越偏。
  • 版本化的对象是「行为的全部输入」。 Prompt、模型 pin、工具的 schema 和描述文字、系统指令——任何一个变了行为都会变。只给 prompt 记版本,出了回归你归因不到模型静默升级上。金丝雀(第 4 步)是最后一道网,因为内部 golden set 和真实流量永远有分布差——一半企业发布过「过了内部 eval、死在真实客户手里」的 agent,别指望自己是另一半。

拼回去:五个回路,一个循环

把五段伪代码装回同一个骨架,就是这个系统的全貌:

def run_task(task):
    log = EventLog.open(task.id)                 # ② 可恢复:外置状态
    state = replay(log)

    while not state.done:
        decision = agent_step(task, state)       # ① 可观测:语义 trace 包住内核

        if (v := guard(decision, state)) is Reject:
            state = apply(state, retry_with(v.reason)); continue   # ③ 可验证:护栏

        if decision.kind == "tool_call":
            if (g := gate(decision.action, ctx)) is not Allow:
                state = apply(state, blocked(g)); continue         # ④ 可治理:关卡
            result = execute_idempotent(log, state, decision)      # ② 幂等副作用

        state = apply(state, decision, result)
        log.checkpoint(state)

# 循环之外,异步地转:
evolve()   # ⑤ 可进化:轨迹 → 用例 → 版本 → 评测门 → 金丝雀

如果资源有限,每个「可」先做最小的那一件事:

「可」如果只做一件事
可观测每次内核调用记录完整 prompt + 模型版本 + 决策,能重建现场
可恢复决策先写日志再执行;副作用带幂等键
可验证输出过 schema 校验;改 prompt 前跑一遍 golden set 对比基线
可治理把动作分成「可逆自动 / 不可逆审批 / 禁止」三档
可进化每个线上事故变成一条永久评测用例

最后回到那条分水岭。五个「可」看起来是五种能力,拆到代码层面其实只有一个动作:在唯一允许不确定的 llm() 周围,用日志、护栏、关卡和回流把确定性一层层围回来。内核负责聪明,外壳负责可靠——工程师的全部工作,是让内核「不够聪明的那些时刻」不至于成为事故。


参考

本站前文

论文与标准