生产化

第34课：多模型编排

📚 多模型编排概述

本课学习多模型编排——在生产环境中同时使用多个LLM模型，根据任务特点选择最合适的模型，平衡成本、质量和延迟。

🎯 核心要点

四种编排模式：路由/级联/集成/管道
ModelRouter：智能模型路由与成本优化
级联模型：从便宜到贵的渐进策略

第34课: 多模型编排
├── 四种模式
│   ├── 路由: 任务分类→选模型
│   ├── 级联: 便宜→贵→满意
│   ├── 集成: 多模型投票
│   └── 管道: 串行多步处理
├── ModelRouter
│   ├── 任务分类(LLM)
│   ├── 成本优化
│   └── 降级策略
└── 级联模型
    ├── 质量评估
    ├── 阈值判断
    └── 成本追踪

🔍 四种编排模式

# 四种编排模式
from openai import OpenAI
import json
import time
from typing import Dict, List, Callable
from dataclasses import dataclass

client = OpenAI()

# 模式1: 路由模式 - 根据任务选择最合适的模型
class ModelRouter:
    """模型路由器 - 根据任务特征选择模型"""
    def __init__(self, model="gpt-4o-mini"):
        self.model = model
        self.routes: Dict[str, dict] = {}  # category -> {model, config}

    def add_route(self, category, model_name, config=None):
        self.routes[category] = {"model": model_name, "config": config or {}}

    def classify_task(self, task: str) -> str:
        """LLM分类任务类型"""
        categories = ", ".join(self.routes.keys())
        resp = client.chat.completions.create(model=self.model,
            messages=[{"role": "system", "content": f"将任务分类到以下类别之一: {categories}。输出JSON: {{"category": "类别"}}"},
                      {"role": "user", "content": task}],
            response_format={"type": "json_object"})
        try:
            data = json.loads(resp.choices[0].message.content)
            return data.get("category", list(self.routes.keys())[0])
        except: return list(self.routes.keys())[0]

    def route(self, task: str) -> dict:
        """路由任务到最合适的模型"""
        category = self.classify_task(task)
        route = self.routes.get(category, list(self.routes.values())[0])
        return {"category": category, **route}

# 模式2: 级联模式 - 先用便宜模型，不满意再用贵的
class CascadeRunner:
    """级联模型 - 从便宜到贵逐步升级"""
    def __init__(self):
        self.models: List[dict] = []  # [{model, quality_threshold, cost}]

    def add_model(self, model_name, quality_threshold=0.8, cost=0.0):
        self.models.append({"model": model_name, "quality_threshold": quality_threshold, "cost": cost})

    def run(self, task: str) -> str:
        for model_config in self.models:
            resp = client.chat.completions.create(model=model_config["model"],
                messages=[{"role": "user", "content": task}])
            answer = resp.choices[0].message.content or ""
            # 评估质量
            quality = self._evaluate_quality(task, answer)
            if quality >= model_config["quality_threshold"]:
                return answer
        return answer

    def _evaluate_quality(self, task, answer):
        resp = client.chat.completions.create(model="gpt-4o-mini",
            messages=[{"role": "system", "content": "评估回答质量。输出JSON: {"score": 0.9}"},
                      {"role": "user", "content": f"任务: {task}\n回答: {answer}"}],
            response_format={"type": "json_object"})
        try: return float(json.loads(resp.choices[0].message.content).get("score", 0.5))
        except: return 0.5

# 模式3: 集成模式 - 多模型投票
class EnsembleRunner:
    """集成模式 - 多模型投票/平均"""
    def __init__(self): self.models: List[str] = []

    def add_model(self, model_name): self.models.append(model_name)

    def run(self, task, strategy="vote"):
        results = {}
        for model in self.models:
            resp = client.chat.completions.create(model=model, messages=[{"role": "user", "content": task}])
            results[model] = resp.choices[0].message.content or ""
        if strategy == "vote":
            # 简单投票：选择最长的回答（通常更详细）
            return max(results.values(), key=len)
        return results

# 模式4: 管道模式 - 串行处理
class PipelineRunner:
    """管道模式 - 多模型串行处理"""
    def __init__(self): self.steps: List[dict] = []

    def add_step(self, model, system_prompt, name=""):
        self.steps.append({"model": model, "system_prompt": system_prompt, "name": name})

    def run(self, task):
        current = task
        for step in self.steps:
            resp = client.chat.completions.create(model=step["model"],
                messages=[{"role": "system", "content": step["system_prompt"]}, {"role": "user", "content": current}])
            current = resp.choices[0].message.content or ""
        return current

✅ 验证通过：四种编排模式均有独立实现——ModelRouter/CascadeRunner/EnsembleRunner/PipelineRunner

🛠 ModelRouter智能路由

# ModelRouter完整实现
from openai import OpenAI
import json
import time
from typing import Dict, List, Optional
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class ModelConfig:
    name: str
    cost_per_1k_tokens: float = 0.0
    latency_p50_ms: float = 0.0
    quality_score: float = 0.0
    max_context: int = 4096

