Persistenza del contesto tra le sessioni
Salvataggio e caricamento delle preferenze dell’utente, della cronologia delle conversazioni e dello stato delle attività.
Persistenza del contesto tra le sessioni è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Perché rendere persistente il contesto
Senza la persistenza, ogni sessione dell'agente ricomincia da zero. Un agente con contesto persistente ricorda il nome dell'utente, le conversazioni precedenti, le preferenze dichiarate e i fatti condivisi dall'utente. Questo rende le interazioni continue e personali.
SQLite per lo stato della sessione
SQLite è perfetto per gli agenti personali con un singolo utente: non richiede un server, è basato su file ed è affidabile. Salvi la cronologia delle conversazioni e i dati dell'utente in un database locale.
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')Salvataggio e caricamento dei messaggi
Ogni messaggio dell'utente e ogni risposta dell'agente devono essere salvati con il relativo ID di sessione. Quando si riprende una sessione, carichi gli ultimi N messaggi per ripristinare il contesto della conversazione.
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 per lo stato della sessione
Per gli agenti web o le distribuzioni su più server, Redis è preferibile a SQLite. Salvi i dati della sessione come JSON con un TTL, in modo che le sessioni obsolete vengano rimosse automaticamente.
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)Caricamento delle preferenze dell'utente all'avvio
Quando inizia una nuova sessione, carichi le preferenze salvate dell'utente: lingua, fuso orario, impostazioni delle notifiche ed eventuali configurazioni specifiche dell'agente. Inserisca questi dati nel prompt di sistema.
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)Archiviazione dei riepiloghi delle conversazioni
Archiviare l'intera cronologia delle conversazioni è costoso in termini di token. Utilizzi il riepilogo generato da un LLM per creare una sintesi compressa delle conversazioni meno recenti, quindi carichi solo il riepilogo per la cronologia più lontana.
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')Estrazione e archiviazione dei fatti sull'utente
Quando l'utente condivide informazioni su di sé, le estragga e le salvi. In questo modo costruisce una memoria a lungo termine che persiste in tutte le sessioni future.
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)Caricamento del contesto rilevante per il turno corrente
All'inizio di ogni turno dell'agente, componga il contesto a partire da: messaggi recenti, preferenze dell'utente e fatti rilevanti. In questo modo fornisce all'LLM tutto ciò di cui ha bisogno senza superare la finestra di contesto.
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]}...')Recupero della memoria tra le sessioni
Per gli agenti in esecuzione per periodi molto lunghi, utilizzi la ricerca vettoriale per trovare conversazioni passate rilevanti, non solo quelle recenti. In questo modo l'agente può ricordare il contesto pertinente di mesi prima.
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')Privacy e gestione dei dati
Gli agenti sempre attivi memorizzano dati personali sensibili. Implementi limiti di conservazione dei dati, consenta agli utenti di eliminarli e non registri mai il contenuto grezzo delle conversazioni nei log dell'applicazione.
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')Controllo della continuità della sessione
Quando una sessione riprende dopo una lunga pausa, fornisca all'agente un riepilogo della pausa: per quanto tempo l'utente è rimasto assente e cosa è cambiato. In questo modo evita salti disorientanti nel contesto.
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')Verifica delle conoscenze: persistenza del contesto
Verifichi la sua comprensione della persistenza del contesto tra le sessioni dell'agente.
Riepilogo della persistenza del contesto
Gli agenti con contesto persistente archiviano lo stato della sessione in SQLite (singolo utente) o Redis (più server), salvano ogni messaggio per la cronologia, estraggono e archiviano i fatti sull'utente per la memoria a lungo termine, creano prompt di sistema arricchiti con le preferenze dell'utente, utilizzano il riepilogo delle conversazioni per gestire i limiti della finestra di contesto e applicano criteri di conservazione dei dati per la conformità alla privacy.
Domande Frequenti
La lezione «Persistenza del contesto tra le sessioni» è gratuita?
Sì — il testo completo di «Persistenza del contesto tra le sessioni» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Persistenza del contesto tra le sessioni»?
Salvataggio e caricamento delle preferenze dell’utente, della cronologia delle conversazioni e dello stato delle attività. Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.
Quanto tempo richiede la lezione «Persistenza del contesto tra le sessioni»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Pattern di progettazione per agenti sempre attivi
- Sistemi proattivi di notifica e avviso
- Persistenza del contesto tra le sessioni
- Creazione di un agente per il briefing quotidiano