🏆 前沿模型横评（2024-2026）¶

# GRPO 核心流程伪代码
def grpo_step(policy, prompts, reward_fn, G=16, epsilon=0.2, beta=0.01):
    """
    policy: 当前策略模型 π_θ
    prompts: 一批问题
    reward_fn: 规则化奖励函数（如：答案正确 → 1，错误 → 0）
    G: 每个问题采样的候选回答数
    """
    all_loss = []

    for q in prompts:
        # Step 1: 组采样 — 对同一问题生成 G 个回答
        responses = [policy.generate(q) for _ in range(G)]

        # Step 2: 计算奖励（规则化，无需 Reward Model）
        rewards = [reward_fn(q, resp) for resp in responses]

        # Step 3: 组内标准化
        mean_r = sum(rewards) / G
        std_r = (sum((r - mean_r)**2 for r in rewards) / G) ** 0.5
        advantages = [(r - mean_r) / (std_r + 1e-8) for r in rewards]

        # Step 4: PPO-clip 式策略优化
        for resp, adv in zip(responses, advantages):  # zip按位置配对多个可迭代对象
            ratio = policy.log_prob(resp, q) - old_policy.log_prob(resp, q)
            ratio = ratio.exp()
            clipped = torch.clamp(ratio, 1 - epsilon, 1 + epsilon)
            loss = -torch.min(ratio * adv, clipped * adv)
            all_loss.append(loss)

    # 加上 KL 散度惩罚
    total_loss = torch.stack(all_loss).mean() + beta * kl_divergence(policy, ref_policy)
    total_loss.backward()
    optimizer.step()

"""
多模型对比评测工具
用于在相同测试集上自动评估多个模型的质量、延迟和成本
"""
import time
import json
import os
from dataclasses import dataclass, field
from openai import OpenAI
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

@dataclass  # @dataclass自动生成__init__等方法
class EvalResult:
    """单次评测结果"""
    model: str
    question: str
    answer: str
    latency_ms: float
    input_tokens: int
    output_tokens: int
    cost_usd: float

@dataclass
class ModelConfig:
    """模型配置"""
    name: str
    provider: str       # "openai", "anthropic", "deepseek", "ollama"
    model_id: str
    input_price: float  # per 1M tokens
    output_price: float # per 1M tokens
    temperature: float = 0.3

# ---------- 待评测模型列表 ----------
MODELS = [
    ModelConfig("GPT-4o-mini", "openai", "gpt-4o-mini", 0.15, 0.60),
    ModelConfig("GPT-4o", "openai", "gpt-4o", 2.50, 10.00),
    ModelConfig("DeepSeek-V3", "deepseek", "deepseek-chat", 0.27, 1.10),
    ModelConfig("Claude Sonnet", "anthropic", "claude-sonnet-4-20250514", 3.00, 15.00),
    # 本地模型（Ollama）— 成本为0
    # ModelConfig("Qwen3-8B", "ollama", "qwen3:8b", 0.0, 0.0),
]

# ---------- 评测数据集 ----------
TEST_CASES = [
    {
        "question": "用Python实现二分查找，要求支持查找第一个大于等于target的位置",
        "category": "代码生成",
        "reference": "def bisect_left",  # 简单的参考答案关键词
    },
    {
        "question": "解释TCP三次握手的过程，以及为什么不能是两次握手",
        "category": "知识问答",
        "reference": "SYN",
    },
    {
        "question": "一个水池有两个进水管和一个出水管。进水管A每小时注水3吨，进水管B每小时注水2吨，出水管每小时排水1吨。水池容量为40吨，从空池开始，多长时间能注满？",
        "category": "数学推理",
        "reference": "10",
    },
    {
        "question": "将以下英文翻译为中文，保持专业术语准确：'The transformer architecture uses self-attention mechanisms to process sequential data in parallel, significantly reducing training time compared to recurrent neural networks.'",
        "category": "翻译",
        "reference": "自注意力",
    },
]