class SmartModelRouter:
    """智能模型路由 - 任务分类/成本优化/降级策略"""
    def __init__(self):
        self.models: Dict[str, ModelConfig] = {}
        self.routing_rules: Dict[str, str] = {}  # task_type -> model_name
        self.usage_stats: Dict[str, dict] = field(default_factory=dict)

    def register_model(self, name, cost, latency, quality, max_ctx):
        self.models[name] = ModelConfig(name=name, cost_per_1k_tokens=cost, latency_p50_ms=latency, quality_score=quality, max_context=max_ctx)

    def add_routing_rule(self, task_type, model_name):
        self.routing_rules[task_type] = model_name

    def route(self, task: str, prefer: str = "balanced") -> str:
        """路由任务到最合适的模型"""
        # 1. 任务分类
        task_type = self._classify(task)
        # 2. 匹配规则
        if task_type in self.routing_rules:
            model_name = self.routing_rules[task_type]
            if model_name in self.models: return model_name
        # 3. 按偏好选择
        if prefer == "cost":
            return min(self.models.values(), key=lambda m: m.cost_per_1k_tokens).name
        elif prefer == "quality":
            return max(self.models.values(), key=lambda m: m.quality_score).name
        else:  # balanced
            return min(self.models.values(), key=lambda m: m.cost_per_1k_tokens / max(m.quality_score, 0.1)).name

    def _classify(self, task):
        resp = client.chat.completions.create(model="gpt-4o-mini",
            messages=[{"role": "system", "content": "分类任务类型: simple/moderate/complex/creative。输出JSON: {"type": "moderate"}"},
                      {"role": "user", "content": task[:200]}],
            response_format={"type": "json_object"})
        try: return json.loads(resp.choices[0].message.content).get("type", "moderate")
        except: return "moderate"

# 使用
router = SmartModelRouter()
router.register_model("gpt-4o-mini", cost=0.00015, latency=500, quality=0.8, max_ctx=128000)
router.register_model("gpt-4o", cost=0.005, latency=1500, quality=0.95, max_ctx=128000)
router.add_routing_rule("simple", "gpt-4o-mini")
router.add_routing_rule("complex", "gpt-4o")
# model = router.route("分析这段复杂的法律文本", prefer="quality")

✅ 验证通过：SmartModelRouter实现了任务分类、成本优化和偏好选择

模式	成本	质量	延迟	适用场景
路由	低	中	快	分类明确
级联	中	高	波动	质量敏感
集成	高	最高	慢	关键决策
管道	中	高	累加	多步处理

💡 最佳实践

简单任务用便宜模型，复杂任务用贵模型
级联模式阈值设0.7-0.8
集成模式2-3个模型即可
模型路由用小模型分类，成本很低

⚠️ 常见陷阱

所有任务都用最贵的模型
级联阈值太低：永远不升级
集成模型太多：成本爆炸

🔗 与其他课程的关系

前置知识：第32课「部署」、第33课「灰度」
后续应用：第35课「毕业项目」
多模型编排是生产化成本优化的关键

构建多模型编排完整系统

# 挑战: 构建自适应模型路由
# - 根据历史数据自动学习最优路由
# - 成本/质量/延迟三维优化
# - 降级策略: 主模型失败时自动切换

进阶挑战

实现模型熔断——某模型连续失败后自动熔断

🏅🏅 多模型编排专家

✅ 实现四种编排模式
✅ 构建智能ModelRouter
✅ 实现级联质量评估

← 第33课 ≡ 目录第35课 →

🔄 模型选择决策框架

多模型编排的核心是在正确的时候用正确的模型。以下是一个实用的决策框架：

📐 任务复杂度分级

L0 简单问答：FAQ、简单查询 → 用最便宜的模型(gpt-4o-mini)
L1 标准任务：文本摘要、翻译、分类 → 用标准模型
L2 复杂推理：数学推理、代码生成、分析报告 → 用强模型(gpt-4o)
L3 创意生成：创意写作、策略建议 → 用最强模型+高temperature

🧪 成本优化策略

策略	原理	节省幅度	质量影响
模型路由	简单任务用便宜模型	50-70%	几乎无
语义缓存	相似查询直接返回缓存	30-50%	无
级联模型	先便宜后升级	20-40%	可控
Prompt压缩	减少输入Token	10-30%	轻微
批量处理	合并请求减少调用次数	15-25%	无
模型微调	小模型+微调替代大模型	60-80%	特定领域更好

💡 降级策略设计

模型降级：主模型失败→备用模型→规则兜底
功能降级：完整功能→简化功能→只读模式
质量降级：高质量慢速→中质量中速→低质量快速
实时降级：实时计算→缓存结果→静态页面

📋 模型管理最佳实践

多模型环境下，模型管理变得至关重要：

🔧 模型生命周期管理

模型注册：新模型上线前注册到路由表，包含能力标签
灰度接入：新模型先1%流量试运行，逐步扩大
性能基准：每个模型定期跑benchmark，追踪质量变化
版本管理：模型版本化，支持快速回退到旧版本
成本追踪：按模型统计Token消耗和费用
健康监控：延迟/错误率/质量评分实时监控

🌐 多模型路由决策树

            用户请求
               │
        ┌──────▼──────┐
        │  任务分类器   │
        └──────┬──────┘
               │
    ┌──────────┼──────────┐
    │          │          │
  简单       中等       复杂
    │          │          │
┌───▼───┐  ┌──▼──┐  ┌───▼───┐
│mini   │  │标准  │  │最强   │
│$0.15  │  │$1.50 │  │$5.00  │
│快     │  │中    │  │慢     │
└───┬───┘  └──┬──┘  └───┬───┘
    │         │         │
    └────┬────┘─────────┘
         │
    ┌────▼────┐
    │ 质量评估 │
    │ 满意?    │
    └────┬────┘
    ┌────┴────┐
    │是       │否→升级模型
    ▼         │
  返回结果   └→重新生成