前言为什么需要 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:float0.7,max_tokens:int2048)-Dict[str,Any]:同步对话passabstractmethodasyncdefchat_stream(self,messages:List[Dict],temperature:float0.7,max_tokens:int2048)-AsyncIterator[str]:流式对话passabstractmethodasyncdefchat_with_tools(self,messages:List[Dict],tools:List[Dict],temperature:float0.7,max_tokens:int4096)-Dict[str,Any]:带工具调用的对话passstaticmethoddefformat_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.configconfig self.clienthttpx.AsyncClient(base_urlconfig.base_url,headers{Authorization:fBearer{config.api_key}},timeouthttpx.Timeout(120.0,connect10.0),)asyncdefclose(self):释放连接池资源awaitself.client.aclose()retry(stopstop_after_attempt(3),waitwait_exponential(multiplier1,min1,max10),)asyncdef_make_request(self,payload:Dict)-Dict:带重试机制的基础请求方法responseawaitself.client.post(/chat/completions,jsonpayload)response.raise_for_status()returnresponse.json()asyncdefchat(self,messages:List[Dict],temperature:float0.7,max_tokens:int2048)-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:float0.7,max_tokens:int2048)-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,jsonpayload)asresponse:response.raise_for_status()asyncforlineinresponse.aiter_lines():ifline.startswith(data: ):dataline[6:]ifdata.strip()[DONE]:breakchunkjson.loads(data)deltachunk.get(choices,[{}])[0].get(delta,{})contentdelta.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_typeretry(stopstop_after_attempt(3),waitwait_exponential(multiplier1,min1,max10),retryretry_if_exception_type((httpx.TimeoutException,httpx.HTTPStatusError)),)asyncdef_make_request(self,payload:Dict)-Dict:三次重试间隔 1s → 2s → 4sresponseawaitself.client.post(/chat/completions,jsonpayload)response.raise_for_status()returnresponse.json()重试策略说明最多重试 3 次等待时间呈指数增长1秒 → 2秒 → 4秒只对超时和 HTTP 错误进行重试三次都失败后向上层抛出异常由调用方处理第四层LLMConfig 工厂函数配置数据类和工厂函数实现配置到适配器的转换fromdataclassesimportdataclassdataclassclassLLMConfig:provider:strapi_key:strbase_url:strmodel:strdefcreate_llm_adapter(config:LLMConfig)-BaseLLMAdapter:工厂函数根据配置创建对应的适配器ifconfig.providerclaude: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:strqwen_base_url:strhttps://dashscope.aliyuncs.com/compatible-mode/v1qwen_model:strqwen-plus# DeepSeek 配置deepseek_api_key:strdeepseek_base_url:strhttps://api.deepseek.com/v1deepseek_model:strdeepseek-chat# OpenAI 配置openai_api_key:stropenai_base_url:strhttps://api.openai.com/v1openai_model:strgpt-4o-minidefget_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,},}configprovider_map[self.llm_provider]returnLLMConfig(providerself.llm_provider,api_keyconfig[api_key],base_urlconfig[base_url],modelconfig[model],)# 使用示例settingsSettings()llm_configsettings.get_llm_config()adaptercreate_llm_adapter(llm_config).env 文件配置示例# LLM_PROVIDERqwen → 使用阿里百炼 Qwen# LLM_PROVIDERdeepseek → 使用 DeepSeek# LLM_PROVIDERopenai → 使用 OpenAILLM_PROVIDERqwen# Qwen 配置QWEN_API_KEYyour_qwen_api_key_hereQWEN_BASE_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1QWEN_MODELqwen-plus# DeepSeek 配置DEEPSEEK_API_KEYyour_deepseek_api_key_hereDEEPSEEK_BASE_URLhttps://api.deepseek.com/v1DEEPSEEK_MODELdeepseek-chat# OpenAI 配置OPENAI_API_KEYyour_openai_api_key_hereOPENAI_BASE_URLhttps://api.openai.com/v1OPENAI_MODELgpt-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 统计usageresponse.get(usage,{})forkeyinself.total_usage:self.total_usage[key]usage.get(key,0)defget_summary(self)-Dict[str,int]:获取汇总统计returnself.total_usage.copy()# 使用示例trackerTokenTracker()foriterationinrange(max_iterations):responseawaitadapter.chat_with_tools(messages,tools)tracker.update(response)print(f总消耗 tokens:{tracker.total_usage[total_tokens]})完整使用流程# 1. 加载配置settingsSettings()llm_configsettings.get_llm_config()# 2. 创建适配器adaptercreate_llm_adapter(llm_config)# 3. 格式化消息messagesBaseLLMAdapter.format_messages(system_prompt你是一个智能助手,history[],user_message你好请介绍一下自己)# 4. 调用 LLMtry:responseawaitadapter.chat(messages)print(response[choices][0][message][content])finally:awaitadapter.close()# 确保释放资源总结LLM 适配层总共约 250 行 Python 代码实现了以下功能抽象基类定义统一接口强制子类实现单一适配器利用 API 兼容性一个适配器通吃三家指数退避重试优雅处理网络异常和限流SSE 流解析支持流式输出Token 追踪精确统计每次对话的 token 消耗核心设计模式适配器模式 工厂模式。适配器封装 API 差异工厂隔离创建逻辑。最终效果改一行 .env 配置业务代码零修改。下一篇将深入讲解 ReAct Agent 循环的手写实现Think → Act → Observe 的完整流程。