GPT-5.6 Sol/Terra/Luna三模型亮相：从Codex代码泄露到7月7日精准卡点，OpenAI的"趁火打劫"商业围猎战略深度解析

// ============================================================
// GPT-5.6 Sol Ultra 子Agent调度框架实现
// 模拟GPT-5.6 Ultra模式下多Agent并行推理的核心机制
// ============================================================
package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

// Task 定义需要处理的任务
type Task struct {
    ID          string
    Description string
    Type        TaskType
    Complexity  int // 1-10
    Dependencies []string
}

type TaskType int

const (
    TaskCoding     TaskType = iota
    TaskAnalysis
    TaskResearch
    TaskSecurity
    TaskPlanning
)

// SubAgent 子Agent定义
type SubAgent struct {
    ID        string
    Expertise TaskType
    Capacity  int // 并发处理能力
    Status    string
    results   chan Result
    ctx       context.Context
    cancel    context.CancelFunc
}

// Result 处理结果
type Result struct {
    TaskID    string
    AgentID   string
    Output    string
    TokensUsed int
    Duration  time.Duration
    Error     error
}

// UltraScheduler Ultra模式调度器
type UltraScheduler struct {
    agents    []*SubAgent
    taskQueue chan Task
    results   map[string]Result
    mu        sync.RWMutex
    wg        sync.WaitGroup
}

func NewUltraScheduler() *UltraScheduler {
    s := &UltraScheduler{
        taskQueue: make(chan Task, 100),
        results:   make(map[string]Result),
    }
    
    // 初始化专业子Agent池
    // 对应GPT-5.6 Ultra模式的多Agent协同架构
    specializations := []struct {
        id       string
        expertise TaskType
    }{
        {"code-agent-1", TaskCoding},
        {"code-agent-2", TaskCoding},
        {"analysis-agent", TaskAnalysis},
        {"research-agent", TaskResearch},
        {"security-agent", TaskSecurity},
        {"planning-agent", TaskPlanning},
    }
    
    for _, spec := range specializations {
        ctx, cancel := context.WithCancel(context.Background())
        agent := &SubAgent{
            ID:        spec.id,
            Expertise: spec.expertise,
            Capacity:  3,
            Status:    "idle",
            results:   make(chan Result, 10),
            ctx:       ctx,
            cancel:    cancel,
        }
        s.agents = append(s.agents, agent)
    }
    
    return s
}

// DispatchTask 根据任务类型调度到合适的子Agent
func (s *UltraScheduler) DispatchTask(task Task) {
    s.wg.Add(1)
    go func() {
        defer s.wg.Done()
        
        // 找到最适合处理该任务的子Agent
        agent := s.selectOptimalAgent(task)
        if agent == nil {
            fmt.Printf("[Ultra] No available agent for task %s\n", task.ID)
            return
        }
        
        fmt.Printf("[Ultra] Dispatching task %s (%s) to agent %s (%v)\n",
            task.ID, task.Description, agent.ID, agent.Expertise)
        
        // 模拟子Agent处理
        start := time.Now()
        time.Sleep(time.Duration(100+task.Complexity*50) * time.Millisecond)
        
        result := Result{
            TaskID:     task.ID,
            AgentID:    agent.ID,
            Output:     fmt.Sprintf("Task %s completed by %s", task.ID, agent.ID),
            TokensUsed: 1000 + task.Complexity*500,
            Duration:   time.Since(start),
        }
        
        s.mu.Lock()
        s.results[task.ID] = result
        s.mu.Unlock()
        
        fmt.Printf("[Ultra] Task %s completed in %v (tokens: %d)\n",
            task.ID, result.Duration, result.TokensUsed)
    }()
}

// selectOptimalAgent 根据任务特征选择最优子Agent
// 对应GPT-5.6 Ultra模式的智能路由机制
func (s *UltraScheduler) selectOptimalAgent(task Task) *SubAgent {
    var bestAgent *SubAgent
    bestScore := -1
    
    for _, agent := range s.agents {
        score := 0
        
        // 专业匹配度评分
        if agent.Expertise == task.Type {
            score += 50
        }
        
        // 关联领域加分
        switch task.Type {
        case TaskCoding:
            if agent.Expertise == TaskSecurity {
                score += 20 // 安全编码需要安全Agent辅助
            }
        case TaskAnalysis:
            if agent.Expertise == TaskResearch {
                score += 25 // 分析任务常需要研究Agent提供背景
            }
        }
        
        // 负载评分（轻负载优先）
        score -= len(agent.results) * 5
        
        if score > bestScore {
            bestScore = score
            bestAgent = agent
        }
    }
    
    return bestAgent
}

// AggregateResults 汇总所有子Agent结果
func (s *UltraScheduler) AggregateResults() []Result {
    s.mu.RLock()
    defer s.mu.RUnlock()
    
    results := make([]Result, 0, len(s.results))
    for _, result := range s.results {
        results = append(results, result)
    }
    return results
}

