Spaces:
Sleeping
Sleeping
File size: 1,800 Bytes
f37a598 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
"""Simple conversation history storage using SQLite."""
import sqlite3
import os
from datetime import datetime
from typing import List, Dict, Optional
DB_PATH = os.getenv("HISTORY_DB_PATH", "conversation_history.db")
def init_db():
"""Initialize the conversation history database."""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS conversations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
prompt TEXT NOT NULL,
response TEXT NOT NULL,
source TEXT DEFAULT 'demo',
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
conn.close()
def save_conversation(session_id: str, prompt: str, response: str, source: str = "demo"):
"""Save a conversation turn to the database."""
init_db()
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute(
"INSERT INTO conversations (session_id, prompt, response, source) VALUES (?, ?, ?, ?)",
(session_id, prompt, response, source),
)
conn.commit()
conn.close()
def get_conversation_history(session_id: str, limit: int = 10) -> List[Dict]:
"""Retrieve conversation history for a session."""
init_db()
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute(
"SELECT prompt, response, source, timestamp FROM conversations WHERE session_id = ? ORDER BY timestamp DESC LIMIT ?",
(session_id, limit),
)
rows = cursor.fetchall()
conn.close()
return [
{
"prompt": row[0],
"response": row[1],
"source": row[2],
"timestamp": row[3],
}
for row in reversed(rows)
]
|