本页内容

AI 信任阶梯:让 AI 一步步赢得进入你产品的资格

一架梯子从红色区域斜靠进深蓝色区域,最上面的一级横档以黄色突出,最下面几级用虚线画出

今年每位 CTO 都开过的那场会

管理层有人问:“我们什么时候加 AI?”

你心里清楚,诚实的答案并不简单。你的产品有付费客户,代码库已经八年了。客服队列、计费逻辑和数据库 schema 里,沉淀着多年来的各种决定,没人愿意在周五下午去碰。

你看过那些 Demo。你也见过模型一本正经地编造出一个订单号。

于是讨论往往只剩下两个糟糕的选项:

  • 一步到位。 把 AI 功能直接推给所有用户,紧盯着,然后祈祷。
  • 一直等待。 等技术“成熟”再说,而竞争对手在这段时间里学到了你没学到的东西。

其实还有第三个选项,而且并不新鲜。工程团队几十年来一直用它上线高风险变更:分阶段赢得信任。先让新系统只看不说,再让它只说不做,只有在有证据表明它做得对的地方,才让它独立行动。

四个方块一步步走向一个布满圆点的大圆,从虚线轮廓逐渐变成实心色块,离得最近的那个刚好与圆的边缘重叠

本文把这个思路变成一份具体的 AI 上线计划。我们称之为 AI 信任阶梯。


一张表看懂这架阶梯

AI 信任阶梯的四个级别呈阶梯状上升:影子、建议、审批、限制内执行,AI 的影响范围逐级扩大

级别AI 做什么谁来决定可能出什么问题凭什么晋级
0. 影子在真实流量上运行。输出只保存,从不展示。你现有的流程,保持不变几乎不会影响用户在重要的类别上,它与团队实际决定的一致程度足够高
1. 建议预填草稿或标签。由人接受、修改或忽略。人,每一次人们过度信任它人们很少需要修改它的建议,而且你确认过他们真的在看
2. 审批后执行准备一个具体操作。人批准后才会执行。人,一键决定审批沦为走过场在一个界定清楚的细分场景里,审批几乎从不被推翻
3. 限制内执行自行执行,但只能在代码和数据库强制执行的硬性限制之内。你的规则,由人抽样审计有边界的错误,会被限制和审计拦住可以一直停在这一级;同时还需要写好退回下一级的规则

比起细节,这张表有两点更重要。

第一,每一级只改变一件事:AI 的输出可以在多大程度上影响生产环境。随着级别上升,模型、prompt 和集成方式几乎可以保持不变。改变的是出错的后果。

第二,你不需要让整个产品一起爬完这架阶梯。你一次只爬一个任务,有时甚至一次只爬一个类别。“给新工单打标签”可能一个月就到第 3 级;“发起退款”可能永远停在第 2 级。这不是失败,而是阶梯在发挥作用。

示意图:五个任务停在不同的级别上。工单类别和小额重复扣款退款处于限制内执行,工单优先级处于建议,大额退款和删除账号按设计停在审批

它和我们另一架阶梯的关系。 在你真的需要 AI Agent 吗?中,我们描述过一架架构阶梯:一次模型调用、工作流、有边界的模型决策步骤、Agent。那架阶梯回答的是“谁控制执行路径?”,而这一架回答的是另一个问题:“输出可以在多大程度上改变生产环境?”两者可以同时使用。即使是一次模型调用,也可以处在信任阶梯的任何一级。


为什么分阶段建立信任有效(以及为什么它不只是谨慎)

这些都不是我们的发明。这架阶梯借鉴了三个久经检验的思路。

自动化等级。 几十年来,人因工程研究者一直强调,自动化不是“开”或“关”二选一。在 2000 年一篇被广泛引用的论文中,Parasuraman、Sheridan 和 Wickens 写道:“自动化可以应用在从低到高的一系列连续等级上,即从完全手动到完全自动。” 在他们列出的选择等级的标准中,有这样两条:“自动化的可靠性,以及决策/行动后果的代价。” 这两条标准正是这架阶梯的核心:它有多常做对,以及做错时代价有多大。

暗发布与金丝雀发布。 软件团队早就知道如何在真实流量上测试高风险变更。Martin Fowler 把暗发布(dark launch)描述为这样一种功能:“它会完成正式发布后要做的全部工作,但没人能看到它在做这些事。” 云端机器学习平台已经把同样的思路做成了产品:在 AWS 的影子测试(shadow test)中,“只有生产变体的响应会返回给调用方应用。” Google SRE 团队把金丝雀发布(canary)定义为:“对服务中某项变更进行局部、限时的部署,并对其进行评估。” 第 0 级就是暗发布;第 1 到第 3 级则是针对决策的金丝雀发布。

