マルチモデルシステム設計：単一のモデルでは不十分な場合
最もシンプルなパターンを選びましょう。
シングルモデルシステムはシンプルです。マルチモデルシステムは強力です。課題となるのはモデルの選択ではなく、それらをオーケストレーションするアーキテクチャの設計です。
マルチモデルシステムとは、単にモデルを増やすことではありません。正しいモデルを、正しいタスクに、正しいタイミングで適用することです。
アーキテクチャパターン
5つのパターンが、ほとんどのユースケースをカバーします。
| パターン | 複雑度 | 使用する場合 | トレードオフ |
|---|---|---|---|
| シングルモデル | 最低 | プロトタイピング、単純なタスク | 機能に制限がある |
| シーケンシャル | 低 | 複数ステップのワークフロー | レイテンシが高くなる |
| パラレル | 中 | 独立したタスク | コストが高くなる |
| 階層型 | 高 | 複雑な推論 | オーケストレーションが複雑 |
| アンサンブル | 最高 | 重要な意思決定 | コストが最も高い |
動作する最もシンプルなパターンを選択してください。複雑さは実在し、累積していきます。
シーケンシャルアーキテクチャ
モデルのチェーンを通じてタスクを処理し、各モデルが特定のステップを専門に行います。
パターン1: パイプライン
パイプラインパターン — 各モデルの出力が次のモデルに入力されます。
class ModelPipeline:
def __init__(self):
self.models = [
{"model": "qwen2.5-1.5b", "task": "classify"},
{"model": "qwen2.5-7b", "task": "extract"},
{"model": "qwen2.5-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 = "qwen2.5-1.5b"
self.specialists = {
"code": "qwen2.5-coder-7b",
"math": "qwen2.5-32b",
"creative": "claude-sonnet-4",
"general": "qwen2.5-7b",
}
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 = [
"qwen2.5-7b",
"qwen2.5-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 = [
"qwen2.5-7b",
"qwen2.5-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 = "qwen2.5-32b"
self.executors = {
"code": "qwen2.5-coder-7b",
"search": "qwen2.5-7b",
"math": "qwen2.5-7b",
}
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"], "qwen2.5-7b")
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 = "qwen2.5-32b"
self.workers = ["qwen2.5-7b", "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}")
スーパーバイザーがボトルネックとなります。計画、委任、レビューを行います。十分な速度があることを確認し、全体システムが遅延しないよう配慮してください。
アンサンブルアーキテクチャ
重要な意思決定のために複数のモデルを組み合わせます。
パターン1: 重み付きアンサンブル
重み付きアンサンブル — 各モデルの出力にスコアを付け、最高スコアのものを選択します。
class WeightedEnsemble:
def __init__(self):
self.models = {
"qwen2.5-32b": 0.5,
"claude-sonnet-4": 0.3,
"qwen2.5-7b": 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 = [
"qwen2.5-32b",
"claude-sonnet-4",
"qwen2.5-7b",
]
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("qwen2.5-32b", prompt)
閾値がコンセンサスの厳格さを制御します。0.7は3分の2の合意を意味します。より迅速な決定には閾値を下げ、より高い自信を得るには閾値を上げます。
マルチモデルシステムが適している場合
混合ワークロードがある場合、重要な意思決定に高品質を必要とする場合、またはコストやレイテンシを最適化する必要がある場合に、マルチモデルシステムは理にかなっています。
すべてのタスクが同様の複雑度である場合、プロトタイピング中である場合、または最適化よりもシンプルさが優先される場合は、マルチモデルシステムは適していません。
経験則として、1つのモデルから始めましょう。コスト、レイテンシ、品質といった実際の制約に直面した時点で、モデルを追加してください。必要な前に複雑なアーキテクチャを採用しないでください。
トレードオフ
| パターン | コスト | レイテンシ | 品質 | 複雑度 |
|---|---|---|---|---|
| シングルモデル | 最低 | 最低 | 変動 | 最低 |
| シーケンシャル | 中 | 高 | 高 | 中 |
| パラレル | 高 | 低 | 高 | 中 |
| 階層型 | 高 | 高 | 最高 | 高 |
| アンサンブル | 最高 | 中 | 最高 | 最高 |
すべてのパターンは何かをトレードオフします。あなたの制約に適合するものを選択してください。
関連リンク
- モデルルーティング戦略 — 機能ベース、コスト考慮、レイテンシ考慮のルーティング
- LLMシステムのコスト最適化 — トークン予算、フォールバックモデル、キャッシング
- 実践的なLLMガードレール — 入力検証、出力フィルタリング、安全性
- LLMアーキテクチャ — システム設計の柱: ルーティング、コスト、ガードレール、およびオーケストレーション
