能做事的 Agent，需要一个推荐系统

                    ┌─────────────────────────────────────────┐
                    │          TMS 三维匹配模型                 │
                    └─────────────────────────────────────────┘

    ┌──────────────┐       ┌──────────────┐       ┌──────────────┐
    │   Task 维度   │       │   Model 维度  │       │  Sandbox 维度 │
    │              │       │              │       │              │
    │  · 复杂度     │       │  · 能力等级   │       │  · 执行能力   │
    │  · 领域标签   │       │  · 推理成本   │       │  · 隔离级别   │
    │  · 工具需求   │       │  · 响应延迟   │       │  · 资源开销   │
    │  · 延迟敏感度 │       │  · 上下文窗口 │       │  · 安全边界   │
    │  · 成本敏感度 │       │  · 专项能力   │       │  · 状态保持   │
    └──────┬───────┘       └──────┬───────┘       └──────┬───────┘
           │                      │                      │
           │         ┌────────────┴────────────┐         │
           │         │    匹配引擎（推荐系统）    │         │
           │         │                         │         │
           └────────>│  Task → (Model, Sandbox) │<────────┘
                     │                         │
                     │  目标函数：               │
                     │  max Quality - λ·Cost    │
                     │  s.t. Latency < SLA      │
                     └─────────────────────────┘

    ┌─────────────────────────────────────────────────────────────┐
    │                  Agent 推荐系统架构                          │
    └─────────────────────────────────────────────────────────────┘

    用户任务
       │
       ▼
    ┌─────────────────┐
    │  Task Encoder    │  自然语言 → 特征向量
    │  (LLM 分类器)    │  复杂度/领域/工具需求/敏感度
    └────────┬────────┘
             │ task_vector
             ▼
    ┌─────────────────┐     ┌──────────────────────────────┐
    │  Candidate       │     │  Model Registry              │
    │  Generator       │<────│  Sandbox Registry            │
    │  (规则 + 向量召回)│     │  (可用资源池)                 │
    └────────┬────────┘     └──────────────────────────────┘
             │ top-k candidates [(model_i, sandbox_j), ...]
             ▼
    ┌─────────────────┐
    │  Ranker          │  对每个 candidate 打分
    │  (轻量模型/规则)  │  score = f(quality, cost, latency)
    └────────┬────────┘
             │ best (model, sandbox) pair
             ▼
    ┌─────────────────┐
    │  Executor        │  在选定沙箱中用选定模型执行任务
    │  (dispatch)      │
    └────────┬────────┘
             │ execution result
             ▼
    ┌─────────────────┐
    │  Feedback Loop   │  记录执行质量 → 更新模型/沙箱画像
    │  (在线学习)       │
    └─────────────────┘

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import json
import time


class IsolationLevel(Enum):
    NONE = "none"
    PROCESS = "process"
    CONTAINER = "container"
    VM = "vm"


@dataclass
class TaskProfile:
    """任务画像 — 由 Task Encoder (LLM) 从用户输入生成"""
    complexity: int  # 1-5
    domains: list[str]
    tools_required: list[str]
    latency_sensitivity: int  # 1-5, 5=实时
    cost_sensitivity: int  # 1-5, 5=高敏感
    context_size: int  # 预估 token 数
    raw_query: str  # 原始用户输入

    # generated by hugo AI


@dataclass
class ModelProfile:
    """模型画像 — 注册时提供，运行时通过反馈更新"""
    name: str
    capability_score: float  # 1-5
    cost_per_1k_input: float
    cost_per_1k_output: float
    avg_latency_ms: int
    max_context: int
    specialties: list[str]
    tool_calling_level: int  # 0-3
    success_rate: float = 0.0  # 在线反馈更新
    total_executions: int = 0

    # generated by hugo AI


@dataclass
class SandboxProfile:
    """沙箱画像"""
    name: str
    capabilities: list[str]
    isolation: IsolationLevel
    startup_time_ms: int
    max_duration_s: int
    stateful: bool
    security_risk: str  # low/medium/high

    # generated by hugo AI


