Google Chat BotにFirestoreで会話履歴を持たせてフォローアップ質問に対応する

メッセージ受信 → KB検索 → 回答生成 → 送信

メッセージ受信 → 履歴取得 → クエリ再構成 → KB検索 → 回答生成(+履歴) → 送信 → 履歴保存

doc_id: "spaces_ABC_threads_XYZ"
{
  turns: [
    { role: "user", text: "VPNの設定方法を教えて", timestamp: ... },
    { role: "model", text: "VPN接続の手順は以下の通りです...", timestamp: ... },
    ...
  ],
  expired_at: <現在時刻 + 24時間>   // TTL用
}

import logging
import threading
from datetime import datetime, timedelta, timezone

from google.cloud import firestore

logger = logging.getLogger(__name__)

_COLLECTION = "chat_threads"
_TTL_HOURS = 24
_client = None
_lock = threading.Lock()

def _get_client() -> firestore.Client:
    global _client
    if _client is None:
        with _lock:
            if _client is None:
                _client = firestore.Client()
    return _client

def _encode_doc_id(thread_name: str) -> str:
    return thread_name.replace("/", "_")

def get_history(thread_name: str) -> list[dict]:
    if not thread_name:
        return []
    try:
        doc = _get_client().collection(_COLLECTION).document(
            _encode_doc_id(thread_name)
        ).get()
        if not doc.exists:
            return []
        turns = doc.to_dict().get("turns", [])
        return [{"role": t["role"], "text": t["text"]} for t in turns]
    except Exception:
        logger.warning("Failed to read history for %s", thread_name, exc_info=True)
        return []

def save_turn(thread_name: str, user_text: str, model_text: str) -> None:
    if not thread_name:
        return
    try:
        doc_ref = _get_client().collection(_COLLECTION).document(
            _encode_doc_id(thread_name)
        )
        doc = doc_ref.get()

        turns = []
        if doc.exists:
            turns = doc.to_dict().get("turns", [])

        now = datetime.now(timezone.utc)
        turns.append({"role": "user", "text": user_text, "timestamp": now})
        turns.append({"role": "model", "text": model_text, "timestamp": now})

        doc_ref.set({
            "turns": turns,
            "expired_at": now + timedelta(hours=_TTL_HOURS),
        })
    except Exception:
        logger.warning("Failed to save turn for %s", thread_name, exc_info=True)

REFORMULATE_TEMPLATE = """\
以下は社内ITサポートチャットの会話履歴です。

## 会話履歴
{history}

## 最新の質問
{question}

## タスク
最新の質問が会話履歴の文脈に依存している場合（例：「それ」「その方法」「他には？」など）、\
ナレッジベース検索に適した自己完結型の質問文に書き換えてください。
すでに自己完結している質問はそのまま返してください。
書き換えた質問文だけを返してください。説明や前置きは不要です。
"""

def _format_history_text(history: list[dict]) -> str:
    lines = []
    for turn in history:
        role_label = "ユーザー" if turn["role"] == "user" else "アシスタント"
        lines.append(f"{role_label}: {turn['text']}")
    return "\n".join(lines)

def reformulate_query(question: str, history: list[dict]) -> str:
    if not history:
        return question

    history_str = _format_history_text(history)
    prompt = REFORMULATE_TEMPLATE.format(history=history_str, question=question)

    try:
        client = _get_genai_client()
        response = client.models.generate_content(
            model=MODEL_ID,
            contents=prompt,
        )
        reformulated = response.text.strip()
        return reformulated if reformulated else question
    except Exception:
        logger.warning("Query reformulation failed, using original", exc_info=True)
        return question

GENERATION_TEMPLATE_WITH_HISTORY = """\
以下のナレッジベースの情報と会話履歴を参考に、ユーザーの質問に回答してください。

## 会話履歴

{history}

## ナレッジベース検索結果

{context}

## ユーザーの質問

{question}

## 回答
"""

def generate_answer(question: str, contexts: list[dict],
                    history: list[dict] | None = None) -> RagAnswer:
    # ... 省略 ...

    if history:
        history_str = _format_history_text(history)
        prompt = GENERATION_TEMPLATE_WITH_HISTORY.format(
            history=history_str, context=context_str, question=question,
        )
    else:
        prompt = GENERATION_TEMPLATE.format(context=context_str, question=question)

    # ... 以降は同じ（Gemini呼び出し + structured output） ...

- スレッド内での会話では、前のやり取りの文脈を踏まえて回答してください
- ただし、前の回答を繰り返さず、新しい質問に対する回答に集中してください
- 過去の回答を参照する場合でも、必ずナレッジベースの情報に基づいてください

from bot.history import get_history, save_turn
from bot.rag import reformulate_query

# パイプライン本体（主要部分のみ抜粋）
history = get_history(thread_name)

# Step 2: 検索クエリ作成（履歴があれば再構成）
if history:
    search_query = reformulate_query(user_text, history)
else:
    search_query = user_text

# Step 3: KB検索（再構成後のクエリで検索）
contexts = retrieve_context(search_query)

# Step 4: 回答生成（元の質問文 + 履歴を渡す）
result = generate_answer(user_text, contexts, history=history or None)

# フォローアップのヒント（初回応答時のみ）
if not history and not is_fallback:
    state.content_paragraphs.append(FOLLOWUP_HINT)

# 成功時のみ履歴を保存
if answer_text is not None:
    save_turn(thread_name, user_text, answer_text)

# 1. Firestore APIを有効化
gcloud services enable firestore.googleapis.com --project=YOUR_PROJECT_ID

# 2. データベースを作成（Nativeモード）
gcloud firestore databases create \
    --location=asia-northeast1 \
    --type=firestore-native \
    --project=YOUR_PROJECT_ID

# 3. TTLポリシーを設定（24時間後に自動削除）
gcloud firestore fields ttls update updated_at \
    --collection-group=chat_threads \
    --enable-ttl \
    --project=YOUR_PROJECT_ID