取り組みの背景 — なぜ「長期記憶」が必要なのか
Gemini API の startChat() を使えば、短い会話は簡単に実装できます。しかし本番環境で稼働するチャットボットでは、数十ターンにわたる長い会話、ユーザーごとのセッション管理、サーバー再起動後の会話再開といった課題に直面します。
Gemini 2.5 Pro のコンテキストウィンドウは最大 100 万トークンと非常に大きいものの、すべての会話履歴をそのまま送り続けるのはコスト面でもレイテンシ面でも現実的ではありません。ここではプロダクション環境で実際に動く長期記憶・セッション永続化の設計パターンを、コード例とともに体系的に解説します。
対象読者は、Gemini API の基本的なチャット実装を理解している中〜上級の開発者です。基本的なマルチターンチャットの実装については、Gemini API マルチターンチャット入門をご参照ください。
会話メモリの3層アーキテクチャ
大規模チャットボットのメモリ管理では、以下の3層構造が実践的に有効です。
第1層: ワーキングメモリ（直近の会話）
直近 N ターン（通常5〜10ターン）の会話をそのまま保持します。ユーザーが「さっき言ったこと」に言及する際の即座の応答に不可欠です。
第2層: 要約メモリ（圧縮された過去の会話）
ワーキングメモリから押し出された古い会話を、Gemini 自身を使って要約・圧縮します。数百ターンの会話を数百トークンに凝縮することで、長期的な文脈を維持しつつトークンコストを劇的に削減します。
第3層: ファクトメモリ（抽出された永続的な事実）
会話から抽出されたユーザーに関する永続的な事実（名前、好み、過去の決定事項など）を構造化データとして保存します。セッションをまたいでも参照でき、パーソナライゼーションの基盤になります。
# メモリマネージャーの基本構造
import json
from google import genai
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
class ConversationMemory:
"""3層メモリアーキテクチャの実装"""
def __init__(self, working_memory_limit=10):
self.working_memory = [] # 第1層: 直近の会話
self.summary = "" # 第2層: 過去の要約
self.facts = {} # 第3層: 永続的な事実
self.working_memory_limit = working_memory_limit
self.total_turns = 0
def add_turn(self, role: str, content: str):
"""会話ターンを追加し、必要に応じてメモリを圧縮"""
self.working_memory.append({"role": role, "parts": [{"text": content}]})
self.total_turns += 1
# ワーキングメモリが上限を超えたら要約を生成
if len(self.working_memory) > self.working_memory_limit:
self._compress_memory()
def _compress_memory(self):
"""古い会話を要約に圧縮する"""
# 最も古い半分を取り出して要約
half = len(self.working_memory) // 2
old_turns = self.working_memory[:half]
self.working_memory = self.working_memory[half:]
old_text = "\n".join(
f"{t['role']}: {t['parts'][0]['text']}" for t in old_turns
)
# Gemini を使って要約を生成
prompt = f"""以下の会話履歴を、重要な情報を失わないように
3〜5文で要約してください。既存の要約がある場合はそれも統合してください。
既存の要約:
{self.summary if self.summary else "(なし)"}
新しい会話:
{old_text}
統合要約:"""
response = client.models.generate_content(
model="gemini-2.5-flash", # 要約にはFlashでコスト削減
contents=prompt
)
self.summary = response.text
def build_context(self) -> list:
"""API に送信するコンテキストを構築"""
context = []
# 要約があれば system instruction 的に先頭に配置
if self.summary:
context.append({
"role": "user",
"parts": [{"text": f"[会話の背景情報] {self.summary}"}]
})
context.append({
"role": "model",
"parts": [{"text": "承知しました。この背景情報を踏まえて会話を続けます。"}]
})
# ファクトメモリがあれば追加
if self.facts:
facts_text = "ユーザーについて: " + ", ".join(
f"{k}: {v}" for k, v in self.facts.items()
)
context.append({
"role": "user",
"parts": [{"text": f"[ユーザー情報] {facts_text}"}]
})
context.append({
"role": "model",
"parts": [{"text": "ユーザー情報を把握しました。"}]
})
# ワーキングメモリを追加
context.extend(self.working_memory)
return contextこのアーキテクチャにより、100ターンを超える会話でも、API に送信するトークン数を常に一定範囲内に保つことができます。
トークン予算制御 — コストとレイテンシの最適化
プロダクション環境では、1リクエストあたりのトークン数を厳密に管理する必要があります。Gemini API の count_tokens() メソッドを活用した予算制御の実装を見ていきましょう。
class TokenBudgetManager:
"""トークン予算を管理し、コンテキストウィンドウの使用量を最適化"""
def __init__(
self,
model_name: str = "gemini-2.5-pro",
max_input_tokens: int = 30000, # 入力トークン上限
reserved_output_tokens: int = 4000 # 出力用に予約
):
self.model_name = model_name
self.max_input_tokens = max_input_tokens
self.reserved_output_tokens = reserved_output_tokens
self.client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
def count_tokens(self, contents: list) -> int:
"""コンテンツのトークン数をカウント"""
response = self.client.models.count_tokens(
model=self.model_name,
contents=contents
)
return response.total_tokens
def fit_to_budget(self, memory: ConversationMemory) -> list:
"""メモリの内容をトークン予算内に収める"""
context = memory.build_context()
token_count = self.count_tokens(context)
# 予算内ならそのまま返す
if token_count <= self.max_input_tokens:
return context
# 予算超過: ワーキングメモリを段階的に圧縮
while token_count > self.max_input_tokens and len(memory.working_memory) > 2:
memory._compress_memory()
context = memory.build_context()
token_count = self.count_tokens(context)
return context
def estimate_cost(self, input_tokens: int, output_tokens: int) -> dict:
"""概算コストを計算（USD）
※ 料金は2026年3月時点の参考値"""
# Gemini 2.5 Pro の料金体系（参考値）
input_cost = (input_tokens / 1_000_000) * 1.25
output_cost = (output_tokens / 1_000_000) * 10.00
return {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"input_cost_usd": round(input_cost, 6),
"output_cost_usd": round(output_cost, 6),
"total_cost_usd": round(input_cost + output_cost, 6)
}トークン予算の設計指針
トークン予算は、コスト、レイテンシ、会話品質のバランスで決まります。以下に実践的な指針をまとめます。
- 入力上限 30,000トークン: 一般的なチャットボットでは十分な文脈量。コストとレスポンス速度のバランスが良い
- 入力上限 100,000トークン: ドキュメント分析や専門的なアドバイザーなど、深い文脈が必要なユースケース向け
- 出力予約 2,000〜8,000トークン: 回答の長さに応じて調整。FAQボットなら2,000、コーディングアシスタントなら8,000が目安
- 要約タイミング: ワーキングメモリが入力上限の60%を超えたら要約を開始するのが安全
セッション永続化 — Redis / Cloudflare KV 実装
サーバー再起動やスケールアウト時にも会話を継続するには、セッションデータを外部ストレージに永続化する必要があります。
Redis を使ったセッション永続化
import redis
import json
import time
class RedisSessionStore:
"""Redis ベースのセッション永続化"""
def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 86400):
self.redis = redis.from_url(redis_url)
self.ttl = ttl # セッション有効期限（デフォルト24時間）
def save_session(self, session_id: str, memory: ConversationMemory):
"""セッションを Redis に保存"""
data = {
"working_memory": memory.working_memory,
"summary": memory.summary,
"facts": memory.facts,
"total_turns": memory.total_turns,
"updated_at": time.time()
}
self.redis.setex(
f"chat:session:{session_id}",
self.ttl,
json.dumps(data, ensure_ascii=False)
)
def load_session(self, session_id: str) -> ConversationMemory | None:
"""Redis からセッションを復元"""
raw = self.redis.get(f"chat:session:{session_id}")
if raw is None:
return None
data = json.loads(raw)
memory = ConversationMemory()
memory.working_memory = data["working_memory"]
memory.summary = data["summary"]
memory.facts = data.get("facts", {})
memory.total_turns = data.get("total_turns", 0)
return memory
def delete_session(self, session_id: str):
"""セッションを削除"""
self.redis.delete(f"chat:session:{session_id}")
def extend_ttl(self, session_id: str):
"""アクティブなセッションの有効期限を延長"""
self.redis.expire(f"chat:session:{session_id}", self.ttl)Cloudflare KV を使ったセッション永続化（エッジ対応）
Cloudflare Workers 環境でチャットボットを動かす場合は、KV ストアが適しています。
// Cloudflare Workers + KV でのセッション管理
interface SessionData {
workingMemory: Array<{ role: string; parts: Array<{ text: string }> }>;
summary: string;
facts: Record<string, string>;
totalTurns: number;
updatedAt: number;
}
export class KVSessionStore {
constructor(private kv: KVNamespace, private ttlSeconds: number = 86400) {}
async saveSession(sessionId: string, data: SessionData): Promise<void> {
await this.kv.put(
`chat:session:${sessionId}`,
JSON.stringify(data),
{ expirationTtl: this.ttlSeconds }
);
}
async loadSession(sessionId: string): Promise<SessionData | null> {
const raw = await this.kv.get(`chat:session:${sessionId}`);
if (!raw) return null;
return JSON.parse(raw) as SessionData;
}
async deleteSession(sessionId: string): Promise<void> {
await this.kv.delete(`chat:session:${sessionId}`);
}
}
// Worker のエントリポイントでの使用例
export default {
async fetch(request: Request, env: Env): Promise<Response> {
const store = new KVSessionStore(env.CHAT_SESSIONS);
const sessionId = getCookieValue(request, "session_id");
// セッション復元
const existing = await store.loadSession(sessionId);
// ... Gemini API 呼び出し ...
// セッション保存
await store.saveSession(sessionId, updatedData);
return new Response(JSON.stringify(result), {
headers: { "Content-Type": "application/json" }
});
}
};ファクト抽出 — 会話からユーザー情報を自動学習
第3層のファクトメモリを効果的に構築するには、会話の中からユーザーに関する永続的な事実を自動抽出する仕組みが必要です。
class FactExtractor:
"""会話からユーザーの永続的な事実を抽出"""
EXTRACTION_PROMPT = """以下の会話から、ユーザーに関する永続的な事実を
JSON形式で抽出してください。一時的な話題は含めないでください。
抽出対象:
- 名前、職業、所属
- 技術スキル、使用言語
- プロジェクトの目的や制約
- 明示された好み・要望
既存の事実（更新・補完してください）:
{existing_facts}
会話:
{conversation}
JSON形式で出力:
{{"name": "...", "role": "...", "skills": ["..."], ...}}"""
def __init__(self):
self.client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
def extract(
self,
conversation: list,
existing_facts: dict
) -> dict:
"""会話から事実を抽出して既存の事実とマージ"""
conv_text = "\n".join(
f"{t['role']}: {t['parts'][0]['text']}" for t in conversation
)
prompt = self.EXTRACTION_PROMPT.format(
existing_facts=json.dumps(existing_facts, ensure_ascii=False),
conversation=conv_text
)
response = self.client.models.generate_content(
model="gemini-2.5-flash",
contents=prompt,
config={
"response_mime_type": "application/json"
}
)
try:
new_facts = json.loads(response.text)
# 既存の事実とマージ（新しい情報で上書き）
merged = {**existing_facts, **new_facts}
return merged
except json.JSONDecodeError:
return existing_factsファクト抽出は毎ターン実行する必要はありません。5〜10ターンごと、またはセッション終了時にバッチ実行するのがコスト効率の良い方法です。
マルチユーザー・マルチセッション管理
本番環境では、数千〜数万のユーザーが同時にチャットを行う状況に対応する必要があります。
import uuid
from datetime import datetime
class ChatSessionManager:
"""マルチユーザー対応のセッション管理"""
def __init__(self, store: RedisSessionStore):
self.store = store
self.budget_manager = TokenBudgetManager()
self.fact_extractor = FactExtractor()
def create_session(self, user_id: str) -> str:
"""新しいチャットセッションを作成"""
session_id = f"{user_id}:{uuid.uuid4().hex[:12]}"
memory = ConversationMemory(working_memory_limit=10)
# ユーザーの過去のファクトメモリがあれば引き継ぐ
user_facts = self._load_user_facts(user_id)
if user_facts:
memory.facts = user_facts
self.store.save_session(session_id, memory)
return session_id
def chat(self, session_id: str, user_message: str) -> str:
"""チャットメッセージを処理して応答を返す"""
# セッション復元
memory = self.store.load_session(session_id)
if memory is None:
raise ValueError(f"Session not found: {session_id}")
# ユーザーメッセージを追加
memory.add_turn("user", user_message)
# トークン予算内にコンテキストを調整
context = self.budget_manager.fit_to_budget(memory)
# Gemini API 呼び出し
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=context,
config={
"system_instruction": self._build_system_instruction(memory),
"temperature": 0.7,
"max_output_tokens": 4000
}
)
assistant_message = response.text
memory.add_turn("model", assistant_message)
# 定期的にファクト抽出（10ターンごと）
if memory.total_turns % 10 == 0:
memory.facts = self.fact_extractor.extract(
memory.working_memory, memory.facts
)
# セッション保存
self.store.save_session(session_id, memory)
self.store.extend_ttl(session_id)
return assistant_message
def _build_system_instruction(self, memory: ConversationMemory) -> str:
"""System Instruction を動的に構築"""
base = "あなたは親切で知識豊富なアシスタントです。"
if memory.facts:
facts_str = ", ".join(f"{k}: {v}" for k, v in memory.facts.items())
base += f"\n\nユーザーについて把握している情報: {facts_str}"
if memory.summary:
base += f"\n\nこれまでの会話の要約: {memory.summary}"
return base
def _load_user_facts(self, user_id: str) -> dict:
"""ユーザーの永続的なファクトメモリを読み込む"""
raw = self.store.redis.get(f"user:facts:{user_id}")
if raw:
return json.loads(raw)
return {}
def save_user_facts(self, user_id: str, session_id: str):
"""セッション終了時にユーザーのファクトメモリを永続化"""
memory = self.store.load_session(session_id)
if memory and memory.facts:
self.store.redis.set(
f"user:facts:{user_id}",
json.dumps(memory.facts, ensure_ascii=False)
)障害復旧とグレースフルデグラデーション
本番環境では、Redis の接続断、Gemini API のレート制限、ネットワーク障害など、さまざまな障害に備える必要があります。
import logging
from functools import wraps
from tenacity import retry, stop_after_attempt, wait_exponential
logger = logging.getLogger(__name__)
class ResilientChatBot:
"""障害に強いチャットボット実装"""
def __init__(self, session_manager: ChatSessionManager):
self.session_manager = session_manager
self.fallback_memory = {} # インメモリのフォールバック
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10)
)
def _call_gemini_with_retry(self, context: list, system_instruction: str) -> str:
"""リトライ付きの Gemini API 呼び出し"""
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=context,
config={
"system_instruction": system_instruction,
"temperature": 0.7,
"max_output_tokens": 4000
}
)
return response.text
def chat_with_fallback(self, session_id: str, user_message: str) -> str:
"""フォールバック機構付きのチャット処理"""
try:
return self.session_manager.chat(session_id, user_message)
except redis.ConnectionError:
logger.warning("Redis connection failed, using in-memory fallback")
return self._fallback_chat(session_id, user_message)
except Exception as e:
if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e):
logger.warning("Rate limited, using smaller model")
return self._reduced_model_chat(session_id, user_message)
raise
def _fallback_chat(self, session_id: str, user_message: str) -> str:
"""Redis が使えない場合のインメモリフォールバック"""
if session_id not in self.fallback_memory:
self.fallback_memory[session_id] = ConversationMemory()
memory = self.fallback_memory[session_id]
memory.add_turn("user", user_message)
context = memory.build_context()
response_text = self._call_gemini_with_retry(context, "あなたは親切なアシスタントです。")
memory.add_turn("model", response_text)
return response_text
def _reduced_model_chat(self, session_id: str, user_message: str) -> str:
"""レート制限時に小さいモデルにフォールバック"""
memory = self.session_manager.store.load_session(session_id)
if memory is None:
memory = ConversationMemory()
memory.add_turn("user", user_message)
# 直近3ターンだけに絞ってFlashモデルで応答
short_context = memory.working_memory[-6:] # 最後の3往復
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
response = client.models.generate_content(
model="gemini-2.5-flash", # 軽量モデルにフォールバック
contents=short_context,
config={"temperature": 0.7, "max_output_tokens": 2000}
)
response_text = response.text
memory.add_turn("model", response_text)
self.session_manager.store.save_session(session_id, memory)
return response_textコンテキストキャッシング API との統合
Gemini API のコンテキストキャッシング機能を活用すると、繰り返し送信する System Instruction や共通のプレフィックスコンテンツのコストを大幅に削減できます。長期記憶アーキテクチャとの組み合わせ方を見てみましょう。
from google import genai
from google.genai import types
class CachedContextManager:
"""コンテキストキャッシングを活用したコスト最適化"""
def __init__(self):
self.client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
self.cached_content = None
def create_cached_context(
self,
system_instruction: str,
shared_knowledge: str
):
"""共通コンテキストをキャッシュ化"""
self.cached_content = self.client.caches.create(
model="gemini-2.5-pro",
config=types.CreateCachedContentConfig(
display_name="chatbot-shared-context",
system_instruction=system_instruction,
contents=[{
"role": "user",
"parts": [{"text": shared_knowledge}]
}, {
"role": "model",
"parts": [{"text": "理解しました。この知識ベースを活用して回答します。"}]
}],
ttl="3600s" # 1時間のキャッシュ
)
)
return self.cached_content.name
def chat_with_cache(self, user_context: list) -> str:
"""キャッシュ済みコンテキストを使って応答を生成"""
if self.cached_content is None:
raise ValueError("キャッシュが未作成です")
response = self.client.models.generate_content(
model="gemini-2.5-pro",
contents=user_context,
config=types.GenerateContentConfig(
cached_content=self.cached_content.name
)
)
return response.textコンテキストキャッシングの詳しい活用法については、Gemini API コンテキストキャッシングでコスト最適化で詳しく解説しています。
本番デプロイ時の監視とメトリクス
チャットボットを本番運用する際には、会話品質とシステムパフォーマンスの両方を監視する仕組みが欠かせません。
import time
from dataclasses import dataclass, field
@dataclass
class ChatMetrics:
"""チャットボットの運用メトリクス"""
total_sessions: int = 0
active_sessions: int = 0
total_messages: int = 0
total_input_tokens: int = 0
total_output_tokens: int = 0
avg_response_time_ms: float = 0.0
compression_events: int = 0 # メモリ圧縮が発生した回数
fallback_events: int = 0 # フォールバックが発生した回数
error_count: int = 0
_response_times: list = field(default_factory=list)
def record_response(self, duration_ms: float, input_tokens: int, output_tokens: int):
self.total_messages += 1
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
self._response_times.append(duration_ms)
# 直近100件の平均レスポンスタイム
recent = self._response_times[-100:]
self.avg_response_time_ms = sum(recent) / len(recent)
def to_dict(self) -> dict:
return {
"total_sessions": self.total_sessions,
"active_sessions": self.active_sessions,
"total_messages": self.total_messages,
"total_input_tokens": self.total_input_tokens,
"total_output_tokens": self.total_output_tokens,
"avg_response_time_ms": round(self.avg_response_time_ms, 2),
"compression_events": self.compression_events,
"fallback_events": self.fallback_events,
"error_rate": round(
self.error_count / max(self.total_messages, 1) * 100, 2
)
}監視すべき重要なメトリクスとして、平均レスポンスタイム（目標: 2秒以内）、メモリ圧縮頻度（高すぎるとワーキングメモリの上限を上げるべき）、フォールバック発生率（5%以下が理想）、セッションあたりの平均ターン数（ユーザーエンゲージメントの指標）があります。
まとめ
大規模チャットボットの長期記憶管理は、単に会話履歴を送り続けるだけでは成り立ちません。3層メモリアーキテクチャ（ワーキングメモリ・要約メモリ・ファクトメモリ）、トークン予算制御、セッション永続化、障害復旧パターンを組み合わせることで、コスト効率が高く、スケーラブルで、ユーザー体験の優れたプロダクション級チャットボットが実現できます。
Gemini API の強力なコンテキストウィンドウとコンテキストキャッシング機能を最大限に活かしつつ、現実のインフラ制約に対応する設計が、成功するAIチャットプロダクトの鍵です。