@dataclass
class Candidate:
    """一个候选的 (模型, 沙箱) 组合"""
    model: ModelProfile
    sandbox: SandboxProfile
    score: float = 0.0
    breakdown: dict = field(default_factory=dict)

    # generated by hugo AI


class TMSEncoder:
    """
    Task Encoder — 将自然语言任务转化为结构化画像。
    
    实际生产环境中，这里用一个轻量 LLM（如 7B-14B）
    做结构化抽取，而不是用最强的推理模型。
    原因：编码本身只需要理解意图，不需要推理，
    用强模型是浪费。
    """

    SYSTEM_PROMPT = """你是一个任务分析器。将用户的自然语言请求转化为结构化任务画像。
严格输出 JSON 格式，不要输出其他内容。

字段说明：
- complexity: 1-5，1=简单问答，5=多步复杂任务
- domains: 领域标签列表，从 [coding, web, data-analysis, writing, 
  reasoning, file-operation, system-admin, creative] 中选择
- tools_required: 需要的工具列表，从 [browser, terminal, code-exec, 
  file, search, api-call] 中选择
- latency_sensitivity: 1-5，5=需要实时响应
- cost_sensitivity: 1-5，5=对成本极度敏感
- context_size: 预估需要的上下文 token 数"""

    def encode(self, query: str) -> TaskProfile:
        """
        实际实现中，这里调用一个轻量 LLM 做结构化抽取。
        为了演示，用规则模拟。
        """
        # 生产环境：response = llm_call(self.SYSTEM_PROMPT, query)
        # 演示：基于关键词的简单规则
        complexity = self._estimate_complexity(query)
        domains = self._detect_domains(query)
        tools = self._detect_tools(query)
        latency_s = self._estimate_latency_sensitivity(query)
        cost_s = 3  # 默认中等敏感度
        context = self._estimate_context(query)

        return TaskProfile(
            complexity=complexity,
            domains=domains,
            tools_required=tools,
            latency_sensitivity=latency_s,
            cost_sensitivity=cost_s,
            context_size=context,
            raw_query=query,
        )

    def _estimate_complexity(self, query: str) -> int:
        keywords_complex = ["分析", "对比", "实现", "构建", "优化", "debug", "调试"]
        keywords_simple = ["是什么", "总结", "翻译", "解释", "帮我写"]
        
        if any(k in query for k in keywords_complex):
            return 4
        elif any(k in query for k in keywords_simple):
            return 2
        return 3

    def _detect_domains(self, query: str) -> list[str]:
        domain_map = {
            "coding": ["代码", "编程", "函数", "bug", "debug", "实现", "API"],
            "web": ["网页", "网站", "URL", "爬虫", "浏览器"],
            "data-analysis": ["数据", "分析", "统计", "图表", "报告"],
            "writing": ["写", "文章", "邮件", "文案", "总结"],
        }
        return [
            domain for domain, keywords in domain_map.items()
            if any(k in query for k in keywords)
        ] or ["general"]

    def _detect_tools(self, query: str) -> list[str]:
        tool_map = {
            "browser": ["网页", "网站", "URL", "打开网页", "浏览"],
            "terminal": ["命令", "终端", "部署", "安装", "git"],
            "code-exec": ["代码", "运行", "执行", "函数", "脚本"],
            "file": ["文件", "读取", "保存", "文档"],
            "search": ["搜索", "查找", "查一下"],
        }
        return [
            tool for tool, keywords in tool_map.items()
            if any(k in query for k in keywords)
        ]

    def _estimate_latency_sensitivity(self, query: str) -> int:
        urgent_keywords = ["快", "马上", "立刻", "急", "实时"]
        if any(k in query for k in urgent_keywords):
            return 5
        return 3

    def _estimate_context(self, query: str) -> int:
        if len(query) > 500:
            return 32000
        elif len(query) > 100:
            return 8000
        return 4000

    # generated by hugo AI