面向 AI 的指南也是同样的结论。 OpenAI 的构建 Agent 实用指南说得很直接:“敏感、不可逆或高风险的操作,应当触发人工监督,直到对 Agent 可靠性的信心逐步建立起来。” 谈到构建 Agent 的整体方法时,同一份指南还提到:“客户采用渐进式方法时,通常能取得更大的成功。” OWASP Top 10 for LLM Applications 建议:“采用人在回路(human-in-the-loop)控制,要求高影响操作在执行前必须经过人工批准。”

这架阶梯之所以有效,还有一个原因,而且正是它能帮你赢下预算讨论:每一级都会产生数据。第 0 级告诉你 AI 和团队的判断有多常一致;第 1 级告诉你人们有多常修改它的结果;第 2 级告诉你审批人有多常说“不”。你不必再争论 AI “是否准备好了”,而是直接看来自你自己生产流量的数字。


贯穿全文的例子

为了讲得具体,我们会带着同一个功能走完每一级。

假设有一家 B2B 软件公司,客服收件箱非常繁忙。客服会阅读每张工单,选择一个类别(账单、缺陷、使用咨询、账号访问、取消订阅),设置优先级,有时还会发起退款。公司希望 AI 分担一部分工作。

现有系统很普通:一个 Web 应用、一个关系型数据库、一个后台任务队列,以及一张用了很多年的 tickets 表。它们没有一样是为 AI 设计的。而这正是重点。

示例中的 AI 部分使用 ModelRiver:工作流保存模型、指令和输出 schema,你的应用通过 HTTP 调用它。但这架阶梯本身是一种上线纪律,而不是某个产品功能。我们会讲清楚哪些部分由 ModelRiver 处理,哪些部分留在你的代码里。


第 0 级之前:先打通链路,不花一个 token

在 AI 接触任何一张真实工单之前,你需要先把管道接通:请求、webhook、数据库写入,以及各种错误路径。你不会想在调试 JSON 解析器的时候还要为 token 付费。

ModelRiver 的测试模式(Test Mode)正是为这一步设计的。工作流处于测试模式时,会直接返回结构化输出中的示例数据,而不调用 provider。没有 provider 调用,没有 token 费用,测试流量也不计入计费请求数。对于异步请求,webhook 和后端回调照常触发,所以整套集成可以端到端跑通。测试请求同样会被记录,用请求日志中的 Test 来源筛选器,就能准确看到应用发出了什么。

先定义输出契约。对于工单分类,它可能是这样的:

JSON
{
"type": "object",
"properties": {
"category": {
"type": "string",
"enum": ["billing", "bug", "how_to", "account_access", "cancellation", "other"]
},
"priority": { "type": "string", "enum": ["low", "normal", "high", "urgent"] },
"reason": { "type": "string", "maxLength": 280 }
},
"required": ["category", "priority", "reason"],
"additionalProperties": false
}

注意这些 enum 值。它们应当与数据库里已有的值一致,而不是专门为 AI 新造一套。如果 tickets.category 列允许六个值,schema 里就应该是同样的六个。之后 AI 的输出开始写入真实数据表时,这一个决定就能帮你避开一整类 bug。

ModelRiver 会在返回每个响应之前,按这个 schema 进行验证。如果模型返回的内容不匹配,这次尝试就算失败;如果你配置了备用模型,请求会继续交给备用模型。它不会悄悄把格式错误的数据交给你。(为什么这很重要,可以参阅我们把同一个 JSON Schema 发给 5 个 LLM Provider,看看哪里会坏。)

当链路在测试模式下跑通后,把工作流切换到真实模型。现在你可以接入真实流量了。


第 0 级:影子模式

它是什么

在第 0 级,每张新工单仍然完全按原来的流程处理。由人来分类。客户和客服都感觉不到任何变化。

但在后台,同一张工单也会被发给 AI。AI 的答案被保存到一张表里,没有人看到。之后,你再把 AI 的答案和团队的实际处理结果做对比。

这是让 AI 接触真实流量最安全的方式。如果 AI 很慢、答错或者宕机,没有人会察觉,因为没有人在等它。

请求路径保持不变;在后台,任务发起 AI 请求,ModelRiver 返回 202 和 channel_id,webhook 按 channel_id 保存结果

为什么必须是真实流量

你可能会问:为什么不直接拿一批历史工单来测试?

这也应该做。但历史工单和实时流量并不一样。Google 的 ML Test Score 论文说得很直接:“无论离线测试多么全面,它本身都无法保证模型在真实生产环境中表现良好。” NIST 的生成式 AI 概况(Generative AI Profile)也指出了同一点:“实验室环境与真实环境之间的不匹配,可能导致测量上的缺口。”

真实工单里有错别字、用文字描述的截图、一条消息里塞了三个问题、愤怒的客户,还有新的产品名称。影子模式可以免费、无风险地在所有这些情况下进行测试。

影子模式的唯一规则

影子调用绝不能影响它所跟随的那个请求。

