25 KiB
25 KiB
多专业小模型 + 路由模型 —— 实现方案
本方案从零开始，逐步构建可运行的原型系统。以代码生成为首个验证领域。
一、总体架构
┌───────────────────────────────────────────────────────────────┐
│ 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 环境准备
# 项目结构
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 构建分类路由器
推荐方案：使用小模型做意图识别
# 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 构建代码专家模型
# 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）
# 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 主路由逻辑
# 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 网关
# 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 运行与测试
# 安装依赖
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 领域专家微调流水线
# 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 缓存层设计
# 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 监控与评估
# 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 自动模型替换机制
# 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 年前沿研究与开源生态