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 完全ガイドをご覧ください。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 8：LINE・Telegram・Discord に接続する — プラットフォーム固有の実装差分

これまではターミナルや自分の Web API でチャットボットを動かしてきました。実際に人に使ってもらうとなると、LINE・Telegram・Discord のようなメッセージングプラットフォームに載せる場面が多いはずです。

私自身、個人開発で作ったチャットボットを最初に LINE へ載せたとき、いちばん面食らったのが reply token の寿命でした。

どちらも「プラットフォームが Webhook であなたのサーバーに POST する → 署名を検証する → Claude に投げる → 返信する」という骨格は同じです。違いはその細部に出ます。ここでつまずきやすい差分を整理しておきます。

LINE：reply token の寿命が落とし穴

LINE Messaging API で最初にハマるのが、reply token の扱いです。reply token は 1 回しか使えず、有効期限も短い（発行から 1 分ほど）。Claude の応答に数秒かかる程度なら間に合いますが、長文生成やツール呼び出しを挟むと失効してしまうことがあります。

実装上は「まず loading アニメーションを表示して reply token を温存し、応答が固まってから push message で送る」という二段構えが安定します。

署名検証も必須です。X-Line-Signature ヘッダーを検証しないと、誰でもあなたのエンドポイントに偽のイベントを送れてしまいます。

import hmac
import hashlib
import base64
import os
import requests
from fastapi import FastAPI, Request, HTTPException
 
app = FastAPI()
LINE_CHANNEL_SECRET = os.environ["LINE_CHANNEL_SECRET"]
LINE_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
 
def verify_line_signature(body: bytes, signature: str) -> bool:
    """X-Line-Signature を HMAC-SHA256 で検証する"""
    digest = hmac.new(
        LINE_CHANNEL_SECRET.encode("utf-8"), body, hashlib.sha256
    ).digest()
    expected = base64.b64encode(digest).decode("utf-8")
    return hmac.compare_digest(expected, signature)
 
def line_push(user_id: str, text: str) -> None:
    """reply token に依存しない push message で送る"""
    requests.post(
        "https://api.line.me/v2/bot/message/push",
        headers={"Authorization": f"Bearer {LINE_ACCESS_TOKEN}"},
        json={"to": user_id, "messages": [{"type": "text", "text": text}]},
        timeout=10,
    )
 
@app.post("/line/webhook")
async def line_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Line-Signature", "")
    if not verify_line_signature(body, signature):
        raise HTTPException(status_code=403, detail="signature mismatch")
 
    events = (await request.json()).get("events", [])
    for event in events:
        if event.get("type") == "message" and event["message"]["type"] == "text":
            user_id = event["source"]["userId"]
            user_text = event["message"]["text"]
            # ユーザーごとに履歴を持つ（STEP 7 の ProductionChatBot を流用）
            reply = get_bot_for(user_id).send(user_text)
            line_push(user_id, reply)
    return {"ok": True}

get_bot_for(user_id) は、ユーザー ID をキーに ProductionChatBot のインスタンスを引く想定です。プロセス内の辞書で持つと再起動で消えるので、本番では会話履歴を Redis などの外部ストアに逃がしておくと安心です。

Telegram：webhook か long polling かを使い分ける

Telegram には受信方法が 2 通りあります。getUpdates をループで叩く long polling と、公開エンドポイントに POST させる webhook です。

開発中は polling が圧倒的に楽です。公開 URL も HTTPS 証明書も要らず、ローカルでそのまま動きます。一方、本番でサーバーを常時起動しておくなら webhook のほうが無駄がありません。webhook を使うときは、登録時に secret_token を指定し、受信側で X-Telegram-Bot-Api-Secret-Token ヘッダーを突き合わせて検証します。

応答に時間がかかる Claude では、sendChatAction で「入力中…」を出しておくと体感がぐっと良くなります。

import os
import requests
 
TG_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
API = f"https://api.telegram.org/bot{TG_TOKEN}"
 
def tg_typing(chat_id: int) -> None:
    requests.post(f"{API}/sendChatAction",
                  json={"chat_id": chat_id, "action": "typing"}, timeout=10)
 
def tg_send(chat_id: int, text: str) -> None:
    requests.post(f"{API}/sendMessage",
                  json={"chat_id": chat_id, "text": text}, timeout=10)
 
