Claude APIでAIチャットボットを実装する — ストリーミング・会話履歴・コスト最適化まで

取り組みの背景：なぜ今、Claude APIでチャットボットを作るのか

市販のチャットボットツールは手軽ですが、自分のユースケースに合わせたカスタマイズには限界があります。Claude APIを直接叩けば、システムプロンプトの完全制御、社内データとの連携、コスト最適化、既存サービスへの組み込みが自由自在になります。

ℹ️

本記事のコードはすべて **Python 3.10以上** で動作確認済みです。APIキーの取得については [Claude API クイックスタート](/articles/api-sdk/api-quickstart) を参照してください。

STEP 1：最小構成のチャットボット

まず、最もシンプルな実装から始めましょう。

import anthropic
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
def chat(user_message: str) -> str:
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": user_message}
        ]
    )
    return message.content[0].text
 
# 実行
response = chat("Pythonで素数を判定する関数を書いてください")
print(response)

これだけで動きます。しかし会話の文脈が保持されないという致命的な問題があります。次のメッセージを送ると、前の会話を覚えていません。

STEP 2：会話履歴を管理する

Claude APIはステートレスなので、会話履歴をクライアント側で管理する必要があります。

import anthropic
from typing import List
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
class ChatSession:
    def __init__(self, system_prompt: str = ""):
        self.history: List[dict] = []
        self.system_prompt = system_prompt
 
    def send(self, user_message: str) -> str:
        # ユーザーメッセージを履歴に追加
        self.history.append({
            "role": "user",
            "content": user_message
        })
 
        # API呼び出し（システムプロンプト + 全履歴を送信）
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=self.system_prompt,
            messages=self.history
        )
 
        assistant_message = response.content[0].text
 
        # アシスタントの返答を履歴に追加
        self.history.append({
            "role": "assistant",
            "content": assistant_message
        })
 
        return assistant_message
 
    def clear(self):
        """会話をリセット"""
        self.history = []
 
# 使用例
session = ChatSession(
    system_prompt="あなたはPython専門のコードレビュアーです。"
                  "コードの問題点を指摘し、改善案を提示してください。"
)
 
print(session.send("このコードをレビューしてください：\ndef add(a, b): return a+b"))
print(session.send("では、型ヒントを追加したバージョンも書いてください"))  # 前の文脈を覚えている

ℹ️

`system` パラメータは会話履歴に含めず、毎回APIに渡すことで、どのターンでも一貫した振る舞いを保てます。

STEP 3：ストリーミング応答を実装する

長い応答を待たせると UX が悪化します。ストリーミングで文字を順次表示しましょう。

import anthropic
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
class StreamingChatSession:
    def __init__(self, system_prompt: str = ""):
        self.history = []
        self.system_prompt = system_prompt
 
    def send_stream(self, user_message: str):
        """ストリーミングでテキストをyieldする"""
        self.history.append({"role": "user", "content": user_message})
 
        full_response = ""
 
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=self.system_prompt,
            messages=self.history
        ) as stream:
            for text in stream.text_stream:
                full_response += text
                yield text  # テキストを逐次返す
 
        # 完全な応答を履歴に保存
        self.history.append({"role": "assistant", "content": full_response})
 
# CLIでの使用例
session = StreamingChatSession(system_prompt="あなたは親切なアシスタントです。")
 
while True:
    user_input = input("\nYou: ").strip()
    if user_input.lower() in ["quit", "exit", "終了"]:
        break
 
    print("Claude: ", end="", flush=True)
    for chunk in session.send_stream(user_input):
        print(chunk, end="", flush=True)
    print()  # 改行

STEP 4：FastAPIでWeb APIとして公開する

CLIだけでなく、WebアプリからもアクセスできるAPIとして実装します。

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import anthropic
import json
from typing import Optional
 
app = FastAPI()
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
# シンプルなインメモリセッション管理（本番ではRedis推奨）
sessions: dict[str, list] = {}
 
class ChatRequest(BaseModel):
    session_id: str
    message: str
    system_prompt: Optional[str] = "あなたは親切なアシスタントです。"
 
class ChatResponse(BaseModel):
    response: str
    session_id: str
 
