マルチモデルシステム設計：単一モデルでは不十分な場合
「機能する最もシンプルなパターンを選びましょう。」
シングルモデルのシステムはシンプルです。マルチモデルのシステムは強力です。課題はモデルを選ぶことではなく、それらを調整するアーキテクチャを設計することにあります。
マルチモデルシステムとは、単に多くのモデルを持つことではありません。それは、適切なタイミングで適切なタスクに対して、適切なモデルを使用することです。
アーキテクチャパターン
主に以下の5つのパターンが、ほとんどのユースケースをカバーします。
| パターン | 複雑度 | 使用時期 | トレードオフ |
|---|---|---|---|
| シングルモデル | 最低 | プロトタイピング、単純なタスク | 能力が限定的 |
| シーケンシャル（逐次処理） | 低 | マルチステップのワークフロー | レイテンシが高い |
| パラレル（並列処理） | 中 | 独立したタスク | コストが高い |
| ヒエラルキー（階層型） | 高 | 複雑な推論 | 調整（オーケストレーション）が複雑 |
| アンサンブル | 最高 | 重要な決定 | コストが最高 |
機能する最もシンプルなものを選びましょう。複雑さは実在し、累積していきます。
シーケンシャルアーキテクチャ
モデルのチェーンを通じてタスクを処理し、それぞれが特定のステップに特化します。
パターン1: パイプライン
パイプラインパターン — 各モデルの出力が次のモデルに入力されます:
class ModelPipeline:
def __init__(self):
self.models = [
{"model": "qwen3-1.7b", "task": "classify"},
{"model": "qwen3-8b", "task": "extract"},
{"model": "qwen3-32b", "task": "reason"},
]
def process(self, input: str) -> str:
current = input
for model_config in self.models:
current = self.call_model(
model_config["model"],
self.create_prompt(model_config["task"], current)
)
return current
レイテンシは累積します。3つのモデルを逐次実行する場合、レイテンシは3倍になります。各ステップが実際に異なるモデルを必要とする場合にのみ、これを使用してください。
パターン2: ラーター
ルーティングパターン — タスクを分類し、専門モデルにルーティングします:
class ModelRouter:
def __init__(self):
self.classifier = "qwen3-1.7b"
self.specialists = {
"code": "qwen2.5-coder-7b",
"math": "qwen3-32b",
"creative": "claude-sonnet-4",
"general": "qwen3-8b",
}
def route(self, prompt: str) -> str:
task_type = self.classify(prompt)
model = self.specialists.get(task_type, self.specialists["general"])
return self.call_model(model, prompt)
分類器は弱点となります。誤分類すると、間違ったモデルにルーティングされ、品質が低下します。十分な性能を持つ分類器を使用してください。カテゴリが明確であれば、小さなモデルでも機能します。
パラレルアーキテクチャ
独立したタスクを同時に処理します。
パターン1: ファンアウト
ファンアウト — 同じプロンプトを複数のモデルで実行します:
import asyncio
class ModelFanOut:
def __init__(self):
self.models = [
"qwen3-8b",
"qwen3-32b",
"claude-sonnet-4",
]
async def process(self, prompt: str) -> list[str]:
tasks = [self.call_model(model, prompt) for model in self.models]
return await asyncio.gather(*tasks)
比較、A/Bテスト、または最適な出力を選択したい場合に有用です。コストは高くなりますが、重要な決定において品質向上に見合う価値があります。
パターン2: 投票
投票 — コンセンサス（合意）によって出力を結合します:
class ModelVoting:
def __init__(self):
self.models = [
"qwen3-8b",
"qwen3-32b",
"claude-sonnet-4",
]
def vote(self, prompt: str) -> str:
responses = [self.call_model(model, prompt) for model in self.models]
from collections import Counter
votes = Counter(responses)
return votes.most_common(1)[0][0]
多数決は分類タスクに有効です。生成タスクでは難しく、完全な一致ではなく意味的な類似性が必要です。
ヒエラルキーアーキテクチャ
異なる抽象レベルでモデルを使用します。
パターン1: プランナー-エグゼキューター
プランナー-エグゼキューター — 強力なモデルが計画し、小さなモデルが実行します:
class PlannerExecutor:
def __init__(self):
self.planner = "qwen3-32b"
self.executors = {
"code": "qwen2.5-coder-7b",
"search": "qwen3-8b",
"math": "qwen3-8b",
}
def process(self, task: str) -> str:
plan = self.call_model(self.planner, f"Plan: {task}")
results = []
for step in self.parse_plan(plan):
executor = self.executors.get(step["type"], "qwen3-8b")
result = self.call_model(executor, step["prompt"])
results.append(result)
return self.call_model(self.planner, f"Synthesize: {results}")
プランナーが重い処理を行います。エグゼキューターが特定のタスクを処理します。このパターンは、計画ステップが高コストだが、実行ステップが低コストな場合に有効です。
パターン2: サーパーバイザー-ワーカー
スーパーバイザー-ワーカー — スーパーバイザーが委任し、レビューします:
class SupervisorWorker:
def __init__(self):
self.supervisor = "qwen3-32b"
self.workers = ["qwen3-8b", "qwen2.5-coder-7b"]
def process(self, task: str) -> str:
assignments = self.call_model(self.supervisor, f"Assign: {task}")
results = []
for assignment in self.parse_assignments(assignments):
result = self.call_model(
assignment["worker"], assignment["task"]
)
results.append(result)
return self.call_model(self.supervisor, f"Review: {results}")
スーパーバイザーがボトルネックとなります。計画、委任、レビューを行います。十分に高速であることを確認 otherwise、システム全体が遅くなります。
アンサンブルアーキテクチャ
重要な決定のために複数のモデルを結合します。
パターン1: 加重アンサンブル
加重アンサンブル — 各モデルの出力にスコアを付け、最高スコアを選択します:
class WeightedEnsemble:
def __init__(self):
self.models = {
"qwen3-32b": 0.5,
"claude-sonnet-4": 0.3,
"qwen3-8b": 0.2,
}
def decide(self, prompt: str) -> str:
responses = {
model: self.call_model(model, prompt)
for model in self.models
}
scores = {}
for model, response in responses.items():
score = self.evaluate(response) * self.models[model]
scores[response] = scores.get(response, 0) + score
return max(scores, key=scores.get)
重みは各モデルへの信頼度を表します。ベンチマークではなく、実際の性能に基づいて調整してください。
パターン2: コンセンサスアンサンブル
コンセンサスアンサンブル — 合意を必要とし、合意がない場合はエスカレーションします:
class ConsensusEnsemble:
def __init__(self, threshold: float = 0.7):
self.threshold = threshold
self.models = [
"qwen3-32b",
"claude-sonnet-4",
"qwen3-8b",
]
def decide(self, prompt: str) -> str:
responses = [
self.call_model(model, prompt)
for model in self.models
]
from collections import Counter
votes = Counter(responses)
max_votes = max(votes.values())
if max_votes / len(self.models) >= self.threshold:
return votes.most_common(1)[0][0]
return self.call_model("qwen3-32b", prompt)
閾値はコンセンサスの厳格さを制御します。0.7は3分の2の合意を意味します。より迅速な決定のために閾値を下げ、より高い信頼性のために閾値を上げます。
マルチモデルシステムが有効な場合
マルチモデルシステムは、ワークロードが混合している場合、重要な決定に高い品質が必要な場合、またはコストやレイテンシを最適化する場合に有効です。
すべてのタスクの複雑さが類似している場合、プロトタイピング段階である場合、または最適化よりもシンプルさが優先される場合は、有効ではありません。
経験則として、1つのモデルから始めます。コスト、レイテンシ、品質といった実際の制約に直面した時点で、モデルを追加してください。必要になる前に複雑なアーキテクチャを設計しないでください。
トレードオフ
| パターン | コスト | レイテンシ | 品質 | 複雑度 |
|---|---|---|---|---|
| シングルモデル | 最低 | 最低 | 変動あり | 最低 |
| シーケンシャル | 中 | 高 | 高 | 中 |
| パラレル | 高 | 低 | 高 | 中 |
| ヒエラルキー | 高 | 高 | 最高 | 高 |
| アンサンブル | 最高 | 中 | 最高 | 最高 |
すべてのパターンは何かを犠牲にします。あなたの制約に合うものを選びましょう。
関連項目
- モデルルーティング戦略 — 機能ベース、コスト考慮、レイテンシ考慮のルーティング
- LLMシステムのコスト最適化 — トークン予算、フォールバックモデル、キャッシング
- 実践的なLLMガードレール — 入力検証、出力フィルタリング、安全性
- LLMアーキテクチャ — システム設計の柱: ルーティング、コスト、ガードレール、およびオーケストレーション