def run_polling():
    """開発向け: long polling で受信する"""
    offset = None
    while True:
        resp = requests.get(f"{API}/getUpdates",
                            params={"timeout": 30, "offset": offset}, timeout=40)
        for update in resp.json().get("result", []):
            offset = update["update_id"] + 1
            msg = update.get("message")
            if not msg or "text" not in msg:
                continue
            chat_id = msg["chat"]["id"]
            tg_typing(chat_id)
            reply = get_bot_for(str(chat_id)).send(msg["text"])
            tg_send(chat_id, reply)

Discord：3 秒以内に「受け取った」と返す

LINE と Telegram はどちらも Webhook 方式でした。Discord も同じつもりで書き始めたところ、最初に作ったスラッシュコマンドがことごとく無反応で終わり、しばらく手が止まりました。

discord.py で組む場合、イベントは Webhook ではなく 常時接続の WebSocket（Gateway） 経由で届きます。署名検証は不要です。その代わりに別の締め切りがあります。スラッシュコマンドは 3 秒以内に何らかの応答を返さないと、その対話そのものが無効になる。

Claude の生成には数秒かかります。素直に「質問を受け取る → Claude を呼ぶ → 返信する」と書くと、ほぼ確実にこの 3 秒を踏み抜きます。

鍵は ctx.defer() です。最初にこれを呼んでおくと「処理中」の状態を Discord に伝えられ、実際の返信は followup.send() で後から送れます。LINE で reply token を温存してから push message に切り替える発想と、狙いはほとんど同じです。

もう一つの制約が 1 メッセージ 2000 文字 の上限です。Claude は放っておくと平気で超えてきますので、システムプロンプトで長さを促したうえで、送信側でも切り詰めておきます。

import discord
from discord import option
from discord.ext import commands
 
SYSTEM_PROMPT = """あなたは Discord サーバーに常駐する AI アシスタントです。
回答は簡潔にまとめ、1 メッセージ 2000 文字の上限を意識してください。
コードは必ずコードブロックで囲んでください。"""
 
class ClaudeCog(commands.Cog):
    def __init__(self, bot: commands.Bot):
        self.bot = bot
 
    @discord.slash_command(name="ask", description="Claude に質問します")
    @option("question", description="質問を入力してください")
    async def ask(self, ctx: discord.ApplicationContext, question: str):
        # 3 秒の締め切りをここで回避する。先に「処理中」を返しておく
        await ctx.defer()
 
        # STEP 7 の ProductionChatBot をユーザー単位で引く
        reply = get_bot_for(str(ctx.author.id)).send(question)
 
        # 2000 文字の上限に収める
        if len(reply) > 1900:
            reply = reply[:1900] + "\n\n…（長いため省略しました）"
 
        await ctx.followup.send(reply)
 
def setup(bot: commands.Bot):
    bot.add_cog(ClaudeCog(bot))

起動側では Message Content Intent を明示的に有効にします。コード側と Developer Portal 側の設定が両方揃っていないと、Bot は起動時点で例外を吐いて落ちます。

intents = discord.Intents.default()
intents.message_content = True  # Developer Portal 側でも ON にしておく
 
bot = commands.Bot(command_prefix="!", intents=intents)

Discord で私が実際に踏んだつまずきを、3 つだけ残しておきます。

症状	原因	対処
起動時に `PrivilegedIntentsRequired`	Developer Portal で Message Content Intent が OFF	Bot タブ → Privileged Gateway Intents → Message Content Intent を ON
スラッシュコマンドが候補に出てこない	招待 URL に `applications.commands` スコープが無い	OAuth2 URL Generator でスコープを追加し、招待し直す
コマンドを直したのに反映されない	グローバルコマンドは反映に最大 1 時間かかる	開発中はギルドコマンド（特定サーバー限定）で登録して即時反映させる

3 つ目は地味ですが、私はここで一番時間を溶かしました。コードが悪いと思い込んで書き直した先で、実はただ反映を待っていただけ、という徒労です。開発中はギルド限定で登録する。これだけで検証のテンポが変わります。

3 つのプラットフォームを並べて見る

こうして並べると、差分は「受信方式」「応答の締め切り」「返信の作法」の 3 点に収束します。

	LINE	Telegram	Discord
受信方式	Webhook	Webhook / long polling	Gateway（WebSocket 常時接続）
受信の検証	`X-Line-Signature` を HMAC-SHA256 で検証	`secret_token` ヘッダーを突き合わせ	不要（Gateway が認証済み）
応答の締め切り	reply token が約 1 分で失効	実質なし	スラッシュコマンドは 3 秒
締め切りの回避策	loading 表示 → push message	—	`ctx.defer()` → `followup.send()`
「入力中」表示	loading アニメーション API	`sendChatAction`	`channel.typing()`
1 通の文字数上限	5,000 文字	4,096 文字	2,000 文字

