Gemini API でカスタマーサポートチャットボットを段階的に育てる実装ノート

import google.generativeai as genai
import os
 
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
 
SYSTEM_PROMPT = """あなたは「TechShop」の親切なカスタマーサポート担当者です。
 
【役割と責務】
- 製品に関する質問に、丁寧かつ正確に回答する
- 問題解決の手順を、分かりやすい言葉で説明する
- 自力で解決できない場合は、担当者への転送を提案する
 
【応答のガイドライン】
- 常に敬語を使用し、押し付けがましくない口調を保つ
- 手順は番号付きの短いステップで提示する
- 推測で回答せず、不明な点は正直に伝える
- 一度に複数の情報を詰め込まず、段階的に説明する
 
【エスカレーション条件】
- 返金・返品に関する複雑な案件
- 技術的な不具合の詳細調査が必要な場合
- ユーザーが「人間のサポートを希望」と明示した場合
- クレームや法的問題が示唆される場合
 
エスカレーションが必要と判断した場合は、回答の末尾に「[ESCALATE]」というキーワードを追加してください。"""

import google.generativeai as genai
import os
 
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
 
class CustomerSupportBot:
    def __init__(self, system_prompt: str):
        self.model = genai.GenerativeModel(
            model_name="gemini-2.5-flash",
            system_instruction=system_prompt,
        )
        self.chat = self.model.start_chat(history=[])
        self.escalated = False
        self.conversation_log = []
 
    def send_message(self, user_message: str) -> dict:
        if self.escalated:
            return {
                "response_text": "担当者への転送処理を進めています。少々お待ちください。",
                "escalate": True,
            }
 
        try:
            response = self.chat.send_message(user_message)
            response_text = response.text
 
            self.conversation_log.append({"role": "user", "content": user_message})
            self.conversation_log.append({"role": "assistant", "content": response_text})
 
            escalate = "[ESCALATE]" in response_text
            if escalate:
                response_text = response_text.replace("[ESCALATE]", "").strip()
                self.escalated = True
 
            return {"response_text": response_text, "escalate": escalate}
 
        except Exception:
            return {
                "response_text": "申し訳ございません。一時的なエラーが発生いたしました。少し時間をおいて再度お試しください。",
                "escalate": False,
            }
 
 
bot = CustomerSupportBot(SYSTEM_PROMPT)
for msg in ["注文した商品がまだ届きません", "注文番号は12345です", "もう返金してほしいのですが"]:
    print(f"ユーザー: {msg}")
    result = bot.send_message(msg)
    print(f"Bot: {result['response_text']}")
    if result["escalate"]:
        print(">>> 担当者へのエスカレーションが必要です <<<")
    print()

import google.generativeai as genai
import os
 
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
 
FAQ_DB = {
    "配送": {
        "answer": "通常 2〜5 営業日でお届けします。お急ぎの場合は速達オプション（+500 円）をご利用ください。",
        "related": "配送追跡は注文確認メールのリンクからご確認いただけます。",
    },
    "返品": {
        "answer": "商品到着後 14 日以内であれば返品可能です。未開封・未使用のものに限ります。",
        "related": "返品用ラベルはマイページからダウンロードできます。",
    },
    "保証": {
        "answer": "全製品に 1 年間のメーカー保証が付いています。",
        "related": "保証期間内の修理は無償で対応いたします。",
    },
    "支払い": {
        "answer": "クレジットカード、銀行振込、コンビニ決済に対応しています。",
        "related": "分割払いはクレジットカードのご利用条件によって異なります。",
    },
}
 
def search_faq(query: str) -> str:
    results = []
    for keyword, info in FAQ_DB.items():
        if keyword in query:
            results.append(f"{info['answer']} {info['related']}")
    return " ".join(results) if results else "該当する情報が見つかりませんでした。担当者に確認の上、回答いたします。"
 
faq_tool = {
    "function_declarations": [
        {
            "name": "search_faq",
            "description": "FAQ データベースを検索して、製品・サービスに関する情報を取得する",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "検索するキーワードや質問内容"},
                },
                "required": ["query"],
            },
        }
    ]
}
 
model_with_faq = genai.GenerativeModel(
    model_name="gemini-2.5-flash",
    system_instruction=SYSTEM_PROMPT,
    tools=[faq_tool],
)
chat_with_faq = model_with_faq.start_chat(history=[])
 
