AI-agenten · Les

Context persistent maken tussen sessies

Gebruikersvoorkeuren, gespreksgeschiedenis en taakstatus opslaan en laden.

Les 3 van 413 stappen

Context persistent maken tussen sessies is een gratis AI-agenten-les op CoddyKit. Dit is les 3 van 4. Je kunt de volledige les hieronder gratis lezen en daarna in de browser praktisch oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject AI-agenten. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus AI-agenten bevat in totaal 4 lessen.

Waarom context bewaren?

Zonder bewaren van context begint elke agentsessie vanaf nul. Een agent met blijvende context onthoudt de naam van de gebruiker, eerdere gesprekken, uitgesproken voorkeuren en feiten die de gebruiker heeft gedeeld. Daardoor voelen interacties doorlopend en persoonlijk aan.

SQLite voor sessiestatus

SQLite is ideaal voor persoonlijke agents voor één gebruiker: geen serveroverhead, gebaseerd op bestanden en betrouwbaar. Sla de gespreksgeschiedenis en gebruikersgegevens op in een lokale database.

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

Berichten opslaan en laden

Elk gebruikersbericht en elk antwoord van de agent moet samen met het sessie-ID worden opgeslagen. Laad bij het hervatten van een sessie de meest recente N berichten om de gesprekscontext te herstellen.

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 voor sessiestatus

Voor webagents of implementaties met meerdere servers is Redis geschikter dan SQLite. Sla sessiegegevens op als JSON met een TTL, zodat verouderde sessies automatisch worden opgeruimd.

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)

Gebruikersvoorkeuren laden bij het opstarten

Laad bij het starten van een nieuwe sessie de opgeslagen voorkeuren van de gebruiker: taal, tijdzone, meldingsinstellingen en configuraties die specifiek zijn voor de agent. Voeg deze toe aan de systeemprompt.

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)

Samenvattingen van gesprekken opslaan

Het opslaan van de volledige gespreksgeschiedenis kost veel tokens. Gebruik samenvatting door een LLM om een compacte samenvatting van oudere gesprekken te maken en laad voor verre geschiedenis alleen die samenvatting.

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

Feiten over gebruikers verzamelen en opslaan

Wanneer de gebruiker feiten over zichzelf deelt, verzamel en sla je die op. Zo bouw je een langetermijngeheugen op dat in alle toekomstige sessies blijft bestaan.

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)

Relevante context voor de huidige beurt laden

Stel aan het begin van elke beurt van de agent context samen uit recente berichten, gebruikersvoorkeuren en relevante feiten. Zo krijgt de LLM alles wat hij nodig heeft zonder de limiet van het contextvenster te overschrijden.

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

Geheugen tussen sessies ophalen

Gebruik voor agents die zeer lang actief blijven een vectorzoekopdracht om relevante eerdere gesprekken te vinden — niet alleen recente gesprekken. Zo kan de agent relevante context van maanden geleden terughalen.

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- en gegevensbeheer

Altijd actieve agents slaan gevoelige persoonlijke gegevens op. Implementeer bewaartermijnen, geef gebruikers de mogelijkheid hun gegevens te verwijderen en schrijf nooit onbewerkte gespreksinhoud naar toepassingslogboeken.

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

Controle op sessiecontinuïteit

Informeer de agent wanneer een sessie na een lange onderbreking wordt hervat over die onderbreking: hoe lang de gebruiker weg was en wat er is veranderd. Zo voorkom je desoriënterende sprongen in de context.

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

Kenniscontrole: context bewaren

Toets je begrip van het bewaren van context tussen agentsessies.

Samenvatting van contextbehoud

Agents met blijvende context slaan sessiestatus op in SQLite (voor één gebruiker) of Redis (voor meerdere servers), bewaren elk bericht voor de geschiedenis, verzamelen en bewaren feiten over gebruikers voor langetermijngeheugen, bouwen systeemprompts op die zijn verrijkt met gebruikersvoorkeuren, gebruiken samenvattingen van gesprekken om limieten van het contextvenster te beheren en passen bewaarbeleid voor gegevens toe om aan privacyregels te voldoen.

Gratis beginnen

Leer AI-agenten met een AI-tutor — gratis

Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.

Cursussen
60
Lessen
239

Veelgestelde vragen

Is de les “Context persistent maken tussen sessies” gratis?

Ja — de volledige tekst van “Context persistent maken tussen sessies” kun je hier gratis op het web lezen. Als je interactief wilt oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is, en de rest van de cursus AI-agenten wilt ontgrendelen, kun je upgraden naar CoddyKit PRO. De cursus AI-agenten bevat in totaal 4 lessen.

Wat leer ik in “Context persistent maken tussen sessies”?

Gebruikersvoorkeuren, gespreksgeschiedenis en taakstatus opslaan en laden. Je oefent met AI-agenten door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.

Heb ik ervaring nodig om met AI-agenten te beginnen?

Ervaring vooraf is niet nodig. AI-agenten op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 3 van 4.

Hoe lang duurt de les “Context persistent maken tussen sessies”?

De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.

Kan ik code schrijven en uitvoeren in deze les over AI-agenten?

Ja. Elke les over AI-agenten bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.

Alle lessen in deze cursus

  1. Ontwerppatronen voor altijd actieve agents
  2. Proactieve meldings- en waarschuwingssystemen
  3. Context persistent maken tussen sessies
  4. Een agent voor dagelijkse briefings bouwen
← Terug naar AI-agenten