682 lines
25 KiB
Markdown
682 lines
25 KiB
Markdown
# 多专业小模型 + 路由模型 —— 实现方案
|
||
|
||
> 本方案从零开始，逐步构建可运行的原型系统。以代码生成为首个验证领域。
|
||
|
||
---
|
||
|
||
## 一、总体架构
|
||
|
||
```
|
||
┌───────────────────────────────────────────────────────────────┐
|
||
│ 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 年前沿研究与开源生态*
|