API 聚合平台接入实战：OpenAI SDK 一行代码切换 GPT/Claude/DeepSeek，多模型统一调用的完整解决方案 - 微元算力(weytoken)

m0_37841076

490人浏览 · 2026-06-24 06:46:11

m0_37841076 · 2026-06-24 06:46:11 发布

摘要

2026 年以来，大模型领域进入了"百模大战"的白热化阶段：OpenAI 的 GPT-5.5、Anthropic 的 Claude Opus 4.8、Google 的 Gemini 3.5、DeepSeek-V4、阿里的 Qwen 系列……每个模型都有自己的优势场景，企业技术团队往往需要同时对接多个模型来满足不同业务需求。然而，传统方式下分别对接各家 API 意味着需要管理多套密钥、多套 SDK、多套计费账单，运维复杂度呈指数级增长。

本文将以 微元算力(weytoken) 大模型 API 聚合平台为核心，演示如何通过 OpenAI SDK 的 base_url 参数，一行代码实现 GPT、Claude、DeepSeek、Gemini 等多模型的统一调用。文章涵盖模型切换、智能路由、故障自动切换、成本追踪、工具链适配以及企业级子账号管理等完整实战场景，所有代码均可直接运行。

一、环境准备
二、统一 API 接入：一行代码切换模型
三、多模型智能路由：按任务自动选择最优模型
四、故障自动切换：主备模型降级方案
五、成本追踪：实时 Token 消耗与费用统计
六、工具适配：Claude Code / Codex / Cherry Studio 零成本接入
七、企业级管理：子账号与用量限额
八、总结

一、环境准备

1.1 痛点回顾：多模型接入的"切肤之痛"

在正式编码之前，我们先回顾一下企业接入多模型时的典型困境：

维度	传统方式	聚合平台方式
API 密钥	每个模型厂商一套，至少 5 套密钥	1 套密钥统一管理
SDK 依赖	OpenAI SDK + Anthropic SDK + Google SDK + …	仅需 OpenAI SDK
计费账单	每月对 5 张账单，财务头疼	1 张账单，统一结算
模型切换	改代码、改 SDK、改参数格式	改一行 `model` 参数
故障切换	需要自行实现重试与降级逻辑	基于统一接口，降级仅需改 model
数据合规	数据分散在多个境外服务器	国内平台，数据安全可控

微元算力(weytoken) 作为企业级大模型 API 聚合平台，提供了统一的 OpenAI 兼容接口，开发团队只需对接一次，即可调用平台上所有主流模型，同时满足数据安全和企业合规需求。

1.2 Python 环境与依赖安装

# 创建虚拟环境（推荐）
python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# macOS / Linux:
source venv/bin/activate

# 安装依赖
pip install openai>=1.0.0

仅需一个依赖包：openai。不需要安装 Anthropic SDK、Google GenAI SDK 等任何其他模型厂商的 SDK。

1.3 API 密钥配置

在 微元算力(weytoken) 注册并获取 API Key 后，建议通过环境变量管理密钥，避免硬编码在代码中：

import os

# 方式一：直接设置（仅用于快速测试，生产环境请用方式二）
os.environ["WEYTOKEN_API_KEY"] = "sk-your-api-key-here"

# 方式二：从 .env 文件加载（推荐）
# 先在项目根目录创建 .env 文件，写入：
# WEYTOKEN_API_KEY=sk-your-api-key-here

# .env 文件内容示例
WEYTOKEN_API_KEY=sk-your-api-key-here
WEYTOKEN_BASE_URL=https://api.weytoken.com/v1

二、统一 API 接入：一行代码切换模型

2.1 核心原理

微元算力平台提供了完全兼容 OpenAI SDK 的 API 接口。只需将 base_url 指向 https://api.weytoken.com/v1，即可通过标准的 OpenAI SDK 调用平台上所有模型。切换模型仅需修改 model 参数，无需改动任何其他代码。

2.2 基础接入代码

import os
from openai import OpenAI

# ============================================================
# 统一接入配置：所有模型共用一个 client
# ============================================================
client = OpenAI(
    api_key=os.environ.get("WEYTOKEN_API_KEY", "sk-your-api-key-here"),
    base_url="https://api.weytoken.com/v1",
)

# ============================================================
# 一行代码切换模型：只需改 model 参数
# ============================================================

def call_model(model: str, prompt: str) -> str:
    """通用模型调用函数，切换模型只需修改 model 参数"""
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "你是一个专业的技术助手。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.7,
        max_tokens=2048,
    )
    return response.choices[0].message.content


# ----- 测试不同模型 -----
prompt = "请用一句话解释什么是大模型 API 聚合平台。"

# GPT-5.5
print("=== GPT-5.5 ===")
print(call_model("gpt-5.5", prompt))

