セッションをまたぐコンテキストの永続化
ユーザー設定、会話履歴、タスク状態を保存・読み込みします。
「セッションをまたぐコンテキストの永続化」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
コンテキストを永続化する理由
永続化しなければ、エージェントのセッションは毎回最初から始まります。コンテキスト永続化エージェントは、ユーザーの名前、過去の会話、明示された設定、ユーザーが共有した事実を記憶します。これにより、継続性があり、パーソナルなインタラクションを実現できます。
セッション状態へのSQLiteの利用
SQLiteは、1人のユーザーが利用する個人向けエージェントに最適です。サーバーのオーバーヘッドがなく、ファイルベースで、信頼性にも優れています。会話履歴とユーザーデータをローカルデータベースに保存します。
import sqlite3
from datetime import datetime
def init_db(db_path: str = 'agent_memory.db') -> sqlite3.Connection:
conn = sqlite3.connect(db_path, check_same_thread=False)
conn.row_factory = sqlite3.Row # Dict-like access
conn.executescript('''
CREATE TABLE IF NOT EXISTS sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT UNIQUE NOT NULL,
user_id TEXT NOT NULL,
started_at TEXT,
last_active TEXT
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
timestamp TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS user_facts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
fact_key TEXT NOT NULL,
fact_value TEXT NOT NULL,
created_at TEXT,
UNIQUE(user_id, fact_key)
);
''')
conn.commit()
return conn
conn = init_db()
print('Database initialized')メッセージの保存と読み込み
すべてのユーザーメッセージとエージェントの返信を、セッションIDとともに保存します。セッションを再開するときは、直近のN件のメッセージを読み込み、会話のコンテキストを復元します。
import sqlite3
from datetime import datetime
def save_message(conn: sqlite3.Connection, session_id: str, role: str, content: str):
conn.execute(
'INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)',
(session_id, role, content, datetime.utcnow().isoformat())
)
conn.commit()
def load_recent_messages(conn: sqlite3.Connection, session_id: str, limit: int = 20) -> list:
rows = conn.execute(
'SELECT role, content FROM messages WHERE session_id = ? ORDER BY id DESC LIMIT ?',
(session_id, limit)
).fetchall()
# Reverse to get chronological order
return [{'role': r['role'], 'content': r['content']} for r in reversed(rows)]
def load_all_session_messages(conn: sqlite3.Connection, user_id: str, days: int = 30) -> list:
from datetime import timedelta
since = (datetime.utcnow() - timedelta(days=days)).isoformat()
rows = conn.execute(
'SELECT m.role, m.content, m.timestamp FROM messages m '
'JOIN sessions s ON s.session_id = m.session_id '
'WHERE s.user_id = ? AND m.timestamp >= ? ORDER BY m.id',
(user_id, since)
).fetchall()
return [dict(r) for r in rows]
if __name__ == '__main__':
conn = sqlite3.connect(':memory:')
conn.row_factory = sqlite3.Row
conn.execute('CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id TEXT, role TEXT, content TEXT, timestamp TEXT)')
save_message(conn, 'sess1', 'user', 'Hello agent')
save_message(conn, 'sess1', 'assistant', 'Hi! How can I help?')
for msg in load_recent_messages(conn, 'sess1'):
print(f"{msg['role']}: {msg['content']}")
セッション状態へのRedisの利用
Webエージェントや複数サーバーへのデプロイでは、SQLiteよりもRedisが適しています。セッションデータをTTL付きのJSONとして保存すると、古くなったセッションを自動的にクリーンアップできます。
import redis
import json
from datetime import datetime
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
SESSION_TTL = 3600 * 24 * 7 # 7 days
def save_session_state(session_id: str, state: dict):
key = f'session:{session_id}'
state['last_updated'] = datetime.utcnow().isoformat()
r.setex(key, SESSION_TTL, json.dumps(state))
def load_session_state(session_id: str) -> dict:
key = f'session:{session_id}'
raw = r.get(key)
if not raw:
return {}
state = json.loads(raw)
# Refresh TTL on access
r.expire(key, SESSION_TTL)
return state
def append_to_session_history(session_id: str, role: str, content: str):
state = load_session_state(session_id)
history = state.get('history', [])
history.append({'role': role, 'content': content})
# Keep last 50 messages
state['history'] = history[-50:]
save_session_state(session_id, state)
# Test
save_session_state('sess-abc', {'user_name': 'Alice', 'history': []})
append_to_session_history('sess-abc', 'user', 'Hello!')
state = load_session_state('sess-abc')
print('Session state:', state)起動時のユーザー設定の読み込み
新しいセッションが始まったら、ユーザーが保存した設定を読み込みます。設定には、言語、タイムゾーン、通知設定、エージェント固有の構成などが含まれます。これらをシステムプロンプトに注入します。
import sqlite3
def load_user_preferences(conn: sqlite3.Connection, user_id: str) -> dict:
rows = conn.execute(
'SELECT fact_key, fact_value FROM user_facts WHERE user_id = ?',
(user_id,)
).fetchall()
return {row['fact_key']: row['fact_value'] for row in rows}
def build_system_prompt_with_preferences(base_prompt: str, user_id: str, conn: sqlite3.Connection) -> str:
prefs = load_user_preferences(conn, user_id)
if not prefs:
return base_prompt
pref_lines = []
if 'name' in prefs:
pref_lines.append(f'The user\'s name is {prefs["name"]}.')
if 'timezone' in prefs:
pref_lines.append(f'The user is in timezone {prefs["timezone"]}.')
if 'language' in prefs:
pref_lines.append(f'Respond in {prefs["language"]}.')
if 'profession' in prefs:
pref_lines.append(f'The user is a {prefs["profession"]}.')
prefs_text = ' '.join(pref_lines)
return f'{base_prompt}\n\nUser context: {prefs_text}'
conn = init_db()
enhanced_prompt = build_system_prompt_with_preferences(
'You are a helpful assistant.',
'user-42',
conn
)
print('System prompt:', enhanced_prompt)会話サマリーの保存
会話履歴をすべて保存すると、トークンの消費が大きくなります。LLMによる要約を使って古い会話を圧縮したサマリーに変換し、過去の履歴についてはサマリーだけを読み込みます。
import openai
client = openai.OpenAI(api_key='sk-...')
def summarize_conversation(messages: list) -> str:
if not messages:
return ''
conversation_text = '\n'.join([
f'{m["role"].upper()}: {m["content"]}'
for m in messages
])
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{
'role': 'user',
'content': (
'Summarize this conversation in 2-3 sentences, '
'focusing on key facts, decisions, and user preferences revealed:\n\n'
f'{conversation_text}'
)
}]
)
return response.choices[0].message.content
def compress_old_history(conn: sqlite3.Connection, session_id: str, keep_recent: int = 10):
all_messages = load_recent_messages(conn, session_id, limit=1000)
if len(all_messages) <= keep_recent:
return
old_messages = all_messages[:-keep_recent]
summary = summarize_conversation(old_messages)
# Store summary as a special message
save_message(conn, session_id, 'summary', f'[Previous conversation summary]: {summary}')
print(f'Compressed {len(old_messages)} old messages into summary')ユーザー情報の抽出と保存
ユーザーが自分自身に関する事実を共有したら、それを抽出して保存します。これにより、今後のすべてのセッションにわたって維持される長期記憶を構築できます。
import openai
import json
client = openai.OpenAI(api_key='sk-...')
def extract_user_facts(message: str) -> dict:
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{
'role': 'user',
'content': (
f'Extract any personal facts the user revealed in this message: "{message}"\n'
'Return JSON with fields: name, location, profession, preferences, or other relevant facts. '
'Return empty dict {{}} if no facts were revealed.'
)
}],
response_format={'type': 'json_object'}
)
return json.loads(response.choices[0].message.content)
def store_user_facts(conn: sqlite3.Connection, user_id: str, facts: dict):
for key, value in facts.items():
if value: # Skip empty values
conn.execute(
'INSERT OR REPLACE INTO user_facts (user_id, fact_key, fact_value, created_at) VALUES (?, ?, ?, ?)',
(user_id, key, str(value), __import__('datetime').datetime.utcnow().isoformat())
)
conn.commit()
if facts:
print(f'Stored {len(facts)} facts for user {user_id}: {list(facts.keys())}')
conn = init_db()
facts = extract_user_facts('I am a software engineer in Berlin working on AI projects')
store_user_facts(conn, 'user-42', facts)
print('Facts extracted:', facts)現在のターンに関連するコンテキストの読み込み
エージェントが各ターンを開始するときに、最近のメッセージ、ユーザー設定、関連する事実からコンテキストを組み立てます。これにより、コンテキストウィンドウの上限を超えることなく、LLMに必要なすべての情報を渡せます。
def build_agent_context(conn: sqlite3.Connection, session_id: str, user_id: str, new_message: str) -> list:
messages = []
# Step 1: System prompt with user preferences
prefs = load_user_preferences(conn, user_id)
system_content = 'You are a helpful personal AI assistant.'
if prefs:
facts_text = ', '.join([f'{k}: {v}' for k, v in prefs.items()])
system_content += f' User context: {facts_text}'
messages.append({'role': 'system', 'content': system_content})
# Step 2: Load conversation history (last 15 messages)
history = load_recent_messages(conn, session_id, limit=15)
messages.extend(history)
# Step 3: Add the new user message
messages.append({'role': 'user', 'content': new_message})
return messages
conn = init_db()
context = build_agent_context(conn, 'sess-abc', 'user-42', 'What should I work on today?')
print(f'Context assembled: {len(context)} messages')
for m in context:
print(f' {m["role"]}: {m["content"][:60]}...')セッションをまたいだメモリの検索
長期間稼働するエージェントでは、最近の会話だけでなく、ベクトル検索を使って関連する過去の会話を探します。これにより、数か月前の関連コンテキストもエージェントが思い出せるようになります。
import openai
import chromadb
client = openai.OpenAI(api_key='sk-...')
chroma = chromadb.Client()
memory_collection = chroma.get_or_create_collection('user_memory')
def store_memory(user_id: str, content: str, metadata: dict):
emb_response = client.embeddings.create(
model='text-embedding-3-small', input=content
)
embedding = emb_response.data[0].embedding
import hashlib
doc_id = f'{user_id}_{hashlib.md5(content.encode()).hexdigest()[:8]}'
memory_collection.add(
ids=[doc_id],
embeddings=[embedding],
documents=[content],
metadatas=[{'user_id': user_id, **metadata}]
)
def retrieve_relevant_memories(user_id: str, current_query: str, top_k: int = 3) -> list:
emb_response = client.embeddings.create(
model='text-embedding-3-small', input=current_query
)
query_embedding = emb_response.data[0].embedding
results = memory_collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
where={'user_id': user_id}
)
return results['documents'][0] if results['documents'] else []
print('Cross-session memory retrieval defined')プライバシーとデータ管理
常時稼働エージェントは、機密性の高い個人データを保存します。データ保持期間の制限を実装し、ユーザーが自分のデータを削除できるようにします。また、会話内容をそのままアプリケーションログに記録してはいけません。
import sqlite3
from datetime import datetime, timedelta
def delete_user_data(conn: sqlite3.Connection, user_id: str):
'''Fully delete all data for a user (right to erasure).'''
conn.execute(
'DELETE FROM messages WHERE session_id IN (SELECT session_id FROM sessions WHERE user_id = ?)',
(user_id,)
)
conn.execute('DELETE FROM sessions WHERE user_id = ?', (user_id,))
conn.execute('DELETE FROM user_facts WHERE user_id = ?', (user_id,))
conn.commit()
print(f'All data deleted for user {user_id}')
def purge_old_messages(conn: sqlite3.Connection, retention_days: int = 90):
'''Remove messages older than retention period.'''
cutoff = (datetime.utcnow() - timedelta(days=retention_days)).isoformat()
cursor = conn.execute(
'DELETE FROM messages WHERE timestamp < ?', (cutoff,)
)
conn.commit()
print(f'Purged {cursor.rowcount} messages older than {retention_days} days')
# Run nightly retention cleanup
conn = init_db()
purge_old_messages(conn, retention_days=90)
print('Retention policy applied')セッション継続性の確認
長い間隔を空けてセッションを再開するときは、ユーザーがどれくらい離れていたか、その間に何が変わったかをエージェントに簡潔に伝えます。これにより、コンテキストが突然切り替わることによる混乱を防げます。
import sqlite3
from datetime import datetime, timedelta
def get_session_gap_context(conn: sqlite3.Connection, session_id: str) -> str:
row = conn.execute(
'SELECT last_active FROM sessions WHERE session_id = ?',
(session_id,)
).fetchone()
if not row or not row['last_active']:
return ''
last_active = datetime.fromisoformat(row['last_active'])
gap = datetime.utcnow() - last_active
if gap < timedelta(minutes=30):
return '' # Recent session, no gap context needed
elif gap < timedelta(hours=12):
return f'Note: The user was last active {int(gap.total_seconds() / 3600)} hours ago.'
elif gap < timedelta(days=7):
return f'Note: The user was last active {gap.days} days ago.'
else:
return f'Note: The user returns after {gap.days} days away. Welcome them back warmly.'
conn = init_db()
gap = get_session_gap_context(conn, 'sess-abc')
if gap:
print('Gap context:', gap)
else:
print('No gap context needed')理解度チェック：コンテキストの永続化
エージェントのセッションをまたいだコンテキストの永続化について理解度を確認します。
コンテキスト永続化のまとめ
コンテキスト永続化エージェントは、セッション状態をSQLite（単一ユーザー）またはRedis（複数サーバー）に保存し、履歴のためにすべてのメッセージを保存し、長期記憶のためにユーザー情報を抽出して保存します。また、ユーザー設定を加えたシステムプロンプトを構築し、コンテキストウィンドウの制限に対応するために会話の要約を使用し、プライバシー規制への準拠に向けたデータ保持ポリシーを適用します。
よくある質問
「セッションをまたぐコンテキストの永続化」レッスンは無料ですか？
はい。「セッションをまたぐコンテキストの永続化」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「セッションをまたぐコンテキストの永続化」で何を学びますか？
ユーザー設定、会話履歴、タスク状態を保存・読み込みします。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「セッションをまたぐコンテキストの永続化」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- 常時稼働エージェントの設計パターン
- プロアクティブな通知・アラートシステム
- セッションをまたぐコンテキストの永続化
- 毎日のブリーフィングエージェントを構築する