Ollama: マルチモデルオーケストレーション

マルチモデルオーケストレーションはAIのチームワーク——小モデルが先陣を切り、大モデルが背骨となり、それぞれが役割を果たす。

💡 ヒント: マルチモデルルーティング戦略の中核は「タスク別に割り当て」——簡単なFAQは3B小モデル（1秒応答、GPU消費低）、複雑な推論は8B大モデル（5秒応答、高精度）を使用。また「負荷別に割り当て」も可能——ピーク時は小モデルでより多くのリクエストを処理し、閑散時は大モデルに切替えて高品質化。リクエストの80%を小モデルにルーティングすると平均レイテンシを50%削減可能。

📋 前提条件: まず以下を習得していること

レッスン13: LangChain統合

1. 学べること

マルチモデルアーキテクチャパターン：Router / Cascade / Parallel Voting
Routerパターン：小モデルが分類 → 大モデルが深く回答
Cascadeパターン：下書き → 推敲 → フォーマットパイプライン
Python asyncioで複数Ollamaインスタンスへの同時呼び出し
AliceのSupportBot V4：Routerベースのトラフィック分割

2. SaaS起業家のリアルな事例

ℹ️ 情報: マルチモデルオーケストレーションでは、各モデルを独立してVRAMにロードする必要がある。2つのモデル（3B + 8B）を同時常駐させるには約10GB VRAMが必要。VRAMが不足する場合、Ollamaは非アクティブなモデルを自動的にアンロードするが、切替には5〜30秒の再ロード遅延が発生する。

(1) ペインポイント：大モデルが簡単な質問にリソースを浪費

AliceはSupportBotの質問の80%（「返品ポリシーは？」「配送状況確認方法は？」）が小モデルで回答可能だが、すべて大モデルで処理しているためGPUリソースを無駄にし、速度も遅いことに気づいた。

(2) ソリューション：Routerパターンでトラフィック分割

小モデル（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つのオーケストレーションパターン

⚠️ 注: Parallel Votingパターンは最高信頼性だが最高コストでもある——各リクエストで3〜5モデルの呼び出しが必要で、GPU計算量が倍増。重要な意思決定シナリオ（医療アドバイス、法的判断など）のみで使用；日常的なカスタマーサービスにはRouterパターンで十分。

💡 ヒント: Routerパターンは最もコスパの良いオーケストレーション方法——簡単なリクエストの80%は3Bモデル（1秒応答）を使用し、複雑なリクエストの20%のみ8Bモデルにルーティング。全体の平均応答時間は5秒から2秒に低下し、GPU使用率も60%削減。

(1) パターン比較

パターン	フロー	ユースケース	レイテンシ	コスト
Router	分類 → 振り分け	カスタマーサービスルーティング、タスク分類	低	低
Cascade	下書き → 推敲 → フォーマット	コード生成、コンテンツ作成	中	中
Parallel Voting	マルチモデル → 投票/統合	高信頼性意思決定	高	高

flowchart TD
    subgraph Router
        R1[入力] --> R2[分類器<br/>3Bモデル]
        R2 -->|簡単| R3[小モデル回答]
        R2 -->|複雑| R4[大モデル回答]
    end

    subgraph Cascade
        C1[入力] --> C2[下書き<br/>3Bモデル]
        C2 --> C3[推敲<br/>8Bモデル]
        C3 --> C4[フォーマット<br/>3Bモデル]
    end

    subgraph Voting
        V1[入力] --> V2[モデルA]
        V1 --> V3[モデルB]
        V1 --> V4[モデルC]
        V2 --> V5[多数決]
        V3 --> V5
        V4 --> V5
    end

4. Routerパターン

(1) Routerアーキテクチャ

sequenceDiagram
    participant U as ユーザー
    participant C as 分類器 (3B)
    participant S as 小モデル (3B)
    participant L as 大モデル (8B)
    U->>C: "返金したい"
    C-->>C: 意図: 返金 (簡単)
    C->>S: 小モデルにルーティング
    S-->>U: 返金ポリシー回答
    U->>C: "3製品の保証条件を比較して"
    C-->>C: 意図: 比較 (複雑)
    C->>L: 大モデルにルーティング
    L-->>U: 詳細比較

(2) 意図分類ラベル設計

ラベル	複雑さ	ルーティング先	質問例
faq	簡単	小モデル	「返品ポリシーは？」
order_status	簡単	小モデル	「注文#12345はどこ？」
product_info	中程度	中モデル	「このヘッドホンはBluetooth 5.3対応？」
comparison	複雑	大モデル	「3つのヘッドホンモデルを比較して」
complaint	複雑	大モデル	「届いた商品が説明と全く違う」

(3) ▶ サンプル：Routerパターン実装

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

# Function defined successfully

5. Cascadeパターン

(1) Cascadeアーキテクチャ詳細

ステージ	モデル	タスク	パラメータ
下書き	3B	素早い下書き	temperature=0.5
推敲	8B	深い推敲	temperature=0.3
フォーマット	3B	出力フォーマット	temperature=0.1

(2) ▶ サンプル：Cascadeコンテンツ生成

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. Parallel Votingとasyncio同時実行

(1) Parallel Votingアーキテクチャ

パターン	モデル数	決定方法	ユースケース
多数決	3以上	多数意見を採用	事実判断
加重平均	2以上	モデル品質で重み付け	スコアリングタスク
最良選択	2以上	最も詳細/信頼性の高いものを選択	自由回答形式

(2) ▶ サンプル：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(a[: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

# Function defined successfully

(3) ▶ サンプル：加重アンサンブル

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

# Function defined successfully

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 Cascadeパターンは大モデル直接使用より良い？

A シナリオによる。Cascadeの合計レイテンシ＝全ステージの合計だが、各ステージは高速。下書き＋推敲が必要なシナリオ（執筆、コード）に適する。単純Q&AにはCascade不要。

Q Parallel VotingはGPUリソースを3倍消費する？

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はより詳細にできる。

📖 まとめ

3つのオーケストレーションパターン：Router（トラフィック分割）、Cascade（パイプライン）、Parallel Voting（高信頼性）
Routerパターン：3B分類 + 8B深い回答、リクエストの80%を小モデル、平均レイテンシ50%削減
Cascadeパターン：下書き(3B) → 推敲(8B) → フォーマット(3B)、コンテンツ作成に適する
asyncioでマルチモデル同時呼び出し；単一GPUではキューイングに注意
SupportBot V4はRouterベース分割：簡単な質問2秒、複雑な質問5秒
ルーティング比率、レイテンシ改善、ユーザー満足度でRouter効果を評価

📝 練習問題

基本（難易度 ⭐）：シンプルな双方向ルーターを実装——FAQは小モデル、その他は大モデル。5つの質問でテストする。
中級（難易度 ⭐⭐）：3段階Cascadeパイプラインを実装——下書き → 推敲 → フォーマット。単一モデル直接出力と品質を比較する。
上級（難易度 ⭐⭐⭐）：SupportBot V4の完全なルーターシステムを実装し、意図分類、ルーティング統計、レイテンシモニタリングを含め、ルーティング効果レポートを出力する。

Web-Tutorial 技術チーム

複数の開発者によって共同維持されているプログラミングチュートリアルプラットフォーム。各チュートリアルは専門分野の開発者が執筆・レビューしています。正確で信頼性の高いコンテンツを目指しています — 問題を見つけた場合はお知らせください。

私たちについて