与模型打交道 · 第 1 篇 | 预估阅读：12 分钟

4 个星期，4 个 LLM，47 次代码修改

小禾以为后端架构搞定了，可以安心写业务了。

直到老板开始"关心"技术选型。

第一周：

老板：“我们要用最好的！上 GPT-5.1！”

小禾屁颠屁颠地接入了 OpenAI：

from openai import OpenAI

client = OpenAI(api_key="sk-xxx")

def generate_story(prompt):
    response = client.chat.completions.create(
        model="gpt-5.1",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

效果确实好，账单也确实好看——一个月烧了两万块。

第二周：

老板看了账单：“换 Gemini 3.0 吧，Google 有免费额度。”

小禾开始改代码：

import google.generativeai as genai

genai.configure(api_key="xxx")
model = genai.GenerativeModel('gemini-3.0-pro')

def generate_story(prompt):
    response = model.generate_content(prompt)
    return response.text

API 完全不一样，消息格式不一样，响应结构也不一样。

小禾改了两天代码。

第三周：

老板：“数据安全很重要！我们用本地的 Ollama，跑 Qwen 模型。”

import requests

def generate_story(prompt):
    response = requests.post(
        "http://localhost:11434/api/generate",
        json={"model": "qwen2.5:32b", "prompt": prompt}
    )
    return response.json()["response"]

又是完全不同的接口。小禾又改了两天。

第四周：

客户说：“我们公司只能用 Claude，合规要求。”

import anthropic

client = anthropic.Anthropic(api_key="xxx")

def generate_story(prompt):
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.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.py
from abc import ABC, abstractmethod
from typing import List, Optional, Iterator
from dataclasses import dataclass

@dataclass
class Message:
    """统一的消息格式"""
    role: str  # "system", "user", "assistant"
    content: str

@dataclass
class GenerationConfig:
    """生成配置"""
    temperature: float = 0.7
    max_tokens: Optional[int] = None
    stop_sequences: Optional[List[str]] = None

class LLMAdapter(ABC):
    """LLM 适配器基类"""

    @abstractmethod
    def generate(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> str:
        """生成回复"""
        pass

    @abstractmethod
    def generate_stream(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> Iterator[str]:
        """流式生成"""
        pass

    @property
    @abstractmethod
    def model_name(self) -> str:
        """模型名称，用于日志和调试"""
        pass

    @property
    def supports_streaming(self) -> bool:
        """是否支持流式输出"""
        return True

接口很简单：

• Message：统一的消息格式，不管哪个 LLM 都用这个
• GenerationConfig：生成参数，温度、最大长度等
• generate：一次性生成
• generate_stream：流式生成

各平台的差异，由各自的适配器处理。

---

实现 OpenAI 适配器

# app/adapters/llm/openai_adapter.py
from openai import OpenAI
from typing import List, Optional, Iterator
from .base import LLMAdapter, Message, GenerationConfig

class OpenAIAdapter(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 = model

    def generate(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> str:
        config = config or GenerationConfig()

        # 转换为 OpenAI 的消息格式
        openai_messages = [
            {"role": m.role, "content": m.content}
            for m in messages
        ]

        response = self.client.chat.completions.create(
            model=self._model,
            messages=openai_messages,
            temperature=config.temperature,
            max_tokens=config.max_tokens,
            stop=config.stop_sequences
        )

        return response.choices[0].message.content

    def generate_stream(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> Iterator[str]:
        config = config or GenerationConfig()

        openai_messages = [
            {"role": m.role, "content": m.content}
            for m in messages
        ]

        stream = self.client.chat.completions.create(
            model=self._model,
            messages=openai_messages,
            temperature=config.temperature,
            stream=True
        )

        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

    @property
    def model_name(self) -> str:
        return f"openai/{self._model}"

OpenAI 的适配器最简单，因为我们的接口设计本来就参考了 OpenAI 的风格。

---

实现 Gemini 适配器

Gemini 的 API 风格不太一样，需要做转换：

# app/adapters/llm/gemini_adapter.py
import google.generativeai as genai
from typing import List, Optional, Iterator
from .base import LLMAdapter, Message, GenerationConfig

class GeminiAdapter(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)

    def generate(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> str:
        config = config or GenerationConfig()

        # 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
        )

        return response.text

    def generate_stream(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> Iterator[str]:
        config = config or GenerationConfig()
        gemini_messages = self._convert_messages(messages)

        response = self.model.generate_content(
            gemini_messages,
            generation_config=genai.GenerationConfig(
                temperature=config.temperature
            ),
            stream=True
        )

        for chunk in response:
            if chunk.text:
                yield chunk.text

    def _convert_messages(self, messages: List[Message]) -> List[dict]:
        """转换消息格式"""
        result = []
        system_content = ""

        for m in messages:
            if m.role == "system":
                system_content = m.content
            elif m.role == "user":
                content = m.content
                if system_content:
                    content = f"{system_content}\n\n{content}"
                    system_content = ""
                result.append({"role": "user", "parts": [content]})
            elif m.role == "assistant":
                result.append({"role": "model", "parts": [m.content]})

        return result

    @property
    def model_name(self) -> str:
        return f"gemini/{self._model_name}"

Gemini 的坑：

1. 没有 system role，要把 system 消息合并到 user 消息里
2. assistant 在 Gemini 里叫 model
3. 消息内容要放在 parts 数组里

这些差异都被适配器消化了，业务代码完全感知不到。

---

实现 Ollama 适配器

本地部署的 Ollama，用的是 REST API：

# app/adapters/llm/ollama_adapter.py
import requests
import json
from typing import List, Optional, Iterator
from .base import LLMAdapter, Message, GenerationConfig

class OllamaAdapter(LLMAdapter):
    """本地 Ollama 适配器"""

    def __init__(
        self,
        base_url: str = "http://localhost:11434",
        model: str = "qwen2.5:32b"
    ):
        self.base_url = base_url
        self._model = model

    def generate(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> str:
        config = config or GenerationConfig()

        ollama_messages = [
            {"role": m.role, "content": m.content}
            for m in messages
        ]

        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()

        return response.json()["message"]["content"]

    def generate_stream(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> Iterator[str]:
        config = config or GenerationConfig()

        ollama_messages = [
            {"role": m.role, "content": m.content}
            for m in messages
        ]

        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
        )

        for line in response.iter_lines():
            if line:
                data = json.loads(line)
                if "message" in data and "content" in data["message"]:
                    yield data["message"]["content"]

    @property
    def model_name(self) -> str:
        return f"ollama/{self._model}"

Ollama 的好处是消息格式和 OpenAI 兼容，转换比较简单。

---

实现 Claude 适配器

Claude 有自己的特色：

# app/adapters/llm/claude_adapter.py
import anthropic
from typing import List, Optional, Iterator
from .base import LLMAdapter, Message, GenerationConfig

class ClaudeAdapter(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 = model

    def generate(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> str:
        config = config or GenerationConfig()

        # Claude 的 system 消息要单独传
        system_msg = None
        claude_messages = []

        for m in messages:
            if m.role == "system":
                system_msg = m.content
            else:
                claude_messages.append({
                    "role": m.role,
                    "content": m.content
                })

        kwargs = {
            "model": self._model,
            "max_tokens": config.max_tokens or 4096,
            "messages": claude_messages,
        }

        if system_msg:
            kwargs["system"] = system_msg

        if config.temperature is not None:
            kwargs["temperature"] = config.temperature

        response = self.client.messages.create(**kwargs)

        return response.content[0].text

    def generate_stream(
        self,
        messages: List[Message],
        config: Optional[GenerationConfig] = None
    ) -> Iterator[str]:
        config = config or GenerationConfig()

        system_msg = None
        claude_messages = []

        for m in messages:
            if m.role == "system":
                system_msg = m.content
            else:
                claude_messages.append({
                    "role": m.role,
                    "content": m.content
                })

        kwargs = {
            "model": self._model,
            "max_tokens": config.max_tokens or 4096,
            "messages": claude_messages,
        }

        if system_msg:
            kwargs["system"] = system_msg

        with self.client.messages.stream(**kwargs) as stream:
            for text in stream.text_stream:
                yield text

    @property
    def model_name(self) -> str:
        return f"anthropic/{self._model}"

Claude 的坑：

1. system 消息要单独传，不能放在 messages 里
2. 必须指定 max_tokens
3. 流式输出的 API 不一样

---

工厂模式统一创建

现在有四个适配器了，需要一个统一的入口来创建：

# app/adapters/llm/factory.py
from typing import Dict, Type, Optional
from .base import LLMAdapter
from .openai_adapter import OpenAIAdapter
from .gemini_adapter import GeminiAdapter
from .ollama_adapter import OllamaAdapter
from .claude_adapter import ClaudeAdapter
from app.core.config import settings

class LLMFactory:
    """LLM 适配器工厂"""

    _adapters: Dict[str, Type[LLMAdapter]] = {
        "openai": OpenAIAdapter,
        "gemini": GeminiAdapter,
        "ollama": OllamaAdapter,
        "claude": ClaudeAdapter,
    }

    _instance: Optional[LLMAdapter] = None

    @classmethod
    def create(cls, adapter_type: str, **kwargs) -> LLMAdapter:
        """创建适配器实例"""
        if adapter_type not in cls._adapters:
            available = ", ".join(cls._adapters.keys())
            raise ValueError(
                f"Unknown adapter: {adapter_type}. "
                f"Available: {available}"
            )

        return cls._adapters[adapter_type](**kwargs)

    @classmethod
    def get_default(cls) -> LLMAdapter:
        """获取默认适配器（单例）"""
        if cls._instance is None:
            cls._instance = cls._create_from_settings()
        return cls._instance

    @classmethod
    def _create_from_settings(cls) -> LLMAdapter:
        """从配置创建适配器"""
        llm_type = settings.LLM_TYPE

        if llm_type == "openai":
            return cls.create(
                "openai",
                api_key=settings.OPENAI_API_KEY,
                model=settings.OPENAI_MODEL
            )
        elif llm_type == "gemini":
            return cls.create(
                "gemini",
                api_key=settings.GEMINI_API_KEY,
                model=settings.GEMINI_MODEL
            )
        elif llm_type == "ollama":
            return cls.create(
                "ollama",
                base_url=settings.OLLAMA_URL,
                model=settings.OLLAMA_MODEL
            )
        elif llm_type == "claude":
            return cls.create(
                "claude",
                api_key=settings.ANTHROPIC_API_KEY,
                model=settings.CLAUDE_MODEL
            )
        else:
            raise ValueError(f"Unknown LLM type: {llm_type}")

    @classmethod
    def register(cls, name: str, adapter_class: Type[LLMAdapter]):
        """注册新适配器"""
        cls._adapters[name] = adapter_class

    @classmethod
    def reset(cls):
        """重置单例（测试用）"""
        cls._instance = None

---

业务代码怎么写？

现在业务代码变得无比简洁：

# app/services/story_generator.py
from app.adapters.llm.factory import LLMFactory
from app.adapters.llm.base import Message, GenerationConfig

def generate_story(user_prompt: str) -> str:
    """生成故事"""
    llm = LLMFactory.get_default()

    messages = [
        Message(
            role="system",
            content="你是一个专业的故事创作者，擅长写引人入胜的短故事。"
        ),
        Message(role="user", content=user_prompt)
    ]

    return llm.generate(messages)


def generate_story_stream(user_prompt: str):
    """流式生成故事"""
    llm = LLMFactory.get_default()

    messages = [
        Message(
            role="system",
            content="你是一个专业的故事创作者，擅长写引人入胜的短故事。"
        ),
        Message(role="user", content=user_prompt)
    ]

    for chunk in llm.generate_stream(messages):
        yield chunk

注意看：业务代码里没有任何 OpenAI、Gemini、Claude 的影子。

它只知道有一个 llm，可以 generate。

用的是 GPT-5.1 还是本地 Qwen？业务代码不关心，也不需要关心。

---

切换模型：只改配置

现在老板说要换模型，小禾只需要：

# .env 文件

# 用 GPT-5.1
LLM_TYPE=openai
OPENAI_API_KEY=sk-xxx
OPENAI_MODEL=gpt-5.1

# 换成 Gemini 3.0
LLM_TYPE=gemini
GEMINI_API_KEY=xxx
GEMINI_MODEL=gemini-3.0-pro

# 换成本地 Ollama
LLM_TYPE=ollama
OLLAMA_URL=http://localhost:11434
OLLAMA_MODEL=qwen2.5:32b

# 换成 Claude
LLM_TYPE=claude
ANTHROPIC_API_KEY=xxx
CLAUDE_MODEL=claude-sonnet-4-20250514

改一行配置，重启服务，完事。

业务代码？一行不改。

---

加个新模型要多久？

后来老板说要支持某个客户自己的私有模型。

小禾花了半小时写了个新适配器：

# app/adapters/llm/custom_adapter.py
class CustomLLMAdapter(LLMAdapter):
    """客户私有模型适配器"""

    def __init__(self, endpoint: str, api_key: str):
        self.endpoint = endpoint
        self.api_key = api_key

    def generate(self, messages, config=None):
        # 调用客户的 API
        response = requests.post(
            self.endpoint,
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"messages": [{"role": m.role, "content": m.content} for m in messages]}
        )
        return response.json()["result"]

    # ... 其他方法

然后注册一下：

LLMFactory.register("custom", CustomLLMAdapter)

配置文件加一行：

LLM_TYPE=custom

搞定。

---

复盘总结

小禾算了笔账：

指标	改造前	改造后
切换 LLM 改动量	47 处	1 行配置
切换 LLM 耗时	2 天	2 分钟
新增 LLM 耗时	2 天	30 分钟
业务代码耦合	强耦合	零耦合
单元测试难度	困难	简单（可 mock）

老板再也不能用"换个模型"来折腾他了。

---

小禾的感悟

变化是永恒的，
代码要为变化而设计。

今天是 GPT，
明天是 Gemini，
后天是什么？
谁也不知道。

但有了适配器，
我不再害怕。

业务代码只知道接口，
不知道实现，
这就是解耦的力量。

抽象不是过度设计，
是对未来的保险。

当老板说"换个模型"时，
我终于可以微笑着说：
"好的，稍等两分钟。"

小禾关掉 IDE，心情舒畅。

以后不管换多少次模型，他都不怕了。

---

下一篇预告：显存爆了，服务挂了，半夜被叫起来

GPU 资源管理，不是加显存就能解决的。

敬请期待。