这意味着:

  • 不要在用户请求内部调用 AI 并等待结果。
  • 不要让 AI 的错误以异常的形式抛进现有代码。
  • 不要让 AI 超时导致数据库事务一直挂着。

在老代码库里遵守这条规则,最简单的方法是把工作交给你已有的后台任务系统。下面是 Node.js 应用中的写法;同样的结构也适用于 Rails、Laravel、Django、Spring 或 .NET。

JS
// 你现有的工单处理函数。唯一新增的一行是入队。
async function createTicket(req, res) {
const ticket = await db.tickets.insert(req.body); // 现有逻辑,保持不变
await jobs.enqueue('ai_shadow_triage', { ticketId: ticket.id }); // 新增,发出即走,不等待结果
res.status(201).json(ticket);
}
JS
// 后台任务。它只负责*发起* AI 请求,然后返回。
export async function aiShadowTriage({ ticketId }) {
const ticket = await db.tickets.find(ticketId);
 
try {
const res = await fetch('https://api.modelriver.com/v1/ai/async', {
method: 'POST',
headers: {
Authorization: `Bearer ${process.env.MODELRIVER_SHADOW_API_KEY}`,
'Content-Type': 'application/json',
},
body: JSON.stringify({
workflow: 'ticket_triage',
messages: [{ role: 'user', content: ticket.body }],
}),
signal: AbortSignal.timeout(5000),
});
 
if (res.status !== 202) {
logger.warn('shadow triage not accepted', { ticketId, status: res.status });
return;
}
 
// 记下哪个请求对应哪张工单。
const { channel_id } = await res.json();
try {
await db.aiShadowRequests.insert({ channel_id, ticket_id: ticketId });
} catch (err) {
logger.warn('shadow triage started but not recorded', { ticketId, channel_id, error: err.message });
}
} catch (err) {
// 影子调用失败是数据,不是事故。
logger.warn('shadow triage failed to start', { ticketId, error: err.message });
}
}

/v1/ai/async 端点会立即返回 202 Accepted 和一个 channel_id。模型在后台运行,完成后,ModelRiver 会把结果发送到你的 webhook。

channel_id 用来把答案和工单对应起来。这个请求的每一次 webhook 投递都带着同一个 channel_id,包括错误投递和重试,所以只要把它和工单 ID 存在一起,就足以完成关联。

接收答案

你的 webhook 处理函数验证签名、保存结果,除此之外什么也不做。

JS
import crypto from 'node:crypto';
 
// 签名覆盖时间戳和请求体中 "data" 字段的原始 JSON 文本,
// 因此要在解析 JSON 之前,基于原始请求体进行校验。
function verifyModelRiverSignature(rawBody, timestamp, signature) {
const start = rawBody.indexOf('"data":');
const end = rawBody.lastIndexOf(',"timestamp":');
if (start === -1 || end === -1 || !timestamp || !signature) return false;
 
const expected = crypto
.createHmac('sha256', process.env.MODELRIVER_WEBHOOK_SECRET)
.update(`${timestamp}.${rawBody.slice(start + '"data":'.length, end)}`)
.digest('hex');
 
return (
signature.length === expected.length &&
crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))
);
}
 
app.post('/webhooks/modelriver', express.raw({ type: 'application/json' }), async (req, res) => {
const rawBody = req.body.toString('utf8');
const verified = verifyModelRiverSignature(
rawBody,
req.get('x-modelriver-timestamp'),
req.get('x-modelriver-signature')
);
if (!verified) return res.sendStatus(401);
 
const body = JSON.parse(rawBody);
const event = body.data;
 
// 以 channel_id 为键;通过 ai_shadow_requests 关联到工单。
await db.aiShadowResults.upsert({
channel_id: body.channel_id, // 唯一键:重试投递只会覆盖,不会产生重复
status: event.status, // "success" 或 "error"
ai_category: event.data?.category ?? null,
ai_priority: event.data?.priority ?? null,
ai_reason: event.data?.reason ?? null,
used_model: event.meta?.used_model ?? null,
duration_ms: event.meta?.duration_ms ?? null,
});
 
res.sendStatus(200);
});

有三个细节让它更稳健:

  • 基于原始请求体校验。 签名覆盖的是 ModelRiver 发送的原始文本。如果先解析 JSON 再重新编码,细微的差异(例如 4.0 变成 4)就会让校验失败。签名验证指南中提供了同一函数的 Python、PHP 和 Ruby 版本。
  • 按 channel_id 做 upsert。 投递失败时,ModelRiver 会重试。用 channel ID 作为唯一键,重试只会覆盖原有的行,而不会产生重复记录。
  • 错误也要保存。 即使工作流里的所有模型都失败了,或者请求被支出上限拦截,webhook 依然会送达,带着 status: "error" 和同一个 channel_id。这是有用的数据。如果某个模型在 4% 的真实工单上超时,你会希望在进入第 1 级之前就知道。