class TMSRecommender:
    """
    TMS 推荐系统 — 核心匹配引擎。
    
    两阶段架构：
    1. Candidate Generator: 硬约束过滤 + 向量召回
    2. Ranker: 多目标排序（质量、成本、延迟）
    """

    def __init__(
        self,
        models: list[ModelProfile],
        sandboxes: list[SandboxProfile],
        quality_weight: float = 0.5,
        cost_weight: float = 0.3,
        latency_weight: float = 0.2,
    ):
        self.models = models
        self.sandboxes = sandboxes
        self.encoder = TMSEncoder()
        
        # 排序权重 — 可根据业务场景调整
        self.w_quality = quality_weight
        self.w_cost = cost_weight
        self.w_latency = latency_weight

    def recommend(self, query: str) -> Candidate:
        """端到端推荐：输入自然语言，输出最优 (model, sandbox) 组合"""
        task = self.encoder.encode(query)
        candidates = self._generate_candidates(task)
        ranked = self._rank(task, candidates)
        
        if not ranked:
            # fallback: 用最强模型 + 最通用沙箱
            return self._fallback(task)
        
        return ranked[0]

    def _generate_candidates(self, task: TaskProfile) -> list[Candidate]:
        """
        阶段1：候选生成
        
        先用硬约束过滤（如工具需求、上下文窗口），
        再做笛卡尔积生成候选组合。
        """
        # 过滤满足工具需求的沙箱
        valid_sandboxes = [
            s for s in self.sandboxes
            if all(t in s.capabilities for t in task.tools_required)
        ]
        
        # 如果没有沙箱满足所有工具需求，放宽到"至少满足一个"
        if not valid_sandboxes:
            valid_sandboxes = [
                s for s in self.sandboxes
                if any(t in s.capabilities for t in task.tools_required)
            ] or self.sandboxes  # 最终 fallback: 全部沙箱

        # 过滤满足上下文需求的模型
        valid_models = [
            m for m in self.models
            if m.max_context >= task.context_size
        ] or self.models  # fallback: 全部模型

        # 笛卡尔积
        candidates = []
        for model in valid_models:
            for sandbox in valid_sandboxes:
                candidates.append(Candidate(model=model, sandbox=sandbox))

        return candidates

    def _rank(self, task: TaskProfile, candidates: list[Candidate]) -> list[Candidate]:
        """
        阶段2：多目标排序
        
        对每个候选计算综合得分：
        score = w_q * quality - w_c * cost - w_l * latency_penalty
        
        关键设计：权重不是固定的，
        而是根据任务的敏感度动态调整。
        """
        # 根据任务特征动态调整权重
        w_q = self.w_quality
        w_c = self.w_cost * (task.cost_sensitivity / 3.0)
        w_l = self.w_latency * (task.latency_sensitivity / 3.0)

        for cand in candidates:
            # 质量分：模型能力与任务复杂度的匹配度
            quality = self._score_quality(task, cand.model)
            
            # 成本分：归一化的预估成本（越低越好）
            cost = self._score_cost(task, cand.model)
            
            # 延迟分：归一化的预估延迟（越低越好）
            latency = self._score_latency(task, cand.model, cand.sandbox)
            
            # 综合得分
            cand.score = w_q * quality - w_c * cost - w_l * latency
            cand.breakdown = {
                "quality": round(quality, 3),
                "cost": round(cost, 3),
                "latency": round(latency, 3),
                "weights": {
                    "quality": round(w_q, 2),
                    "cost": round(w_c, 2),
                    "latency": round(w_l, 2),
                },
            }

        # 降序排列
        candidates.sort(key=lambda c: c.score, reverse=True)
        return candidates

    def _score_quality(self, task: TaskProfile, model: ModelProfile) -> float:
        """
        质量评分 — 模型能力与任务需求的匹配度。
        
        不是"能力越强分数越高"，而是"能力刚好匹配任务复杂度时分数最高"。
        过度配置（用推理模型做简单任务）会有轻微惩罚。
        """
        capability_gap = model.capability_score - task.complexity
        
        if capability_gap >= 0:
            # 能力足够：轻微惩罚过度配置
            return 1.0 - 0.05 * capability_gap
        else:
            # 能力不足：严重惩罚
            return 1.0 - 0.3 * abs(capability_gap)

    def _score_cost(self, task: TaskProfile, model: ModelProfile) -> float:
        """成本评分 — 归一化到 0-1，越低表示成本越高"""
        # 预估 token 用量（输入 + 输出）
        estimated_tokens = task.context_size + 1000  # 简单估算
        cost = estimated_tokens / 1000 * (
            model.cost_per_1k_input + model.cost_per_1k_output
        )
        # 归一化：假设最大成本为 $1.0
        return min(cost / 1.0, 1.0)

    def _score_latency(self, task: TaskProfile, model: ModelProfile, 
                       sandbox: SandboxProfile) -> float:
        """延迟评分 — 归一化到 0-1，越低表示延迟越高"""
        total_ms = model.avg_latency_ms + sandbox.startup_time_ms
        # 归一化：假设最大可接受延迟为 30s
        return min(total_ms / 30000, 1.0)

    def _fallback(self, task: TaskProfile) -> Candidate:
        """Fallback: 用能力最强的模型和最通用的沙箱"""
        best_model = max(self.models, key=lambda m: m.capability_score)
        best_sandbox = max(self.sandboxes, key=lambda s: len(s.capabilities))
        return Candidate(model=best_model, sandbox=best_sandbox)

    # generated by hugo AI


