Provider 无关的模型层

30 秒导读: Upsonic 支持 25+ 家模型 provider(OpenAI、Anthropic、Google、Bedrock、Groq、Ollama……),但上层代码从不关心你用哪家。秘密是一个抽象基类 Model 定义了统一的三个方法(请求 / 数 token / 流式),外加一个 infer_model() 分发表把 'anthropic/claude-sonnet-4-5' 这样的字符串解析成对应的 Model 子类实例。本章讲这套字符串 → 真实 API 调用的完整链路。

本章只讲模型侧的抽象。一次完整运行怎么被 24 步管线调度(ModelExecutionStep 何时调用 model.request)是 02-execution-pipeline.md 的内容;这里我们把镜头对准"管线拿到 self.model 之后,那一次调用内部发生了什么"。

---

1. 这是什么(零基础也能懂)

一句话定义: 模型层是 Upsonic 里"对接大模型 API"的那一层——它把 25+ 家 provider 各不相同的 SDK、请求格式、返回格式,藏在一个统一接口后面。

它要解决的痛点: 每家模型厂商的 API 都长得不一样。

差异点	Anthropic	OpenAI	Google
系统提示	顶层独立的 system 参数	混在 messages 里的一条 role=system	独立的 system_instruction
工具调用返回	BetaToolUseBlock	tool_calls[].function	functionCall part
流式事件	content_block_delta 等	chat.completion.chunk	GenerateContentResponse 分片

如果每个用到模型的地方都写 if provider == 'anthropic': ... elif ...,代码会立刻失控。模型层的职责就是:在一处把这些差异吸收掉,对外只暴露一套统一的输入/输出。

用起来什么样: 上层永远只给一个字符串,拿回一个统一的 ModelResponse。

# 示意,非源码 —— 体会"字符串进,统一对象出"
from upsonic import infer_model

model = infer_model("anthropic/claude-sonnet-4-5")   # 字符串 → AnthropicModel 实例
resp = await model.request(messages, settings, params)  # 统一接口
print(resp.parts)   # 统一的 ModelResponse,不管底层是哪家

换成 "openai/gpt-4o" 或 "ollama/llama3.1",上面这段一个字都不用改——这就是"provider 无关"。

一句话直觉/类比: 把它当成模型界的电源转换插头。世界各地插座形状不同(各家 API),你的电器(Agent 主逻辑)只认一种插头(Model 接口),转换插头(各 Model 子类)负责适配当地插座。

---

2. 顶层全景(它大概怎么转)

一次 infer_model("anthropic/claude-sonnet-4-5") 到真实 HTTP 请求,数据这样流:

 "anthropic/claude-sonnet-4-5"
        │
        ▼
 ┌──────────────────┐   ① 归一化别名 + 拆 provider/model
 │  infer_model()   │      normalize_model_id → "anthropic","claude-sonnet-4-5"
 │  (分发表)         │   ② provider 名 → Provider 对象(持有 SDK client)
 └──────────────────┘   ③ 按 provider 家族 new 出对应 Model 子类
        │
        ▼
 ┌──────────────────┐   持有: client(厂商 SDK)、profile(能力画像)、settings
 │  AnthropicModel  │   继承自 Model(统一接口)
 └──────────────────┘
        │  model.request(messages, settings, params)
        ▼
 ┌──────────────────┐   prepare_request(): 合并 settings、按 profile 裁剪 params
 │  统一前处理       │   校验:该模型支不支持 native 输出 / tool 输出 / 图像输出?
 └──────────────────┘
        │
        ▼
 ┌──────────────────┐   _map_message(): upsonic 消息 → Anthropic 消息格式
 │  厂商映射         │   _map_tool_definition(): ToolDefinition → BetaToolParam
 │  (子类私有)       │   client.beta.messages.create(...)  ← 真·API 调用
 └──────────────────┘
        │  厂商返回 (BetaMessage / 流式事件)
        ▼
 ┌──────────────────┐   _process_response(): BetaTextBlock → TextPart …
 │  统一后处理       │   → ModelResponse(parts=[...], usage=..., finish_reason=...)
 └──────────────────┘
        │
        ▼
   ModelResponse(统一对象,交回管线)