把影子流量放进单独的项目

这是一个小决定,回报却很大。在 ModelRiver 中,webhook、支出上限、输入护栏设置和请求日志都以项目为范围。一个项目里的每个异步请求,都会投递到该项目中所有已启用的 webhook。

所以,请为影子流量单独创建一个项目,配上独立的 API key、独立的 webhook 和独立的支出上限。这样你会得到:

  • 成本上限。 为项目设置支出上限,比如每日上限。影子项目达到上限后,ModelRiver 会停止调用模型,直到下一个周期。异步请求仍会以 202 被接受,每个请求都会以一条 error.type: "project_budget_exceeded" 的错误 webhook 返回。在影子模式下,这正是你想要的:实验暂停,生产环境丝毫不受影响。上限是根据近期支出来检查的,所以一波并发请求可能会略微超出。
  • 干净的日志。 影子请求不会混进生产环境的仪表盘。
  • 不会有意外投递。 生产环境的 webhook 处理函数永远不会收到影子事件。

有一点是单独的项目隔离不了的:影子请求仍然会计入你所在组织的套餐请求配额。规划影子流量的规模时要考虑到这一点。

想清楚你发送的是哪些数据

影子模式会把真实的客户消息发送给模型 provider。像对待任何新的数据处理方一样对待它:检查你的协议、隐私政策,以及你对客户做过的承诺。

有两个地方可以帮上忙。messages 里放什么文本由你决定,所以去掉任务用不到的内容。项目设置 “Save full request and response text”(保存完整的请求与响应文本)决定请求日志是否保存完整的 prompt 和响应。你可以在影子项目中关闭它,这样请求日志就不再保存完整的 prompt 和响应。有两个限制需要知道:webhook 投递日志仍会记录它发送的内容;只要项目中有工作流使用会话记忆(session memory),这个设置就无法关闭。

诚实地衡量一致性

一两周之后,你会得到任何 Demo 都给不了的东西:针对同一批真实工单,AI 的答案和团队的答案。一个简单的查询就能看出两者在哪里一致。

SQL
SELECT t.category AS human_category,
s.ai_category,
count(*) AS tickets
FROM tickets t
JOIN ai_shadow_requests r ON r.ticket_id = t.id
JOIN ai_shadow_results s ON s.channel_id = r.channel_id
WHERE s.status = 'success'
AND t.triaged_at > now() - interval '14 days'
GROUP BY 1, 2
ORDER BY 1, 2;

这就是一张混淆矩阵。读它的时候要注意:

  • 按类别看,而不只看整体。 “88% 一致”可能掩盖了“取消订阅类只有 40% 一致”。而取消订阅恰恰可能是出错代价最高的地方。
  • 小心稀有类别。 如果 90% 的工单都是使用咨询,一个永远回答 "how_to" 的模型看起来也有 90% 的准确率。统计学家正是因此才使用 Cohen's kappa 这类指标:单纯的一致百分比没有考虑偶然一致的情况。Hamel Husain 在他的 LLM 评审指南中给出了实用建议:分别报告每个类别被正确识别的比例。
  • 你的团队并不总是对的。 当 AI 和人的判断不一致时,去读那张工单。有时是 AI 错了;有时是处理的人太匆忙,或者你的类别本身有重叠。分歧往往正是你发现流程本身不清晰的地方。
  • 读具体的例子。 数字告诉你该看哪里,读 50 个分歧案例才能告诉你原因。正如 Hamel Husain 在 Your AI Product Needs Evals 中所说:“如果你没有在看大量数据,那你的做法就是错的。”

示意数字:88% 的整体一致率掩盖了各类别的差异,从 how_to 的 96% 一直到 cancellation 的 40%

顺便从同一张表里查看错误率和响应时间。它们会告诉你,这个工作流将来是否有可能放进面向用户的路径。

不动生产代码也能改进

你会想修改 prompt、增加示例,或者换一个模型试试。工作流有 Draft、Live 和 History。你编辑 Draft,在 Playground 里用 Draft、Live 或更早的修订版本进行试验,然后发布。API 流量始终运行 Live 修订版本,所以编辑 Draft 永远不会改变生产环境。如果新版本效果更差,就回滚到之前的修订版本。

每条请求日志都会记录实际运行的工作流修订版本号,所以你随时能看到某个答案出自哪个修订版本。要在自己的查询里对比修订版本,就把一个修订版本标签和每条结果存在一起。这个标签应该从运行时配置或 flag 值中读取,并在每次发布或回滚时更新,而不是从只有部署时才会变化的环境变量中读取。发布前后仍在处理中的请求,标签可能不准;需要某个请求的确切修订版本时,以它的请求日志为准。这样,“新 prompt 有没有帮助?”就从一场争论变成了一条 GROUP BY revision 查询。