# Claude Opus 4.8
print("\n=== Claude Opus 4.8 ===")
print(call_model("claude-opus-4-8-20250514", prompt))

# DeepSeek-V4
print("\n=== DeepSeek-V4 ===")
print(call_model("deepseek-v4", prompt))

# Gemini 3.5 Pro
print("\n=== Gemini 3.5 Pro ===")
print(call_model("gemini-3.5-pro", prompt))

# Qwen-Max
print("\n=== Qwen-Max ===")
print(call_model("qwen-max", prompt))

2.3 流式输出（Streaming）

流式输出是聊天应用的标配，同样一行配置即可：

def call_model_stream(model: str, prompt: str):
    """流式调用，实时打印输出"""
    stream = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "你是一个专业的技术助手。"},
            {"role": "user", "content": prompt},
        ],
        stream=True,
        temperature=0.7,
        max_tokens=2048,
    )

    full_response = ""
    for chunk in stream:
        if chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            print(content, end="", flush=True)
            full_response += content
    print()  # 换行
    return full_response


# 使用示例
call_model_stream("deepseek-v4", "写一首关于AI编程的五言绝句")

2.4 多模型并发调用

实际业务中经常需要同时调用多个模型进行对比，以下是一个并发调用示例：

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    api_key=os.environ.get("WEYTOKEN_API_KEY", "sk-your-api-key-here"),
    base_url="https://api.weytoken.com/v1",
)

async def async_call_model(model: str, prompt: str) -> dict:
    """异步调用单个模型"""
    response = await async_client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
        max_tokens=1024,
    )
    return {
        "model": model,
        "content": response.choices[0].message.content,
        "tokens": response.usage.total_tokens,
    }


async def compare_models(prompt: str):
    """并发调用多个模型，对比输出"""
    models = ["gpt-5.5", "claude-opus-4-8-20250514", "deepseek-v4", "gemini-3.5-pro"]
    tasks = [async_call_model(model, prompt) for model in models]
    results = await asyncio.gather(*tasks)

    for r in results:
        print(f"\n{'='*60}")
        print(f"模型: {r['model']} | Token 消耗: {r['tokens']}")
        print(f"{'='*60}")
        print(r["content"][:200] + "..." if len(r["content"]) > 200 else r["content"])


# 运行
# asyncio.run(compare_models("如何设计一个高可用的微服务架构？"))

三、多模型智能路由：按任务自动选择最优模型

3.1 设计思路

不同模型在不同任务上表现差异明显。例如，Claude 在长文本理解和代码生成上表现突出，GPT 在创意写作和通用推理上更为均衡，DeepSeek 在中文场景和数学推理上性价比极高。

智能路由的核心思想是：根据任务类型，自动选择最适合的模型，在保证质量的同时最大化性价比。

3.2 智能路由实现

from enum import Enum
from typing import Optional
from dataclasses import dataclass


class TaskType(Enum):
    """任务类型枚举"""
    CODE_GENERATION = "code_generation"       # 代码生成
    CODE_REVIEW = "code_review"               # 代码审查
    TEXT_ANALYSIS = "text_analysis"           # 长文本分析
    CREATIVE_WRITING = "creative_writing"     # 创意写作
    CHINESE_PROCESSING = "chinese_processing" # 中文处理
    MATH_REASONING = "math_reasoning"         # 数学推理
    TRANSLATION = "translation"               # 翻译
    GENERAL_CHAT = "general_chat"             # 通用对话
    DATA_EXTRACTION = "data_extraction"       # 数据提取
    SUMMARIZATION = "summarization"           # 摘要总结


@dataclass
class ModelRoute:
    """模型路由配置"""
    model: str
    priority: int  # 优先级，数字越小越优先
    max_tokens: int = 4096
    temperature: float = 0.7


