Stripe斥资75亿美元收购OpenRouter：AI经济的基础设施之战

传统模式（碎片化）：
开发者 ----> OpenAI API (独立API Key + 独立计费)
       ----> Anthropic API (独立API Key + 独立计费)
       ----> Google Gemini API (独立API Key + 独立计费)
       ----> Meta Llama API (独立API Key + 独立计费)
       ----> Mistral API (独立API Key + 独立计费)
       ----> Cohere API (独立API Key + 独立计费)

OpenRouter模式（统一网关）：
开发者 ----> OpenRouter API (统一API Key + 统一计费)
                 |--> OpenAI
                 |--> Anthropic
                 |--> Google Gemini
                 |--> Meta Llama
                 |--> Mistral
                 |--> Cohere

# OpenRouter API 使用示例
import requests
import json

class OpenRouterClient:
    '''
    OpenRouter 统一AI模型网关客户端
    展示其核心价值：统一接口 + 智能路由 + 成本聚合
    '''

    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://openrouter.ai/api/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "HTTP-Referer": "https://example.com",
            "X-Title": "My AI App"
        }
        self.usage_log = []

    def chat(self, messages, model="auto", **kwargs):
        '''
        统一的聊天接口
        model="auto" 时自动选择最佳模型
        '''
        if model == "auto":
            model = self._select_best_model(messages)
            print(f"  [路由] 自动选择模型: {model}")

        payload = {
            "model": model,
            "messages": messages,
            **kwargs
        }

        resp = requests.post(
            f"{self.base_url}/chat/completions",
            json=payload,
            headers=self.headers
        )

        data = resp.json()

        # 记录使用情况和成本
        usage = data.get("usage", {})
        self.usage_log.append({
            "model": model,
            "prompt_tokens": usage.get("prompt_tokens", 0),
            "completion_tokens": usage.get("completion_tokens", 0),
            "total_tokens": usage.get("total_tokens", 0),
            "cost": self._calculate_cost(model, usage)
        })

        return data

    def _select_best_model(self, messages):
        '''
        智能路由：根据任务类型选择模型
        '''
        last_message = messages[-1]["content"].lower() if messages else ""

        if "code" in last_message or "function" in last_message:
            return "anthropic/claude-3.5-sonnet"
        elif "image" in last_message:
            return "openai/gpt-4o"
        elif len(last_message) > 5000:
            return "google/gemini-1.5-pro"
        else:
            return "openai/gpt-4o-mini"

    def _calculate_cost(self, model, usage):
        '''计算调用成本'''
        pricing = {
            "openai/gpt-4o": {"input": 0.005, "output": 0.015},
            "openai/gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
            "anthropic/claude-3.5-sonnet": {"input": 0.003, "output": 0.015},
            "google/gemini-1.5-pro": {"input": 0.00125, "output": 0.005},
        }

        rates = pricing.get(model, {"input": 0.001, "output": 0.002})
        cost = (
            usage.get("prompt_tokens", 0) * rates["input"] / 1000 +
            usage.get("completion_tokens", 0) * rates["output"] / 1000
        )
        return round(cost, 6)

    def get_usage_summary(self):
        '''获取使用汇总（成本聚合功能）'''
        total_cost = sum(log["cost"] for log in self.usage_log)
        total_tokens = sum(log["total_tokens"] for log in self.usage_log)

        model_breakdown = {}
        for log in self.usage_log:
            model = log["model"]
            if model not in model_breakdown:
                model_breakdown[model] = {"calls": 0, "cost": 0, "tokens": 0}
            model_breakdown[model]["calls"] += 1
            model_breakdown[model]["cost"] += log["cost"]
            model_breakdown[model]["tokens"] += log["total_tokens"]

        return {
            "total_cost": round(total_cost, 4),
            "total_tokens": total_tokens,
            "total_calls": len(self.usage_log),
            "by_model": model_breakdown
        }


# 使用示例
client = OpenRouterClient("your-api-key")

# 同一个接口，自动路由到不同模型
messages_list = [
    [{"role": "user", "content": "写一个Python排序算法"}],
    [{"role": "user", "content": "解释量子计算的基本原理"}],
    [{"role": "user", "content": "分析这段长文本...（5000字以上）"}],
]

for msgs in messages_list:
    result = client.chat(msgs, model="auto")

# 查看统一的使用和成本汇总
summary = client.get_usage_summary()
print(json.dumps(summary, indent=2, ensure_ascii=False))

传统SaaS支付模式：
用户 --> 月度/年度订阅 --> Stripe ($99/月)

AI经济支付模式：
开发者 --> API调用 --> OpenRouter --> 多个AI模型 (按token计费)
                            |
                     Stripe统一结算 (成本聚合)
                            |
                     开发者收到统一账单

