max_tokens是调用AI API时最容易被忽视、却对成本和性能影响最大的参数之一。在本文ază完整的案例研究中，展示如何通过科学设置该参数实现70%成本reduktion和3倍Latenzverbesserung。

案例研究：慕尼黑电商团队的AI转型之路

企业背景与业务挑战

这家总部位于慕尼黑的时尚电商企业，拥有约200万月度活跃用户，此前依赖某美国主流AI服务商处理商品描述生成、客服问答和搜索推荐功能。业务团队反馈的三大痛点：

迁移至HolySheep的决策过程

经过技术评估，团队选择了HolySheep AI作为核心AI基础设施，核心优势包括：

具体迁移步骤

1. 基础配置切换

# 迁移前配置（某美国服务商）
import openai

client = openai.OpenAI(
    api_key="OLD_API_KEY",
    base_url="https://api.openai.com/v1"  # ← 需替换
)

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "生成商品描述"}],
    max_tokens=2000  # ← 过大的默认值
)


迁移后配置（HolySheep）

import openai

client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"  # ← 仅修改此处
)

response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": "生成商品描述"}],
    max_tokens=256  # ← 精准设置
)

2. Canary Deployment策略

import os
import random
from openai import OpenAI

class AIBridge:
    def __init__(self):
        self.primary = OpenAI(
            api_key=os.environ.get("HOLYSHEEP_API_KEY"),
            base_url="https://api.holysheep.ai/v1"
        )
        self.canary_ratio = 0.1  # 10%流量走新服务
    
    def complete(self, prompt: str, context: str = "", max_tokens: int = 256) -> dict:
        messages = []
        if context:
            messages.append({"role": "assistant", "content": context})
        messages.append({"role": "user", "content": prompt})
        
        # Canary判断
        if random.random() < self.canary_ratio:
            try:
                return self._call_holysheep(messages, max_tokens)
            except Exception as e:
                print(f"Canary失败，回退: {e}")
        
        return self._call_holysheep(messages, max_tokens)
    
    def _call_holysheep(self, messages: list, max_tokens: int) -> dict:
        response = self.primary.chat.completions.create(
            model="deepseek-v3.2",
            messages=messages,
            max_tokens=max_tokens,
            temperature=0.7
        )
        return {
            "content": response.choices[0].message.content,
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens
            }
        }


使用示例

bridge = AIBridge()
result = bridge.complete("为这件红色连衣裙写一段描述", max_tokens=128)
print(f"消耗Tokens: {result['usage']['total_tokens']}")

3. API Key轮换机制

import os
import time
from datetime import datetime, timedelta
from openai import OpenAI

class KeyRotator:
    def __init__(self, api_keys: list):
        self.keys = api_keys
        self.current_index = 0
        self.last_rotation = datetime.now()
        self.rotation_interval = timedelta(hours=24)
        self.usage_count = {k: 0 for k in api_keys}
    
    def get_client(self) -> OpenAI:
        self._check_rotation()
        return OpenAI(
            api_key=self.keys[self.current_index],
            base_url="https://api.holysheep.ai/v1"
        )
    
    def _check_rotation(self):
        if datetime.now() - self.last_rotation > self.rotation_interval:
            self.current_index = (self.current_index + 1) % len(self.keys)
            self.last_rotation = datetime.now()
            print(f"Key轮换至索引{self.current_index}")
    
    def record_usage(self, tokens: int):
        self.usage_count[self.keys[self.current_index]] += tokens
    
    def get_stats(self) -> dict:
        return {
            "current_key": self.keys[self.current_index][:8] + "***",
            "usage_distribution": self.usage_count,
            "uptime_hours": (datetime.now() - self.last_rotation).total_seconds() / 3600
        }


生产环境使用

keys = [
    os.environ.get("HOLYSHEEP_KEY_1"),
    os.environ.get("HOLYSHEEP_KEY_2"),
    os.environ.get("HOLYSHEEP_KEY_3")
]
rotator = KeyRotator(keys)