class FeedbackCollector:
    """
    反馈收集器 — 推荐系统的闭环关键。
    
    每次任务执行完成后，收集实际执行质量数据，
    用于更新模型画像和排序策略。
    """

    @dataclass
    class ExecutionRecord:
        task: TaskProfile
        model_used: str
        sandbox_used: str
        success: bool
        actual_latency_ms: int
        actual_cost: float
        user_satisfaction: Optional[int]  # 1-5, None=未反馈

        # generated by hugo AI

    def __init__(self):
        self.records: list[self.ExecutionRecord] = []

    def record(self, record: ExecutionRecord) -> None:
        self.records.append(record)

    def get_model_stats(self, model_name: str) -> dict:
        """计算模型在线统计数据"""
        model_records = [
            r for r in self.records if r.model_used == model_name
        ]
        if not model_records:
            return {"success_rate": 0.0, "count": 0}
        
        successes = sum(1 for r in model_records if r.success)
        return {
            "success_rate": successes / len(model_records),
            "total_executions": len(model_records),
            "avg_latency_ms": (
                sum(r.actual_latency_ms for r in model_records) 
                / len(model_records)
            ),
            "avg_cost": (
                sum(r.actual_cost for r in model_records) 
                / len(model_records)
            ),
        }

    # generated by hugo AI

# 注册可用模型
models = [
    ModelProfile(
        name="gpt-4o-mini",
        capability_score=3.0,
        cost_per_1k_input=0.00015,
        cost_per_1k_output=0.0006,
        avg_latency_ms=800,
        max_context=128000,
        specialties=["general", "coding"],
        tool_calling_level=2,
    ),
    ModelProfile(
        name="claude-sonnet-4",
        capability_score=4.5,
        cost_per_1k_input=0.003,
        cost_per_1k_output=0.015,
        avg_latency_ms=2000,
        max_context=200000,
        specialties=["reasoning", "coding", "writing"],
        tool_calling_level=3,
    ),
    ModelProfile(
        name="o3-pro",
        capability_score=5.0,
        cost_per_1k_input=0.01,
        cost_per_1k_output=0.04,
        avg_latency_ms=15000,
        max_context=200000,
        specialties=["reasoning", "math", "coding"],
        tool_calling_level=3,
    ),
]

