マルチエージェント横断の共通記憶エンジン「Engrim」：SQLiteとMCPで実現するローカルファーストなエピソード記憶アーキテクチャ

TL;DR

• Google AntigravityやCursor、Claude Codeなどの複数開発エージェント間におけるコンテキスト断絶を解消する、ローカルファーストな共通記憶エンジン「Engrim」の設計思想を解説します。
• SQLite FTS5（BM25）とModel2VecによるReciprocal Rank Fusion（RRF）ハイブリッド検索を採用し、外部APIに依存しないゼロレイテンシでの高精度コンテキスト復元を実現します。
• MCP（Model Context Protocol）とCLIフックによる透過的な統合により、15万トークン超の会話履歴を1,000トークン未満の高品質プロンプトへ圧縮し、セッション再ロードコストを99%削減します。

背景と技術課題

現在のLLMベースのコーディングエージェント（Google Antigravity、Claude Code、Cursor、Windsurfなど）は強力な推論能力と自律性を備えている一方で、セッション間やツール間における「記憶のサイロ化」という重大な構造的課題を抱えています。

1. Attention Dilution（注意希薄化）とコンテキストコストの爆発
プロジェクトが進行し、ファイル群や会話ログのコンテキストウィンドウが数十万トークン規模に達すると、LLMの文脈把握精度（Needle In A Haystack耐性など）は著しく低下します。また、新規セッションを開始するたびに膨大な履歴テキストをモデルへ再ロードする必要があり、API利用コストとTTFT（Time To First Token）が急激に増大します。

2. セッションリセットに伴う文脈喪失（エージェントの健忘症）
ターミナル上のCLIエージェントとIDE内のエージェントを併用する際、以前のセッションで決定した設計方針や、あえて採用を見送った理由（Why not）などの経緯が引き継がれません。その結果、エージェントは過去の決定事項と矛盾するコードを生成し、人間が同一のレビュー指摘を何度も繰り返すことになります。

これらの課題を根本から解決するためには、エージェントの内部状態に依存せず、必要なタイミングで関連知識のみを動的に注入する「クロスエージェントな外部永続化レイヤー」が不可欠となります。

コアアーキテクチャ・仕組みの解説

Engrimは、プロジェクトルートに単一の .engrim.db（SQLite）を配置し、エージェントのエピソード記憶を管理するデーモンレス・アーキテクチャを採用しています。

ハイブリッド検索パイプライン（Zero-Latency Hybrid Search）
外部のマネージドVector DBや商用Embeddings APIを介さず、完全にローカル環境内で完結する検索パイプラインを構築しています。

• 疎な検索（Sparse）: SQLiteの FTS5 拡張を利用したBM25全文検索。シンボル名、エラーログ、ファイルパスなどの完全一致・キーワード照合を担います。
• 密な検索（Dense）: CPU上で高速推論が可能な静的埋め込みモデル（Model2Vec）によるベクトル検索。自然言語による意図や抽象的な概念類似度を評価します。

これら2系統の検索スコアを Reciprocal Rank Fusion（RRF）で統合することで、キーワードの完全一致と文脈的な類似性の双方を両立した記憶抽出を行います。

Agent Provenance Tracking（決定プロベナンス追跡）
すべての記憶レコードには source_agent（例: cursor, antigravity）や session_id、タイムスタンプが記録されます。複数のエージェントが協調する現場において、「どのツールが、どのような根拠でその設計を決定したのか」をフライトレコーダーのように透過的に監査可能です。

graph TD
    subgraph Agents
        A["Cursor / Windsurf"]
        B["Google Antigravity"]
        C["Claude Code"]
    end

    subgraph Integration Layer
        M["MCP stdio Server"]
        H["CLI Hooks / PreInvocation"]
    end

    subgraph Engrim Engine
        S[("SQLite .engrim.db")]
        F["FTS5 (BM25)"]
        V["Model2Vec Embedding"]
        R["RRF Score Ranker"]
    end

    A <-->|MCP Protocol| M
    B <-->|Hook Execution| H
    C <-->|Hook Execution| H

    M <--> S
    H <--> S
    S --> F
    S --> V
    F --> R
    V --> R

ハンズオン・実装/コード例

Engrimは、リポジトリ単位のセットアップコマンドを実行することで、各エージェントの設定ファイルへ自動的に連携フックを注入します。

# Engrimの初期化とエージェント環境の自動検出
$ engrim setup

Detected environments:
- Cursor: Injecting MCP configuration to .cursor/mcp.json
- Antigravity: Installing PreInvocation hook to .gemini/hooks/

以下は、記憶の保存とModel2Vecを用いたエンベディング生成、およびRRFハイブリッド検索を担うコア実装例です。外部ライブラリへの依存を最小限に抑え、CPU環境下でも高速に動作します。

import sqlite3
import json
import numpy as np
from model2vec import StaticModel

