Ollama: 多模型编排
多模型编排是 AI 的团队作战——小模型冲锋，大模型压阵，各司其职。
💡 提示：多模型路由策略的核心是"按任务分配"——简单 FAQ 用 3B 小模型（1 秒响应，低 GPU 消耗），复杂推理用 8B 大模型（5 秒响应，高精度）。也可"按负载分配"——高峰期用小模型处理更多请求，低谷期切大模型提升质量。80% 请求走小模型可将平均延迟降低 50%。
📋 前置知识：需要先掌握以下内容
- 第13课：LangChain集成
1. 你将学到
- 多模型架构模式：路由器 / 级联 / 并行投票
- 路由器模式：小模型分类 → 大模型精答
- 级联模式：粗筛 → 精炼 → 格式化流水线
- Python asyncio 并发调用多个 Ollama 实例
- Alice 的 SupportBot V4：路由器分流
2. 一个 SaaS 创业者的真实故事
ℹ️ 信息： 多模型编排时，每个模型需要独立加载到 VRAM。3B + 8B 两个模型同时驻留约需 10GB VRAM。如果 VRAM 不足，Ollama 会自动卸载不活跃模型，但切换时会有 5-30 秒的重新加载延迟。
(1) 痛点：大模型处理简单问题浪费资源
Alice 发现 SupportBot 80% 的问题（"退货政策是什么？""怎么查物流？"）只需小模型就能回答，但全部走大模型浪费 GPU 资源且速度慢。
(2) 解法：路由器模式分流
小模型（3B）1 秒分类意图，简单问题直接回答，复杂问题路由到大模型（8B）：
PYTHON
# Router pattern: small model classifies, large model handles complex
intent = small_model.classify(question) # 1 second
if intent == "simple":
answer = small_model.answer(question) # 2 seconds
else:
answer = large_model.answer(question) # 5 seconds
3. 三种编排模式
⚠️ 注意：并行投票模式虽然可靠性最高，但成本也最高——每个请求需要调用 3-5 个模型，GPU 计算量成倍增加。仅在关键决策场景（如医疗建议、法律判断）使用，日常客服场景路由器模式足够。
💡 提示： 路由器模式是性价比最高的编排方式——80% 的简单请求用 3B 模型（1 秒响应），仅 20% 复杂请求路由到 8B 模型。整体平均响应时间从 5 秒降到 2 秒，GPU 利用率降低 60%。
(1) 模式对比
| 模式 | 流程 | 适用场景 | 延迟 | 成本 |
|---|---|---|---|---|
| 路由器 | 分类 → 分发 | 客服分流、任务分类 | 低 | 低 |
| 级联 | 粗筛 → 精炼 → 格式化 | 代码生成、内容创作 | 中 | 中 |
| 并行投票 | 多模型 → 投票/合并 | 高可靠性决策 | 高 | 高 |
flowchart TD
subgraph Router
R1[Input] --> R2[Classifier<br/>3B Model]
R2 -->|Simple| R3[Small Model Answer]
R2 -->|Complex| R4[Large Model Answer]
end
subgraph Cascade
C1[Input] --> C2[Draft<br/>3B Model]
C2 --> C3[Refine<br/>8B Model]
C3 --> C4[Format<br/>3B Model]
end
subgraph Voting
V1[Input] --> V2[Model A]
V1 --> V3[Model B]
V1 --> V4[Model C]
V2 --> V5[Majority Vote]
V3 --> V5
V4 --> V5
end
4. 路由器模式
(1) 路由器架构
sequenceDiagram
participant U as User
participant C as Classifier (3B)
participant S as Small Model (3B)
participant L as Large Model (8B)
U->>C: "I want a refund"
C-->>C: Intent: refund (simple)
C->>S: Route to small model
S-->>U: Refund policy answer
U->>C: "Compare warranty terms for 3 products"
C-->>C: Intent: comparison (complex)
C->>L: Route to large model
L-->>U: Detailed comparison
(2) 意图分类标签设计
| 标签 | 复杂度 | 路由目标 | 示例问题 |
|---|---|---|---|
| faq | 简单 | 小模型 | "退货政策是什么？" |
| order_status | 简单 | 小模型 | "订单 #12345 到哪了？" |
| product_info | 中等 | 中模型 | "这个耳机支持蓝牙 5.3 吗？" |
| comparison | 复杂 | 大模型 | "比较这 3 款耳机的区别" |
| complaint | 复杂 | 大模型 | "收到的商品与描述严重不符" |
▶ 示例 1: 路由器模式实现
PYTHON
import ollama
import json
from dataclasses import dataclass
from typing import Optional
@dataclass
class ModelRouter:
classifier_model: str = "llama3.2:3b"
small_model: str = "llama3.2:3b"
large_model: str = "qwen2.5"
INTENTS = {
"faq": "simple",
"order_status": "simple",
"product_info": "medium",
"comparison": "complex",
"complaint": "complex"
}
def classify(self, question: str) -> dict:
response = ollama.chat(
model=self.classifier_model,
messages=[{
"role": "user",
"content": f"""Classify this customer question.
Return JSON: {{"intent": "faq|order_status|product_info|comparison|complaint", "complexity": "simple|medium|complex"}}
Question: {question}"""
}],
format="json",
stream=False,
options={"temperature": 0.1}
)
return json.loads(response["message"]["content"])
def route(self, question: str, system: str = "") -> str:
intent_data = self.classify(question)
complexity = intent_data.get("complexity", "simple")
model = self.small_model if complexity == "simple" else self.large_model
messages = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": question})
response = ollama.chat(model=model, messages=messages,
stream=False, options={"temperature": 0.3})
return response["message"]["content"], model, intent_data
# Usage
router = ModelRouter()
questions = [
"What is the return policy?",
"Compare the 3 wireless headphones you sell",
"Where is order #88765?"
]
for q in questions:
answer, model_used, intent = router.route(q, system="You are SupportBot.")
print(f"Q: {q}")
print(f"Intent: {intent}, Model: {model_used}")
print(f"A: {answer[:100]}...\n")
输出：
TEXT
📖 仅展示
# 函数定义成功
5. 级联模式
(1) 级联架构详解
| 阶段 | 模型 | 任务 | 参数 |
|---|---|---|---|
| Draft | 3B | 快速草稿 | temperature=0.5 |
| Refine | 8B | 深度优化 | temperature=0.3 |
| Format | 3B | 格式化输出 | temperature=0.1 |
▶ 示例 2: 级联内容生成
PYTHON
import ollama
class CascadePipeline:
def __init__(self):
self.draft_model = "llama3.2:3b"
self.refine_model = "qwen2.5"
self.format_model = "llama3.2:3b"
def generate(self, topic: str) -> dict:
# Stage 1: Draft
draft = ollama.chat(
model=self.draft_model,
messages=[{"role": "user",
"content": f"Write a brief draft about: {topic}"}],
stream=False, options={"temperature": 0.5}
)["message"]["content"]
# Stage 2: Refine
refined = ollama.chat(
model=self.refine_model,
messages=[
{"role": "system", "content": "Improve the following text. Add details and fix errors."},
{"role": "user", "content": draft}
],
stream=False, options={"temperature": 0.3}
)["message"]["content"]
# Stage 3: Format
formatted = ollama.chat(
model=self.format_model,
messages=[
{"role": "system", "content": "Format this text as a professional product description with bullet points."},
{"role": "user", "content": refined}
],
stream=False, options={"temperature": 0.1}
)["message"]["content"]
return {"draft": draft, "refined": refined, "formatted": formatted}
pipeline = CascadePipeline()
result = pipeline.generate("wireless noise-cancelling headphones")
print("=== Final Output ===")
print(result["formatted"])
输出：
TEXT
📖 仅展示
=== Final Output ===
6. 并行投票与 asyncio 并发
(1) 并行投票架构
| 模式 | 模型数 | 决策方式 | 适用 |
|---|---|---|---|
| 多数投票 | 3+ | 取多数一致答案 | 事实判断 |
| 加权平均 | 2+ | 按模型质量加权 | 评分任务 |
| 最佳选择 | 2+ | 选最详细/最可信 | 开放问题 |
▶ 示例 3: asyncio 并发调用
PYTHON
import asyncio
import ollama
from dataclasses import dataclass
@dataclass
class ParallelVoter:
models: list[str] = None
def __post_init__(self):
if self.models is None:
self.models = ["llama3.2:3b", "qwen2.5", "mistral"]
async def query_model(self, model: str, question: str) -> dict:
client = ollama.AsyncClient()
response = await client.chat(
model=model,
messages=[{"role": "user", "content": question}],
stream=False,
options={"temperature": 0.3}
)
return {"model": model, "answer": response["message"]["content"]}
async def vote(self, question: str) -> dict:
tasks = [self.query_model(m, question) for m in self.models]
results = await asyncio.gather(*tasks)
# Simple voting: check for consensus
answers = [r["answer"] for r in results]
return {
"question": question,
"results": results,
"consensus": len(set(answers[:50] for a in answers)) == 1
}
# Usage
async def main():
voter = ParallelVoter()
result = await voter.vote("Is 2+2 equal to 4?")
for r in result["results"]:
print(f"{r['model']}: {r['answer'][:80]}")
asyncio.run(main())
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例 4: 加权集成
PYTHON
import ollama
def weighted_ensemble(question: str, models: list[dict]) -> str:
"""Query multiple models and select the best response."""
responses = []
for m in models:
resp = ollama.chat(
model=m["name"],
messages=[{"role": "user", "content": question}],
stream=False,
options={"temperature": m.get("temperature", 0.3)}
)
responses.append({
"model": m["name"],
"answer": resp["message"]["content"],
"weight": m.get("weight", 1.0),
"length": len(resp["message"]["content"])
})
# Select longest response from highest-weight model
responses.sort(key=lambda x: x["weight"] * x["length"], reverse=True)
return responses[0]["answer"]
models = [
{"name": "qwen2.5", "weight": 1.5, "temperature": 0.3},
{"name": "llama3.2:3b", "weight": 0.8, "temperature": 0.5},
]
result = weighted_ensemble("Explain the benefits of local AI", models)
print(result)
输出：
TEXT
📖 仅展示
# 函数定义成功
7. 综合示例：SupportBot V4 路由器系统
PYTHON
# ============================================
# Comprehensive: SupportBot V4 with Router
# Multi-model routing for e-commerce support
# ============================================
import ollama
import json
import time
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class SupportBotV4:
classifier: str = "llama3.2:3b"
simple_model: str = "llama3.2:3b"
complex_model: str = "qwen2.5"
system_prompt: str = "You are SupportBot for GlobalShop e-commerce. Be polite and concise."
INTENT_MAP: dict = field(default_factory=lambda: {
"faq": "simple",
"order_status": "simple",
"shipping": "simple",
"product_info": "complex",
"comparison": "complex",
"complaint": "complex",
"refund": "complex"
})
def classify_intent(self, question: str) -> tuple[str, str]:
"""Classify intent and determine complexity."""
response = ollama.chat(
model=self.classifier,
messages=[{
"role": "user",
"content": f"""Classify intent (one word): {question}
Categories: faq, order_status, shipping, product_info, comparison, complaint, refund
Reply with just the category word."""
}],
stream=False,
options={"temperature": 0.0}
)
intent = response["message"]["content"].strip().lower()
for key in self.INTENT_MAP:
if key in intent:
return key, self.INTENT_MAP[key]
return "faq", "simple"
def answer(self, question: str) -> dict:
"""Route question and generate answer."""
start = time.time()
intent, complexity = self.classify_intent(question)
model = self.simple_model if complexity == "simple" else self.complex_model
response = ollama.chat(
model=model,
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": question}
],
stream=False,
options={"temperature": 0.3 if complexity == "simple" else 0.4}
)
elapsed = time.time() - start
return {
"question": question,
"intent": intent,
"complexity": complexity,
"model_used": model,
"answer": response["message"]["content"],
"latency_s": round(elapsed, 2)
}
def batch_process(self, questions: list[str]) -> list[dict]:
"""Process multiple questions and show routing stats."""
results = [self.answer(q) for q in questions]
simple_count = sum(1 for r in results if r["complexity"] == "simple")
avg_simple = sum(r["latency_s"] for r in results if r["complexity"] == "simple") / max(simple_count, 1)
complex_count = len(results) - simple_count
avg_complex = sum(r["latency_s"] for r in results if r["complexity"] == "complex") / max(complex_count, 1)
print(f"=== Routing Stats ===")
print(f"Simple: {simple_count}/{len(results)} (avg {avg_simple:.1f}s)")
print(f"Complex: {complex_count}/{len(results)} (avg {avg_complex:.1f}s)")
print(f"Cost saving: ~{simple_count * 60}% of queries use small model")
return results
# Usage
if __name__ == "__main__":
bot = SupportBotV4()
questions = [
"What is the return policy?",
"Where is my order #88765?",
"Do you ship internationally?",
"Compare the 3 headphone models in detail",
"I received a completely wrong item, this is unacceptable!",
"What are the warranty terms for electronics?",
"Analyze which laptop is best for a graphic designer under $1000",
]
results = bot.batch_process(questions)
for r in results:
print(f"\n[{r['complexity'].upper()}|{r['model_used']}] {r['intent']}")
print(f" Q: {r['question']}")
print(f" A: {r['answer'][:100]}... ({r['latency_s']}s)")
❓ 常见问题
Q 路由器分类不准怎么办？
A 用更好的分类器。小模型（3B）分类准确率约 85%，8B 约 92%。也可用关键词匹配预筛，模型只处理模糊情况。
Q 级联模式比直接用大模型好吗？
A 看场景。级联模式总延迟 = 各阶段之和，但每阶段更快。适合需要草稿+精炼的场景（写作、代码）。简单问答不需要级联。
Q 并行投票会 3 倍消耗 GPU 资源吗？
A 是的。3 个模型同时推理需要 3 倍 VRAM。可以顺序调用 3 个模型避免 OOM，但失去并行优势。
Q asyncio 并发和顺序调用速度差异大吗？
A 取决于 GPU。单 GPU 时 Ollama 内部排队，并发不加速。多 GPU 或 CPU+GPU 混合时，asyncio 可并发不同模型。
Q 如何评估路由器的分流效果？
A 统计分流比例、各类延迟、用户满意度。目标：80% 请求走小模型，平均延迟降低 50%+，满意度不下降。
Q 小模型和大模型的 System Prompt 要一样吗？
A 建议一致，保证用户体验统一。但可以针对不同复杂度微调——大模型的 Prompt 可以更细致。
📖 小节
- 三种编排模式：路由器（分流）、级联（流水线）、并行投票（高可靠）
- 路由器模式：3B 分类 + 8B 精答，80% 请求走小模型，平均延迟降 50%
- 级联模式：Draft(3B) → Refine(8B) → Format(3B)，适合内容创作
- asyncio 并发调用多模型，单 GPU 时需注意排队
- SupportBot V4 用路由器分流，简单问题 2s 复杂问题 5s
- 评估路由器看分流比例、延迟改善、用户满意度三指标
📝 作业
- 基础题（难度⭐）：实现一个简单的二路由器——FAQ 走小模型，其他走大模型，测试 5 个问题。
- 进阶题（难度⭐⭐）：实现三级联流水线——草稿→精炼→格式化，对比单模型直接输出的质量差异。
- 挑战题（难度⭐⭐⭐）：为 SupportBot V4 实现完整路由系统，包含意图分类、分流统计、延迟监控，并输出路由效果报告。
