SQLite в Python: персистентная память для агентов

SQLite — встроенная в Python реляционная база данных. Хранит данные в одном файле, не требует сервера. Идеальна для персистентной памяти агентов.

Подключение

import sqlite3

conn = sqlite3.connect("memory.db")  # создаст файл если нет
cursor = conn.cursor()
conn.close()

Или через контекстный менеджер:

with sqlite3.connect("memory.db") as conn:
    cursor = conn.cursor()
    # работаем...
# conn.close() вызывается автоматически

Создание таблицы

def init_db(db_path: str = "chat_history.db"):
    with sqlite3.connect(db_path) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS messages (
                id        INTEGER PRIMARY KEY AUTOINCREMENT,
                session   TEXT NOT NULL,
                role      TEXT NOT NULL,
                content   TEXT NOT NULL,
                created   TEXT DEFAULT (datetime('now'))
            )
        """)
        conn.commit()

Запись сообщений

def save_message(session: str, role: str, content: str, db_path: str = "chat_history.db"):
    with sqlite3.connect(db_path) as conn:
        conn.execute(
            "INSERT INTO messages (session, role, content) VALUES (?, ?, ?)",
            (session, role, content)
        )
        conn.commit()

Всегда используй ? плейсхолдеры — защита от SQL-инъекций.

Загрузка истории

def load_history(session: str, db_path: str = "chat_history.db") -> list[dict]:
    with sqlite3.connect(db_path) as conn:
        conn.row_factory = sqlite3.Row  # доступ по имени колонки
        rows = conn.execute(
            "SELECT role, content FROM messages WHERE session = ? ORDER BY id",
            (session,)
        ).fetchall()
    return [{"role": row["role"], "content": row["content"]} for row in rows]

Список сессий

def list_sessions(db_path: str = "chat_history.db") -> list[str]:
    with sqlite3.connect(db_path) as conn:
        rows = conn.execute(
            "SELECT DISTINCT session, MIN(created) as started FROM messages GROUP BY session ORDER BY started DESC"
        ).fetchall()
    return [row[0] for row in rows]

Удаление сессии

def delete_session(session: str, db_path: str = "chat_history.db"):
    with sqlite3.connect(db_path) as conn:
        conn.execute("DELETE FROM messages WHERE session = ?", (session,))
        conn.commit()

Полный пример: память агента

import sqlite3
import anthropic

client = anthropic.Anthropic()
DB = "agent_memory.db"

def init_db():
    with sqlite3.connect(DB) as conn:
        conn.execute("CREATE TABLE IF NOT EXISTS messages (session TEXT, role TEXT, content TEXT)")

def chat(session: str, user_input: str) -> str:
    save_message(session, "user", user_input)
    history = load_history(session)

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=history
    )
    answer = response.content[0].text
    save_message(session, "assistant", answer)
    return answer