模拟API调用

client = rotator.get_client()
response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": "测试消息"}],
    max_tokens=64
)
rotator.record_usage(response.usage.total_tokens)
print(rotator.get_stats())

30天关键指标对比

指标迁移前迁移后改善幅度
平均响应延迟420ms180ms-57.1%
月API账单$4.200$680-83.8%
Token利用率34%89%+161.8%
P99延迟2.800ms620ms-77.9%
超时错误率3.2%0.08%-97.5%

max_tokens参数深度解析

什么是max_tokens？

max_tokens定义了模型单次响应生成最大token数量上限。注意：这是上限而非目标值，实际输出取决于模型对任务复杂度的判断。

HolySheep 2026年最新定价

模型价格/MTok适用场景
DeepSeek V3.2$0.42长文本生成、批量处理
Gemini 2.5 Flash$2.50快速响应、实时交互
GPT-4.1$8.00复杂推理、高质量输出
Claude Sonnet 4.5$15.00长文档分析、代码生成

场景化max_tokens设置策略

1. 短问答场景（max_tokens: 64-128）

import openai

client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

def quick_qa(question: str) -> str:
    response = client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[
            {"role": "system", "content": "你是一个简洁的助手，直接回答问题。"},
            {"role": "user", "content": question}
        ],
        max_tokens=96,  # 约60-80个中文字
        temperature=0.3,
        timeout=5.0
    )
    return response.choices[0].message.content


测试

print(quick_qa("德语中'你好'怎么说？"))

预期输出简短，96 tokens完全足够

2. 商品描述生成（max_tokens: 150-300）

import openai
import json

client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

def generate_product_description(product_info: dict) -> str:
    prompt = f"""为以下商品生成一段150字以内的中文描述：

商品名称：{product_info['name']}
材质：{product_info.get('material', '优质面料')}
特点：{product_info.get('features', '舒适耐穿')}
洗涤建议：{product_info.get('care', '手洗')}

格式要求：
- 使用情感化语言
- 突出核心卖点
- 包含购买引导"""

    response = client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=256,  # 留出余量，允许更详细输出
        temperature=0.7
    )
    return response.choices[0].message.content


批量处理示例

products = [
    {"name": "北欧风纯棉T恤", "material": "100%有机棉", "features": "透气、速干"},
    {"name": "修身牛仔裤", "material": " denim", "features": "高弹、显瘦"}
]

for p in products:
    desc = generate_product_description(p)
    print(f"【{p['name']}】{desc}\n")

3. 结构化JSON输出（max_tokens: 512-1024）

import openai
import json
from typing import TypedDict

class ProductSchema(TypedDict):
    title: str
    price: float
    currency: str
    description: str
    tags: list[str]

client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

def extract_product_structured(raw_text: str) -> ProductSchema:
    """从非结构化文本提取产品信息"""
    
    response = client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[
            {
                "role": "system", 
                "content": "你是一个JSON生成器。只输出有效的JSON，不要任何其他文字。"
            },
            {
                "role": "user", 
                "content": f"从以下文本提取产品信息：\n\n{raw_text}\n\nJSON格式：{{title, price, currency, description, tags}}"
            }
        ],
        max_tokens=512,
        response_format={"type": "json_object"}
    )
    
    raw = response.choices[0].message.content
    # 解析并验证
    data = json.loads(raw)
    
    # 关键：验证并补充默认值
    return ProductSchema(
        title=data.get("title", ""),
        price=float(data.get("price", 0)),
        currency=data.get("currency", "CNY"),
        description=data.get("description", "")[:200],  # 截断保护
        tags=data.get("tags", [])[:10]  # 限制标签数量
    )


测试

test_text = "这件红色连衣裙售价299元，采用丝绸面料，适合春秋季节穿着，优雅大方"
result = extract_product_structured(test_text)
print(json.dumps(result, ensure_ascii=False, indent=2))

