0Pricing
AI Agents · Leçon

Persistance du contexte entre les sessions

Enregistrer et charger les préférences utilisateur, l’historique des conversations et l’état des tâches.

Persistance du contexte entre les sessions est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 3 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.

Pourquoi conserver le contexte ?

Sans persistance, chaque session de l’agent repart de zéro. Un agent à contexte persistant se souvient du nom de l’utilisateur, des conversations passées, de ses préférences déclarées et des faits qu’il a communiqués. Les interactions paraissent ainsi continues et personnalisées.

SQLite pour l’état des sessions

SQLite est idéal pour les agents personnels destinés à un seul utilisateur : il ne nécessite aucun serveur, repose sur des fichiers et est fiable. Stockez l’historique des conversations et les données utilisateur dans une base de données 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')

Enregistrement et chargement des messages

Chaque message de l’utilisateur et chaque réponse de l’agent doivent être enregistrés avec leur ID de session. Lors de la reprise d’une session, chargez les N messages les plus récents afin de restaurer le contexte de la conversation.

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 pour l’état des sessions

Pour les agents Web ou les déploiements sur plusieurs serveurs, Redis est préférable à SQLite. Stockez les données de session au format JSON avec un TTL afin que les sessions obsolètes soient automatiquement supprimées.

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)

Chargement des préférences de l’utilisateur au démarrage

Lorsqu’une nouvelle session commence, chargez les préférences enregistrées de l’utilisateur : langue, fuseau horaire, paramètres de notification et configurations propres à l’agent. Injectez-les dans l’invite système.

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)

Stockage du résumé des conversations

Conserver l’historique complet des conversations consomme beaucoup de jetons. Utilisez la synthèse par un LLM pour créer un résumé condensé des conversations anciennes, puis ne chargez que ce résumé pour l’historique lointain.

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')

Extraction et stockage des faits sur l’utilisateur

Lorsque l’utilisateur communique des informations personnelles, extrayez-les et enregistrez-les. Vous construisez ainsi une mémoire à long terme qui persiste dans toutes les sessions futures.

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)

Chargement du contexte pertinent pour le tour actuel

Au début de chaque tour de l’agent, assemblez le contexte à partir des messages récents, des préférences de l’utilisateur et des faits pertinents. Le LLM dispose ainsi de tout ce dont il a besoin sans dépasser la fenêtre de contexte.

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]}...')

Récupération de la mémoire entre les sessions

Pour les agents qui fonctionnent pendant de très longues périodes, utilisez une recherche vectorielle afin de retrouver les conversations passées pertinentes, et pas seulement les conversations récentes. L’agent peut ainsi se rappeler un contexte pertinent datant de plusieurs mois.

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')

Confidentialité et gestion des données

Les agents toujours actifs stockent des données personnelles sensibles. Mettez en place des limites de conservation, permettez aux utilisateurs de supprimer leurs données et n’enregistrez jamais le contenu brut des conversations dans les journaux de l’application.

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')

Vérification de la continuité de session

Lorsqu’une session reprend après une longue interruption, informez brièvement l’agent de cette interruption : indiquez depuis combien de temps l’utilisateur était absent et ce qui a changé. Cela évite les changements de contexte déroutants.

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')

Vérification des connaissances : persistance du contexte

Vérifiez votre compréhension de la persistance du contexte entre les sessions de l’agent.

Résumé de la persistance du contexte

Les agents à contexte persistant stockent l’état des sessions dans SQLite (pour un seul utilisateur) ou Redis (pour plusieurs serveurs), enregistrent chaque message dans l’historique, extraient et stockent les faits sur l’utilisateur pour conserver une mémoire à long terme, construisent des invites système enrichies par les préférences de l’utilisateur, utilisent la synthèse des conversations pour gérer les limites de la fenêtre de contexte et appliquent des politiques de conservation des données pour respecter les exigences de confidentialité.

Questions Fréquemment Posées

La leçon « Persistance du contexte entre les sessions » est-elle gratuite ?

Oui — le texte complet de « Persistance du contexte entre les sessions » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Persistance du contexte entre les sessions » ?

Enregistrer et charger les préférences utilisateur, l’historique des conversations et l’état des tâches. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Agents ?

Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 3 sur 4.

Combien de temps prend la leçon « Persistance du contexte entre les sessions » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?

Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Schémas de conception pour des agents toujours actifs
  2. Systèmes proactifs de notification et d’alerte
  3. Persistance du contexte entre les sessions
  4. Construire un agent de briefing quotidien
← Retour à AI Agents