LLM 应用从演示走向生产，难点往往不再是“能否生成答案”，而是能否在模型超时、账号隔离、流式输出和数字密集型任务中持续给出可信结果。NarrateAI 展示了一套运行在 Amazon Bedrock 上的质量保障方案，结合自适应编排、跨账号多模型故障转移、实时流式评估、复合评分和数据准确性验证，在其场景中实现了约 99% 的数值准确率，同时保留实时响应能力。
这里的关键并不是某一个提示词，而是把质量控制做成一条可观测、可降级、可阻断的工程流水线。
把一次模型调用拆成可编排的流水线
生产请求不应全部走同一条路径。一个开放式文案请求和一个包含金额、比例、日期的财务问答，风险完全不同。自适应编排可以先识别任务类型，再决定需要哪些步骤：
- 低风险生成：直接调用主模型，执行基础安全检查。
- 数据密集型问答：先检索或读取权威数据，再要求模型基于证据回答。
- 高风险计算：生成答案后，由确定性代码重新计算并核对数字。
- 复杂推理：增加独立评估模型，检查完整性、证据覆盖和指令遵循情况。
可以用一份简单配置描述这些策略，而不是把判断逻辑散落在业务代码里：
pipelines:
default:
evaluators:
- safety
- instruction_following
threshold: 0.75
numeric:
evaluators:
- safety
- required_evidence
- numeric_accuracy
threshold: 0.92
stream_policy: sentence_buffered
routing:
numeric_keywords:
- revenue
- margin
- percentage
- total
- 增长率
- 合计
- 利润率
配置中的阈值需要通过自己的标注集校准。不能因为某个系统报告了约 99% 的数值准确率，就把这一结果直接视为其他数据集、模型和提示词上的保证。
跨账号、多模型故障转移不能只做“重试”
Amazon Bedrock 的生产部署经常涉及多个 AWS 账号、区域或模型。跨账号隔离有助于控制权限和故障域，但故障转移需要处理几个容易忽视的问题：
- 使用 AWS STS 承担目标账号角色，并给予最小化的 Bedrock 调用权限。
- 为不同模型准备兼容的请求模板，而不是假设所有模型参数完全一致。
- 只对限流、暂时不可用和服务端错误执行故障转移；权限错误和输入错误应直接暴露。
- 如果已经向用户发送了部分 token，不要悄悄切换模型并从头生成，否则可能出现重复或语义跳变。
- 记录实际使用的账号、区域、模型、重试次数和评估结果，便于追踪质量回退。
下面的 Python 示例展示了一种可以改造的最小实现：它通过环境变量配置多个 Bedrock 目标，逐个承担角色并调用 ConverseStream；只有在尚未输出任何文本时，才允许切换到备用模型。示例还在流式阶段执行基础监控，并在结束后计算复合质量分数。
运行前需要安装 boto3，并确保当前身份拥有 sts:AssumeRole 权限；目标角色还需要相应的 Bedrock 模型调用权限。模型 ID、区域和角色 ARN 必须替换成自己的值。
python -m pip install 'boto3>=1.34.131'
export BEDROCK_TARGETS='[
{"region":"us-east-1","role_arn":null,"model_id":"YOUR_PRIMARY_MODEL_ID"},
{"region":"us-west-2","role_arn":"arn:aws:iam::123456789012:role/BedrockInvokeRole","model_id":"YOUR_FALLBACK_MODEL_ID"}
]'
python bedrock_qa.py
将下面内容保存为 bedrock_qa.py：
import json
import os
import re
import sys
from decimal import Decimal, InvalidOperation
import boto3
from botocore.exceptions import ClientError
TARGETS = json.loads(os.environ["BEDROCK_TARGETS"])
RETRYABLE = {
"ThrottlingException",
"ServiceUnavailableException",
"InternalServerException",
"ModelTimeoutException",
}
def runtime_client(target):
if not target.get("role_arn"):
return boto3.client("bedrock-runtime", region_name=target["region"])
sts = boto3.client("sts")
credentials = sts.assume_role(
RoleArn=target["role_arn"],
RoleSessionName="bedrock-quality-gateway",
)["Credentials"]
return boto3.client(
"bedrock-runtime",
region_name=target["region"],
aws_access_key_id=credentials["AccessKeyId"],
aws_secret_access_key=credentials["SecretAccessKey"],
aws_session_token=credentials["SessionToken"],
)
def normalize_number(value):
try:
return Decimal(value.replace(",", "").rstrip("%"))
except InvalidOperation:
return None
def extract_numbers(text):
matches = re.findall(r"-?\d[\d,]*(?:\.\d+)?%?", text)
return {n for raw in matches if (n := normalize_number(raw)) is not None}
def evaluate(text, expected_numbers, required_terms):
actual = extract_numbers(text)
expected = {Decimal(str(number)) for number in expected_numbers}
numeric_score = (
len(expected & actual) / len(expected) if expected else 1.0
)
term_score = (
sum(term.lower() in text.lower() for term in required_terms)
/ len(required_terms)
if required_terms
else 1.0
)
safety_score = 0.0 if "BEGIN PRIVATE KEY" in text else 1.0
composite = (
0.60 * numeric_score
+ 0.25 * term_score
+ 0.15 * safety_score
)
return {
"numeric_score": round(numeric_score, 3),
"required_term_score": round(term_score, 3),
"safety_score": round(safety_score, 3),
"composite_score": round(composite, 3),
"numbers_seen": sorted(str(number) for number in actual),
}
def stream_with_failover(prompt):
for index, target in enumerate(TARGETS):
emitted = False
pieces = []
try:
response = runtime_client(target).converse_stream(
modelId=target["model_id"],
messages=[{
"role": "user",
"content": [{"text": prompt}],
}],
inferenceConfig={
"maxTokens": 500,
"temperature": 0.0,
},
)
for event in response["stream"]:
delta = event.get("contentBlockDelta", {}).get("delta", {})
text = delta.get("text")
if not text:
continue
emitted = True
pieces.append(text)
sys.stdout.write(text)
sys.stdout.flush()
# 轻量级实时检查；复杂事实校验应放在句子缓冲区之后。
if sum(map(len, pieces)) > 20_000:
raise RuntimeError("Response exceeded the configured limit")
print()
return "".join(pieces), target
except ClientError as exc:
code = exc.response.get("Error", {}).get("Code", "Unknown")
can_fail_over = (
not emitted
and code in RETRYABLE
and index < len(TARGETS) - 1
)
if can_fail_over:
print(
f"Primary target failed with {code}; trying fallback",
file=sys.stderr,
)
continue
raise
raise RuntimeError("No Bedrock target produced a response")
if __name__ == "__main__":
expected_revenue = 125.4
expected_margin = 18.2
prompt = f"""
请仅根据以下已验证数据回答，并保留单位：
- 本季度收入：{expected_revenue} 百万元
- 利润率：{expected_margin}%
用一句话总结本季度表现。不要推测未提供的数据。
""".strip()
answer, selected_target = stream_with_failover(prompt)
report = evaluate(
answer,
expected_numbers=[expected_revenue, expected_margin],
required_terms=["收入", "利润率"],
)
print(json.dumps({
"selected_target": selected_target,
"evaluation": report,
"accepted": report["composite_score"] >= 0.92,
}, ensure_ascii=False, indent=2))
这个示例故意保持了边界清晰：它能验证预期数字是否出现，却不能理解“18.2%”与“0.182”是否等价，也无法判断年份、币种和数字之间的关系。生产实现应把数值抽取升级为结构化字段校验，例如同时验证 metric、period、value、unit 和 source_record_id。
流式评估的核心是决定何时允许内容离开系统
“边生成边评估”并不意味着检查完成前就把所有 token 原样发给用户。不同风险等级需要不同的输出策略：
- 直接透传：延迟最低，适用于低风险内容；发现问题后通常已经无法撤回。
- 固定 token 缓冲：积累几十个 token 后检查，再批量发送，能拦截部分敏感内容。
- 按句缓冲：等待句号或段落结束，对完整语义单元执行检查，适合事实和数字验证。
- 完整答案门控：全部生成并通过评估后才返回，质量控制最强，但失去真正的流式体验。
数字校验尤其适合按句缓冲，因为数字 token 可能被拆分，单独检查每个增量容易产生误报。若检测到错误，可以重写当前句、回退到备用模型，或向客户端发送结构化错误事件，而不是继续输出未经验证的内容。
对于 Server-Sent Events，可以约定业务事件类型：
event: delta
data: {"text":"本季度收入为 125.4 百万元"}
event: quality
data: {"numeric_score":1.0,"composite_score":0.95}
event: done
data: {"accepted":true,"model":"fallback-or-primary-id"}
客户端不应只拼接 delta。它还需要处理 quality、blocked、retrying 和 done 等状态，才能把后端质量控制真实反映给用户。
复合评分比单一“正确/错误”更适合生产决策
单个评估器容易掩盖问题。例如答案包含所有正确数字，但把收入说成利润；或者事实正确，却泄露了不应返回的内部内容。复合评估可以把多个维度组合起来：
- 数值准确性与单位一致性；
- 是否引用了要求使用的证据；
- 回答是否覆盖用户问题；
- 安全、隐私和合规检查；
- 格式、长度和指令遵循情况。
不过，加权平均也有陷阱。安全检查不应被其他高分抵消，因此更合理的策略是同时使用硬门槛和软评分：
接受条件 = 安全检查通过
AND 数值准确率 >= 0.98
AND 复合分数 >= 0.92
权重和阈值必须基于真实业务数据调优。评估集至少应覆盖正常请求、边界值、缺失数据、冲突来源、模型拒答、限流和流式中断等情况。
上线前的工程检查清单
将这套思路落地时，可以按以下顺序推进：
- 为请求标注风险等级，不让所有任务共享一条生成路径。
- 使用结构化事实作为数字校验依据，避免再调用一个 LLM 来“猜”数字是否正确。
- 为每个备用模型执行兼容性测试，并记录实际发生的模型切换。
- 只在安全的边界进行流式故障转移，避免重复输出和语义断裂。
- 对高风险内容采用句子缓冲或完整门控，而不是无条件透传 token。
- 将原始回答、证据版本、模型 ID、评估器版本和各项分数写入审计日志。
- 用自己的黄金数据集持续回归测试，不把约 99% 当成脱离场景的通用承诺。
真正可上线的 LLM 质量保障，不是给模型调用包一层重试，而是把路由、冗余、验证、流式协议和审计连接成一个闭环。这样即使底层模型或账号发生变化，应用仍然拥有明确、可测量的质量边界。