4. 长文档处理（max_tokens: 2048-4096）

import openai
from concurrent.futures import ThreadPoolExecutor

client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

def summarize_long_text(text: str, chunk_size: int = 2000) -> str:
    """分块处理长文本，智能合并摘要"""
    
    # 文本分块（按字符，保留边界）
    chunks = []
    for i in range(0, len(text), chunk_size):
        chunk = text[i:i+chunk_size]
        # 确保在句号处断开
        if i + chunk_size < len(text):
            last_period = chunk.rfind('。')
            if last_period > chunk_size * 0.7:
                chunks.append(chunk[:last_period+1])
                i = i + last_period
            else:
                chunks.append(chunk)
        else:
            chunks.append(chunk)
    
    def summarize_chunk(chunk: str) -> str:
        response = client.chat.completions.create(
            model="deepseek-v3.2",
            messages=[
                {"role": "system", "content": "提取本段核心观点，用一句话概括。"},
                {"role": "user", "content": chunk}
            ],
            max_tokens=256,
            temperature=0.3
        )
        return response.choices[0].message.content
    
    # 并行处理各块
    with ThreadPoolExecutor(max_workers=5) as executor:
        summaries = list(executor.map(summarize_chunk, chunks))
    
    # 合并所有摘要
    combined = " ".join(summaries)
    
    # 最终整合
    final_response = client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[
            {"role": "system", "content": "基于以下要点，生成一篇连贯的摘要。"},
            {"role": "user", "content": combined}
        ],
        max_tokens=512,
        temperature=0.5
    )
    
    return final_response.choices[0].message.content


性能监控

import time
start = time.time()
long_text = "..." * 1000  # 模拟长文本
result = summarize_long_text(long_text)
print(f"处理耗时: {time.time() - start:.2f}秒")

自适应max_tokens实现方案

import openai
from enum import Enum
from typing import Optional

class TaskComplexity(Enum):
    TRIVIAL = ("trivial", 64, 96)
    SIMPLE = ("simple", 128, 192)
    STANDARD = ("standard", 256, 384)
    COMPLEX = ("complex", 512, 768)
    EXPERT = ("expert", 1024, 1536)
    
    def __init__(self, name: str, min_tokens: int, max_tokens: int):
        self.name = name
        self.min_tokens = min_tokens
        self.max_tokens = max_tokens

class AdaptiveAI:
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"
        )
        self.default_model = "deepseek-v3.2"
        self.usage_log = []
    
    def estimate_complexity(self, prompt: str) -> TaskComplexity:
        """基于启发式规则估算任务复杂度"""
        word_count = len(prompt)
        has_code = "```" in prompt or "def " in prompt
        has_list = any(kw in prompt for kw in ["列表", "列出", "列出10"])
        has_analyze = any(kw in prompt for kw in ["分析", "比较", "评估"])
        has_long = any(kw in prompt for kw in ["详细", "完整", "深入"])
        
        score = 0
        if word_count > 200: score += 2
        elif word_count > 100: score += 1
        if has_code: score += 2
        if has_list: score += 1
        if has_analyze: score += 2
        if has_long: score += 1
        
        if score >= 7: return TaskComplexity.EXPERT
        if score >= 5: return TaskComplexity.COMPLEX
        if score >= 3: return TaskComplexity.STANDARD
        if score >= 1: return TaskComplexity.SIMPLE
        return TaskComplexity.TRIVIAL
    
    def complete(self, prompt: str, complexity: Optional[TaskComplexity] = None) -> dict:
        if complexity is None:
            complexity = self.estimate_complexity(prompt)
        
        response = self.client.chat.completions.create(
            model=self.default_model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=complexity.max_tokens
        )
        
        result = {
            "content": response.choices[0].message.content,
            "complexity": complexity.name,
            "tokens_used": response.usage.total_tokens,
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens
        }
        
        self.usage_log.append(result)
        return result
    
    def get_cost_optimization_report(self) -> dict:
        """生成成本优化报告"""
        if not self.usage_log:
            return {"message": "暂无数据"}
        
        total_tokens = sum(r["tokens_used"] for r in self.usage_log)
        complexity_dist = {}
        for entry in self.usage_log:
            c = entry["complexity"]
            complexity_dist[c] = complexity_dist.get(c, 0) + 1
        
        avg_usage_ratio = sum(
            entry["completion_tokens"] / entry["tokens_used"] 
            for entry in self.usage_log
        ) / len(self.usage_log) * 100
        
        estimated_cost = total_tokens / 1_000_000 * 0.42  # DeepSeek价格
        
        return {
            "total_requests": len(self.usage_log),
            "total_tokens": total_tokens,
            "complexity_distribution": complexity_dist,
            "avg_token_usage_ratio": f"{avg_usage_ratio:.1f}%",
            "estimated_cost_usd": f"${estimated_cost:.4f}",
            "optimization_tip": "建议降低complexity阈值" if avg_usage_ratio < 60 else "使用合理"
        }


