# 多专业小模型 + 路由模型 —— 实现方案

> 本方案从零开始，逐步构建可运行的原型系统。以代码生成为首个验证领域。

---

## 一、总体架构

```
┌───────────────────────────────────────────────────────────────┐
│                        API Gateway                            │
└─────────────────────┬─────────────────────────────────────────┘
                      │
                      ▼
┌───────────────────────────────────────────────────────────────┐
│                     Router (路由层)                             │
│  ┌─────────────┐  ┌─────────────┐  ┌───────────────────────┐  │
│  │Classifier   │  │Difficulty   │  │Router Cache           │  │
│  │(意图识别)    │  │Estimator    │  │(频繁查询→直达路径)    │  │
│  └──────┬──────┘  └──────┬──────┘  └───────────────────────┘  │
└─────────┼────────────────┼────────────────────────────────────┘
          │                │
          ▼                ▼
┌───────────────────────────────────────────────────────────────┐
│                 Expert Pool (专家模型池)                         │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────┐ │
│  │Code      │ │Math      │ │Legal     │ │Medical   │ │...  │ │
│  │Expert    │ │Expert    │ │Expert    │ │Expert    │ │     │ │
│  │(1.5-7B)  │ │(1.5-4B)  │ │(3-7B)    │ │(3-7B)    │ │     │ │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘ └─────┘ │
└──────────────────────┬────────────────────────────────────────┘
                       │
                       ▼
┌───────────────────────────────────────────────────────────────┐
│                   Quality Controller (质量层)                   │
│  ┌─────────────────┐  ┌──────────────────────────────────┐   │
│  │Judge Model      │  │Fallback Trigger                  │   │
│  │(验证输出质量)    │  │(不合格时升级到大模型)             │   │
│  └─────────────────┘  └──────────────────────────────────┘   │
└──────────────────────┬────────────────────────────────────────┘
                       │ (必要时)
                       ▼
┌───────────────────────────────────────────────────────────────┐
│               Large Model Fallback (大模型回退)                 │
│               GPT-4 / Claude / DeepSeek 等                     │
└───────────────────────────────────────────────────────────────┘
```

---

## 二、分阶段实现路线

### 第一阶段：MVP（最小可行原型）— 4–6 周

**目标：** 构建一个端到端原型，聚焦单个领域（代码生成），验证路由 + 小模型的可行性。

#### 2.1 环境准备

```bash
# 项目结构
project_root/
├── models/                # 模型权重（Git LFS 或 symlink）
├── router/                # 路由模块
│   ├── classifier.py      # 分类器
│   ├── config.yaml        # 路由配置
│   └── router.py          # 主路由逻辑
├── experts/               # 专家模型封装
│   ├── code_expert.py     # 代码专家
│   └── base_expert.py     # 专家基类
├── judge/                 # 质量控制器
│   └── judge.py
├── gateway/               # API 网关
│   └── api.py
├── data/                  # 训练/评估数据
├── scripts/               # 训练和评估脚本
├── requirements.txt
└── README.md
```

#### 2.2 构建分类路由器

**推荐方案：使用小模型做意图识别**

```python
# router/classifier.py — 核心分类器
# 方案 A：用 Qwen3-0.5B / Llama-3.2-1B 做 few-shot 分类
# 方案 B：训练一个轻量 BERT 分类器（更小、更快）

from transformers import AutoModelForSequenceClassification, AutoTokenizer

class IntentClassifier:
    """意图分类器：识别查询属于哪个领域和难度"""
    
    def __init__(self, model_name="Qwen/Qwen3-0.5B"):
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name, 
            num_labels=6  # code, math, legal, medical, general, unknown
        )
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
    
    def classify(self, query: str) -> dict:
        """返回 {domain: str, confidence: float, difficulty: str}"""
        # 实现分类逻辑 + 置信度评估
        pass
    
    def should_fallback(self, query: str) -> bool:
        """判断是否直接走大模型（低置信度查询）"""
        result = self.classify(query)
        return result["confidence"] < 0.6
```

**关键设计指标：**
- 分类准确率 ≥ 95%
- 推理延迟 < 50ms
- 参数量建议 ≤ 1B

#### 2.3 构建代码专家模型

