数据截至 (上游 commit 0b46c8430636)
模型抽象层:统一接口与多 provider
30 秒导读: Strands 号 称「model-agnostic(与模型无关)」——同一段 agent 代码,底层可以是 Bedrock、Anthropic、OpenAI、Gemini、Ollama…… 换模型只改一行。这靠的是一个抽象基类
Model(abc.ABC):它把「和一个大模型对话」这件事,压成 4 个抽象方法 + 2 个属性 + 1 个默认实现。每个 provider 只要实现这套接口、把它翻译到自家 SDK,就能无缝插进 agent 事件循环。本章讲清这套抽象长什么样、Bedrock 这个默认实现怎么落地、以及各 provider 的差异点在哪。
1. 这是什么(为什么需要一层抽象)
先说痛点。 每家模型厂商的 SDK 长得都不一样:Bedrock 叫 converse_stream、Anthropic 叫 client.messages.stream、OpenAI 是 chat.completions.create。请求格式、流式事件、错误类型、工具 schema 全都对不上。如果 agent 主循环直接调某一家的 SDK,那这段代码就和那家厂商焊死了。
这一层要解决的就是它: 在 agent 主循环和五花八门的厂商 SDK 之间,插一层统一的窄接口。主循环只认这层接口;每个厂商写一个「适配器」把接口翻译到自己的 SDK。
- 一句话定义:
Model是所有模型 provider 的抽象基类,定义了「配置模型 + 流式对话 + 结构化输出」的标准契约。 - 给谁用 / 解决什么: 让 agent 作者不必关心底层是谁;让新 provider 的接入只是「填一个类 」。
- 一句话直觉: 把它想成数据库里的 ODBC/JDBC——上层写一套 SQL,底层换 MySQL 还是 Postgres 只换一个驱动。这里「SQL」就是
stream(),「驱动」就是BedrockModel、AnthropicModel……
用起来什么样——换 provider 只动构造那一行,Agent(...) 和后面所有代码都不变:
# 示意,非源码:model-agnostic 的直观体现
from strands import Agent
from strands.models import BedrockModel, AnthropicModel
model = BedrockModel(model_id="global.anthropic.claude-sonnet-4-6") # 默认 provider
# 想换成 Anthropic 直连?只改这一行:
# model = AnthropicModel(model_id="claude-sonnet-4-6", client_args={"api_key": "..."})
agent = Agent(model=model) # 下面全部与 provider 无关
agent("帮我把这个函数改成异步的") # 主循环只调 model.stream(...),不认识 boto3/anthropic
本节不碰底层细节。记住一点:下面所有内容,都是在解释「那一行之下、主循环之上」的这层薄抽象怎么工作。
2. 顶层全景(它大概怎么转)
一次模型调用在这层里的流向,是「统一请求下沉、厂商 SDK 干活、统一事件上浮」:
统一世界(与 provider 无关) 厂商世界(各家 SDK)
┌───────────────┐ messages/tool_specs ┌──────────────────┐ converse_stream() ┌──────────┐
│ agent 事件循环 │ ───────────────────────▶│ 某个 Model 子类 │ ─────────────────────▶│ 厂商 API │
│ (第1章 主线) │ 调 model.stream() │ (如 BedrockModel) │ format_request() │ Bedrock… │
└───────────────┘ └──────────────────┘ └──────────┘
▲ │ 厂商原生 chunk │
│ 标准 StreamEvent 事件流 │ format_chunk() 翻译回统一格式 │
└────────────────────────────────────────────┴◀──────────────────────────────────────┘
messageStart / contentBlockDelta / messageStop / metadata …
怎么读这张图: 左到右是「下沉」——统一的 messages/tool_specs 经 format_request 变成厂商请求;右到左是「上浮」——厂商的原生流式 chunk 经 format_chunk 翻译回一套标准事件再交还主循环。抽象层的全部工作,就是这两次翻译。
各部件职责:
| 部件 | 干什么 | 在哪 |
|---|---|---|
Model(abc.ABC) | 定义统一契约:4 抽象方法 + 2 属性 + 默认 count_tokens | models/model.py:158 |
stream() 契约 | 产出标准 StreamEvent 事件流,与第1章 streaming 对接 | types/streaming.py:208 StreamEvent |
| provider 子类 | 把统一接口翻译到自家 SDK(默认 BedrockModel) | models/bedrock.py:84 |
format_request / format_chunk | 请求下沉翻译 / 事件上浮翻译(约定俗成,非抽象) | 各 provider 文件内 |
_validation / _strict_schema | 配置键校验、strict JSON schema 转换等共享工具 | models/_validation.py、models/_strict_schema.py |
ModelRetryStrategy | 节流(throttling)时按指数退避重试 | event_loop/_retry.py:21 |
主线走一遍(高层): 主循环拿着 messages + tool_specs 调 model.stream(...) → 子类 format_request 把它翻成厂商请求 → 调厂商 SDK 拿到原生流 → 逐块 format_chunk 翻回标准 StreamEvent → yield 回主循环,由主循环组装成消息 / 触发工具。
3. 统一接口:Model(abc.ABC)
这节讲契约本身——所有 provider 必须遵守的最小面。它就定义在 models/model.py:158 的 class Model(abc.ABC)。
3.1 四个抽象方法 + 两个属性
抽象基类刻意做得很窄:只强制四个方法,子类不实现就无法实例化(abc.ABC 的语义)。
| 成员 | 类型 | 位置 | 作用 |
|---|---|---|---|
stream(...) | 抽象方法 | model.py:227 | 核心:把一轮对话变成标准事件流(见 §4) |
structured_output(...) | 抽象方法 | model.py:206 | 让模型产出符合某个 Pydantic 模型的结构化结果 |
update_config(**cfg) | 抽象方法 | model.py:186 | 运行期改配置(温度、model_id 等) |
get_config() | 抽象方法 | model.py:196 | 取回当前配置 |
stateful | 属性(默认 False) | model.py:166 | 模型是否服务端托管会话状态 |
context_window_limit | 属性 | model.py:175 | 上下文窗口 token 上限(读配置) |
count_tokens(...) | 默认实现(可覆盖) | model.py:263 | 发送前估算 token 数(见 §3.2) |
stream 是心脏。 它的签名把「一轮对话」需要的一切摆平了:messages、tool_specs、system_prompt,以及 tool_choice、system_prompt_content、invocation_state 等仅限关 键字参数:
# 真实签名节选,models/model.py:230 Model.stream
def stream(
self,
messages: Messages,
tool_specs: list[ToolSpec] | None = None,
system_prompt: str | None = None,
*, # ← 之后全是仅限关键字(keyword-only)
tool_choice: ToolChoice | None = None,
system_prompt_content: list[SystemContentBlock] | None = None,
invocation_state: dict[str, Any] | None = None,
**kwargs: Any,
) -> AsyncIterable[StreamEvent]: ...
- 那个裸
*是刻意的约定:system_prompt之后的参数一律仅限关键字。这样以后加新选项,不会打乱主循环的调用点——这是「统一接口能演进而不破坏调用方」的关键设计。 - 返回类型基类写
AsyncIterable[StreamEvent],provider 覆盖时收窄成AsyncGenerator[StreamEvent, None]。子类用@override并配# type: ignore[override],因为它收窄了**kwargs(见bedrock.py:248的update_config)。
structured_output(model.py:206) 要求「模型输出必须匹配给定的 output_model: type[T](一个 pydantic.BaseModel)」;它同样是异步生成器,最后一个 yield 的事件带 {"output": <实例>},前面的事件是普通流。做法各家不 同(见 §6)。
stateful(model.py:166)默认 False。 它是 model-agnostic 里一个精妙的钩子:大多数模型是无状态的(每轮都要把完整历史传回去),但少数 provider(如 OpenAI Responses API,openai_responses.py:195 覆盖了这个属性)在服务端记住会话。§8 会讲这个属性如何驱动一次消息清理。
context_window_limit(model.py:175) 只是读配置里的 context_window_limit(定义在 BaseModelConfig,model.py:121),给上下文管理(如到阈值触发压缩,见第5章)用。
3.2 默认 count_tokens:启发式兜底
要解决的小问题: 主动上下文管理需要「在发送前」知道大概多少 token,好决定要不要压缩。但不是每个 provider 都有 token 计数 API。
思路: 基类给一个依赖无关的启发式兜底——有 tiktoken 就用它,没有就按字符数硬估。count_tokens(model.py:263)默认就走这条兜底,provider 可以覆盖成原生 API(Bedrock 就覆盖了,见 §5)。
启发式的核心是几个小函数,规则很直白:
| 函数 | 位置 | 规则 |
|---|---|---|
_heuristic_estimate_text | model.py:27 | 文本按 字符数 / 4 向上取整 |
_heuristic_estimate_json | model.py:32 | JSON 对象按 序列化后字符数 / 2 |
_count_content_block_tokens | model.py:40 | 拆开一个内容块,分别对 text / toolUse / toolResult / reasoning / 引用等累加 |
_estimate_tokens_with_heuristic | model.py:91 | 遍历所有消息 + tool_specs + system,汇总 |
一个关键取舍写在注释里:toolResult 里的图片 / 文档是二进制,启发式故意不计(model.py:62)。文档也诚实说明:精度因模型而异,不用于计费或精确配额(model.py:274)。
3.3 缓存配置:CacheConfig / CacheToolsConfig
抽象层还提供两个跨 provider 的缓存配置数据类(prompt caching = 让厂商缓存重复的前缀,省 token/延迟):
| 数据类 | 位置 | 字段 | 干什么 |
|---|---|---|---|
CacheConfig | model.py:134 | strategy("auto"/"anthropic")、ttl | 自动注入 cachePoint 的策略与过期时间 |
CacheToolsConfig | model.py:153 | type("default")、ttl | 给 toolConfig 那块单独设缓存点 |
它们是普通 @dataclass(不是 wire shape,所以不用 TypedDict),被 provider 的配置引用——例如 BedrockConfig.cache_config(bedrock.py:151)。真正「往请求里注入缓存点」的逻辑在各 provider,见 §5 的 _inject_cache_point。
4. stream() 的标准事件契约
这节讲上浮方向的统一格式——无论底层是谁,stream() 吐出来的事件都长同一个样,这样第1章的流式回合逻辑才能与 provider 解耦。
标准事件类型全都定义在 types/streaming.py:208 的 StreamEvent(一个 total=False 的 TypedDict,建模自 Bedrock 的 Converse API——所以 Bedrock 是「最省翻译」的一家):
| 事件键 | 含义 | 对应类型 |
|---|---|---|
messageStart | 一条消息开始(带 role) | MessageStartEvent (streaming.py:16) |
contentBlockStart | 一个内容块开始(如工具调用的 id/name) | ContentBlockStartEvent (streaming.py:26) |
contentBlockDelta | 增量:文本 / 工具输入 / 推理 / 引用 | ContentBlockDeltaEvent (streaming.py:123) |
contentBlockStop | 一个内容块结束 | ContentBlockStopEvent (streaming.py:136) |
messageStop | 消息结束(带 stopReason) | MessageStopEvent (streaming.py:147) |
metadata | usage / metrics / trace | MetadataEvent (streaming.py:159) |
redactContent | 护栏触发时的内容抹除 | RedactContentEvent (streaming.py:195) |
一次典型回合,provider 必须按这个次序把厂商流翻译出来:
messageStart
└─ contentBlockStart ─▶ contentBlockDelta × N ─▶ contentBlockStop (可重复:文本块、工具块…)
messageStop (stopReason)
metadata (usage / metrics)
这就是「契约」的意义:主循环只写一次「拼装这串事件」的逻辑,底层换谁都不用改。Bedrock 因为标准事件本就仿它,format_chunk 几乎是直传;其他家则要把自己的 SSE/事件对象映射成这套键(见 §6)。structured_output 复用了这条流:它内部调 stream(),再经 process_stream(event_loop/streaming.py:394)聚合(见 §5)。
5. 参照实现:BedrockModel(默认 provider)
这节把默认实现读透——class BedrockModel(Model)(bedrock.py:84)。它是 Agent() 不指定模型时的兜底,也是理解「统一接口如何落到一家 SDK」的最佳样本。
5.1 配置与默认模型
配置是嵌套 TypedDict BedrockConfig(BaseModelConfig, total=False)(bedrock.py:96),继承了公共的 context_window_limit,再加一堆 Bedrock 特有项(guardrail、cache、strict_tools、service_tier…)。
构造(bedrock.py:165)只接 boto 相关参数 + **model_config,先 validate_config_keys 校验键,再存进 TypedDict。默认 model id 由 _get_default_model_with_warning(bedrock.py:1334)按 region 前缀推断(us/eu/ap),常量 DEFAULT_BEDROCK_MODEL_ID = "global.anthropic.claude-sonnet-4-6"(bedrock.py:42);不认识的 region 会告警并提示显式传 model_id。
5.2 下沉翻译:format_request
format_request(bedrock.py:257)是「统一 → Bedrock」的核心。它把 messages/tool_specs/system 拼成 Converse 请求,几处值得注意:
- 工具 schema 直接用统一格式——
tool_spec["inputSchema"](Bedrock 本就用这套),仅当开了strict_tools才过ensure_strict_json_schema并加strict: True(bedrock.py:314-319)。 - 缓存点注入:
_inject_cache_point(bedrock.py:418)在strategy="auto"且模型是 Claude 系时,把 cachePoint 追加到最后一条 user 消息(_cache_strategy,bedrock.py:220)。 - 严格字段过滤:
_format_bedrock_messages(bedrock.py:489)会逐块删掉 Bedrock 不认识的字段——因为 Bedrock 对未知字段直接抛校验异常(注释见bedrock.py:512),这与「其他 API 忽略未知字段」不同,是这家的坑。
5.3 异步↔同步桥:stream + _stream
boto3 是同步的,而抽象接口是 async。Bedrock 用「后台线程 + 队列」搭桥:
stream() [async] _stream() [在 asyncio.to_thread 里同步跑]
│ 建 asyncio.Queue │ format_request()
│ create_task(_stream) ───────────────▶ │ client.converse_stream(**request)
│ │ for chunk in stream: callback(chunk)
│ while: event = await queue.get() ◀──── callback: loop.call_soon_threadsafe(put)
│ yield event │ finally: callback(None) # 哨兵=结束
▼ ▼
stream(bedrock.py:970)开一个asyncio.Queue,把阻塞的_stream丢进asyncio.to_thread,自己await queue.get()逐个yield;callback(None)是结束哨兵(bedrock.py:1002)。_stream(bedrock.py:1029)按streaming配置走converse_stream(流式)或converse(非流式,再由convert_non_streaming_to_streaming(bedrock.py:1145)手工拼出 §4 那串标准事件)。- 这正是[第3章外]AGENTS 约定的「阻塞调用包进
asyncio.to_thread、绝不阻塞事件循环」。