DeepSeek 模型多渠道 API 性能评测（20250217 版）

import time
import json
import csv
import concurrent.futures
import datetime
import pytz
import statistics
from openai import OpenAI
from google.colab import userdata
 
 
def test_provider(provider_config, prompts):
    """
    测试单个提供商的 API，支持并发请求、多个 prompt，并将结果保存到 CSV 文件。
    使用 OpenAI 客户端。
    增加了平均首字时间和平均 token/s 的计算和输出。
    """
    provider_name = provider_config.get("name", "Unnamed Provider")
    model_name = provider_config.get("model", "Unknown Model")
    num_threads = 3  # 设置为 3 个线程
    num_prompts_per_thread = 4 # 每个线程处理的任务数
 
    print(f"\n--- 开始测试服务商：{provider_name} (模型：{model_name}) (线程数: {num_threads}) ---")
 
    api_key = userdata.get(provider_config["api_key_env"])
    base_url = provider_config.get("base_url")
 
    if not api_key:
        print(f"错误：未找到 {provider_name} 的 API Key 环境变量 ({provider_config['api_key_env']})")
        return
 
    # 初始化 OpenAI 客户端
    client = OpenAI(api_key=api_key, base_url=base_url)
 
    def single_request(prompt):
        """
        处理单个请求 (与原脚本计算逻辑一致)。
        在请求结束日志中添加了指标信息。
        """
        request_id = f"{provider_name}-{model_name}-{time.time()}"
        print(f"[{request_id}] 请求开始: {prompt[:50]}...")
 
        try:
            start_time = time.time()
            first_token_time = None
            reasoning_tokens = 0
            content_tokens = 0
            prompt_tokens = 0
            total_tokens = 0
            reasoning_time = 0
            content_time = 0
            full_content = ""
            error_message = ""
            reasoning_start_time = None
            content_start_time = None
            last_reasoning_time = None
            last_content_time = None
 
            messages = [{"role": "user", "content": prompt}]
 
            # 发起流式请求
            response = client.chat.completions.create(
                model=model_name,
                messages=messages,
                stream=True,
                temperature=0.6,
                stream_options={"include_usage": True},
            )
 
            for chunk in response:
                if chunk.usage:
                    prompt_tokens = chunk.usage.prompt_tokens
                    completion_tokens = chunk.usage.completion_tokens
                    total_tokens = chunk.usage.total_tokens
                    if chunk.usage.completion_tokens_details is not None:
                        reasoning_tokens = chunk.usage.completion_tokens_details.reasoning_tokens
                    else:
                        reasoning_tokens = 0
                    content_tokens = completion_tokens - reasoning_tokens
 
                if not chunk.choices:
                    continue
                delta = chunk.choices[0].delta
 
                reasoning_piece = getattr(delta, 'reasoning_content', "")
                content_piece = getattr(delta, 'content', "")
 
                if first_token_time is None and (reasoning_piece or content_piece):
                    first_token_time = time.time() - start_time
 
                current_time = time.time()
 
                if reasoning_piece:
                    if reasoning_start_time is None:
                        reasoning_start_time = current_time
                    if last_reasoning_time is not None:
                        reasoning_time += current_time - last_reasoning_time
                    last_reasoning_time = current_time
                    full_content += reasoning_piece
 
                elif content_piece:
                    if content_start_time is None:
                        content_start_time = current_time
                        if last_reasoning_time is not None:
                            reasoning_time += last_reasoning_time - reasoning_start_time
                    if last_content_time is not None:
                        content_time += current_time - last_content_time
                    last_content_time = current_time
                    full_content += content_piece
 
            total_time = time.time() - start_time
            if content_start_time is not None and last_content_time is not None:
                content_time += time.time() - last_content_time
 
            # 计算本次请求的 tokens/s
            tokens_per_second = completion_tokens / total_time if total_time > 0 else None
 
            print(
                f"[{request_id}] 请求结束, "
                f"首字时间: {first_token_time:.3f}s, "
                f"推理tokens: {reasoning_tokens}, "
                f"推理时间: {reasoning_time:.3f}s, "
                f"内容tokens: {content_tokens}, "
                f"内容时间: {content_time:.3f}s, "
                f"总tokens: {total_tokens}, "
                f"总时间: {total_time:.3f}s, "
                f"tokens/s: {tokens_per_second:.2f}" if tokens_per_second is not None else "tokens/s: N/A"
            )
 
            return {
                "provider": provider_name,
                "model": model_name,
                "prompt": prompt,
                "first_token_time": round(first_token_time, 3) if first_token_time is not None else None,
                "reasoning_tokens": reasoning_tokens if reasoning_tokens > 0 else None,
                "reasoning_time": round(reasoning_time, 3) if reasoning_time > 0 else None,
                "content_tokens": content_tokens,
                "content_time": round(content_time, 3) if content_time > 0 else None,
                "total_tokens": total_tokens,
                "total_time": round(total_time, 3),
                "full_content": full_content,
                "temperature": 0.6,
                "error": error_message,
                "success": True,
                "completion_tokens": completion_tokens,
            }
 
        except Exception as e:
            return process_error(f"发生错误：{e}", prompt)
 
    def process_error(error_message, prompt):
        """处理错误，返回错误信息字典"""
        request_id = f"{provider_name}-{model_name}-{time.time()}"
        print(f"[{request_id}] 请求错误: {error_message}")
        return {
            "provider": provider_name,
            "model": model_name,
            "prompt": prompt,
            "first_token_time": None,
            "reasoning_tokens": None,
            "reasoning_time": None,
            "content_tokens": None,
            "content_time": None,
            "total_tokens": None,
            "total_time": None,
            "full_content": None,
            "temperature": 0.6,
            "error": error_message,
            "success": False,
            "completion_tokens": None,
        }
 
    # 使用 ThreadPoolExecutor，固定 3 个线程
    with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
        #futures = [executor.submit(single_request, prompt) for prompt in prompts] # 原来的方式
        futures = [executor.submit(single_request, prompt) for prompt in prompts[:num_threads * num_prompts_per_thread]]
        results = [future.result() for future in concurrent.futures.as_completed(futures)]
 
    timestamp = datetime.datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y%m%d_%H%M%S')
    csv_filename = f"results_{provider_name.replace('/', '_')}_{model_name.replace('/', '_')}_{timestamp}.csv"
 
    with open(csv_filename, 'w', newline='', encoding='utf-8') as csvfile:
        fieldnames = ["provider", "model", "prompt", "first_token_time", "reasoning_tokens",
                      "reasoning_time", "content_tokens", "content_time", "total_tokens", "total_time",
                      "full_content", "temperature", "error", "success", "completion_tokens"]
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(filter(None, results))
 
    success_count = sum(1 for r in results if r and r['success'])
    failure_count = len(results) - success_count
    total_requests = len(results)
    success_rate = (success_count / total_requests) * 100 if total_requests > 0 else 0.0
 
    valid_first_token_times = [
        r['first_token_time'] for r in results if r and r['success'] and r['first_token_time'] is not None
    ]
    avg_first_token_time = statistics.mean(valid_first_token_times) if valid_first_token_times else None
 
    valid_completion_tokens = [
        r['completion_tokens'] for r in results if r and r['success'] and r['completion_tokens'] is not None
    ]
    valid_total_times = [r['total_time'] for r in results if r and r['success'] and r['total_time'] is not None]
 
    if len(valid_completion_tokens) > 0 and len(valid_completion_tokens) == len(valid_total_times):
        avg_tokens_per_second = statistics.mean(
            [tokens / time for tokens, time in zip(valid_completion_tokens, valid_total_times)]
        ) if valid_total_times else None
    else:
        avg_tokens_per_second = None
 
    print(f"测试完成, 结果已保存到文件: {csv_filename}")
    print(f"总请求数: {total_requests}, 成功数: {success_count}, 失败数: {failure_count}, 成功率: {success_rate:.2f}%")
    if avg_first_token_time is not None:
        print(f"平均首字时间: {avg_first_token_time:.3f} 秒")
    else:
        print("平均首字时间: 无法计算 (没有成功的请求或首字时间数据)")
    if avg_tokens_per_second is not None:
        print(f"平均 token/s: {avg_tokens_per_second:.2f}")
    else:
        print("平均 token/s: 无法计算 (没有成功的请求或 token/时间数据)")
 
 