使用示例

ai = AdaptiveAI("YOUR_HOLYSHEEP_API_KEY")

test_prompts = [
    "你好",  # TRIVIAL
    "解释什么是机器学习",  # SIMPLE
    "帮我写一个Python函数计算斐波那契数列",  # STANDARD
    "详细分析2024年中国电商市场趋势，包含数据对比和未来预测",  # COMPLEX
]

for prompt in test_prompts:
    result = ai.complete(prompt)
    print(f"[{result['complexity']}] {result['tokens_used']} tokens")

成本计算与预算控制

import openai
from datetime import datetime, timedelta
from typing import Optional

class BudgetController:
    """ HolySheep API 成本控制器 """
    
    PRICES = {
        "deepseek-v3.2": 0.42,
        "gemini-2.5-flash": 2.50,
        "gpt-4.1": 8.00,
        "claude-sonnet-4.5": 15.00
    }
    
    def __init__(self, api_key: str, monthly_budget_usd: float):
        self.client = openai.OpenAI(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1"
        )
        self.monthly_budget = monthly_budget_usd
        self.daily_limit = monthly_budget_usd / 30
        self.current_month_spend = 0.0
        self.current_day_spend = 0.0
        self.month_start = datetime.now().replace(day=1)
        self.day_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
    
    def _reset_daily_if_needed(self):
        if datetime.now().date() > self.day_start.date():
            self.current_day_spend = 0.0
            self.day_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
    
    def _reset_monthly_if_needed(self):
        if datetime.now().month != self.month_start.month:
            self.current_month_spend = 0.0
            self.month_start = datetime.now().replace(day=1)
    
    def _calculate_cost(self, model: str, tokens: int) -> float:
        return tokens / 1_000_000 * self.PRICES.get(model, 1.0)
    
    def can_proceed(self, model: str, estimated_tokens: int = 1000) -> tuple[bool, str]:
        self._reset_daily_if_needed()
        self._reset_monthly_if_needed()
        
        estimated_cost = self._calculate_cost(model, estimated_tokens)
        
        if self.current_month_spend + estimated_cost > self.monthly_budget:
            return False, f"月度预算超限 (剩余: ${self.monthly_budget - self.current_month_spend:.2f})"
        
        if self.current_day_spend + estimated_cost > self.daily_limit:
            return False, f"日度限额超限 (限制: ${self.daily_limit:.2f})"
        
        return True, "OK"
    
    def complete_with_budget_check(self, prompt: str, model: str = "deepseek-v3.2", 
                                   max_tokens: int = 256) -> Optional[dict]:
        """带预算检查的API调用"""
        
        can_proceed, reason = self.can_proceed(model)
        if not can_proceed:
            print(f"⚠️ 请求被阻止: {reason}")
            return None
        
        response = self.client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=max_tokens
        )
        
        tokens = response.usage.total_tokens
        cost = self._calculate_cost(model, tokens)
        
        self.current_month_spend += cost
        self.current_day_spend += cost
        
        return {
            "content": response.choices[0].message.content,
            "tokens": tokens,
            "cost_usd": cost,
            "month_total": self.current_month_spend,
            "day_total": self.current_day_spend
        }
    
    def get_status(self) -> dict:
        self._reset_daily_if_needed()
        return {
            "monthly_budget": f"${self.monthly_budget:.2f}",
            "month_spent": f"${self.current_month_spend:.2f}",
            "month_remaining": f"${self.monthly_budget - self.current_month_spend:.2f}",
            "daily_limit": f"${self.daily_limit:.2f}",
            "day_spent": f"${self.current_day_spend:.2f}",
            "day_remaining": f"${self.daily_limit - self.current_day_spend:.2f}",
            "usage_percentage": f"{self.current_month_spend / self.monthly_budget * 100:.1f}%"
        }


