AWSクレジットが使える！ BedrockのKimi K2を調教して、StrandsのClaudeコストを節約しよう

この記事は、投稿主のみのるん氏がAIエージェントを構築する過程で遭遇したトラブルと解決策を、一緒に奮闘した私Claude Codeがまとめたものです。

AI生成ブログのアンチとして有名なみのるんが、一体どんな記事をClaudeに書かせたのか？ぜひ楽しんで読んでくださいね。

きっかけとなった個人開発アプリ「パワポ作るマン」

みのるんは普段、AWSのBedrock AgentCoreとStrands Agentsを使ってAIエージェントを構築しています。
最近リリースした「パワポ作るマン」というWebアプリもその一つで、チャットで指示するだけでMarpベースのスライドを自動生成してくれるサービスです。

公開から1週間で260人超のユーザーが登録し、Claude APIの推論コストが1週間で1万5千円ほど発生してしまいました。そこで彼が目をつけたのが、Moonshot AIが提供するオープンウェイトモデル「Kimi K2 Thinking」です。

これはBedrockでサードパーティモデル扱いとなるClaudeシリーズと違い、みのるんがジャブジャブ余らせているAWSクレジットを利用できるため、Claudeからの乗り換えでコスト削減を図ろうと考えました。

ところが、実際に実装してみるとClaudeとは挙動が大きく異なり、数々のハマりポイントに遭遇することになりました。この記事では、その経験から得られたナレッジを共有します。

なぜKimiなのか？

AWSでAIエージェントを構築する場合、まず選択肢に上がるのがAnthropicのClaudeシリーズです。
最新版のSonnetモデルを選定すれば間違いないのですが、今回のような個人開発アプリをポケットマネーで運用していると、ユーザー数が多いこともあって毎月数万円の運用コストがかかりそうで、さすがに家計に響きます。

代替となる賢いモデルは、なかなかBedrockで他の選択肢がありません。特にオープンウェイトのモデルはどれも、ClaudeやGPTといったフロンティアモデルに全然太刀打ちできないレベルのものが、こと日本語においては多かったです。

しかしKimiがすごいのは卓越した日本語の賢さ。Claude Haikuに匹敵するぐらいの高度な日本語表現と知性を感じるモデルで、これはもしかするとClaudeの代替になるんじゃないかとみのるんが思い立ったのが、今回の奮闘のきっかけです。

両モデルのBedrockにおける違い

まず、基本的なスペック差を整理しておきます。

特徴	Claude Sonnet 4.5	Kimi K2 Thinking
入/出力コスト（1Mトークン）	＄3.0 / ＄15.0	＄0.6 / ＄2.5
クロスリージョン推論	対応	対応
ツール利用	対応	対応
拡張思考	対応	対応
プロンプトキャッシュ	対応	非対応

Kimi K2には「Thinking」という名前の通り、最近のLLMによくある推論（思考）過程を出力する機能があります。実装面ではこれが様々なトラブルの原因になります。

【ハマりポイント1】 プロンプトキャッシュが使えない

最初のハマりポイントは、Strands AgentsのBedrockModel設定です。Claudeで使っていた設定をそのまま流用したところ、いきなりAccessDeniedExceptionが発生しました。

from strands.models import BedrockModel

# ❌ NG: Kimi K2でキャッシュオプションを使用するとエラー
model = BedrockModel(
    model_id="moonshot.kimi-k2-thinking",
    cache_prompt="default",  # AccessDeniedException!
    cache_tools=True,        # AccessDeniedException!
)

原因はシンプルで、Kimi K2はプロンプトキャッシュに対応していないためです。以下のようにキャッシュオプションを外すことで解決します。

キャッシュが使えないのはそれはそれで痛いですが、KimiモデルであればAWSクレジット内で全て無料利用できるため、まぁ良いでしょう。

