Have a Google Chat Bot retain conversation history in Firestore 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, without referencing any 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 Analysis

The existing pipeline followed this flow:

Receive message → KB search → Generate answer → Send

With this structure, the following type of exchange within a thread was impossible:

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

Naturally, a RAG search on the string "Can you tell me more" returns no results. What's needed is:

1. Save & retrieve conversation history — Retain past exchanges
2. Query reformulation — Convert "Can you tell me more?" → "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)

Retrieving messages within a thread via the Chat API would be ideal, but spaces.messages.list does not support the chat.bot scope. It requires the chat.messages.readonly scope, which needs Workspace administrator approval. Since it can't be used by a Bot alone, this option was ruled out.

Cloud Logging

The Bot already records queries and responses 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 (chosen)

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 steps in the pipeline remains the same four steps as before; history retrieval/saving is 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, and the document is automatically deleted after 24 hours
• No upper limit on the number of turns — TTL provides a natural limit (IT support conversations typically run 2–4 turns)

Implementation

1. Firestore CRUD (bot/history.py)

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

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 only log a warning on Firestore errors without raising exceptions. Even if history cannot be retrieved, the Bot continues operating normally (without history).

2. Query Reformulation

Using follow-up questions directly for RAG search returns no meaningful results. An LLM is used 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", "what else?", etc.), \
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 text is passed as-is to answer generation. This allows the LLM to understand the context of "tell me more," while ensuring an appropriate query is used for search.

3. Answer Generation with History

When conversation history exists, it switches to a prompt template that includes the history.

GENERATION_TEMPLATE_WITH_HISTORY = """\
Please answer the user's question using the following knowledge base information and conversation history as reference.

## 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, answer while taking into account the context of previous exchanges
- However, do not repeat previous answers; focus on answering the new question
- Even when referencing past answers, always base your response on information from the knowledge base

4. Integration into the Pipeline

The main changes to worker.py:

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

# Pipeline body (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 with reformulated query)
contexts = retrieve_context(search_query)

# Step 4: Generate answer (pass original question text + 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 inform users that follow-up is possible. It is not shown from the second response onward (since the user is already following up, it's unnecessary).

GCP Setup

Firestore setup is completed with three commands.

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

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

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

For IAM, if the Cloud Functions default service account has the Editor role, it already includes the datastore.user permission required for Firestore read/write, 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 to grant datastore.owner temporarily.

Verification

After deployment, I tested actual 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 tell me more?
Bot: Let me explain VPN configuration 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 rewritten appropriately.

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

Summary

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

Here is a summary of the key implementation points:

• Firestore for storage — Low latency, automatic cleanup via TTL, sufficient free tier
• Query reformulation — Follow-up questions are converted by an LLM into self-contained queries before RAG search
• Answer generation with the original question — The reformulated query is used only for RAG search; the user's original question is passed to answer generation
• Graceful degradation — The Bot continues normal operation even 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 a 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 uses intent classification to branch between "clarification" and "new topic," but the benefits didn't justify the risk of misclassification and the added complexity, so I adopted the simple "always use RAG" approach.