[小ネタ] Strands Agents に Model Routing が追加されていたので試してみた

uv init --python 3.13 strands-model-routing
cd strands-model-routing
uv add strands-agents

import logging

from strands import Agent
from strands.models import BedrockModel, ModelRouter

logging.basicConfig(format="%(name)s | %(message)s")
logging.getLogger("strands.models.routing").setLevel(logging.INFO)

# わざと存在しないモデルIDを1番手にする
primary_model = BedrockModel(model_id="jp.anthropic.claude-not-exist-v1:0")
backup_model = BedrockModel(model_id="jp.anthropic.claude-haiku-4-5-20251001-v1:0")

router = ModelRouter(models=[primary_model, backup_model], max_switches=1)
agent = Agent(model=router, callback_handler=None)

result = agent("日本の首都を一言で答えてください。")
print(result)

uv run fallback.py

strands.models.routing.router | strategy=<FallbackStrategy>, candidate=<BedrockModel/jp.anthropic.claude-not-exist-v1:0>, model=<BedrockModel/jp.anthropic.claude-not-exist-v1:0> | candidate selected
strands.models.routing.router | from_candidate=<BedrockModel/jp.anthropic.claude-not-exist-v1:0>, to_candidate=<BedrockModel/jp.anthropic.claude-haiku-4-5-20251001-v1:0>, error=<ValidationException> | model call failed, switching candidate
東京

import logging

from strands import Agent
from strands.models import BedrockModel, ClassifierStrategy, ModelRouter, RoutingCandidate

logging.basicConfig(format="%(name)s | %(message)s")
logging.getLogger("strands.models.routing").setLevel(logging.INFO)

# 振り分けを判断するモデル
classifier_model = BedrockModel(
    model_id="jp.anthropic.claude-haiku-4-5-20251001-v1:0",
    max_tokens=64,
    streaming=False,
    temperature=0,
)

router = ModelRouter(
    models=[
        RoutingCandidate(
            BedrockModel(model_id="jp.anthropic.claude-haiku-4-5-20251001-v1:0"),
            name="fast",
            description="あいさつ、短い質問、簡単な翻訳など、すぐ答えられる依頼",
            metadata={"cost": "low", "latency": "low"},
        ),
        RoutingCandidate(
            BedrockModel(model_id="jp.anthropic.claude-sonnet-4-6"),
            name="strong",
            description="複数の条件を踏まえた設計、比較検討、デバッグなど、考える量が多い依頼",
            metadata={"cost": "high", "latency": "medium"},
        ),
    ],
    strategy=ClassifierStrategy(classifier_model),
)

requests = [
    "「おはようございます」を英語にしてください。",
    "マルチリージョンのアクティブ/アクティブ構成で、DynamoDBの書き込み競合をどう設計すべきか比較してください。",
]
for request in requests:
    print(f"\n### {request}")
    agent = Agent(model=router, callback_handler=None)
    result = agent(request)
    print(str(result)[:80].replace("\n", " "), "...")

uv run classifier.py

### 「おはようございます」を英語にしてください。
strands.models.routing.router | strategy=<ClassifierStrategy>, candidate=<fast>, model=<BedrockModel/jp.anthropic.claude-haiku-4-5-20251001-v1:0> | candidate selected
「おはようございます」の英語訳は：  **Good morning.**  より丁寧な表現としては： - **Good morning, sir/madam.* ...

### マルチリージョンのアクティブ/アクティブ構成で、DynamoDBの書き込み競合をどう設計すべきか比較してください。
strands.models.routing.router | strategy=<ClassifierStrategy>, candidate=<strong>, model=<BedrockModel/jp.anthropic.claude-sonnet-4-6> | candidate selected
# DynamoDB マルチリージョン アクティブ/アクティブ 書き込み競合設計  ## 競合が発生するメカニズム  ``` Region A (ap-nort ...

class ClassifyThenFallback:
    def __init__(self, classifier: ClassifierStrategy) -> None:
        self._classifier = classifier
        self._fallback = FallbackStrategy()

    async def select(self, context: RoutingContext, **kwargs: Any) -> RoutingCandidate | None:
        # 最初の選択は分類、失敗後の選択はフォールバックに任せる
        if not context.attempts:
            return await self._classifier.select(context, **kwargs)
        return await self._fallback.select(context, **kwargs)

uv run custom.py

strands.models.routing.router | strategy=<ClassifyThenFallback>, candidate=<strong>, model=<BedrockModel/jp.anthropic.claude-not-exist-v1:0> | candidate selected
strands.models.routing.router | from_candidate=<strong>, to_candidate=<fast>, error=<ValidationException> | model call failed, switching candidate
# DynamoDBマルチリージョン構成での書き込み競合設計  マルチリージョンのアクティブ/アクティブ構成における書き込み競合対策を、複数のアプローチで比較し ...