夜雨聆风学习资料网

ARTICLE · 1080424

AWS 给4000名高管用了一个 AI 助手:它为什么不敢只依赖一个模型?

AWS 给4000名高管用了一个 AI 助手:它为什么不敢只依赖一个模型?
很多大模型 Demo 看起来都很惊艳。

输入一段问题,模型几秒钟后就能生成一份结构完整、语言流畅的回答。

但当 AI 真正进入生产环境,问题会完全不同:

  • 模型偶尔返回错误数字怎么办?
  • 上游服务突然限流怎么办?
  • 流式输出了一半才发现结果不可信怎么办?
  • 主模型不可用,能不能自动切换?
  • 切换模型后,输出格式是否还能保持一致?
  • 如何知道错误发生在哪一步?

AWS 最近公布了一个很有代表性的生产案例。

其内部系统 NarrateAI 服务超过4000名 AWS 高管,用于在业务会议中实时查询数据。官方称,这套系统通过多层质量控制,实现了约99%的数字准确率。

值得注意的是,它没有把可靠性全部寄托在某一个最强模型上,而是同时采用了多模型故障转移、实时流式评测和数据准确性验证等机制。

AWS:NarrateAI 的生产级大模型质量保障架构[1]

这说明了一件事:

🌈

大模型进入生产环境后,决定系统可靠性的往往不只是模型能力,而是模型之外的工程架构。


一、为什么高管使用的 AI 助手更难做?

普通聊天助手偶尔出现一个错误,用户可能重新提问一次。

但会议场景完全不同。

假设一名高管在业务会议中询问:

过去三个季度,亚太地区企业客户的收入增长率是多少?

AI 如果回答:

亚太地区企业客户收入增长了18.7%。

这个数字看起来非常合理。

但如果真实数据是8.7%,模型只是多生成了一个"1",错误就可能直接影响会议判断。

这种错误最危险的地方是:

  • 语言非常流畅
  • 格式非常专业
  • 数字看起来很合理
  • 用户很难第一时间发现问题

所以,生产级 AI 系统不能只追求"回答得像不像",还要验证:

  • 数据是否来自正确的数据源
  • 计算过程是否正确
  • 时间范围是否匹配
  • 单位是否一致
  • 引用是否能够追溯
  • 最终数字是否通过验证

二、AWS 使用了哪五层质量控制?

根据 AWS 的技术说明,NarrateAI 的实时问答层主要组合了五类机制:

动态工作流编排      ↓多模型故障转移      ↓实时流式评测      ↓组合式质量评价      ↓数据准确性验证

这些机制共同解决不同类型的问题。


三、第一层:动态工作流编排

并不是所有问题都需要调用最强模型。

例如下面几个问题,复杂程度完全不同:

"今天的总销售额是多少?"
"为什么本季度利润率下降?"
"结合过去三年的数据,分析不同区域利润率变化的主要原因。"

第一个问题主要是数据查询。

第二个问题需要汇总多个指标。

第三个问题则需要更复杂的关联分析和解释。

因此,系统应该先识别问题类型,再决定后续流程:

用户问题   ↓意图识别   ├── 简单指标查询   ├── 多指标对比   ├── 趋势分析   └── 复杂原因推理

不同任务可以采用不同处理方式:

任务类型
推荐处理方式
字段提取
低成本模型
查询语句生成
编程或推理模型
数据验证
规则引擎或独立模型
原因分析
强推理模型
报告改写
快速通用模型

如果所有请求都交给同一个模型,不仅成本更高,也很难针对不同风险设置验证规则。


四、第二层:多模型故障转移

生产环境中,模型调用失败并不罕见。

常见原因包括:

  • 上游限流
  • 网络超时
  • 服务暂时不可用
  • 上下文超过限制
  • 返回格式错误
  • 内容安全策略拒绝
  • 单个模型所在区域出现异常

如果应用只配置一个模型,一次上游故障就可能导致整个功能不可用。

因此,需要设置主模型和备用模型:

请求进入   ↓调用主模型   ↓成功 ─────────→ 返回结果   ↓ 失败判断错误类型   ↓调用备用模型   ↓重新验证结果

一个简单的 Python 实现如下。

