混合 Provider 策略：成本与质量的最优平衡

                    低复杂度                    高复杂度
               ┌─────────────────────┬─────────────────────┐
  高延迟要求   │  本地模型            │  Claude Sonnet      │
  (实时响应)   │  (Llama/Qwen)       │  (编码/分析)        │
               ├─────────────────────┼─────────────────────┤
  低延迟要求   │  Claude Haiku        │  GPT-4o             │
  (批量处理)   │  (分类/摘要)         │  (审查/关键决策)    │
               └─────────────────────┴─────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                      用户请求                                │
└────────────────────────┬────────────────────────────────────┘
                         ▼
┌─────────────────────────────────────────────────────────────┐
│                   Hermes Router                             │
│                                                             │
│  ┌───────────┐  ┌───────────┐  ┌───────────┐  ┌──────────┐ │
│  │ 任务分类器 │─▶│ 复杂度评估│─▶│ 策略选择器 │─▶│ Provider │ │
│  │ (本地)     │  │ (本地)    │  │ (规则引擎) │  │ 分发     │ │
│  └───────────┘  └───────────┘  └───────────┘  └──────────┘ │
│                                                             │
│  路由决策因素：                                              │
│  • 任务类型 (编码/分析/翻译/摘要/审查)                       │
│  • 复杂度评分 (1-10)                                        │
│  • 上下文长度                                               │
│  • 延迟要求                                                 │
│  • 预算约束                                                 │
│  • 质量要求                                                 │
└────────────────────────┬────────────────────────────────────┘
                         ▼
         ┌───────────────┼───────────────┐
         ▼               ▼               ▼
   ┌───────────┐  ┌───────────┐  ┌───────────┐
   │ 本地模型   │  │ Claude    │  │ OpenAI    │
   │ Llama 70B │  │ Sonnet 4  │  │ GPT-4o    │
   │ Qwen 32B  │  │ Haiku     │  │ o4-mini   │
   └───────────┘  └───────────┘  └───────────┘

# config/providers.yaml
providers:
  # 本地模型 - 低成本优先
  local-llama:
    type: "vllm"
    model: "meta-llama/Llama-3.1-70B-Instruct"
    endpoint: "http://localhost:8001/v1"
    cost_per_1m_input: 0.10
    cost_per_1m_output: 0.10
    max_context: 128000
    capabilities:
      - "chat"
      - "completion"
      - "embedding"
    priority: 1  # 优先使用（成本最低）

  local-qwen:
    type: "vllm"
    model: "Qwen/Qwen2.5-32B-Instruct"
    endpoint: "http://localhost:8002/v1"
    cost_per_1m_input: 0.05
    cost_per_1m_output: 0.05
    max_context: 128000
    capabilities:
      - "chat"
      - "completion"
      - "chinese"  # 中文优化
    priority: 2

  # Claude 系列 - 高质量
  claude-sonnet:
    type: "anthropic"
    model: "claude-sonnet-4-20250514"
    cost_per_1m_input: 3.00
    cost_per_1m_output: 15.00
    max_context: 200000
    capabilities:
      - "chat"
      - "coding"
      - "analysis"
      - "long_context"
    priority: 3

  claude-haiku:
    type: "anthropic"
    model: "claude-haiku-3.5"
    cost_per_1m_input: 0.25
    cost_per_1m_output: 1.25
    max_context: 200000
    capabilities:
      - "chat"
      - "classification"
      - "summarization"
    priority: 4

  # OpenAI 系列 - 审查验证
  gpt-4o:
    type: "openai"
    model: "gpt-4o"
    cost_per_1m_input: 2.50
    cost_per_1m_output: 10.00
    max_context: 128000
    capabilities:
      - "chat"
      - "review"
      - "multimodal"
    priority: 5

  o4-mini:
    type: "openai"
    model: "o4-mini"
    cost_per_1m_input: 1.10
    cost_per_1m_output: 4.40
    max_context: 200000
    capabilities:
      - "chat"
      - "reasoning"
    priority: 6

# src/router/provider_router.py
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from enum import Enum
import asyncio
import time