def call_model(config: ModelConfig, question: str, system: str = "You are a helpful assistant.") -> EvalResult:
    """调用单个模型并记录结果"""
    start = time.time()

    if config.provider == "openai":
        client = OpenAI()
        resp = client.chat.completions.create(
            model=config.model_id,
            messages=[{"role": "system", "content": system}, {"role": "user", "content": question}],
            temperature=config.temperature,
            max_tokens=1024,
        )
        answer = resp.choices[0].message.content
        in_tok = resp.usage.prompt_tokens
        out_tok = resp.usage.completion_tokens

    elif config.provider == "deepseek":
        client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"], base_url="https://api.deepseek.com")
        resp = client.chat.completions.create(
            model=config.model_id,
            messages=[{"role": "system", "content": system}, {"role": "user", "content": question}],
            temperature=config.temperature,
            max_tokens=1024,
        )
        answer = resp.choices[0].message.content
        in_tok = resp.usage.prompt_tokens
        out_tok = resp.usage.completion_tokens

    elif config.provider == "anthropic":
        client = Anthropic()
        resp = client.messages.create(
            model=config.model_id,
            max_tokens=1024,
            system=system,
            messages=[{"role": "user", "content": question}],
            temperature=config.temperature,
        )
        answer = resp.content[0].text
        in_tok = resp.usage.input_tokens
        out_tok = resp.usage.output_tokens

    elif config.provider == "ollama":
        client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
        resp = client.chat.completions.create(
            model=config.model_id,
            messages=[{"role": "system", "content": system}, {"role": "user", "content": question}],
            temperature=config.temperature,
            max_tokens=1024,
        )
        answer = resp.choices[0].message.content
        in_tok = resp.usage.prompt_tokens if resp.usage else 0  # 三元空值守卫：部分模型可能不返回usage统计，此时默认为0
        out_tok = resp.usage.completion_tokens if resp.usage else 0

    else:
        raise ValueError(f"Unknown provider: {config.provider}")

    latency = (time.time() - start) * 1000
    cost = (in_tok * config.input_price + out_tok * config.output_price) / 1_000_000

    return EvalResult(
        model=config.name,
        question=question[:50],
        answer=answer,
        latency_ms=latency,
        input_tokens=in_tok,
        output_tokens=out_tok,
        cost_usd=cost,
    )

def run_evaluation():
    """运行完整评测"""
    results = []

    for model_cfg in MODELS:
        print(f"\n{'='*60}")
        print(f"评测模型: {model_cfg.name}")
        print(f"{'='*60}")

        model_results = []
        total_cost = 0
        total_latency = 0

        for i, test in enumerate(TEST_CASES):  # enumerate同时获取索引和元素
            try:  # try/except捕获异常，防止程序崩溃
                result = call_model(model_cfg, test["question"])
                model_results.append(result)
                total_cost += result.cost_usd
                total_latency += result.latency_ms

                # 简单的参考答案匹配检查
                hit = "✓" if test["reference"].lower() in result.answer.lower() else "✗"
                print(f"  [{hit}] {test['category']}: {result.latency_ms:.0f}ms, "
                      f"${result.cost_usd:.6f}, {result.output_tokens} tokens")

            except Exception as e:
                print(f"  [!] {test['category']}: 错误 - {e}")

        avg_latency = total_latency / len(TEST_CASES) if TEST_CASES else 0
        print(f"\n  汇总: 平均延迟 {avg_latency:.0f}ms, 总成本 ${total_cost:.6f}")
        results.extend(model_results)

    # 保存结果
    with open("eval_results.json", "w", encoding="utf-8") as f:  # with自动管理文件关闭
        json.dump([{
            "model": r.model,
            "question": r.question,
            "latency_ms": r.latency_ms,
            "cost_usd": r.cost_usd,
            "output_tokens": r.output_tokens,
        } for r in results], f, ensure_ascii=False, indent=2)

    print(f"\n评测结果已保存到 eval_results.json")

# 运行
# run_evaluation()