LLM 适配层实战：一套代码无缝切换 Qwen / DeepSeek / OpenAI

前言：为什么需要 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 适配器必须实现的三个核心方法：

fromabcimportABC,abstractmethodfromtypingimportAsyncIterator,List,Dict,AnyclassBaseLLMAdapter(ABC):@abstractmethodasyncdefchat(self,messages:List[Dict],temperature:float=0.7,max_tokens:int=2048)->Dict[str,Any]:"""同步对话"""pass@abstractmethodasyncdefchat_stream(self,messages:List[Dict],temperature:float=0.7,max_tokens:int=2048)->AsyncIterator[str]:"""流式对话"""pass@abstractmethodasyncdefchat_with_tools(self,messages:List[Dict],tools:List[Dict],temperature:float=0.7,max_tokens:int=4096)->Dict[str,Any]:"""带工具调用的对话"""pass@staticmethoddefformat_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})returnmessages

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

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

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

importhttpxfromtenacityimportretry,stop_after_attempt,wait_exponentialclassOpenAIAdapter(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),)asyncdefclose(self):"""释放连接池资源"""awaitself.client.aclose()@retry(stop=stop_after_attempt(3),wait=wait_exponential(multiplier=1,min=1,max=10),)asyncdef_make_request(self,payload:Dict)->Dict:"""带重试机制的基础请求方法"""response=awaitself.client.post("/chat/completions",json=payload)response.raise_for_status()returnresponse.json()asyncdefchat(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,}returnawaitself._make_request(payload)asyncdefchat_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,}asyncforchunkinself._make_stream_request(payload):yieldchunkasyncdef_make_stream_request(self,payload:Dict)->AsyncIterator[str]:"""SSE 流解析"""payload["stream"]=Trueasyncwithself.client.stream("POST","/chat/completions",json=payload)asresponse:response.raise_for_status()asyncforlineinresponse.aiter_lines():ifline.startswith("data: "):data=line[6:]ifdata.strip()=="[DONE]":breakchunk=json.loads(data)delta=chunk.get("choices",[{}])[0].get("delta",{})content=delta.get("content","")ifcontent:yieldcontent

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

第三层：重试策略

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

fromtenacityimportretry,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)),)asyncdef_make_request(self,payload:Dict)->Dict:"""三次重试，间隔 1s → 2s → 4s"""response=awaitself.client.post("/chat/completions",json=payload)response.raise_for_status()returnresponse.json()

重试策略说明：

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

第四层：LLMConfig + 工厂函数

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

fromdataclassesimportdataclass@dataclassclassLLMConfig:provider:strapi_key:strbase_url:strmodel:strdefcreate_llm_adapter(config:LLMConfig)->BaseLLMAdapter:"""工厂函数：根据配置创建对应的适配器"""ifconfig.provider=="claude":returnClaudeAdapter(config)# 预留扩展returnOpenAIAdapter(config)

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

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

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

frompydantic_settingsimportBaseSettingsfromtypingimportLiteralclassSettings(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"defget_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]returnLLMConfig(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 → 使用 OpenAILLM_PROVIDER=qwen# Qwen 配置QWEN_API_KEY=your_qwen_api_key_hereQWEN_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1QWEN_MODEL=qwen-plus# DeepSeek 配置DEEPSEEK_API_KEY=your_deepseek_api_key_hereDEEPSEEK_BASE_URL=https://api.deepseek.com/v1DEEPSEEK_MODEL=deepseek-chat# OpenAI 配置OPENAI_API_KEY=your_openai_api_key_hereOPENAI_BASE_URL=https://api.openai.com/v1OPENAI_MODEL=gpt-4o-mini

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

Token 追踪实现

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

classTokenTracker:def__init__(self):self.total_usage={"prompt_tokens":0,"completion_tokens":0,"total_tokens":0,}defupdate(self,response:Dict[str,Any]):"""更新 token 统计"""usage=response.get("usage",{})forkeyinself.total_usage:self.total_usage[key]+=usage.get(key,0)defget_summary(self)->Dict[str,int]:"""获取汇总统计"""returnself.total_usage.copy()# 使用示例tracker=TokenTracker()foriterationinrange(max_iterations):response=awaitadapter.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. 调用 LLMtry:response=awaitadapter.chat(messages)print(response["choices"][0]["message"]["content"])finally:awaitadapter.close()# 确保释放资源

总结

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

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

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

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