Google Chat Bot に構造化ログとフィードバックダイアログを実装してKB改善サイクルを回す

import json
import os
import threading
from typing import Any

_local = threading.local()
_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "")

def set_request_context(*, trace_header: str = "") -> None:
    """リクエストごとのトレースコンテキストを設定する。"""
    trace = ""
    if trace_header and _PROJECT_ID:
        trace_id = trace_header.split("/")[0]
        trace = f"projects/{_PROJECT_ID}/traces/{trace_id}"
    _local.trace = trace

def log_event(severity: str, event_type: str, **fields: Any) -> None:
    """構造化 JSON ログエントリを stdout に書き出す。"""
    entry: dict[str, Any] = {
        "severity": severity,
        "event_type": event_type,
        **fields,
    }
    trace = getattr(_local, "trace", "")
    if trace:
        entry["logging.googleapis.com/trace"] = trace

    print(json.dumps(entry, ensure_ascii=False, default=str), flush=True)

{
  "severity": "INFO",
  "event_type": "usage",
  "user_id": "users/123456",
  "space_id": "spaces/AAAA",
  "message_id": "spaces/AAAA/messages/BBBB",
  "query_text": "有給の申請方法は？",
  "query_length": 10,
  "rag_confidence": 0.35,
  "rag_sources_count": 3,
  "response_text": "有給休暇の申請は、社内ポータルの「勤怠管理」から...",
  "response_length": 520,
  "latency_ms": 3200,
  "is_fallback": false,
  "error": null
}

{
  "severity": "INFO",
  "event_type": "feedback",
  "user_id": "users/123456",
  "space_id": "spaces/AAAA",
  "message_id": "spaces/AAAA/messages/BBBB",
  "vote": "down",
  "reason_category": "inaccurate",
  "comment_text": "参照先が古い"
}

def _build_feedback_section(message_name, endpoint_url):
    msg_id = message_name or ""
    function_url = endpoint_url or ""
    return {
        "widgets": [{
            "buttonList": {
                "buttons": [
                    {
                        "text": "参考になった",
                        "onClick": {
                            "action": {
                                "function": function_url,
                                "parameters": [
                                    {"key": "action", "value": "feedback"},
                                    {"key": "vote", "value": "up"},
                                    {"key": "message_id", "value": msg_id},
                                ],
                            }
                        },
                    },
                    {
                        "text": "改善が必要",
                        "onClick": {
                            "action": {
                                "function": function_url,
                                "interaction": "OPEN_DIALOG",  # ダイアログを開く
                                "parameters": [
                                    {"key": "action", "value": "feedback"},
                                    {"key": "vote", "value": "down"},
                                    {"key": "message_id", "value": msg_id},
                                ],
                            }
                        },
                    },
                ]
            }
        }]
    }

def build_feedback_dialog(message_id, endpoint_url):
    function_url = endpoint_url or ""
    msg_id = message_id or ""
    return {
        "action": {
            "navigations": [{
                "pushCard": {
                    "sections": [{
                        "header": "フィードバック",
                        "widgets": [
                            {
                                "textParagraph": {
                                    "text": "回答の改善のため、詳細を教えてください。",
                                }
                            },
                            {
                                "selectionInput": {
                                    "type": "DROPDOWN",
                                    "label": "理由",
                                    "name": "reason_category",
                                    "items": [
                                        {"text": "不正確", "value": "inaccurate", "selected": True},
                                        {"text": "関連性が低い", "value": "irrelevant"},
                                        {"text": "不完全", "value": "incomplete"},
                                        {"text": "その他", "value": "other"},
                                    ],
                                }
                            },
                            {
                                "textInput": {
                                    "label": "コメント（任意）",
                                    "type": "MULTIPLE_LINE",
                                    "name": "comment_text",
                                }
                            },
                            {
                                "buttonList": {
                                    "buttons": [
                                        {
                                            "text": "送信",
                                            "onClick": {
                                                "action": {
                                                    "function": function_url,
                                                    "parameters": [
                                                        {"key": "action", "value": "feedback"},
                                                        {"key": "vote", "value": "down"},
                                                        {"key": "message_id", "value": msg_id},
                                                        {"key": "submit", "value": "true"},
                                                    ],
                                                }
                                            },
                                        },
                                        {
                                            "text": "スキップ",
                                            "onClick": {
                                                "action": {
                                                    "function": function_url,
                                                    "parameters": [
                                                        {"key": "action", "value": "feedback"},
                                                        {"key": "vote", "value": "down"},
                                                        {"key": "message_id", "value": msg_id},
                                                        {"key": "submit", "value": "true"},
                                                        {"key": "skipped", "value": "true"},
                                                    ],
                                                }
                                            },
                                        },
                                    ]
                                }
                            },
                        ]
                    }]
                }
            }]
        }
    }

chat = body.get("chat", {})
common_event = body.get("commonEventObject", {})
params = common_event.get("parameters", {})

