通用桌面 Agent 的 Query 分类与智能调度：四层意图漏斗设计

User Query
    │
    ▼
┌──────────────────────────────────────┐
│ Stage 1: 规则引擎                     │
│ 正则 + 关键词白名单，< 1ms              │
│ 命中率 ~50%，零 LLM 成本               │
└────────────┬─────────────────────────┘
             │ 未命中
             ▼
┌──────────────────────────────────────┐
│ Stage 2: Embedding 缓存               │
│ FAISS 内存索引，< 20ms                 │
│ 命中率 ~25%（累计划 75%）              │
└────────────┬─────────────────────────┘
             │ 置信度 < 0.92
             ▼
┌──────────────────────────────────────┐
│ Stage 3: 微调小模型                    │
│ DistilBERT 领域微调，< 50ms            │
│ 命中率 ~15%（累计划 90%）              │
└────────────┬─────────────────────────┘
             │ 模糊 / 复合意图
             ▼
┌──────────────────────────────────────┐
│ Stage 4: 完整 LLM                     │
│ Claude Sonnet / GPT-4o，< 800ms       │
│ 覆盖率 ~10%，处理真正的复杂推理         │
└──────────────────────────────────────┘

                        延迟（ms）   累计覆盖率   每次成本
Stage 1: 规则引擎         < 1         ~50%        $0
Stage 2: Embedding 缓存   < 20        ~75%        $0.0001
Stage 3: DistilBERT      < 50        ~90%        $0.0003
Stage 4: LLM 兜底         < 800       ~100%       $0.005

import re
from dataclasses import dataclass
from enum import Enum
from typing import Optional


class Intent(Enum):
    FILE_READ = "file_read"
    FILE_WRITE = "file_write"
    WEB_SEARCH = "web_search"
    CODE_EXECUTE = "code_execute"
    APP_LAUNCH = "app_launch"
    SYSTEM_CONTROL = "system_control"


@dataclass
class RuleMatch:
    intent: Intent
    confidence: float
    extracted_args: dict


RULES: list[tuple[str, Intent, dict]] = [
    # 文件操作
    (r"\b(?:打开|读|查看|显示)\s*(?:一下)?\s*(?:这个|那个|的)?\s*文件\b", Intent.FILE_READ, {}),
    (r"\b(?:创建|新建|写入|保存)\s*(?:一个|新的)?\s*文件\b", Intent.FILE_WRITE, {}),
    # Web 搜索
    (r"\b(?:搜索|搜|查|百度|Google)\s+(?:一下|一下)?", Intent.WEB_SEARCH, {}),
    # 代码执行
    (r"\b(?:运行|执行|跑)\s*(?:一下|这个)?\s*(?:脚本|代码|python|bash|命令)\b", Intent.CODE_EXECUTE, {}),
    # 应用启动
    (r"\b(?:打开|启动|运行)\s+(?:VS Code|Chrome|Terminal|Excel|Word)\b", Intent.APP_LAUNCH, {}),
]


def rule_classify(query: str) -> Optional[RuleMatch]:
    for pattern, intent, args_template in RULES:
        if match := re.search(pattern, query, re.IGNORECASE):
            return RuleMatch(
                intent=intent,
                confidence=0.95,
                extracted_args=args_template | match.groupdict(),
            )
    return None
# generated by hugo AI

import faiss
import numpy as np
from dataclasses import dataclass, field


@dataclass
class CachedClassification:
    query: str
    intent: Intent
    timestamp: float


class EmbeddingCache:
    def __init__(self, embedding_dim: int = 768, max_size: int = 10000):
        self.index = faiss.IndexFlatIP(embedding_dim)  # inner product for cosine
        self.entries: list[CachedClassification] = []
        self.max_size = max_size

    def search(self, query_embedding: np.ndarray, threshold: float = 0.92) -> Optional[Intent]:
        if self.index.ntotal == 0:
            return None
        scores, indices = self.index.search(query_embedding.reshape(1, -1), k=1)
        return self.entries[indices[0][0]].intent if scores[0][0] > threshold else None

    def add(self, query: str, query_embedding: np.ndarray, intent: Intent) -> None:
        self.entries.append(CachedClassification(query=query, intent=intent, timestamp=time.time()))
        self.index.add(query_embedding.reshape(1, -1))
        # LRU 淘汰
        if len(self.entries) > self.max_size:
            oldest_idx = min(range(len(self.entries)), key=lambda i: self.entries[i].timestamp)
            self.index.remove_ids(np.array([oldest_idx]))
            del self.entries[oldest_idx]
