Gemini APIの会話履歴をローリングサマリで圧縮する — 数百ターン続くチャットボットの設計

# pip install google-genai==1.* tiktoken-style
from google import genai
from google.genai import types
import os
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
MODEL = "gemini-2.5-flash"  # 要約は安くて速い Flash で十分
 
# 圧縮を発動するトークン閾値（コスト基準で決める）
COMPRESS_THRESHOLD = 8000
# 逐語で残す直近メッセージ数（ユーザー＋モデルで2の倍数が扱いやすい）
KEEP_RECENT = 6
 
def count_tokens(history: list[types.Content]) -> int:
    """会話履歴の合計トークン数を返します。"""
    resp = client.models.count_tokens(model=MODEL, contents=history)
    return resp.total_tokens
 
def summarize(messages: list[types.Content]) -> str:
    """古い会話部分を1段落の要約に変換します。"""
    # 要約専用プロンプト。「事実」「前提」「未解決の質問」を必ず保持させる
    transcript = "\n".join(
        f"{m.role}: {m.parts[0].text}" for m in messages if m.parts
    )
    instruction = (
        "次の会話を、後続のチャットで失ってはいけない情報に絞って5〜8行で要約してください。"
        "ユーザーが共有した固有名詞・数値・好み・未解決の質問は必ず保持してください。"
        "会話の雰囲気や挨拶は省略して構いません。\n\n"
        f"---\n{transcript}\n---"
    )
    resp = client.models.generate_content(
        model=MODEL,
        contents=instruction,
        config=types.GenerateContentConfig(temperature=0.2),
    )
    return resp.text
 
def compress(history: list[types.Content]) -> list[types.Content]:
    """履歴を「要約 + 直近 KEEP_RECENT 件」に圧縮します。"""
    if len(history) <= KEEP_RECENT:
        return history
    old, recent = history[:-KEEP_RECENT], history[-KEEP_RECENT:]
    summary = summarize(old)
    summary_msg = types.Content(
        role="user",
        parts=[types.Part.from_text(text=f"[これまでの会話の要約]\n{summary}")],
    )
    return [summary_msg] + recent

def chat_turn(history, user_message):
    history.append(types.Content(
        role="user",
        parts=[types.Part.from_text(text=user_message)],
    ))
    if count_tokens(history) > COMPRESS_THRESHOLD:
        history = compress(history)
    resp = client.models.generate_content(model="gemini-2.5-pro", contents=history)
    history.append(resp.candidates[0].content)
    return history, resp.text

# 要約モデルは「世代名」ではなく安定したIDでピン留めする
SUMMARY_MODEL = "gemini-2.5-flash"  # 例: 検証済みの安定版を明示的に固定
 
def summarize_pinned(messages: list[types.Content]) -> tuple[str, str]:
    """要約テキストと、実際に応答したモデル名を返します。"""
    transcript = "\n".join(
        f"{m.role}: {m.parts[0].text}" for m in messages if m.parts
    )
    instruction = (
        "次の会話を、後続のチャットで失ってはいけない情報に絞って5〜8行で要約してください。"
        "固有名詞・数値・URL は原文のまま残してください。\n\n"
        f"---\n{transcript}\n---"
    )
    resp = client.models.generate_content(
        model=SUMMARY_MODEL,
        contents=instruction,
        config=types.GenerateContentConfig(temperature=0.2),
    )
    # 応答メタデータから実際に処理したモデルを取得（SDKのバージョン差を吸収）
    served = getattr(resp, "model_version", None) or SUMMARY_MODEL
    if not served.startswith(SUMMARY_MODEL):
        # 既定モデルの差し替えを検知。ここでログ＆アラートを上げる
        print(f"[warn] summary model drift: requested={SUMMARY_MODEL} served={served}")
    return resp.text, served

import re
 
URL_RE = re.compile(r"https?://[^\s)]+")
NUM_RE = re.compile(r"\b\d[\d,\.]{2,}\b")          # 予約番号・金額・桁数の大きい数値
PROPER_RE = re.compile(r"[A-Z][a-zA-Z0-9_\-]{2,}") # 製品名・ID・コード片の近似
 
def critical_tokens(text: str) -> set[str]:
    """落としてはいけない要素を原文から抽出します。"""
    toks = set(URL_RE.findall(text)) | set(NUM_RE.findall(text)) | set(PROPER_RE.findall(text))
    # 一般的すぎる語を除外（必要に応じて辞書化）
    return {t for t in toks if t.lower() not in {"http", "https", "the"}}
 
def retention_ratio(original: str, summary: str) -> float:
    """要約が原文の重要要素をどれだけ保持したかを 0.0〜1.0 で返します。"""
    must_keep = critical_tokens(original)
    if not must_keep:
        return 1.0
    survived = sum(1 for t in must_keep if t in summary)
    return survived / len(must_keep)
 
def compress_with_gate(history: list[types.Content], min_retention: float = 0.9):
    """生存率ゲートつきの圧縮。閾値を割ったら逐語保持にフォールバックします。"""
    if len(history) <= KEEP_RECENT:
        return history, 1.0
    old, recent = history[:-KEEP_RECENT], history[-KEEP_RECENT:]
    original = "\n".join(m.parts[0].text for m in old if m.parts)
    summary, _ = summarize_pinned(old)
    ratio = retention_ratio(original, summary)
    if ratio < min_retention:
        # 重要要素が落ちた → 要約を採用せず、直近を多めに逐語で残す保守側に倒す
        print(f"[warn] summary retention {ratio:.2f} < {min_retention}; keeping more verbatim")
        return history[-(KEEP_RECENT * 2):], ratio
    summary_msg = types.Content(
        role="user",
        parts=[types.Part.from_text(text=f"[これまでの会話の要約]\n{summary}")],
    )
    return [summary_msg] + recent, ratio

import threading
from collections import defaultdict
 
_locks: dict[str, threading.Lock] = defaultdict(threading.Lock)
 
# セッションごとの「絶対に消さない」構造化メモリ（DB列でもよい）
sticky_memory: dict[str, dict] = defaultdict(dict)
 
def build_prompt(session_id: str, history: list[types.Content]) -> list[types.Content]:
    """構造化メモリを先頭に固定し、その後ろに圧縮済み履歴を続けます。"""
    mem = sticky_memory[session_id]
    header = types.Content(
        role="user",
        parts=[types.Part.from_text(
            text="[セッション固定情報・要約対象外]\n" + str(mem)
        )],
    ) if mem else None
    return ([header] if header else []) + history
 
def chat_turn_safe(session_id: str, history, user_message):
    history.append(types.Content(
        role="user", parts=[types.Part.from_text(text=user_message)]))
    # 同一セッションの圧縮を直列化して二重圧縮を防ぐ
    if count_tokens(history) > COMPRESS_THRESHOLD:
        with _locks[session_id]:
            if count_tokens(history) > COMPRESS_THRESHOLD:  # ロック取得後に再確認
                history, _ = compress_with_gate(history)
    prompt = build_prompt(session_id, history)
    resp = client.models.generate_content(model="gemini-2.5-pro", contents=prompt)
    history.append(resp.candidates[0].content)
    return history, resp.text