聊天日志记录器Skill chat-logger

---

名称: 聊天日志记录器 描述: “将所有聊天消息记录到SQLite数据库中，用于可搜索的历史记录和审计。使用场景：(1) 构建聊天历史，(2) 审计对话，(3) 搜索过去消息，或 (4) 用户要求记录聊天。”

聊天日志记录器

将所有传入和传出的聊天消息记录到SQLite数据库中，用于可搜索的历史记录、分析和审计。适用于任何聊天系统或代理框架。

使用时机

• 构建可搜索的聊天历史系统
• 审计和审查过去对话
• 创建聊天交互的分析
• 调试聊天流程和响应
• 用户要求跟踪或搜索对话历史

所需工具/API

• Python标准库 (sqlite3, datetime, json)
• 任何支持SQLite的编程语言

无需外部API或服务。

数据库模式

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

字段:

• id - 自增主键
• timestamp - ISO 8601 时间戳的消息时间
• session_id - 可选会话/对话标识符
• sender - 消息发送者 (‘user’, ‘assistant’, 或自定义ID)
• content - 消息文本内容
• metadata - JSON字段，用于额外数据 (channel, tools, context)
• created_at - 数据库插入时间戳

基本实现

Python

初始化数据库:

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

记录消息:

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"})

查询消息:

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快速查询

# 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.json

集成示例

通用聊天应用

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 response

装饰器模式

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}"

代理提示

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.

最佳实践

1. 截断长消息 以避免数据库膨胀（例如，10,000字符）
2. 使用索引 在timestamp、session_id和sender上，用于快速查询
3. 将元数据存储为JSON 以增加灵活性
4. 使用ISO 8601时间戳 以保持一致性
5. 会话ID 帮助组织对话
6. 隐私考虑：注意存储敏感数据
7. 定期备份：SQLite文件易于备份/恢复

故障排除

数据库锁定错误：

• 正确关闭所有连接，使用 conn.close()
• 对于高流量，使用连接池

大数据库文件：

• 运行 VACUUM 来压缩：sqlite3 messages.db "VACUUM"
• 定期归档旧消息

查询性能：

• 确保索引已创建 (timestamp, session_id, sender)
• 在查询中使用LIMIT
• 对于大结果集，考虑分页

另请参阅

• …/file-tracker/SKILL.md — 跟踪文件修改
• …/web-search-api/SKILL.md — 搜索外部内容