Chat Logger
Log all incoming and outgoing chat messages to a SQLite database for searchable history, analytics, and auditing. Works with any chat system or agent framework.
When to use
- Building a searchable chat history system
- Auditing and reviewing past conversations
- Creating analytics on chat interactions
- Debugging chat flows and responses
- User asks to track or search conversation history
Required tools / APIs
- Python standard library (sqlite3, datetime, json)
- Any programming language with SQLite support
No external APIs or services required.
Database Schema
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
session_id TEXT,
sender TEXT NOT NULL, -- 'user', 'assistant', or identifier
content TEXT,
metadata TEXT, -- JSON: channel, tools_used, etc.
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_timestamp ON messages(timestamp);
CREATE INDEX idx_session ON messages(session_id);
CREATE INDEX idx_sender ON messages(sender);
-- Automatic purge: delete records older than 1 year
DELETE FROM messages WHERE created_at < datetime('now', '-1 year');Fields:
id- Auto-incrementing primary keytimestamp- ISO 8601 timestamp of the messagesession_id- Optional session/conversation identifiersender- Message sender ('user', 'assistant', or custom ID)content- Message text contentmetadata- JSON field for additional data (channel, tools, context)created_at- Database insertion timestamp
Basic Implementation
Python
Initialize database:
import sqlite3
from datetime import datetime
from pathlib import Path
import json
# Configure database path
DB_PATH = Path.home() / ".chat_logs" / "messages.db"
def init_db():
"""Initialize database and create tables."""
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(DB_PATH))
conn.execute("""
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
session_id TEXT,
sender TEXT NOT NULL,
content TEXT,
metadata TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON messages(timestamp)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_session ON messages(session_id)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_sender ON messages(sender)")
conn.commit()
conn.close()
def purge_old_messages():
"""Delete messages older than 1 year to keep the database size sane."""
conn = sqlite3.connect(str(DB_PATH))
conn.execute("DELETE FROM messages WHERE created_at < datetime('now', '-1 year')")
conn.commit()
conn.close()
# Initialize on import and purge old records
init_db()
purge_old_messages()Log messages:
def log_message(sender: str, content: str, session_id: str = None, metadata: dict = None):
"""Log a chat message to the database."""
conn = sqlite3.connect(str(DB_PATH))
try:
conn.execute(
"""INSERT INTO messages (timestamp, session_id, sender, content, metadata)
VALUES (?, ?, ?, ?, ?)""",
(
datetime.utcnow().isoformat(),
session_id,
sender,
content[:10000] if content else None, # Truncate long messages
json.dumps(metadata) if metadata else None
)
)
conn.commit()
finally:
conn.close()
# Usage examples
log_message("user", "Hello, how are you?", session_id="session_123")
log_message("assistant", "I'm doing well, thank you!", session_id="session_123")
log_message("user", "Help me deploy a website", session_id="session_456",
metadata={"channel": "web", "ip": "192.168.1.1"})Query messages:
def get_recent_messages(limit: int = 50):
"""Get recent messages."""
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
cursor = conn.execute(
"SELECT * FROM messages ORDER BY timestamp DESC LIMIT ?",
(limit,)
)
results = cursor.fetchall()
conn.close()
return results
def get_session_history(session_id: str):
"""Get all messages from a specific session."""
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
cursor = conn.execute(
"SELECT * FROM messages WHERE session_id = ? ORDER BY timestamp ASC",
(session_id,)
)
results = cursor.fetchall()
conn.close()
return results
def search_messages(query: str, limit: int = 20):
"""Search message content."""
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
cursor = conn.execute(
"SELECT * FROM messages WHERE content LIKE ? ORDER BY timestamp DESC LIMIT ?",
(f"%{query}%", limit)
)
results = cursor.fetchall()
conn.close()
return results
# Usage
messages = get_recent_messages(10)
for msg in messages:
print(f"[{msg['timestamp']}] {msg['sender']}: {msg['content'][:100]}")
# Search
results = search_messages("deploy website")
print(f"Found {len(results)} messages about deploying websites")Node.js
import sqlite3 from "sqlite3";
import { promisify } from "util";
import path from "path";
import os from "os";
const DB_PATH = path.join(os.homedir(), ".chat_logs", "messages.db");
// Initialize database
const db = new sqlite3.Database(DB_PATH);
const run = promisify(db.run.bind(db));
const all = promisify(db.all.bind(db));
await run(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
session_id TEXT,
sender TEXT NOT NULL,
content TEXT,
metadata TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
`);
// Log message
async function logMessage(sender, content, sessionId = null, metadata = null) {
await run(
`INSERT INTO messages (timestamp, session_id, sender, content, metadata)
VALUES (?, ?, ?, ?, ?)`,
[
new Date().toISOString(),
sessionId,
sender,
content,
metadata ? JSON.stringify(metadata) : null,
]
);
}
// Query messages
async function getRecentMessages(limit = 50) {
return await all(
`SELECT * FROM messages ORDER BY timestamp DESC LIMIT ?`,
[limit]
);
}
// Usage
await logMessage("user", "Hello!", "session_123");
await logMessage("assistant", "Hi there!", "session_123");
const messages = await getRecentMessages(10);
console.log(messages);Bash Quick Queries
# View recent messages
sqlite3 ~/.chat_logs/messages.db "SELECT timestamp, sender, substr(content, 1, 80) FROM messages ORDER BY timestamp DESC LIMIT 20"
# Search for specific content
sqlite3 ~/.chat_logs/messages.db "SELECT * FROM messages WHERE content LIKE '%docker%' ORDER BY timestamp DESC"
# Count messages by sender
sqlite3 ~/.chat_logs/messages.db "SELECT sender, COUNT(*) as count FROM messages GROUP BY sender"
# Export session to JSON
sqlite3 -json ~/.chat_logs/messages.db "SELECT * FROM messages WHERE session_id='session_123' ORDER BY timestamp ASC" > conversation.jsonIntegration Examples
Generic Chat Application
class ChatLogger:
"""Simple chat logger that can wrap any chat system."""
def __init__(self, db_path: str = None):
self.db_path = db_path or str(Path.home() / ".chat_logs" / "messages.db")
self._init_db()
def _init_db(self):
# Same as init_db() above
pass
def log_user_message(self, content: str, session_id: str = None, **metadata):
return log_message("user", content, session_id, metadata)
def log_assistant_message(self, content: str, session_id: str = None, **metadata):
return log_message("assistant", content, session_id, metadata)
def get_conversation(self, session_id: str):
return get_session_history(session_id)
# Usage in any chat system
logger = ChatLogger()
# In your chat handler
def handle_message(user_input, session_id):
logger.log_user_message(user_input, session_id=session_id)
# Process message...
response = generate_response(user_input)
logger.log_assistant_message(response, session_id=session_id)
return responseDecorator Pattern
def with_logging(session_id: str = None):
"""Decorator to automatically log chat interactions."""
def decorator(func):
def wrapper(user_message, *args, **kwargs):
# Log user message
log_message("user", user_message, session_id=session_id)
# Call original function
response = func(user_message, *args, **kwargs)
# Log assistant response
log_message("assistant", response, session_id=session_id)
return response
return wrapper
return decorator
# Usage
@with_logging(session_id="session_123")
def chat_handler(message):
return f"You said: {message}"Agent Prompt
You have chat logging capability. All conversations are logged to a SQLite database.
When user asks to:
- Search past conversations
- Find specific messages
- Review conversation history
- Export chat logs
Use the SQLite database at ~/.chat_logs/messages.db with this schema:
- messages table (id, timestamp, session_id, sender, content, metadata)
Query examples:
1. Recent history: SELECT * FROM messages ORDER BY timestamp DESC LIMIT 50
2. Search content: SELECT * FROM messages WHERE content LIKE '%keyword%'
3. Session history: SELECT * FROM messages WHERE session_id = ? ORDER BY timestamp ASC
Always use SQL queries to retrieve information and present results clearly to the user.Best Practices
- Truncate long messages to avoid database bloat (e.g., 10,000 chars)
- Use indexes on timestamp, session_id, and sender for fast queries
- Store metadata as JSON for flexibility
- Use ISO 8601 timestamps for consistency
- Session IDs help organize conversations
- Privacy considerations: be mindful of storing sensitive data
- Regular backups: SQLite files are easy to backup/restore
Troubleshooting
Database locked error:
- Close all connections properly with
conn.close() - Use connection pooling for high traffic
Large database file:
- Run
VACUUMto compact:sqlite3 messages.db "VACUUM" - Archive old messages periodically
Query performance:
- Ensure indexes are created (timestamp, session_id, sender)
- Use LIMIT on queries
- Consider pagination for large result sets
See also
- ../file-tracker/SKILL.md — Track file modifications
- ../web-search-api/SKILL.md — Search external content