if __name__ == "__main__":
    prompts = [
      # 文本任务 (10个)
      "解读一下这条推文：'@naval The currency of life isn't money. It is not even time. It's attention.' 请结合 Naval Ravikant 的背景和观点进行解释。",
 
      "将以下中文推文翻译成英文，并以自然的口感回复给推特网友：'目前的流程和提示词仅支持英文翻译为中文，不过只要适当调整一下流程和提示词，应该可以支持任意语言的翻译。'",
 
      "帮我润色并扩充这篇推文，使其更适合在技术社区 V2EX 上发布，并吸引开发者关注：\n```\n花了两天和 Cursor 重新设计了个人博客，这风格实在是太爱了，最近把之前的文章搬过来，2025 年好好写博客。\n博客是基于 Next.js 和 mdx 的静态博客，可以在本地使用 cursor 编写文章。\n网址：ginonotes.com\n```",
 
      "将以下段落用中文重写，保持原有含义，使其更符合中文技术博客的表达习惯：\n```\nWhen building LLM applications, you'll probably want to experiment with different cognitive architectures just as frequently as you experiment with prompts. We're building LangChain and LangGraph to enable that. Most of our development efforts over the past year have gone into building low-level, highly controllable orchestration frameworks (LCEL and LangGraph). \nThis is a bit of a departure from early LangChain which focused on easy-to-use, off-the-shelf chains. These were great for getting started but tough to customize and experiment with. This was fine early on, as everyone was just trying to get started, but as the space matured, the design pretty quickly hit its limits.\n```",
 
      "为以下文章撰写一段摘要（约150字），要求概括文章的核心观点、主要内容和结论：\n```\n围绕 DeepSeek 成就的广泛叙述引发了关于 Scaling Law的论断。行业有理由质疑模型性能是否总是越大越好。但这种观点忽略了关键的转变：我们看到的并非是scaling的终结，而是其向新维度的演变。正如我们在 "2025 State of AI"（https://www.leoniscap.com/deep-dives/the-state-of-ai-in-2025）报告中提到的那样，行业正从一个由预训练扩展（Pre-training Scaling）主导的时代转向一个后训练优化（Post-training Optimization）提供巨大未开发潜力的时代。\nAnthropic 的 CEO Dario Amodei 也指出，后训练阶段扩展 / Post-training Scaling机会可能会带来比传统预训练扩展 / Pre-training Scaling更大的改进。从 2020 年到 2023 年，主要scaling的是在互联网文本上训练的预训练模型。随着这种"大力出奇迹"方法带来的性能提升走到尽头，像 OpenAI 和 Anthropic 这样的厂商转向使用强化学习和思维链 / CoT作为新的 Scaling 重点。这种方法在提高数学、编码和科学推理等可衡量任务的性能方面特别有效。在 2025 年，我们可以期待这种以后训练为中心的Scaling会带来更多成果，包括更复杂的强化学习方法、更好的合成数据生成以及更高效的推理时计算——这些领域我们才刚刚开始探索。\n然而，DeepSeek 的成就确实表明，支出与性能之间的关系并非线性。通过像 MoE 和 MLA 这样的架构创新，他们用更少的资源实现了更多的成果——这类似于半导体行业在 2000 年代末遇到 CPU 时钟速度墙时的演变。正如芯片制造商在原始时钟速度之外找到了新的改进向量——多核架构、先进封装和并行计算——AI 开发并非撞上了墙，而是在多样化其扩展向量。 \n主要大模型厂商的行动支持了这种更细致的观点。大型模厂商仍在继续进行大规模的基础设施投资。OpenAI 刚刚宣布了其 5000 亿美元的 Stargate 基础设施计划，xAI 推出了其 10 万卡 GPU 的 Colossus 超级数据中心，Meta 正在计划 2026 年的 2GW的数据中心，其他公司也在预测数十亿美元的 AI 开发投资。这些投资表明，Scaling 并未终结——它正在演变。未来可能会同时出现这两种情况：通过架构创新继续提高效率，以及进行大规模基础设施投资以推动可能性的边界。\n```",
 
      "总结以下这段文字的主要观点（不超过三句话），并分析作者可能存在的偏见或局限性：\n```\nAnthropic 的 CEO Dario Amodei 在报告中说，几周前，我曾撰文呼吁美国应加强对华芯片出口管制。此后不久，中国人工智能公司 DeepSeek 便成功地——至少在某些方面——以更低的成本，实现了与美国顶尖人工智能模型相近的性能水平。\n在此，我暂且不讨论 DeepSeek 是否对 Anthropic 等美国人工智能企业构成威胁（尽管我认为许多关于 DeepSeek 威胁美国人工智能领导地位的说法被严重夸大了）。\n我更关注的是，DeepSeek 的成果发布是否削弱了芯片出口管制政策的合理性。我的看法是否定的。事实上，我认为 DeepSeek 的进展反而令出口管制政策显得比一周前更具存在意义上的重要性。\n出口管制服务于一个至关重要的目标：确保民主国家在人工智能发展中保持领先地位。需要明确的是，出口管制并非逃避美中竞争的手段。最终，如果美国和其他民主国家的 AI 公司想要胜出，就必须开发出比中国更卓越的模型。但是，在力所能及的情况下，我们不应将技术优势拱手让给中国。\n```",
 
      "将以下新闻标题改写成更吸引眼球的形式，并分别针对科技爱好者、普通大众和投资者三个群体进行改写：\n'Anthropic 创始人发声：DeepSeek 事件前所未有，美国要继续加强出口管制'",
 
      "为以下产品描述生成3个不同的社交媒体帖子，分别适用于 Twitter、Facebook 和 LinkedIn，并考虑不同平台的特点和受众：\n```\n文润是一个基于先进 AI 技术的智能文本润色平台。通过多轮评审和优化的 AI 工作流，我们能帮助用户快速提升文本质量，解决传统文本创作中耗时费力、质量不稳定、翻译生硬等痛点问题。无论是技术文档、学术论文、营销文案还是日常写作，文润都能让您的文字更加清晰、准确、流畅、地道。支持文本直接输入、网页抓取、文件上传等多种方式，并提供多种文本风格和语气选项，满足不同场景的写作需求。\n```",
 
      "将以下技术文档中的一段翻译成更通俗易懂的语言，并添加适当的例子或图示（如果可能），以便非技术背景的读者也能理解：\n```\n当 OpenAI Assistants API 发布时，这标志着 AI 领域向智能体时代迈出了重要一步。它使 OpenAI 从一家提供大型语言模型（LLM）API 的公司，转变为一家提供智能体 API 的企业。\n我认为 OpenAI Assistants API 在多个方面都表现出色，它引入了许多创新且实用的基础设施，专门用于运行智能体应用。然而，这也在某种程度上限制了开发者在其基础上构建真正复杂（且极具价值！）智能体时所能采用的"认知架构"的灵活性。\n这凸显了"智能体基础设施"与"认知架构"之间的本质区别。正如亚马逊创始人杰夫·贝索斯那句经典箴言所言："专注于让你的啤酒更美味"。\n```",
 
      "用更简洁、更专业、更有说服力的语言改写以下句子，并指出原句存在的问题：\n'由于 DeepSeek-V3 的性能不如那些美国前沿模型——假设在规模曲线上落后约 2 倍，我认为这对于 DeepSeek-V3 来说已经相当慷慨了——这意味着，如果 DeepSeek-V3 的训练成本比美国一年前开发的现有模型低约 8 倍，那将是完全正常、完全符合"趋势"的。我不会给出具体数字，但从前一点可以清楚地看出，即使你完全相信 DeepSeek 宣称的训练成本，他们的表现充其量也只是符合趋势，甚至可能还达不到。例如，这远不如最初的 GPT-4 到 Claude 3.5 Sonnet 的推理价格差异（10 倍），而 3.5 Sonnet 是一款比 GPT-4 更出色的模型。'",
 
      "什么是 AI Agent？它有哪些组成部分？其核心原理是什么？这项技术重要吗？如何开始学习？请结合实际应用场景（如智能客服、自动驾驶、游戏NPC等）进行说明，并推荐一些学习资源。",
      
      "您认为未来五年内，AI Agent 在哪些应用领域最具潜力取得突破性进展？为什么？请分析这些领域的技术现状、市场需求、潜在挑战和发展趋势，并给出您的预测。"
    ]
 
    # 定义各服务商的配置, 使用 api_key_env 字段
    providers = [
        {
            "name": "火山引擎",
            "api_key_env": 'Key_Volces',
            "base_url": "https://ark.cn-beijing.volces.com/api/v3",
            "model": 'ep-20250217091332-8g29h',
            "modelType": "R1"
        },
       {
            "name": "火山引擎",
            "api_key_env": 'Key_Volces',
            "base_url": "https://ark.cn-beijing.volces.com/api/v3",
            "model": "ep-20250217102601-knmvv",
            "modelType": "V3"
        },
        {
            "name": "OpenRouter",
            "api_key_env": 'Key_OpenRouter',
            "base_url": "https://openrouter.ai/api/v1",
            "model": 'deepseek/deepseek-r1',
            "modelType": "R1"
        },
       {
            "name": "OpenRouter",
            "api_key_env": 'Key_OpenRouter',
            "base_url": "https://openrouter.ai/api/v1",
            "model": "deepseek/deepseek-chat",
            "modelType": "V3"
        },
        {
            "name": "DeepSeek官网",
            "api_key_env": 'Key_DeepSeek',
            "base_url": "https://api.deepseek.com",
            "model": 'deepseek-reasoner',
            "modelType": "R1"
        },
       {
            "name": "DeepSeek官网",
            "api_key_env": 'Key_DeepSeek',
            "base_url": "https://api.deepseek.com",
            "model": "deepseek-chat",
            "modelType": "V3"
        },
    ]
    print(f"本次测试开始于中国时间：{datetime.datetime.now(pytz.timezone('Asia/Shanghai')).strftime('%Y-%m-%d %H:%M:%S')}")
 
    # 循环对每个服务商进行测试
    for provider in providers:
        test_provider(provider, prompts)