什么时候离开第 0 级

没有放之四海皆准的数字。门槛由你来定,而且要在看结果之前,根据出错的代价来定。一条合理的晋级规则大致如下:

  • 你关心的每个类别都有足够多的真实案例,包括稀有类别。
  • 每个类别的一致率都高于你事先设定的阈值。
  • 错误率和超时率足够低,缺少建议的情况会很少见。
  • 你读过那些分歧案例,并且理解它们。

在开始之前把它写下来。看到数字之后才定阈值的团队,往往会刚好选中他们刚得到的那个数字。


第 1 级:建议

它是什么

在第 1 级,AI 的答案变得可见,但只作为建议。客服打开工单时,类别和优先级字段已经预先填好。客服可以一键接受、修改,或者直接忽略。

每个决定仍然由人来做。改变的是他们要打多少字,以及你能学到什么。

第 1 级最重要的设计原则

AI 失败时,旧体验原样回来。

如果 AI 很慢、宕机,或者没有返回有用的内容,客服看到的就是他们用了多年的空白表单。缺少建议绝不应该变成一个错误页面。

你可以在工单到达时生成建议(沿用第 0 级的异步流程,写入一张 suggestions 表),也可以在客服打开工单时生成(使用同步调用)。下面是同步版本:

JS
async function getTriageSuggestion(ticket) {
try {
const res = await fetch('https://api.modelriver.com/v1/ai', {
method: 'POST',
headers: {
Authorization: `Bearer ${process.env.MODELRIVER_API_KEY}`,
'Content-Type': 'application/json',
},
body: JSON.stringify({
workflow: 'ticket_triage',
messages: [{ role: 'user', content: ticket.body }],
}),
signal: AbortSignal.timeout(8000),
});
 
const body = await res.json();
if (body.status !== 'success') return null; // 所有模型都失败了:显示普通表单
 
return {
...body.data, // { category, priority, reason }
revision: await config.get('ticket_triage_revision'), // 发布和回滚时更新
model: body.meta?.used_model,
};
} catch {
return null; // 超时或网络错误:显示普通表单
}
}

注意对 body.status 的检查。当工作流中的所有模型(包括备用模型)都失败时,ModelRiver 仍然会返回一个 JSON 信封,HTTP 状态码往往是 200,但其中是 status: "error",并在 meta.attempts 中列出每次尝试的情况。要检查这个字段,而不只是看 HTTP 状态码。

在工作流上至少配置一个备用模型,最好来自不同的 provider。如果主模型出现故障、被限流,或者返回的输出没有通过 schema 验证,ModelRiver 会先尝试备用模型,再宣告失败。响应会在 meta.used_provider 和 meta.used_model 中告诉你实际是哪个模型给出了答案,每次失败的尝试也都会被记录下来。

记录人们如何处理建议

这正是让你晋级到第 2 级的数据,它应该放在你的数据库里,紧挨着工单。ModelRiver 没有内置的反馈或评分存储,你也不会希望它放在那里:处理结果是关于你的工单的业务事实,应该和你的其他业务数据放在一起。

SQL
CREATE TABLE ai_suggestion_outcomes (
ticket_id bigint NOT NULL REFERENCES tickets(id),
field text NOT NULL, -- 'category' 或 'priority'
suggested_value text NOT NULL,
final_value text NOT NULL,
accepted_unchanged boolean NOT NULL,
workflow_revision text,
decided_by bigint NOT NULL, -- 处理工单的客服
decided_at timestamptz NOT NULL DEFAULT now()
);

仅凭这一张表,你就能按类别、按客服、按工作流修订版本计算修改率。

第 1 级的隐藏风险:人们不再检查

这里有一个被充分研究过的陷阱,它有一个名字:自动化偏差(automation bias)。JAMIA 上的一篇系统综述把它定义为:“过度依赖自动化的倾向”。Parasuraman 和 Manzey 的 2010 年综述发现,这种偏差的特点是:“在新手和专家参与者中都会出现,无法通过培训或指示来预防”。NIST 的生成式 AI 概况同样点出了这一问题:“对自动化系统的过度顺从”。

说白了:如果建议通常是对的,人们就会开始不看内容直接点“接受”。这时你的修改率看起来漂亮极了,但它在骗你。

几种实用的防御方法:

  • 展示理由,而不只是答案。 schema 里的 reason 字段就是为此而设。“账单:客户提到发票 #4471 被重复扣款”比一个单独的“账单”更容易核对。
  • 不要预先选中高风险字段。 对于优先级这种低风险字段,可以预填;对于“按取消订阅关闭”这种高风险字段,把建议显示在字段旁边,让人自己选择。
  • 警惕可疑的速度。 如果某位客服一整天都在两秒内接受建议,他就没有在审核。
  • 已接受的建议也要审计。 每周请人独立地对一小部分随机抽取的、建议被原样接受的工单重新分类。这是发现“走过场”的唯一方法。