// SimulateUltraMode 模拟Ultra模式处理复杂任务
func SimulateUltraMode() {
    scheduler := NewUltraScheduler()
    
    // 构造一个复杂任务集（模拟GPT-5.6 Ultra模式的实际工作负载）
    tasks := []Task{
        {ID: "T1", Description: "分析代码库架构", Type: TaskAnalysis, Complexity: 7},
        {ID: "T2", Description: "重构模块A的代码", Type: TaskCoding, Complexity: 8},
        {ID: "T3", Description: "安全审计依赖库", Type: TaskSecurity, Complexity: 6},
        {ID: "T4", Description: "生成单元测试", Type: TaskCoding, Complexity: 5},
        {ID: "T5", Description: "研究新API兼容性", Type: TaskResearch, Complexity: 4},
        {ID: "T6", Description: "规划发布策略", Type: TaskPlanning, Complexity: 3},
    }
    
    fmt.Println("=" + strings.Repeat("=", 59))
    fmt.Println("  GPT-5.6 Sol Ultra 模式子Agent调度模拟")
    fmt.Println("=" + strings.Repeat("=", 59))
    fmt.Printf("\n任务总数: %d\n", len(tasks))
    fmt.Printf("子Agent池: %d个专业Agent\n\n", len(scheduler.agents))
    
    // 并行调度所有任务
    for _, task := range tasks {
        scheduler.DispatchTask(task)
    }
    
    scheduler.wg.Wait()
    
    results := scheduler.AggregateResults()
    fmt.Printf("\n[Ultra模式完成] %d个任务全部处理完毕\n", len(results))
    
    // 计算性能指标
    var totalTokens, totalDuration int64
    for _, r := range results {
        totalTokens += int64(r.TokensUsed)
        totalDuration += r.Duration.Milliseconds()
    }
    
    fmt.Printf("总Token消耗: %d\n", totalTokens)
    fmt.Printf("总耗时: %dms\n", totalDuration)
    fmt.Printf("平均Token/任务: %d\n", totalTokens/int64(len(results)))
    fmt.Printf("并发度: %d个并行Agent\n\n", len(scheduler.agents))
    
    // 对比：单路径推理（非Ultra模式）
    singlePathTokens := 0
    for _, task := range tasks {
        // 单路径模式下，每个任务消耗更多token（无子Agent优化）
        singlePathTokens += 2000 + task.Complexity*800
    }
    
    fmt.Println("性能对比（Ultra vs 单路径推理）:")
    fmt.Printf("  Ultra模式 Token: %d\n", totalTokens)
    fmt.Printf("  单路径 Token:    %d\n", singlePathTokens)
    fmt.Printf("  节省比例:        %.1f%%\n",
        (1-float64(totalTokens)/float64(singlePathTokens))*100)
}

func main() {
    SimulateUltraMode()
}

#!/usr/bin/env python3
"""
GPT-5.6 Speed Dial: 动态推理算力分配机制实现
模拟速度拨盘如何在同一个模型上实现不同推理强度的调节
"""

import time
import math
from enum import Enum
from typing import Any, Dict, Optional, Tuple


class SpeedDialPosition(Enum):
    """速度拨盘档位"""
    TURBO = 0.0      # 极速模式 - 最短推理路径
    FAST = 0.25      # 快速模式 - 精简推理
    BALANCED = 0.5   # 均衡模式 - 默认推理
    PRECISE = 0.75   # 精确模式 - 增强推理
    ULTRA = 1.0      # 极限模式 - 最大推理深度


