Lab 007：Claude / GPT / DeepSeek / 本地模型编码任务横评

# model_benchmark_runner.py
"""
多模型编码任务横评 Runner。
统一任务集 × 多模型，自动执行和评分。
"""
import json
import time
from dataclasses import dataclass, asdict, field

@dataclass
class TaskResult:
    model: str
    task_id: str
    task_type: str          # bugfix / review / test / docs
    difficulty: str         # easy / medium / hard
    
    # 质量指标
    score: float            # 0-1 综合得分
    tests_passed: int = 0
    tests_total: int = 0
    issues_found: int = 0   # Code Review 发现的问题数
    mutation_score: float = 0.0  # Test Generation 的变异测试得分
    
    # 成本指标
    input_tokens: int = 0
    output_tokens: int = 0
    cost_usd: float = 0.0
    time_seconds: float = 0.0
    
    # 执行信息
    success: bool = False
    error: str = ""

PRICING = {
    "claude-opus": {"input": 15.0, "output": 75.0},
    "claude-sonnet": {"input": 3.0, "output": 15.0},
    "gpt-4o": {"input": 5.0, "output": 15.0},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "deepseek-v3": {"input": 0.27, "output": 1.10},
    "llama3-70b": {"input": 0.0, "output": 0.0},  # 本地部署
}

def run_task(model: str, task: dict, runner) -> TaskResult:
    """执行单个任务"""
    result = TaskResult(
        model=model,
        task_id=task["id"],
        task_type=task["type"],
        difficulty=task["difficulty"],
    )
    
    start = time.time()
    
    try:
        # 1. 构建 Prompt
        prompt = build_prompt(task, model)
        
        # 2. 调用模型
        response = runner.call_model(model, prompt)
        
        # 3. 解析结果
        result.input_tokens = response.usage.input_tokens
        result.output_tokens = response.usage.output_tokens
        
        pricing = PRICING[model]
        result.cost_usd = (
            result.input_tokens * pricing["input"] / 1_000_000 +
            result.output_tokens * pricing["output"] / 1_000_000
        )
        
        # 4. 评分
        if task["type"] == "bugfix":
            result = evaluate_bugfix(result, response, task)
        elif task["type"] == "review":
            result = evaluate_review(result, response, task)
        elif task["type"] == "test":
            result = evaluate_test(result, response, task)
        elif task["type"] == "docs":
            result = evaluate_docs(result, response, task)
        
        result.success = result.score >= 0.7
        
    except Exception as e:
        result.error = str(e)
        result.success = False
    
    result.time_seconds = time.time() - start
    return result

def evaluate_bugfix(result: TaskResult, response, task: dict) -> TaskResult:
    """Bugfix 评分：应用补丁后运行回归测试"""
    # 应用补丁
    apply_patch(response.content)
    
    # 运行测试
    test_output = run_tests(task["test_command"])
    result.tests_passed = test_output.passed
    result.tests_total = test_output.total
    
    # 评分 = 通过率
    result.score = result.tests_passed / result.tests_total if result.tests_total > 0 else 0
    return result

def evaluate_review(result: TaskResult, response, task: dict) -> TaskResult:
    """Code Review 评分：对比发现的问题和已知问题"""
    found_issues = parse_review_comments(response.content)
    known_issues = task["known_issues"]
    
    # 计算召回率和准确率
    true_positives = sum(1 for f in found_issues if matches_any(f, known_issues))
    result.issues_found = len(found_issues)
    
    recall = true_positives / len(known_issues) if known_issues else 0
    precision = true_positives / len(found_issues) if found_issues else 0
    
    # F1 分数
    result.score = 2 * recall * precision / (recall + precision) if (recall + precision) > 0 else 0
    return result

def run_full_benchmark(models: list, tasks: list, runner):
    """运行完整评测"""
    all_results = []
    
    for model in models:
        print(f"\n{'='*60}")
        print(f"评测模型: {model}")
        print(f"{'='*60}")
        
        for task in tasks:
            # 每个任务运行 3 次
            task_results = []
            for run in range(3):
                result = run_task(model, task, runner)
                task_results.append(result)
                print(f"  [{task['type']}] {task['id']} run {run+1}: score={result.score:.2f} cost=${result.cost_usd:.4f}")
            
            # 取均值
            avg_result = average_results(task_results)
            all_results.append(avg_result)
    
    # 保存结果
    with open("benchmark-results.json", "w") as f:
        json.dump([asdict(r) for r in all_results], f, indent=2, ensure_ascii=False)
    
    return all_results

Bugfix 成功率（按难度）：
                简单    中等    困难
Claude Opus     100%    88%    63%
Claude Sonnet    94%    75%    38%
GPT-4o           88%    75%    38%
DeepSeek-V3      81%    69%    25%
GPT-4o-mini      69%    50%    13%
Llama 3 70B      56%    44%     0%

结论：
- 简单任务：所有云端模型差异不大（>80%）
- 困难任务：Opus 明显领先（63% vs 第二的 38%）
- 本地模型在困难任务上基本不可用

# 推荐的模型预算分配
monthly_budget: $2000
allocation:
  claude_opus:
    percentage: 40%      # $800
    use_for: ["complex_bugfix", "security_review", "architecture"]
  
  claude_sonnet:
    percentage: 35%      # $700
    use_for: ["standard_bugfix", "code_review", "test_generation"]
  
  deepseek_v3:
    percentage: 15%      # $300
    use_for: ["batch_formatting", "simple_docs", "lint_fix"]
  
  reserve:
    percentage: 10%      # $200
    use_for: "应急和新模型试验"