Claude Code のセッション記憶を永続化する自作メモリシステム｜JSONL×Git×FTS5×ベクトル検索で揮発性を解消する

export/
  2026-03_desktop-win.jsonl   # Windowsデスクトップが追記するファイル
  2026-03_laptop-mac.jsonl    # Macノートが追記するファイル
  2026-03_server-linux.jsonl  # 開発サーバーが追記するファイル

[general]
machine_name = "desktop-win"   # このデバイス固有の識別子（スペースなし推奨）
project      = "main"          # プロジェクト名（複数プロジェクトを分離）
export_dir   = "export"        # JSONLを保存するディレクトリ（リポジトリ内）
db_path      = ".cache/memory.db"  # SQLiteキャッシュ（.gitignore対象）

{
  "id":         "550e8400-e29b-41d4-a716-446655440000",
  "session_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "project":    "main",
  "machine":    "desktop-win",
  "timestamp":  "2026-03-25T14:32:11.000Z",
  "question":   "FastAPI の依存性注入でDBセッションを渡すとき、テスト時だけ差し替えるには？",
  "answer":     "dependency_overrides を使います。app.dependency_overrides[get_db] = get_test_db のように...",
  "embedding":  [0.0145, -0.0281, 0.0093, "... (1024次元) ..."]
}

-- キーワード検索用（trigramトークナイザで日本語対応）
CREATE VIRTUAL TABLE IF NOT EXISTS memory_fts USING fts5(
    id        UNINDEXED,
    session_id UNINDEXED,
    project   UNINDEXED,
    machine   UNINDEXED,
    timestamp UNINDEXED,
    question,          -- 検索対象
    answer,            -- 検索対象
    tokenize = "trigram"
);

-- ベクトル検索用（sqlite-vec の仮想テーブル）
CREATE VIRTUAL TABLE IF NOT EXISTS memory_vec USING vec0(
    id        TEXT PRIMARY KEY,
    embedding FLOAT[1024]
);

import sqlite3
import sqlite_vec
import json

def open_db(db_path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)
    conn.enable_load_extension(True)
    sqlite_vec.load(conn)          # sqlite-vec を拡張機能としてロード
    conn.enable_load_extension(False)
    conn.row_factory = sqlite3.Row
    return conn

def insert_chunk(conn: sqlite3.Connection, rec: dict, embedding: list[float]):
    """FTS5 と vec0 に同時挿入。id 衝突時は上書き（冪等）。"""
    conn.execute(
        """INSERT OR REPLACE INTO memory_fts
           (id, session_id, project, machine, timestamp, question, answer)
           VALUES (?, ?, ?, ?, ?, ?, ?)""",
        (rec["id"], rec.get("session_id",""), rec.get("project",""),
         rec.get("machine",""), rec["timestamp"],
         rec["question"], rec["answer"])
    )
    conn.execute(
        "INSERT OR REPLACE INTO memory_vec(id, embedding) VALUES (?, ?)",
        (rec["id"], json.dumps(embedding))
    )
    conn.commit()

import json
import uuid
from pathlib import Path
from datetime import datetime, timezone


def load_session_log(project_dir: str) -> list[dict]:
    """~/.claude/projects/{project}/ 以下の最新セッションログを読み込む。"""
    base = Path(project_dir)
    # セッションログは更新日時が新しいものを使用
    logs = sorted(base.glob("*.jsonl"), key=lambda p: p.stat().st_mtime, reverse=True)
    if not logs:
        return []
    chunks = []
    with open(logs[0], encoding="utf-8") as f:
        for line in f:
            try:
                chunks.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return chunks


def extract_qa_pairs(log_entries: list[dict]) -> list[dict]:
    """
    セッションログから Q&A ペアを抽出する。
    - role="user" の直後に role="assistant" が来るペアを1チャンクとする。
    - 質問が10文字未満のペアはスキップ。
    """
    session_id = str(uuid.uuid4())
    pairs = []
    i = 0
    while i < len(log_entries) - 1:
        entry = log_entries[i]
        next_entry = log_entries[i + 1]

        # ユーザー→AI の順のペアを探す
        if (entry.get("type") == "user" and
                next_entry.get("type") == "assistant"):
            question = _extract_text(entry)
            answer   = _extract_text(next_entry)
            ts       = entry.get("timestamp") or datetime.now(timezone.utc).isoformat()

            if len(question) >= 10 and len(answer) >= 10:
                pairs.append({
                    "id":         str(uuid.uuid4()),
                    "session_id": session_id,
                    "timestamp":  ts,
                    "question":   question[:2000],   # 上限2000文字
                    "answer":     answer[:4000],     # 上限4000文字
                })
            i += 2
        else:
            i += 1
    return pairs


