Google Chat Bot に聞き返しウィジェットを実装してみた — cardsV2 のインラインフォーム活用

ユーザー: 「マクロ」
ボット:   Excel のマクロについて説明します...（ユーザーは Outlook のマクロを聞きたかった）

ユーザー: 「マクロ」
ボット:   「マクロ」について、どの内容をお探しですか？
          [Excel のマクロ] [Outlook のマクロ] [スプレッドシートのマクロ]
          [自由入力欄] [送信]
ユーザー: [Excel のマクロ] をクリック
ボット:   Excel のマクロの設定手順は以下の通りです...

from pydantic import BaseModel, Field

class RagAnswer(BaseModel):
    """Structured output from Gemini for RAG responses."""

    answer: str = Field(
        description="ナレッジベースの情報に基づく回答（Markdown形式）"
    )
    is_answerable: bool = Field(
        description="ナレッジベースに該当する情報があり回答できた場合はtrue"
    )
    needs_clarification: bool = Field(
        default=False,
        description="クエリが曖昧で意図を特定できない場合はtrue。"
        "trueの場合、answerに聞き返しの文を入れ、"
        "clarification_optionsに選択肢を入れる",
    )
    clarification_options: list[str] = Field(
        default_factory=list,
        description="聞き返す場合の選択肢（最大3つ）。"
        "needs_clarificationがtrueの場合のみ使用",
    )

## 曖昧なクエリの処理
- ユーザーの質問が曖昧で、ナレッジベースの検索結果が複数の異なるトピックに該当する場合：
  - needs_clarification を true に設定する
  - answer に聞き返しの文を入れる
  - clarification_options に検索結果に基づく選択肢を最大3つ入れる
- 質問が一語のみでも、検索結果が1つのトピックに集中している場合は聞き返さず即回答する
- 検索結果が0件の場合は聞き返さず、is_answerable を false にする

from google import genai
from google.genai import types

client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)

response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=prompt,
    config=types.GenerateContentConfig(
        system_instruction=SYSTEM_PROMPT,
        response_schema=RagAnswer,
        response_mime_type="application/json",
    ),
)

result = response.parsed  # → RagAnswer インスタンス

def _build_clarification_section(clarification_text, options,
                                 message_name, endpoint_url,
                                 space_name, thread_name, user_name,
                                 original_query=""):
    function_url = endpoint_url or ""
    base_params = [
        {"key": "action", "value": "clarify"},
        {"key": "message_name", "value": message_name or ""},
        {"key": "space_name", "value": space_name or ""},
        {"key": "thread_name", "value": thread_name or ""},
        {"key": "user_name", "value": user_name or ""},
        {"key": "original_query", "value": original_query or ""},
    ]

    widgets = [
        {"textParagraph": {"text": clarification_text}},
    ]

    # 選択肢ボタン（最大3つ）
    if options:
        buttons = []
        for option in options[:3]:
            buttons.append({
                "text": option,
                "onClick": {
                    "action": {
                        "function": function_url,
                        "parameters": base_params + [
                            {"key": "query", "value": option},
                        ],
                    }
                },
            })
        widgets.append({"buttonList": {"buttons": buttons}})

    # 自由入力欄
    widgets.append({
        "textInput": {
            "label": "上記以外はこちらに入力",
            "type": "SINGLE_LINE",
            "name": "clarify_text",
        }
    })

    # 送信ボタン（自由入力用）
    widgets.append({
        "buttonList": {
            "buttons": [{
                "text": "送信",
                "onClick": {
                    "action": {
                        "function": function_url,
                        "parameters": base_params + [
                            {"key": "query", "value": ""},
                        ],
                    }
                },
            }]
        }
    })

    return {"widgets": widgets}

{
  "commonEventObject": {
    "formInputs": {
      "clarify_text": {
        "stringInputs": {
          "value": ["ユーザーが入力したテキスト"]
        }
      }
    },
    "parameters": {
      "action": "clarify",
      "query": ""
    }
  }
}

if "buttonClickedPayload" in chat:
    action = params.get("action")

    if action == "clarify":
        return handle_clarify(body, endpoint_url=endpoint_url)

    if action == "feedback":
        # 既存のフィードバック処理
        ...

