智能体长期记忆设计：该记什么、何时遗忘，以及如何落地

import argparse
import re
import sqlite3
import time
import uuid

DB_PATH = "agent_memory.db"
SENSITIVE_PATTERNS = [
    re.compile(r"\b(?:sk|ghp)_[A-Za-z0-9_-]{16,}\b"),
    re.compile(r"\bpassword\s*[:=]\s*\S+", re.I),
    re.compile(r"\b\d{17}[0-9Xx]\b"),
]
TTL_BY_SCOPE = {
    "session": 0,
    "task": 24 * 3600,
    "workspace": 30 * 24 * 3600,
    "user": None,
}


def connect():
    db = sqlite3.connect(DB_PATH)
    db.execute("""
        CREATE TABLE IF NOT EXISTS memories (
            id TEXT PRIMARY KEY,
            scope TEXT NOT NULL,
            content TEXT NOT NULL,
            source TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            expires_at INTEGER
        )
    """)
    return db


def contains_sensitive_data(text):
    return any(pattern.search(text) for pattern in SENSITIVE_PATTERNS)


def remember(scope, content, source="user-confirmed"):
    if scope not in TTL_BY_SCOPE:
        raise ValueError(f"unsupported scope: {scope}")
    if scope == "session":
        return "Skipped: session data is not persisted."
    if contains_sensitive_data(content):
        return "Rejected: possible sensitive data."

    now = int(time.time())
    ttl = TTL_BY_SCOPE[scope]
    expires_at = now + ttl if ttl is not None else None
    memory_id = str(uuid.uuid4())

    with connect() as db:
        db.execute(
            "INSERT INTO memories VALUES (?, ?, ?, ?, ?, ?)",
            (memory_id, scope, content, source, now, expires_at),
        )
    return memory_id


def list_active(scope=None):
    now = int(time.time())
    with connect() as db:
        db.execute("DELETE FROM memories WHERE expires_at <= ?", (now,))
        sql = "SELECT id, scope, content, source, expires_at FROM memories"
        params = ()
        if scope:
            sql += " WHERE scope = ?"
            params = (scope,)
        return db.execute(sql, params).fetchall()


def forget(memory_id):
    with connect() as db:
        result = db.execute("DELETE FROM memories WHERE id = ?", (memory_id,))
        return result.rowcount > 0


def main():
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="command", required=True)

    add = sub.add_parser("remember")
    add.add_argument("scope", choices=TTL_BY_SCOPE)
    add.add_argument("content")

    show = sub.add_parser("list")
    show.add_argument("--scope", choices=TTL_BY_SCOPE)

    delete = sub.add_parser("forget")
    delete.add_argument("id")

    args = parser.parse_args()
    if args.command == "remember":
        print(remember(args.scope, args.content))
    elif args.command == "list":
        for row in list_active(args.scope):
            print(row)
    else:
        print("Deleted" if forget(args.id) else "Not found")


if __name__ == "__main__":
    main()

python memory_gate.py remember user "回答代码问题时优先使用 Python"
python memory_gate.py remember workspace "测试命令是 pytest -q"
python memory_gate.py remember session "当前正在排查登录超时"
python memory_gate.py remember user "password=correct-horse-battery-staple"
python memory_gate.py list