# generated by hugo AI

生产环境 query 日志
    → Stage 4 LLM 分类（离线批量，不要求实时）
    → 人工抽样校验（抽 5% 检查准确率）
    → 达标后作为 DistilBERT 微调数据
    → 部署，持续监控 Stage 3 vs Stage 4 的一致率

CLASSIFIER_SYSTEM_PROMPT = """You are an intent classifier for a desktop AI agent.
Given a user query, classify it into EXACTLY ONE primary intent.

Available intents and their definitions:
- file_read: Read or access file contents
- file_write: Create, write, or modify files
- file_organize: Move, rename, delete, or batch-process files
- web_search: Search or browse the internet
- web_action: Fill forms, submit, interact with web pages
- code_execute: Run scripts, execute commands, git operations
- app_automation: Launch and interact with desktop applications
- system_control: Change system settings, manage processes
- communication: Send emails, messages, schedule meetings
- media_process: Screenshots, image/video editing, OCR

For ambiguous queries, choose the most likely primary intent.
Reply with ONLY the intent name, nothing else.
"""
# generated by hugo AI

@dataclass
class RouteContext:
    active_app: str
    recent_actions: list[str]
    user_preferences: dict


class ContextAwareRouter:
    def route(self, intent: Intent, query: str, ctx: RouteContext) -> str:
        # 根据活跃应用微调工具选择
        tool_priority = self._compute_priority(intent, ctx.active_app)
        # 根据历史成功率加权
        best_tool = max(
            tool_priority,
            key=lambda t: self.tool_success_rates.get(t, {}).get(intent.value, 0.5),
        )
        return best_tool
# generated by hugo AI

Action selection:
    ├── 有 API？→ 直接调用（< 50ms，确定性 100%）
    ├── 有 URL Scheme？→ deep link 调用
    └── 都没有 → 走 GUI 自动化（500ms-2000ms，非确定性）

┌─────────────────────────────────────────────┐
│              自进化反馈闭环                    │
│                                              │
│  用户纠错 / 隐式反馈（undo、重试）              │
│      │                                       │
│      ▼                                       │
│  记录 (query, 预测意图, 实际意图, 成功/失败)    │
│      │                                       │
│      ▼                                       │
│  每 N 次检查：准确率 < 85%？                   │
│      │                                       │
│      ├── 是 → 触发分析                        │
│      │      ├── 同 pattern 3+ 次 → 更新规则   │
│      │      ├── Embedding 漂移 → 重建缓存     │
│      │      └── 分布整体偏移 → 重新微调        │
│      │                                       │
│      └── 否 → 继续监控                        │
└─────────────────────────────────────────────┘

from collections import defaultdict


class SelfEvolvingRouter:
    def __init__(self, drift_check_interval: int = 100, accuracy_threshold: float = 0.85):
        self.history: list[dict] = []
        self.drift_check_interval = drift_check_interval
        self.accuracy_threshold = accuracy_threshold
        # pattern → 纠错计数
        self.correction_patterns: defaultdict[str, int] = defaultdict(int)

    def record(self, query: str, predicted: Intent, actual: Intent, success: bool) -> None:
        self.history.append({
            "query": query,
            "predicted": predicted,
            "actual": actual,
            "success": success,
        })

        if not success and predicted != actual:
            pattern_key = f"{predicted.value}→{actual.value}"
            self.correction_patterns[pattern_key] += 1

        if len(self.history) % self.drift_check_interval == 0:
            self._check_drift()

    def _check_drift(self) -> None:
        recent = self.history[-self.drift_check_interval:]
        accuracy = sum(1 for r in recent if r["success"]) / len(recent)

        if accuracy < self.accuracy_threshold:
            # 分析高频纠错 pattern
            for pattern, count in self.correction_patterns.items():
                if count >= 3:
                    predicted_intent, actual_intent = pattern.split("→")
                    self._update_rules(predicted_intent, actual_intent)
            self.correction_patterns.clear()

    def _update_rules(self, wrong: str, correct: str) -> None:
        # 基于高频纠错模式自动更新规则引擎的优先级
        # 实现细节取决于具体规则引擎的设计
        pass
# generated by hugo AI