我在过去两年帮助超过30家企业完成了 AI API 的迁移与成本优化，见过太多团队在毫不知情的情况下每月烧掉上万美元。作为一个深度使用 AI API 的开发者，我今天把我在生产环境中验证过的完整成本控制方案分享出来，包括架构设计、代码实现、benchmark 数据，以及 HolySheep 这类中转服务的选型逻辑。

为什么 AI API 成本会失控

根据我对接入项目的统计，超过 70% 的团队在第三个月时会发现 API 费用超出预期 3-5 倍。核心问题往往不是业务增长，而是以下几个系统性漏洞：

完整预算控制架构

我推荐的架构分为五层：计量层 → 策略层 → 缓存层 → 限流层 → 监控层。每一层都有对应的代码实现。

计量层：精确统计每个请求的 Token 消耗

"""
AI API 成本计量中间件
支持 HolySheep API (https://api.holysheep.ai/v1) 及官方 API
"""

import time
import json
import tiktoken
from dataclasses import dataclass, asdict
from typing import Optional, List, Dict, Any
from datetime import datetime
import sqlite3
import threading

@dataclass
class TokenUsage:
    """Token 使用记录"""
    timestamp: str
    model: str
    input_tokens: int
    output_tokens: int
    total_tokens: int
    cost_usd: float
    cost_cny: float
    provider: str  # "holysheep" or "openai"
    request_id: str

