Zachowywanie kontekstu między sesjami
Zapisywanie i wczytywanie preferencji użytkownika, historii rozmów i stanu zadań.
Zachowywanie kontekstu między sesjami to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 3 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.
Dlaczego warto utrwalać kontekst?
Bez utrwalania każda sesja agenta zaczyna się od początku. Agent zachowujący kontekst pamięta imię użytkownika, wcześniejsze rozmowy, określone preferencje oraz informacje udostępnione przez użytkownika. Dzięki temu interakcje wydają się ciągłe i spersonalizowane.
SQLite jako stan sesji
SQLite doskonale nadaje się do osobistych agentów jednego użytkownika: nie wymaga narzutu serwera, działa na plikach i jest niezawodny. Historię rozmów oraz dane użytkownika należy przechowywać w lokalnej bazie danych.
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')Zapisywanie i wczytywanie wiadomości
Każdą wiadomość użytkownika i każdą odpowiedź agenta należy zapisywać wraz z identyfikatorem sesji. Przy wznawianiu sesji należy wczytać ostatnich N wiadomości, aby odtworzyć kontekst rozmowy.
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 jako stan sesji
W przypadku agentów internetowych lub wdrożeń na wielu serwerach Redis sprawdza się lepiej niż SQLite. Dane sesji należy przechowywać jako JSON z ustawionym TTL, aby nieaktualne sesje były automatycznie usuwane.
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)Wczytywanie preferencji użytkownika podczas uruchamiania
Po rozpoczęciu nowej sesji należy wczytać zapisane preferencje użytkownika: język, strefę czasową, ustawienia powiadomień oraz konfiguracje właściwe dla agenta. Następnie należy uwzględnić je w promptcie systemowym.
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)Przechowywanie podsumowań rozmów
Przechowywanie pełnej historii rozmów jest kosztowne pod względem liczby tokenów. Należy użyć podsumowywania przez LLM, aby utworzyć skompresowane podsumowanie starszych rozmów, a następnie wczytywać tylko to podsumowanie dla odleglejszej historii.
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')Wyodrębnianie i przechowywanie faktów o użytkowniku
Gdy użytkownik udostępnia informacje o sobie, należy je wyodrębnić i zapisać. W ten sposób powstaje długoterminowa pamięć, która zachowuje się we wszystkich przyszłych sesjach.
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)Wczytywanie kontekstu odpowiedniego dla bieżącej tury
Na początku każdej tury agenta należy zebrać kontekst obejmujący: ostatnie wiadomości, preferencje użytkownika oraz istotne fakty. Dzięki temu LLM otrzymuje wszystkie potrzebne informacje bez przekraczania rozmiaru okna kontekstu.
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]}...')Pobieranie pamięci między sesjami
W przypadku agentów działających przez bardzo długi czas należy używać wyszukiwania wektorowego do znajdowania istotnych wcześniejszych rozmów, a nie tylko ostatnich. Pozwala to agentowi przywoływać istotny kontekst sprzed wielu miesięcy.
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')Prywatność i zarządzanie danymi
Agenci działający stale przechowują poufne dane osobowe. Należy wdrożyć limity przechowywania danych, umożliwić użytkownikom usuwanie ich danych i nigdy nie rejestrować surowej treści rozmów w logach aplikacji.
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')Kontrola ciągłości sesji
Gdy sesja zostaje wznowiona po długiej przerwie, należy krótko poinformować agenta o tej przerwie: jak długo użytkownik był nieobecny i co się zmieniło. Zapobiega to dezorientującym przeskokom kontekstu.
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')Sprawdzenie wiedzy: utrwalanie kontekstu
Sprawdź swoją wiedzę na temat utrwalania kontekstu między sesjami agenta.
Podsumowanie utrwalania kontekstu
Agenci zachowujący kontekst przechowują stan sesji w SQLite (dla jednego użytkownika) lub Redisie (na wielu serwerach), zapisują każdą wiadomość w historii, wyodrębniają i przechowują fakty o użytkowniku na potrzeby długoterminowej pamięci, tworzą prompty systemowe wzbogacone o preferencje użytkownika, używają podsumowywania rozmów do zarządzania limitami okna kontekstu oraz stosują zasady przechowywania danych w celu zapewnienia zgodności z wymogami prywatności.
Ucz się AI Agents dzięki korepetycjom AI — za darmo
Pisz i uruchamiaj kod w przeglądarce, otrzymuj natychmiastową pomoc od korepetytora AI dostępnego 24/7 i kontynuuj naukę w sieci lub w aplikacji.
- Kursy
- 60
- Lekcje
- 239
Często zadawane pytania
Czy lekcja „Zachowywanie kontekstu między sesjami” jest bezpłatna?
Tak — pełny tekst „Zachowywanie kontekstu między sesjami” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.
Co nauczysz się w „Zachowywanie kontekstu między sesjami”?
Zapisywanie i wczytywanie preferencji użytkownika, historii rozmów i stanu zadań. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć AI Agents?
Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 3 z 4.
Ile czasu zajmuje lekcja „Zachowywanie kontekstu między sesjami”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?
Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Wzorce projektowe agentów działających stale
- Proaktywne systemy powiadomień i alertów
- Zachowywanie kontekstu między sesjami
- Budowanie agenta codziennego briefingu