# 注册可用沙箱
sandboxes = [
    SandboxProfile(
        name="text-chat",
        capabilities=[],  # 纯对话，不需要工具
        isolation=IsolationLevel.NONE,
        startup_time_ms=0,
        max_duration_s=30,
        stateful=True,
        security_risk="low",
    ),
    SandboxProfile(
        name="code-exec",
        capabilities=["code-exec", "file"],
        isolation=IsolationLevel.CONTAINER,
        startup_time_ms=2000,
        max_duration_s=300,
        stateful=True,
        security_risk="medium",
    ),
    SandboxProfile(
        name="browser",
        capabilities=["browser", "search"],
        isolation=IsolationLevel.PROCESS,
        startup_time_ms=1500,
        max_duration_s=120,
        stateful=True,
        security_risk="medium",
    ),
]

recommender = TMSRecommender(models, sandboxes)

# 场景1：简单问答
result = recommender.recommend("Python 的 GIL 是什么？")
# → 匹配: gpt-4o-mini + text-chat
#   原因: 低复杂度任务，不需要工具，轻量模型足够

# 场景2：需要浏览器的任务
result = recommender.recommend("帮我打开这个网页并总结内容：https://example.com")
# → 匹配: gpt-4o-mini + browser
#   原因: 中等复杂度，需要浏览器工具，不需要强推理

# 场景3：复杂编码任务
result = recommender.recommend("帮我实现一个分布式任务调度器，支持失败重试和优先级队列")
# → 匹配: claude-sonnet-4 + code-exec
#   原因: 高复杂度，需要强编码能力和代码执行环境

# 场景4：高延迟敏感 + 低成本敏感
result = recommender.recommend("快！帮我分析这段代码的 bug：[大段代码]")
# → 匹配: claude-sonnet-4 + text-chat
#   原因: 高延迟敏感（"快！"），需要推理能力，
#         但不需要执行代码（纯分析）

# generated by hugo AI

from dataclasses import dataclass


@dataclass
class RankingModel:
    """
    学习型排序器 — 用历史反馈数据训练。
    
    实际生产中可以用 LightGBM / XGBoost 
    或简单的线性回归。
    不需要深度学习——特征维度低（< 20），
    样本量足够时传统 ML 效果更好且更快。
    """
    weights: dict[str, float] = None

    def train(self, records: list[FeedbackCollector.ExecutionRecord]) -> None:
        """
        从历史执行记录中学习最优权重。
        
        目标：找到一组权重，使得历史上成功的执行
        得分高于失败的执行。
        
        实际实现中使用 pairwise ranking loss：
        L = Σ max(0, 1 - score(positive) + score(negative))
        """
        # 简化演示：用成功记录的平均特征作为权重
        successful = [r for r in records if r.success]
        if not successful:
            return

        # 实际实现中这里用梯度下降优化
        # 演示：用统计方法估算权重
        self.weights = {
            "quality": 0.5,
            "cost": 0.3,
            "latency": 0.2,
        }

    def predict_score(
        self, 
        task: TaskProfile, 
        model: ModelProfile, 
        sandbox: SandboxProfile
    ) -> float:
        """用学习到的权重计算得分"""
        if not self.weights:
            return 0.0
        
        # 特征工程：将 (task, model, sandbox) 转化为特征向量
        features = {
            "capability_match": model.capability_score - task.complexity,
            "cost_normalized": model.cost_per_1k_input * task.context_size / 1000,
            "latency_total": model.avg_latency_ms + sandbox.startup_time_ms,
            "domain_overlap": len(
                set(task.domains) & set(model.specialties)
            ) / max(len(task.domains), 1),
            "tool_coverage": len(
                set(task.tools_required) & set(sandbox.capabilities)
            ) / max(len(task.tools_required), 1),
        }

        return sum(
            self.weights.get(k, 0) * v 
            for k, v in features.items()
        )

    # generated by hugo AI