Memory & History

Dua tempat agent "inget", dengan tujuan berbeda.

	Memory	History
Fungsi	Preferensi user, fakta stabil	Chat thread per user
Storage	memory.json (1 file)	history/<user_id>.json (1 file per user)
Lifetime	Persistent across sessions	Sliding window (last N messages)
Update	Explicit (/remember) atau auto-detect koreksi	Otomatis tiap chat
Load	Append ke system prompt	Append ke messages array

Memory implementation

Struktur

{
  "notes": [
    "ringkas output >20 baris kecuali user minta",
    "command destructive selalu risk=high",
    "user prefer gue/lo bukan saya/Anda"
  ],
  "user_name": "Gutluc",
  "preferred_language": "id",
  "default_branch": "main",
  "wallet_main_address": "0xabc..."
}

notes adalah list of strings. Fields lain bebas, sesuai kebutuhan.

Load & save

import json
from pathlib import Path

MEMORY_FILE = Path.home() / "agent" / "data" / "memory.json"

def load_memory() -> dict:
    if MEMORY_FILE.exists():
        try:
            return json.loads(MEMORY_FILE.read_text())
        except (json.JSONDecodeError, OSError):
            pass
    return {"notes": [], "user_name": ""}

def save_memory(data: dict):
    MEMORY_FILE.parent.mkdir(parents=True, exist_ok=True)
    MEMORY_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2))

def add_memory_note(note: str):
    m = load_memory()
    if note not in m.get("notes", []):
        m.setdefault("notes", []).append(note)
        m["notes"] = m["notes"][-50:]  # keep last 50
        save_memory(m)

Inject ke system prompt

def build_system_prompt():
    soul = load_soul()
    memory = load_memory()

    memory_text = ""
    if memory.get("notes"):
        memory_text = "\n\nYang lo ingat tentang user:\n" + \
            "\n".join(f"- {n}" for n in memory["notes"][-20:])

    if memory.get("user_name"):
        memory_text = f"\n\nNama user: {memory['user_name']}" + memory_text

    return f"{soul}{memory_text}"

[-20:] = limit 20 notes terakhir biar context ga bloat.

Command handlers

async def remember_cmd(update, context):
    """User: /remember <text>"""
    text = " ".join(context.args)
    if not text:
        await update.message.reply_text(
            "Format: /remember <text>\n"
            "Contoh: /remember selalu ringkas output >20 baris"
        )
        return
    if is_suspicious_note(text):
        await update.message.reply_text(
            "Note ini keliatan kayak prompt injection / system instruction. "
            "Ga gue save. Tulis lebih natural ya."
        )
        return
    add_memory_note(text)
    await update.message.reply_text(f"OK, gue inget: {text}")

async def forget_cmd(update, context):
    """User: /forget <substring>"""
    substring = " ".join(context.args)
    if not substring:
        await update.message.reply_text("Format: /forget <substring>")
        return
    m = load_memory()
    before = len(m.get("notes", []))
    m["notes"] = [
        n for n in m.get("notes", []) 
        if substring.lower() not in n.lower()
    ]
    save_memory(m)
    removed = before - len(m["notes"])
    await update.message.reply_text(f"Removed {removed} note(s).")

async def memory_cmd(update, context):
    """User: /memory (list all notes)"""
    m = load_memory()
    if not m.get("notes"):
        await update.message.reply_text("Memory kosong.")
        return
    text = "Memory notes:\n" + "\n".join(
        f"{i+1}. {n}" for i, n in enumerate(m["notes"])
    )
    await update.message.reply_text(text[:4000])

def is_suspicious_note(text: str) -> bool:
    suspicious = [
        r'absolute mode', r'eliminate emojis', r'you are now',
        r'ignore previous', r'reply in the language', r'as an AI'
    ]
    text_lower = text.lower()
    for p in suspicious:
        if re.search(p, text_lower):
            return True
    return len(text) > 500

History implementation

Struktur

[
  {"role": "user", "content": "halo"},
  {"role": "assistant", "content": "Halo Gutluc! Lagi ngapain?"},
  {"role": "user", "content": "cek disk"},
  {"role": "assistant", "content": "<tool_use>...</tool_use>"},
  {"role": "user", "content": "Tool execution. Hasil: ..."},
  {"role": "assistant", "content": "Free 1.5GB dari 6.7GB."}
]

Roles: user (dari user atau dari tool result), assistant (dari LLM).

Load & save

HISTORY_DIR = Path.home() / "agent" / "data" / "history"

def load_history(chat_id: int) -> list:
    f = HISTORY_DIR / f"{chat_id}.json"
    if f.exists():
        try:
            return json.loads(f.read_text())
        except (json.JSONDecodeError, OSError):
            pass
    return []

