LLM Provider 抽象层设计
如何设计一个可切换多模型的 LLM 调用抽象层:Protocol 接口定义、统一消息格式、流式累积封装、环境变量切换。
#type / concept
#status / growing
#tech / ai
[!info] related notes
- 前置概念: Function Calling
- 流式细节: Function Calling 流式累积
- 上层应用: 咨询 Agent 工作流
LLM Provider 抽象层设计
核心问题
AI 产品需要在不同模型间切换:开发时用便宜模型,生产时用强模型,某个服务商挂了时快速切换。如果每个调用点都写死 model="gpt-4o",切换模型要改所有代码。
设计目标
- 一处配置,全局生效 — 环境变量切换模型,代码零修改
- 上层不感知模型细节 — 统一的 chat 和 chat_stream 接口
- 流式累积封装在内部 — 上层不用关心 tool_call 的 chunk 拼接
- 可测试 — 通过 Protocol 接口注入 mock
接口设计
class LLMProvider(Protocol):
async def chat(
self,
messages: list[ChatMessage],
tools: list[ToolDefinition] | None = None,
temperature: float = 0.7,
max_tokens: int = 2048,
) -> ChatMessage: ...
async def chat_stream(
self,
messages: list[ChatMessage],
tools: list[ToolDefinition] | None = None,
temperature: float = 0.7,
max_tokens: int = 2048,
) -> AsyncIterator[StreamChunk]: ...
用 Protocol 而不是 ABC:不需要显式 implements,只要方法签名匹配就行。
统一数据结构
@dataclass
class ChatMessage:
role: str # "system" | "user" | "assistant" | "tool"
content: str | None
tool_calls: list[ToolCall] | None
tool_call_id: str | None # tool 结果的关联 ID
@dataclass
class StreamChunk:
delta: str = "" # 文本增量
tool_calls: list[ToolCall] | None = None # 流结束时才填充完整 tool_calls
finished: bool = False
关键设计:StreamChunk 把 tool_calls 的累积逻辑封装在 Provider 内部。上层只需要在 finished=True 时拿到完整的 tool_calls,不需要自己拼 chunk。
为什么选 OpenAI 兼容 API
几乎所有主流模型都兼容 OpenAI API 格式:
| 服务商 | base_url | 说明 |
|---|---|---|
| OpenAI | 默认 | 原生 |
| DeepSeek | https://api.deepseek.com/v1 | 完全兼容 |
| 阿里通义 | https://dashscope.aliyuncs.com/compatible-mode/v1 | 兼容 |
| OpenRouter | https://openrouter.ai/api/v1 | 聚合多模型 |
只需要一个 OpenAICompatibleProvider,通过 base_url 切换。
环境变量配置
LLM_MODEL=gpt-4o-mini
LLM_API_KEY=sk-xxx
LLM_BASE_URL=https://api.deepseek.com/v1 # 切换模型只改这一行
单例 vs 依赖注入
# 全局单例 — 生产环境
_default_provider: OpenAICompatibleProvider | None = None
def get_llm_provider() -> OpenAICompatibleProvider:
global _default_provider
if _default_provider is None:
_default_provider = OpenAICompatibleProvider()
return _default_provider
单例的好处是避免重复创建 HTTP 连接。测试时可以通过依赖注入传 mock。
与 LangChain 的区别
| 维度 | 自建 Provider | LangChain |
|---|---|---|
| 复杂度 | ~200 行 | 框架级 |
| 依赖 | 只有 openai SDK | 大量传递依赖 |
| 灵活性 | 完全可控 | 受框架约束 |
| 适用 | 接口简单、模型少 | 需要 chain/agent/memory |
如果只需要”发消息、收回复、调工具”,自建就够了。如果需要复杂的 chain 编排、memory 管理、多 agent 协作,考虑 LangChain 或 LangGraph。
常见错误
硬编码模型名
# ❌ 每个调用点写死模型
response = await client.chat.completions.create(model="gpt-4o", ...)
# ✅ 通过 Provider 统一管理
provider = get_llm_provider()
response = await provider.chat(messages, tools=tools)
API Key 泄露
永远不要把 API Key 硬编码在代码里。用环境变量或 secret manager。
不处理模型不支持 function calling
不是所有模型都支持 function calling。Provider 应该在 tools 不为 None 时检查模型能力,不支持时降级为纯文本提示。