前言：为什么需要 LLM 适配层

在实际项目中，我们经常需要同时对接多个大模型提供商：阿里百炼的 Qwen、DeepSeek、OpenAI。虽然它们都提供 OpenAI-compatible 的 API 接口，但 Base URL、Model ID 以及一些细微的差异（比如 Qwen 对 system role 的支持、DeepSeek 的 token 计数方式）使得直接硬编码调用存在很大问题。

核心痛点：每次切换模型都需要修改业务代码，耦合度高、维护成本大。

本文的目标是：通过适配层设计，实现只改一行 .env 配置即可切换 LLM 提供商，业务代码零修改。

整体架构设计

适配层采用五层结构，层层解耦：

Settings (.env) → LLMConfig (dataclass) → create_llm_adapter (工厂) → OpenAIAdapter (实现) → BaseLLMAdapter (接口)

下面逐层深入讲解。

第一层：BaseLLMAdapter — 定义接口契约

作为抽象基类，它定义了所有 LLM 适配器必须实现的三个核心方法：

from abc import ABC, abstractmethod
from typing import AsyncIterator, List, Dict, Any

class BaseLLMAdapter(ABC):
    @abstractmethod
    async def chat(self, messages: List[Dict], temperature: float = 0.7, max_tokens: int = 2048) -> Dict[str, Any]:
        """同步对话"""
        pass
    
    @abstractmethod
    async def chat_stream(self, messages: List[Dict], temperature: float = 0.7, max_tokens: int = 2048) -> AsyncIterator[str]:
        """流式对话"""
        pass
    
    @abstractmethod
    async def chat_with_tools(self, messages: List[Dict], tools: List[Dict], temperature: float = 0.7, max_tokens: int = 4096) -> Dict[str, Any]:
        """带工具调用的对话"""
        pass
    
    @staticmethod
    def format_messages(system_prompt: str, history: List[Dict], user_message: str) -> List[Dict]:
        """格式化消息，所有适配器共用"""
        messages = [{"role": "system", "content": system_prompt}]
        messages.extend(history)
        messages.append({"role": "user", "content": user_message})
        return messages

为什么用 ABC 而不是 Protocol？
因为这是给开发者使用的基类，需要强制子类实现所有抽象方法。使用 ABC 后，如果子类漏写某个方法，IDE 会直接报错，避免运行时才发现问题。

第二层：OpenAIAdapter — 一个适配器通吃三家

利用 Qwen、DeepSeek、OpenAI 都兼容 OpenAI API 格式的特性，我们只需要一个适配器，通过配置不同的 base_url 和 api_key 即可：

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class OpenAIAdapter(BaseLLMAdapter):
    def __init__(self, config: "LLMConfig"):
        self.config = config
        self.client = httpx.AsyncClient(
            base_url=config.base_url,
            headers={"Authorization": f"Bearer {config.api_key}"},
            timeout=httpx.Timeout(120.0, connect=10.0),
        )
    
    async def close(self):
        """释放连接池资源"""
        await self.client.aclose()
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
    )
    async def _make_request(self, payload: Dict) -> Dict:
        """带重试机制的基础请求方法"""
        response = await self.client.post("/chat/completions", json=payload)
        response.raise_for_status()
        return response.json()
    
    async def chat(self, messages: List[Dict], temperature: float = 0.7, max_tokens: int = 2048) -> Dict[str, Any]:
        payload = {
            "model": self.config.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }
        return await self._make_request(payload)
    
    async def chat_stream(self, messages: List[Dict], temperature: float = 0.7, max_tokens: int = 2048) -> AsyncIterator[str]:
        """流式输出实现"""
        payload = {
            "model": self.config.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": True,
        }
        async for chunk in self._make_stream_request(payload):
            yield chunk
    
    async def _make_stream_request(self, payload: Dict) -> AsyncIterator[str]:
        """SSE 流解析"""
        payload["stream"] = True
        async with self.client.stream("POST", "/chat/completions", json=payload) as response:
            response.raise_for_status()
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    data = line[6:]
                    if data.strip() == "[DONE]":
                        break
                    chunk = json.loads(data)
                    delta = chunk.get("choices", [{}])[0].get("delta", {})
                    content = delta.get("content", "")
                    if content:
                        yield content

关键工程细节：httpx 连接生命周期管理
AsyncClient 在 __init__ 中创建，必须在 close() 中释放。每个 Agent 请求结束后，通过 try/finally 确保连接池被正确关闭，防止泄漏。

第三层：重试策略

LLM API 调用经常会遇到网络超时、429 限流、服务端 500 等问题。使用 tenacity 库实现指数退避重试：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((httpx.TimeoutException, httpx.HTTPStatusError)),
)
async def _make_request(self, payload: Dict) -> Dict:
    """三次重试，间隔 1s → 2s → 4s"""
    response = await self.client.post("/chat/completions", json=payload)
    response.raise_for_status()
    return response.json()

重试策略说明：

  • 最多重试 3 次
  • 等待时间呈指数增长：1秒 → 2秒 → 4秒
  • 只对超时和 HTTP 错误进行重试
  • 三次都失败后向上层抛出异常，由调用方处理

第四层：LLMConfig + 工厂函数

配置数据类和工厂函数，实现配置到适配器的转换：

from dataclasses import dataclass

@dataclass
class LLMConfig:
    provider: str
    api_key: str
    base_url: str
    model: str