import osfrom openai import OpenAIclient = OpenAI(    api_key=os.environ["GENVIS_API_KEY"],    base_url="https://genvis.xyz/v1")PRIMARY_MODEL = os.environ["PRIMARY_MODEL"]FALLBACK_MODEL = os.environ["FALLBACK_MODEL"]defcall_model(model: str, prompt: str) -> str:    response = client.chat.completions.create(        model=model,        messages=[            {"role": "system","content": ("你是一名企业数据分析助手。""只能基于提供的数据回答,不能猜测不存在的指标。"                )            },            {"role": "user","content": prompt            }        ],        temperature=0    )return response.choices[0].message.contentdefcall_with_fallback(prompt: str) -> dict:try:        result = call_model(PRIMARY_MODEL, prompt)return {"model": PRIMARY_MODEL,"fallback_used": False,"result": result        }except Exception as primary_error:        result = call_model(FALLBACK_MODEL, prompt)return {"model": FALLBACK_MODEL,"fallback_used": True,"primary_error": str(primary_error),"result": result        }

这里没有把模型名称写死,而是通过环境变量配置:

export PRIMARY_MODEL="your-primary-model"export FALLBACK_MODEL="your-fallback-model"

实际模型名称和可用范围,应以当前平台控制台为准。


五、Fallback 不是简单地"换一个模型"

很多系统虽然设置了备用模型,但实际上仍然不够可靠。

例如主模型返回的格式是:

{"answer":"本季度收入增长12.4%","confidence":0.92,"sources":["revenue_q3"]}

备用模型可能返回:

根据相关数据,本季度收入大约增长了12.4%。

人可以看懂,但程序可能直接解析失败。

因此,多模型切换至少要保证:

  • 输入协议一致
  • 输出结构一致
  • 字段名称一致
  • 错误格式可以识别
  • Token 限制满足任务要求
  • 工具调用方式兼容
  • 切换后重新进行结果验证

可以要求模型按照固定 JSON Schema 输出:

response = client.chat.completions.create(    model=model,    messages=messages,    response_format={"type": "json_schema","json_schema": {"name": "business_answer","schema": {"type": "object","properties": {"answer": {"type": "string"                    },"confidence": {"type": "number"                    },"sources": {"type": "array","items": {"type": "string"                        }                    }                },"required": ["answer","confidence","sources"                ],"additionalProperties": False            }        }    })

如果当前模型或接口不支持 JSON Schema,也可以在应用层解析和校验 JSON。


六、第三层:实时流式评测

为了降低等待感,很多 AI 应用采用流式输出:

正在分析……本季度……亚太地区收入……同比增长……

但流式输出带来了新的问题:

🌈

如果答案已经显示给用户,系统才发现数字错误,应该怎么办?

所以,生产系统不能只在最终结果生成后评测,还需要在生成过程中同步检查。

一个简化的架构可以是:

模型流式输出   ├──→ 用户界面缓冲区   └──→ 实时评测模块              ↓       发现异常数字或格式              ↓       停止输出并重新生成

注意,这里的"实时"不一定意味着每生成一个字就调用一次评测模型。

更合理的方式是:

  • 按完整句子检查
  • 按数据段落检查
  • 对数字和单位单独检查
  • 先写入缓冲区,通过后再显示
  • 高风险回答延迟展示,优先完成验证

对于会议、金融、医疗等高风险场景,多等待几百毫秒,往往比快速显示错误答案更合理。


七、第四层:组合式质量评价

只用一个指标评价答案是不够的。

一份回答可能:

  • 语言非常流畅
  • 数据来源正确
  • 但计算过程错误

也可能:

  • 数字完全正确
  • 但没有回答用户真正的问题

因此,需要把质量拆成多个维度:

defcalculate_quality_score(metrics: dict) -> float:return (        metrics["data_accuracy"] * 0.40        + metrics["source_grounding"] * 0.25        + metrics["instruction_following"] * 0.20        + metrics["format_validity"] * 0.15    )

例如:

metrics = {"data_accuracy": 1.0,"source_grounding": 0.9,"instruction_following": 0.8,"format_validity": 1.0}score = calculate_quality_score(metrics)

根据分数设置处理方式:

0.90—1.00  → 直接返回0.75—0.89  → 二次验证0.60—0.74  → 更换模型重新生成低于0.60   → 转人工审核

权重不能照搬,需要根据业务风险调整。

例如财务系统应该提高数据准确性权重,客服系统则可能更关注指令遵循和回答完整度。


八、第五层:数据准确性验证

这是 NarrateAI 场景中最重要的一层。

如果用户问:

本季度收入是多少?

模型应该负责理解问题和组织语言,但最终数字不应该由模型凭记忆生成。

正确流程应该是:

用户问题   ↓模型生成查询计划   ↓调用真实数据源   ↓应用层执行计算   ↓模型解释结果   ↓应用层再次核对数字

假设数据库返回:

{"quarter":"2026-Q3","revenue":128500000,"currency":"USD"}

