Have Firestore maintain conversation history in a Google Chat Bot to handle follow-up questions

Introduction

While operating a RAG-based Google Chat Bot, I noticed that when users asked "Can you tell me more about that?", the Bot would lose context.

The cause was simple: the Bot was sending only the latest message to the LLM each time, with no reference to past exchanges within the thread. In other words, when told "tell me more about that", it had no idea what "that" referred to.

In this article, I'll introduce an implementation that saves conversation history in Firestore to handle follow-up questions.

Prerequisites & Environment

• Google Cloud Functions (2nd gen) + Python 3.14
• Gemini 2.5 Flash (google-genai SDK)
• Vertex AI RAG Engine (knowledge base search)
• Google Chat API (Bot responses)

Problem Definition

The existing pipeline works as follows:

Receive message → KB search → Generate answer → Send

With this setup, the following exchange within a thread is not possible:

User: How do I configure VPN?
Bot: (Responds with VPN steps)
User: Can you explain in more detail?
Bot: No relevant information found ← Searches KB using "Can you explain in more detail?"

Naturally, searching RAG with "Can you explain in more detail?" returns no hits. What's needed is:

1. Saving and retrieving conversation history — retain past exchanges
2. Query reformulation — convert "Can you explain in more detail?" → "Detailed VPN configuration steps"
3. Answer generation with history — pass past context to the LLM

Storage Selection

I considered three options for storing conversation history.

Google Chat API (spaces.messages.list)

Ideally, I could retrieve messages within a thread using the Chat API, but spaces.messages.list does not support the chat.bot scope. The chat.messages.readonly scope is required, which needs Workspace admin approval. Since it can't be used by a standalone Bot, I ruled this out.

Cloud Logging

The Bot already logs queries and answers to Cloud Logging as usage logs. However, Cloud Logging has an ingestion delay of 30 seconds to 5 minutes, and queries also take 2–10 seconds. It's not suitable for interactive conversations.

Firestore (selected)

Read latency is in the tens of milliseconds, and it supports automatic deletion via TTL policies. It fits comfortably within the free tier (1 GiB storage, 50,000 reads/day) with virtually no additional cost.

Architecture

Here is the revised pipeline:

Receive message → Retrieve history → Reformulate query → KB search → Generate answer (+history) → Send → Save history

The number of visible pipeline steps remains the same as the existing four steps; history retrieval and saving are added before and after the steps, and query reformulation is incorporated into the "Creating search query" step.

Firestore Data Model

One document per thread is stored in the chat_threads collection.

doc_id: "spaces_ABC_threads_XYZ"
{
  turns: [
    { role: "user", text: "How do I configure VPN?", timestamp: ... },
    { role: "model", text: "The steps for VPN connection are as follows...", timestamp: ... },
    ...
  ],
  expired_at: <current time + 24 hours>   // for TTL
}

• doc_id is the thread name (spaces/ABC/threads/XYZ) with / replaced by _ (since / cannot be used in Firestore document IDs)
• A TTL anchor is set on expired_at, triggering automatic deletion after 24 hours
• No upper limit on turn count — naturally constrained by TTL (IT support conversations are typically 2–4 turns)

Implementation

1. Firestore CRUD (bot/history.py)

The Firestore client is initialized using the singleton pattern, the same pattern used for the existing Chat API and genai clients.

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)

The key point is graceful degradation. Both get_history and save_turn simply log errors and do not raise exceptions when Firestore encounters an error. Even if history cannot be retrieved, the Bot continues to operate normally (without history).

2. Query Reformulation

Sending follow-up questions directly to RAG search returns no meaningful results. We use an LLM to rewrite them into self-contained queries.

REFORMULATE_TEMPLATE = """\
The following is a conversation history from an internal IT support chat.

## Conversation History
{history}

## Latest Question
{question}

## Task
If the latest question depends on the context of the conversation history (e.g., "that", "that method", "anything else?"), \
rewrite it into a self-contained question suitable for knowledge base search.
If the question is already self-contained, return it as-is.
Return only the rewritten question. No explanation or preamble is needed.
"""

def _format_history_text(history: list[dict]) -> str:
    lines = []
    for turn in history:
        role_label = "User" if turn["role"] == "user" else "Assistant"
        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