host = request.headers.get("X-Forwarded-Host") or request.headers.get("Host", "")
scheme = request.headers.get("X-Forwarded-Proto", "https")
service = os.environ.get("K_SERVICE", "")
endpoint_url = f"{scheme}://{host}/{service}" if host else ""

if "buttonClickedPayload" in chat:
    if params.get("action") != "feedback":
        return {}

    # ダイアログ送信（送信/スキップボタン）
    if params.get("submit") == "true":
        return feedback.handle_dialog_submit(body)

    # カードボタンクリック（参考になった / 改善が必要）
    return feedback.handle_card_click(body, endpoint_url=endpoint_url)

_THANK_YOU_TEXT = "フィードバックありがとうございます！"

def _build_thank_you_body(current_message):
    """既存セクションを保持し、フィードバックボタンだけを「ありがとう」に差し替える。"""
    cards_v2 = current_message.get("cardsV2", [])
    if not cards_v2:
        return _fallback_thank_you_card()

    card = cards_v2[0].get("card", {})
    sections = list(card.get("sections", []))
    if not sections:
        return _fallback_thank_you_card()

    last_section = sections[-1]
    last_widgets = last_section.get("widgets", [])

    # 冪等性ガード: 既に「ありがとう」テキストがあればスキップ
    if (len(last_widgets) == 1
            and "textParagraph" in last_widgets[0]
            and _THANK_YOU_TEXT in last_widgets[0]["textParagraph"].get("text", "")):
        return None

    # フィードバックボタンのセクションを除去
    if any("buttonList" in w for w in last_widgets):
        sections = sections[:-1]

    sections.append({
        "widgets": [{"textParagraph": {"text": _THANK_YOU_TEXT}}]
    })
    return {
        "cardsV2": [{
            "cardId": cards_v2[0].get("cardId", "progressive-card"),
            "card": {"sections": sections},
        }]
    }

def _replace_feedback_with_thanks(client, message_id):
    """GET → 加工 → PATCH でフィードバックボタンを「ありがとう」に差し替える。"""
    current_message = client.get_message(message_id)
    body = _build_thank_you_body(current_message)
    if body is not None:
        client.patch_message(message_id, body, "cardsV2")

def handle_card_click(event_body, endpoint_url=""):
    vote, message_id, user_id, space_id = _extract_feedback_params(event_body)

    if vote == "up":
        log_event("INFO", "feedback",
                  user_id=user_id, space_id=space_id,
                  message_id=message_id, vote=vote,
                  reason_category=None, comment_text=None)
        try:
            client = ChatApiClient()
            _replace_feedback_with_thanks(client, message_id)
        except Exception:
            logger.warning("Failed to patch message after upvote", exc_info=True)
        return {}

    # vote == "down": ダイアログを開く（この時点ではログしない）
    return build_feedback_dialog(message_id, endpoint_url)

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

    vote, message_id, user_id, space_id = _extract_feedback_params(event_body)
    skipped = params.get("skipped", "false") == "true"

    reason_category = None
    comment_text = None
    if not skipped:
        reason_raw = form_inputs.get("reason_category", {})
        reason_values = reason_raw.get("stringInputs", {}).get("value", [])
        reason_category = reason_values[0] if reason_values else None

        comment_raw = form_inputs.get("comment_text", {})
        comment_values = comment_raw.get("stringInputs", {}).get("value", [])
        comment_text = comment_values[0] if comment_values else None

    log_event("INFO", "feedback",
              user_id=user_id, space_id=space_id,
              message_id=message_id, vote=vote,
              reason_category=reason_category, comment_text=comment_text)

    # 元のメッセージのフィードバックボタンを「ありがとう」に差し替え
    try:
        client = ChatApiClient()
        _replace_feedback_with_thanks(client, message_id)
    except Exception:
        logger.warning("Failed to patch message after feedback", exc_info=True)

    # ダイアログを閉じる
    return {
        "action": {
            "navigations": [{
                "endNavigation": {"action": "CLOSE_DIALOG"},
            }],
            "notification": {
                "text": _THANK_YOU_TEXT,
            },
        }
    }

def _emit_usage_event(start_time, user_name, space_name, message_name,
                      user_text, contexts, state, error_str, is_fallback):
    latency_ms = int((time.monotonic() - start_time) * 1000)
    best_score = contexts[0]["score"] if contexts else None
    severity = "ERROR" if error_str else ("WARNING" if is_fallback else "INFO")
    log_event(
        severity, "usage",
        user_id=user_name, space_id=space_name,
        message_id=message_name,
        query_text=user_text, query_length=len(user_text),
        rag_confidence=best_score, rag_sources_count=len(contexts),
        response_text="\n".join(state.content_paragraphs),
        response_length=sum(len(p) for p in state.content_paragraphs),
        latency_ms=latency_ms, is_fallback=is_fallback, error=error_str,
    )