class TaskType(Enum):
    CODING = "coding"
    ANALYSIS = "analysis"
    TRANSLATION = "translation"
    SUMMARIZATION = "summarization"
    CLASSIFICATION = "classification"
    CHAT = "chat"
    REVIEW = "review"
    EXTRACTION = "extraction"

class ComplexityLevel(Enum):
    LOW = 1        # 简单问答、格式转换
    MEDIUM = 2     # 需要推理、多步操作
    HIGH = 3       # 复杂编码、深度分析

class QualityRequirement(Enum):
    LOW = "low"       # 允许一定误差
    MEDIUM = "medium" # 需要基本准确
    HIGH = "high"     # 必须精确，用于审查

@dataclass
class RouteRequest:
    content: str
    task_type: TaskType
    context_length: int
    quality_requirement: QualityRequirement
    latency_requirement: float  # 秒
    budget_limit: float  # 美元
    is_streaming: bool = False
    user_tier: str = "standard"  # free/standard/premium

@dataclass
class RouteDecision:
    provider: str
    model: str
    reason: str
    estimated_cost: float
    estimated_latency: float
    fallback_provider: Optional[str] = None

class ProviderRouter:
    """智能 Provider 路由决策引擎"""

    def __init__(self, config: Dict[str, Any]):
        self.providers = config["providers"]
        self.rules = config.get("routing_rules", [])
        self.cost_budget = config.get("monthly_budget", 1000)
        self.spent_this_month = 0
        self.classifier = self._load_classifier()

    def route(self, request: RouteRequest) -> RouteDecision:
        """为核心路由方法 - 决定使用哪个 Provider"""

        # 1. 评估任务复杂度
        complexity = self._assess_complexity(request)

        # 2. 获取候选 Provider 列表
        candidates = self._get_candidates(request, complexity)

        # 3. 评分排序
        scored = [(p, self._score_provider(p, request, complexity))
                  for p in candidates]
        scored.sort(key=lambda x: x[1], reverse=True)

        # 4. 选择最佳 Provider
        best_provider, best_score = scored[0]

        # 5. 构建决策
        decision = RouteDecision(
            provider=best_provider["name"],
            model=best_provider["model"],
            reason=self._explain_decision(best_provider, request, complexity),
            estimated_cost=self._estimate_cost(best_provider, request),
            estimated_latency=self._estimate_latency(best_provider, request),
            fallback_provider=scored[1][0]["name"] if len(scored) > 1 else None,
        )

        return decision

    def _assess_complexity(self, request: RouteRequest) -> ComplexityLevel:
        """评估任务复杂度"""
        score = 0

        # 基于任务类型
        type_complexity = {
            TaskType.CLASSIFICATION: 0,
            TaskType.SUMMARIZATION: 1,
            TaskType.TRANSLATION: 1,
            TaskType.CHAT: 1,
            TaskType.EXTRACTION: 1,
            TaskType.ANALYSIS: 2,
            TaskType.CODING: 2,
            TaskType.REVIEW: 2,
        }
        score += type_complexity.get(request.task_type, 1)

        # 基于上下文长度
        if request.context_length > 50000:
            score += 1
        if request.context_length > 100000:
            score += 1

        # 基于内容关键词（启发式）
        content = request.content.lower()
        complex_indicators = [
            "实现", "设计", "架构", "优化", "重构",
            "implement", "design", "architecture", "optimize",
            "分析原因", "为什么", "对比", "评估",
            "analyze", "why", "compare", "evaluate",
        ]
        complexity_count = sum(1 for kw in complex_indicators if kw in content)
        if complexity_count >= 3:
            score += 1

        if score <= 1:
            return ComplexityLevel.LOW
        elif score <= 2:
            return ComplexityLevel.MEDIUM
        else:
            return ComplexityLevel.HIGH

    def _get_candidates(self, request: RouteRequest,
                       complexity: ComplexityLevel) -> List[Dict]:
        """根据请求获取候选 Provider"""
        candidates = []

        for name, config in self.providers.items():
            # 预算过滤
            estimated = self._estimate_cost(config, request)
            if self.spent_this_month + estimated > self.cost_budget:
                continue

            # 能力匹配
            if not self._check_capability(config, request):
                continue

            # 上下文长度匹配
            if request.context_length > config["max_context"]:
                continue

            # 质量要求匹配
            if request.quality_requirement == QualityRequirement.HIGH:
                # 高质量要求只考虑顶级模型
                if name not in ["gpt-4o", "claude-sonnet"]:
                    continue

            # 复杂度匹配
            if complexity == ComplexityLevel.LOW:
                # 低复杂度优先本地模型
                if name.startswith("local-"):
                    candidates.append({"name": name, **config})
            elif complexity == ComplexityLevel.MEDIUM:
                # 中等复杂度考虑 Claude Haiku 或本地
                if name.startswith("local-") or name == "claude-haiku":
                    candidates.append({"name": name, **config})
            else:
                # 高复杂度考虑 Sonnet/GPT-4o
                if name in ["claude-sonnet", "gpt-4o", "o4-mini"]:
                    candidates.append({"name": name, **config})

        # 如果没有候选者，回退到默认
        if not candidates:
            candidates = [{"name": "local-llama", **self.providers["local-llama"]}]

        return candidates

    def _score_provider(self, provider: Dict, request: RouteRequest,
                       complexity: ComplexityLevel) -> float:
        """为 Provider 评分"""
        score = 0.0

        # 成本分（越低越好）- 权重 30%
        cost = self._estimate_cost(provider, request)
        max_cost = 10.0  # 最高成本上限
        cost_score = max(0, 1 - cost / max_cost)
        score += cost_score * 0.30

        # 质量分 - 权重 40%
        quality_scores = {
            "local-llama": 0.65,
            "local-qwen": 0.60,
            "claude-haiku": 0.75,
            "o4-mini": 0.85,
            "claude-sonnet": 0.92,
            "gpt-4o": 0.95,
        }
        quality_score = quality_scores.get(provider["name"], 0.5)
        score += quality_score * 0.40

        # 延迟分 - 权重 20%
        latency = self._estimate_latency(provider, request)
        max_latency = 30.0
        latency_score = max(0, 1 - latency / max_latency)
        score += latency_score * 0.20

        # 特殊能力加分 - 权重 10%
        if request.task_type == TaskType.CODING and "coding" in provider.get("capabilities", []):
            score += 0.10
        if request.task_type == TaskType.REVIEW and "review" in provider.get("capabilities", []):
            score += 0.10
        if "chinese" in provider.get("capabilities", []) and self._is_chinese(request.content):
            score += 0.05

        return score

    def _check_capability(self, provider: Dict, request: RouteRequest) -> bool:
        """检查 Provider 是否满足请求能力要求"""
        capabilities = provider.get("capabilities", [])

        task_capability_map = {
            TaskType.CODING: "coding",
            TaskType.ANALYSIS: "analysis",
            TaskType.REVIEW: "review",
            TaskType.CLASSIFICATION: "classification",
        }

        required = task_capability_map.get(request.task_type)
        if required:
            return required in capabilities

        return True  # 通用任务不需要特殊能力

    def _is_chinese(self, text: str) -> bool:
        """检测是否为中文内容"""
        chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        return chinese_chars > len(text) * 0.3