class EngrimMemory:
    def __init__(self, db_path: str = ".engrim.db"):
        self.db = sqlite3.connect(db_path)
        # 依存関係が極小かつCPU推論に最適化された静的モデルをロード
        self.embedder = StaticModel.from_pretrained("minishlab/M2V_base_output")
        self._init_db()

    def _init_db(self):
        # メタデータ・FTS5全文検索・ベクトル保存用テーブルの初期化
        self.db.executescript("""
            CREATE TABLE IF NOT EXISTS meta (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                agent TEXT NOT NULL,
                tags TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );
            CREATE VIRTUAL TABLE IF NOT EXISTS memories USING fts5(
                content,
                content='meta',
                content_rowid='id'
            );
            CREATE TABLE IF NOT EXISTS embeddings (
                rowid INTEGER PRIMARY KEY,
                vector BLOB NOT NULL
            );
        """)

    def store_decision(self, agent: str, content: str, tags: list[str]) -> int:
        cursor = self.db.cursor()
        cursor.execute(
            "INSERT INTO meta(agent, tags) VALUES(?, ?)",
            (agent, json.dumps(tags))
        )
        rowid = cursor.lastrowid

        # FTS5インデックスへの登録
        cursor.execute(
            "INSERT INTO memories(rowid, content) VALUES(?, ?)",
            (rowid, content)
        )

        # CPU上で即時エンベディングを生成（数ミリ秒レイテンシ）
        vector = self.embedder.encode([content])[0].astype(np.float32)
        cursor.execute(
            "INSERT INTO embeddings(rowid, vector) VALUES(?, ?)",
            (rowid, vector.tobytes())
        )
        self.db.commit()
        return rowid

    def retrieve_context(self, query: str, top_k: int = 5, rrf_k: int = 60) -> list[dict]:
        cursor = self.db.cursor()

        # 1. 疎な検索 (SQLite FTS5 BM25)
        fts_cursor = cursor.execute("""
            SELECT rowid, rank
            FROM memories
            WHERE memories MATCH ?
            ORDER BY rank
            LIMIT ?
        """, (query, top_k * 2))
        sparse_ranks = {row[0]: rank for rank, row in enumerate(fts_cursor, start=1)}

        # 2. 密な検索 (Model2Vec コサイン類似度)
        query_vec = self.embedder.encode([query])[0].astype(np.float32)
        query_norm = np.linalg.norm(query_vec)

        cursor.execute("SELECT rowid, vector FROM embeddings")
        dense_scores = []
        for rowid, blob in cursor.fetchall():
            doc_vec = np.frombuffer(blob, dtype=np.float32)
            denom = query_norm * np.linalg.norm(doc_vec)
            sim = float(np.dot(query_vec, doc_vec) / denom) if denom > 0 else 0.0
            dense_scores.append((rowid, sim))

        dense_scores.sort(key=lambda x: x[1], reverse=True)
        dense_ranks = {rowid: rank for rank, (rowid, _) in enumerate(dense_scores[:top_k * 2], start=1)}

        # 3. Reciprocal Rank Fusion (RRF) によるスコア統合
        all_rowids = set(sparse_ranks.keys()) | set(dense_ranks.keys())
        rrf_scores = {}
        for rowid in all_rowids:
            score = 0.0
            if rowid in sparse_ranks:
                score += 1.0 / (rrf_k + sparse_ranks[rowid])
            if rowid in dense_ranks:
                score += 1.0 / (rrf_k + dense_ranks[rowid])
            rrf_scores[rowid] = score

        top_rowids = sorted(rrf_scores.keys(), key=lambda r: rrf_scores[r], reverse=True)[:top_k]

        results = []
        for rowid in top_rowids:
            cursor.execute(
                "SELECT m.content, t.agent, t.tags FROM memories m JOIN meta t ON m.rowid = t.id WHERE m.rowid = ?",
                (rowid,)
            )
            content, agent, tags = cursor.fetchone()
            results.append({
                "rowid": rowid,
                "content": content,
                "agent": agent,
                "tags": json.loads(tags),
                "rrf_score": rrf_scores[rowid]
            })
        return results

ベンチマークと実務でのトレードオフ

実運用中のアルゴリズム取引システムリポジトリ（計105セッション、累積履歴15万トークン超）を対象に、Engrimの性能評価を行いました。

• コンテキスト圧縮率とトークン削減: セッション起動時に15万トークンを静的に読み込ませる従来方式から、EngrimのRRF検索で現在のアクティブコンテキストに関連するTop-K記憶（約800トークン）のみを動的注入する方式へ移行しました。これにより、プロンプトサイズが劇的に圧縮され、再ロードに伴うトークンコストを99%以上削減できています。
• レイテンシ特性: Model2VecとSQLiteの組み合わせにより、埋め込み生成および検索のオーバーヘッドはローカルCPU環境下で平均10ms未満に収まります。外部API通信特有のネットワークレイテンシが発生しないため、エージェントの応答速度を一切損ないません。
• アーキテクチャ上のトレードオフ: 静的埋め込みモデル（Model2Vec）は推論速度とメモリ効率に優れる反面、プロジェクト固有のドメイン語彙や新規ライブラリに関する抽象的な概念把握においては、大規模な埋め込みモデル（text-embedding-3-large など）に比べて劣る場合があります。この点については、SQLite FTS5の厳密なキーワードマッチング（Sparse検索）を組み合わせることで、語彙の脱落を実用上十分にカバーできる設計としています。

まとめと展望

Engrimは、SQLiteとローカル推論技術を融合することで、トークンウィンドウの制限やAPIコストに依存しない持続可能なマルチエージェント開発基盤を提供します。MCPとCLIフックを活用した透過的な連携により、既存の開発ワークフローを妨げることなく、エピソード記憶のメリットを最大限に享受できます。

今後の課題としては、蓄積された記憶ログをエージェント自身が定期的に要約・圧縮するガベージコレクション機能の実装や、複数リポジトリ間で横断的にナレッジを共有するグローバルストレージ機構への拡張を計画しています。コンテキストウィンドウの長大化が進む現代においても、「何をプロンプトに含めないか」を精密に制御するローカルファーストな記憶エンジンの価値は揺るぎません。