AI 创业营收 89% 被两家吞掉——开发者该怎么面对这个双寡头现实

# ai_gateway.py — 最小可运行的多提供商抽象层
# 依赖：pip install openai anthropic

from openai import OpenAI
from anthropic import Anthropic
from dataclasses import dataclass
from typing import Optional

@dataclass
class LLMResponse:
    content: str
    model: str
    provider: str
    input_tokens: int
    output_tokens: int

class LLMGateway:
    """统一调用入口，隔离提供商差异"""

    PROVIDERS = {
        "openai": {
            "client_init": lambda: OpenAI(),          # 自动读取 OPENAI_API_KEY
            "default_model": "gpt-4o",
        },
        "anthropic": {
            "client_init": lambda: Anthropic(),       # 自动读取 ANTHROPIC_API_KEY
            "default_model": "claude-sonnet-4-20250514",
        },
    }

    def __init__(self, provider: str = "openai", model: Optional[str] = None):
        if provider not in self.PROVIDERS:
            raise ValueError(f"不支持的提供商: {provider}，可选: {list(self.PROVIDERS.keys())}")
        cfg = self.PROVIDERS[provider]
        self.provider = provider
        self.client = cfg["client_init"]()
        self.model = model or cfg["default_model"]

    def chat(self, system: str, user: str, max_tokens: int = 1024) -> LLMResponse:
        if self.provider == "openai":
            return self._call_openai(system, user, max_tokens)
        elif self.provider == "anthropic":
            return self._call_anthropic(system, user, max_tokens)

    def _call_openai(self, system: str, user: str, max_tokens: int) -> LLMResponse:
        resp = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": system},
                {"role": "user", "content": user},
            ],
            max_tokens=max_tokens,
        )
        choice = resp.choices[0]
        usage = resp.usage
        return LLMResponse(
            content=choice.message.content,
            model=resp.model,
            provider="openai",
            input_tokens=usage.prompt_tokens,
            output_tokens=usage.completion_tokens,
        )

    def _call_anthropic(self, system: str, user: str, max_tokens: int) -> LLMResponse:
        resp = self.client.messages.create(
            model=self.model,
            max_tokens=max_tokens,
            system=system,
            messages=[{"role": "user", "content": user}],
        )
        block = resp.content[0]
        usage = resp.usage
        return LLMResponse(
            content=block.text,
            model=resp.model,
            provider="anthropic",
            input_tokens=usage.input_tokens,
            output_tokens=usage.output_tokens,
        )

    def switch(self, provider: str, model: Optional[str] = None):
        """热切换提供商，无需重建上层逻辑"""
        new_cfg = self.PROVIDERS[provider]
        self.provider = provider
        self.client = new_cfg["client_init"]()
        self.model = model or new_cfg["default_model"]


# ===== 使用示例 =====
if __name__ == "__main__":
    # 默认走 OpenAI
    gw = LLMGateway(provider="openai")
    r1 = gw.chat(
        system="你是一位简洁的技术顾问",
        user="用一句话解释什么是 token",
    )
    print(f"[{r1.provider}/{r1.model}] {r1.content}")
    print(f"  tokens: in={r1.input_tokens}, out={r1.output_tokens}")

    # 切到 Anthropic，同一套调用代码
    gw.switch("anthropic")
    r2 = gw.chat(
        system="你是一位简洁的技术顾问",
        user="用一句话解释什么是 token",
    )
    print(f"[{r2.provider}/{r2.model}] {r2.content}")
    print(f"  tokens: in={r2.input_tokens}, out={r2.output_tokens}")

# .env 或直接 export
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."

pip install openai anthropic
python ai_gateway.py

# 一个最小评估流程的伪命令
python run_eval.py \
  --dataset ./eval_set.jsonl \
  --providers openai,anthropic \
  --models gpt-4o,claude-sonnet-4-20250514 \
  --metrics accuracy,latency,cost_per_1k_tokens \
  --output ./eval_results.csv

# fallback_config.yaml — 生产环境的提供商切换规则
default_provider: openai
rules:
  - trigger: cost_threshold
    provider: openai
    daily_limit_usd: 500
    fallback: anthropic
  - trigger: error_rate
    provider: openai
    threshold_pct: 5
    window_minutes: 10
    fallback: anthropic
  - trigger: latency
    provider: openai
    p99_ms: 5000
    fallback: anthropic