```python
# experts/code_expert.py

from transformers import AutoModelForCausalLM, AutoTokenizer

class CodeExpert:
    """代码生成专业模型"""
    
    def __init__(self, model_name="Qwen/Qwen2.5-Coder-7B-Instruct"):
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            device_map="auto",
            torch_dtype="bfloat16"
        )
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
    
    def generate(self, prompt: str, **kwargs) -> str:
        """生成代码回复"""
        messages = [{"role": "user", "content": prompt}]
        text = self.tokenizer.apply_chat_template(
            messages, tokenize=False
        )
        inputs = self.tokenizer(text, return_tensors="pt").to(self.model.device)
        outputs = self.model.generate(
            **inputs,
            max_new_tokens=kwargs.get("max_tokens", 2048),
            temperature=kwargs.get("temperature", 0.2),  # 代码生成用低温度
        )
        return self.tokenizer.decode(outputs[0], skip_special_tokens=True)
```

**模型选择依据：**

| 模型 | 参数量 | 代码能力 | 推理成本 | 推荐场景 |
|------|--------|---------|---------|---------|
| Qwen2.5-Coder-7B | 7B | ★★★★★ | 中 | 主要代码专家 |
| DeepSeek-Coder-1.3B | 1.3B | ★★★★ | 低 | 轻量代码任务 |
| BokantLM-0.5B | 0.5B | ★★★ | 极低 | 简单代码补全 |

#### 2.4 构建质量控制器（Judge）

```python
# judge/judge.py

class QualityJudge:
    """输出质量评估器——判断是否需要升级到大模型"""
    
    def __init__(self, model_name="Qwen3-1B-Instruct"):
        # 使用小模型作为 Judge，避免引入新的瓶颈
        pass
    
    def evaluate(self, query: str, response: str, domain: str) -> dict:
        """评估输出质量，返回评分和是否建议升级"""
        # 评估维度：
        # - 相关性：是否回答了问题
        # - 正确性：领域知识是否正确
        # - 完整性：是否遗漏关键信息
        # - 安全性：是否包含有害内容
        pass
    
    def needs_fallback(self, evaluation: dict) -> bool:
        """判断是否需要升级到大模型"""
        score = evaluation["overall_score"]
        return score < 0.7  # 低于阈值→升级
```

#### 2.5 主路由逻辑

```python
# router/router.py

class Router:
    """主路由器——协调整个系统"""
    
    def __init__(self, classifier, experts: dict, judge, fallback_model):
        self.classifier = classifier
        self.experts = experts
        self.judge = judge
        self.fallback = fallback_model
        self.cache = {}  # 简单缓存：高频查询跳过路由
    
    async def route(self, query: str) -> dict:
        """路由一个查询"""
        
        # Step 1: 检查缓存
        cache_key = self._cache_key(query)
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        # Step 2: 评估是否需要跳过路由直接走大模型
        if self.classifier.should_fallback(query):
            return await self._call_fallback(query)
        
        # Step 3: 分类
        classification = self.classifier.classify(query)
        domain = classification["domain"]
        
        # Step 4: 选择专家模型
        expert = self.experts.get(domain) or self.experts["general"]
        
        # Step 5: 生成回复
        response = expert.generate(query)
        
        # Step 6: 质量评估
        evaluation = self.judge.evaluate(query, response, domain)
        
        # Step 7: 决定是否升级
        if self.judge.needs_fallback(evaluation):
            response = await self._call_fallback(query)
            evaluation["upgraded"] = True
        
        result = {
            "response": response,
            "domain": domain,
            "upgraded": evaluation.get("upgraded", False),
            "quality_score": evaluation.get("overall_score"),
            "model_used": domain if not evaluation.get("upgraded") else "fallback"
        }
        
        # 高频查询写入缓存
        self.cache[cache_key] = result
        
        return result
    
    async def _call_fallback(self, query: str) -> str:
        """回退到大模型"""
        return await self.fallback.generate(query)
```

#### 2.6 API 网关

```python
# gateway/api.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Expert Router API")

class QueryRequest(BaseModel):
    query: str
    stream: bool = False

class QueryResponse(BaseModel):
    response: str
    domain: str
    upgraded: bool
    model_used: str

@app.post("/chat", response_model=QueryResponse)
async def chat(request: QueryRequest):
    return await router.route(request.query)

@app.get("/health")
async def health():
    return {"status": "ok", "experts": list(router.experts.keys())}

@app.get("/metrics")
async def metrics():
    """暴露路由统计信息"""
    return {
        "total_requests": stats.requests,
        "fallback_rate": stats.fallback_rate,
        "avg_latency": stats.avg_latency,
        "domain_distribution": stats.domain_distribution
    }
```

#### 2.7 运行与测试

```bash
# 安装依赖
pip install torch transformers fastapi uvicorn pyyaml

# 启动服务
python gateway/api.py

# 测试
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"query": "用Python写一个快速排序"}'
```