什么时候离开第 1 级

你要找的是:在某个特定的细分场景中,修改率很低,并且经过审计确认。这个范围往往比你希望的更窄。“提到发票号的账单工单”可能远比“所有工单”更早准备好。这没问题。晋级的是细分场景,而不是整个功能。


第 2 级:审批后执行

它是什么

在第 2 级,AI 不只是填一个字段。它会准备一个包含全部细节的具体操作,由人一键批准或拒绝。

在我们的例子中:客户要求退款。AI 读完工单后提议:“为订单 1234 退款 $42.00,原因:重复扣款。”客服主管看到这个提案和相应证据,点击批准。只有到这时,你的后端才会发起退款。

这正是 OWASP 为高影响操作推荐的模式:在操作执行之前由人批准。OpenAI 的指南也把“授权大额退款”列为应当触发人工监督的操作。模型提议,人来批准,系统执行。

把 AI 的提案当作用户输入

OWASP 关于不当输出处理(Improper Output Handling)的指南给出了正确的思维模型:“像对待任何其他用户一样对待模型,采用零信任方法,并对模型传给后端函数的响应进行恰当的输入验证。”

所以,一个提案在被人看到之前,要先经过三道检查。

1. Schema。 工作流的输出契约限制了 AI 能提出什么:

JSON
{
"type": "object",
"properties": {
"action": { "type": "string", "enum": ["refund", "escalate", "no_action"] },
"order_id": { "type": "string", "maxLength": 40 },
"amount_cents": { "type": "integer", "minimum": 1, "maximum": 50000 },
"reason_code": { "type": "string", "enum": ["duplicate_charge", "service_outage", "cancellation", "other"] },
"evidence": { "type": "string", "maxLength": 500 }
},
"required": ["action", "reason_code", "evidence"],
"additionalProperties": false
}

ModelRiver 会强制校验 enum、minimum 和 maximum、maxLength、required 以及 additionalProperties。一个 $900 的提案无法通过验证,永远不会作为有效提案到达你的代码。有一个限制需要坦白说明:ModelRiver 不会强制校验 format 关键字,例如 "format": "date" 或 "format": "email"。这些请在你自己的代码里验证。

2. 你的业务规则。 Schema 无法知道订单 1234 是否存在、是否属于这位客户、是否已经退过款。只有你的后端知道。

JS
async function validateRefundProposal(proposal, ticket) {
if (proposal.action !== 'refund') return { ok: true };
 
const order = await db.orders.find(proposal.order_id);
if (!order) return { ok: false, why: 'order_not_found' };
if (order.customer_id !== ticket.customer_id) return { ok: false, why: 'order_belongs_to_other_customer' };
 
const refundable = order.amount_paid_cents - order.amount_refunded_cents;
if (proposal.amount_cents > refundable) return { ok: false, why: 'exceeds_refundable_amount' };
 
return { ok: true, order };
}

3. 人。 有效的提案进入你应用里的审批队列。无效的提案也进入同一个队列,但会标注“AI 提案未通过检查:订单属于其他客户”。这对审核人有用,对你也有用。

各部分如何衔接

这个流程复用了第 0 级和第 1 级的全部内容:

TEXT
工单到达
│
▼
你的任务队列 ──► POST /v1/ai/async (workflow: refund_proposal)
│
▼ 模型在后台运行
Webhook 发送到你的后端
│
▼
ModelRiver 已检查 schema
你的代码检查业务规则
│
▼
审批队列(你的应用)
│
人工批准
│
▼
你的后端发起退款(幂等)

ModelRiver 的后端管道还可以在 AI 与最终结果之间运行自动化步骤:AI 起草,你的后端核实并补充信息,结果再流向下一步或用户。我们在为什么 AI 需要不止一次调用你的后端中完整演示过类似的退款流程。

对审批来说,有一个限制很关键:每个管道步骤最多等待你的后端回调 10 分钟。这对自动化检查绰绰有余,但对一个正在吃午饭的人来说远远不够。所以,人工审批不要让管道步骤一直挂着。保存提案,完成流程中自动化的部分,然后让审批在你自己的队列里待多久都行。

让执行具备幂等性

投递失败时 webhook 会重试,人也会双击按钮。你的退款代码必须能安全地运行两次。

最简单的做法是在数据库层面,把提案 ID 作为幂等键:

SQL
CREATE TABLE refunds (
id bigserial PRIMARY KEY,
order_id bigint NOT NULL REFERENCES orders(id),
amount_cents integer NOT NULL CHECK (amount_cents > 0),
ai_proposal_id bigint UNIQUE, -- 每个提案永远只对应一笔退款
approved_by bigint, -- 第 2 级:批准的人
executed_by_rule text, -- 第 3 级:具名的自动规则
created_at timestamptz NOT NULL DEFAULT now(),
CHECK (approved_by IS NOT NULL OR executed_by_rule IS NOT NULL)
);