# ✅ OK: キャッシュオプションなしで設定
model = BedrockModel(
    model_id="moonshot.kimi-k2-thinking",
)

ClaudeとKimi K2を動的に切り替えるアプリケーションでは、以下のようにモデル固有の設定を分岐させると良いでしょう。

def _get_model_config(model_type: str = "claude") -> dict:
    if model_type == "kimi":
        return {
            "model_id": "moonshot.kimi-k2-thinking",
            # キャッシュ非対応なので指定しない
        }
    else:
        return {
            "model_id": "us.anthropic.claude-sonnet-4-5-20250929-v1:0",
            "cache_prompt": "default",
            "cache_tools": True,
        }

【ハマりポイント2】 Reasoningが別イベントとして出てくる

Kimi K2 Thinkingは、通常の data イベントに加えて reasoning イベント（思考プロセス）を発火します。これを適切に処理しないと、UIに思考プロセスが表示されてしまったり、逆に何も表示されなかったりします。

async for event in agent.stream_async(prompt):
    # 思考プロセスは無視して最終回答のみ表示する場合
    if event.get("reasoning"):
        continue

    if "data" in event:
        yield {"type": "text", "data": event["data"]}
    elif "result" in event:
        result = event["result"]
        if hasattr(result, 'message') and result.message:
            for content in getattr(result.message, 'content', []):
                # reasoningContentも無視
                if hasattr(content, 'reasoningContent'):
                    continue
                if hasattr(content, 'text') and content.text:
                    yield {"type": "text", "data": content.text}

思考プロセスを表示したい場合は reasoningText イベントを処理すればOKです。ただし、後述するトラブルの原因にもなるので、基本的には無視することをおすすめします。

【ハマりポイント3】 ツール名がぶっ壊れる

これが最も厄介な問題でした。ツール呼び出し時にツール名が破損し、内部トークンが混入してしまいます。。

例えば web_search を呼び出そうとすると、以下のようになってしまいます。

web_search<|tooluse_end|><|ASSISTANT|><|reasoning|>

この問題への対策として、ツール名の破損を検出してリトライするロジックを実装しました。

VALID_TOOL_NAMES = {"web_search", "output_slide", "generate_tweet_url"}
MAX_RETRY_COUNT = 5

def is_tool_name_corrupted(tool_name: str) -> bool:
    """ツール名が破損しているかチェック"""
    if not tool_name:
        return False
    # 有効なツール名でなければ破損
    if tool_name not in VALID_TOOL_NAMES:
        return True
    # 内部トークンが混入していたら破損
    if "<|" in tool_name or "tooluse_" in tool_name:
        return True
    return False

ストリーミング処理内でこの関数を使い、破損を検出したらリトライします。

retry_count = 0
while retry_count <= MAX_RETRY_COUNT:
    tool_name_corrupted = False
    stream = agent.stream_async(user_message)

    async for event in stream:
        if "current_tool_use" in event:
            tool_name = event["current_tool_use"].get("name", "")
            if is_tool_name_corrupted(tool_name):
                tool_name_corrupted = True
                continue  # 破損したツール呼び出しは無視

    if tool_name_corrupted:
        retry_count += 1
        agent.messages.clear()  # 破損した履歴をクリア
        continue
    break

ポイントは agent.messages.clear() で会話履歴をクリアしてからリトライすることです。破損した履歴を引き継ぐと、次も破損する可能性が高くなります。

【ハマりポイント4】 ツール呼び出しがreasoningに迷いこむ

さらに厄介なのが、ツール呼び出しが reasoningText（思考プロセス）内にテキストとして埋め込まれてしまうケースです。この場合、 current_tool_use イベントが発火せず、何も起きないまま終了します。

ログを見ると、こんな感じになっています。

{
  "reasoningText": {
    "text": "...Web検索します。 <|tool_calls_section_begin|> <|tool_call_begin|> functions.web_search:0 <|tool_call_argument_begin|> {\"query\": \"...\"} <|tool_call_end|> <|tool_calls_section_end|>"
  },
  "finish_reason": "end_turn"
}

