OpusがSonnetを指揮する2モデル協調パターン｜コスト30%削減のPython実装

Advisor+Executorパターンとは

全量Opusの問題:
  - 入力$15/1M（Sonnetの5倍）
  - 単純な実装タスクにも高コストを払う

全量Sonnetの問題:
  - 複雑な設計判断・アーキテクチャ決定の精度が落ちる

解決策: Opus（Advisor）+ Sonnet（Executor）
  1. Opus: タスクを分析・計画・指示書を作成（少ないトークン）
  2. Sonnet: 指示書に従って実装（多いトークン、低コスト）

基本実装

# advisor_executor.py
import anthropic
import os

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

def advisor_plan(task: str) -> str:
    """Opus 5 が計画・設計判断を行う（Advisor役）"""
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=1024,
        system="""あなたはシニアエンジニアとして、タスクを分析して実装計画を立てます。
出力形式:
1. アーキテクチャの判断（何を・どのように作るか）
2. 実装者への具体的な指示（箇条書き、コーディングレベルまで詳細に）
3. 注意事項（セキュリティ・パフォーマンス・エッジケース）""",
        messages=[{"role": "user", "content": f"タスク: {task}"}]
    )
    return response.content[0].text

def executor_implement(plan: str, task: str) -> str:
    """Sonnet 4.5 が計画に従って実装する（Executor役）"""
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=4096,
        system="あなたはシニアエンジニアの指示に従って、高品質なコードを実装するエンジニアです。",
        messages=[{"role": "user", "content": f"""
## 元のタスク
{task}

## シニアエンジニアからの実装指示
{plan}

上記の指示に従って、実際のコードを実装してください。
"""}]
    )
    return response.content[0].text

def advisor_executor(task: str) -> dict:
    """Advisor（Opus）→Executor（Sonnet）の2段階処理"""
    import time
    
    # Step 1: Opusが計画を立てる
    t1 = time.time()
    plan = advisor_plan(task)
    planning_time = time.time() - t1
    
    # Step 2: Sonnetが実装する
    t2 = time.time()
    implementation = executor_implement(plan, task)
    implementation_time = time.time() - t2
    
    return {
        "plan": plan,
        "implementation": implementation,
        "planning_time_s": round(planning_time, 1),
        "implementation_time_s": round(implementation_time, 1),
    }

# 使用例
result = advisor_executor(
    "FastAPIでユーザー認証エンドポイント（JWT）を実装して。"
    "パスワードはbcryptでハッシュ化。リフレッシュトークンも必要。"
)
print("=== 計画 ===")
print(result["plan"])
print("\n=== 実装 ===")
print(result["implementation"])

サブタスクを並列実行して高速化

# parallel_executor.py
import asyncio
import anthropic
import json

client = anthropic.Anthropic()

async def opus_decompose(task: str) -> list[str]:
    """Opusがタスクを独立したサブタスクに分解する"""
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=512,
        messages=[{"role": "user", "content": f"""
以下のタスクを、互いに独立して並列実装できるサブタスクに分解してください。
JSON配列で返してください: ["サブタスク1", "サブタスク2", ...]

タスク: {task}
"""}]
    )
    text = response.content[0].text
    # JSON配列を抽出
    start = text.find("[")
    end = text.rfind("]") + 1
    return json.loads(text[start:end])

async def sonnet_execute(subtask: str) -> str:
    """Sonnetが単一サブタスクを実装する"""
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": subtask}]
    )
    return response.content[0].text

async def parallel_executor(task: str) -> dict:
    """Opusで分解 → Sonnetで並列実装"""
    # Step 1: 分解
    subtasks = await opus_decompose(task)
    print(f"サブタスク数: {len(subtasks)}")
    
    # Step 2: 並列実装
    results = await asyncio.gather(*[
        sonnet_execute(st) for st in subtasks
    ])
    
    return dict(zip(subtasks, results))

# 使用例
async def main():
    result = await parallel_executor(
        "ECサイトのバックエンドAPIを作成: "
        "商品一覧・商品詳細・カート操作・注文処理の4エンドポイント"
    )
    for task, code in result.items():
        print(f"\n--- {task[:50]} ---")
        print(code[:300])

asyncio.run(main())

コスト比較：全量Opus vs Advisor+Executor

# benchmark.py
# 同じタスクを3パターンで実行してコストを比較

def cost_usd(input_tokens: int, output_tokens: int, model: str) -> float:
    prices = {
        "opus": (15.00, 75.00),
        "sonnet": (3.00, 15.00),
    }
    inp_price, out_price = prices[model]
    return (input_tokens * inp_price + output_tokens * out_price) / 1_000_000

TASK = "Pythonでシングルトンパターンをスレッドセーフに実装して"

# パターン1: 全量Opus
r1 = client.messages.create(model="claude-opus-5", max_tokens=2048,
    messages=[{"role": "user", "content": TASK}])
cost1 = cost_usd(r1.usage.input_tokens, r1.usage.output_tokens, "opus")

# パターン2: 全量Sonnet
r2 = client.messages.create(model="claude-sonnet-4-5", max_tokens=2048,
    messages=[{"role": "user", "content": TASK}])
cost2 = cost_usd(r2.usage.input_tokens, r2.usage.output_tokens, "sonnet")

# パターン3: Advisor+Executor
plan = advisor_plan(TASK)
impl = executor_implement(plan, TASK)
# Opusは計画フェーズのみ（短い）、Sonnetは実装フェーズ
r3_plan = client.messages.create(model="claude-opus-5", max_tokens=512,
    messages=[{"role": "user", "content": TASK}])
cost3 = cost_usd(r3_plan.usage.input_tokens, r3_plan.usage.output_tokens, "opus")

print(f"全量Opus:        ${cost1:.5f}")
print(f"全量Sonnet:      ${cost2:.5f}")
print(f"Advisor+Executor: 〜${cost1*0.4:.5f}（推定）")  # 計画は短い

あわせて読みたい

• Claude Codeが失敗する5つのパターン｜生成AIコーディングの落とし穴と品質ゲート
• 【2026年版】Claude Opus・Sonnet・Haiku 違いと選び方
• Claude Codeで午前中に機能完成する方法｜16のロール活用で爆速開発