The reformulated query is used only for RAG search. The user's original question is passed as-is to answer generation. This allows the LLM to understand the context of "Can you explain in more detail?" while using an appropriate query for search.

3. Answer Generation with History

When conversation history exists, we switch to a prompt template that includes the history.

GENERATION_TEMPLATE_WITH_HISTORY = """\
Please answer the user's question based on the following knowledge base information and conversation history.

## Conversation History

{history}

## Knowledge Base Search Results

{context}

## User's Question

{question}

## Answer
"""

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

    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)

    # ... the rest is the same (Gemini call + structured output) ...

Instructions for handling conversation history have also been added to the system prompt:

- When conversing within a thread, respond taking into account the context of previous exchanges
- However, do not repeat previous answers; focus on responding to the new question
- Even when referencing past answers, always base your response on knowledge base information

4. Integration into the Pipeline

The main changes in worker.py:

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

# Main pipeline (key parts only)
history = get_history(thread_name)

# Step 2: Create search query (reformulate if history exists)
if history:
    search_query = reformulate_query(user_text, history)
else:
    search_query = user_text

# Step 3: KB search (search using reformulated query)
contexts = retrieve_context(search_query)

# Step 4: Generate answer (pass original question + history)
result = generate_answer(user_text, contexts, history=history or None)

# Follow-up hint (only on first response)
if not history and not is_fallback:
    state.content_paragraphs.append(FOLLOWUP_HINT)

# Save history only on success
if answer_text is not None:
    save_turn(thread_name, user_text, answer_text)

Only on the first response, a hint saying "You can also ask additional questions in this thread (within 24 hours)" is displayed to let users know that follow-ups are possible. It is not shown on subsequent responses (since the user is already following up, it's unnecessary).

GCP Setup

Firestore setup is completed with three commands:

# 1. Enable Firestore API
gcloud services enable firestore.googleapis.com --project=YOUR_PROJECT_ID

# 2. Create database (Native mode)
gcloud firestore databases create \
    --location=asia-northeast1 \
    --type=firestore-native \
    --project=YOUR_PROJECT_ID

# 3. Set TTL policy (automatic deletion after 24 hours)
gcloud firestore fields ttls update updated_at \
    --collection-group=chat_threads \
    --enable-ttl \
    --project=YOUR_PROJECT_ID

Regarding IAM, if the Cloud Functions default service account has the Editor role, the datastore.user permission required for Firestore read/write is already included, so no additional configuration is needed.

Note that creating a new database requires the datastore.owner role. The Editor role has insufficient permissions, so you'll need to ask a project owner or have the datastore.owner role temporarily granted.

Verification

After deployment, I tested follow-up questions in Google Chat.

User: How do I configure VPN?
Bot: The steps for VPN connection are as follows... (detailed steps provided)

User: Can you explain in more detail?
Bot: Let me explain VPN setup in more detail... (supplementary information building on the previous answer)

In the "Creating search query" step of the progressive card, the reformulated query (e.g., "Detailed VPN configuration steps") is displayed, confirming that the query is being properly rewritten.

In the Firestore console, I also confirmed that a document was created in the chat_threads collection.

Summary

By adding Firestore-based conversation history to the Google Chat Bot, it can now handle follow-up questions within a thread.

Here's a summary of the key implementation points:

• Firestore for storage — low latency, automatic cleanup via TTL, sufficient within the free tier
• Query reformulation — use an LLM to convert follow-up questions into self-contained queries before RAG search
• Use original question for answer generation — the reformulated query is used only for RAG search; the user's original question is passed to answer generation
• Graceful degradation — the Bot continues to operate normally during Firestore failures (falls back to no history)
• No turn count limit — TTL (24 hours) acts as a natural limit; IT support conversations are short, so turn management is unnecessary

Running RAG search on every follow-up question may seem wasteful at first glance, but considering the possibility that users might switch to a new topic, always going through RAG is simpler and safer. I also considered a design that classifies intent to branch between "clarification" and "new topic", but the risk of misclassification and the added complexity did not justify the benefits, so I adopted the simple "always use RAG" approach.