def save_history(chat_id: int, history: list):
    HISTORY_DIR.mkdir(parents=True, exist_ok=True)
    sanitized = []
    for msg in history:
        if isinstance(msg, dict) and "content" in msg:
            sanitized.append({
                **msg, 
                "content": redact_secrets(msg["content"])
            })
        else:
            sanitized.append(msg)
    (HISTORY_DIR / f"{chat_id}.json").write_text(
        json.dumps(sanitized, ensure_ascii=False, indent=2)
    )

Sanitize wajib sebelum save — biar credential yang tau-tau muncul di output ga ke-persist.

Sliding window

def ask_ai_agentic(chat_id, user_text):
    history = load_history(chat_id)

    # Build messages: system + last N history + new user msg
    messages = [
        {"role": "system", "content": build_system_prompt()},
        *history[-30:],  # last 30 messages
        {"role": "user", "content": user_text}
    ]

    # ... agentic loop

    # After loop, save updated history
    history.append({"role": "user", "content": user_text})
    history.append({"role": "assistant", "content": final_answer})

    # Optionally save tool_use turns juga
    save_history(chat_id, history[-60:])  # keep last 60

/reset command

async def reset_cmd(update, context):
    """User: /reset (archive + clear history)"""
    chat_id = update.effective_chat.id
    f = HISTORY_DIR / f"{chat_id}.json"

    if f.exists():
        # Archive
        archive_dir = HISTORY_DIR / "archive"
        archive_dir.mkdir(exist_ok=True)
        archive_file = archive_dir / f"{chat_id}-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json"
        archive_file.write_text(f.read_text())

        # Reset
        f.write_text("[]")

    await update.message.reply_text(
        "Chat history di-reset. Mulai dari nol."
    )

Pattern auto-detect koreksi

Beyond explicit /remember, agent bisa auto-detect koreksi dari user:

KOREKSI_PATTERNS = [
    r'\b(stop|jangan|ga usah|gak usah) (?P<verb>\w+)',
    r'\b(?P<verb>\w+) jangan (?P<obj>\w+)',
    r'gue (mau|prefer|suka) (?P<pref>.+) bukan',
    r'kalo (saya|gue|gua) bilang (?P<trigger>.+), kasih (?P<response>.+)',
]

def detect_correction(user_text: str) -> str | None:
    for pattern in KOREKSI_PATTERNS:
        match = re.search(pattern, user_text, re.IGNORECASE)
        if match:
            return user_text  # full text as note
    return None

async def handle_message(update, context):
    text = update.message.text

    correction = detect_correction(text)
    if correction:
        add_memory_note(correction)
        # Lanjut process normal

    response = ask_ai_agentic(chat_id, text)
    await update.message.reply_text(response)

Anti-patterns

❌ Memory tanpa size limit

def add_memory_note(note):
    m = load_memory()
    m["notes"].append(note)  # ← grows forever
    save_memory(m)

Solusi: m["notes"] = m["notes"][-50:] setelah append.

❌ History tanpa redaction

def save_history(chat_id, history):
    Path(f"history/{chat_id}.json").write_text(json.dumps(history))
    # ← credential bisa ke-leak

Wajib redact:

sanitized = [{**msg, "content": redact_secrets(msg["content"])} for msg in history]

❌ History terlalu panjang di context

messages = [system, *history, new_user_msg]  # all history

Bloat context, latency naik. Limit:

messages = [system, *history[-30:], new_user_msg]

❌ Memory dan history bercampur

# Salah: simpan task progress di memory
add_memory_note("Sedang ngerjain feature X")

# Salah: simpan preferensi di history
history.append({"role": "system", "content": "user prefer gue/lo"})

Beda tempat: preferensi → memory, conversation → history.

❌ Reset memory tanpa konfirmasi

@command("/reset_memory")
def reset_memory_cmd(update, context):
    save_memory({"notes": []})  # ← user yang lupa bisa hapus semua

Tambah konfirmasi:

@command("/reset_memory")
def reset_memory_cmd(update, context):
    chat_id = update.effective_chat.id
    pending_confirmations[chat_id] = "reset_memory"
    await update.message.reply_text(
        f"Ada {len(load_memory()['notes'])} notes. "
        "Yakin mau hapus semua? Balas 'ya hapus semua' untuk konfirmasi."
    )

Audit memory periodically

Tiap minggu / bulan:

cat ~/agent/data/memory.json | jq '.notes' | less

Hapus yang stale, redundant, atau prompt injection:

# Manual edit
vim ~/agent/data/memory.json

# Atau via command:
/forget <substring>

Backup memory

# Periodic backup
cp ~/agent/data/memory.json ~/agent/data/backups/memory-$(date +%Y%m%d).json

# Atau ke S3 / cloud:
aws s3 cp ~/agent/data/memory.json s3://my-backups/agent/memory-$(date +%Y%m%d).json

History per-user juga di-backup kalo penting, tapi biasanya size kecil dan ga critical (vs memory yang berisi preferensi penting).