注意最后那个 CHECK。每一笔退款都必须有一个人工批准者,或者一条具名的自动规则。数据库本身就会拒绝一个无人负责的操作。

第 2 级要衡量什么

  • 推翻率: 审批人拒绝或修改有效提案的频率。按操作类型、原因代码和金额区间分别跟踪。
  • 检查失败率: 提案未通过业务规则的频率。这个比率上升,往往说明有什么变了:新产品、新的退款政策,或者模型更新。
  • 审批耗时: 如果每次审批只花两秒,你就又回到了第 1 级的自动化偏差问题。要抽样审计。

什么时候离开第 2 级

很多操作永远不应该离开第 2 级。大额退款、删除账号、任何涉及法律或医疗的事项:让人永久留在回路中。这是好结果,而不是项目停滞。AI 仍然完成了大部分工作,最终决定仍然由人来做。

但对于一个范围很窄、理解充分的细分场景,比如针对重复扣款的小额退款,你可能会发现审批人几乎从不说“不”。这个细分场景就是第 3 级的候选。


第 3 级:限制内执行

它是什么

在第 3 级,AI 的提案无需人事先批准就会执行,但只能在你的系统强制执行的限制之内,并由人事后抽样审计。

关键词是限制。第 3 级不是“信任 AI”,而是“信任一组硬性边界,并让 AI 在边界内运作”。

分层构建限制

每一层拦截不同的错误,没有任何一层依赖模型表现良好。

AI 的操作位于层层嵌套的限制之中:输出 schema、业务规则、数据库约束、策略闸门、数量与支出上限,以及审计

层级示例拦截什么在哪里实现
输出 schemaamount_cents 最大值、reason_code 枚举无效或超出范围的提案ModelRiver 结构化输出
业务规则订单存在、属于该客户、金额 ≤ 可退金额与实际情况不符的提案你的代码
数据库约束UNIQUE (ai_proposal_id)、CHECK (amount_cents > 0)重复和不可能的值,即使代码有 bug 也能拦住你的数据库
策略闸门仅限 duplicate_charge、仅限 $50 以下、仅在 flag 开启时超出你在第 2 级验证过的细分场景的操作你的代码
数量与支出上限每小时自动退款次数上限;项目支出上限失控循环和成本激增你的代码和 ModelRiver
审计由人审查随机抽取的自动操作样本隐蔽的系统性错误你的流程

策略闸门就是阶梯变成代码的地方:

JS
const AUTO_REFUND_REASONS = new Set(['duplicate_charge']);
const AUTO_REFUND_MAX_CENTS = 5000;
 
async function decideRefundPath(proposal, check) {
const autoAllowed =
(await flags.isEnabled('ai_auto_refunds')) && // 紧急开关(kill switch)
check.ok &&
proposal.action === 'refund' &&
AUTO_REFUND_REASONS.has(proposal.reason_code) &&
proposal.amount_cents <= AUTO_REFUND_MAX_CENTS &&
(await autoRefundsInLastHour()) < 50; // 数量上限
 
return autoAllowed ? 'execute_with_rule' : 'send_to_approval_queue';
}

所有没通过闸门的操作都会退回第 2 级。所以闸门过严的代价,只是由人来批准一下,而这本来就是他们昨天在做的事。

支出上限就是熔断器

在 ModelRiver 中,你可以设置项目支出上限(按小时、天、周或月),也可以为工作流中的每个模型槽位单独设置上限。项目达到上限后,ModelRiver 会停止调用模型:/v1/ai 返回 402,/v1/ai/async 则投递一条 error.type: "project_budget_exceeded" 的错误 webhook。某个模型槽位达到上限时,会跳过该槽位,尝试下一个备用模型。

在第 3 级,这一点的意义不止于成本。如果某个 bug 让你的应用把同一张工单发给 AI 一千次,即使没人盯着,支出上限也会让循环停下来。同时还要配合你自己对操作的数量上限,因为 AI 调用便宜,并不代表退款也便宜。

紧急开关

NIST 的 AI 风险管理框架要求具备:“机制……用于接管、脱离或停用那些表现或结果与预期用途不一致的 AI 系统”。Pete Hodgson 关于 feature toggle(功能开关)的文章给同一件事起了个更亲切的名字:“长期存在的‘紧急开关’(Kill Switches)”。

在第 3 级,你需要两个这样的开关:

  1. 一个 feature flag(上面的 ai_auto_refunds),可以立即把所有操作送回审批队列。你的 feature flag 系统本来就能做到这一点,直接用它。
  2. 一次工作流回滚。 如果问题是某次 prompt 或模型变更引起的,就把工作流回滚到上一个 Live 修订版本。生产环境无需部署代码就会切回之前的配置。