怎么读这张图: 从上到下是一次调用的时间顺序。中间三段("统一前处理 → 厂商映射 → 统一后处理")就是模型层的全部工作——两头是统一格式,只有中间那一段是 provider 特有的。

各部件一句话职责:

部件	干什么	在哪(相对 src/upsonic/)
infer_model()	字符串 → Model 子类实例的分发表	models/__init__.py:2064
Model	抽象基类,定义统一接口 request/count_tokens/request_stream	models/__init__.py:655
ModelRequestParameters	一次请求的工具/输出配置(与 settings 分开)	models/__init__.py:628
ModelSettings	采样参数(温度、max_tokens……),跨 provider 通用子集	models/settings.py:8
StreamedResponse	流式返回的统一抽象,把厂商事件翻成统一事件	models/__init__.py:1671
Provider	持有厂商 SDK client、base_url、能力 profile	providers/__init__.py:16
各 *Model 子类	每家一份,负责真实请求/返回映射	models/anthropic.py 等 26 个文件
WrapperModel	装饰器基类,透明包住另一个 Model	models/wrapper.py:18
InstrumentedModel	给请求加 OpenTelemetry 埋点的包装	models/instrumented.py:371
model_registry / model_selector	模型元数据库 + 按任务选型	models/model_registry.py、models/model_selector.py

---

3. 核心原理(逐个机制,由浅入深)

3.1 Model 抽象基类:三个方法定义"一个模型能干什么"

要解决的小问题: 上层要能对任何模型做三件事——发一次请求、(可选)提前数 token、以流式方式发请求。基类就把这三件事钉死成接口。

Model 继承自 Runnable(UEL 的可组合单元,让 prompt | model 这种管道成立),核心是三个方法(models/__init__.py:727-759):

方法	作用	是否必须实现
request(...)	发一次请求,返回完整 ModelResponse	是(@abstractmethod)
count_tokens(...)	请求前数入参 token(给用量上限用)	否,默认抛 NotImplementedError
request_stream(...)	发请求并流式返回 StreamedResponse	否,默认抛 NotImplementedError

三者签名完全一致,都吃这三样东西:

• messages: list[ModelMessage] —— 统一的对话消息(系统/用户/助手/工具返回)。
• model_settings: ModelSettings | None —— 采样参数(见 §3.2)。
• model_request_parameters: ModelRequestParameters —— 工具与输出配置(见 §3.3)。

关键设计:两类参数刻意分开。 ModelSettings 是"调模型的旋钮"(温度、top_p……),ModelRequestParameters 是"这次要挂哪些工具、要不要结构化输出"。前者可被 agent 级默认值与单次调用值合并,后者由 agent 每轮重新构建。

除了三个动作方法,基类还有一批元信息属性,让上层不碰厂商 SDK 就能了解一个模型(models/__init__.py:1512-1615):

# 示意,非源码 —— 基类暴露的只读元信息
model.model_name    # 'claude-sonnet-4-5'
model.system        # 'anthropic'  (OTel gen_ai.system 用)
model.model_id      # 'anthropic:claude-sonnet-4-5'
model.label         # 'Claude Sonnet 4.5'  (给人看的展示名)
model.profile       # ModelProfile:这模型支不支持 json schema 输出 / 工具 / 图像输出…

其中 profile(能力画像)是后面统一前处理做校验的依据——它是"这模型到底能干什么"的单一事实源(models/__init__.py:1563-1586)。

3.2 ModelSettings:跨 provider 的采样参数公共子集

要解决的小问题: 温度、max_tokens 这些参数几乎人人都有,但不完全一样。怎么让一份配置能喂给任何模型?

答案是一个 TypedDict,只收多家共有的字段,并在文档字符串里逐字段标注"哪些 provider 支持"(models/settings.py:8):

# 摘自 settings.py,已精简
class ModelSettings(TypedDict, total=False):
    max_tokens: int
    temperature: float
    top_p: float
    timeout: float | Timeout
    parallel_tool_calls: bool
    stop_sequences: list[str]
    extra_headers: dict[str, str]
    extra_body: object
    # ……

