AI 后端开发 · 第 1 篇 | 预估阅读:12 分钟
4 个星期,4 个 LLM,47 次代码修改
小禾以为后端架构搞定了,可以安心写业务了。
直到老板开始"关心"技术选型。
第一周:
老板:“我们要用最好的!上 GPT-5.1!”
小禾屁颠屁颠地接入了 OpenAI:
fromopenaiimportOpenAI client=OpenAI(api_key="sk-xxx")defgenerate_story(prompt):response=client.chat.completions.create(model="gpt-5.1",messages=[{"role":"user","content":prompt}])returnresponse.choices[0].message.content效果确实好,账单也确实好看——一个月烧了两万块。
第二周:
老板看了账单:“换 Gemini 3.0 吧,Google 有免费额度。”
小禾开始改代码:
importgoogle.generativeaiasgenai genai.configure(api_key="xxx")model=genai.GenerativeModel('gemini-3.0-pro')defgenerate_story(prompt):response=model.generate_content(prompt)returnresponse.textAPI 完全不一样,消息格式不一样,响应结构也不一样。
小禾改了两天代码。
第三周:
老板:“数据安全很重要!我们用本地的 Ollama,跑 Qwen 模型。”
importrequestsdefgenerate_story(prompt):response=requests.post("http://localhost:11434/api/generate",json={"model":"qwen2.5:32b","prompt":prompt})returnresponse.json()["response"]又是完全不同的接口。小禾又改了两天。
第四周:
客户说:“我们公司只能用 Claude,合规要求。”
importanthropic client=anthropic.Anthropic(api_key="xxx")defgenerate_story(prompt):response=client.messages.create(model="claude-sonnet-4-20250514",max_tokens=4096,messages=[{"role":"user","content":prompt}])returnresponse.content[0].text小禾崩溃了。
4 个星期,4 个 LLM,业务代码改了 47 处。
每次改完还要回归测试,生怕哪里漏了。
“这日子没法过了。”
问题出在哪?
小禾冷静下来分析,发现问题的根源是:业务代码和 LLM 实现强耦合。
业务代码里到处都是:
# 生成故事response=client.chat.completions.create(...)# 生成分镜response=client.chat.completions.create(...)# 生成角色描述response=client.chat.completions.create(...)# 生成画面提示词response=client.chat.completions.create(...)换一次 LLM,这些地方全要改。
小禾想起了之前学过的设计模式:适配器模式。
如果在业务代码和 LLM 之间加一层抽象,是不是就能解决问题?
设计统一抽象层
小禾画了张新的架构图:
业务代码只依赖抽象接口,不关心具体用哪个 LLM。
切换 LLM?换个适配器就行,业务代码一行不改。
定义统一接口
首先,定义统一的消息格式和生成接口:
# app/adapters/llm/base.pyfromabcimportABC,abstractmethodfromtypingimportList,Optional,Iteratorfromdataclassesimportdataclass@dataclassclassMessage:"""统一的消息格式"""role:str# "system", "user", "assistant"content:str@dataclassclassGenerationConfig:"""生成配置"""temperature:float=0.7max_tokens:Optional[int]=Nonestop_sequences:Optional[List[str]]=NoneclassLLMAdapter(ABC):"""LLM 适配器基类"""@abstractmethoddefgenerate(self,messages:List[Message],config:Optional[GenerationConfig]=None)->str:"""生成回复"""pass@abstractmethoddefgenerate_stream(self,messages:List[Message],config:Optional[GenerationConfig]=None)->Iterator[str]:"""流式生成"""pass@property@abstractmethoddefmodel_name(self)->str:"""模型名称,用于日志和调试"""pass@propertydefsupports_streaming(self)->bool:"""是否支持流式输出"""returnTrue接口很简单:
Message:统一的消息格式,不管哪个 LLM 都用这个GenerationConfig:生成参数,温度、最大长度等generate:一次性生成generate_stream:流式生成
各平台的差异,由各自的适配器处理。
实现 OpenAI 适配器
# app/adapters/llm/openai_adapter.pyfromopenaiimportOpenAIfromtypingimportList,Optional,Iteratorfrom.baseimportLLMAdapter,Message,GenerationConfigclassOpenAIAdapter(LLMAdapter):"""OpenAI GPT 系列适配器"""def__init__(self,api_key:str,model:str="gpt-5.1",base_url:Optional[str]=None):self.client=OpenAI(api_key=api_key,base_url=base_url)self._model=modeldefgenerate(self,messages:List[Message],config:Optional[GenerationConfig]=None)->str:config=configorGenerationConfig()# 转换为 OpenAI 的消息格式openai_messages=[{"role":m.role,"content":m.content}forminmessages]response=self.client.chat.completions.create(model=self._model,messages=openai_messages,temperature=config.temperature,max_tokens=config.max_tokens,stop=config.stop_sequences)returnresponse.choices[0].message.contentdefgenerate_stream(self,messages:List[Message],config:Optional[GenerationConfig]=None)->Iterator[str]:config=configorGenerationConfig()openai_messages=[{"role":m.role,"content":m.content}forminmessages]stream=self.client.chat.completions.create(model=self._model,messages=openai_messages,temperature=config.temperature,stream=True)forchunkinstream:ifchunk.choices[0].delta.content:yieldchunk.choices[0].delta.content@propertydefmodel_name(self)->str:returnf"openai/{self._model}"OpenAI 的适配器最简单,因为我们的接口设计本来就参考了 OpenAI 的风格。
实现 Gemini 适配器
Gemini 的 API 风格不太一样,需要做转换:
# app/adapters/llm/gemini_adapter.pyimportgoogle.generativeaiasgenaifromtypingimportList,Optional,Iteratorfrom.baseimportLLMAdapter,Message,GenerationConfigclassGeminiAdapter(LLMAdapter):"""Google Gemini 适配器"""def__init__(self,api_key:str,model:str="gemini-3.0-pro"):genai.configure(api_key=api_key)self._model_name=model self.model=genai.GenerativeModel(model)defgenerate(self,messages:List[Message],config:Optional[GenerationConfig]=None)->str:config=configorGenerationConfig()# Gemini 的消息格式不同# 需要把 system 消息合并到第一条 user 消息gemini_messages=self._convert_messages(messages)generation_config=genai.GenerationConfig(temperature=config.temperature,max_output_tokens=config.max_tokens,stop_sequences=config.stop_sequences)response=self.model.generate_content(gemini_messages,generation_config=generation_config)returnresponse.textdefgenerate_stream(self,messages:List[Message],config:Optional[GenerationConfig]=None)->Iterator[str]:config=configorGenerationConfig()gemini_messages=self._convert_messages(messages)response=self.model.generate_content(gemini_messages,generation_config=genai.GenerationConfig(temperature=config.temperature),stream=True)forchunkinresponse:ifchunk.text:yieldchunk.textdef_convert_messages(self,messages:List[Message])->List[dict]:"""转换消息格式"""result=[]system_content=""forminmessages:ifm.role=="system":system_content=m.contentelifm.role=="user":content=m.contentifsystem_content:content=f"{system_content}\n\n{content}"system_content=""result.append({"role":"user","parts":[content]})elifm.role=="assistant":result.append({"role":"model","parts":[m.content]})returnresult@propertydefmodel_name(self)->str:returnf"gemini/{self._model_name}"Gemini 的坑:
- 没有 system role,要把 system 消息合并到 user 消息里
- assistant 在 Gemini 里叫 model
- 消息内容要放在 parts 数组里
这些差异都被适配器消化了,业务代码完全感知不到。
实现 Ollama 适配器
本地部署的 Ollama,用的是 REST API:
# app/adapters/llm/ollama_adapter.pyimportrequestsimportjsonfromtypingimportList,Optional,Iteratorfrom.baseimportLLMAdapter,Message,GenerationConfigclassOllamaAdapter(LLMAdapter):"""本地 Ollama 适配器"""def__init__(self,base_url:str="http://localhost:11434",model:str="qwen2.5:32b"):self.base_url=base_url self._model=modeldefgenerate(self,messages:List[Message],config:Optional[GenerationConfig]=None)->str:config=configorGenerationConfig()ollama_messages=[{"role":m.role,"content":m.content}forminmessages]response=requests.post(f"{self.base_url}/api/chat",json={"model":self._model,"messages":ollama_messages,"options":{"temperature":config.temperature,"num_predict":config.max_tokens},"stream":False},timeout=300# 本地模型可能比较慢)response.raise_for_status()returnresponse.json()["message"]["content"]defgenerate_stream(self,messages:List[Message],config:Optional[GenerationConfig]=None)->Iterator[str]:config=configorGenerationConfig()ollama_messages=[{"role":m.role,"content":m.content}forminmessages]response=requests.post(f"{self.base_url}/api/chat",json={"model":self._model,"messages":ollama_messages,"options":{"temperature":config.temperature},"stream":True},stream=True,timeout=300)forlineinresponse.iter_lines():ifline:data=json.loads(line)if"message"indataand"content"indata["message"]:yielddata["message"]["content"]@propertydefmodel_name(self)->str:returnf"ollama/{self._model}"Ollama 的好处是消息格式和 OpenAI 兼容,转换比较简单。
实现 Claude 适配器
Claude 有自己的特色:
# app/adapters/llm/claude_adapter.pyimportanthropicfromtypingimportList,Optional,Iteratorfrom.baseimportLLMAdapter,Message,GenerationConfigclassClaudeAdapter(LLMAdapter):"""Anthropic Claude 适配器"""def__init__(self,api_key:str,model:str="claude-sonnet-4-20250514"):self.client=anthropic.Anthropic(api_key=api_key)self._model=modeldefgenerate(self,messages:List[Message],config:Optional[GenerationConfig]=None)->str:config=configorGenerationConfig()# Claude 的 system 消息要单独传system_msg=Noneclaude_messages=[]forminmessages:ifm.role=="system":system_msg=m.contentelse:claude_messages.append({"role":m.role,"content":m.content})kwargs={"model":self._model,"max_tokens":config.max_tokensor4096,"messages":claude_messages,}ifsystem_msg:kwargs["system"]=system_msgifconfig.temperatureisnotNone:kwargs["temperature"]=config.temperature response=self.client.messages.create(**kwargs)returnresponse.content[0].textdefgenerate_stream(self,messages:List[Message],config:Optional[GenerationConfig]=None)->Iterator[str]:config=configorGenerationConfig()system_msg=Noneclaude_messages=[]forminmessages:ifm.role=="system":system_msg=m.contentelse:claude_messages.append({"role":m.role,"content":m.content})kwargs={"model":self._model,"max_tokens":config.max_tokensor4096,"messages":claude_messages,}ifsystem_msg:kwargs["system"]=system_msgwithself.client.messages.stream(**kwargs)asstream:fortextinstream.text_stream:yieldtext@propertydefmodel_name(self)->str:returnf"anthropic/{self._model}"Claude 的坑:
- system 消息要单独传,不能放在 messages 里
- 必须指定 max_tokens
- 流式输出的 API 不一样
工厂模式统一创建
现在有四个适配器了,需要一个统一的入口来创建:
# app/adapters/llm/factory.pyfromtypingimportDict,Type,Optionalfrom.baseimportLLMAdapterfrom.openai_adapterimportOpenAIAdapterfrom.gemini_adapterimportGeminiAdapterfrom.ollama_adapterimportOllamaAdapterfrom.claude_adapterimportClaudeAdapterfromapp.core.configimportsettingsclassLLMFactory:"""LLM 适配器工厂"""_adapters:Dict[str,Type[LLMAdapter]]={"openai":OpenAIAdapter,"gemini":GeminiAdapter,"ollama":OllamaAdapter,"claude":ClaudeAdapter,}_instance:Optional[LLMAdapter]=None@classmethoddefcreate(cls,adapter_type:str,**kwargs)->LLMAdapter:"""创建适配器实例"""ifadapter_typenotincls._adapters:available=", ".join(cls._adapters.keys())raiseValueError(f"Unknown adapter:{adapter_type}. "f"Available:{available}")returncls._adapters[adapter_type](**kwargs)@classmethoddefget_default(cls)->LLMAdapter:"""获取默认适配器(单例)"""ifcls._instanceisNone:cls._instance=cls._create_from_settings()returncls._instance@classmethoddef_create_from_settings(cls)->LLMAdapter:"""从配置创建适配器"""llm_type=settings.LLM_TYPEifllm_type=="openai":returncls.create("openai",api_key=settings.OPENAI_API_KEY,model=settings.OPENAI_MODEL)elifllm_type=="gemini":returncls.create("gemini",api_key=settings.GEMINI_API_KEY,model=settings.GEMINI_MODEL)elifllm_type=="ollama":returncls.create("ollama",base_url=settings.OLLAMA_URL,model=settings.OLLAMA_MODEL)elifllm_type=="claude":returncls.create("claude",api_key=settings.ANTHROPIC_API_KEY,model=settings.CLAUDE_MODEL)else:raiseValueError(f"Unknown LLM type:{llm_type}")@classmethoddefregister(cls,name:str,adapter_class:Type[LLMAdapter]):"""注册新适配器"""cls._adapters[name]=adapter_class@classmethoddefreset(cls):"""重置单例(测试用)"""cls._instance=None业务代码怎么写?
现在业务代码变得无比简洁:
# app/services/story_generator.pyfromapp.adapters.llm.factoryimportLLMFactoryfromapp.adapters.llm.baseimportMessage,GenerationConfigdefgenerate_story(user_prompt:str)->str:"""生成故事"""llm=LLMFactory.get_default()messages=[Message(role="system",content="你是一个专业的故事创作者,擅长写引人入胜的短故事。"),Message(role="user",content=user_prompt)]returnllm.generate(messages)defgenerate_story_stream(user_prompt:str):"""流式生成故事"""llm=LLMFactory.get_default()messages=[Message(role="system",content="你是一个专业的故事创作者,擅长写引人入胜的短故事。"),Message(role="user",content=user_prompt)]forchunkinllm.generate_stream(messages):yieldchunk注意看:业务代码里没有任何 OpenAI、Gemini、Claude 的影子。
它只知道有一个llm,可以generate。
用的是 GPT-5.1 还是本地 Qwen?业务代码不关心,也不需要关心。
切换模型:只改配置
现在老板说要换模型,小禾只需要:
# .env 文件# 用 GPT-5.1LLM_TYPE=openaiOPENAI_API_KEY=sk-xxxOPENAI_MODEL=gpt-5.1# 换成 Gemini 3.0LLM_TYPE=geminiGEMINI_API_KEY=xxxGEMINI_MODEL=gemini-3.0-pro# 换成本地 OllamaLLM_TYPE=ollamaOLLAMA_URL=http://localhost:11434OLLAMA_MODEL=qwen2.5:32b# 换成 ClaudeLLM_TYPE=claudeANTHROPIC_API_KEY=xxxCLAUDE_MODEL=claude-sonnet-4-20250514改一行配置,重启服务,完事。
业务代码?一行不改。
加个新模型要多久?
后来老板说要支持某个客户自己的私有模型。
小禾花了半小时写了个新适配器:
# app/adapters/llm/custom_adapter.pyclassCustomLLMAdapter(LLMAdapter):"""客户私有模型适配器"""def__init__(self,endpoint:str,api_key:str):self.endpoint=endpoint self.api_key=api_keydefgenerate(self,messages,config=None):# 调用客户的 APIresponse=requests.post(self.endpoint,headers={"Authorization":f"Bearer{self.api_key}"},json={"messages":[{"role":m.role,"content":m.content}forminmessages]})returnresponse.json()["result"]# ... 其他方法然后注册一下:
LLMFactory.register("custom",CustomLLMAdapter)配置文件加一行:
LLM_TYPE=custom搞定。
复盘总结
小禾算了笔账:
| 指标 | 改造前 | 改造后 |
|---|---|---|
| 切换 LLM 改动量 | 47 处 | 1 行配置 |
| 切换 LLM 耗时 | 2 天 | 2 分钟 |
| 新增 LLM 耗时 | 2 天 | 30 分钟 |
| 业务代码耦合 | 强耦合 | 零耦合 |
| 单元测试难度 | 困难 | 简单(可 mock) |
老板再也不能用"换个模型"来折腾他了。
小禾的感悟
变化是永恒的, 代码要为变化而设计。 今天是 GPT, 明天是 Gemini, 后天是什么? 谁也不知道。 但有了适配器, 我不再害怕。 业务代码只知道接口, 不知道实现, 这就是解耦的力量。 抽象不是过度设计, 是对未来的保险。 当老板说"换个模型"时, 我终于可以微笑着说: "好的,稍等两分钟。"小禾关掉 IDE,心情舒畅。
以后不管换多少次模型,他都不怕了。
下一篇预告:显存爆了,服务挂了,半夜被叫起来
GPU 资源管理,不是加显存就能解决的。
敬请期待。