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

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