本文是「成都硅基边界」零代码构建平台旗下 Silicon-AI 智能体引擎实战复盘系列第七篇。前面聊过智能体、RAG、MCP、工作流引擎与数据库，这一篇讲平台的地基之一——多 LLM 抽象层：Qwen、DeepSeek、Doubao、GLM、OpenAI 五家模型，业务代码如何做到"换模型只换配置、不改一行代码"。文中代码均为示意代码（非项目真实源码），按生产级 Python 规范书写。
一、背景："接模型"为什么会越接越乱
接一家新模型，表面看只是换个 API 地址，实际要处理一堆"隐形差异"：
- 思考模式协议不同：同是"开启深度思考"，Qwen 是
enable_thinking，GLM/DeepSeek/Doubao 是extra_body.thinking，OpenAI 是reasoning_effort； - 推理内容格式不同：思考过程的增量藏在
delta.reasoning_content，部分 SDK 根本不解析； - 模型上下架频繁：用户历史配置里的模型名，三个月后可能已下架；
- 多模态参数各异：同样是"传一段音频"，有的要显式
format字段，有的不用； - 密钥安全：API Key 绝不能明文落库；
- 计量计费：每家价格结构不同（输入/输出差价、计价单位），流式调用还要想办法拿到 usage。
这些差异如果散落在业务代码里，每接一家模型就要改一遍所有调用点。Silicon-AI 的解法是建一层抽象，把差异全部收编。整体结构如下：
flowchart TD
BIZ[业务代码<br/>智能体 / 工作流 / RAG] --> FAC[LLMComponent 统一门面<br/>bind_tools / invoke / astream / 结构化输出]
FAC --> F{providerFormat 工厂}
F --> Q[ChatQwen<br/>阿里百炼]
F --> D[ChatDeepSeek<br/>DeepSeek 官方]
F --> DB[ChatDoubao<br/>火山方舟]
F --> G[ChatGLM<br/>智谱 BigModel]
F --> O[ChatOpenAI 兜底<br/>任意 OpenAI 兼容厂商]
Q & D & DB & G & O --> CB[LLMUsageCallback<br/>统一计量计费]
CB --> DB2[(用量 / 费用落库)]
二、亮点一：统一 OpenAI 兼容底座 + 工厂装配
五家模型全部基于 OpenAI 兼容协议接入：Qwen 走百炼兼容模式、GLM 走智谱兼容模式、Doubao 走火山方舟（原生兼容 OpenAI SDK）、DeepSeek 用官方 ChatDeepSeek。工厂按 providerFormat 装配，业务侧面对的是同一个 LLMComponent 门面：
"""llm_factory.py —— 模型工厂：按 providerFormat 装配统一的对话模型实例。"""
from __future__ import annotations
from collections.abc import Sequence
from typing import Any
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_openai import ChatOpenAI
from model.enums.llm_provider_enum import LLMProviderEnum
def build_chat_model(
llm: "LLMChat",
callbacks: Sequence[BaseCallbackHandler],
) -> BaseChatModel:
"""按厂商格式装配对话模型实例。
Args:
llm: 已解密凭据的模型配置（model / base_url / api_key / temperature 等）。
callbacks: 生命周期回调（计量计费等），对所有厂商统一挂载。
Returns:
可直接用于 invoke / astream / bind_tools 的对话模型实例。
Raises:
ValueError: model 或 api_key 为空时抛出。
"""
common: dict[str, Any] = {
"model": llm.model,
"base_url": llm.base_url,
"api_key": llm.api_key,
"temperature": llm.temperature,
"timeout": llm.timeout,
"streaming": llm.streaming,
"stream_usage": True, # 流式调用也要返回 usage，否则 token 统计恒为 0
"callbacks": callbacks,
}
match llm.format:
case LLMProviderEnum.QWEN:
return ChatQwen(**common, enable_thinking=llm.enable_thinking)
case LLMProviderEnum.GLM:
return ChatGLM(**common, enable_thinking=llm.enable_thinking)
case LLMProviderEnum.DEEPSEEK | LLMProviderEnum.DOUBAO:
return ChatDeepSeek(**common, extra_body=thinking_body(llm)) \
if llm.format is LLMProviderEnum.DEEPSEEK \
else ChatDoubao(**common, extra_body=thinking_body(llm))
case _:
# 兜底：任意 OpenAI 兼容厂商，配置 base_url 即插即用
return ChatOpenAI(**common)
注意最后的兜底分支：凡是 OpenAI 兼容的新厂商，不用写任何适配代码，配置 base_url 即插即用。"支持 N 家模型"的成本，从 O(N) 次业务改造降为 O(1) 次配置。
三、亮点二：一个开关，五种"思考"协议
"深度思考"这个业务开关，在各家模型上的协议写法完全不同。抽象层把它收敛为一个 enable_thinking 字段，由适配器统一翻译：
| 厂商 | 开启思考的写法 |
|---|---|
| Qwen（思考模型） | enable_thinking=True，可配 thinking_budget |
| GLM | extra_body={"thinking": {"type": "enabled"}} |
| DeepSeek | extra_body={"thinking": {"type": "enabled"/"disabled"}} |
| Doubao | extra_body={"thinking": {"type": "enabled"/"disabled"}} |
| OpenAI | reasoning_effort="high" |
这里藏着一个真实踩过的坑：部分思考模型默认开启 thinking，而思考开启时 tool_choice 等结构化输出能力受限——智能体一旦绑定工具就报错，且报错信息完全不指向根因。解法是显式传 enabled/disabled，绝不依赖服务端默认值：
def thinking_body(llm: "LLMChat") -> dict[str, Any]:
"""将统一的 enable_thinking 开关翻译为 DeepSeek/Doubao 私有协议。
必须显式传入 enabled/disabled：部分思考模型默认开启 thinking，
会限制 tool_choice 等结构化输出能力，表现为"绑定工具后调用失败"，
且报错信息与真实原因相距甚远——绝不依赖服务端默认值。
"""
mode = "enabled" if llm.enable_thinking else "disabled"
return {
"thinking": {
"type": mode}}
四、亮点三：reasoning_content 解析补丁，推理/正文分离
深度思考模型的"思考过程"以 delta.reasoning_content 增量返回，而所用版本的 langchain_openai 父类不解析这个字段——思考内容会直接丢失。各适配子类统一做了流式 chunk 解析的 override，把思考内容归并进 additional_kwargs["reasoning_content"]：
class ChatQwen(ChatOpenAI):
"""Qwen 对话模型（百炼 OpenAI 兼容模式），补齐 reasoning_content 解析。"""
enable_thinking: bool | None = Field(default=None)
thinking_budget: int | None = Field(default=None)
@staticmethod
def merge_reasoning(
chunk: AIMessageChunk,
delta: "ChoiceDelta",
) -> AIMessageChunk:
"""将流式增量中的 reasoning_content 归并进 additional_kwargs。
父类（langchain_openai 1.1.6）不解析 delta.reasoning_content，
思考内容会在流式路径上丢失；此处显式归并，供前端分离渲染
「思考过程 / 正式回答」，并供回调层测量思考耗时。
"""
if (reasoning := getattr(delta, "reasoning_content", None)) is None:
return chunk
merged = chunk.additional_kwargs.get("reasoning_content", "") + reasoning
return chunk.model_copy(
update={
"additional_kwargs": {
**chunk.additional_kwargs,
"reasoning_content": merged}}
)
收益立竿见影：前端可以把"思考过程"和"正式回答"分开渲染（打字机先出思考、再出正文），回调层还能据此测量思考耗时；同一套分离逻辑在 Qwen / GLM / Doubao 三家复用。
五、亮点四：下架模型自动路由，旧配置不断裂
模型厂商上下架频繁，用户三个月前保存的"qwen-plus"可能已经下架，直接报错会把用户旧数据变成"死配置"。适配层用不可变映射表 + model_validator 在实例化前自动路由：
from types import MappingProxyType
from typing import Final
_DEPRECATED_MODELS: Final[Mapping[str, str]] = MappingProxyType({
"qwen-plus": "qwen3-plus",
"qwen-flash": "qwen3-flash",
"qwq-plus": "qwq-32b",
})
class ChatQwen(ChatOpenAI):
@model_validator(mode="before")
@classmethod
def _route_deprecated_model(cls, values: dict[str, Any]) -> dict[str, Any]:
"""实例化前将已下架模型名路由至官方替代型号，保证旧配置持续可用。"""
if isinstance(values, dict) and (model := values.get("model")) in _DEPRECATED_MODELS:
replacement = _DEPRECATED_MODELS[model]
logger.warning("对话模型 %s 已下架，自动路由至 %s", model, replacement)
values = {
**values, "model": replacement}
return values
两个专业细节：映射表用 MappingProxyType 包成只读，防止运行时被误改；用 model_validator(mode="before") 在字段校验前拦截，替换发生在任何请求发出之前。下架不等于断裂：旧配置自动升级到替代模型，只在日志里留痕。
六、亮点五：多模态 content blocks 转换，抹平各家参数差异
多模态输入（图片、音频、视频、文档）在各家协议里同样不对齐。适配层统一做 content blocks 转换：
- 图片：多数厂商
image_url原生支持，直接透传； - 音频：火山方舟要求显式
format字段——从 URL 路径后缀推断，推断不出时兜底mp3，并转成 OpenAI 标准input_audio格式； - 文档：不支持文档输入的厂商（如火山方舟 Chat API 不收 PDF）直接不进 content blocks，引导走文件解析工具。
from pathlib import PurePosixPath
from urllib.parse import urlparse
_AUDIO_FORMAT_BY_EXT: Final[Mapping[str, str]] = MappingProxyType({
".mp3": "mp3", ".wav": "wav", ".m4a": "m4a",
".aac": "aac", ".flac": "flac", ".ogg": "ogg", ".opus": "opus",
})
_DEFAULT_AUDIO_FORMAT: Final[str] = "mp3"
def guess_audio_format(url: str) -> str:
"""从 URL 路径后缀推断音频格式。
火山方舟 Chat API 的音频输入必须显式携带 format 字段；
后缀缺失或不可识别时回退 mp3，避免请求被服务端直接拒绝。
"""
suffix = PurePosixPath(urlparse(url).path).suffix.lower()
return _AUDIO_FORMAT_BY_EXT.get(suffix, _DEFAULT_AUDIO_FORMAT)
多模态差异和思考模式一样，全部消化在适配器内部——业务代码只管把"一段音频"扔进来。
七、亮点六：回调式计量计费，一次挂载全局生效
抽象层挂了一个 LLMUsageCallback（LangChain 回调处理器），把计量计费从业务代码里彻底剥离：
from dataclasses import dataclass
from decimal import Decimal
@dataclass(slots=True, frozen=True)
class UsageSnapshot:
"""一次模型调用的用量快照。"""
prompt_tokens: int = 0
completion_tokens: int = 0
elapsed_ms: int = 0
payment: Decimal = Decimal("0")
class LLMUsageCallback(BaseCallbackHandler):
"""模型调用生命周期回调：统一完成计量、计费与落库。
在工厂装配阶段挂载，业务代码零感知；
on_llm_error 同样落库（费用记零），保证账单闭环可对账。
"""
def __init__(self, llm_chat: "LLMChat", balance: "BalanceConsumePO | None" = None) -> None:
self._llm = llm_chat
self._balance = balance
self._t0: float | None = None
def on_llm_start(self, serialized, prompts, **kwargs) -> None:
self._t0 = time.monotonic()
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
usage: dict = (response.llm_output or {
}).get("token_usage") or {
}
in_tok = int(usage.get("prompt_tokens") or 0)
out_tok = int(usage.get("completion_tokens") or 0)
in_price, out_price = self._llm.resolve_price(in_tok) # 输入/输出差价
payment = ceil_particle(
Decimal(in_tok) * Decimal(str(in_price))
+ Decimal(out_tok) * Decimal(str(out_price))
)
elapsed_ms = int((time.monotonic() - self._t0) * 1000) if self._t0 else 0
self._persist(UsageSnapshot(in_tok, out_tok, elapsed_ms, payment))
def on_llm_error(self, error: BaseException, **kwargs) -> None:
self._persist(UsageSnapshot()) # 失败调用也落库，费用记零
def _persist(self, snapshot: UsageSnapshot) -> None:
... # 写入 LLMChatStatPO 并累加 balance_consume，此处省略
几个细节值得强调：stream_usage=True 保证流式调用也能拿到 usage（否则流式场景 token 统计永远是 0）；输入/输出差价与计价单位换算都在 resolve_price 一处收敛；失败的调用也要落库（费用记零），否则账单对不上；快照用 frozen dataclass，计量数据一旦生成不可篡改。这一层挂上后，智能体、工作流、RAG 所有调用点零改动自动具备计量能力。
八、亮点七：密钥加密落库 + 结构化输出 + 小模型分层
还有三处容易被忽略的工程细节：
- API Key 密文落库：模型服务商的 Key 加密存储，使用时才解密注入——数据库拖走也不泄露密钥；
- 结构化输出：门面提供
invoke_structured_output(schema)，支持从"输出变量定义"递归构建 Pydantic 模型（含嵌套子结构），工作流 LLM 节点的"多输出变量"就靠它实现：
_PYTHON_TYPES: Final[Mapping[str, type]] = MappingProxyType({
"string": str, "integer": int, "number": float, "boolean": bool,
})
def build_output_model(variables: Sequence["OutputVariable"]) -> type[BaseModel]:
"""由输出变量定义递归构建 Pydantic 模型（支持嵌套对象与列表）。
Args:
variables: 工作流节点上声明的输出变量（可能带 children 子结构）。
Returns:
可直接传给 with_structured_output 的动态模型类型。
"""
fields: dict[str, tuple[Any, FieldInfo]] = {
}
for var in variables:
if var.children:
child = build_output_model(var.children) # 嵌套 → 递归建子模型
annotation: Any = list[child] if var.is_list else child
else:
annotation = _PYTHON_TYPES[var.type]
fields[var.name] = (annotation,
Field(description=var.description or var.name))
return create_model("WorkflowOutput",
__config__=ConfigDict(strict=False), **fields)
- 小模型分层：问题重写、推荐问题、意图分类这类轻任务统一走
ChatSmallLLM（小模型），重任务才用旗舰模型——按任务分级选模型，是成本优化的第一杠杆。
九、结语：抽象层的价值是"业务零感知"
回看这层抽象做了什么：协议差异（思考模式、reasoning、多模态）收进适配器，生命周期事件（上下架、密钥、计量计费）收进回调和校验器，业务代码只面对一个门面。换模型 = 改一条配置；接新厂商 = 写一个继承 ChatOpenAI 的适配类；计费策略调整 = 只改回调一处。
如果你也要做多模型平台，建议按这个优先级投入：先统一底座（OpenAI 兼容 + 工厂），再收编计量计费（回调挂载），最后打磨体验细节（思考分离、下架路由、多模态转换）。模型市场变化越快，这层抽象的复利就越大。