def create_llm_adapter(config: LLMConfig) -> BaseLLMAdapter:
    """工厂函数：根据配置创建对应的适配器"""
    if config.provider == "claude":
        return ClaudeAdapter(config)  # 预留扩展
    return OpenAIAdapter(config)

工厂模式的价值：
即使目前只有 OpenAIAdapter，也要保留工厂函数。当未来需要接入不兼容 OpenAI 格式的 Provider（如 Claude 的 SDK），只需在工厂函数中添加判断，调用方完全不需要修改。

第五层：Settings — .env 驱动切换

使用 Pydantic Settings 从环境变量读取配置：

from pydantic_settings import BaseSettings
from typing import Literal

class Settings(BaseSettings):
    # 当前使用的 LLM 提供商
    llm_provider: Literal["qwen", "openai", "deepseek"] = "qwen"
    
    # Qwen 配置
    qwen_api_key: str = ""
    qwen_base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1"
    qwen_model: str = "qwen-plus"
    
    # DeepSeek 配置
    deepseek_api_key: str = ""
    deepseek_base_url: str = "https://api.deepseek.com/v1"
    deepseek_model: str = "deepseek-chat"
    
    # OpenAI 配置
    openai_api_key: str = ""
    openai_base_url: str = "https://api.openai.com/v1"
    openai_model: str = "gpt-4o-mini"
    
    def get_llm_config(self) -> LLMConfig:
        """根据当前 provider 返回对应配置"""
        provider_map = {
            "qwen": {
                "api_key": self.qwen_api_key,
                "base_url": self.qwen_base_url,
                "model": self.qwen_model,
            },
            "deepseek": {
                "api_key": self.deepseek_api_key,
                "base_url": self.deepseek_base_url,
                "model": self.deepseek_model,
            },
            "openai": {
                "api_key": self.openai_api_key,
                "base_url": self.openai_base_url,
                "model": self.openai_model,
            },
        }
        config = provider_map[self.llm_provider]
        return LLMConfig(
            provider=self.llm_provider,
            api_key=config["api_key"],
            base_url=config["base_url"],
            model=config["model"],
        )

# 使用示例
settings = Settings()
llm_config = settings.get_llm_config()
adapter = create_llm_adapter(llm_config)

.env 文件配置示例：

# LLM_PROVIDER=qwen   → 使用阿里百炼 Qwen
# LLM_PROVIDER=deepseek → 使用 DeepSeek
# LLM_PROVIDER=openai → 使用 OpenAI
LLM_PROVIDER=qwen

# Qwen 配置
QWEN_API_KEY=your_qwen_api_key_here
QWEN_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
QWEN_MODEL=qwen-plus

# DeepSeek 配置
DEEPSEEK_API_KEY=your_deepseek_api_key_here
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1
DEEPSEEK_MODEL=deepseek-chat

# OpenAI 配置
OPENAI_API_KEY=your_openai_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini

切换模型时，只需修改 .env 文件中的 LLM_PROVIDER 一行，业务代码完全不需要改动。

Token 追踪实现

在 Agent 循环中累加每次调用的 token 消耗：

class TokenTracker:
    def __init__(self):
        self.total_usage = {
            "prompt_tokens": 0,
            "completion_tokens": 0,
            "total_tokens": 0,
        }
    
    def update(self, response: Dict[str, Any]):
        """更新 token 统计"""
        usage = response.get("usage", {})
        for key in self.total_usage:
            self.total_usage[key] += usage.get(key, 0)
    
    def get_summary(self) -> Dict[str, int]:
        """获取汇总统计"""
        return self.total_usage.copy()

# 使用示例
tracker = TokenTracker()
for iteration in range(max_iterations):
    response = await adapter.chat_with_tools(messages, tools)
    tracker.update(response)
    
print(f"总消耗 tokens: {tracker.total_usage['total_tokens']}")

完整使用流程

# 1. 加载配置
settings = Settings()
llm_config = settings.get_llm_config()

# 2. 创建适配器
adapter = create_llm_adapter(llm_config)

# 3. 格式化消息
messages = BaseLLMAdapter.format_messages(
    system_prompt="你是一个智能助手",
    history=[],
    user_message="你好，请介绍一下自己"
)

# 4. 调用 LLM
try:
    response = await adapter.chat(messages)
    print(response["choices"][0]["message"]["content"])
finally:
    await adapter.close()  # 确保释放资源

总结

LLM 适配层总共约 250 行 Python 代码，实现了以下功能：

  1. 抽象基类：定义统一接口，强制子类实现
  2. 单一适配器：利用 API 兼容性，一个适配器通吃三家
  3. 指数退避重试：优雅处理网络异常和限流
  4. SSE 流解析：支持流式输出
  5. Token 追踪：精确统计每次对话的 token 消耗

核心设计模式：适配器模式 + 工厂模式。适配器封装 API 差异，工厂隔离创建逻辑。最终效果：改一行 .env 配置，业务代码零修改。

下一篇将深入讲解 ReAct Agent 循环的手写实现：Think → Act → Observe 的完整流程。

# python # 人工智能
Logo
DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

cover

半小时上手蓝耘元生代 MaaS：用 OpenAI SDK 调用 DeepSeek，把大模型接进自己的项目

avatar DeepSeek技术社区
cover

GEO生成式引擎优化：AI时代的内容战略重构

avatar DeepSeek技术社区
cover

从零开始做一个属于自己的 Skill（保姆级教程）

avatar DeepSeek技术社区