[!info] related notes

• 所属 MOC: AI Agent Application MOC
• 相关: Model Gateway, Model Router, Fallback Strategy
• 实践: 多模型 LLM Provider 抽象层, OpenAI 兼容 LLM Provider

LLM Provider Abstraction

一句话定义

LLM Provider Abstraction 是通过统一接口对接多个 LLM 提供商的抽象层。它让上层代码不需要关心底层用的是 OpenAI、Anthropic 还是本地模型，只需要调用统一的 chat() 方法。

它解决什么问题

直接对接每个 LLM Provider 的 API 会导致：

• 代码和某个 Provider 强耦合
• 切换模型需要改大量代码
• 无法实现 Fallback（一个 Provider 挂了无法切换到另一个）
• 无法统一追踪 token 用量和成本

核心原理

统一接口

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class ChatMessage:
    role: str  # system, user, assistant, tool
    content: str
    tool_calls: list = None
    tool_call_id: str = None

@dataclass
class ChatResponse:
    content: str
    tool_calls: list = None
    finish_reason: str = None
    usage: dict = None  # {input_tokens, output_tokens}

class LLMProvider(ABC):
    @abstractmethod
    async def chat(
        self,
        messages: list[ChatMessage],
        model: str,
        tools: list = None,
        temperature: float = 0.7,
        max_tokens: int = 4096,
        stream: bool = False,
    ) -> ChatResponse:
        ...

    @abstractmethod
    async def chat_stream(
        self,
        messages: list[ChatMessage],
        model: str,
        **kwargs,
    ) -> AsyncIterator[str]:
        ...

Provider 实现

class OpenAIProvider(LLMProvider):
    def __init__(self, api_key: str):
        self.client = AsyncOpenAI(api_key=api_key)

    async def chat(self, messages, model, tools=None, **kwargs):
        response = await self.client.chat.completions.create(
            model=model,
            messages=[asdict(m) for m in messages],
            tools=tools,
            **kwargs,
        )
        return ChatResponse(
            content=response.choices[0].message.content,
            tool_calls=response.choices[0].message.tool_calls,
            usage={
                "input_tokens": response.usage.prompt_tokens,
                "output_tokens": response.usage.completion_tokens,
            },
        )

class AnthropicProvider(LLMProvider):
    def __init__(self, api_key: str):
        self.client = AsyncAnthropic(api_key=api_key)

    async def chat(self, messages, model, tools=None, **kwargs):
        # Anthropic 的 API 格式和 OpenAI 不同
        system_msg = [m for m in messages if m.role == "system"]
        other_msgs = [m for m in messages if m.role != "system"]

        response = await self.client.messages.create(
            model=model,
            system=system_msg[0].content if system_msg else "",
            messages=[asdict(m) for m in other_msgs],
            tools=tools,
            **kwargs,
        )
        return ChatResponse(
            content=response.content[0].text,
            usage={
                "input_tokens": response.usage.input_tokens,
                "output_tokens": response.usage.output_tokens,
            },
        )

Provider Factory

class LLMProviderFactory:
    def __init__(self):
        self._providers = {}

    def register(self, name: str, provider: LLMProvider):
        self._providers[name] = provider

    def get(self, name: str) -> LLMProvider:
        return self._providers[name]

# 注册
factory = LLMProviderFactory()
factory.register("openai", OpenAIProvider(api_key="..."))
factory.register("anthropic", AnthropicProvider(api_key="..."))

常见坑

1. 接口不统一: OpenAI 和 Anthropic 的 tool_call 格式不同，需要归一化
2. 流式响应格式不同: 每个 Provider 的 SSE 格式有差异
3. 不做 Fallback: 一个 Provider 挂了整个服务不可用
4. 不做 token 归一化: 不同 Provider 的 token 计算方式不同

参考资料

• OpenAI API
• Anthropic API
• OpenAI 兼容 LLM Provider