AI巨头IPO竞速与苹果WWDC 2026：AI资本化与消费级AI的新篇章

// llm_gateway.go
// 高性能AI多模型路由网关 - Go实现
// 支持OpenAI、Anthropic、Google Gemini多模型路由与负载均衡

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"math"
	"net/http"
	"sort"
	"strings"
	"sync"
	"time"
)

// ProviderType 模型提供商类型
type ProviderType string

const (
	ProviderOpenAI    ProviderType = "openai"
	ProviderAnthropic ProviderType = "anthropic"
	ProviderGemini    ProviderType = "gemini"
)

// ModelCapability 模型能力描述
type ModelCapability struct {
	Provider      ProviderType  `json:"provider"`
	ModelName     string        `json:"model_name"`
	CostPer1KIn   float64       `json:"cost_per_1k_in"`
	CostPer1KOut  float64       `json:"cost_per_1k_out"`
	ContextWindow int           `json:"context_window"` // token数
	AvgLatency    time.Duration `json:"avg_latency"`
	IsAvailable   bool          `json:"is_available"`
	Priority      int           `json:"priority"` // 路由优先级
}

// ModelRegistry 模型注册表
type ModelRegistry struct {
	mu     sync.RWMutex
	models map[string]*ModelCapability
}

func NewModelRegistry() *ModelRegistry {
	return &ModelRegistry{
		models: make(map[string]*ModelCapability),
	}
}

func (r *ModelRegistry) Register(key string, m *ModelCapability) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.models[key] = m
}

func (r *ModelRegistry) Get(key string) *ModelCapability {
	r.mu.RLock()
	defer r.mu.RUnlock()
	return r.models[key]
}

func (r *ModelRegistry) ListAvailable() []*ModelCapability {
	r.mu.RLock()
	defer r.mu.RUnlock()
	var result []*ModelCapability
	for _, m := range r.models {
		if m.IsAvailable {
			result = append(result, m)
		}
	}
	return result
}

// RouterStrategy 路由策略接口
type RouterStrategy interface {
	Select(models []*ModelCapability, req *ChatRequest) *ModelCapability
}

// CostOptimizedStrategy 成本优化路由策略
type CostOptimizedStrategy struct{}

func (s *CostOptimizedStrategy) Select(models []*ModelCapability, req *ChatRequest) *ModelCapability {
	if len(models) == 0 {
		return nil
	}
	sort.Slice(models, func(i, j int) bool {
		costI := models[i].CostPer1KIn + models[i].CostPer1KOut
		costJ := models[j].CostPer1KIn + models[j].CostPer1KOut
		return costI < costJ
	})
	// 检查上下文窗口是否满足需求
	for _, m := range models {
		if req.EstimatedTokens <= m.ContextWindow {
			return m
		}
	}
	return models[0]
}

// LatencyOptimizedStrategy 延迟优化路由策略
type LatencyOptimizedStrategy struct{}

func (s *LatencyOptimizedStrategy) Select(models []*ModelCapability, req *ChatRequest) *ModelCapability {
	if len(models) == 0 {
		return nil
	}
	sort.Slice(models, func(i, j int) bool {
		return models[i].AvgLatency < models[j].AvgLatency
	})
	for _, m := range models {
		if req.EstimatedTokens <= m.ContextWindow {
			return m
		}
	}
	return models[0]
}

// PriorityFailoverStrategy 优先级故障转移策略
type PriorityFailoverStrategy struct{}

func (s *PriorityFailoverStrategy) Select(models []*ModelCapability, req *ChatRequest) *ModelCapability {
	if len(models) == 0 {
		return nil
	}
	sort.Slice(models, func(i, j int) bool {
		return models[i].Priority < models[j].Priority
	})
	for _, m := range models {
		if m.IsAvailable && req.EstimatedTokens <= m.ContextWindow {
			return m
		}
	}
	// 所有模型都不满足上下文窗口要求，选优先级最高的
	for _, m := range models {
		if m.IsAvailable {
			return m
		}
	}
	return nil
}

// ChatRequest 统一的聊天请求
type ChatRequest struct {
	Messages        []Message `json:"messages"`
	EstimatedTokens int       `json:"estimated_tokens"`
	RouteStrategy   string    `json:"route_strategy,omitempty"`
	UserID          string    `json:"user_id,omitempty"`
	Tier            string    `json:"tier,omitempty"` // premium, standard, free
}

type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

// AIAdapter 模型适配器接口
type AIAdapter interface {
	Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error)
	Stream(ctx context.Context, req *ChatRequest) (<-chan TokenChunk, error)
}