使用示例

controller = BudgetController(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    monthly_budget_usd=500.0
)


检查状态

print(controller.get_status())


执行调用

result = controller.complete_with_budget_check(
    prompt="为我的电商网站生成5个产品标签",
    model="deepseek-v3.2",
    max_tokens=128
)

if result:
    print(f"✅ 成功: {result['tokens']} tokens, 成本: ${result['cost_usd']:.4f}")

我的实战经验：max_tokens优化的三个关键认知

作为一名长期从事AI应用开发的技术负责人，我在实际项目中总结出max_tokens优化的三个核心原则：

1. 先测量，后优化

切忌凭直觉设置max_tokens。我建议在正式上线前，至少收集1000次真实调用的token分布数据。通过分析p25/p50/p75/p95/p99分位数，可以找到最优的平衡点。

2. 场景分级策略

不同业务场景对输出长度需求差异巨大：

3. 动态调整机制

静态配置无法适应业务变化。建议实现基于历史数据的自适应调节：每24小时重新评估各场景的token消耗中位数，自动微调max_tokens阈值。

Häufige Fehler und Lösungen

错误1：max_tokens设置过大导致成本浪费

问题现象：每次调用无论问题复杂度，响应token数都接近max_tokens上限，造成严重浪费。

# ❌ 错误配置：所有请求都用4096 tokens上限
response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": "今天天气如何？"}],
    max_tokens=4096  # 完全没必要！
)


✅ 正确配置：根据场景选择合适上限

response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": "今天天气如何？"}],
    max_tokens=64  # 简短回答足够
)

错误2：max_tokens过小导致输出截断

问题现象：生成的内容在中间被截断，用户看到不完整的信息。

# ❌ 错误配置：为长文本生成设置过小上限
response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": "请详细解释量子计算的原理和发展历史..."}],
    max_tokens=128  # 严重不足！
)


✅ 正确配置：为复杂任务预留足够空间

response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": "请详细解释量子计算的原理和发展历史..."}],
    max_tokens=2048,  # 留足空间
    stop=["---", "注："]  # 设置停止词防止无限制输出
)


截断检测和补救

if response.choices[0].finish_reason == "length":
    print("⚠️ 输出可能被截断，考虑增加max_tokens")

错误3：未处理API超时和重试

问题现象：网络波动时直接失败，用户体验差，无重试机制。

import time
import openai
from openai import RateLimitError, APIError


❌ 错误配置：无任何错误处理

response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": "生成报告"}],
    max_tokens=1024
)


✅ 正确配置：完整重试机制