厂商私有的参数则由子类扩展这个 TypedDict,并强制加前缀。例如 AnthropicModelSettings(models/anthropic.py:169)加了 anthropic_thinking、anthropic_cache_instructions 等,注释明写规矩:

"ALL FIELDS MUST BE anthropic_ PREFIXED SO YOU CAN MERGE THEM WITH OTHER MODELS."

前缀的意义:因为多个模型的 settings 会用 | 合并成一份(merge_model_settings, models/settings.py:187),加前缀才能保证 fallback 到别家时不会把 anthropic_thinking 误喂给 OpenAI。合并规则很朴素——base | overrides,后者覆盖前者,用于"agent 级默认 settings"叠加"单次调用 settings"。

3.3 请求怎么组装:结构化输出如何变成一个 tool

要解决的小问题: 用户想让模型返回一个 Pydantic 对象(结构化输出),但很多模型并不原生支持"给我一段符合此 JSON schema 的输出"。怎么办?

Upsonic 的招数:把结构化输出伪装成一次工具调用。 请求组装发生在 agent 侧(agent/agent.py:2203 _build_model_request_parameters):

1. 收集函数工具定义(agent 级 + task 级),塞进 function_tools。
2. 若 task.response_format 是个 Pydantic 模型,就:
   • output_mode 设为 'auto'、allow_text_output=False;
   • 用 model_json_schema() 生成 schema,包成 OutputObjectDefinition;
   • 额外造一个"输出工具" 塞进 output_tools。

那个"输出工具"就是把 schema 当成工具参数的一次普通 ToolDefinition(agent/agent.py:2261 _build_output_tools):

# 摘自 agent.py:_build_output_tools,已精简
return [ToolDefinition(
    name=DEFAULT_OUTPUT_TOOL_NAME,          # 约定名 'final_result'
    parameters_json_schema=schema,           # 用户 Pydantic 模型的 schema
    description=response_format.__doc__ or ...,
    kind='output',                           # 标记:这是输出工具,不是普通工具
    strict=True,
)]

于是"要一个结构化对象"就退化成了"请调用 final_result(...) 这个工具"——模型只需做它本就擅长的工具调用,Upsonic 再把工具入参校验成 Pydantic 对象。这就是结构化输出转 tool 的核心思路。

组装好的 ModelRequestParameters(models/__init__.py:628)是个 dataclass,承载这次请求的全部"工具与输出"信息:

字段	含义
function_tools	普通函数工具定义
builtin_tools	内置工具(网页搜索、代码执行等)
output_mode	text / tool / native / prompted / auto
output_object	结构化输出的 schema 描述
output_tools	上面那个"输出工具"(tool 模式用)
allow_text_output	是否允许纯文本输出

3.4 统一前处理:prepare_request 按 profile 裁剪并校验

要解决的小问题: 上面组装出的 params 是"理想请求",但具体模型未必支持。谁来把理想请求落到某个模型的现实能力上?

答案是基类的 prepare_request()(models/__init__.py:782),每个子类的 request() 开头都会先调它。它做三件事:

1. 合并 settings —— merge_model_settings(self.settings, model_settings),模型自带默认叠加本次调用。
2. 裁剪 params —— 若 output_mode == 'auto',用 profile.default_structured_output_mode 敲定实际模式;并清掉与模式不符的字段(如非 tool 模式就清空 output_tools)。
3. 能力校验(会抛错) —— 拿 profile 逐条查:

# 摘自 __init__.py:prepare_request,已精简
if params.output_mode == 'native' and not self.profile.supports_json_schema_output:
    raise UserError('Native structured output is not supported by this model.')
if params.output_mode == 'tool' and not self.profile.supports_tools:
    raise UserError('Tool output is not supported by this model.')
if params.allow_image_output and not self.profile.supports_image_output:
    raise UserError('Image output is not supported by this model.')

重点看: profile 让"不支持"在发请求前就被拦下并给出清楚报错,而不是等厂商 API 返回一个晦涩的 400。子类还能覆写 prepare_request 追加自家逻辑——例如 Anthropic 覆写版(models/anthropic.py:362)专门处理"思考模式与输出工具不能同时用"这类厂商约束。