# src/router/executor.py
import asyncio
import httpx
from typing import Dict, Any, Optional

class RequestExecutor:
    """请求执行器 - 实际调用选定的 Provider"""

    def __init__(self, providers_config: Dict[str, Any]):
        self.providers = providers_config
        self.clients = {}
        self.stats = {
            "total_requests": 0,
            "provider_usage": {},
            "total_cost": 0,
            "errors": 0,
        }

    async def execute(self, request: Dict[str, Any],
                     decision: RouteDecision) -> Dict[str, Any]:
        """执行请求到选定的 Provider"""
        provider_name = decision.provider

        try:
            # 记录开始
            start_time = time.time()

            # 调用 Provider
            result = await self._call_provider(
                provider_name, request
            )

            elapsed = time.time() - start_time

            # 计算成本
            cost = self._calculate_cost(provider_name, result)

            # 更新统计
            self._update_stats(provider_name, elapsed, cost)

            result["metadata"] = {
                "provider": provider_name,
                "model": decision.model,
                "cost": cost,
                "latency": elapsed,
            }

            return result

        except Exception as e:
            self.stats["errors"] += 1

            # 如果有 fallback，尝试 fallback
            if decision.fallback_provider:
                print(f"⚠️ {provider_name} 失败，回退到 {decision.fallback_provider}")
                return await self._call_fallback(
                    decision.fallback_provider, request
                )

            raise

    async def _call_provider(self, provider_name: str,
                            request: Dict) -> Dict:
        """调用具体 Provider"""
        config = self.providers[provider_name]

        if provider_name.startswith("local-"):
            return await self._call_vllm(config, request)
        elif config["type"] == "anthropic":
            return await self._call_anthropic(config, request)
        elif config["type"] == "openai":
            return await self._call_openai(config, request)
        else:
            raise ValueError(f"不支持的 Provider 类型: {config['type']}")

    async def _call_vllm(self, config: Dict, request: Dict) -> Dict:
        """调用本地 vLLM 服务"""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{config['endpoint']}/chat/completions",
                json={
                    "model": config["model"],
                    "messages": request["messages"],
                    "max_tokens": request.get("max_tokens", 4096),
                    "temperature": request.get("temperature", 0.7),
                },
                timeout=60,
            )
            response.raise_for_status()
            return response.json()

    async def _call_anthropic(self, config: Dict, request: Dict) -> Dict:
        """调用 Claude API"""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://api.anthropic.com/v1/messages",
                headers={
                    "x-api-key": config["api_key"],
                    "anthropic-version": "2024-02-15",
                    "content-type": "application/json",
                },
                json={
                    "model": config["model"],
                    "messages": request["messages"],
                    "max_tokens": request.get("max_tokens", 8192),
                    "temperature": request.get("temperature", 0.7),
                },
                timeout=120,
            )
            response.raise_for_status()
            return response.json()

    async def _call_openai(self, config: Dict, request: Dict) -> Dict:
        """调用 OpenAI API"""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://api.openai.com/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {config['api_key']}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": config["model"],
                    "messages": request["messages"],
                    "max_tokens": request.get("max_tokens", 4096),
                    "temperature": request.get("temperature", 0.7),
                },
                timeout=120,
            )
            response.raise_for_status()
            return response.json()