def robust_complete(prompt: str, max_tokens: int = 1024, max_retries: int = 3) -> dict:
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="deepseek-v3.2",
                messages=[{"role": "user", "content": prompt}],
                max_tokens=max_tokens,
                timeout=30.0
            )
            return {
                "content": response.choices[0].message.content,
                "tokens": response.usage.total_tokens,
                "attempts": attempt + 1
            }
            
        except RateLimitError:
            wait_time = 2 ** attempt  # 指数退避
            print(f"限流，等待{wait_time}秒...")
            time.sleep(wait_time)
            
        except APIError as e:
            if "context_length" in str(e):
                # 处理上下文超限
                print("⚠️ 上下文超限，尝试分段处理")
                return {"error": "context_too_long", "content": None}
            wait_time = 2 ** attempt
            print(f"API错误，等待{wait_time}秒重试...")
            time.sleep(wait_time)
            
        except Exception as e:
            print(f"未知错误: {e}")
            return {"error": str(e), "content": None}
    
    return {"error": "max_retries_exceeded", "content": None}


使用

result = robust_complete("生成完整的产品分析报告", max_tokens=2048)
if result.get("error"):
    print(f"请求失败: {result['error']}")

错误4：多语言混合场景下的token估算偏差

问题现象：中文和英文混合内容导致token数远超预期。

# ❌ 错误配置：假设中文token与英文1:1
MAX_CHARS = 1000
response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": prompt}],
    max_tokens=MAX_CHARS  # 中文约2000 tokens，严重超预算
)


✅ 正确配置：考虑中英文差异（中文≈1.5 tokens/字）

MAX_CHARS = 1000
estimated_tokens = len(prompt) * 1.5 + MAX_CHARS * 1.5  # 双向估算
max_tokens = min(MAX_CHARS * 2, 4096)  # 设置合理上限

response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": prompt}],
    max_tokens=int(max_tokens)
)


监控实际使用

print(f"估算tokens: {estimated_tokens}")
print(f"实际tokens: {response.usage.total_tokens}")
print(f"效率: {response.usage.completion_tokens / response.usage.total_tokens * 100:.1f}%")

错误5：忽视finish_reason判断

问题现象：无法判断响应是否完整，影响业务逻辑判断。

# ❌ 错误配置：只取内容，不检查状态
response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": prompt}],
    max_tokens=512
)
content = response.choices[0].message.content  # 不知道是否完整！


✅ 正确配置：完整检查响应状态

response = client.chat.completions.create(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": prompt}],
    max_tokens=512
)

choice = response.choices[0]
content = choice.message.content
finish_reason = choice.finish_reason

if finish_reason == "stop":
    print("✅ 正常完成")
elif finish_reason == "length":
    print("⚠️ 达到token限制，内容可能被截断")
    # 可选：增加tokens重试
    retry_response = client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[
            {"role": "assistant", "content": content},
            {"role": "user", "content": "请继续上述内容"}
        ],
        max_tokens=512
    )
    content += retry_response.choices[0].message.content
elif finish_reason == "content_filter":
    print("❌ 内容被过滤，请调整prompt")
elif finish_reason == "tool_calls":
    print("🔧 模型请求调用工具，需要处理tool_calls")

性能监控仪表盘实现

import time
from collections import defaultdict
from datetime import datetime

class PerformanceMonitor:
    """ HolySheep API 性能监控器 """
    
    def __init__(self):
        self.requests = []
        self.errors = []
        self.costs = []
    
    def record(self, model: str, tokens: int, latency_ms: float, 
               success: bool, error_msg: str = None):
        entry = {
            "timestamp": datetime.now(),
            "model": model,
            "tokens": tokens,
            "latency_ms": latency_ms,
            "success": success,
            "error": error_msg
        }
        self.requests.append(entry)
        
        if not success:
            self.errors.append(entry)
        else:
            self.costs.append(tokens / 1_000_000 * 0.42)  # DeepSeek价格
    
    def get_stats(self, hours: int = 24) -> dict:
        cutoff = datetime.now().timestamp() - hours * 3600
        recent = [r for r

Verwandte Ressourcen
Verwandte Artikel