Couchbase 为 Capella iQ 引入 Amazon Bedrock，并使用 Anthropic Claude 系列模型提供生成式 AI 能力。值得关注的不只是“调用了哪个模型”，而是它在生产环境中采用了多模型架构：应用不再与单一模型紧密绑定，而是在统一入口后面完成模型选择、请求治理和运行监控。
这类设计适合需要持续迭代的 AI 产品。模型能力、价格、延迟和上下文窗口都在变化，如果业务代码直接依赖某个模型的请求格式，每次升级都可能演变成一次跨模块改造。
多模型不是多写几个模型 ID
多模型架构的核心是把模型差异限制在推理层，而不是散落到产品代码中。一个可维护的调用链通常可以拆成四层：
- 产品层负责定义任务，例如解释查询、生成代码或总结文档。
- 路由层根据任务、延迟预算和模型能力选择目标模型。
- 适配层把统一请求转换为 Bedrock 所需的模型输入，并规范化响应。
- 治理层记录延迟、令牌消耗、错误类型和模型版本，执行重试与降级。
Amazon Bedrock 在这里承担统一的模型访问入口。应用可以通过相对稳定的 AWS API 和权限体系调用 Claude 系列模型，而无需单独维护模型托管基础设施。对生产团队而言，这也意味着身份认证、区域配置和审计可以进入现有的 AWS 运维体系。
不过，多模型并不等于每次请求都动态选择模型。对于高风险任务，更稳妥的方式通常是维护明确的路由规则：某类任务固定使用经过评测的模型，只有在版本发布或故障降级时才切换。
路由决策要由评测数据驱动
不同任务对模型的要求并不相同。生成一段简短解释可能更看重响应速度，而分析复杂查询或处理长上下文则更看重推理质量。路由器至少需要考虑以下信号：
- 任务类型和提示词模板版本
- 输入长度及预期输出长度
- 延迟目标与调用成本
- 模型所在区域及当前可用性
- 离线评测分数和线上失败率
- 数据敏感级别与访问策略
路由规则必须能被测试。不要只用“回答看起来不错”作为上线标准，可以准备一组脱敏样本，分别记录正确性、结构合规率、拒答率、P95 延迟和单次请求成本。更新模型前，用同一数据集比较候选版本，并为关键任务设置最低通过线。
模型输出也不能直接当成可信指令执行。如果 Capella iQ 一类助手生成数据库查询或操作建议，产品层仍应增加语法解析、只读限制、资源配额和人工确认。多模型架构解决的是模型演进与运行治理问题，不会自动消除提示词注入、错误生成或越权操作。
可以这样实践：用 Bedrock 构建一个最小模型路由器
下面是一个可运行并改造的 Python 示例。它根据任务复杂度选择不同的 Claude 模型，并在首选模型调用失败时尝试后备模型。示例不代表 Capella iQ 的内部实现，只展示这种架构可以如何落地。
运行前需要：
- 安装并配置 AWS CLI 凭证；
- 在目标 AWS 区域启用相应的 Bedrock 模型访问权限；
- 将环境变量中的模型 ID 替换为账号和区域实际可用的 Claude 模型 ID。
python -m pip install boto3
export AWS_REGION=us-east-1
export BEDROCK_FAST_MODEL_ID='replace-with-your-fast-claude-model-id'
export BEDROCK_REASONING_MODEL_ID='replace-with-your-reasoning-claude-model-id'
创建 app.py：
import os
import time
from dataclasses import dataclass
import boto3
from botocore.exceptions import BotoCoreError, ClientError
@dataclass(frozen=True)
class Task:
prompt: str
complexity: str = "low"
max_tokens: int = 500
REGION = os.getenv("AWS_REGION", "us-east-1")
FAST_MODEL = os.environ["BEDROCK_FAST_MODEL_ID"]
REASONING_MODEL = os.environ["BEDROCK_REASONING_MODEL_ID"]
client = boto3.client("bedrock-runtime", region_name=REGION)
def model_order(task: Task) -> list[str]:
if task.complexity == "high" or len(task.prompt) > 4_000:
return [REASONING_MODEL, FAST_MODEL]
return [FAST_MODEL, REASONING_MODEL]
def invoke(model_id: str, task: Task) -> str:
started = time.perf_counter()
response = client.converse(
modelId=model_id,
system=[
{
"text": (
"You are a database assistant. Explain uncertainty, "
"and never claim that an unexecuted query succeeded."
)
}
],
messages=[
{
"role": "user",
"content": [{"text": task.prompt}],
}
],
inferenceConfig={
"maxTokens": task.max_tokens,
"temperature": 0.1,
},
)
elapsed_ms = int((time.perf_counter() - started) * 1000)
usage = response.get("usage", {})
print(
{
"model_id": model_id,
"latency_ms": elapsed_ms,
"input_tokens": usage.get("inputTokens"),
"output_tokens": usage.get("outputTokens"),
}
)
return response["output"]["message"]["content"][0]["text"]
def run(task: Task) -> str:
errors = []
for model_id in model_order(task):
try:
return invoke(model_id, task)
except (ClientError, BotoCoreError) as exc:
errors.append(f"{model_id}: {exc}")
raise RuntimeError("All configured models failed: " + " | ".join(errors))
if __name__ == "__main__":
result = run(
Task(
prompt=(
"Explain why a database query might stop using an index after "
"a schema or data-distribution change. Return three checks."
),
complexity="high",
)
)
print(result)
运行命令：
python app.py
示例使用 Bedrock Converse API 统一消息格式，并把路由逻辑放在独立函数中。生产实现还应区分不可重试错误与限流、超时等短暂故障；后备模型也应通过兼容性评测，不能仅因为 API 调用成功就认为它可以替代首选模型。
生产收益来自标准化，而不只是模型能力
Couchbase 的案例说明，多模型路线的运营价值来自统一管理。团队可以在同一架构内评估 Claude 系列的不同模型，逐步发布新版本，并围绕模型调用建立一致的监控和故障处理方式。这比在每个功能中单独集成模型更容易控制变更范围。
要让这些收益真正落地，建议至少记录以下字段：请求 ID、任务类型、模型 ID、提示词版本、输入与输出令牌数、首令牌延迟、总延迟、重试次数、结果状态和安全过滤结果。日志中不要保存原始敏感数据；可以记录经过脱敏的摘要、哈希值或数据分类标签。
缓存也需要谨慎使用。确定性较高的解释类请求可以按“模型版本 + 提示词版本 + 规范化输入”生成缓存键，但涉及实时数据、用户权限或数据库状态的回答不应复用过期结果。
采用前检查清单
引入类似架构时，可以按下面的顺序推进：
- 为每种任务定义质量、延迟和成本目标，而不是先选模型再寻找用途。
- 建立统一模型适配层，禁止业务模块直接拼接供应商请求体。
- 用脱敏评测集验证每个模型和提示词版本。
- 配置超时、有限次数重试、并发限制和明确的降级路径。
- 对模型生成的查询执行解析、权限检查和资源限制。
- 记录模型版本与提示词版本，使线上问题可以复现。
- 通过小流量或影子请求比较新旧模型，再逐步扩大范围。
多模型架构会增加路由、评测和可观测性的工程成本，但它换来的是更可控的模型升级和故障处理。对于 Capella iQ 这类持续演进的 AI 产品，这层抽象的价值不在于追逐模型数量，而在于让模型变化不再牵动整个应用。