def handle_with_faq(user_message: str) -> str:
    response = chat_with_faq.send_message(user_message)
    for part in response.candidates[0].content.parts:
        if hasattr(part, "function_call") and part.function_call.name == "search_faq":
            query = part.function_call.args.get("query", user_message)
            faq_result = search_faq(query)
            response = chat_with_faq.send_message(
                genai.protos.Content(parts=[
                    genai.protos.Part(
                        function_response=genai.protos.FunctionResponse(
                            name="search_faq",
                            response={"result": faq_result},
                        )
                    )
                ])
            )
    return response.text

import google.generativeai as genai
import os
from datetime import datetime
 
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
 
def summarize_conversation(conversation_log: list) -> str:
    if not conversation_log:
        return "会話履歴なし"
 
    model = genai.GenerativeModel("gemini-2.5-flash")
    history_text = "\n".join(f"{m['role']}: {m['content']}" for m in conversation_log)
    prompt = f"""以下の会話履歴を 3 文以内で要約してください。
問題の概要・現状・ユーザーの感情状態を簡潔に記載してください。
 
会話履歴:
{history_text}"""
    response = model.generate_content(prompt)
    return response.text
 
def assess_priority(summary: str) -> str:
    high_priority = ["緊急", "至急", "返金", "法的", "クレーム", "詐欺", "被害"]
    return "高" if any(k in summary for k in high_priority) else "通常"
 
def create_escalation_ticket(conversation_log: list, user_id: str = "anonymous") -> dict:
    summary = summarize_conversation(conversation_log)
    priority = assess_priority(summary)
 
    ticket = {
        "ticket_id": f"ESC-{datetime.now().strftime('%Y%m%d%H%M%S')}",
        "created_at": datetime.now().isoformat(),
        "user_id": user_id,
        "summary": summary,
        "priority": priority,
        "conversation_log": conversation_log,
        "status": "open",
        "assigned_to": None,
    }
 
    print(f"[エスカレーション] {ticket['ticket_id']} 優先度={ticket['priority']}")
    return ticket

import google.generativeai as genai
 
SUMMARIZE_AFTER_TURNS = 10   # このターン数を超えたら圧縮
KEEP_RECENT_TURNS = 4        # 直近はそのまま残す
FALLBACK_TEXT = "申し訳ありません、この内容はこちらでお答えできませんでした。担当者におつなぎします。"
 
 
def compact_history(history: list) -> list:
    """古い履歴を要約 1 件に畳み、直近ターンだけ生で残す。"""
    if len(history) <= SUMMARIZE_AFTER_TURNS:
        return history
 
    old, recent = history[:-KEEP_RECENT_TURNS], history[-KEEP_RECENT_TURNS:]
    text = "\n".join(f"{m['role']}: {m['content']}" for m in old)
    model = genai.GenerativeModel("gemini-2.5-flash")
    summary = model.generate_content(
        "以下のサポート会話を、未解決の論点と確定した事実だけ箇条書きで残して要約してください。"
        "推測や感想は書かないでください。\n\n" + text
    ).text
    return [{"role": "user", "content": f"[これまでの経緯]\n{summary}"}] + recent
 
 
def safe_text(response) -> tuple[str, str | None]:
    """空応答を UI に出さない。(本文, 打ち切り理由) を返す。"""
    candidates = getattr(response, "candidates", None) or []
    if not candidates:
        return FALLBACK_TEXT, "NO_CANDIDATE"
 
    reason = getattr(candidates[0], "finish_reason", None)
    reason = getattr(reason, "name", str(reason))
    if reason in ("SAFETY", "RECITATION", "OTHER"):
        return FALLBACK_TEXT, reason
 
    text = (getattr(response, "text", "") or "").strip()
    if not text:
        return FALLBACK_TEXT, reason or "EMPTY"
    return text, None

TEST_CASES = [
    {
        "input": "注文番号 12345 の商品がまだ届きません",
        "expect_no_escalate": True,
        "expect_keywords": ["配送", "確認"],
    },
    {
        "input": "もう返金してほしい、本当に頭にきました",
        "expect_escalate": True,
    },
    {
        "input": "Hi, I haven't received my order yet",
        "expect_no_escalate": True,
        "expect_lang": "en",
    },
]
 
def run_tests(bot_factory):
    results = []
    for case in TEST_CASES:
        bot = bot_factory()
        out = bot.send_message(case["input"])
        passed = True
        if case.get("expect_escalate") and not out["escalate"]:
            passed = False
        if case.get("expect_no_escalate") and out["escalate"]:
            passed = False
        for kw in case.get("expect_keywords", []):
            if kw not in out["response_text"]:
                passed = False
        results.append((case["input"][:30], passed))
    return results