def _extract_text(entry: dict) -> str:
    """メッセージエントリからテキストを取り出す。"""
    content = entry.get("message", {}).get("content", "")
    if isinstance(content, str):
        return content.strip()
    if isinstance(content, list):
        # content がブロックリストの場合（tool_use 等が含まれる）
        texts = [b.get("text","") for b in content if b.get("type") == "text"]
        return " ".join(texts).strip()
    return ""

import json
import math
from datetime import datetime, timezone


def hybrid_search(
    conn,
    query: str,
    query_embedding: list[float],
    top_k: int = 5,
    rrf_k: int = 60,
    half_life_days: float = 30.0,
    project: str | None = None,
) -> list[dict]:
    """
    FTS5 + ベクトル検索を RRF で統合し、時間減衰を乗算して返す。
    project を指定するとそのプロジェクトの記憶のみを対象にする。
    """
    project_filter = f"AND project = '{project}'" if project else ""

    # --- FTS5 キーワード検索 ---
    fts_rows = conn.execute(
        f"""SELECT id, project, machine, timestamp, question, answer
            FROM memory_fts
            WHERE memory_fts MATCH ? {project_filter}
            ORDER BY rank
            LIMIT ?""",
        (query, top_k * 4)
    ).fetchall()

    # --- ベクトル検索 ---
    vec_rows = conn.execute(
        f"""SELECT f.id, f.project, f.machine, f.timestamp, f.question, f.answer,
                   vec_distance_cosine(v.embedding, ?) AS distance
            FROM memory_vec v
            JOIN memory_fts f ON v.id = f.id
            WHERE 1=1 {project_filter.replace("project", "f.project")}
            ORDER BY distance ASC
            LIMIT ?""",
        (json.dumps(query_embedding), top_k * 4)
    ).fetchall()

    # --- RRF スコアを計算 ---
    scores: dict[str, dict] = {}

    for rank, row in enumerate(fts_rows):
        r = dict(row)
        scores[r["id"]] = {"data": r, "rrf": 1.0 / (rrf_k + rank)}

    for rank, row in enumerate(vec_rows):
        r = dict(row)
        if r["id"] not in scores:
            scores[r["id"]] = {"data": r, "rrf": 0.0}
        scores[r["id"]]["rrf"] += 1.0 / (rrf_k + rank)

    # --- 時間減衰を乗算（直近の記憶を優先）---
    for entry in scores.values():
        decay = _time_decay(entry["data"]["timestamp"], half_life_days)
        entry["final_score"] = entry["rrf"] * decay

    ranked = sorted(scores.values(), key=lambda x: x["final_score"], reverse=True)
    return [e["data"] for e in ranked[:top_k]]


def _time_decay(timestamp_str: str, half_life_days: float) -> float:
    """
    半減期を使った時間減衰。
    経過日数が half_life_days になるとスコアが 0.5 になる。
    30日後 → 0.50、60日後 → 0.25、90日後 → 0.125
    """
    try:
        ts  = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
        now = datetime.now(timezone.utc)
        elapsed = (now - ts).total_seconds() / 86400.0
        return math.pow(0.5, elapsed / half_life_days)
    except Exception:
        return 1.0

{
  "env": {
    "MEMORY_DIR": "C:/Users/yourname/memory-engine"
  },
  "hooks": {
    "SessionStart": [
      {
        "matcher": "",
        "hooks": [
          {
            "type": "command",
            "command": "cd \"${MEMORY_DIR}\" && git pull --quiet --rebase && uv run python -m src session-start --inject"
          }
        ]
      }
    ],
    "Stop": [
      {
        "matcher": "",
        "hooks": [
          {
            "type": "command",
            "command": "cd \"${MEMORY_DIR}\" && uv run python -m src session-end && git add export/ && git commit -m \"memory: session \"$(date +%Y%m%d-%H%M%S)\" \" --quiet && git push --quiet"
          }
        ]
      }
    ]
  }
}