type ChatResponse struct {
	Content      string `json:"content"`
	Model        string `json:"model"`
	Provider     string `json:"provider"`
	TokensIn     int    `json:"tokens_in"`
	TokensOut    int    `json:"tokens_out"`
	LatencyMs    int64  `json:"latency_ms"`
}

type TokenChunk struct {
	Content string `json:"content"`
	Done    bool   `json:"done"`
}

// OpenAIAdapter OpenAI模型适配器
type OpenAIAdapter struct {
	apiKey  string
	baseURL string
	client  *http.Client
}

func NewOpenAIAdapter(apiKey string) *OpenAIAdapter {
	return &OpenAIAdapter{
		apiKey:  apiKey,
		baseURL: "https://api.openai.com/v1",
		client:  &http.Client{Timeout: 60 * time.Second},
	}
}

func (a *OpenAIAdapter) Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error) {
	payload := map[string]interface{}{
		"model":    "gpt-4o",
		"messages": req.Messages,
	}
	body, _ := json.Marshal(payload)
	httpReq, _ := http.NewRequestWithContext(ctx, "POST",
		a.baseURL+"/chat/completions", strings.NewReader(string(body)))
	httpReq.Header.Set("Authorization", "Bearer "+a.apiKey)
	httpReq.Header.Set("Content-Type", "application/json")

	start := time.Now()
	resp, err := a.client.Do(httpReq)
	if err != nil {
		return nil, fmt.Errorf("openai request failed: %w", err)
	}
	defer resp.Body.Close()

	respBody, _ := io.ReadAll(resp.Body)
	var result struct {
		Choices []struct {
			Message struct {
				Content string `json:"content"`
			} `json:"message"`
		} `json:"choices"`
		Usage struct {
			PromptTokens     int `json:"prompt_tokens"`
			CompletionTokens int `json:"completion_tokens"`
		} `json:"usage"`
	}
	json.Unmarshal(respBody, &result)

	latency := time.Since(start).Milliseconds()
	content := ""
	if len(result.Choices) > 0 {
		content = result.Choices[0].Message.Content
	}

	return &ChatResponse{
		Content:   content,
		Model:     "gpt-4o",
		Provider:  string(ProviderOpenAI),
		TokensIn:  result.Usage.PromptTokens,
		TokensOut: result.Usage.CompletionTokens,
		LatencyMs: latency,
	}, nil
}

func (a *OpenAIAdapter) Stream(ctx context.Context, req *ChatRequest) (<-chan TokenChunk, error) {
	// 流式实现略
	return nil, nil
}

// AnthropicAdapter Anthropic模型适配器
type AnthropicAdapter struct {
	apiKey  string
	client  *http.Client
}

func NewAnthropicAdapter(apiKey string) *AnthropicAdapter {
	return &AnthropicAdapter{
		apiKey: apiKey,
		client: &http.Client{Timeout: 60 * time.Second},
	}
}

func (a *AnthropicAdapter) Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error) {
	// 将统一格式转换为Anthropic格式
	var anthropicMessages []map[string]interface{}
	for _, msg := range req.Messages {
		anthropicMessages = append(anthropicMessages, map[string]interface{}{
			"role":    msg.Role,
			"content": msg.Content,
		})
	}
	payload := map[string]interface{}{
		"model":         "claude-sonnet-4-6",
		"max_tokens":    4096,
		"messages":      anthropicMessages,
	}
	body, _ := json.Marshal(payload)
	httpReq, _ := http.NewRequestWithContext(ctx, "POST",
		"https://api.anthropic.com/v1/messages", strings.NewReader(string(body)))
	httpReq.Header.Set("x-api-key", a.apiKey)
	httpReq.Header.Set("anthropic-version", "2023-06-01")
	httpReq.Header.Set("Content-Type", "application/json")

	start := time.Now()
	resp, err := a.client.Do(httpReq)
	if err != nil {
		return nil, fmt.Errorf("anthropic request failed: %w", err)
	}
	defer resp.Body.Close()

	respBody, _ := io.ReadAll(resp.Body)
	var result struct {
		Content []struct {
			Text string `json:"text"`
		} `json:"content"`
		Usage struct {
			InputTokens  int `json:"input_tokens"`
			OutputTokens int `json:"output_tokens"`
		} `json:"usage"`
	}
	json.Unmarshal(respBody, &result)

	latency := time.Since(start).Milliseconds()
	content := ""
	if len(result.Content) > 0 {
		content = result.Content[0].Text
	}

	return &ChatResponse{
		Content:   content,
		Model:     "claude-sonnet-4-6",
		Provider:  string(ProviderAnthropic),
		TokensIn:  result.Usage.InputTokens,
		TokensOut: result.Usage.OutputTokens,
		LatencyMs: latency,
	}, nil
}