3.5 一次真实映射:AnthropicModel 怎么把请求打出去

思路: 前处理之后,剩下的就是纯翻译——把统一格式翻成 Anthropic SDK 认识的格式,调 client,再把返回翻回统一格式。

request() 的主干只有四步(models/anthropic.py:310-325):

# 摘自 anthropic.py:AnthropicModel.request
async def request(self, messages, model_settings, model_request_parameters):
    check_allow_model_requests()                              # 全局开关,测试时可禁网
    model_settings, model_request_parameters = self.prepare_request(...)  # §3.4 前处理
    response = await self._messages_create(messages, False, ...)          # 真·调用
    return self._process_response(response)                   # 返回翻回统一格式

真正拼请求体的是 _messages_create()(models/anthropic.py:415)。它把统一格式逐项映射到 Anthropic SDK 的 messages.create 参数:

• 工具:_map_tool_definition()(models/anthropic.py:1148)把 ToolDefinition 翻成 BetaToolParam,即 {name, description, input_schema}。
• 消息:_map_message() 把 upsonic 的 ModelMessage 拆成 Anthropic 的 system 顶层参数 + messages 列表(注意:系统提示在 Anthropic 是独立顶层参数,不像 OpenAI 混在 messages 里)。
• 最后落到唯一一处真实网络调用(models/anthropic.py:439):

# 摘自 anthropic.py:_messages_create,已精简
return await self.client.beta.messages.create(
    max_tokens=model_settings.get('max_tokens', 4096),
    system=system_prompt or OMIT,
    messages=anthropic_messages,
    model=self._model_name,
    tools=tools or OMIT,
    tool_choice=tool_choice or OMIT,
    temperature=model_settings.get('temperature', OMIT),
    stream=stream,
    ...
)

返回方向由 _process_response()(models/anthropic.py:550)负责:遍历 response.content 里的每个块,把 BetaTextBlock → TextPart、BetaToolUseBlock → ToolCallPart、BetaThinkingBlock → ThinkingPart……再连同 usage、finish_reason 打包成统一的 ModelResponse。至此,"是哪家模型"的信息被彻底抹平。

API 错误也统一。 _messages_create 把 SDK 的 APIStatusError/APIConnectionError 转成 Upsonic 自己的 ModelHTTPError/ModelAPIError(models/anthropic.py:460-465),这样上层的重试/降级逻辑不必认识每家 SDK 的异常类型。

3.6 流式如何统一回传:StreamedResponse

要解决的小问题: 各家的流式格式天差地别(Anthropic 发 content_block_delta,OpenAI 发 chat.completion.chunk)。上层想要"一种事件流"。

StreamedResponse 是个抽象基类(models/__init__.py:1671),它规定:子类只需实现一个 _get_event_iterator(),把厂商事件翻成统一的 ModelResponseStreamEvent;基类的 __aiter__ 再在外面裹两层通用逻辑:

厂商原始流
   │
   ▼
_get_event_iterator()   ← 子类实现:厂商事件 → 统一事件(用 _parts_manager 累积增量)
   │
   ▼
iterator_with_final_event()  ← 基类:命中输出 schema 时补发一个 FinalResultEvent
   │
   ▼
iterator_with_part_end()     ← 基类:在每个 Part 结束时补发 PartEndEvent
   │
   ▼
统一事件流(交给上层消费)

关键角色是 _parts_manager(ModelResponsePartsManager):流式返回是一串增量(delta),它负责把碎片按 vendor_part_id 累积成完整的 TextPart/ThinkingPart/ToolCallPart。以 Anthropic 为例(models/anthropic.py:1224 _get_event_iterator),收到 BetaToolUseBlock 分片时:

# 摘自 anthropic.py,已精简 —— 把厂商增量交给统一的 parts_manager
maybe_event = self._parts_manager.handle_tool_call_delta(
    vendor_part_id=event.index,
    tool_name=current_block.name,
    args=cast(dict, current_block.input) or None,
    tool_call_id=current_block.id,
)
if maybe_event is not None:
    yield maybe_event   # 产出的是统一事件,不是 Anthropic 事件

