Claude の Dreaming を長期稼働エージェントのメモリ整理に組み込む

import json
import os
from datetime import datetime, timezone
from pathlib import Path
from anthropic import Anthropic
 
client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
MEMORY_FILE = Path("memory.jsonl")
MODEL = "claude-sonnet-4-6"
 
def load_memories() -> list[dict]:
    if not MEMORY_FILE.exists():
        return []
    rows = []
    for line in MEMORY_FILE.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line:
            rows.append(json.loads(line))
    # status が active のものだけを整理対象にする
    return [m for m in rows if m.get("status", "active") == "active"]
 
CONSOLIDATION_PROMPT = """\
あなたは長期稼働エージェントのメモリ管理担当です。
以下は時系列で蓄積された運用メモの一覧です（古い順）。
次の方針で整理し、JSON だけを返してください。
 
1. 同じ対象について新旧で矛盾するメモがあれば、最新を正とし古い方を stale にする
2. 個別の出来事を3件以上横断して同じ教訓が出ているなら、1件の一般化された方針に畳み込む
3. 一度きりで再現性のない記録は archive にする
4. 残すメモは原文の事実を改変しない
 
返却フォーマット:
{
  "keep": [{"id": "...", "text": "...", "merged_from": ["id1","id2"]}],
  "stale": ["id3", "id4"],
  "archive": ["id5"]
}
 
メモ一覧:
"""
 
def consolidate(memories: list[dict]) -> dict:
    listing = "\n".join(
        f'- id={m["id"]} created={m["created"]} : {m["text"]}'
        for m in memories
    )
    resp = client.messages.create(
        model=MODEL,
        max_tokens=4096,
        messages=[{"role": "user", "content": CONSOLIDATION_PROMPT + listing}],
    )
    raw = resp.content[0].text
    start, end = raw.find("{"), raw.rfind("}")
    return json.loads(raw[start : end + 1])
 
def apply_plan(memories: list[dict], plan: dict) -> list[dict]:
    by_id = {m["id"]: m for m in memories}
    now = datetime.now(timezone.utc).isoformat()
    result = []
    demoted = set(plan.get("stale", [])) | set(plan.get("archive", []))
    for item in plan.get("keep", []):
        base = by_id.get(item["id"], {})
        result.append({
            "id": item["id"],
            "created": base.get("created", now),
            "updated": now,
            "text": item["text"],
            "status": "active",
            "merged_from": item.get("merged_from", []),
        })
    for mid in demoted:
        if mid in by_id:
            row = dict(by_id[mid])
            row["status"] = "stale" if mid in plan.get("stale", []) else "archived"
            row["updated"] = now
            result.append(row)
    return result
 
def main():
    memories = load_memories()
    if len(memories) < 10:
        print(f"active memories = {len(memories)}: 整理は不要")
        return
    plan = consolidate(memories)
    merged = apply_plan(memories, plan)
    MEMORY_FILE.write_text(
        "\n".join(json.dumps(m, ensure_ascii=False) for m in merged) + "\n",
        encoding="utf-8",
    )
    kept = sum(1 for m in merged if m["status"] == "active")
    print(f"整理完了: active {len(memories)} -> {kept} 件")
 
if __name__ == "__main__":
    main()

残す: 再現性のある判断基準・複数回確認された事実・現在も有効な制約
stale にする: 新しいメモと矛盾する古い事実
archive にする: 一度きりの出来事・特定セッションでしか意味を持たない記録
触らない: 残すと決めたメモの「事実」部分（表現は整えてよいが内容は改変しない）