本来はtool_useイベントとして発火すべきものが、テキストとして埋め込まれてしまっていますね。。

対策として、 reasoningText 内にツール呼び出しの痕跡がある場合もリトライ対象にします。

if hasattr(reasoning_text, 'text') and reasoning_text.text:
    text = reasoning_text.text
    # ツール呼び出しがテキストとして埋め込まれている場合を検出
    if "<|tool_call" in text or "functions.web_search" in text or "functions.output_slide" in text:
        tool_name_corrupted = True
        print(f"[WARN] Tool call found in reasoning text (retry ...)")

さらに、JSON引数からマークダウンを抽出するフォールバック関数も用意しておくと安心です。

def extract_marp_markdown_from_text(text: str) -> str | None:
    import re
    import json

    if not text:
        return None

    # ケース1: JSON引数内のマークダウンを抽出
    json_arg_pattern = r'<\|tool_call_argument_begin\|>\s*(\{[\s\S]*?\})\s*<\|tool_call_end\|>'
    json_match = re.search(json_arg_pattern, text)
    if json_match:
        try:
            data = json.loads(json_match.group(1))
            if isinstance(data, dict) and "markdown" in data:
                markdown = data["markdown"]
                if markdown and "marp: true" in markdown:
                    return markdown
        except json.JSONDecodeError:
            pass

    # ケース2: 直接的なマークダウンを抽出
    if "marp: true" in text:
        pattern = r'(---\s*\nmarp:\s*true[\s\S]*?)(?:<\|tool_call|$)'
        match = re.search(pattern, text)
        if match:
            markdown = match.group(1).strip()
            markdown = re.sub(r'<\|[^>]+\|>', '', markdown)
            return markdown

    return None

【ハマりポイント5】 ツールを呼ばずに通常テキストへ出力してしまう

もう一つ、Kimi K2特有の問題があります。ツールを呼ばずに、テキストストリーム（ data イベント）へツール引数を直接出力してしまうケースです。

この場合、今回のアプリではチャット欄にスライド用のマークダウンがそのまま表示されてしまい、Tool Useは発動しません。
対策として、テキストをバッファリングしてマークダウンを検出し、フォールバックとして抽出しました。

kimi_text_buffer = "" if model_type == "kimi" else None
kimi_skip_text = False

async for event in stream:
    if "data" in event:
        chunk = event["data"]
        if model_type == "kimi":
            kimi_text_buffer += chunk
            # マークダウン開始を検出したらテキスト送信をスキップ
            if not kimi_skip_text and "marp: true" in kimi_text_buffer.lower():
                kimi_skip_text = True
            if not kimi_skip_text:
                yield {"type": "text", "data": chunk}
        else:
            yield {"type": "text", "data": chunk}

# ストリーム終了後、バッファからマークダウンを抽出
if model_type == "kimi" and kimi_text_buffer:
    extracted = extract_marp_markdown_from_text(kimi_text_buffer)
    if extracted:
        fallback_markdown = extracted

今回のアプリではKimi K2の場合のみテキストをバッファリングし、Marp用の出力である marp: true を検出したら以降のテキスト送信をスキップすることで、チャット欄への混入を防ぎます。

おわりに

これだけ頑張って調教したKimi K2、皆さんもぜひ以下のアプリでモデル切り替えして体験してみてください。Claudeと比較するとさらに面白いです。

世間ではKimi K2.5 Thinkingが登場し、Claude超えか!? と大きな話題になっていますね。ちょうどClaude CodeのOSS代替であるOpenCodeが流行し、Anthropicモデルが速攻使えなくなったことによって、タイミングよくKimi K2.5にスポットライトが当たっています。

上記のような苦労も軽減してるかな？？ Bedrockにも早くK2.5来てくれ〜！