@cli.command("session-start")
@click.option("--project", default="main")
@click.option("--inject", is_flag=True,
              help="直近の記憶サマリーをstdoutに出力しセッション開始時に注入する")
def session_start(project: str, inject: bool):
    """
    1. git pull は Hook コマンド内で実行済み
    2. 当月の全JSONLをSQLiteに再構築（モデルロードなし・embedはJSONLから読む）
    3. --inject のときだけ直近5件をstdoutへ（セッション冒頭に表示される）
    """
    cfg = load_config()
    # embedがJSONLに保存されているため load_model=False で高速再構築
    db.rebuild_from_jsonl(cfg, project=project, load_model=False)

    if inject:
        recent = db.get_recent(cfg, project=project, limit=5)
        if recent:
            print("\n=== 直近の会話記憶（自動注入） ===")
            for r in recent:
                date = r["timestamp"][:10]
                mach = r.get("machine", "?")
                print(f"\n[{date} / {mach}]")
                print(f"Q: {r['question'][:100]}...")
                print(f"A: {r['answer'][:150]}...")
            print("=== ここまで ===\n")

@cli.command("session-end")
@click.option("--project", default="main")
def session_end(project: str):
    """
    1. セッションログを読み込んでQ&Aチャンクに分割
    2. Ruri v3-310mでベクトルを計算（ここだけモデルをロード）
    3. SQLiteとJSONLに保存（冪等：同じidは上書き）
    4. git commit/push はHookコマンド内で実行
    """
    cfg = load_config()
    project_dir = cfg.get("claude_projects_dir", "~/.claude/projects")
    log_entries = chunker.load_session_log(os.path.expanduser(project_dir))
    pairs = chunker.extract_qa_pairs(log_entries)

    if not pairs:
        print("[memory] 保存するチャンクがありません。")
        return

    # モデルロードはここだけ（セッション開始時は不要）
    model = embedder.load()
    conn  = db.open_db(cfg)
    jsonl = export.get_jsonl_path(cfg, project)  # 例: export/2026-03_desktop-win.jsonl

    saved = 0
    for pair in pairs:
        pair["project"] = project
        pair["machine"] = cfg["machine_name"]
        text = f"{pair['question']} {pair['answer']}"
        emb  = embedder.embed_texts([text], model)[0]
        pair["embedding"] = emb
        db.insert_chunk(conn, pair, emb)
        export.append_jsonl(jsonl, pair)
        saved += 1

    conn.close()
    print(f"[memory] {saved}件の記憶を保存しました → {jsonl}")

---
name: memory-recall
description: |
  過去の会話記憶をハイブリッド検索で引き出す。
  「思い出して」「前に話した」「あの時の結論は」「覚えている？」
  などのトリガーが出たとき使う。
  検索はローカルSQLiteで実行するためLLMのAPIコストはゼロ。
allowed-tools:
  - Bash
---

# 記憶の検索

## 手順

1. ユーザーが参照したい内容をクエリとして抽出する
2. 以下のコマンドを実行する：

```
cd "${MEMORY_DIR}" && uv run python -m src search "クエリ"
```

3. 返ってきた記憶の中から関連するものを選んでユーザーに提示する
4. 記憶の日付・デバイスも一緒に伝える（「〇月〇日にWindowsで話した内容です」）

## 検索オプション

- `--top-k 10`：より多くの候補を取得（デフォルト5）
- `--project work`：特定プロジェクトのみ対象
- `--half-life 60`：時間減衰を緩め（デフォルト30日）

## 注意

- 記憶が見つからない場合は「該当する記憶がありません」と正直に伝える
- 古い記憶（半年以上前）は情報が古い可能性があると注記する

@cli.command("search")
@click.argument("query")
@click.option("--project", default=None, help="プロジェクトでフィルタ")
@click.option("--top-k", default=5)
@click.option("--half-life", default=30.0)
def search_cmd(query: str, project: str | None, top_k: int, half_life: float):
    """ハイブリッド検索を実行して関連記憶を返す（APIコストゼロ）。"""
    cfg   = load_config()
    model = embedder.load()
    qemb  = embedder.embed_query(query, model)
    conn  = db.open_db(cfg)
    results = search.hybrid_search(conn, query, qemb,
                                   top_k=top_k,
                                   half_life_days=half_life,
                                   project=project)
    conn.close()

    if not results:
        print("該当する記憶が見つかりませんでした。")
        return

    for i, r in enumerate(results, 1):
        date = r["timestamp"][:10]
        mach = r.get("machine", "?")
        proj = r.get("project", "?")
        print(f"\n[{i}] {date} / {mach} / project:{proj}")
        print(f"Q: {r['question']}")
        print(f"A: {r['answer'][:300]}..." if len(r['answer']) > 300 else f"A: {r['answer']}")