# src/review/dual_review.py
import asyncio
from typing import Dict, Any

class DualReviewSystem:
    """双重审查系统 - 低成本生成 + 高质量验证"""

    def __init__(self, router: ProviderRouter, executor: RequestExecutor):
        self.router = router
        self.executor = executor
        self.review_threshold = 0.8  # 验证通过阈值

    async def process_with_review(self, request: Dict) -> Dict:
        """处理请求并审查"""
        # 阶段 1: 使用低成本模型生成
        generate_request = RouteRequest(
            content=request["content"],
            task_type=request["task_type"],
            context_length=request.get("context_length", 0),
            quality_requirement=QualityRequirement.MEDIUM,
            latency_requirement=request.get("latency_requirement", 10),
            budget_limit=request.get("budget_limit", 1.0),
        )

        gen_decision = self.router.route(generate_request)
        gen_result = await self.executor.execute(request, gen_decision)

        # 阶段 2: 使用高质量模型审查
        review_prompt = self._build_review_prompt(
            request["content"], gen_result
        )

        review_request = RouteRequest(
            content=review_prompt,
            task_type=TaskType.REVIEW,
            context_length=len(review_prompt),
            quality_requirement=QualityRequirement.HIGH,
            latency_requirement=15,
            budget_limit=2.0,
        )

        review_decision = self.router.route(review_request)
        review_result = await self.executor.execute(
            {"messages": [{"role": "user", "content": review_prompt}]},
            review_decision,
        )

        # 阶段 3: 解析审查结果
        review_score = self._parse_review_score(review_result)

        if review_score >= self.review_threshold:
            # 审查通过
            return {
                "output": gen_result["output"],
                "review_status": "passed",
                "review_score": review_score,
                "total_cost": (
                    gen_result["metadata"]["cost"] +
                    review_result["metadata"]["cost"]
                ),
            }
        else:
            # 审查不通过，重新生成
            regenerate_result = await self._regenerate(
                request, gen_result, review_result
            )
            return {
                "output": regenerate_result["output"],
                "review_status": "regenerated",
                "review_score": review_score,
                "total_cost": regenerate_result["metadata"]["cost"],
            }

    def _build_review_prompt(self, original: str, result: Dict) -> str:
        """构建审查提示词"""
        return f"""请审查以下任务的完成质量。

原始任务：
{original}

生成结果：
{result.get("output", "")}

请从以下维度评分（0-1）：
1. 准确性：结果是否正确
2. 完整性：是否覆盖所有要求
3. 代码质量（如适用）：是否规范、高效
4. 安全性：是否存在安全隐患

请以 JSON 格式返回：
{{
    "accuracy": 0.0-1.0,
    "completeness": 0.0-1.0,
    "code_quality": 0.0-1.0,
    "safety": 0.0-1.0,
    "overall": 0.0-1.0,
    "feedback": "具体改进建议"
}}

开始审查。"""

