摘要

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 端点：

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

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 聚合平台的多模型统一接入方案，核心要点回顾：

1. 统一接入：仅需 OpenAI SDK，通过 base_url="https://api.weytoken.com/v1" 一行配置，即可调用 GPT、Claude、DeepSeek、Gemini、Qwen 等主流模型，切换模型只需修改 model 参数。

2. 智能路由：根据任务类型（代码生成、中文处理、翻译、创意写作等）自动选择最优模型，在保证质量的同时最大化性价比。

3. 故障切换：内置重试、指数退避和熔断器机制，主模型不可用时自动降级到备用模型，确保业务连续性。

4. 成本追踪：实时统计每个模型、每个项目的 Token 消耗和费用，支持导出 CSV 报告，让成本一目了然。

5. 工具链适配：Claude Code、Codex CLI、Cherry Studio 等主流工具均可通过配置自定义 API 端点实现零成本接入。

6. 企业级管理：子账号创建、预算限额、用量监控等企业级功能，满足团队协作和合规需求。

对于企业级用户来说，选择微元算力这样的国内聚合平台，除了技术上的便利，更重要的是数据安全和企业合规方面的保障。相比直接对接境外模型厂商，国内聚合平台在数据不出境、审计合规、财务结算等方面具有明显优势。