模型可以负责把它改写成:

2026年第三季度收入为1.285亿美元。

但系统仍然需要检查:

  • 128500000 是否正确转换为 1.285亿
  • 币种是否遗漏
  • 时间范围是否一致
  • 是否把季度数据误写成年度数据

可以在应用层进行确定性验证:

defvalidate_answer(answer: dict, source: dict) -> bool:return (        answer["raw_value"] == source["revenue"]and answer["currency"] == source["currency"]and answer["period"] == source["quarter"]    )

能用程序确认的内容,尽量不要再交给模型猜测。


九、统一 API 网关在生产系统里有什么作用?

多模型故障转移看起来只需要写几个 try...except,但生产环境要处理的问题远不止这些。

网关层通常需要统一管理:

  • API Key
  • 模型路由
  • 请求超时
  • 重试策略
  • Fallback
  • 用户配额
  • Token 统计
  • 调用日志
  • 成本核算
  • 错误分类
  • 上游健康状态

整体架构可以写成:

业务应用   ↓统一 API 网关   ├── 鉴权   ├── 模型路由   ├── 限流   ├── 重试   ├── Fallback   ├── 日志   └── 成本统计        ↓  多个模型服务

这样,业务团队不需要在每个项目里重复实现模型切换。


十、哪些错误应该重试,哪些不应该?

不是所有错误都适合自动重试。

错误类型
建议处理
网络超时
可以重试
上游限流
延迟重试或切换模型
服务暂时不可用
切换备用模型
JSON 格式错误
可重新生成一次
上下文超过限制
缩短上下文后重试
API Key 无效
不要重试
用户余额不足
不要重试
内容策略拒绝
不应通过换模型绕过
业务数据缺失
转人工或提示用户

如果不区分错误类型,系统可能陷入无意义重试:

请求失败→ 重试→ 再次失败→ 切换模型→ 继续失败→ Token 和费用不断增加

因此,每次失败都应该记录原因和处理结果。


十一、生产环境还应该记录哪些数据?

建议至少记录以下字段:

{"request_id":"req_20260926_001","user_id":"user_1024","task_type":"business_metrics","primary_model":"primary-model","final_model":"fallback-model","fallback_used":true,"latency_ms":1860,"input_tokens":2840,"output_tokens":462,"quality_score":0.94,"validation_passed":true}

这些数据可以帮助开发者回答:

  • 哪个模型最容易超时?
  • 哪类任务最常触发 Fallback?
  • 备用模型的结果是否更差?
  • 哪些用户消耗最多 Token?
  • 哪个业务流程成本最高?
  • 模型升级后质量是否下降?

没有这些记录,模型切换就只能依靠感觉。


十二、开发者可以从小规模方案开始

并不是所有项目都要一开始就复制 AWS 的完整架构。

一个小型项目可以先实现四项能力:

第一步:统一模型调用入口

不要在各个业务模块中分别初始化客户端。

第二步:配置主模型和备用模型

模型名称通过配置或环境变量管理。

第三步:校验结构化输出

保证切换模型后,程序仍然能正确解析结果。

第四步:记录调用日志

至少记录模型、耗时、Token、错误和是否发生切换。

等请求量增加后,再逐渐增加:

  • 实时评测
  • 动态模型路由
  • 自动质量评分
  • 成本预算
  • 人工审核
  • 完整链路追踪

结语

AWS 的 NarrateAI 案例说明,生产级 AI 系统的可靠性并不来自某一个"永远不会出错"的模型。

它来自一整套控制机制:

选择合适的模型       +模型失败自动切换       +输出过程实时评测       +结果质量综合打分       +关键数据确定性验证

模型能力仍然重要,但模型越多、调用量越大,工程层的重要性就越明显。

对于开发者来说,真正需要建设的不是一个简单的模型调用接口,而是一套能够回答下面这些问题的系统:

  • 这次请求调用了哪个模型?
  • 为什么选择它?
  • 它失败后切换到了哪里?
  • 最终结果是否经过验证?
  • 这次调用花了多少成本?
  • 出现错误后能否完整追踪?

当 AI 开始服务真实业务时,最强模型只是起点。

模型路由、Fallback、验证、审计和成本控制,才是系统能否稳定运行的关键。


入口见公众号菜单栏,统一模型接口与多模型路由、Fallback、审计方案见 https://genvis.xyz/v1。

引用链接

[1]AWS:NarrateAI 的生产级大模型质量保障架构: https://aws.amazon.com/blogs/machine-learning/narrateai-production-ready-llm-quality-assurance-on-amazon-bedrock/

相关学习资料