resource.type="cloud_run_revision"
AND resource.labels.service_name="your-bot-name"
AND jsonPayload.event_type="feedback"
AND jsonPayload.vote="down"

# BigQuery データセットを作成
bq mk --dataset --location=asia-northeast1 YOUR_PROJECT:chat_analytics

# ログシンクを作成（usage と feedback イベントのみ転送）
gcloud logging sinks create kb-analytics-sink \
  bigquery.googleapis.com/projects/YOUR_PROJECT/datasets/chat_analytics \
  --log-filter='jsonPayload.event_type=("usage" OR "feedback")'

-- usage イベント用ビュー
CREATE OR REPLACE VIEW `chat_analytics.usage_events` AS
SELECT
  timestamp,
  jsonPayload.user_id,
  jsonPayload.space_id,
  jsonPayload.message_id,
  jsonPayload.query_text,
  jsonPayload.response_text,
  jsonPayload.rag_confidence,
  jsonPayload.rag_sources_count,
  jsonPayload.latency_ms,
  jsonPayload.is_fallback,
  jsonPayload.error
FROM `chat_analytics.YOUR_TABLE`
WHERE jsonPayload.event_type = 'usage';

-- feedback イベント用ビュー
CREATE OR REPLACE VIEW `chat_analytics.feedback_events` AS
SELECT
  timestamp,
  jsonPayload.user_id,
  jsonPayload.space_id,
  jsonPayload.message_id,
  jsonPayload.vote,
  jsonPayload.reason_category,
  jsonPayload.comment_text
FROM `chat_analytics.YOUR_TABLE`
WHERE jsonPayload.event_type = 'feedback';

SELECT query_text, COUNT(*) as count
FROM `chat_analytics.usage_events`
WHERE is_fallback = true
GROUP BY query_text
ORDER BY count DESC

SELECT u.query_text, u.response_text,
       f.reason_category, f.comment_text
FROM `chat_analytics.usage_events` u
JOIN `chat_analytics.feedback_events` f
  ON u.message_id = f.message_id
WHERE f.vote = 'down'
ORDER BY f.timestamp DESC

# ❌ 動かない（空文字列は falsy）
if common_event.get("invokedFunction"):
    ...

# ✅ 正しい判定方法
if "buttonClickedPayload" in chat:
    ...

host = request.headers.get("X-Forwarded-Host") or request.headers.get("Host", "")
scheme = request.headers.get("X-Forwarded-Proto", "https")
service = os.environ.get("K_SERVICE", "")
endpoint_url = f"{scheme}://{host}/{service}" if host else ""

# ❌ Add-ons では効果なし
return {"actionResponse": {"type": "UPDATE_MESSAGE"}, **thank_you_card}

# ✅ GET-then-PATCH で元の回答を保持しつつボタンだけ差し替え
client = ChatApiClient()
current_message = client.get_message(message_id)
body = _build_thank_you_body(current_message)  # ボタン除去 + ありがとう追加
if body is not None:
    client.patch_message(message_id, body, "cardsV2")
return {}

# ❌ 通常の Chat API 形式（Add-ons では動かない）
return {
    "actionResponse": {
        "type": "DIALOG",
        "dialogAction": {
            "dialog": {"body": {"sections": [...]}}
        }
    }
}

# ✅ Add-ons でダイアログを開く
return {
    "action": {
        "navigations": [{
            "pushCard": {"sections": [...]}
        }]
    }
}

# ✅ Add-ons でダイアログを閉じる
return {
    "action": {
        "navigations": [{
            "endNavigation": {"action": "CLOSE_DIALOG"}
        }],
        "notification": {"text": "完了しました"},
    }
}

# ❌ Add-ons では dialogEventType が存在しない
if body.get("dialogEventType") == "SUBMIT_DIALOG":
    ...

# ダイアログ内の送信ボタンに submit=true パラメータを付与
{"key": "submit", "value": "true"}

# ルーティングで判定
if params.get("submit") == "true":
    return feedback.handle_dialog_submit(body)  # ダイアログ送信
return feedback.handle_card_click(body, ...)     # カードボタンクリック

from pydantic import BaseModel, Field

class RagAnswer(BaseModel):
    answer: str = Field(description="ナレッジベースの情報に基づく回答（Markdown形式）")
    is_answerable: bool = Field(
        description="ナレッジベースに質問に該当する情報があり回答できた場合はtrue、"
        "該当情報がなく回答できなかった場合はfalse",
    )

from google import genai
from google.genai import types

def generate_answer(question: str, contexts: list[dict]) -> RagAnswer:
    # ...
    response = client.models.generate_content(
        model=MODEL_ID,
        contents=prompt,
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_PROMPT,
            response_schema=RagAnswer,          # Pydantic モデルを直接渡せる
            response_mime_type="application/json",
        ),
    )
    return response.parsed  # バリデーション済みの RagAnswer インスタンス

result = generate_answer(user_text, contexts)
is_fallback = not result.is_answerable  # キーワードマッチング不要