# ============================================================
# 智能路由表：根据任务类型选择最优模型
# ============================================================
ROUTE_TABLE: dict[TaskType, list[ModelRoute]] = {
    TaskType.CODE_GENERATION: [
        ModelRoute("claude-opus-4-8-20250514", 1, max_tokens=8192, temperature=0.3),
        ModelRoute("deepseek-v4", 2, max_tokens=4096, temperature=0.3),
        ModelRoute("gpt-5.5", 3, max_tokens=4096, temperature=0.3),
    ],
    TaskType.CODE_REVIEW: [
        ModelRoute("claude-opus-4-8-20250514", 1, max_tokens=4096, temperature=0.2),
        ModelRoute("gpt-5.5", 2, max_tokens=4096, temperature=0.2),
    ],
    TaskType.TEXT_ANALYSIS: [
        ModelRoute("claude-opus-4-8-20250514", 1, max_tokens=8192, temperature=0.5),
        ModelRoute("gemini-3.5-pro", 2, max_tokens=4096, temperature=0.5),
    ],
    TaskType.CREATIVE_WRITING: [
        ModelRoute("gpt-5.5", 1, max_tokens=4096, temperature=0.9),
        ModelRoute("claude-opus-4-8-20250514", 2, max_tokens=4096, temperature=0.9),
    ],
    TaskType.CHINESE_PROCESSING: [
        ModelRoute("deepseek-v4", 1, max_tokens=4096, temperature=0.7),
        ModelRoute("qwen-max", 2, max_tokens=4096, temperature=0.7),
        ModelRoute("gpt-5.5", 3, max_tokens=4096, temperature=0.7),
    ],
    TaskType.MATH_REASONING: [
        ModelRoute("deepseek-v4", 1, max_tokens=4096, temperature=0.1),
        ModelRoute("gpt-5.5", 2, max_tokens=4096, temperature=0.1),
    ],
    TaskType.TRANSLATION: [
        ModelRoute("gpt-5.5", 1, max_tokens=4096, temperature=0.3),
        ModelRoute("deepseek-v4", 2, max_tokens=4096, temperature=0.3),
    ],
    TaskType.GENERAL_CHAT: [
        ModelRoute("gpt-5.5", 1, max_tokens=2048, temperature=0.7),
    ],
    TaskType.DATA_EXTRACTION: [
        ModelRoute("gpt-5.5", 1, max_tokens=2048, temperature=0.1),
        ModelRoute("deepseek-v4", 2, max_tokens=2048, temperature=0.1),
    ],
    TaskType.SUMMARIZATION: [
        ModelRoute("claude-opus-4-8-20250514", 1, max_tokens=2048, temperature=0.3),
        ModelRoute("gpt-5.5", 2, max_tokens=2048, temperature=0.3),
    ],
}


class SmartRouter:
    """智能路由器：根据任务类型自动选择最优模型"""

    def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
        self.client = OpenAI(
            api_key=api_key or os.environ.get("WEYTOKEN_API_KEY"),
            base_url=base_url or "https://api.weytoken.com/v1",
        )

    def _auto_detect_task(self, prompt: str) -> TaskType:
        """基于关键词自动检测任务类型"""
        prompt_lower = prompt.lower()

        code_keywords = ["代码", "code", "函数", "function", "类", "class", "bug",
                         "debug", "编程", "programming", "实现", "implement"]
        review_keywords = ["审查", "review", "优化", "optimize", "重构", "refactor",
                           "性能", "performance"]
        math_keywords = ["计算", "calculate", "数学", "math", "公式", "formula",
                         "证明", "prove", "推理", "reasoning"]
        translation_keywords = ["翻译", "translate", "英文", "english", "中文",
                                "chinese", "日语", "japanese"]
        extraction_keywords = ["提取", "extract", "解析", "parse", "结构化", "json",
                               "正则", "regex"]
        summary_keywords = ["总结", "summarize", "摘要", "概括", "summary", "归纳"]
        creative_keywords = ["写", "创作", "故事", "story", "诗歌", "poem", "文案",
                             "copywriting", "创意", "creative"]

        if any(kw in prompt_lower for kw in code_keywords):
            return TaskType.CODE_GENERATION
        if any(kw in prompt_lower for kw in review_keywords):
            return TaskType.CODE_REVIEW
        if any(kw in prompt_lower for kw in math_keywords):
            return TaskType.MATH_REASONING
        if any(kw in prompt_lower for kw in translation_keywords):
            return TaskType.TRANSLATION
        if any(kw in prompt_lower for kw in extraction_keywords):
            return TaskType.DATA_EXTRACTION
        if any(kw in prompt_lower for kw in summary_keywords):
            return TaskType.SUMMARIZATION
        if any(kw in prompt_lower for kw in creative_keywords):
            return TaskType.CREATIVE_WRITING

        return TaskType.GENERAL_CHAT

    def route(self, prompt: str, task_type: Optional[TaskType] = None,
              use_fallback: bool = True) -> dict:
        """
        智能路由调用

        Args:
            prompt: 用户输入
            task_type: 任务类型，不传则自动检测
            use_fallback: 是否启用故障自动切换

        Returns:
            dict: 包含 model、content、tokens 等信息
        """
        if task_type is None:
            task_type = self._auto_detect_task(prompt)

        routes = ROUTE_TABLE.get(task_type, ROUTE_TABLE[TaskType.GENERAL_CHAT])
        routes = sorted(routes, key=lambda r: r.priority)

        last_error = None
        for route in routes:
            try:
                response = self.client.chat.completions.create(
                    model=route.model,
                    messages=[{"role": "user", "content": prompt}],
                    temperature=route.temperature,
                    max_tokens=route.max_tokens,
                )
                return {
                    "model": route.model,
                    "task_type": task_type.value,
                    "content": response.choices[0].message.content,
                    "total_tokens": response.usage.total_tokens,
                    "prompt_tokens": response.usage.prompt_tokens,
                    "completion_tokens": response.usage.completion_tokens,
                    "fallback_used": routes.index(route) > 0,
                }
            except Exception as e:
                last_error = e
                if not use_fallback:
                    raise
                print(f"[警告] 模型 {route.model} 调用失败: {e}，尝试下一个...")
                continue

        raise RuntimeError(f"所有模型均调用失败，最后错误: {last_error}")