func (a *AnthropicAdapter) Stream(ctx context.Context, req *ChatRequest) (<-chan TokenChunk, error) {
	return nil, nil
}

// GeminiAdapter Google Gemini模型适配器
type GeminiAdapter struct {
	apiKey  string
	client  *http.Client
}

func NewGeminiAdapter(apiKey string) *GeminiAdapter {
	return &GeminiAdapter{
		apiKey: apiKey,
		client: &http.Client{Timeout: 60 * time.Second},
	}
}

func (a *GeminiAdapter) Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error) {
	var contents []map[string]interface{}
	for _, msg := range req.Messages {
		role := "user"
		if msg.Role == "assistant" || msg.Role == "model" {
			role = "model"
		}
		contents = append(contents, map[string]interface{}{
			"role": role,
			"parts": []map[string]string{
				{"text": msg.Content},
			},
		})
	}
	payload := map[string]interface{}{
		"contents": contents,
	}
	body, _ := json.Marshal(payload)
	httpReq, _ := http.NewRequestWithContext(ctx, "POST",
		fmt.Sprintf("https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-pro:generateContent?key=%s", a.apiKey),
		strings.NewReader(string(body)))
	httpReq.Header.Set("Content-Type", "application/json")

	start := time.Now()
	resp, err := a.client.Do(httpReq)
	if err != nil {
		return nil, fmt.Errorf("gemini request failed: %w", err)
	}
	defer resp.Body.Close()

	respBody, _ := io.ReadAll(resp.Body)
	var result struct {
		Candidates []struct {
			Content struct {
				Parts []struct {
					Text string `json:"text"`
				} `json:"parts"`
			} `json:"content"`
		} `json:"candidates"`
		UsageMetadata struct {
			PromptTokenCount     int `json:"promptTokenCount"`
			CandidatesTokenCount int `json:"candidatesTokenCount"`
		} `json:"usageMetadata"`
	}
	json.Unmarshal(respBody, &result)

	latency := time.Since(start).Milliseconds()
	content := ""
	if len(result.Candidates) > 0 && len(result.Candidates[0].Content.Parts) > 0 {
		content = result.Candidates[0].Content.Parts[0].Text
	}

	return &ChatResponse{
		Content:   content,
		Model:     "gemini-1.5-pro",
		Provider:  string(ProviderGemini),
		TokensIn:  result.UsageMetadata.PromptTokenCount,
		TokensOut: result.UsageMetadata.CandidatesTokenCount,
		LatencyMs: latency,
	}, nil
}

func (a *GeminiAdapter) Stream(ctx context.Context, req *ChatRequest) (<-chan TokenChunk, error) {
	return nil, nil
}

// AIGateway AI服务网关 - 统一入口
type AIGateway struct {
	registry  *ModelRegistry
	adapters  map[ProviderType]AIAdapter
	strategies map[string]RouterStrategy
	stats     *GatewayStats
	mu        sync.RWMutex
}

type GatewayStats struct {
	mu           sync.Mutex
	TotalReqs    int64             `json:"total_requests"`
	SuccessReqs  int64             `json:"success_requests"`
	FailReqs     int64             `json:"fail_requests"`
	LatencySum   int64             `json:"latency_sum_ms"`
	ModelCounter map[string]int64  `json:"model_counter"`
	ProviderCost map[string]float64 `json:"provider_cost"`
}

func NewGatewayStats() *GatewayStats {
	return &GatewayStats{
		ModelCounter: make(map[string]int64),
		ProviderCost: make(map[string]float64),
	}
}

func (s *GatewayStats) Record(req *ChatRequest, resp *ChatResponse) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.TotalReqs++
	s.LatencySum += resp.LatencyMs
	s.ModelCounter[resp.Model]++
	// 估算成本
	cost := float64(resp.TokensIn)*0.000015 + float64(resp.TokensOut)*0.00006
	s.ProviderCost[resp.Provider] += cost
}