# 月初めに手動実行、またはタスクスケジューラ/launchd で自動化
cd "${MEMORY_DIR}" && uv run python -m src monthly-merge

-- クラウドストレージ上のアーカイブDBをアタッチ
ATTACH DATABASE "/path/to/cloud/archive.db" AS arc;

-- 当月 + アーカイブを UNION して検索
SELECT * FROM memory_fts WHERE memory_fts MATCH ?
UNION ALL
SELECT * FROM arc.memory_fts WHERE arc.memory_fts MATCH ?
ORDER BY timestamp DESC;

@cli.command("rebuild")
@click.option("--project", default=None)
@click.option("--with-model", is_flag=True,
              help="embeddingがないJSONLレコードも再計算する（時間がかかる）")
def rebuild(project: str | None, with_model: bool):
    """
    全JSONLからSQLiteを完全再構築する。
    DBが破損した場合や新しいデバイスへの移行時に使用する。
    """
    cfg = load_config()
    model = embedder.load() if with_model else None
    conn  = db.open_db(cfg, recreate=True)  # テーブルを作り直す
    jsonl_dir = Path(cfg["export_dir"])
    total = 0

    for jsonl_path in sorted(jsonl_dir.glob("*.jsonl")):
        if project and f"_{project}_" not in jsonl_path.name:
            continue
        for rec in export.iter_jsonl(jsonl_path):
            emb = rec.get("embedding")
            if not emb:
                if not model:
                    continue  # --with-model なしはスキップ
                text = f"{rec['question']} {rec['answer']}"
                emb  = embedder.embed_texts([text], model)[0]
            db.insert_chunk(conn, rec, emb)
            total += 1

    conn.close()
    print(f"[memory] 再構築完了：{total}件")

memory-engine/
  pyproject.toml          # プロジェクト設定・依存パッケージ
  config.toml             # デバイス固有設定（machine_name等）
  .gitignore              # .cache/・*.db を除外
  .gitattributes          # *.jsonl を LF に統一
  export/                 # JSONL ファイル（Git管理される正本）
    2026-03_desktop-win.jsonl
    2026-03_laptop-mac.jsonl
  .cache/                 # SQLiteキャッシュ（Git管理外）
    memory.db
  src/
    __init__.py
    __main__.py           # CLI（session-start/end, search, rebuild）
    db.py                 # SQLite操作（open_db, insert_chunk, rebuild_from_jsonl）
    search.py             # ハイブリッド検索（hybrid_search, _time_decay）
    export.py             # JSONL操作（get_jsonl_path, append_jsonl, iter_jsonl）
    embedder.py           # Ruri v3-310m（load, embed_texts, embed_query）
    chunker.py            # チャンク化（load_session_log, extract_qa_pairs）

[project]
name = "memory-engine"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "sentence-transformers>=3.0",  # Ruri v3-310m のロード・推論
    "sqlite-vec>=0.1.6",           # ベクトル検索（SQLite拡張）
    "click>=8.0",                  # CLI フレームワーク
]

[project.scripts]
memory = "src.__main__:cli"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

# 1. uv のインストール（まだの場合）
curl -LsSf https://astral.sh/uv/install.sh | sh
# Windows: powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

# 2. リポジトリをクローン（または新規作成）
git clone git@github.com:your-account/memory-engine.git
cd memory-engine

# 3. 依存パッケージをインストール
uv sync

# 4. 設定ファイルを編集（machine_name を現在のデバイス名に変更）
# config.toml の machine_name を設定

# 5. 動作確認
uv run python -m src --help

# 6. 初回DB構築（すでにJSONLがある場合）
uv run python -m src rebuild

.cache/
*.db
__pycache__/
.venv/

*.jsonl  text eol=lf
*.toml   text eol=lf
*.md     text eol=lf
*.py     text eol=lf