# ============================================================
# 使用示例
# ============================================================
router = SmartRouter()

# 示例 1：代码生成任务，自动路由到 Claude Opus
result = router.route("用 Python 实现一个线程安全的 LRU 缓存")
print(f"[{result['task_type']}] 使用模型: {result['model']}, Token: {result['total_tokens']}")
print(result["content"][:300])

# 示例 2：中文任务，自动路由到 DeepSeek
result = router.route("帮我写一篇关于人工智能发展史的科普文章，面向中文读者")
print(f"\n[{result['task_type']}] 使用模型: {result['model']}, Token: {result['total_tokens']}")

# 示例 3：手动指定任务类型
result = router.route(
    "Translate the following to Chinese: 'The quick brown fox jumps over the lazy dog.'",
    task_type=TaskType.TRANSLATION
)
print(f"\n[{result['task_type']}] 使用模型: {result['model']}, Token: {result['total_tokens']}")
print(result["content"])

四、故障自动切换：主备模型降级方案

4.1 为什么需要故障切换

在生产环境中，模型 API 可能因为以下原因不可用：

上游模型厂商服务故障（如 OpenAI 宕机）
速率限制（Rate Limit）触发
账户余额不足
网络波动导致超时

聚合平台的优势在于：当主模型不可用时，可以无缝切换到备用模型，保证业务不中断。

4.2 带重试和降级的健壮调用

import time
import random
from functools import wraps
from typing import Callable, Any


class RobustModelClient:
    """
    健壮模型客户端：内置重试、指数退避、故障切换
    """

    def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
        self.client = OpenAI(
            api_key=api_key or os.environ.get("WEYTOKEN_API_KEY"),
            base_url=base_url or "https://api.weytoken.com/v1",
            timeout=60.0,  # 60 秒超时
            max_retries=0,  # 我们自己控制重试
        )

    def call_with_fallback(
        self,
        prompt: str,
        primary_model: str = "claude-opus-4-8-20250514",
        fallback_models: Optional[list[str]] = None,
        max_retries: int = 3,
        system_prompt: str = "你是一个专业的技术助手。",
        **kwargs,
    ) -> dict:
        """
        带故障切换的模型调用

        Args:
            prompt: 用户输入
            primary_model: 主模型
            fallback_models: 备用模型列表（按优先级排列）
            max_retries: 每个模型的最大重试次数
            system_prompt: 系统提示词

        Returns:
            dict: 包含响应内容和调用详情
        """
        if fallback_models is None:
            fallback_models = ["gpt-5.5", "deepseek-v4", "gemini-3.5-pro"]

        # 模型调用序列：主模型 + 备用模型
        model_sequence = [primary_model] + [
            m for m in fallback_models if m != primary_model
        ]

        last_error = None
        call_history = []

        for model in model_sequence:
            for attempt in range(1, max_retries + 1):
                try:
                    start_time = time.time()
                    response = self.client.chat.completions.create(
                        model=model,
                        messages=[
                            {"role": "system", "content": system_prompt},
                            {"role": "user", "content": prompt},
                        ],
                        **kwargs,
                    )
                    elapsed = time.time() - start_time

                    result = {
                        "success": True,
                        "model": model,
                        "content": response.choices[0].message.content,
                        "total_tokens": response.usage.total_tokens,
                        "latency_seconds": round(elapsed, 2),
                        "attempts": attempt,
                        "is_fallback": (model != primary_model),
                        "call_history": call_history,
                    }
                    return result

                except Exception as e:
                    error_info = {
                        "model": model,
                        "attempt": attempt,
                        "error": str(e),
                    }
                    call_history.append(error_info)
                    last_error = e

                    if attempt < max_retries:
                        # 指数退避 + 随机抖动
                        wait_time = (2 ** attempt) + random.uniform(0, 1)
                        print(f"[重试] 模型 {model} 第 {attempt} 次失败，"
                              f"{wait_time:.1f}s 后重试: {e}")
                        time.sleep(wait_time)
                    else:
                        print(f"[切换] 模型 {model} 重试 {max_retries} 次后仍失败，"
                              f"切换到下一个备用模型")

        # 所有模型都失败
        return {
            "success": False,
            "content": None,
            "error": str(last_error),
            "call_history": call_history,
        }