func NewAIGateway(openAIKey, anthropicKey, geminiKey string) *AIGateway {
	gw := &AIGateway{
		registry:   NewModelRegistry(),
		adapters:   make(map[ProviderType]AIAdapter),
		strategies: make(map[string]RouterStrategy),
		stats:      NewGatewayStats(),
	}

	// 注册模型
	gw.registry.Register("gpt-4o", &ModelCapability{
		Provider: ProviderOpenAI, ModelName: "gpt-4o",
		CostPer1KIn: 0.0025, CostPer1KOut: 0.01,
		ContextWindow: 128000, AvgLatency: 1800 * time.Millisecond,
		IsAvailable: true, Priority: 1,
	})
	gw.registry.Register("claude-sonnet-4-6", &ModelCapability{
		Provider: ProviderAnthropic, ModelName: "claude-sonnet-4-6",
		CostPer1KIn: 0.003, CostPer1KOut: 0.015,
		ContextWindow: 200000, AvgLatency: 2200 * time.Millisecond,
		IsAvailable: true, Priority: 1,
	})
	gw.registry.Register("gemini-1.5-pro", &ModelCapability{
		Provider: ProviderGemini, ModelName: "gemini-1.5-pro",
		CostPer1KIn: 0.00125, CostPer1KOut: 0.005,
		ContextWindow: 1000000, AvgLatency: 1500 * time.Millisecond,
		IsAvailable: true, Priority: 2,
	})
	// 低成本模型
	gw.registry.Register("gpt-4o-mini", &ModelCapability{
		Provider: ProviderOpenAI, ModelName: "gpt-4o-mini",
		CostPer1KIn: 0.00015, CostPer1KOut: 0.0006,
		ContextWindow: 128000, AvgLatency: 800 * time.Millisecond,
		IsAvailable: true, Priority: 3,
	})
	gw.registry.Register("claude-haiku-4-5", &ModelCapability{
		Provider: ProviderAnthropic, ModelName: "claude-haiku-4-5",
		CostPer1KIn: 0.00025, CostPer1KOut: 0.00125,
		ContextWindow: 200000, AvgLatency: 600 * time.Millisecond,
		IsAvailable: true, Priority: 3,
	})

	// 初始化适配器
	gw.adapters[ProviderOpenAI] = NewOpenAIAdapter(openAIKey)
	gw.adapters[ProviderAnthropic] = NewAnthropicAdapter(anthropicKey)
	gw.adapters[ProviderGemini] = NewGeminiAdapter(geminiKey)

	// 注册路由策略
	gw.strategies["cost"] = &CostOptimizedStrategy{}
	gw.strategies["latency"] = &LatencyOptimizedStrategy{}
	gw.strategies["failover"] = &PriorityFailoverStrategy{}

	return gw
}

// Route 路由请求到合适模型
func (gw *AIGateway) Route(ctx context.Context, req *ChatRequest) (*ChatResponse, error) {
	// 1. 选择路由策略
	strategyName := req.RouteStrategy
	if strategyName == "" {
		// 根据用户层级选择默认策略
		switch req.Tier {
		case "premium":
			strategyName = "failover" // 优先高质量
		case "standard":
			strategyName = "latency"  // 优先速度
		default:
			strategyName = "cost"     // 优先成本
		}
	}

	strategy, ok := gw.strategies[strategyName]
	if !ok {
		return nil, fmt.Errorf("unknown strategy: %s", strategyName)
	}

	// 2. 获取可用模型
	available := gw.registry.ListAvailable()
	if len(available) == 0 {
		return nil, fmt.Errorf("no available models")
	}

	// 3. 路由选择
	selected := strategy.Select(available, req)
	if selected == nil {
		return nil, fmt.Errorf("no suitable model found")
	}

	// 4. 获取适配器并执行
	adapter, ok := gw.adapters[selected.Provider]
	if !ok {
		return nil, fmt.Errorf("no adapter for provider: %s", selected.Provider)
	}

	// 5. 添加重试和故障转移
	maxRetries := 2
	var lastErr error
	for attempt := 0; attempt <= maxRetries; attempt++ {
		resp, err := adapter.Chat(ctx, req)
		if err == nil {
			gw.stats.Record(req, resp)
			return resp, nil
		}
		lastErr = err
		// 标记当前模型不可用，尝试下一个
		gw.mu.Lock()
		if m := gw.registry.Get(string(selected.Provider) + "-" + selected.ModelName); m != nil {
			m.IsAvailable = false
		}
		gw.mu.Unlock()

		// 故障转移：选择下一个可用模型
		available = gw.registry.ListAvailable()
		selected = strategy.Select(available, req)
		if selected == nil {
			break
		}
		adapter = gw.adapters[selected.Provider]
	}

	return nil, fmt.Errorf("all models failed after retries, last error: %w", lastErr)
}

// GetStats 获取网关统计
func (gw *AIGateway) GetStats() *GatewayStats {
	return gw.stats
}

// HealthCheck 健康检查
func (gw *AIGateway) HealthCheck() map[string]bool {
	result := make(map[string]bool)
	for _, m := range gw.registry.ListAvailable() {
		result[string(m.Provider)+"/"+m.ModelName] = m.IsAvailable
	}
	return result
}