---

### 第二阶段：多领域扩展 — 4–6 周

#### 3.1 领域专家微调流水线

```python
# scripts/train_expert.py

"""
微调流水线：使用 LoRA 高效微调领域专家模型

使用方法：
    python train_expert.py --domain math --base_model Qwen/Qwen3-4B-Instruct
"""

from datasets import load_dataset
from transformers import (
    AutoModelForCausalLM, AutoTokenizer, TrainingArguments
)
from peft import LoraConfig, get_peft_model

def train_domain_expert(domain: str, base_model: str):
    """使用 LoRA 微调领域专家模型"""
    
    print(f"开始微调 {domain} 专家模型，基础模型: {base_model}")
    
    # 1. 加载领域数据集
    # 数学：GSM8K, MATH
    # 代码：CodeAlpaca, DomainCodeBench
    # 法律：LAiW, UCL-Bench
    # 医学：MedQA, PubMedQA
    dataset = load_dataset(f"domain_data/{domain}_train")
    
    # 2. 加载基础模型
    model = AutoModelForCausalLM.from_pretrained(base_model)
    tokenizer = AutoTokenizer.from_pretrained(base_model)
    
    # 3. 配置 LoRA
    lora_config = LoraConfig(
        r=16,         # 秩
        lora_alpha=32,
        target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
        lora_dropout=0.1,
        bias="none",
        task_type="CAUSAL_LM"
    )
    model = get_peft_model(model, lora_config)
    
    # 4. 训练
    training_args = TrainingArguments(
        output_dir=f"./models/{domain}_expert_lora",
        num_train_epochs=3,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        learning_rate=2e-4,
        logging_steps=10,
        save_steps=500,
    )
    
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=dataset,
    )
    trainer.train()
    
    # 5. 保存
    model.save_pretrained(f"./models/{domain}_expert_lora")
    print(f"{domain} 专家模型微调完成")
```

**推荐领域扩展优先级：**

```
阶段一             阶段二             阶段三
┌────────┐      ┌────────┐       ┌────────┐
│ 代码   │ ──── │ 数学   │ ────  │ 法律   │
│ (7B)   │      │ (4B)   │       │ (7B)   │
└────────┘      └────────┘       └────────┘
                      │                │
                      ▼                ▼
                 ┌────────┐       ┌────────┐
                 │ 通用   │       │ 医学   │
                 │ (1.5B) │       │ (7B)   │
                 └────────┘       └────────┘
```

#### 3.2 缓存层设计

```python
# router/cache.py

class RouterCache:
    """
    两阶段缓存：
    L1: 精确匹配缓存（完全相同查询）
    L2: 语义相似缓存（相似查询命中，使用 embedding 检索）
    """
    
    def __init__(self, embedding_model="BAAI/bge-small-zh-v1.5"):
        self.embedder = self._load_embedder(embedding_model)
        self.exact_cache = {}       # {query_hash: result}
        self.semantic_store = []    # [(embedding, query, result)]
        self.similarity_threshold = 0.92
    
    def get(self, query: str):
        """尝试从缓存获取"""
        # L1: 精确匹配
        if query in self.exact_cache:
            return self.exact_cache[query]
        
        # L2: 语义匹配
        q_emb = self._embed(query)
        for emb, cached_query, result in self.semantic_store:
            if self._cosine_sim(q_emb, emb) > self.similarity_threshold:
                return result
        
        return None
    
    def put(self, query: str, result: dict, frequency: int = 1):
        """写入缓存"""
        if frequency > 5:  # 高频查询→精确缓存
            self.exact_cache[query] = result
        else:
            self.semantic_store.append((self._embed(query), query, result))
```

---

### 第三阶段：生产化 — 4–6 周

#### 4.1 监控与评估

