AI 路由层正在成为基础设施——OpenRouter 1.13 亿美元融资背后的信号

import os
import requests
from datetime import datetime

OPENROUTER_API_KEY = os.environ.get("OPENROUTER_API_KEY")
OPENROUTER_BASE = "https://openrouter.ai/api/v1"

# 定义路由策略：不同任务类型走不同模型
ROUTING_TABLE = {
    "complex_reasoning": {
        "model": "anthropic/claude-opus-4",
        "max_tokens": 4096,
        "reason": "复杂推理需要最强模型",
    },
    "code_generation": {
        "model": "deepseek/deepseek-coder-v2",
        "max_tokens": 2048,
        "reason": "代码生成用专域模型，成本更低",
    },
    "quick_summary": {
        "model": "google/gemini-2.5-flash-preview",
        "max_tokens": 512,
        "reason": "摘要任务用快速模型，延迟最低",
    },
}

def classify_task(prompt: str) -> str:
    """简单的任务分类器——生产环境可替换为更精细的规则或小模型"""
    if any(kw in prompt.lower() for kw in ["分析", "推理", "决策", "比较优劣"]):
        return "complex_reasoning"
    if any(kw in prompt.lower() for kw in ["写代码", "实现函数", "debug", "refactor"]):
        return "code_generation"
    return "quick_summary"

def call_openrouter(prompt: str, task_type: str = None):
    """通过 OpenRouter 路由调用，自动选择模型"""
    if task_type is None:
        task_type = classify_task(prompt)

    route = ROUTING_TABLE[task_type]
    model = route["model"]

    headers = {
        "Authorization": f"Bearer {OPENROUTER_API_KEY}",
        "Content-Type": "application/json",
        # OpenRouter 支持在 header 中传递应用信息，用于统计和排名
        "HTTP-Referer": "https://my-app.example",
        "X-Title": "Multi-Model Router Demo",
    }

    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": route["max_tokens"],
    }

    print(f"[{datetime.now().isoformat()}] 任务类型={task_type} → 模型={model} ({route['reason']})")

    resp = requests.post(f"{OPENROUTER_BASE}/chat/completions", headers=headers, json=payload, timeout=60)
    resp.raise_for_status()
    data = resp.json()

    # OpenRouter 返回中包含用量和成本信息
    usage = data.get("usage", {})
    content = data["choices"][0]["message"]["content"]

    print(f"  tokens: prompt={usage.get('prompt_tokens')}, completion={usage.get('completion_tokens')}")
    print(f"  cost: ${usage.get('total_cost', 'N/A')}")
    print(f"  回复: {content[:200]}...")
    return content

# —— 运行三种不同任务 ——
if __name__ == "__main__":
    # 1. 复杂推理 → Claude Opus
    call_openrouter("分析以下两种微服务架构方案的优劣：基于服务网格 vs 基于 API 网关的中心化路由")

    # 2. 代码生成 → DeepSeek Coder
    call_openrouter("用 Python 实现一个带过期时间的 LRU Cache 类，要求支持 O(1) get 和 put")

    # 3. 快速摘要 → Gemini Flash
    call_openrouter("请用一句话总结：Kubernetes 中 HPA 和 VPA 的核心区别是什么")

export OPENROUTER_API_KEY="sk-or-v1-xxxxxxxxxxxx"
python router_demo.py

# 不指定具体模型，让 OpenRouter 根据成本和性能自动路由
payload = {
    "model": "openrouter/auto",  # 自动路由模式
    "messages": [{"role": "user", "content": prompt}],
    "max_tokens": 1024,
    # 可选：设置成本上限
    "max_price": {"input": 0.005, "output": 0.01},  # 每 1M tokens 的价格上限（美元）
}