流结束后,StreamedResponse.get()(models/__init__.py:1765)用 _parts_manager 里累积的完整 parts 拼出一个和非流式同构的 ModelResponse——所以上层无论走流式还是非流式,拿到的最终对象长得一样。

3.7 provider 家族:OpenAI 兼容者一律复用一份实现

要解决的小问题: 25+ 家里,很大一部分其实是"OpenAI 兼容"接口(Ollama、LiteLLM、Together、OpenRouter、xAI、DeepSeek……)。难道每家都重写一遍映射?

不。infer_model() 里有一步"归一化家族"(models/__init__.py:2152-2155):凡是在 _OPENAI_CHAT_COMPATIBLE_PROVIDERS 集合里的,model_kind 一律改写成 'openai-chat';Google 系列(google-gla/gemini/google-vertex)则归并成 'google'。

# 摘自 __init__.py:infer_model,已精简
if model_kind in _OPENAI_CHAT_COMPATIBLE_PROVIDERS:
    model_kind = 'openai-chat'
elif model_kind in _GOOGLE_PROVIDERS:
    model_kind = 'google'

具体到类,Ollama、LiteLLM 这些"子类"几乎是空壳——只把默认 provider 改一下,全部映射逻辑继承自 OpenAIChatModel(models/ollama.py:23、models/litellm.py:23):

# 摘自 ollama.py,已精简 —— 一个近乎空壳的兼容子类
class OllamaModel(OpenAIChatModel):
    """Convenience wrapper around OpenAIChatModel, default provider='ollama'."""
    def __init__(self, model_name, *, provider='ollama', ...):
        super().__init__(model_name=model_name, provider=provider, ...)

所以真正"重量级"的映射实现只有少数几份(openai.py 3133 行、anthropic.py 1477 行、google.py 1411 行、bedrock.py 1173 行……),其余靠继承 + 换 provider 完成。分发表 + 家族归一 + 子类复用,是这一层能"支持 25+ 家却不失控"的根本原因。

3.8 分发表 infer_model:字符串到底怎么变成实例

要解决的小问题: "anthropic/claude-sonnet-4-5" 这一个字符串,怎么变出一个持有 Anthropic client、带 profile 的对象?

infer_model()(models/__init__.py:2064)是整条链路的入口,步骤:

1. 环境/任务覆盖 —— 若设了 LLM_MODEL_KEY 环境变量或 Celery 任务传了 bypass_llm_model,可全局改写要用的模型。
2. 归一化别名 —— normalize_model_id()(models/__init__.py:2008)把简写补全(如 bedrock/claude-3-5-sonnet:v2 → 完整 Bedrock ID);未知 provider(如本地 ollama/...)则原样透传。
3. 拆 provider/model —— 按 / 切分;没写前缀时按 gpt*/claude*/gemini* 猜 provider 并发弃用警告。
4. 建 Provider —— infer_provider(provider_name)(providers/__init__.py:172)造出 Provider 对象,它持有厂商 SDK client、base_url、能力 model_profile。
5. 家族归一 + 分发 —— 见 §3.7,最后一长串 if model_kind == ... 把控制权交给对应子类构造器:

# 摘自 __init__.py:infer_model,已精简的分发尾部
elif model_kind == 'anthropic':
    from .anthropic import AnthropicModel
    return AnthropicModel(model_name, provider=provider)
elif model_kind == 'bedrock':
    from .bedrock import BedrockConverseModel
    return BedrockConverseModel(model_name, provider=provider)
# ... 其余各家同理;未知则 raise UserError

注意每个分支都是惰性 import(from .anthropic import ... 写在分支内),这样用户只装了 anthropic 依赖时,不会因为 import bedrock 而报缺包。

3.9 包装器:WrapperModel 与 OTel 埋点

要解决的小问题: 想给"任意模型"统一加一层能力(埋点、日志、fallback),又不想改每个子类。

WrapperModel(models/wrapper.py:18)是个透明装饰器基类:它本身也是 Model,内部持有一个 wrapped 模型,所有方法都直接转调 wrapped,连 __getattr__ 都往下透传。它"自己什么都不做",只作基类。