```python
# scripts/evaluate_router.py

"""
路由系统持续评估脚本
定期在 benchmark 上评估路由系统性能
"""

DOMAIN_BENCHMARKS = {
    "code": "humaneval",        # 代码生成
    "math": "gsm8k",            # 数学推理
    "legal": "laiw",            # 法律推理
    "medical": "medqa",         # 医学问答
    "general": "mmlu",          # 通用知识
}

METRICS = {
    "accuracy": None,       # 路由决策准确率
    "fallback_rate": 0.15,  # 目标：≤15% 升级率
    "avg_latency": 500,     # 目标：≤500ms
    "cost_ratio": 0.15,     # 目标：≤大模型的 15%
    "quality_delta": 0.0,   # 目标：质量不下降（Δ≥0）
}

def evaluate_routing_system():
    """
    评估流水线：
    1. 加载 benchmark 数据集
    2. 对每条数据：路由决策 → 专家生成 → 评分
    3. 与大模型 baseline 对比
    4. 输出报告
    """
    results = {}
    
    for domain, benchmark in DOMAIN_BENCHMARKS.items():
        accuracy = run_benchmark(domain, benchmark)
        results[domain] = accuracy
    
    # 计算总体指标
    report = {
        "overall_accuracy": sum(results.values()) / len(results),
        "per_domain": results,
        "fallback_rate": calculate_fallback_rate(),
        "cost_savings": calculate_cost_savings(),
        "latency_p50": calculate_latency_percentile(50),
        "latency_p99": calculate_latency_percentile(99),
    }
    
    return report
```

#### 4.2 自动模型替换机制

```python
# scripts/model_updater.py

"""
利用 Densing Law：每 3.5 个月能力密度翻倍
定期检查是否有更小的新模型可以达到同等性能
"""

class ModelUpdater:
    """自动追踪新模型，评估是否替换旧专家"""
    
    def __init__(self):
        self.models_registry = {}  # {domain: current_model_info}
    
    def check_for_updates(self):
        """定期检查是否有更优模型可用"""
        for domain, current in self.models_registry.items():
            # 搜索该领域的新 SOTA 小模型
            candidates = self._search_new_models(domain)
            
            for candidate in candidates:
                # 如果新模型参数量更小且性能不降
                if (candidate["parameters"] < current["parameters"] and
                    candidate["benchmark_score"] >= current["benchmark_score"] * 0.98):
                    
                    print(f"发现更优模型: {candidate['name']} "
                          f"({candidate['parameters']}B vs {current['parameters']}B)")
                    self._propose_replacement(domain, candidate)
```

#### 4.3 部署架构

```
                          ┌──────────────┐
                          │  Load        │
                          │  Balancer    │
                          └──────┬───────┘
                                 │
                 ┌───────────────┼───────────────┐
                 ▼               ▼               ▼
          ┌────────────┐  ┌────────────┐  ┌────────────┐
          │ Router     │  │ Router     │  │ Router     │
          │ Instance 1 │  │ Instance 2 │  │ Instance 3 │
          └─────┬──────┘  └─────┬──────┘  └─────┬──────┘
                │               │               │
                └───────────────┼───────────────┘
                                │
                    ┌───────────┴───────────┐
                    │                       │
              ┌─────┴──────┐         ┌──────┴─────┐
              │ Expert Pool│         │ Judge/Pool │
              │ (GPU Nodes)│         │ (GPU Nodes)│
              └────────────┘         └────────────┘
```

---

## 三、关键技术决策说明

### 3.1 为什么选择 Qwen 系列作为起点？

| 因素 | Qwen3/Qwen2.5 | Llama 3 | DeepSeek |
|------|--------------|---------|----------|
| 中文能力 | ★★★★★ | ★★★ | ★★★★ |
| 小模型效果 | ★★★★ (0.5B起) | ★★★★★ (1B起) | ★★★★ (1.3B起) |
| 代码专项 | ★★★★★ (有Coder版) | ★★★★ | ★★★★★ |
| 开源协议 | ★★★★ (Apache 2.0) | ★★★★ | ★★★ |
| LoRA 生态 | 成熟 | 成熟 | 成熟 |

**建议：** 中文场景优先选 Qwen 系列；英文/代码场景 Qwen Coder 或 DeepSeek 均可。

### 3.2 路由器选型对比

| 方案 | 延迟 | 准确率 | 成本 | 推荐场景 |
|------|------|--------|------|---------|
| BERT 分类器 (110M) | <10ms | 90-93% | 极低 | 初期/快速原型 |
| 小 LLM 分类 (0.5-1B) | 30-80ms | 94-97% | 低 | 正式环境 |
| 嵌入向量路由 (如 RouterRetriever) | 15-30ms | 93-96% | 低 | 需要动态增减专家 |
| LLM-as-Router (3-7B) | 100-300ms | 96-98% | 中 | 复杂路由逻辑 |

### 3.3 质量控制器方案

**方案 A：小模型 Judge（推荐）**
- 使用 1-3B 模型评估输出质量
- 延迟低（50-150ms）
- 足够判断明显质量问题

**方案 B：LLM-as-Judge**
- 使用 7-13B 模型
- 评估更全面（可覆盖事实性、合规性等）
- 成本较高，仅在 P99 场景使用

