استمرارية السياق عبر الجلسات

لماذا نحتفظ بالسياق؟

SQLite لحالة الجلسة

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

حفظ الرسائل وتحميلها

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 لحالة الجلسة

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)

تخزين ملخص المحادثة

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)

تحميل السياق ذي الصلة بالدور الحالي

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

اختبار المعرفة: استمرارية السياق

ملخص استمرارية السياق