InstrumentedModel(models/instrumented.py:371)就继承它,在转调前后插入 OpenTelemetry span:

# 摘自 instrumented.py:InstrumentedModel.request,已精简
async def request(self, messages, model_settings, model_request_parameters):
    prepared_settings, prepared_parameters = self.wrapped.prepare_request(...)
    with self._instrument(messages, prepared_settings, prepared_parameters) as finish:
        response = await self.wrapped.request(messages, model_settings, model_request_parameters)
        finish(response, prepared_parameters)   # 把 usage / finish_reason 写进 span
        return response

_instrument()(models/instrumented.py:422)开一个 chat {model_name} 的 CLIENT span,写入 gen_ai.operation.name、模型属性、工具定义、采样参数等符合 OTel gen_ai.* 语义约定的 attribute。流式版本同理,在流耗尽的 finally 里用 response_stream.get() 补记最终响应。因为走的是 WrapperModel 透明包装,任何 provider 的模型都能被无侵入地埋点。

3.10 模型选型:registry 元数据 + selector 打分

要解决的小问题: 用户不确定该用哪个模型时,能不能让框架按任务推荐一个?

这套是可选的辅助能力,和上面的执行链路解耦,由两块组成:

• model_registry.py —— 一张手工维护的模型元数据库。每个模型是一条 ModelMetadata(models/model_registry.py:97),记录能力标签、上下文窗口、基准分(MMLU/GPQA/HumanEval……)、cost_tier、speed_tier、擅长场景等。MODEL_REGISTRY 字典(models/model_registry.py:917)汇总所有条目,get_model_metadata() 按名查询。
• model_selector.py —— 两种选法。RuleBasedSelector(models/model_selector.py:80)不用 LLM:从任务描述里按关键词匹配出需要的能力,再对 registry 里每个模型 _score_model() 打分排序;LLMBasedSelector 则让一个 agent 直接推荐。统一入口是 select_model_async()(models/model_selector.py:600),产出一个 ModelRecommendation(含 model_name、reason、confidence_score、备选列表)。

Agent 上的门面是 recommend_model_for_task()(agent/agent.py:3492,异步版 :3415),它只是把 agent 的默认模型名当 fallback 传进 select_model_async,拿回推荐给用户参考——它只推荐,不自动切换,用不用由调用方决定。

3.11 多模态素材:DownloadedItem 与带 SSRF 防护的下载

要解决的小问题: 消息里常带图片/文件的 URL,发给模型前得先下载成字节。下载外部 URL 有安全风险。

DownloadedItem(models/__init__.py:2297)是个泛型 TypedDict,把"下载到的数据 + 数据类型"打包:

# 摘自 __init__.py
class DownloadedItem(TypedDict, Generic[DataT]):
    data: DataT       # bytes 或(base64/文本)str
    data_type: str    # 从 content-type 提取的 MIME 类型

下载走 download_item()(models/__init__.py:2326),可返回 bytes/base64/base64_uri/text 多种格式。它内建 SSRF(服务端请求伪造)防护:只允许 http/https、默认拦私网 IP、永久拦云元数据端点 169.254.169.254、下载前解析主机名以防 DNS rebinding;YouTube 视频链接直接拒下。各 provider 子类在映射含 FileUrl 的消息时调用它,把远端素材变成可发送的字节。

---

4. 巧妙之处(可借鉴的技术)

• 两类参数分离(settings vs. parameters)。 采样旋钮和"工具/输出"配置各走一条路,合并规则、生命周期都不同——settings 可跨调用叠加,parameters 每轮重建。清晰的关注点分离。见 models/settings.py:8 与 models/__init__.py:628。
• 结构化输出 = 一次工具调用。 不依赖模型原生 JSON 能力,把 Pydantic schema 包成 final_result 工具,退化成模型都会的工具调用。agent/agent.py:2261。
• profile 前置校验。 "不支持"在发请求前抛 UserError 给出人话,而非等厂商返回 400。models/__init__.py:829-835。
• 家族归一 + 空壳子类。 OpenAI 兼容的十几家共用一份 OpenAIChatModel,新增一家常常只是继承 + 改默认 provider。models/__init__.py:2152、models/ollama.py:23。
• 惰性 import 分发。 infer_model 每个分支在分支内 import,只装了一家依赖也能跑。models/__init__.py:2213-2254。
• 流式与非流式同构。 StreamedResponse.get() 用同一套 parts 拼出与非流式一致的 ModelResponse,上层消费逻辑不必分叉。models/__init__.py:1765。
• 异常与埋点的统一层。 厂商异常转 ModelHTTPError;WrapperModel 让 OTel 埋点对任何 provider 无侵入。models/anthropic.py:460、models/instrumented.py:371。