# src/strategies.py
from enum import Enum

class ProcessingStrategy(Enum):
    DIRECT = "direct"         # 直接处理，不审查
    REVIEW = "review"         # 生成后审查
    DUAL_GENERATION = "dual"  # 双模型生成 + 对比

class TaskImportance(Enum):
    LOW = "low"       # 日常对话、简单查询
    MEDIUM = "medium" # 代码建议、分析报告
    HIGH = "high"     # 生产代码、安全审计
    CRITICAL = "critical"  # 金融交易、医疗建议

class StrategySelector:
    """策略选择器"""

    STRATEGY_MAP = {
        (TaskImportance.LOW, ComplexityLevel.LOW): ProcessingStrategy.DIRECT,
        (TaskImportance.LOW, ComplexityLevel.MEDIUM): ProcessingStrategy.DIRECT,
        (TaskImportance.MEDIUM, ComplexityLevel.LOW): ProcessingStrategy.DIRECT,
        (TaskImportance.MEDIUM, ComplexityLevel.MEDIUM): ProcessingStrategy.REVIEW,
        (TaskImportance.HIGH, ComplexityLevel.LOW): ProcessingStrategy.REVIEW,
        (TaskImportance.HIGH, ComplexityLevel.MEDIUM): ProcessingStrategy.REVIEW,
        (TaskImportance.HIGH, ComplexityLevel.HIGH): ProcessingStrategy.DUAL_GENERATION,
        (TaskImportance.CRITICAL, ComplexityLevel.HIGH): ProcessingStrategy.DUAL_GENERATION,
    }

    @classmethod
    def select(cls, importance: TaskImportance,
               complexity: ComplexityLevel) -> ProcessingStrategy:
        return cls.STRATEGY_MAP.get(
            (importance, complexity),
            ProcessingStrategy.REVIEW,  # 默认审查
        )

📊 Provider 使用统计 (最近 7 天)

┌──────────────────┬──────────┬──────────┬──────────┬──────────┐
│ Provider         │ 请求数   │ 占比     │ 总成本   │ 平均延迟  │
├──────────────────┼──────────┼──────────┼──────────┼──────────┤
│ local-llama      │ 45,230   │ 64.6%    │ $45.23   │ 0.8s     │
│ local-qwen       │ 8,450    │ 12.1%    │ $8.45    │ 0.6s     │
│ claude-haiku     │ 5,670    │ 8.1%     │ $28.35   │ 1.2s     │
│ claude-sonnet    │ 6,210    │ 8.9%     │ $186.30  │ 3.5s     │
│ gpt-4o           │ 3,890    │ 5.5%     │ $97.25   │ 2.8s     │
│ o4-mini          │ 560      │ 0.8%     │ $12.32   │ 4.2s     │
├──────────────────┼──────────┼──────────┼──────────┼──────────┤
│ 合计             │ 70,010   │ 100%     │ $377.90  │ 1.4s     │
└──────────────────┴──────────┴──────────┴──────────┴──────────┘

💰 成本对比：
   混合策略:    $377.90 / 周
   全部 Sonnet: $2,100.30 / 周 (节省 82%)
   全部 GPT-4o: $1,750.25 / 周 (节省 78%)