**方案 C：规则 + 模型混合**
- 规则快速过滤（格式、长度、关键词）
- 模型处理复杂评估
- 最佳性价比

---

## 四、预算估算

### 4.1 开发阶段硬件

| 项目 | 规格 | 预估成本（月） |
|------|------|--------------|
| GPU 节点 1 | 1× A100 80GB | ¥15,000–20,000 |
| GPU 节点 2 | 1× RTX 4090 24GB | ¥5,000–8,000 |
| CPU 节点 | 8核 32GB | ¥1,000–2,000 |

### 4.2 运行阶段推理成本

假设日均 10 万次请求：

| 方案 | 估算成本（月） | 说明 |
|------|--------------|------|
| 全量大模型 (70B) | ¥50,000–80,000 | 全部请求走 API |
| 路由系统 | ¥5,000–12,000 | 80% 小模型，20% 升级 |
| **节省** | **¥45,000–68,000** | **节省 80–90%** |

### 4.3 开源替代方案（零成本起步）

- 模型：Qwen2.5-Coder / Qwen3 (开源)
- 路由器：BERT 分类器 (自训练)
- Judge：小模型 (开源)
- 推理框架：vLLM / llama.cpp (开源)
- 部署：Docker + 单 GPU (成本可控)

---

## 五、验证指标与成功标准

### 5.1 第一阶段验收标准
- [ ] 路由系统能正确分类 ≥5 种意图
- [ ] 代码专家在 HumanEval 上 pass@1 ≥ 大模型 baseline 的 95%
- [ ] 端到端延迟 < 大模型推理的 1.5×
- [ ] 成本 ≤ 大模型方案的 20%

### 5.2 第二阶段验收标准
- [ ] 3 个以上领域专家均达到或接近 SOTA
- [ ] 质量控制器准确率 ≥90%（与大模型评估的一致性）
- [ ] 缓存命中率 ≥30%
- [ ] 升级率（fallback rate）≤ 20%

### 5.3 第三阶段验收标准
- [ ] 系统可灰度发布、A/B 测试
- [ ] 模型热替换不影响在线服务
- [ ] 整体成本降低 ≥80%

---

## 六、开源工具与参考实现

| 组件 | 推荐工具 | 用途 |
|------|---------|------|
| 模型推理 | vLLM, llama.cpp, TGI | 高性能推理引擎 |
| 微调 | unsloth, LLaMA-Factory, Axolotl | LoRA/QLoRA 高效微调 |
| 路由 | RouterArena (评估), 自建 | 路由决策 |
| 监控 | Prometheus + Grafana | 延迟、升级率、准确率 |
| 嵌入 | BGE, text-embedding-3-small | 语义缓存 |
| 评估 | LM Evaluation Harness | 标准化 benchmark |

---

## 七、时间线总览

```
第 1-2 周     第 3-4 周      第 5-6 周      第 7-10 周      第 11-14 周
┌────────┐   ┌────────┐    ┌────────┐    ┌─────────┐     ┌──────────┐
│ 环境   │→  │ 路由   │→   │ 代码   │→   │ 多领域  │→    │ 生产化   │
│ 搭建   │   │ 原型   │    │ 专家   │    │ 扩展    │     │ 部署     │
└────────┘   └────────┘    └────────┘    └─────────┘     └──────────┘
  · 选型       · 分类器      · 微调      · 数学专家     · 缓存
  · 环境       · Router     · 评估      · 法律专家     · 监控
  · 数据       · Judge      · 迭代      · 医学专家     · CI/CD
                              · RouterArena · 热替换
```

---

## 八、风险清单

| 风险 | 概率 | 影响 | 应对 |
|------|------|------|------|
| 路由误分类导致回答质量下降 | 中 | 高 | 引入置信度阈值，低置信走大模型 |
| 小模型推理天花板（复杂任务） | 高 | 中 | 级联升级机制兜底 |
| 多模型管理复杂度超预期 | 中 | 中 | 使用统一推理框架 + LoRA 统一管理 |
| 缓存击穿导致大模型负载飙升 | 低 | 高 | 限流 + 降级 + 预缓存热门 query |
| 领域数据不足微调效果差 | 中 | 中 | 使用合成数据 + few-shot 先验证 |
| 模型能力密度快速变化导致架构重选 | 低 | 低 | 接口抽象化，模型替换不影响路由层 |

---

*文档生成日期：2026-07-30*
*基于 2025 年前沿研究与开源生态*