@app.post("/chat")
async def chat(request: ChatRequest):
    """通常のチャットエンドポイント"""
    if request.session_id not in sessions:
        sessions[request.session_id] = []
 
    history = sessions[request.session_id]
    history.append({"role": "user", "content": request.message})
 
    try:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=request.system_prompt,
            messages=history
        )
 
        assistant_message = response.content[0].text
        history.append({"role": "assistant", "content": assistant_message})
 
        return ChatResponse(
            response=assistant_message,
            session_id=request.session_id
        )
    except anthropic.APIError as e:
        raise HTTPException(status_code=500, detail=str(e))
 
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    """Server-Sent Events でストリーミング"""
    if request.session_id not in sessions:
        sessions[request.session_id] = []
 
    history = sessions[request.session_id]
    history.append({"role": "user", "content": request.message})
 
    def generate():
        full_response = ""
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=request.system_prompt,
            messages=history
        ) as stream:
            for text in stream.text_stream:
                full_response += text
                # SSE形式で送信
                yield f"data: {json.dumps({'text': text})}\n\n"
 
        history.append({"role": "assistant", "content": full_response})
        yield f"data: {json.dumps({'done': True})}\n\n"
 
    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )
 
@app.delete("/session/{session_id}")
async def clear_session(session_id: str):
    """セッションをリセット"""
    sessions.pop(session_id, None)
    return {"status": "cleared"}

起動方法：

pip install fastapi uvicorn anthropic
uvicorn main:app --reload

STEP 5：会話履歴のコスト最適化

会話が長くなるほどトークン消費が増大します。主要な最適化テクニックを紹介します。

5-1：会話を要約してトリミング

def summarize_and_trim(
    client: anthropic.Anthropic,
    history: list,
    max_turns: int = 10
) -> list:
    """古い会話を要約して履歴を圧縮する"""
    if len(history) <= max_turns * 2:
        return history
 
    # 古い部分を要約
    old_history = history[:-max_turns * 2]
    recent_history = history[-max_turns * 2:]
 
    summary_response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 安価なモデルで要約
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"以下の会話を3文以内で要約してください：\n\n{json.dumps(old_history, ensure_ascii=False)}"
        }]
    )
 
    summary = summary_response.content[0].text
 
    # 要約をシステムメッセージとして先頭に挿入
    return [
        {"role": "user", "content": f"[会話の要約] {summary}"},
        {"role": "assistant", "content": "了解しました。前の会話の文脈を把握しています。"},
        *recent_history
    ]

5-2：Prompt Caching でAPI料金を削減

システムプロンプトが長い場合、キャッシュで最大90%のコスト削減が可能です。

# キャッシュを活用したシステムプロンプト
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "ここに長いシステムプロンプト（社内ドキュメントや仕様書など）...",
            "cache_control": {"type": "ephemeral"}  # キャッシュを有効化
        }
    ],
    messages=history
)

ℹ️

Prompt Caching の詳細は [Prompt Caching 完全ガイド](/articles/api-sdk/prompt-caching) をご覧ください。1000トークン以上のシステムプロンプトで効果が出ます。

5-3：用途に応じてモデルを使い分ける

| ユースケース | 推奨モデル | 理由 | |---|---|---| | 単純なQ&A | claude-haiku-4-5-20251001 | 高速・安価 | | コーディング支援 | claude-sonnet-4-6 | バランス良好 | | 複雑な分析・推論 | claude-opus-4-6 | 最高精度 | | 要約・分類 | claude-haiku-4-5-20251001 | コスト最適 |

def get_model_for_task(task_type: str) -> str:
    """タスクに応じてモデルを自動選択"""
    model_map = {
        "simple_qa": "claude-haiku-4-5-20251001",
        "coding": "claude-sonnet-4-6",
        "analysis": "claude-opus-4-6",
        "summarization": "claude-haiku-4-5-20251001",
    }
    return model_map.get(task_type, "claude-sonnet-4-6")

STEP 6：エラーハンドリングとリトライ

本番環境では適切なエラー処理が不可欠です。

import anthropic
import time
from typing import Optional
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
def safe_chat(
    messages: list,
    system: str = "",
    max_retries: int = 3,
    retry_delay: float = 1.0
) -> Optional[str]:
    """リトライ付きのAPI呼び出し"""
 
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=2048,
                system=system,
                messages=messages
            )
            return response.content[0].text
 
        except anthropic.RateLimitError:
            # レート制限：指数バックオフで待機
            wait_time = retry_delay * (2 ** attempt)
            print(f"レート制限に達しました。{wait_time}秒後にリトライ...")
            time.sleep(wait_time)
 
        except anthropic.APIStatusError as e:
            if e.status_code >= 500:
                # サーバーエラー：リトライ
                print(f"サーバーエラー ({e.status_code})。リトライ中...")
                time.sleep(retry_delay)
            else:
                # クライアントエラー（400番台）：リトライしない
                print(f"クライアントエラー: {e.message}")
                return None
 
        except anthropic.APIConnectionError:
            # 接続エラー：リトライ
            print("接続エラー。リトライ中...")
            time.sleep(retry_delay)
 
    print("最大リトライ回数に達しました")
    return None