在真正需要之前,先把两者都演练一遍。

自动退回下一级

晋级规则只是这套系统的一半。你还需要降级规则,并且要在上线前写下来:

  • 如果自动退款的审计错误率超过阈值,关闭 flag。
  • 如果业务规则失败率突然上升,关闭 flag 并调查原因。
  • 如果你更换了模型或发布了新的 prompt 修订版本,考虑先退回一级,直到新版本重新赢得信任。

最后一条常常让人意外。但新模型就是一个新的决策者。正如我们在生产环境里的 LLM Failover中展示的,同一个 prompt 换一个模型,表现可能就像另一个产品。

四个级别从影子到限制内执行依次叠放,一侧写着每次晋级的条件,另一侧写着每次降级的触发条件


哪些由 ModelRiver 处理,哪些留在你的代码里

我们想把这一点说准确,因为如果在这里夸大其词,就和本文的主旨背道而驰了。

需求ModelRiver你的应用
通过 HTTP 调用模型而不阻塞应用异步请求、webhook、WebSocket 投递任务入队,处理 webhook
保证输出结构结构化输出,返回前完成验证;schema 验证失败会触发故障转移验证 format 字段以及任何需要你自己数据的内容
扛住 provider 故障主模型加最多两个备用模型,每次尝试都有记录全部失败时展示旧体验
安全地修改 prompt 和模型Draft、Live、History、发布、回滚;每条请求日志都记录修订版本号把修订版本和每条结果一起保存
不花 token 就把链路接通测试模式,提供示例数据、webhook 和回调集成代码只写一次
控制成本项目级和单模型支出上限对操作设置数量上限
筛查有害输入输入护栏,覆盖暴力、自残等类别:拦截、仅记录或关闭(未成年人保护始终开启)防御 prompt 注入,例如使用范围很窄的 schema、不提供不必要的工具
关联结果202 响应和每次 webhook 投递都带有同一个 channel_id把 channel_id 和你的记录存在一起
对比 AI 与人工决策不内置你的影子结果表和查询
审批队列不内置你的应用、你的界面、你的审计记录
反馈与修改跟踪不内置你的处理结果表
按比例放量与 A/B 测试不内置你的 feature flag 系统

表格的下半部分是有意为之。审批、反馈和放量关乎的是你的业务决策,它们应该放在你的数据旁边,放在团队已经信任的系统里。网关的职责是让 AI 这部分变得无聊而可靠,好让你把全部注意力放在这些决策上。


一份可以直接带给团队的上线计划

下面是整架阶梯的检查清单。

第 0 周:先定规则,再看数据

  • 选一个任务,而不是整个功能。是“工单分类”,而不是“AI 客服”。
  • 按类别写下答错的代价。
  • 为每一级写好晋级和降级规则,并给出阈值。
  • 确定谁对这些数字负责,以及谁有权拉下紧急开关。

第 0 级之前

  • 定义输出 schema,使用与数据库相同的 enum 值。
  • 为影子流量单独创建一个项目,配备独立的 key、webhook 和支出上限。
  • 在测试模式下完成集成,并检查错误路径。

第 0 级:影子

  • 从后台任务发起 AI 请求,绝不放在用户的请求路径上。
  • 把每个请求的 channel_id 和工单存在一起,并保存每一条结果(包括错误),以 channel_id 为键。
  • 按类别与人工决策做对比,并阅读分歧案例。

第 1 级:建议

  • 展示建议时附上理由;失败时展示旧体验。
  • 记录人们接受、修改或忽略了什么,并附上工作流修订版本。
  • 抽样审计已接受的建议,检查是否在走过场。

第 2 级:审批

  • 在人看到提案之前,先用 schema、再用业务规则验证。
  • 把人工审批放在你自己的队列里,而不是挂在未结束的管道步骤上。
  • 在数据库层面保证执行的幂等性。

第 3 级:限制

  • 按类型、金额、数量和 feature flag 为自动操作设置闸门。
  • 把支出上限设为熔断器。
  • 随机抽样审计自动操作,结果不佳就降级。

这架阶梯的意义

这架阶梯的目标不是放慢 AI,而是把一个令人害怕、非全即无的决定,拆成一系列小决定,每一个都有来自你自己生产流量的证据支撑。

四个方格排成上升的一行,每个方格中被黄色填满的部分都比前一个更大

你从安全的地方起步,让 AI 在团队身边安静地观察。在它赢得信任的地方,让它帮忙。只有在数字表明它配得上的地方,才赋予它真正的责任,而且要放在不依赖它表现良好的限制之内。

每一步,只要看起来不对,就退回一级。什么都不会坏,没有人会被半夜叫醒。你原来的流程依然在那里,因为你从未移除它。

这就是在不拿公司做赌注的前提下,为一个用户早已依赖的产品加入 AI 的方法。


下一步

延伸阅读