class DynamicInferenceEngine:
    """
    动态推理引擎。
    根据速度拨盘位置动态分配推理算力。
    核心机制：通过调整推理深度、采样宽度和自洽校验次数来控制推理成本。
    """
    
    def __init__(self, model_name: str):
        self.model_name = model_name
        # 基础推理参数
        self.base_params = {
            "max_tokens": 4096,
            "temperature": 0.7,
            "top_p": 0.9,
        }
        
        # 各档位对应的推理参数缩放
        self.speed_dial_configs = {
            SpeedDialPosition.TURBO: {
                "description": "极速 - 适合简单问答、头脑风暴",
                "reasoning_depth": 0.2,      # 20%推理深度
                "sampling_width": 1,          # 单次采样
                "self_consistency": 1,        # 不自洽校验
                "think_budget_ratio": 0.1,    # 10%思考预算
                "latency_target_ms": 500,     # 500ms延迟目标
            },
            SpeedDialPosition.FAST: {
                "description": "快速 - 适合日常开发、文档处理",
                "reasoning_depth": 0.4,
                "sampling_width": 2,
                "self_consistency": 1,
                "think_budget_ratio": 0.3,
                "latency_target_ms": 1500,
            },
            SpeedDialPosition.BALANCED: {
                "description": "均衡 - 默认模式，通用场景",
                "reasoning_depth": 0.6,
                "sampling_width": 3,
                "self_consistency": 2,
                "think_budget_ratio": 0.5,
                "latency_target_ms": 3000,
            },
            SpeedDialPosition.PRECISE: {
                "description": "精确 - 适合代码审查、数据分析",
                "reasoning_depth": 0.8,
                "sampling_width": 4,
                "self_consistency": 3,
                "think_budget_ratio": 0.7,
                "latency_target_ms": 5000,
            },
            SpeedDialPosition.ULTRA: {
                "description": "极限 - 适合复杂推理、科研分析",
                "reasoning_depth": 1.0,
                "sampling_width": 6,
                "self_consistency": 5,
                "think_budget_ratio": 1.0,
                "latency_target_ms": 10000,
            },
        }
    
    def estimate_cost(self, 
                     position: SpeedDialPosition, 
                     input_tokens: int,
                     output_tokens: int) -> Dict[str, Any]:
        """
        估算指定速度拨盘位置下的推理成本。
        
        成本模型：
        - 基础成本 = 输入tokens × 输入价格 + 输出tokens × 输出价格
        - 推理成本 = 基础成本 × (1 + reasoning_depth × think_budget_ratio)
        - 自洽校验成本 = 推理成本 × self_consistency
        """
        config = self.speed_dial_configs[position]
        
        # 每百万token价格（按模型不同）
        price_per_million = {
            "Sol": {"input": 5, "output": 30},
            "Terra": {"input": 2.5, "output": 15},
            "Luna": {"input": 1, "output": 6},
        }
        
        prices = price_per_million.get(self.model_name, price_per_million["Terra"])
        
        # 基础token成本
        base_cost = (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000
        
        # 推理深度加权
        depth_factor = 1 + config["reasoning_depth"] * config["think_budget_ratio"]
        reasoning_cost = base_cost * depth_factor
        
        # 自洽校验加权
        total_cost = reasoning_cost * config["self_consistency"]
        
        # 估计延迟
        est_latency = config["latency_target_ms"] * (1 + output_tokens / 1000)
        
        # 估计Token效率
        effective_tokens = output_tokens * config["self_consistency"]
        
        return {
            "position": position.name,
            "description": config["description"],
            "base_cost_usd": round(base_cost, 6),
            "reasoning_cost_usd": round(reasoning_cost, 6),
            "total_cost_usd": round(total_cost, 6),
            "estimated_latency_ms": round(est_latency, 0),
            "effective_output_tokens": output_tokens,
            "total_tokens_consumed": effective_tokens,
            "reasoning_depth": config["reasoning_depth"],
            "self_consistency_checks": config["self_consistency"],
        }
    
    def optimize_speed_dial(self,
                           task_type: str,
                           input_tokens: int,
                           output_tokens: int,
                           max_latency_ms: int = 3000,
                           max_cost_usd: float = 0.1) -> SpeedDialPosition:
        """
        根据任务特征自动优化速度拨盘位置。
        对应GPT-5.6的智能推荐功能。
        """
        # 任务类型到推荐拨盘位置的映射
        task_recommendations = {
            "chat": SpeedDialPosition.FAST,
            "code_completion": SpeedDialPosition.FAST,
            "code_review": SpeedDialPosition.PRECISE,
            "debugging": SpeedDialPosition.PRECISE,
            "architecture": SpeedDialPosition.ULTRA,
            "analysis": SpeedDialPosition.BALANCED,
            "creative_writing": SpeedDialPosition.BALANCED,
            "research": SpeedDialPosition.ULTRA,
            "simple_qa": SpeedDialPosition.TURBO,
            "translation": SpeedDialPosition.FAST,
        }
        
        # 先根据任务类型推荐
        recommended = task_recommendations.get(task_type, SpeedDialPosition.BALANCED)
        
        # 然后根据约束条件调整
        for position in [SpeedDialPosition.FAST, SpeedDialPosition.BALANCED,
                        SpeedDialPosition.PRECISE, SpeedDialPosition.ULTRA]:
            cost_estimate = self.estimate_cost(position, input_tokens, output_tokens)
            
            # 检查延迟约束
            if cost_estimate["estimated_latency_ms"] > max_latency_ms:
                continue
            
            # 检查成本约束
            if cost_estimate["total_cost_usd"] > max_cost_usd:
                continue
            
            return position
        
        # 如果所有档位都超约束，返回最快的
        return SpeedDialPosition.TURBO
    
    def simulate_inference(self, 
                          prompt: str,
                          position: SpeedDialPosition) -> Dict[str, Any]:
        """
        模拟速度拨盘在不同位置下的推理行为。
        """
        config = self.speed_dial_configs[position]
        input_tokens = len(prompt.split())
        
        # 模拟推理时间
        think_time = 0.1 * config["reasoning_depth"] * config["think_budget_ratio"]
        if config["self_consistency"] > 1:
            think_time *= config["self_consistency"]
        
        time.sleep(min(think_time, 0.5))  # 模拟
        
        # 模拟不同推理深度下的输出质量
        quality_metrics = {
            "accuracy": 0.5 + config["reasoning_depth"] * 0.4,
            "completeness": 0.3 + config["reasoning_depth"] * 0.6,
            "creativity": 0.7 - config["reasoning_depth"] * 0.3,
            "consistency": 0.4 + config["self_consistency"] * 0.1,
        }
        
        return {
            "model": self.model_name,
            "speed_dial": position.name,
            "input_tokens": input_tokens,
            "think_time_seconds": round(think_time, 3),
            "quality_metrics": quality_metrics,
            "config": config,
        }


# ============================================================
# 速度拨盘性能对比测试
# ============================================================

def benchmark_speed_dial():
    """对比不同速度拨盘位置下的性能与成本"""
    
    print("=" * 70)
    print("GPT-5.6 速度拨盘性能对比测试")
    print("=" * 70)
    
    test_cases = [
        ("简单问答", "What is the capital of France?", "simple_qa", 50, 50),
        ("代码审查", "Review this Go function for concurrency issues", "code_review", 500, 300),
        ("架构设计", "Design a microservice architecture for an e-commerce platform", "architecture", 1000, 2000),
        ("科研分析", "Analyze the implications of the new transformer architecture", "research", 2000, 1500),
    ]
    
    for task_name, prompt, task_type, in_tokens, out_tokens in test_cases:
        print(f"\n{'='*70}")
        print(f"任务: {task_name} ({task_type})")
        print(f"输入: ~{in_tokens} tokens, 输出: ~{out_tokens} tokens")
        print(f"{'='*70}")
        
        for model in ["Sol", "Terra", "Luna"]:
            engine = DynamicInferenceEngine(model)
            
            # 自动优化
            optimal = engine.optimize_speed_dial(
                task_type, in_tokens, out_tokens,
                max_latency_ms=5000,
                max_cost_usd=0.5
            )
            
            print(f"\n  [{model}] 推荐拨盘: {optimal.name}")
            cost = engine.estimate_cost(optimal, in_tokens, out_tokens)
            print(f"    成本: ${cost['total_cost_usd']:.6f}")
            print(f"    延迟: {cost['estimated_latency_ms']:.0f}ms")
            print(f"    推理深度: {cost['reasoning_depth']:.0%}")
            print(f"    自洽校验: {cost['self_consistency_checks']}次")
            
            # 全档位对比
            print(f"    全档位成本对比:")
            for pos in SpeedDialPosition:
                c = engine.estimate_cost(pos, in_tokens, out_tokens)
                marker = " ← 推荐" if pos == optimal else ""
                print(f"      {pos.name:10s} | ${c['total_cost_usd']:.6f} | {c['estimated_latency_ms']:.0f}ms{marker}")


if __name__ == "__main__":
    benchmark_speed_dial()

// ============================================================
// GPT-5.6 定价策略分析器
// 对比不同模型在不同使用场景下的总拥有成本
// ============================================================
package main

import (
    "encoding/json"
    "fmt"
    "strings"
)

type ModelPricing struct {
    Name        string
    InputPrice  float64 // $ per 1M tokens
    OutputPrice float64 // $ per 1M tokens
}

type UsageScenario struct {
    Name              string
    DailyInputTokens  int64
    DailyOutputTokens int64
    MonthlyActiveDays int
}

type CostAnalysis struct {
    ModelName        string
    ScenarioName     string
    DailyCost        float64
    MonthlyCost      float64
    AnnualCost       float64
    CostPer1000Tasks float64
    RelativeToCheapest float64
}

func analyzeCost(model ModelPricing, scenario UsageScenario) CostAnalysis {
    dailyInputCost := float64(scenario.DailyInputTokens) * model.InputPrice / 1_000_000
    dailyOutputCost := float64(scenario.DailyOutputTokens) * model.OutputPrice / 1_000_000
    dailyCost := dailyInputCost + dailyOutputCost
    monthlyCost := dailyCost * float64(scenario.MonthlyActiveDays)
    annualCost := monthlyCost * 12
    
    // 每1000次任务的成本（假设每次任务平均1000输入+500输出tokens）
    costPerTask := (1000 * model.InputPrice + 500 * model.OutputPrice) / 1_000_000
    costPer1000Tasks := costPerTask * 1000
    
    return CostAnalysis{
        ModelName:        model.Name,
        ScenarioName:     scenario.Name,
        DailyCost:        dailyCost,
        MonthlyCost:      monthlyCost,
        AnnualCost:       annualCost,
        CostPer1000Tasks: costPer1000Tasks,
    }
}

func main() {
    fmt.Println("=" + strings.Repeat("=", 69))
    fmt.Println("  GPT-5.6 定价策略分析 - 全场景成本对比")
    fmt.Println("=" + strings.Repeat("=", 69))
    
    models := []ModelPricing{
        {"GPT-5.6 Sol", 5, 30},
        {"GPT-5.6 Terra", 2.5, 15},
        {"GPT-5.6 Luna", 1, 6},
        {"Claude Fable 5", 10, 50},
        {"Claude Sonnet 5", 2, 10},
        {"DeepSeek V4", 0.435, 0.87},
    }
    
    scenarios := []UsageScenario{
        {"个人开发者", 50000, 20000, 30},
        {"小型团队(10人)", 500000, 200000, 26},
        {"中型企业(100人)", 5000000, 2000000, 22},
        {"大型企业(1000人)", 50000000, 20000000, 22},
        {"API批量处理", 100000000, 50000000, 30},
    }
    
    for _, scenario := range scenarios {
        fmt.Printf("\n📊 %s (日均输入: %dK, 输出: %dK, 月活跃: %d天)\n",
            scenario.Name,
            scenario.DailyInputTokens/1000,
            scenario.DailyOutputTokens/1000,
            scenario.MonthlyActiveDays)
        fmt.Println(strings.Repeat("-", 70))
        fmt.Printf("%-20s %12s %12s %12s %12s\n", "模型", "日成本", "月成本", "年成本", "每千次")
        fmt.Println(strings.Repeat("-", 70))
        
        var analyses []CostAnalysis
        for _, model := range models {
            analysis := analyzeCost(model, scenario)
            analyses = append(analyses, analysis)
            fmt.Printf("%-20s $%10.2f $%10.2f $%10.2f $%10.4f\n",
                model.Name,
                analysis.DailyCost,
                analysis.MonthlyCost,
                analysis.AnnualCost,
                analysis.CostPer1000Tasks)
        }
        
        // 找出最便宜的
        cheapest := analyses[0]
        for _, a := range analyses[1:] {
            if a.MonthlyCost < cheapest.MonthlyCost {
                cheapest = a
            }
        }
        
        fmt.Printf("\n最经济选择: %s (月成本 $%.2f)\n", cheapest.ModelName, cheapest.MonthlyCost)
        
        // 展示GPT-5.6相对于Claude Fable 5的节省
        var fable5, sol, terra, luna CostAnalysis
        for _, a := range analyses {
            switch a.ModelName {
            case "Claude Fable 5":
                fable5 = a
            case "GPT-5.6 Sol":
                sol = a
            case "GPT-5.6 Terra":
                terra = a
            case "GPT-5.6 Luna":
                luna = a
            }
        }
        
        fmt.Printf("GPT-5.6 Sol vs Fable 5: 节省 %.1f%%\n",
            (1-sol.MonthlyCost/fable5.MonthlyCost)*100)
        fmt.Printf("GPT-5.6 Terra vs Fable 5: 节省 %.1f%%\n",
            (1-terra.MonthlyCost/fable5.MonthlyCost)*100)
        fmt.Printf("GPT-5.6 Luna vs Fable 5: 节省 %.1f%%\n",
            (1-luna.MonthlyCost/fable5.MonthlyCost)*100)
    }
    
    // 输出JSON格式供程序化使用
    fmt.Println("\n\nJSON输出:")
    jsonData, _ := json.MarshalIndent(models, "  ", "  ")
    fmt.Println(string(jsonData))
}

// ============================================================
// GPT-5.6 多模型智能路由中间件
// 支持Sol/Terra/Luna三档模型自动路由
// 支持加权随机、成本优先、质量优先、延迟优先四种策略
// ============================================================
package main

import (
    "container/heap"
    "encoding/json"
    "fmt"
    "math/rand"
    "sort"
    "strings"
    "sync"
    "time"
)

// Model 模型定义
type Model struct {
    Name        string  `json:"name"`
    Provider    string  `json:"provider"`
    InputPrice  float64 `json:"input_price"`  // $ per 1M tokens
    OutputPrice float64 `json:"output_price"` // $ per 1M tokens
    Quality     float64 `json:"quality"`      // 0-1, 基于Terminal-Bench 2.1
    Latency     float64 `json:"latency"`      // ms, 平均延迟
    MaxTokens   int     `json:"max_tokens"`
    IsAvailable bool    `json:"is_available"`
}

// RoutingDecision 路由决策结果
type RoutingDecision struct {
    SelectedModel *Model  `json:"selected_model"`
    Strategy      string  `json:"strategy"`
    EstimatedCost float64 `json:"estimated_cost"`
    Reason        string  `json:"reason"`
}

// Request 请求上下文
type Request struct {
    TaskType      string `json:"task_type"`      // coding, analysis, chat, batch
    InputTokens   int    `json:"input_tokens"`
    OutputTokens  int    `json:"output_tokens"`
    MaxLatencyMs  int    `json:"max_latency_ms"`
    MaxCostUsd    float64 `json:"max_cost_usd"`
    MinQuality    float64 `json:"min_quality"`
    Priority      int    `json:"priority"`       // 1-10
}

// ModelRouter 多模型路由器
type ModelRouter struct {
    models    []*Model
    usageLog  []UsageRecord
    mu        sync.RWMutex
    stats     RouterStats
}

type UsageRecord struct {
    Timestamp   time.Time
    ModelName   string
    TaskType    string
    Cost        float64
    Latency     float64
    TokensIn    int
    TokensOut   int
}

type RouterStats struct {
    TotalRequests   int64
    TotalCost       float64
    ModelCounts     map[string]int64
    AverageLatency  float64
}

func NewModelRouter() *ModelRouter {
    r := &ModelRouter{
        models: []*Model{
            // GPT-5.6 家族
            {Name: "GPT-5.6 Sol", Provider: "OpenAI", InputPrice: 5, OutputPrice: 30,
                Quality: 0.95, Latency: 3000, MaxTokens: 1500000, IsAvailable: true},
            {Name: "GPT-5.6 Terra", Provider: "OpenAI", InputPrice: 2.5, OutputPrice: 15,
                Quality: 0.88, Latency: 1500, MaxTokens: 1500000, IsAvailable: true},
            {Name: "GPT-5.6 Luna", Provider: "OpenAI", InputPrice: 1, OutputPrice: 6,
                Quality: 0.78, Latency: 500, MaxTokens: 1500000, IsAvailable: true},
            // 竞品
            {Name: "Claude Fable 5", Provider: "Anthropic", InputPrice: 10, OutputPrice: 50,
                Quality: 0.93, Latency: 2500, MaxTokens: 500000, IsAvailable: true},
            {Name: "Claude Sonnet 5", Provider: "Anthropic", InputPrice: 2, OutputPrice: 10,
                Quality: 0.85, Latency: 1200, MaxTokens: 1000000, IsAvailable: true},
            {Name: "DeepSeek V4", Provider: "DeepSeek", InputPrice: 0.435, OutputPrice: 0.87,
                Quality: 0.75, Latency: 800, MaxTokens: 1000000, IsAvailable: true},
        },
        stats: RouterStats{
            ModelCounts: make(map[string]int64),
        },
    }
    return r
}

// Route 根据请求路由到最优模型
func (r *ModelRouter) Route(req Request, strategy string) RoutingDecision {
    r.mu.Lock()
    defer r.mu.Unlock()
    
    var decision RoutingDecision
    
    switch strategy {
    case "cost_first":
        decision = r.routeCostFirst(req)
    case "quality_first":
        decision = r.routeQualityFirst(req)
    case "latency_first":
        decision = r.routeLatencyFirst(req)
    case "weighted_random":
        decision = r.routeWeightedRandom(req)
    default:
        decision = r.routeBalanced(req)
    }
    
    // 记录使用
    if decision.SelectedModel != nil {
        r.stats.TotalRequests++
        r.stats.TotalCost += decision.EstimatedCost
        r.stats.ModelCounts[decision.SelectedModel.Name]++
        
        r.usageLog = append(r.usageLog, UsageRecord{
            Timestamp: time.Now(),
            ModelName: decision.SelectedModel.Name,
            TaskType:  req.TaskType,
            Cost:      decision.EstimatedCost,
            TokensIn:  req.InputTokens,
            TokensOut: req.OutputTokens,
        })
    }
    
    return decision
}

// routeCostFirst 成本优先策略
func (r *ModelRouter) routeCostFirst(req Request) RoutingDecision {
    var candidates []*Model
    
    for _, m := range r.models {
        if !m.IsAvailable {
            continue
        }
        cost := estimateCost(m, req)
        if req.MaxCostUsd > 0 && cost > req.MaxCostUsd {
            continue
        }
        if m.Quality >= req.MinQuality {
            candidates = append(candidates, m)
        }
    }
    
    if len(candidates) == 0 {
        // 回退到满足质量的最低成本模型
        for _, m := range r.models {
            if m.IsAvailable && m.Quality >= req.MinQuality {
                candidates = append(candidates, m)
            }
        }
    }
    
    if len(candidates) == 0 {
        return RoutingDecision{Reason: "no available models matching criteria"}
    }
    
    // 按成本排序
    sort.Slice(candidates, func(i, j int) bool {
        return estimateCost(candidates[i], req) < estimateCost(candidates[j], req)
    })
    
    best := candidates[0]
    return RoutingDecision{
        SelectedModel: best,
        Strategy:      "cost_first",
        EstimatedCost: estimateCost(best, req),
        Reason:        fmt.Sprintf("最低成本模型: %s ($%.6f)", best.Name, estimateCost(best, req)),
    }
}

// routeQualityFirst 质量优先策略
func (r *ModelRouter) routeQualityFirst(req Request) RoutingDecision {
    var candidates []*Model
    
    for _, m := range r.models {
        if !m.IsAvailable {
            continue
        }
        cost := estimateCost(m, req)
        if req.MaxCostUsd > 0 && cost > req.MaxCostUsd {
            continue
        }
        if m.Latency <= float64(req.MaxLatencyMs) || req.MaxLatencyMs == 0 {
            candidates = append(candidates, m)
        }
    }
    
    if len(candidates) == 0 {
        candidates = r.getAvailableModels()
    }
    
    // 按质量排序
    sort.Slice(candidates, func(i, j int) bool {
        return candidates[i].Quality > candidates[j].Quality
    })
    
    best := candidates[0]
    return RoutingDecision{
        SelectedModel: best,
        Strategy:      "quality_first",
        EstimatedCost: estimateCost(best, req),
        Reason:        fmt.Sprintf("最高质量模型: %s (质量分数: %.2f)", best.Name, best.Quality),
    }
}

// routeLatencyFirst 延迟优先策略
func (r *ModelRouter) routeLatencyFirst(req Request) RoutingDecision {
    var candidates []*Model
    
    for _, m := range r.models {
        if !m.IsAvailable {
            continue
        }
        if m.Quality >= req.MinQuality {
            candidates = append(candidates, m)
        }
    }
    
    // 按延迟排序
    sort.Slice(candidates, func(i, j int) bool {
        return candidates[i].Latency < candidates[j].Latency
    })
    
    best := candidates[0]
    return RoutingDecision{
        SelectedModel: best,
        Strategy:      "latency_first",
        EstimatedCost: estimateCost(best, req),
        Reason:        fmt.Sprintf("最低延迟模型: %s (%.0fms)", best.Name, best.Latency),
    }
}

// routeWeightedRandom 加权随机策略
func (r *ModelRouter) routeWeightedRandom(req Request) RoutingDecision {
    type weightedModel struct {
        model  *Model
        weight float64
    }
    
    var candidates []weightedModel
    
    for _, m := range r.models {
        if !m.IsAvailable || m.Quality < req.MinQuality {
            continue
        }
        
        // 权重 = 质量分数 / 成本
        cost := estimateCost(m, req)
        if cost <= 0 {
            cost = 0.001
        }
        weight := m.Quality / cost
        
        // 任务类型偏好调整
        switch req.TaskType {
        case "coding":
            if m.Name == "GPT-5.6 Sol" || m.Name == "Claude Fable 5" {
                weight *= 1.5
            }
        case "batch":
            if strings.Contains(m.Name, "Luna") || strings.Contains(m.Name, "DeepSeek") {
                weight *= 2.0
            }
        case "chat":
            if strings.Contains(m.Name, "Terra") || strings.Contains(m.Name, "Sonnet") {
                weight *= 1.3
            }
        }
        
        candidates = append(candidates, weightedModel{m, weight})
    }
    
    // 轮盘赌选择
    totalWeight := 0.0
    for _, c := range candidates {
        totalWeight += c.weight
    }
    
    rnd := rand.Float64() * totalWeight
    cumulative := 0.0
    for _, c := range candidates {
        cumulative += c.weight
        if rnd <= cumulative {
            return RoutingDecision{
                SelectedModel: c.model,
                Strategy:      "weighted_random",
                EstimatedCost: estimateCost(c.model, req),
                Reason:        fmt.Sprintf("加权随机选择: %s (权重: %.2f)", c.model.Name, c.weight),
            }
        }
    }
    
    // Fallback
    return RoutingDecision{Reason: "weighted random selection failed"}
}

// routeBalanced 均衡策略（默认）
func (r *ModelRouter) routeBalanced(req Request) RoutingDecision {
    // 根据任务类型推荐
    taskModelMap := map[string]string{
        "coding":    "GPT-5.6 Sol",
        "analysis":  "GPT-5.6 Terra",
        "chat":      "GPT-5.6 Terra",
        "batch":     "GPT-5.6 Luna",
        "research":  "GPT-5.6 Sol",
        "debugging": "GPT-5.6 Sol",
        "review":    "GPT-5.6 Terra",
    }
    
    recommendedName := taskModelMap[req.TaskType]
    if recommendedName == "" {
        recommendedName = "GPT-5.6 Terra"
    }
    
    for _, m := range r.models {
        if m.Name == recommendedName && m.IsAvailable {
            return RoutingDecision{
                SelectedModel: m,
                Strategy:      "balanced",
                EstimatedCost: estimateCost(m, req),
                Reason:        fmt.Sprintf("任务类型推荐: %s → %s", req.TaskType, m.Name),
            }
        }
    }
    
    // Fallback to Terra
    for _, m := range r.models {
        if m.Name == "GPT-5.6 Terra" && m.IsAvailable {
            return RoutingDecision{
                SelectedModel: m,
                Strategy:      "balanced_fallback",
                EstimatedCost: estimateCost(m, req),
                Reason:        "默认回退到 Terra",
            }
        }
    }
    
    return RoutingDecision{Reason: "no available models"}
}

func (r *ModelRouter) getAvailableModels() []*Model {
    var available []*Model
    for _, m := range r.models {
        if m.IsAvailable {
            available = append(available, m)
        }
    }
    return available
}

func estimateCost(m *Model, req Request) float64 {
    inputCost := float64(req.InputTokens) * m.InputPrice / 1_000_000
    outputCost := float64(req.OutputTokens) * m.OutputPrice / 1_000_000
    return inputCost + outputCost
}

// ============================================================
// 优先级队列 - 支持高优先级任务插队
// ============================================================

type PriorityRequest struct {
    Request  Request
    Decision RoutingDecision
    Index    int
}

type PriorityQueue []*PriorityRequest

func (pq PriorityQueue) Len() int { return len(pq) }
func (pq PriorityQueue) Less(i, j int) bool {
    return pq[i].Request.Priority > pq[j].Request.Priority
}
func (pq PriorityQueue) Swap(i, j int) {
    pq[i], pq[j] = pq[j], pq[i]
    pq[i].Index = i
    pq[j].Index = j
}
func (pq *PriorityQueue) Push(x interface{}) {
    n := len(*pq)
    item := x.(*PriorityRequest)
    item.Index = n
    *pq = append(*pq, item)
}
func (pq *PriorityQueue) Pop() interface{} {
    old := *pq
    n := len(old)
    item := old[n-1]
    old[n-1] = nil
    item.Index = -1
    *pq = old[0 : n-1]
    return item
}

// ============================================================
// 性能测试
// ============================================================

func main() {
    fmt.Println("=" + strings.Repeat("=", 69))
    fmt.Println("  GPT-5.6 多模型智能路由中间件 性能测试")
    fmt.Println("=" + strings.Repeat("=", 69))
    
    router := NewModelRouter()
    
    testRequests := []Request{
        {"coding", 2000, 1000, 5000, 0.5, 0.85, 8},
        {"chat", 500, 200, 2000, 0.05, 0.7, 3},
        {"batch", 10000, 5000, 10000, 0.1, 0.6, 1},
        {"analysis", 3000, 1500, 3000, 0.2, 0.8, 5},
        {"research", 5000, 3000, 8000, 1.0, 0.9, 10},
    }
    
    strategies := []string{"cost_first", "quality_first", "latency_first", "weighted_random", "balanced"}
    
    for _, req := range testRequests {
        fmt.Printf("\n📋 任务: %s (输入:%d, 输出:%d, 优先级:%d)\n",
            req.TaskType, req.InputTokens, req.OutputTokens, req.Priority)
        fmt.Println(strings.Repeat("-", 70))
        
        for _, strategy := range strategies {
            decision := router.Route(req, strategy)
            if decision.SelectedModel != nil {
                fmt.Printf("  %-16s → %-20s | $%.6f | %s\n",
                    strategy,
                    decision.SelectedModel.Name,
                    decision.EstimatedCost,
                    decision.Reason[:minInt(40, len(decision.Reason))])
            } else {
                fmt.Printf("  %-16s → ❌ %s\n", strategy, decision.Reason)
            }
        }
    }
    
    fmt.Println("\n" + strings.Repeat("=", 69))
    fmt.Println("  路由统计")
    fmt.Println(strings.Repeat("=", 69))
    fmt.Printf("总请求数: %d\n", router.stats.TotalRequests)
    fmt.Printf("总成本: $%.4f\n", router.stats.TotalCost)
    fmt.Println("各模型使用次数:")
    for name, count := range router.stats.ModelCounts {
        fmt.Printf("  %-20s: %d次\n", name, count)
    }
}

func minInt(a, b int) int {
    if a < b { return a }
    return b
}

#!/usr/bin/env python3
"""
Terminal-Bench 2.1 评测数据分析工具
支持模型性能对比、成本效益分析和场景推荐
"""

import json
import math
from typing import Dict, List, Tuple
from dataclasses import dataclass


@dataclass
class BenchmarkResult:
    """评测结果"""
    model: str
    standard_score: float
    ultra_score: float
    input_price: float
    output_price: float
    category: str  # flagship, balanced, lightweight, opensource


class BenchmarkAnalyzer:
    """评测分析器"""
    
    def __init__(self):
        self.results = [
            BenchmarkResult("GPT-5.6 Sol Ultra", 91.9, 91.9, 5, 30, "flagship"),
            BenchmarkResult("GPT-5.6 Sol", 88.8, 91.9, 5, 30, "flagship"),
            BenchmarkResult("Claude Mythos 5", 88.0, 88.0, 10, 50, "flagship"),
            BenchmarkResult("Claude Fable 5", 84.3, 84.3, 10, 50, "flagship"),
            BenchmarkResult("GPT-5.6 Terra", 82.4, 86.1, 2.5, 15, "balanced"),
            BenchmarkResult("GPT-5.5", 79.2, 83.5, 5, 15, "balanced"),
            BenchmarkResult("GPT-5.6 Luna", 72.1, 75.8, 1, 6, "lightweight"),
            BenchmarkResult("DeepSeek V4-Pro", 72.8, 72.8, 0.435, 0.87, "opensource"),
        ]
    
    def cost_performance_analysis(self, input_tokens: int = 1000, output_tokens: int = 500) -> List[Dict]:
        """
        成本效益分析：每美元获得的性能分数。
        这是GPT-5.6定价策略的核心竞争力指标。
        """
        analysis = []
        
        for r in self.results:
            cost = (input_tokens * r.input_price + output_tokens * r.output_price) / 1_000_000
            score_per_dollar = r.standard_score / cost if cost > 0 else float('inf')
            
            analysis.append({
                "model": r.model,
                "benchmark_score": r.standard_score,
                "cost_per_task": round(cost, 6),
                "score_per_dollar": round(score_per_dollar, 0),
                "category": r.category,
            })
        
        # 按性价比排序
        analysis.sort(key=lambda x: x["score_per_dollar"], reverse=True)
        return analysis
    
    def ultra_mode_improvement(self) -> List[Dict]:
        """分析Ultra模式带来的性能提升"""
        improvements = []
        
        for r in self.results:
            if r.ultra_score > r.standard_score:
                improvement = ((r.ultra_score - r.standard_score) / r.standard_score) * 100
                improvements.append({
                    "model": r.model,
                    "standard": r.standard_score,
                    "ultra": r.ultra_score,
                    "improvement_pct": round(improvement, 2),
                    "category": r.category,
                })
        
        improvements.sort(key=lambda x: x["improvement_pct"], reverse=True)
        return improvements
    
    def recommend_model(self, 
                       budget_constraint: float = 0.1,
                       min_quality: float = 0.7,
                       task_type: str = "general") -> List[Tuple[str, str]]:
        """
        根据约束条件推荐最优模型。
        """
        recommendations = []
        
        for r in self.results:
            cost = (1000 * r.input_price + 500 * r.output_price) / 1_000_000
            score_normalized = r.standard_score / 100.0
            
            if cost > budget_constraint:
                continue
            if score_normalized < min_quality:
                continue
            
            # 任务类型适配
            task_fit = 1.0
            if task_type == "coding" and r.category == "flagship":
                task_fit = 1.3
            elif task_type == "batch" and r.category == "lightweight":
                task_fit = 1.4
            elif task_type == "chat" and r.category == "balanced":
                task_fit = 1.2
            
            effective_score = score_normalized * task_fit / cost
            recommendations.append((r.model, f"效益评分: {effective_score:.1f}"))
        
        recommendations.sort(key=lambda x: float(x[1].split(': ')[1]), reverse=True)
        return recommendations


def main():
    analyzer = BenchmarkAnalyzer()
    
    print("=" * 70)
    print("Terminal-Bench 2.1 评测深度分析")
    print("=" * 70)
    
    # 1. 成本效益分析
    print("\n📊 成本效益分析（每美元得分）")
    print("-" * 70)
    print(f"{'模型':<25} {'得分':<8} {'成本/任务':<12} {'得分/美元':<12} {'类别':<12}")
    print("-" * 70)
    
    for item in analyzer.cost_performance_analysis():
        print(f"{item['model']:<25} {item['benchmark_score']:<8.1f} "
              f"${item['cost_per_task']:<8.6f} {item['score_per_dollar']:<12,.0f} {item['category']:<12}")
    
    # 2. Ultra模式提升分析
    print("\n\n📊 Ultra模式性能提升")
    print("-" * 70)
    print(f"{'模型':<25} {'标准':<8} {'Ultra':<8} {'提升%':<10}")
    print("-" * 70)
    
    for item in analyzer.ultra_mode_improvement():
        print(f"{item['model']:<25} {item['standard']:<8.1f} "
              f"{item['ultra']:<8.1f} +{item['improvement_pct']:<7.2f}%")
    
    # 3. 场景推荐
    print("\n\n🎯 场景推荐")
    print("-" * 70)
    
    scenarios = [
        ("代码开发", "coding", 0.5, 0.85),
        ("批量处理", "batch", 0.05, 0.6),
        ("日常对话", "chat", 0.1, 0.7),
        ("通用任务", "general", 0.2, 0.75),
    ]
    
    for name, task_type, budget, min_qual in scenarios:
        print(f"\n  {name} (预算: ${budget}, 最低质量: {min_qual})")
        recs = analyzer.recommend_model(budget, min_qual, task_type)
        for model, reason in recs[:3]:
            print(f"    → {model:<25} {reason}")


if __name__ == "__main__":
    main()