# ============================================================
# 使用示例
# ============================================================
robust_client = RobustModelClient()

# 正常调用
result = robust_client.call_with_fallback(
    prompt="用 Go 语言实现一个并发安全的计数器",
    primary_model="claude-opus-4-8-20250514",
    temperature=0.3,
    max_tokens=4096,
)

if result["success"]:
    print(f"模型: {result['model']} | 延迟: {result['latency_seconds']}s | "
          f"Token: {result['total_tokens']} | 是否降级: {result['is_fallback']}")
    print(result["content"][:500])
else:
    print(f"调用失败: {result['error']}")
    print(f"调用历史: {result['call_history']}")

4.3 熔断器模式（Circuit Breaker）

对于高并发场景，建议引入熔断器模式，避免在模型持续不可用时重复尝试：

import threading
from datetime import datetime, timedelta


class CircuitBreaker:
    """简单的熔断器实现"""

    def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout  # 秒
        self.failure_count = 0
        self.last_failure_time: Optional[datetime] = None
        self.state = "CLOSED"  # CLOSED / OPEN / HALF_OPEN
        self.lock = threading.Lock()

    def call(self, func: Callable, *args, **kwargs) -> Any:
        with self.lock:
            if self.state == "OPEN":
                if (datetime.now() - self.last_failure_time).total_seconds() > self.recovery_timeout:
                    self.state = "HALF_OPEN"
                    print("[熔断器] 进入半开状态，尝试恢复...")
                else:
                    raise Exception(f"熔断器已打开，请等待 {self.recovery_timeout}s 后重试")

        try:
            result = func(*args, **kwargs)
            with self.lock:
                self.failure_count = 0
                self.state = "CLOSED"
            return result
        except Exception as e:
            with self.lock:
                self.failure_count += 1
                self.last_failure_time = datetime.now()
                if self.failure_count >= self.failure_threshold:
                    self.state = "OPEN"
                    print(f"[熔断器] 连续失败 {self.failure_count} 次，熔断器打开")
            raise e


