Running Gemini Chat History on Redis — Field Notes on Not Losing Conversation State in Production

from google import genai
 
MAX_RECENT_TURNS = 10  # keep the last 10 rounds (= 20 messages) verbatim
SUMMARY_TRIGGER = 14   # summarize once the round count exceeds this
 
def trim_history(history: list, client: genai.Client) -> list:
    """Keep the last N rounds; replace older ones with a summary."""
    turns = len(history) // 2
    if turns <= SUMMARY_TRIGGER:
        return history
 
    cut = (turns - MAX_RECENT_TURNS) * 2
    older, recent = history[:cut], history[cut:]
 
    older_text = "\n".join(
        f"{m['role']}: {m['parts'][0]['text']}" for m in older
    )
    # Summarizing is not a hard task, so a lightweight model is plenty
    summary = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=(
            "Summarize the conversation below in 300 characters or fewer. "
            "Always preserve proper nouns, numbers, and any unfinished task state.\n\n"
            f"{older_text}"
        ),
    ).text
 
    return [
        {"role": "user", "parts": [{"text": f"[Summary of the conversation so far]\n{summary}"}]},
        {"role": "model", "parts": [{"text": "Understood. Let's continue."}]},
        *recent,
    ]

SESSION_TTL = 60 * 60 * 24 * 7  # 7 days, measured from last access
 
r.set(f"chat:{session_id}", json.dumps(new_history), ex=SESSION_TTL)

import uuid
import time
 
def with_session_lock(session_id: str, fn, timeout_ms: int = 5000):
    """Acquire a per-session lock before running fn()."""
    lock_key = f"chat:lock:{session_id}"
    token = str(uuid.uuid4())
    deadline = time.time() + timeout_ms / 1000
 
    while time.time() < deadline:
        if r.set(lock_key, token, nx=True, px=10000):  # lock self-expires in 10s
            try:
                return fn()
            finally:
                # delete only when the token matches (never steal someone else's lock)
                release = (
                    "if redis.call('get', KEYS[1]) == ARGV[1] then "
                    "return redis.call('del', KEYS[1]) else return 0 end"
                )
                r.eval(release, 1, lock_key, token)
        time.sleep(0.05)
 
    raise RuntimeError("Failed to acquire chat session lock")

import time
from dataclasses import dataclass, asdict
from typing import Literal
 
@dataclass
class StoredMessage:
    role: Literal["user", "model"]
    text: str
    ts: float  # creation time (for debugging and analytics)
 
def to_stored(history) -> list[dict]:
    out = []
    for msg in history:
        text = "".join(p.text for p in msg.parts if hasattr(p, "text"))
        out.append(asdict(StoredMessage(role=msg.role, text=text, ts=time.time())))
    return out
 
def from_stored(stored: list[dict]) -> list[dict]:
    """Convert back to the format the Gemini API expects."""
    return [
        {"role": m["role"], "parts": [{"text": m["text"]}]}
        for m in stored
    ]

def load_history(session_id: str) -> list:
    try:
        raw = r.get(f"chat:{session_id}")
        return from_stored(json.loads(raw)) if raw else []
    except redis.RedisError:
        # only the history is missing; the conversation still proceeds (degraded mode)
        return []
 
def save_history(session_id: str, history) -> None:
    try:
        r.set(f"chat:{session_id}", json.dumps(to_stored(history)), ex=SESSION_TTL)
    except redis.RedisError:
        # count the failure as a metric, but still return the response
        pass  # increment a monitoring counter here
 

def chat_persistent(session_id: str, user_message: str) -> str:
    def _run():
        history = load_history(session_id)
        history = trim_history(history, client)  # compress before saving
 
        chat = client.chats.create(model="gemini-3.5-pro", history=history)
        response = chat.send_message(user_message)
 
        save_history(session_id, chat.get_history())
        return response.text
 
    return with_session_lock(session_id, _run)