class CostTracker:
    """成本追踪器"""
    
    # 2026 年主流模型定价 (USD per 1M tokens)
    # Input / Output 价格
    MODEL_PRICES = {
        # GPT-4.1 系列
        "gpt-4.1": (2.0, 8.0),
        "gpt-4.1-mini": (0.4, 1.6),
        # Claude Sonnet 4.5 系列
        "claude-sonnet-4.5": (3.0, 15.0),
        "claude-haiku-3.5": (0.8, 4.0),
        # Gemini 2.5 系列
        "gemini-2.5-flash": (0.35, 2.50),
        "gemini-2.5-pro": (1.25, 10.0),
        # DeepSeek V3.2 系列
        "deepseek-v3.2": (0.14, 0.42),
        "deepseek-chat": (0.12, 0.28),
    }
    
    # 汇率配置
    HOLYSHEEP_RATE = 1.0  # ¥1 = $1 (无损)
    OFFICIAL_RATE = 7.3    # 官方人民币汇率
    
    def __init__(self, db_path: str = "cost_tracker.db"):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self._init_db()
        self._lock = threading.Lock()
        
    def _init_db(self):
        """初始化数据库"""
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS token_usage (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT,
                model TEXT,
                input_tokens INTEGER,
                output_tokens INTEGER,
                total_tokens INTEGER,
                cost_usd REAL,
                cost_cny REAL,
                provider TEXT,
                request_id TEXT
            )
        """)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS daily_budget (
                date TEXT PRIMARY KEY,
                budget_usd REAL,
                spent_usd REAL,
                alert_sent INTEGER DEFAULT 0
            )
        """)
        self.conn.commit()
    
    def calculate_cost(
        self, 
        model: str, 
        input_tokens: int, 
        output_tokens: int,
        provider: str = "holysheep"
    ) -> tuple[float, float]:
        """计算单次请求成本"""
        if model not in self.MODEL_PRICES:
            # 默认按 GPT-4.1-mini 计算
            input_price, output_price = self.MODEL_PRICES["gpt-4.1-mini"]
        else:
            input_price, output_price = self.MODEL_PRICES[model]
        
        # 计算美元成本
        cost_usd = (input_tokens * input_price + output_tokens * output_price) / 1_000_000
        
        # 转换为人民币
        rate = self.HOLYSHEEP_RATE if provider == "holysheep" else self.OFFICIAL_RATE
        cost_cny = cost_usd * rate
        
        return cost_usd, cost_cny
    
    def record(self, usage: TokenUsage):
        """记录一次 Token 使用"""
        with self._lock:
            self.conn.execute("""
                INSERT INTO token_usage 
                (timestamp, model, input_tokens, output_tokens, total_tokens, 
                 cost_usd, cost_cny, provider, request_id)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                usage.timestamp, usage.model, usage.input_tokens, usage.output_tokens,
                usage.total_tokens, usage.cost_usd, usage.cost_cny, 
                usage.provider, usage.request_id
            ))
            self.conn.commit()
    
    def get_daily_spent(self, date: Optional[str] = None) -> Dict[str, float]:
        """获取每日消费统计"""
        if date is None:
            date = datetime.now().strftime("%Y-%m-%d")
        
        cursor = self.conn.execute("""
            SELECT 
                COALESCE(SUM(cost_usd), 0) as total_usd,
                COALESCE(SUM(cost_cny), 0) as total_cny,
                COUNT(*) as request_count
            FROM token_usage 
            WHERE timestamp LIKE ?
        """, (f"{date}%",))
        
        row = cursor.fetchone()
        return {
            "total_usd": row[0],
            "total_cny": row[1],
            "request_count": row[2]
        }
    
    def get_monthly_spent(self, year_month: Optional[str] = None) -> Dict[str, float]:
        """获取月度消费统计"""
        if year_month is None:
            year_month = datetime.now().strftime("%Y-%m")
        
        cursor = self.conn.execute("""
            SELECT 
                COALESCE(SUM(cost_usd), 0) as total_usd,
                COALESCE(SUM(cost_cny), 0) as total_cny,
                COUNT(*) as request_count
            FROM token_usage 
            WHERE timestamp LIKE ?
        """, (f"{year_month}%",))
        
        row = cursor.fetchone()
        return {
            "total_usd": row[0],
            "total_cny": row[1],
            "request_count": row[2]
        }


全局实例

tracker = CostTracker()

策略层：智能模型路由与 Token 优化

"""
AI API 智能路由层
根据请求复杂度自动选择最优模型
"""

import hashlib
import json
from enum import Enum
from typing import Optional, Callable
from dataclasses import dataclass

class RequestComplexity(Enum):
    SIMPLE = "simple"      # 简单问答、翻译
    MODERATE = "moderate"  # 摘要、改写
    COMPLEX = "complex"    # 代码、多轮对话
    ADVANCED = "advanced"  # 复杂推理、长文本

class ModelRouter:
    """智能模型路由"""
    
    # 模型配置：优先选择高性价比方案
    MODEL_MAPPING = {
        RequestComplexity.SIMPLE: {
            "primary": "deepseek-v3.2",      # $0.42/MTok output，性价比最高
            "fallback": "gemini-2.5-flash",  # $2.50/MTok output
            "max_tokens": 512,
            "temperature": 0.3
        },
        RequestComplexity.MODERATE: {
            "primary": "gemini-2.5-flash",   # 平衡成本与效果
            "fallback": "deepseek-chat",
            "max_tokens": 2048,
            "temperature": 0.5
        },
        RequestComplexity.COMPLEX: {
            "primary": "gpt-4.1-mini",       # 比 GPT-4.1 便宜 5 倍
            "fallback": "claude-haiku-3.5",
            "max_tokens": 8192,
            "temperature": 0.7
        },
        RequestComplexity.ADVANCED: {
            "primary": "gpt-4.1",            # 复杂推理专用
            "fallback": "claude-sonnet-4.5",
            "max_tokens": 16384,
            "temperature": 0.7
        }
    }
    
    # 关键词匹配规则
    COMPLEXITY_KEYWORDS = {
        RequestComplexity.SIMPLE: [
            "翻译", "translate", "什么意思", "定义",
            "翻译成", "改写为英文", "一句话概括"
        ],
        RequestComplexity.MODERATE: [
            "总结", "摘要", "概括", "润色", "优化",
            "expand", "summarize", "paraphrase"
        ],
        RequestComplexity.COMPLEX: [
            "写代码", "debug", "解释代码", "实现",
            "write code", "implement", "fix bug", "refactor"
        ],
        RequestComplexity.ADVANCED: [
            "分析", "对比", "推理", "设计架构",
            "analyze", "reasoning", "architecture", "design"
        ]
    }
    
    def classify_request(self, prompt: str) -> RequestComplexity:
        """根据 Prompt 内容分类请求复杂度"""
        prompt_lower = prompt.lower()
        
        # 先检查关键词
        for complexity, keywords in self.COMPLEXITY_KEYWORDS.items():
            for keyword in keywords:
                if keyword.lower() in prompt_lower:
                    return complexity
        
        # 检查 Prompt 长度
        prompt_length = len(prompt)
        if prompt_length > 3000:
            return RequestComplexity.ADVANCED
        elif prompt_length > 1000:
            return RequestComplexity.COMPLEX
        elif prompt_length > 200:
            return RequestComplexity.MODERATE
        else:
            return RequestComplexity.SIMPLE
    
    def get_model_config(
        self, 
        prompt: str, 
        force_model: Optional[str] = None
    ) -> dict:
        """获取模型配置"""
        if force_model:
            return {
                "model": force_model,
                "max_tokens": 4096,
                "temperature": 0.7
            }
        
        complexity = self.classify_request(prompt)
        config = self.MODEL_MAPPING[complexity].copy()
        return config
    
    def estimate_cost(self, prompt: str, model: str) -> dict:
        """预估请求成本"""
        # 粗略估算 Token 数量 (中文约 2 chars/token, 英文约 4 chars/token)
        estimated_input_tokens = len(prompt) // 2
        estimated_output_tokens = 500  # 默认预估
        
        # 获取模型价格
        from cost_tracker import CostTracker
        tracker = CostTracker()
        cost_usd, cost_cny = tracker.calculate_cost(
            model, estimated_input_tokens, estimated_output_tokens
        )
        
        return {
            "estimated_input_tokens": estimated_input_tokens,
            "estimated_output_tokens": estimated_output_tokens,
            "estimated_cost_usd": round(cost_usd, 6),
            "estimated_cost_cny": round(cost_cny, 6)
        }


全局实例

router = ModelRouter()

主流 AI API 价格对比表

在开始成本控制方案前，我们需要清楚了解各平台的价格差异。以下是 2026 年 Q1 最新价格数据：

模型 供应商 Input ($/MTok) Output ($/MTok) 国内延迟 汇率优势 推荐场景
DeepSeek V3.2 HolySheep $0.14 $0.42 <50ms ¥1=$1 简单问答、翻译
Gemini 2.5 Flash HolySheep $0.35 $2.50 <50ms ¥1=$1 中等复杂度任务
GPT-4.1-mini HolySheep $0.40 $1.60 <50ms ¥1=$1 代码生成、复杂对话
GPT-4.1 HolySheep $2.00 $8.00 <50ms ¥1=$1 高精度推理
Claude Sonnet 4.5 HolySheep $3.00 $15.00 <50ms ¥1=$1 复杂分析
GPT-4.1 OpenAI 官方 $2.00 $8.00 200-500ms ¥7.3=$1 通用场景
Claude Sonnet 4.5 Anthropic 官方 $3.00 $15.00 300-800ms ¥7.3=$1 复杂分析

通过对比可以看到，同样使用 HolySheep AI 的 DeepSeek V3.2 比官方 OpenAI GPT-4.1 输出成本低 95%，而且国内延迟从 300ms+ 降低到 50ms 以内。

生产级 API 调用封装

下面是支持 HolySheep API 的完整生产级封装，包含缓存、重试、限流和成本追踪：

"""
生产级 AI API 客户端
支持 HolySheep API (https://api.holysheep.ai/v1)
集成缓存、重试、限流、成本追踪
"""

import os
import time
import hashlib
import asyncio
import aiohttp
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
import json

@dataclass
class APIResponse:
    content: str
    model: str
    input_tokens: int
    output_tokens: int
    total_tokens: int
    cost_usd: float
    cost_cny: float
    latency_ms: float
    cached: bool = False

class AIBudgetClient:
    """带预算控制的 AI API 客户端"""
    
    def __init__(
        self,
        api_key: str = "YOUR_HOLYSHEEP_API_KEY",  # 替换为你的 HolySheep API Key
        base_url: str = "https://api.holysheep.ai/v1",
        daily_budget_usd: float = 100.0,
        monthly_budget_usd: float = 2000.0,
        enable_cache: bool = True,
        cache_ttl: int = 3600
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.daily_budget = daily_budget_usd
        self.monthly_budget = monthly_budget_usd
        self.enable_cache = enable_cache
        self.cache_ttl = cache_ttl
        
        # 导入成本追踪器
        from cost_tracker import CostTracker
        self.tracker = CostTracker()
        
        # 导入模型路由
        from model_router import ModelRouter
        self.router = ModelRouter()
        
        # 内存缓存 (生产环境建议用 Redis)
        self._cache: Dict[str, tuple[str, float]] = {}
        
        # 限流器
        self._rate_limiter = asyncio.Semaphore(50)  # 最大 50 并发
        
        # Session (复用连接)
        self._session: Optional[aiohttp.ClientSession] = None
    
    async def _get_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            timeout = aiohttp.ClientTimeout(total=60)
            self._session = aiohttp.ClientSession(timeout=timeout)
        return self._session
    
    def _get_cache_key(self, model: str, messages: List[Dict]) -> str:
        """生成缓存 Key"""
        content = json.dumps({"model": model, "messages": messages}, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()
    
    def _check_cache(self, cache_key: str) -> Optional[str]:
        """检查缓存"""
        if not self.enable_cache:
            return None
        
        if cache_key in self._cache:
            cached_content, timestamp = self._cache[cache_key]
            if time.time() - timestamp < self.cache_ttl:
                return cached_content
            else:
                del self._cache[cache_key]
        return None
    
    def _set_cache(self, cache_key: str, content: str):
        """设置缓存"""
        if self.enable_cache:
            self._cache[cache_key] = (content, time.time())
    
    def _check_budget(self) -> bool:
        """检查预算限制"""
        daily_spent = self.tracker.get_daily_spent()["total_usd"]
        monthly_spent = self.tracker.get_monthly_spent()["total_usd"]
        
        if daily_spent >= self.daily_budget:
            print(f"⚠️ 每日预算超限: ${daily_spent:.2f} / ${self.daily_budget:.2f}")
            return False
        
        if monthly_spent >= self.monthly_budget:
            print(f"⚠️ 每月预算超限: ${monthly_spent:.2f} / ${self.monthly_budget:.2f}")
            return False
        
        return True
    
    async def chat(
        self,
        messages: List[Dict[str, str]],
        model: Optional[str] = None,
        force_model: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 2048,
        **kwargs
    ) -> APIResponse:
        """发送聊天请求"""
        
        # 预算检查
        if not self._check_budget():
            raise ValueError("Budget exceeded")
        
        # 模型选择
        if force_model:
            selected_model = force_model
        elif model:
            selected_model = model
        else:
            # 从最后一条用户消息推断
            user_prompt = messages[-1]["content"] if messages else ""
            config = self.router.get_model_config(user_prompt)
            selected_model = config["primary"]
            max_tokens = config.get("max_tokens", max_tokens)
            temperature = config.get("temperature", temperature)
        
        # 缓存检查
        cache_key = self._get_cache_key(selected_model, messages)
        cached_content = self._check_cache(cache_key)
        
        if cached_content:
            return APIResponse(
                content=cached_content,
                model=selected_model,
                input_tokens=0,
                output_tokens=0,
                total_tokens=0,
                cost_usd=0,
                cost_cny=0,
                latency_ms=0,
                cached=True
            )
        
        # 限流
        async with self._rate_limiter:
            start_time = time.time()
            
            session = await self._get_session()
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
            
            payload = {
                "model": selected_model,
                "messages": messages,
                "temperature": temperature,
                "max_tokens": max_tokens,
                **kwargs
            }
            
            try:
                async with session.post(
                    f"{self.base_url}/chat/completions",
                    headers=headers,
                    json=payload
                ) as response:
                    if response.status == 429:
                        # 限流重试 with exponential backoff
                        for retry in range(3):
                            await asyncio.sleep(2 ** retry)
                            async with session.post(
                                f"{self.base_url}/chat/completions",
                                headers=headers,
                                json=payload
                            ) as retry_response:
                                if retry_response.status == 200:
                                    response = retry_response
                                    break
                        else:
                            raise Exception("Rate limit exceeded after retries")
                    
                    data = await response.json()
                    
                    if "error" in data:
                        raise Exception(f"API Error: {data['error']}")
                    
                    # 解析响应
                    choice = data["choices"][0]["message"]
                    content = choice["content"]
                    
                    # 计算 Token (需要根据实际响应获取)
                    # HolySheep API 返回 usage 字段
                    usage = data.get("usage", {})
                    input_tokens = usage.get("prompt_tokens", 0)
                    output_tokens = usage.get("completion_tokens", 0)
                    total_tokens = usage.get("total_tokens", input_tokens + output_tokens)
                    
                    # 计算成本
                    cost_usd, cost_cny = self.tracker.calculate_cost(
                        selected_model, input_tokens, output_tokens,
                        provider="holysheep"
                    )
                    
                    latency_ms = (time.time() - start_time) * 1000
                    
                    # 记录使用量
                    from cost_tracker import TokenUsage
                    self.tracker.record(TokenUsage(
                        timestamp=time.strftime("%Y-%m-%d %H:%M:%S"),
                        model=selected_model,
                        input_tokens=input_tokens,
                        output_tokens=output_tokens,
                        total_tokens=total_tokens,
                        cost_usd=cost_usd,
                        cost_cny=cost_cny,
                        provider="holysheep",
                        request_id=data.get("id", "unknown")
                    ))
                    
                    # 设置缓存
                    self._set_cache(cache_key, content)
                    
                    return APIResponse(
                        content=content,
                        model=selected_model,
                        input_tokens=input_tokens,
                        output_tokens=output_tokens,
                        total_tokens=total_tokens,
                        cost_usd=cost_usd,
                        cost_cny=cost_cny,
                        latency_ms=latency_ms,
                        cached=False
                    )
                    
            except aiohttp.ClientError as e:
                print(f"⚠️ Network error: {e}, retrying...")
                raise
    
    async def batch_chat(
        self,
        requests: List[Dict[str, Any]],
        concurrency: int = 10
    ) -> List[APIResponse]:
        """批量请求 (带并发控制)"""
        semaphore = asyncio.Semaphore(concurrency)
        
        async def bounded_chat(req):
            async with semaphore:
                return await self.chat(**req)
        
        tasks = [bounded_chat(req) for req in requests]
        return await asyncio.gather(*tasks, return_exceptions=True)
    
    async def close(self):
        """关闭连接"""
        if self._session and not self._session.closed:
            await self._session.close()


使用示例

async def main():
    client = AIBudgetClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        daily_budget_usd=50.0,  # 每日预算 $50
        monthly_budget_usd=1000.0  # 每月预算 $1000
    )
    
    try:
        # 简单问答 (会自动选择 DeepSeek V3.2)
        response = await client.chat(messages=[
            {"role": "user", "content": "请用一句话解释量子计算"}
        ])
        print(f"回复: {response.content}")
        print(f"模型: {response.model}, 成本: ${response.cost_usd:.6f}")
        
        # 复杂任务 (会自动选择 GPT-4.1-mini)
        response = await client.chat(messages=[
            {"role": "user", "content": "写一个 Python 快排算法，包含单元测试"}
        ])
        print(f"回复: {response.content[:200]}...")
        print(f"模型: {response.model}, 成本: ${response.cost_usd:.6f}")
        
    finally:
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

成本优化实战：从 $3000/月 降到 $400/月

我曾帮助一家内容生成创业公司优化 AI 成本。他们的场景是每天生成 5000 篇 SEO 文章，使用 GPT-4o 官方 API，月账单高达 $3000。下面是我做的优化：

优化一：模型降级

经过分析，80% 的文章生成是简单的内容改写和扩展，完全可以用 DeepSeek V3.2 替代 GPT-4o。

优化二：引入语义缓存

分析日志发现，有 30% 的请求是重复或高度相似的 (相同关键词、不同表述)。引入语义缓存后：

优化三：Prompt 压缩

原始 Prompt 平均长度 2000 tokens，优化后平均 800 tokens：

优化四：切换到 HolySheep

使用 HolySheep AI 后：

最终效果

指标 优化前 优化后 改善
月费用 $3,000 $400 ↓87%
平均延迟 350ms 45ms ↓87%
缓存命中率 0% 28% 新增
模型选择 GPT-4o 100% 混合 智能路由

常见错误与解决方案

错误一：Token 计算错误导致预算偏差

很多开发者直接用字符数除以 4 来估算 Token，这对于英文有效，但中文完全错误。中文平均 1.5-2 个字符对应 1 个 Token。

# ❌ 错误估算
def wrong_token_count(text: str) -> int:
    return len(text) // 4  # 中文严重低估


✅ 正确做法：使用 tiktoken

from tiktoken import encoding_for_model

def correct_token_count(text: str, model: str = "gpt-4") -> int:
    enc = encoding_for_model(model)
    return len(enc.encode(text))


或者使用 HolySheep 返回的 usage 字段


API 响应中的 usage.prompt_tokens 和 usage.completion_tokens

错误二：无限重试导致雪崩

当 API 限流时，无限重试会让问题恶化。我见过一个案例，因为没有退避策略，重试产生的费用是正常请求的 15 倍。

# ❌ 危险重试
while True:
    try:
        response = await client.chat(...)
        break
    except Exception as e:
        print(f"Error: {e}")
        # 没有 sleep，会直接打爆 API


✅ 安全重试 with exponential backoff

async def safe_retry(func, max_retries=3, base_delay=1.0):
    for attempt in range(max_retries):
        try:
            return await func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            
            # 根据错误类型决定是否重试
            error_msg = str(e).lower()
            if "rate limit" in error_msg or "timeout" in error_msg:
                delay = base_delay * (2 ** attempt)  # 1s, 2s, 4s
                print(f"Retry {attempt + 1}/{max_retries} after {delay}s")
                await asyncio.sleep(delay)
            else:
                # 其他错误不重试
                raise

错误三：忽视 Stream 模式的成本

很多人以为 Stream 模式更便宜，实际上成本完全一样，而且因为每次传输更小的数据块，网络开销反而更大。

# Stream vs 非 Stream 成本相同

Stream 的优势是响应更快，用户体验更好


不是节省成本的手段



如果需要节省成本，正确做法是：


1. 减少 max_tokens 限制


2. 使用更小的模型


3. 压缩 Prompt


4. 引入缓存



❌ Stream 不能节省成本

payload = {
    "model": "gpt-4o",
    "messages": messages,
    "stream": True,  # 成本一样
    "max_tokens": 4096
}


✅ 正确节省成本

payload = {
    "model": "deepseek-v3.2",  # 更便宜的模型
    "messages": compress_prompt(messages),  # 压缩 Prompt
    "max_tokens": 512,  # 限制输出长度
    "stream": False
}

常见报错排查

报错一：401 Unauthorized

# 错误信息

{"error": {"message": "Incorrect API key provided", "type": "invalid_request_error"}}



排查步骤


1. 检查 API Key 是否正确


2. 确认使用的是 HolySheep Key，不是 OpenAI Key


3. 检查 Key 是否已过期或被禁用


4. 确认 base_url 是否正确



✅ 正确配置

client = AIBudgetClient(
    api_key="YOUR_HOLYSHEEP_API_KEY",  # 从 https://www.holysheep.ai 获取
    base_url="https://api.holysheep.ai/v1"
)

报错二：429 Rate Limit Exceeded

# 错误信息

{"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}



排查步骤


1. 检查当前并发请求数是否超过限制


2. 查看每日/每月配额是否用完


3. 实现请求队列和限流



✅ 限流实现

class RateLimiter:
    def __init__(self, max_rpm: int = 60):
        self.max_rpm = max_rpm
        self.requests = []
        self.lock = asyncio.Lock()
    
    async def acquire(self):
        async with self.lock:
            now = time.time()
            # 移除一分钟前的请求
            self.requests = [t for t in self.requests if now - t < 60]
            
            if len(self.requests) >= self.max_rpm:
                sleep_time = 60 - (now - self.requests[0])
                await asyncio.sleep(sleep_time)
            
            self.requests.append(now)

报错三：400 Bad Request - context_length_exceeded

# 错误信息

{"error": {"message": "maximum context length exceeded", "type": "invalid_request_error"}}



原因：输入 Token 超过模型上下文窗口


例如 GPT-4.1 支持 128K tokens，但你的请求超过了



✅ 解决方案：截断或压缩输入

def truncate_messages(messages: list, max_tokens: int = 100000):
    """截断消息列表以适应上下文限制"""
    # 计算当前 token 数
    total_tokens = 0
    truncated = []
    
    for msg in reversed(messages):
        # 粗略估算 (

相关资源
相关文章