STEP 7：完成形 — 実用的なチャットボットクラス

これまでの要素をまとめた、本番投入できるクラスです。

import anthropic
import json
import time
import uuid
from typing import Generator, Optional
 
class ProductionChatBot:
    """本番用チャットボットクラス"""
 
    def __init__(
        self,
        api_key: str,
        system_prompt: str = "あなたは親切なアシスタントです。",
        model: str = "claude-sonnet-4-6",
        max_tokens: int = 2048,
        max_history_turns: int = 20,
    ):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.system_prompt = system_prompt
        self.model = model
        self.max_tokens = max_tokens
        self.max_history_turns = max_history_turns
        self.history = []
        self.session_id = str(uuid.uuid4())
        self.total_tokens_used = 0
 
    def send(self, message: str) -> str:
        """メッセージを送信してレスポンスを返す"""
        self.history.append({"role": "user", "content": message})
 
        response = self.client.messages.create(
            model=self.model,
            max_tokens=self.max_tokens,
            system=self.system_prompt,
            messages=self.history
        )
 
        assistant_text = response.content[0].text
        self.history.append({"role": "assistant", "content": assistant_text})
 
        # トークン使用量を記録
        self.total_tokens_used += response.usage.input_tokens + response.usage.output_tokens
 
        # 履歴が長くなりすぎたらトリミング
        if len(self.history) > self.max_history_turns * 2:
            self.history = self.history[-(self.max_history_turns * 2):]
 
        return assistant_text
 
    def stream(self, message: str) -> Generator[str, None, None]:
        """ストリーミングでテキストをyield"""
        self.history.append({"role": "user", "content": message})
        full_response = ""
 
        with self.client.messages.stream(
            model=self.model,
            max_tokens=self.max_tokens,
            system=self.system_prompt,
            messages=self.history
        ) as stream:
            for text in stream.text_stream:
                full_response += text
                yield text
 
        self.history.append({"role": "assistant", "content": full_response})
 
    def get_stats(self) -> dict:
        """使用統計を返す"""
        return {
            "session_id": self.session_id,
            "turns": len(self.history) // 2,
            "total_tokens": self.total_tokens_used,
            "estimated_cost_usd": self.total_tokens_used / 1_000_000 * 3.0  # Sonnet概算
        }
 
    def reset(self):
        """会話をリセット"""
        self.history = []
        self.total_tokens_used = 0
 
# 使用例
bot = ProductionChatBot(
    api_key="YOUR_API_KEY",
    system_prompt="""あなたはプロのPython開発者です。
    コードの品質、可読性、パフォーマンスに強いこだわりを持ち、
    ベストプラクティスに基づいた実装を提案してください。"""
)
 
# ストリーミング応答
print("Claude: ", end="")
for chunk in bot.stream("FastAPIでCRUDアプリを作るベストプラクティスを教えてください"):
    print(chunk, end="", flush=True)
print()
 
print(f"\n📊 統計: {bot.get_stats()}")

まとめと次のステップ

STEP 1-2: 基本実装と会話履歴管理
STEP 3: ストリーミングによるUX改善
STEP 4: FastAPIでのWeb API化
STEP 5: コスト最適化テクニック
STEP 6: 本番向けエラーハンドリング
STEP 7: すべてを統合した実用クラス

次のステップとして、以下の記事も参考にしてください：

ツールユース完全ガイド — チャットボットに検索や計算機能を追加する
マルチモーダル入力ガイド — 画像を理解するチャットボット
Prompt Cachingで高速化 — 大規模システムへの展開

ℹ️

**実装したコードを共有しませんか？** X (Twitter) で **#ClaudeLab** タグをつけてポストしていただくと嬉しいです！

Claude APIでAIチャットボットを実装する — ストリーミング・会話履歴・コスト最適化まで

取り組みの背景：なぜ今、Claude APIでチャットボットを作るのか

STEP 1：最小構成のチャットボット

STEP 2：会話履歴を管理する

STEP 3：ストリーミング応答を実装する

STEP 4：FastAPIでWeb APIとして公開する

STEP 5：会話履歴のコスト最適化

5-1：会話を要約してトリミング

5-2：Prompt Caching でAPI料金を削減

5-3：用途に応じてモデルを使い分ける

STEP 6：エラーハンドリングとリトライ

STEP 7：完成形 — 実用的なチャットボットクラス

まとめと次のステップ

お読みいただきありがとうございます

関連記事