---

5. 边界与局限

• ModelSettings 只是公共子集。 厂商私有能力必须靠带前缀的子类 settings 承载;跨 provider fallback 时,私有 settings 对不认识它的模型无效(靠前缀避免误用,而非自动翻译)。
• model_registry 是手工维护的静态表。 基准分、cost/speed tier 都是人工填的快照,会随模型迭代过时;选型建议仅供参考,recommend_model_for_task 也只推荐不切换。
• 重量级映射仍是每家一份。 家族复用只覆盖 OpenAI/Google 兼容者;Anthropic、Bedrock、Cohere、Mistral 等仍各自维护上千行映射,新增一家不兼容 provider 的成本不低。
• count_tokens 非普遍支持。 基类默认抛错,只有实现了的模型(如 Anthropic)才支持"请求前数 token"的用量上限。
• request_stream 也是可选实现。 未实现流式的模型调用会抛 NotImplementedError。

---

6. 横向对比

模型层是 agent 框架的通用组件,同 shelf 的兄弟项目多半也有一层类似抽象。Upsonic 的取舍偏"pydantic-ai 血统":以 Model 抽象基类 + infer_model 字符串分发 + 每家一个子类为骨架,ModelResponse/ModelRequestParameters/StreamedResponse 这套类型命名也与 pydantic-ai 高度一致。相对轻量的框架常直接依赖 LiteLLM 之类的第三方统一层;Upsonic 则选择自持各家映射(同时保留 litellm 作为其中一个 provider),换来的是对请求/返回映射、异常统一、OTel 埋点的完全掌控。跨库原理对照见本 shelf 总库文档的"模型接入"一节。

---

7. 代码地图(导航索引)

主题	文件路径(相对 src/upsonic/)	符号名
抽象基类 + 三方法	models/__init__.py	Model、Model.request、Model.count_tokens、Model.request_stream
统一前处理/校验	models/__init__.py	Model.prepare_request、Model.customize_request_parameters
请求参数(工具/输出)	models/__init__.py	ModelRequestParameters
采样参数公共子集	models/settings.py	ModelSettings、merge_model_settings
流式统一抽象	models/__init__.py	StreamedResponse、StreamedResponse.get
字符串分发	models/__init__.py	infer_model、normalize_model_id、_OPENAI_CHAT_COMPATIBLE_PROVIDERS
多模态下载	models/__init__.py	DownloadedItem、download_item
Anthropic 真实映射	models/anthropic.py	AnthropicModel.request、_messages_create、_map_tool_definition、_process_response
Anthropic 流式	models/anthropic.py	AnthropicStreamedResponse._get_event_iterator
厂商私有 settings	models/anthropic.py	AnthropicModelSettings
OpenAI 兼容空壳子类	models/ollama.py、models/litellm.py	OllamaModel、LiteLLMModel
透明包装基类	models/wrapper.py	WrapperModel
OTel 埋点包装	models/instrumented.py	InstrumentedModel、InstrumentationSettings、instrument_model
模型元数据库	models/model_registry.py	ModelMetadata、MODEL_REGISTRY、get_model_metadata
任务选型	models/model_selector.py	RuleBasedSelector、select_model_async、ModelRecommendation
结构化输出组装	agent/agent.py	_build_model_request_parameters、_build_output_tools
选型门面	agent/agent.py	recommend_model_for_task、recommend_model_for_task_async
Provider 抽象	providers/__init__.py	Provider、infer_provider