# src/quality_monitor.py
class QualityMonitor:
    """质量监控 - 跟踪各 Provider 的输出质量"""

    def __init__(self):
        self.quality_scores = {
            "local-llama": [],
            "local-qwen": [],
            "claude-haiku": [],
            "claude-sonnet": [],
            "gpt-4o": [],
        }

    def record_quality(self, provider: str, score: float):
        """记录单次质量评分"""
        if provider in self.quality_scores:
            self.quality_scores[provider].append(score)
            # 保留最近 1000 次记录
            if len(self.quality_scores[provider]) > 1000:
                self.quality_scores[provider] = \
                    self.quality_scores[provider][-1000:]

    def get_average_quality(self, provider: str) -> float:
        """获取 Provider 的平均质量分数"""
        scores = self.quality_scores.get(provider, [])
        if not scores:
            return 0.0
        return sum(scores) / len(scores)

    def generate_report(self) -> Dict:
        """生成质量报告"""
        return {
            provider: {
                "avg_quality": self.get_average_quality(provider),
                "sample_count": len(scores),
            }
            for provider, scores in self.quality_scores.items()
        }

# src/router/auto_tuner.py
import json
from pathlib import Path

class AutoTuner:
    """基于反馈的自动路由调优"""

    def __init__(self, config_path: str):
        self.config_path = Path(config_path)
        self.feedback_log = Path("logs/routing_feedback.jsonl")
        self.feedback_data = self._load_feedback()

    def record_feedback(self, request_id: str, provider: str,
                       user_rating: int, auto_quality: float):
        """记录路由反馈"""
        feedback = {
            "request_id": request_id,
            "provider": provider,
            "user_rating": user_rating,  # 1-5
            "auto_quality": auto_quality,
            "timestamp": time.time(),
        }

        with open(self.feedback_log, "a") as f:
            f.write(json.dumps(feedback) + "\n")

        self.feedback_data.append(feedback)

    def analyze_and_update(self) -> Dict:
        """分析反馈并更新路由权重"""
        # 按 Provider 统计满意度
        provider_satisfaction = {}
        for fb in self.feedback_data:
            provider = fb["provider"]
            if provider not in provider_satisfaction:
                provider_satisfaction[provider] = []
            provider_satisfaction[provider].append(
                (fb["user_rating"] + fb["auto_quality"] * 5) / 2
            )

        # 计算平均满意度
        satisfaction_scores = {
            p: sum(s) / len(s) for p, s in provider_satisfaction.items()
        }

        # 更新路由配置
        self._update_routing_weights(satisfaction_scores)

        return satisfaction_scores

    def _update_routing_weights(self, satisfaction: Dict):
        """根据满意度更新路由权重"""
        # 提高高满意度 Provider 的权重
        # 降低低满意度 Provider 的权重
        pass

# src/router/budget_manager.py
class BudgetManager:
    """预算自适应管理"""

    def __init__(self, monthly_budget: float):
        self.monthly_budget = monthly_budget
        self.daily_budget = monthly_budget / 30
        self.spent_today = 0
        self.spent_this_month = 0

    def adjust_strategy(self) -> Dict:
        """根据预算使用情况调整策略"""
        remaining_daily = self.daily_budget - self.spent_today
        remaining_monthly = self.monthly_budget - self.spent_this_month
        days_remaining = 30 - (datetime.now().day - 1)

        # 动态调整每日预算
        adjusted_daily = remaining_monthly / max(days_remaining, 1)

        if adjusted_daily < self.daily_budget * 0.5:
            # 预算紧张，增加本地模型权重
            return {
                "mode": "budget_tight",
                "local_weight": 0.8,
                "cloud_weight": 0.2,
                "review_enabled": False,
            }
        elif adjusted_daily > self.daily_budget * 1.5:
            # 预算充裕，增加高质量模型使用
            return {
                "mode": "budget_relaxed",
                "local_weight": 0.4,
                "cloud_weight": 0.6,
                "review_enabled": True,
            }
        else:
            # 正常模式
            return {
                "mode": "normal",
                "local_weight": 0.6,
                "cloud_weight": 0.4,
                "review_enabled": True,
            }