import threading
from bot.worker import process_message

def handle_clarify(event_body, endpoint_url=""):
    common_event = event_body.get("commonEventObject", {})
    params = common_event.get("parameters", {})
    form_inputs = common_event.get("formInputs", {})

    # ボタンクリック → params["query"] にテキストが入っている
    # 自由入力 + 送信ボタン → params["query"] は空、formInputs にテキスト
    query = params.get("query", "")
    if not query:
        text_input = form_inputs.get("clarify_text", {})
        values = text_input.get("stringInputs", {}).get("value", [])
        query = values[0].strip() if values else ""

    if not query:
        return {}  # 空クエリは無視

    # 元のクエリと聞き返し応答を結合して文脈を保持
    original_query = params.get("original_query", "")
    if original_query and query != original_query:
        query = f"{original_query}（{query}について）"

    message_name = params.get("message_name", "")
    space_name = params.get("space_name", "")
    thread_name = params.get("thread_name", "")
    user_name = params.get("user_name", "")

    user = event_body.get("chat", {}).get("user", {})
    sender = user.get("displayName", "")

    thread = threading.Thread(
        target=process_message,
        args=(space_name, query, sender),
        kwargs={
            "endpoint_url": endpoint_url,
            "thread_name": thread_name or None,
            "user_name": user_name,
            "message_name": message_name or None,
            "is_clarification": True,
        },
    )
    thread.start()

    return {}

# process_message 内
if message_name is None:
    message_name = chat_client.create_message(space_name, body, thread_name=thread_name)
else:
    chat_client.patch_message(message_name, body, "cardsV2")

ユーザー: 「マクロ」
ボット:   聞き返しカード → [Excel のマクロ]
ボット:   また聞き返しカード → 無限ループ！

# 無限ループ防止
if is_clarification and result.needs_clarification:
    result = RagAnswer(
        answer=result.answer,
        is_answerable=result.is_answerable,
        needs_clarification=False,
        clarification_options=[],
    )

## 制約
- パスワードやアカウント情報は直接回答しない
  （手順のみ案内し、回答の末尾に必ず次の一文を追加する）：
  「YOUR_ESCALATION_TEXTにお問い合わせください。」
- ナレッジベースに該当情報がない場合は、回答の末尾に必ず次の一文を追加する：
  「YOUR_ESCALATION_TEXTにお問い合わせください。」

ESCALATION_TEXT = "YOUR_ESCALATION_TEXTにお問い合わせください。"

def _ensure_escalation(result: RagAnswer) -> RagAnswer:
    """Append escalation text to non-answerable responses if missing."""
    if result.needs_clarification:
        return result  # 聞き返し中はスキップ
    if not result.is_answerable and ESCALATION_TEXT not in result.answer:
        return RagAnswer(
            answer=result.answer.rstrip() + "\n" + ESCALATION_TEXT,
            is_answerable=False,
        )
    return result

def generate_answer(question, contexts, history=None):
    if not contexts:
        return RagAnswer(answer=NO_RESULT_MESSAGE, is_answerable=False)

    # ... LLM 呼び出し ...

    try:
        if isinstance(response.parsed, RagAnswer):
            return _ensure_escalation(response.parsed)
        if response.text is not None:
            return _ensure_escalation(RagAnswer.model_validate_json(response.text))
    except (ValueError, TypeError):
        logger.warning("Failed to parse structured response")
    return RagAnswer(answer=NO_RESULT_MESSAGE, is_answerable=False)

ユーザー: 「ツールの申請方法を教えて」
ボット:   聞き返しカード → [kintone] [Salesforce] [Microsoft 365] + 自由入力
ユーザー: 自由入力に「kintone」と入力して送信
ボット:   kintone の一般的な説明を回答 ← 「申請方法」の文脈が消えている！

# cards.py — base_params に original_query を追加
base_params = [
    {"key": "action", "value": "clarify"},
    {"key": "original_query", "value": original_query or ""},
    # ...
]

# clarify.py — 結合してからパイプラインを再実行
original_query = params.get("original_query", "")
if original_query and query != original_query:
    query = f"{original_query}（{query}について）"