Gemini API のチャット履歴を Redis で運用する — 本番で会話状態を失わない実装メモ

from google import genai
 
MAX_RECENT_TURNS = 10  # 直近 10 往復（= 20 メッセージ）はそのまま残す
SUMMARY_TRIGGER = 14   # 往復数がこれを超えたら要約を走らせる
 
def trim_history(history: list, client: genai.Client) -> list:
    """直近 N 往復を残し、それ以前を要約で置き換える"""
    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
    )
    # 要約は難しいタスクではないので軽量モデルで十分
    summary = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=(
            "次の会話を 300 字以内で要約してください。"
            "固有名詞・数値・未完了のタスク状態は必ず残してください。\n\n"
            f"{older_text}"
        ),
    ).text
 
    return [
        {"role": "user", "parts": [{"text": f"[これまでの会話の要約]\n{summary}"}]},
        {"role": "model", "parts": [{"text": "承知しました。続きを進めます。"}]},
        *recent,
    ]

SESSION_TTL = 60 * 60 * 24 * 7  # 7 日間、最終アクセスから起算
 
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):
    """セッション単位の排他ロックを取ってから 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):  # ロックは 10 秒で自動失効
            try:
                return fn()
            finally:
                # 自分のトークンと一致するときだけ削除（他人のロックを奪わない）
                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  # 作成時刻（デバッグ・分析用）
 
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]:
    """Gemini API が期待するフォーマットへ戻す"""
    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:
        # 履歴が引けないだけで会話自体は成立させる（縮退運転）
        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:
        # 保存失敗はメトリクスに数え、応答自体は返す
        pass  # ここで監視カウンタをインクリメントしておく

def chat_persistent(session_id: str, user_message: str) -> str:
    def _run():
        history = load_history(session_id)
        history = trim_history(history, client)  # 保存前に圧縮しておく
 
        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)