# AI经济支付处理的复杂度示例
from datetime import datetime

class AIPaymentProcessor:
    '''
    AI经济支付处理器
    处理微交易、实时计费和多模型聚合
    '''

    def __init__(self):
        self.user_balances = {}
        self.pending_charges = {}
        self.exchange_rates = {"USD": 1.0, "CNY": 7.2, "EUR": 0.92}

    def process_api_call(self, user_id, model, input_tokens,
                         output_tokens, latency_ms):
        '''
        处理单次API调用的计费
        '''
        # 1. 计算费用（微交易）
        cost = self._calculate_micro_cost(
            model, input_tokens, output_tokens
        )

        # 2. 检查用户余额
        balance = self.user_balances.get(user_id, 0)
        if balance < cost:
            return {"status": "insufficient_balance", "required": cost}

        # 3. 扣减余额
        self.user_balances[user_id] -= cost

        # 4. 记录交易
        transaction = {
            "user_id": user_id,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": cost,
            "latency_ms": latency_ms,
            "timestamp": datetime.now().isoformat()
        }

        # 5. 聚合到待结算
        if user_id not in self.pending_charges:
            self.pending_charges[user_id] = []
        self.pending_charges[user_id].append(transaction)

        return {"status": "success", "charged": cost}

    def _calculate_micro_cost(self, model, input_tokens, output_tokens):
        '''计算微交易成本'''
        rates = {
            "gpt-4o": (0.005, 0.015),
            "claude-3.5": (0.003, 0.015),
            "gemini-pro": (0.00125, 0.005),
        }
        in_rate, out_rate = rates.get(model, (0.001, 0.002))
        return round(
            (input_tokens * in_rate + output_tokens * out_rate) / 1000,
            6
        )

    def settle_monthly(self, user_id, preferred_currency="USD"):
        '''月度结算'''
        charges = self.pending_charges.get(user_id, [])
        if not charges:
            return {"total": 0, "currency": preferred_currency}

        total_usd = sum(c["cost_usd"] for c in charges)

        # 按模型分组
        by_model = {}
        for charge in charges:
            model = charge["model"]
            if model not in by_model:
                by_model[model] = {"calls": 0, "cost": 0, "tokens": 0}
            by_model[model]["calls"] += 1
            by_model[model]["cost"] += charge["cost_usd"]
            by_model[model]["tokens"] += (
                charge["input_tokens"] + charge["output_tokens"]
            )

        # 货币转换
        rate = self.exchange_rates.get(preferred_currency, 1.0)
        total_local = round(total_usd * rate, 2)

        # 清空待结算
        self.pending_charges[user_id] = []

        return {
            "total_usd": round(total_usd, 4),
            "total_local": total_local,
            "currency": preferred_currency,
            "exchange_rate": rate,
            "by_model": by_model,
            "total_calls": len(charges)
        }

processor = AIPaymentProcessor()
processor.user_balances["user_123"] = 100.0  # 预付100美元

# 模拟多次API调用
for i in range(5):
    processor.process_api_call(
        "user_123", "gpt-4o",
        input_tokens=500 + i*100,
        output_tokens=200 + i*50,
        latency_ms=800 + i*100
    )

# 月度结算
result = processor.settle_monthly("user_123", "CNY")
print(f"月度总费用: {result['total_local']} CNY")

# 开发者应对策略：多网关抽象层
class MultiGatewayRouter:
    '''
    多网关路由器：避免供应商锁定
    同时支持OpenRouter和其他AI模型网关
    '''

    def __init__(self):
        self.gateways = {
            "openrouter": {
                "client": None,
                "weight": 0.5,
                "priority": 1
            },
            "direct_api": {
                "client": None,
                "weight": 0.3,
                "priority": 2
            },
            "local_model": {
                "client": None,
                "weight": 0.2,
                "priority": 3
            }
        }
        self.health_status = {gw: "healthy" for gw in self.gateways}

    def route_request(self, messages, **kwargs):
        '''路由请求到最佳网关'''
        available = [
            gw for gw, status in self.health_status.items()
            if status == "healthy"
        ]

        available.sort(key=lambda g: self.gateways[g]["priority"])

        for gateway_name in available:
            try:
                gateway = self.gateways[gateway_name]
                if gateway["client"]:
                    result = gateway["client"].chat(messages, **kwargs)
                    return {"gateway": gateway_name, "result": result}
            except Exception as e:
                print(f"  [{gateway_name}] 失败: {e}")
                self.health_status[gateway_name] = "degraded"
                continue

        return {"error": "所有网关都不可用"}