# ============================================================
# 集成熔断器的模型调用
# ============================================================
class ProductionModelClient(RobustModelClient):
    """生产级模型客户端：故障切换 + 熔断器"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.circuit_breakers: dict[str, CircuitBreaker] = {}

    def _get_breaker(self, model: str) -> CircuitBreaker:
        if model not in self.circuit_breakers:
            self.circuit_breakers[model] = CircuitBreaker(
                failure_threshold=3,
                recovery_timeout=30,
            )
        return self.circuit_breakers[model]

    def call_production(self, prompt: str, primary_model: str, **kwargs) -> dict:
        """生产级调用：内置熔断保护"""
        return self.call_with_fallback(
            prompt=prompt,
            primary_model=primary_model,
            **kwargs,
        )

五、成本追踪：实时 Token 消耗与费用统计

5.1 成本追踪的意义

调用多个模型时，如果不做成本追踪，月底账单可能会让你大吃一惊。不同模型的价格差异巨大，例如 Claude Opus 的价格可能是 DeepSeek 的数十倍。精确的成本追踪能帮助你：

实时了解每个模型的花费
按项目/业务线拆分成本
发现异常消耗并及时止损

5.2 完整的成本追踪实现

import json
from collections import defaultdict
from datetime import datetime


# ============================================================
# 各模型定价（示例价格，实际以微元算力平台为准）
# 单位：元 / 百万 Token
# ============================================================
MODEL_PRICING = {
    "gpt-5.5":                    {"input": 15.0,  "output": 60.0},
    "claude-opus-4-8-20250514":   {"input": 75.0,  "output": 300.0},
    "deepseek-v4":                {"input": 2.0,   "output": 8.0},
    "gemini-3.5-pro":             {"input": 7.0,   "output": 28.0},
    "qwen-max":                   {"input": 5.0,   "output": 20.0},
}


class CostTracker:
    """成本追踪器：实时统计 Token 消耗和费用"""

    def __init__(self):
        self.records: list[dict] = []
        self._lock = threading.Lock()

    def record(self, model: str, prompt_tokens: int, completion_tokens: int,
               latency: float = 0, project: str = "default",
               task_type: str = "general") -> dict:
        """记录一次调用"""
        pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
        input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
        output_cost = (completion_tokens / 1_000_000) * pricing["output"]
        total_cost = input_cost + output_cost

        record = {
            "timestamp": datetime.now().isoformat(),
            "model": model,
            "project": project,
            "task_type": task_type,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": prompt_tokens + completion_tokens,
            "input_cost": round(input_cost, 6),
            "output_cost": round(output_cost, 6),
            "total_cost": round(total_cost, 6),
            "latency_seconds": round(latency, 2),
        }

        with self._lock:
            self.records.append(record)

        return record

    def get_summary(self) -> dict:
        """获取汇总统计"""
        with self._lock:
            if not self.records:
                return {"total_calls": 0, "total_cost": 0}

            total_cost = sum(r["total_cost"] for r in self.records)
            total_tokens = sum(r["total_tokens"] for r in self.records)

            # 按模型汇总
            by_model = defaultdict(lambda: {"calls": 0, "tokens": 0, "cost": 0.0})
            for r in self.records:
                by_model[r["model"]]["calls"] += 1
                by_model[r["model"]]["tokens"] += r["total_tokens"]
                by_model[r["model"]]["cost"] += r["total_cost"]

            # 按项目汇总
            by_project = defaultdict(lambda: {"calls": 0, "tokens": 0, "cost": 0.0})
            for r in self.records:
                by_project[r["project"]]["calls"] += 1
                by_project[r["project"]]["tokens"] += r["total_tokens"]
                by_project[r["project"]]["cost"] += r["total_cost"]

            return {
                "total_calls": len(self.records),
                "total_tokens": total_tokens,
                "total_cost": round(total_cost, 4),
                "by_model": dict(by_model),
                "by_project": dict(by_project),
            }

    def print_summary(self):
        """打印格式化汇总报告"""
        summary = self.get_summary()
        print("=" * 70)
        print(f"                    成本追踪汇总报告")
        print("=" * 70)
        print(f"  总调用次数: {summary['total_calls']}")
        print(f"  总 Token 数: {summary['total_tokens']:,}")
        print(f"  总费用:     ${summary['total_cost']:.4f}")
        print("-" * 70)
        print(f"  {'模型':<30} {'调用':>6} {'Token':>10} {'费用':>10}")
        print("-" * 70)
        for model, stats in summary["by_model"].items():
            print(f"  {model:<30} {stats['calls']:>6} "
                  f"{stats['tokens']:>10,} ${stats['cost']:>9.4f}")
        print("-" * 70)
        print(f"  {'项目':<30} {'调用':>6} {'Token':>10} {'费用':>10}")
        print("-" * 70)
        for project, stats in summary["by_project"].items():
            print(f"  {project:<30} {stats['calls']:>6} "
                  f"{stats['tokens']:>10,} ${stats['cost']:>9.4f}")
        print("=" * 70)

    def export_csv(self, filepath: str):
        """导出为 CSV 文件"""
        import csv
        with self._lock:
            if not self.records:
                return
            fieldnames = self.records[0].keys()
            with open(filepath, "w", newline="", encoding="utf-8") as f:
                writer = csv.DictWriter(f, fieldnames=fieldnames)
                writer.writeheader()
                writer.writerows(self.records)
        print(f"[成本追踪] 已导出 {len(self.records)} 条记录到 {filepath}")


# ============================================================
# 集成成本追踪的模型客户端
# ============================================================
class TrackedModelClient:
    """带成本追踪的模型客户端"""

    def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
        self.client = OpenAI(
            api_key=api_key or os.environ.get("WEYTOKEN_API_KEY"),
            base_url=base_url or "https://api.weytoken.com/v1",
        )
        self.tracker = CostTracker()

    def chat(self, model: str, prompt: str, project: str = "default",
             task_type: str = "general", **kwargs) -> dict:
        """带成本追踪的聊天调用"""
        start_time = time.time()

        response = self.client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            **kwargs,
        )

        latency = time.time() - start_time

        # 记录成本
        cost_record = self.tracker.record(
            model=model,
            prompt_tokens=response.usage.prompt_tokens,
            completion_tokens=response.usage.completion_tokens,
            latency=latency,
            project=project,
            task_type=task_type,
        )

        return {
            "content": response.choices[0].message.content,
            "cost": cost_record,
        }


# ============================================================
# 使用示例
# ============================================================
tracked_client = TrackedModelClient()

# 模拟多次调用
test_cases = [
    ("deepseek-v4", "帮我写一个 Python 快速排序", "project_alpha", "code"),
    ("gpt-5.5", "写一篇关于云计算的短文", "project_beta", "writing"),
    ("claude-opus-4-8-20250514", "分析这段代码的性能瓶颈", "project_alpha", "review"),
    ("deepseek-v4", "用中文回答：什么是机器学习？", "project_gamma", "qa"),
    ("gemini-3.5-pro", "翻译：Hello World 到中文", "project_beta", "translation"),
]

for model, prompt, project, task_type in test_cases:
    result = tracked_client.chat(model, prompt, project=project, task_type=task_type)
    print(f"[{project}] {model}: {result['content'][:50]}... "
          f"费用: ${result['cost']['total_cost']:.6f}")

# 打印汇总报告
tracked_client.tracker.print_summary()

# 导出 CSV
# tracked_client.tracker.export_csv("cost_report.csv")

六、工具适配：Claude Code / Codex / Cherry Studio 零成本接入

6.1 适配原理

微元算力平台的 API 完全兼容 OpenAI 接口规范，因此几乎所有支持自定义 base_url 的 AI 工具都可以零成本接入。以下是三款主流开发工具的配置方法。

6.2 Claude Code（CLI AI 编程助手）

Claude Code 是 Anthropic 官方推出的终端 AI 编程助手，通过配置 ANTHROPIC_BASE_URL 环境变量即可接入聚合平台：

# 方式一：环境变量配置（推荐）
export ANTHROPIC_BASE_URL="https://api.weytoken.com/v1"
export ANTHROPIC_API_KEY="sk-your-api-key-here"

# 方式二：在 claude 配置文件中设置
# ~/.claude.json
{
  "apiKey": "sk-your-api-key-here",
  "baseUrl": "https://api.weytoken.com/v1"
}

配置完成后，Claude Code 的所有模型调用都将通过微元算力平台路由，你可以在 Web 控制台随时切换使用的模型。

6.3 OpenAI Codex CLI

OpenAI 最近开源的 Codex CLI 同样支持自定义 API 端点：

# 设置环境变量
export OPENAI_API_KEY="sk-your-api-key-here"
export OPENAI_BASE_URL="https://api.weytoken.com/v1"

# 启动 Codex
codex

6.4 Cherry Studio（桌面 AI 客户端）

Cherry Studio 是一款优秀的桌面 AI 客户端，支持通过图形界面配置自定义 API 端点：

打开 Cherry Studio，进入 设置 > 模型服务
点击 添加提供商
选择 OpenAI 兼容 类型
填写配置：
- API 地址：https://api.weytoken.com/v1
- API 密钥：粘贴你的 API Key
- 模型列表：从平台获取可用模型列表
保存后即可在对话界面选择任意模型

6.5 通用适配脚本

以下是一个通用的适配检测脚本，帮助验证你的工具是否兼容：

import requests


def check_api_compatibility(base_url: str, api_key: str):
    """检测 API 兼容性"""
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    # 1. 检测模型列表
    print("=" * 50)
    print("API 兼容性检测")
    print("=" * 50)

    try:
        resp = requests.get(f"{base_url}/models", headers=headers, timeout=10)
        if resp.status_code == 200:
            models = resp.json().get("data", [])
            print(f"\n[OK] 模型列表获取成功，共 {len(models)} 个可用模型:")
            for m in models[:10]:
                print(f"     - {m['id']}")
            if len(models) > 10:
                print(f"     ... 还有 {len(models) - 10} 个模型")
        else:
            print(f"[FAIL] 模型列表获取失败: {resp.status_code}")
    except Exception as e:
        print(f"[FAIL] 连接失败: {e}")

    # 2. 检测 Chat Completions 端点
    try:
        resp = requests.post(
            f"{base_url}/chat/completions",
            headers=headers,
            json={
                "model": "gpt-5.5",
                "messages": [{"role": "user", "content": "Hi"}],
                "max_tokens": 10,
            },
            timeout=15,
        )
        if resp.status_code == 200:
            print(f"\n[OK] Chat Completions 端点正常")
            data = resp.json()
            print(f"     响应模型: {data.get('model', 'N/A')}")
            print(f"     响应内容: {data['choices'][0]['message']['content'][:50]}...")
        else:
            print(f"\n[FAIL] Chat Completions 端点异常: {resp.status_code}")
            print(f"     {resp.text[:200]}")
    except Exception as e:
        print(f"\n[FAIL] Chat Completions 调用失败: {e}")

    print("=" * 50)


# 运行检测
# check_api_compatibility("https://api.weytoken.com/v1", "sk-your-api-key-here")

七、企业级管理：子账号与用量限额

7.1 企业场景需求

在企业团队中，通常需要：

为不同部门/项目创建独立的子账号
为每个子账号设置用量限额和预算上限
统一的账单管理和审计日志
基于角色的权限控制

7.2 子账号管理 SDK 封装

import requests
from typing import Optional


class WeyTokenAdminClient:
    """
    微元算力平台管理客户端
    用于企业级子账号管理、用量查询和限额设置
    """

    def __init__(self, api_key: str, base_url: str = "https://api.weytoken.com/v1"):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    def get_usage(self, start_date: Optional[str] = None,
                  end_date: Optional[str] = None) -> dict:
        """
        查询用量统计

        Args:
            start_date: 开始日期 (YYYY-MM-DD)
            end_date: 结束日期 (YYYY-MM-DD)
        """
        params = {}
        if start_date:
            params["start_date"] = start_date
        if end_date:
            params["end_date"] = end_date

        resp = requests.get(
            f"{self.base_url}/usage",
            headers=self.headers,
            params=params,
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()

    def list_sub_accounts(self) -> list[dict]:
        """获取子账号列表"""
        resp = requests.get(
            f"{self.base_url}/sub_accounts",
            headers=self.headers,
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json().get("data", [])

    def create_sub_account(self, name: str, monthly_budget: float,
                           allowed_models: Optional[list[str]] = None) -> dict:
        """
        创建子账号

        Args:
            name: 子账号名称（如 "前端开发组"、"AI 测试项目"）
            monthly_budget: 月度预算上限（美元）
            allowed_models: 允许使用的模型列表，None 表示全部可用
        """
        payload = {
            "name": name,
            "monthly_budget": monthly_budget,
        }
        if allowed_models:
            payload["allowed_models"] = allowed_models

        resp = requests.post(
            f"{self.base_url}/sub_accounts",
            headers=self.headers,
            json=payload,
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()

    def update_sub_account_budget(self, sub_account_id: str,
                                  monthly_budget: float) -> dict:
        """更新子账号预算"""
        resp = requests.patch(
            f"{self.base_url}/sub_accounts/{sub_account_id}",
            headers=self.headers,
            json={"monthly_budget": monthly_budget},
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()

    def get_sub_account_usage(self, sub_account_id: str) -> dict:
        """查询子账号用量"""
        resp = requests.get(
            f"{self.base_url}/sub_accounts/{sub_account_id}/usage",
            headers=self.headers,
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()

    def print_team_dashboard(self):
        """打印团队用量仪表盘"""
        accounts = self.list_sub_accounts()
        print("=" * 80)
        print(f"                    团队用量仪表盘")
        print("=" * 80)
        print(f"  {'子账号名称':<20} {'预算':>10} {'已用':>10} {'剩余':>10} {'占比':>8}")
        print("-" * 80)

        for acc in accounts:
            usage = self.get_sub_account_usage(acc["id"])
            budget = acc.get("monthly_budget", 0)
            used = usage.get("total_cost", 0)
            remaining = budget - used
            pct = (used / budget * 100) if budget > 0 else 0

            print(f"  {acc['name']:<20} ${budget:>9.2f} ${used:>9.2f} "
                  f"${remaining:>9.2f} {pct:>7.1f}%")

        print("=" * 80)


# ============================================================
# 使用示例
# ============================================================
# admin = WeyTokenAdminClient(api_key="sk-your-admin-key")
#
# # 创建子账号
# admin.create_sub_account(
#     name="AI产品研发组",
#     monthly_budget=500.0,
#     allowed_models=["gpt-5.5", "claude-opus-4-8-20250514", "deepseek-v4"],
# )
#
# admin.create_sub_account(
#     name="内容创作组",
#     monthly_budget=200.0,
#     allowed_models=["gpt-5.5", "deepseek-v4"],
# )
#
# # 查看仪表盘
# admin.print_team_dashboard()

7.3 企业级最佳实践总结

实践	说明
主账号 + 子账号	主账号统一充值，子账号按项目/部门分配
预算告警	设置 80% 预算告警线，避免超额
模型白名单	限制子账号只能使用指定模型，控制成本
审计日志	定期导出调用记录，用于财务对账
密钥轮换	定期更换 API Key，降低泄露风险

八、总结

本文从实战角度出发，完整演示了基于 微元算力(weytoken) API 聚合平台的多模型统一接入方案，核心要点回顾：

统一接入：仅需 OpenAI SDK，通过 base_url="https://api.weytoken.com/v1" 一行配置，即可调用 GPT、Claude、DeepSeek、Gemini、Qwen 等主流模型，切换模型只需修改 model 参数。
智能路由：根据任务类型（代码生成、中文处理、翻译、创意写作等）自动选择最优模型，在保证质量的同时最大化性价比。
故障切换：内置重试、指数退避和熔断器机制，主模型不可用时自动降级到备用模型，确保业务连续性。
成本追踪：实时统计每个模型、每个项目的 Token 消耗和费用，支持导出 CSV 报告，让成本一目了然。
工具链适配：Claude Code、Codex CLI、Cherry Studio 等主流工具均可通过配置自定义 API 端点实现零成本接入。
企业级管理：子账号创建、预算限额、用量监控等企业级功能，满足团队协作和合规需求。