// HTTP服务入口
func main() {
	gw := NewAIGateway(
		"sk-openai-xxx",
		"sk-ant-xxx",
		"AIzaSyXXX",
	)

	// 初始化模型注册表
	_ = gw

	http.HandleFunc("/v1/chat/completions", func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
			return
		}
		var req ChatRequest
		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		// 估算token数（简化版）
		for _, msg := range req.Messages {
			req.EstimatedTokens += len(strings.Fields(msg.Content)) * 2
		}
		ctx := r.Context()
		resp, err := gw.Route(ctx, &req)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(resp)
	})

	http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
		json.NewEncoder(w).Encode(gw.HealthCheck())
	})

	http.HandleFunc("/stats", func(w http.ResponseWriter, r *http.Request) {
		json.NewEncoder(w).Encode(gw.GetStats())
	})

	log.Println("AI Gateway starting on :8080...")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

"""
ai_model_router.py
智能AI模型路由选择器 - Python实现
基于请求特征实时选择最优模型
"""

import time
import json
import hashlib
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, Callable
from collections import deque
import statistics


class Provider(Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    GEMINI = "gemini"
    LOCAL = "local"


class TaskType(Enum):
    CHAT = "chat"
    CODE = "code"
    REASONING = "reasoning"
    EXTRACTION = "extraction"
    CLASSIFICATION = "classification"
    SUMMARIZATION = "summarization"


@dataclass
class ModelConfig:
    """模型配置"""
    provider: Provider
    model_name: str
    cost_per_1k_input: float  # 美元
    cost_per_1k_output: float
    context_window: int
    latency_p50: float  # 毫秒
    latency_p95: float
    is_available: bool = True
    tasks: list[TaskType] = field(default_factory=list)

    def estimate_cost(self, input_tokens: int, output_tokens: int) -> float:
        """估算请求成本"""
        return (input_tokens / 1000 * self.cost_per_1k_input +
                output_tokens / 1000 * self.cost_per_1k_output)


@dataclass
class RoutingDecision:
    """路由决策记录"""
    model: ModelConfig
    strategy: str
    estimated_cost: float
    decision_time_ms: float
    reason: str


class LatencyTracker:
    """延迟跟踪器 - 滑动窗口P50/P95计算"""

    def __init__(self, window_size: int = 100):
        self.window: deque = deque(maxlen=window_size)

    def record(self, latency_ms: float):
        self.window.append(latency_ms)

    @property
    def p50(self) -> float:
        if not self.window:
            return 0.0
        return statistics.median(self.window)

    @property
    def p95(self) -> float:
        if not self.window:
            return 0.0
        sorted_data = sorted(self.window)
        idx = int(len(sorted_data) * 0.95)
        return sorted_data[min(idx, len(sorted_data) - 1)]


class SemanticCache:
    """语义缓存 - 基于嵌入向量的相似度匹配"""

    def __init__(self, similarity_threshold: float = 0.92):
        self.cache: dict[str, tuple[str, float]] = {}
        self.threshold = similarity_threshold
        self.hits = 0
        self.misses = 0

    def _simple_hash(self, text: str) -> str:
        """简易哈希（生产环境应替换为真实嵌入）"""
        return hashlib.sha256(text.encode()).hexdigest()[:16]

    def get(self, query: str) -> Optional[str]:
        """从缓存获取"""
        key = self._simple_hash(query)
        if key in self.cache:
            self.hits += 1
            return self.cache[key][0]
        # 简化的语义匹配（生产环境应使用向量数据库）
        for cached_key, (cached_response, similarity) in self.cache.items():
            if similarity >= self.threshold:
                self.hits += 1
                return cached_response
        self.misses += 1
        return None

    def set(self, query: str, response: str, similarity: float = 1.0):
        key = self._simple_hash(query)
        self.cache[key] = (response, similarity)

    @property
    def hit_rate(self) -> float:
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0.0


class AIModelRouter:
    """
    AI模型路由器 - 根据成本、延迟、任务类型智能选择模型
    支持成本优化、延迟优化、故障转移等多策略
    """

    def __init__(self):
        self.models: dict[str, ModelConfig] = {}
        self.latency_trackers: dict[str, LatencyTracker] = {}
        self.cache = SemanticCache()
        self.decisions: list[RoutingDecision] = []

        self._init_default_models()

    def _init_default_models(self):
        """初始化默认模型注册表"""
        models = [
            # 旗舰模型 - 高能力高成本
            ModelConfig(Provider.OPENAI, "gpt-4o",
                        0.0025, 0.01, 128000,
                        1800, 3500, True,
                        [TaskType.CHAT, TaskType.REASONING, TaskType.CODE]),
            ModelConfig(Provider.ANTHROPIC, "claude-sonnet-4-6",
                        0.003, 0.015, 200000,
                        2200, 4000, True,
                        [TaskType.CODE, TaskType.REASONING, TaskType.CHAT]),
            ModelConfig(Provider.GEMINI, "gemini-1.5-pro",
                        0.00125, 0.005, 1000000,
                        1500, 2800, True,
                        [TaskType.CHAT, TaskType.REASONING, TaskType.SUMMARIZATION]),
            # 经济型模型 - 成本效益优先
            ModelConfig(Provider.OPENAI, "gpt-4o-mini",
                        0.00015, 0.0006, 128000,
                        800, 1500, True,
                        [TaskType.CHAT, TaskType.CLASSIFICATION,
                         TaskType.EXTRACTION, TaskType.SUMMARIZATION]),
            ModelConfig(Provider.ANTHROPIC, "claude-haiku-4-5",
                        0.00025, 0.00125, 200000,
                        600, 1200, True,
                        [TaskType.CHAT, TaskType.CLASSIFICATION,
                         TaskType.EXTRACTION]),
            ModelConfig(Provider.GEMINI, "gemini-1.5-flash",
                        0.000075, 0.0003, 1000000,
                        500, 1000, True,
                        [TaskType.CHAT, TaskType.CLASSIFICATION,
                         TaskType.EXTRACTION, TaskType.SUMMARIZATION]),
        ]
        for m in models:
            key = f"{m.provider.value}/{m.model_name}"
            self.models[key] = m
            self.latency_trackers[key] = LatencyTracker()

    def register_model(self, config: ModelConfig):
        """注册新模型"""
        key = f"{config.provider.value}/{config.model_name}"
        self.models[key] = config
        if key not in self.latency_trackers:
            self.latency_trackers[key] = LatencyTracker()

    def _estimate_tokens(self, text: str) -> int:
        """简易token估算"""
        return len(text.split()) * 2

    def _get_suitable_models(self, task_type: TaskType,
                              input_tokens: int) -> list[ModelConfig]:
        """获取适合当前任务的可用模型"""
        suitable = []
        for model in self.models.values():
            if not model.is_available:
                continue
            if task_type not in model.tasks:
                continue
            if input_tokens > model.context_window:
                continue
            suitable.append(model)
        return suitable

    def cost_optimized_select(self, task_type: TaskType,
                               input_tokens: int,
                               output_tokens: int = 500) -> Optional[ModelConfig]:
        """成本优化策略 - 选择最便宜的可用模型"""
        suitable = self._get_suitable_models(task_type, input_tokens)
        if not suitable:
            return None
        return min(suitable,
                   key=lambda m: m.estimate_cost(input_tokens, output_tokens))

    def latency_optimized_select(self, task_type: TaskType,
                                  input_tokens: int) -> Optional[ModelConfig]:
        """延迟优化策略 - 选择P50延迟最低的模型"""
        suitable = self._get_suitable_models(task_type, input_tokens)
        if not suitable:
            return None
        return min(suitable, key=lambda m: self.latency_trackers[
            f"{m.provider.value}/{m.model_name}"].p50 or m.latency_p50)

    def quality_optimized_select(self, task_type: TaskType,
                                  input_tokens: int) -> Optional[ModelConfig]:
        """质量优化策略 - 选择能力最强的模型"""
        suitable = self._get_suitable_models(task_type, input_tokens)
        if not suitable:
            return None
        # 按优先级：旗舰全能力模型优先
        priority_order = [TaskType.CODE, TaskType.REASONING,
                          TaskType.CHAT, TaskType.SUMMARIZATION,
                          TaskType.EXTRACTION, TaskType.CLASSIFICATION]
        for priority_task in priority_order:
            for model in suitable:
                if priority_task in model.tasks:
                    return model
        return suitable[0]

    def hybrid_select(self, task_type: TaskType,
                       input_tokens: int,
                       user_tier: str = "standard") -> tuple[ModelConfig, str]:
        """
        混合路由策略 - 根据用户层级和任务类型动态选择

        策略说明：
        - premium用户：质量优先，故障转移到次优模型
        - standard用户：延迟优先，成本考虑
        - free用户：成本优先
        """
        strategy_used = ""

        if user_tier == "premium":
            model = self.quality_optimized_select(task_type, input_tokens)
            strategy_used = "quality"
        elif user_tier == "standard":
            model = self.latency_optimized_select(task_type, input_tokens)
            strategy_used = "latency"
        else:
            model = self.cost_optimized_select(task_type, input_tokens)
            strategy_used = "cost"

        return model, strategy_used

    def route_with_fallback(self, task_type: TaskType,
                             input_text: str,
                             user_tier: str = "standard") -> tuple[ModelConfig, RoutingDecision]:
        """
        智能路由 + 故障转移

        这是核心路由方法，包含：
        1. 语义缓存查询
        2. 智能模型选择
        3. 自动故障转移
        4. 决策记录
        """
        start_time = time.time()
        input_tokens = self._estimate_tokens(input_text)
        output_tokens = min(input_tokens, 4096)

        # 1. 检查缓存
        cached = self.cache.get(input_text[:100])
        if cached:
            # 缓存命中，使用最便宜的模型验证
            model = self.cost_optimized_select(task_type, input_tokens)
            decision = RoutingDecision(
                model=model,
                strategy="cache_hit",
                estimated_cost=0.0,
                decision_time_ms=(time.time() - start_time) * 1000,
                reason=f"缓存命中，跳过模型调用"
            )
            return model, decision

        # 2. 主路由选择
        model, strategy = self.hybrid_select(task_type, input_tokens, user_tier)
        if model is None:
            raise RuntimeError("没有可用模型")

        primary_decision = RoutingDecision(
            model=model,
            strategy=strategy,
            estimated_cost=model.estimate_cost(input_tokens, output_tokens),
            decision_time_ms=(time.time() - start_time) * 1000,
            reason=f"主路由策略={strategy}, 任务={task_type.value}, 层级={user_tier}"
        )

        # 3. 模拟调用（实际使用中会执行真正的API调用）
        latency_ms = self.latency_trackers[
            f"{model.provider.value}/{model.model_name}"].p50 or model.latency_p50

        success = latency_ms < model.latency_p95 * 1.5

        if not success:
            # 故障转移：尝试下一个可用模型
            fallback_model = self.latency_optimized_select(task_type, input_tokens)
            if fallback_model and fallback_model != model:
                decision = RoutingDecision(
                    model=fallback_model,
                    strategy=f"failover_from_{strategy}",
                    estimated_cost=fallback_model.estimate_cost(
                        input_tokens, output_tokens),
                    decision_time_ms=(time.time() - start_time) * 1000,
                    reason=f"主模型{model.model_name}超时，故障转移到{fallback_model.model_name}"
                )
                self.decisions.append(decision)
                return fallback_model, decision

        self.decisions.append(primary_decision)
        return model, primary_decision

    def record_latency(self, model_key: str, latency_ms: float):
        """记录模型延迟"""
        if model_key in self.latency_trackers:
            self.latency_trackers[model_key].record(latency_ms)

    def get_stats(self) -> dict:
        """获取路由统计信息"""
        total_cost = sum(d.estimated_cost for d in self.decisions)
        strategy_count = {}
        for d in self.decisions:
            strategy_count[d.strategy] = strategy_count.get(d.strategy, 0) + 1

        return {
            "total_decisions": len(self.decisions),
            "total_estimated_cost": round(total_cost, 4),
            "cache_hit_rate": round(self.cache.hit_rate, 3),
            "strategy_distribution": strategy_count,
            "model_latency_p50": {
                k: round(t.p50, 1)
                for k, t in self.latency_trackers.items()
                if t.p50 > 0
            }
        }


# === 使用示例 ===

def demo_cost_comparison():
    """演示不同任务下的成本对比"""
    print("=" * 60)
    print("📊 AI模型路由 - 成本对比演示")
    print("=" * 60)

    router = AIModelRouter()

    test_cases = [
        (TaskType.CLASSIFICATION, "将这段文本分类为正面或负面评价", 50),
        (TaskType.CODE, "用Python实现一个二叉树的层序遍历", 500),
        (TaskType.REASONING, "分析这篇20页论文的数学证明过程", 2000),
        (TaskType.SUMMARIZATION, "总结这篇5000字的文章要点", 1000),
    ]

    for task_type, text, output_tokens in test_cases:
        input_tokens = router._estimate_tokens(text)
        print(f"\n📌 任务类型: {task_type.value}")
        print(f"   输入: {input_tokens} tokens, 输出: {output_tokens} tokens")

        # 成本优化选择
        cost_model = router.cost_optimized_select(task_type, input_tokens, output_tokens)
        if cost_model:
            cost = cost_model.estimate_cost(input_tokens, output_tokens)
            print(f"   💰 成本优化: {cost_model.model_name} "
                  f"(${cost:.6f})")

        # 延迟优化选择
        latency_model = router.latency_optimized_select(task_type, input_tokens)
        if latency_model:
            print(f"   ⚡ 延迟优化: {latency_model.model_name} "
                  f"({latency_model.latency_p50}ms P50)")

        # 质量优化选择
        quality_model = router.quality_optimized_select(task_type, input_tokens)
        if quality_model:
            print(f"   🎯 质量优化: {quality_model.model_name}")


def demo_hybrid_routing():
    """演示混合路由策略"""
    print("\n" + "=" * 60)
    print("🔄 AI模型路由 - 混合策略演示")
    print("=" * 60)

    router = AIModelRouter()

    # 模拟不同用户层级的请求
    scenarios = [
        ("premium", TaskType.CODE, "请实现一个分布式锁"),
        ("premium", TaskType.CHAT, "今天天气怎么样？"),
        ("standard", TaskType.CLASSIFICATION, "请分类以下评论..."),
        ("standard", TaskType.SUMMARIZATION, "总结这篇文章..."),
        ("free", TaskType.CHAT, "你好"),
        ("free", TaskType.EXTRACTION, "从文本中提取日期和金额"),
    ]

    for tier, task, text in scenarios:
        try:
            model, decision = router.route_with_fallback(task, text, tier)
            print(f"\n👤 [{tier.upper()}] {task.value}")
            print(f"   模型: {model.provider.value}/{model.model_name}")
            print(f"   策略: {decision.strategy}")
            print(f"   预估成本: ${decision.estimated_cost:.6f}")
            print(f"   原因: {decision.reason}")
        except RuntimeError as e:
            print(f"   ❌ 错误: {e}")


def demo_cache_effectiveness():
    """演示语义缓存的成本节省效果"""
    print("\n" + "=" * 60)
    print("💾 AI模型路由 - 语义缓存效果演示")
    print("=" * 60)

    router = AIModelRouter()
    router.cache.set("什么是人工智能", "人工智能是...", 0.95)
    router.cache.set("AI是什么", "AI是人工智能的简称...", 0.92)

    queries = [
        "什么是人工智能",  # 精确命中
        "AI是什么",         # 精确命中
        "什么叫人工智能",   # 语义近似
        "解释机器学习",     # 未命中
    ]

    for q in queries:
        cached = router.cache.get(q)
        status = "✅ 缓存命中" if cached else "❌ 缓存未命中"
        print(f"   查询: \"{q}\" -> {status}")

    print(f"\n   总命中率: {router.cache.hit_rate * 100:.1f}%")
    print(f"   命中: {router.cache.hits}, 未命中: {router.cache.misses}")


def demo_apple_siri_router():
    """
    模拟苹果Siri AI的三层路由架构

    对应WWDC 2026公布的Siri架构：
    L1 - 设备端Neural Engine
    L2 - Apple Private Cloud Compute
    L3 - Google Cloud (NVIDIA B200)
    """
    print("\n" + "=" * 60)
    print("🍎 Apple Siri AI 三层路由架构模拟")
    print("=" * 60)

    @dataclass
    class SiriTier:
        name: str
        models: list[str]
        latency_range: tuple[float, float]
        cost_per_request: float

    tiers = [
        SiriTier("L1-设备端", ["Apple Neural Engine"],
                 (0.1, 50), 0.0),
        SiriTier("L2-Private Cloud", ["Apple Foundation Model"],
                 (50, 500), 0.0001),
        SiriTier("L3-Google Cloud", ["Gemini 1.2T", "Claude", "ChatGPT"],
                 (500, 3000), 0.005),
    ]

    siri_requests = [
        ("设置10分钟后闹钟", "简单", "L1-设备端"),
        ("从上周的邮件中找到酒店确认号", "中等", "L2-Private Cloud"),
        ("分析这份PDF并总结20页报告的核心观点", "复杂", "L3-Google Cloud"),
        ("帮我规划明天去纽约的行程，包括景点、餐厅和交通", "复杂", "L3-Google Cloud -> Claude"),
        ("搜索昨晚拍的照片中出现的宠物", "简单", "L1-设备端"),
    ]

    for query, complexity, expected_tier in siri_requests:
        print(f"\n   🗣️ \"{query}\"")
        print(f"   复杂度: {complexity}")
        print(f"   路由到: {expected_tier}")
        # 找出对应层级
        for tier in tiers:
            if tier.name.split("-")[0] in expected_tier:
                print(f"   延迟: {tier.latency_range[0]}-{tier.latency_range[1]}ms")
                print(f"   成本: ${tier.cost_per_request}")


if __name__ == "__main__":
    demo_cost_comparison()
    demo_hybrid_routing()
    demo_cache_effectiveness()
    demo_apple_siri_router()

    # 完整统计
    router = AIModelRouter()
    for _ in range(10):
        router.route_with_fallback(TaskType.CHAT, "hello", "free")
    print("\n" + "=" * 60)
    print("📈 完整路由统计")
    import json
    print(json.dumps(router.get_stats(), indent=2, ensure_ascii=False))