見比べていて腑に落ちたのは、どのプラットフォームも結局は同じことを要求している、という点でした。「あなたのサーバーが生きていることを、まず素早く知らせろ」。Claude の生成が終わるのを待ってから返事をしようとすると、名前を変えた同じ壁に必ずぶつかります。

先に「受け取りました」を返し、本文は後から送る。この形さえ最初から組んでおけば、載せ替え先がどこであれ、大きく崩れることはありません。

STEP 9：Claude に個人情報をそのまま渡さない — PII マスキング

個人開発の案件でも、問い合わせ対応をチャットボットに任せようとした途端にこの問題に直面しました。社内データや顧客とのやり取りをチャットボットに通すと、メールアドレスや電話番号といった個人情報（PII）が、そのまま API に流れてしまいます。入力ログにも、モデルへのリクエストにも生の値を残したくない、という要件は本番では珍しくありません。

有効なのは、送信前に PII をプレースホルダーへ置き換え、応答が返ってきてから元に戻すという可逆マスキングです。Claude には {{EMAIL_1}} のような無害化された文字列だけが渡り、ユーザーには復元済みの自然な文章が返ります。

import re
 
class PIIMasker:
    """送信前にPIIを伏字化し、応答後に復元する可逆マスカー"""
 
    PATTERNS = {
        "EMAIL": re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"),
        "PHONE": re.compile(r"0\d{1,4}-\d{1,4}-\d{4}"),
        "CARD": re.compile(r"\b(?:\d[ -]?){13,16}\b"),
    }
 
    def mask(self, text: str):
        mapping = {}
        counters = {}
        def replace(kind, m):
            counters[kind] = counters.get(kind, 0) + 1
            token = f"{{{{{kind}_{counters[kind]}}}}}"
            mapping[token] = m.group(0)
            return token
        for kind, pattern in self.PATTERNS.items():
            text = pattern.sub(lambda m, k=kind: replace(k, m), text)
        return text, mapping
 
    def unmask(self, text: str, mapping: dict) -> str:
        for token, original in mapping.items():
            text = text.replace(token, original)
        return text
 
# 使用例
masker = PIIMasker()
user_text = "請求書を tanaka@example.com に送って、090-1234-5678 にも連絡して"
masked, mapping = masker.mask(user_text)
print(masked)
# → 請求書を {{EMAIL_1}} に送って、{{PHONE_1}} にも連絡して
 
reply = bot.send(masked)          # Claude には伏字だけが渡る
restored = masker.unmask(reply, mapping)  # ユーザーには元の値で返す

ひとつ正直に書いておきたいのは、正規表現によるマスキングは万能ではないということです。メールや電話番号のような形式が決まっているものは捕まえやすいですが、氏名や住所のように文脈で判断する PII は、固有表現抽出（NER）や辞書を併用しないと取りこぼします。

「絶対に外に出してはいけない情報」が決まっているなら、そもそもその項目を Claude に渡さない設計が最も確実です。マスキングは「うっかり混入を減らす一層の防御」と位置づけ、本当に守りたい情報は入力段階で切り離す。この二段構えで考えておくと、後から監査が入っても説明できる基盤になります。

まとめと次のステップ

STEP 1-2: 基本実装と会話履歴管理
STEP 3: ストリーミングによるUX改善
STEP 4: FastAPIでのWeb API化
STEP 5: コスト最適化テクニック
STEP 6: 本番向けエラーハンドリング
STEP 7: すべてを統合した実用クラス
STEP 8: LINE・Telegram への接続と署名検証
STEP 9: PII マスキングで個人情報を守る

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

ツールユース完全ガイド — チャットボットに検索や計算機能を追加する
マルチモーダル入力ガイド — 画像を理解するチャットボット
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：完成形 — 実用的なチャットボットクラス

STEP 8：LINE・Telegram・Discord に接続する — プラットフォーム固有の実装差分

LINE：reply token の寿命が落とし穴

Telegram：webhook か long polling かを使い分ける

Discord：3 秒以内に「受け取った」と返す

3 つのプラットフォームを並べて見る

STEP 9：Claude に個人情報をそのまま渡さない — PII マスキング

まとめと次のステップ

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

関連記事