llive v0.2 — RAD 知識庫 × 外部 LLM 連携

Raptor の Research Aggregation Directory (49 分野 / 44,864 docs / 112 MB) を llive 配下に取り込み、生物学的記憶モデルから書き戻し可能にし、 Claude Desktop / LM Studio / Cursor / Ollama / OpenWebUI から呼べる ローカルファースト記憶 LLM として実用化するまでの設計・実装記録。

2026-05-15 441 → 527 tests / 全 PASS ruff clean commits: a75ccd4 → f361d06

1. はじめに

本資料の目的

本資料は llive v0.2 系の RAD 横断エピックを、後から自分で読み返したときに 「なぜこの形にしたか」が再構成できる粒度で記録するためのものです。API リファレンスというより 設計思想と判断ポイントのスナップショットとして書いています。

対象読者

• llive 本体の今後の保守・拡張をする自分自身
• RAD/RAG・MCP・LLM backend 抽象化を別プロジェクトで再利用したい自分自身
• Phase 1-7 のメインロードマップと、横断エピックの関係を見失ったときの自分自身

前提

• Python 3.11 系 (pyproject.toml で >=3.11,<3.12 に固定)
• Raptor 側に RAD コーパス (D:/docs/<分野>_v2/) が存在
• 本体は stdlib のみで動作。[mcp] / [llm] / [vlm] は extras

2. 全体アーキテクチャ

取り込み (A) → 知識庫 API (B) → 外部 LLM 連携 (C) の 3 層構造で、C 層は さらに backend 抽象化 (C-1)・MCP server (C-2)・OpenAI HTTP server (C-3) の 3 経路に分岐します。

3. Phase A — RAD 取り込み層

3.1 設計判断

• 物理コピー: ハードリンクや symlink ではなく、ファイルを複製。理由は llive 単独で完結する独立した知識庫として配布可能にするため。再現は scripts/import_rad.py が担保。
• stdlib のみ: argparse / pathlib / shutil / hashlib / json だけで成立。 取り込みは PyPI 配布 wheels には含まれない一時操作なので、依存追加すべきでない。
• 差分判定 = size + mtime: SHA-256 ハッシュは 44,864 ファイルで遅すぎる。 --force で完全再コピー可能。
• スマート判定既定: <domain>_v2 を優先、無い分野 (tui_corpus, security_papers_2025_2026) は v1 を採用。--include-legacy で両方含める。
• 書き層を予約: _learned/ ディレクトリを取り込み完了時に作っておく。 Phase B が後から消費。README を一緒に置いて「ここは何か」を残す。

3.2 スクリプト構造

# 主要関数 (~250 行)
def resolve_source(arg) -> Path: # --source > $LLIVE_RAD_SOURCE > $RAPTOR_CORPUS_DIR > D:/docs
def resolve_dest(arg) -> Path:   # --dest > $LLIVE_RAD_DIR > <repo>/data/rad
def list_corpora(source, include_legacy, only) -> list[Path]:  # スマート判定
def sync_corpus(src, dst, *, mirror, dry_run, force) -> CorpusStat:
def write_index(dest, report, *, dry_run) -> Path:  # _index.json 生成
def ensure_learned_dir(dest, *, dry_run):           # 書き層予約

3.3 ディレクトリレイアウト

data/rad/
  <domain>_v2/         # Raptor からの読み専用ミラー
  _learned/<domain>/   # llive 学習堆積 (Phase B で書く)
    <doc_id>.md
    <doc_id>.provenance.json
  _index.json          # 分野/ファイル数/バイト/取り込み日時
  README.md            # 構造説明 (gitignore 除外)

4. Phase B — RadCorpusIndex (読み + 書き)

4.1 読み+書き統合の理由

最初の検討では「読み API」と「書き API」を別クラスにする案がありました。が、 Consolidator (生物学的記憶モデル) の出口から見ると、両方を同じ ハンドルで扱えた方が呼び出し側がシンプルです。

# Bad (2 クラスに分かれている案)
reader = RadCorpusReader()
writer = RadCorpusWriter()
writer.append(...)
hits = reader.query(...)  # reader と writer が同じ root を見ているか保証されない

# Good (実装)
idx = RadCorpusIndex()        # 単一エントリポイント
idx.append_learning(...)
hits = idx.query(...)         # 同じ root、書いた直後に検索可能 (reload 自動)

4.2 API 表面

4.3 パストラバーサル防御

def read_document(self, domain, rel_path):
    info = self.get_domain_info(domain)
    full = (info.path / rel_path).resolve()
    # resolve() 後にドメインルート配下か検証
    try:
        full.relative_to(info.path.resolve())
    except ValueError as exc:
        raise PermissionError(f"path traversal blocked: {rel_path}") from exc

4.4 Consolidator (生物学的記憶モデル) 統合

llive の Consolidator は episodic → semantic への「Wiki Compile」フェーズで ConceptPage を作ります。この出口に RAD 書き戻しを差し込みます。

def _mirror_to_rad(self, page, cluster_events, result):
    if self.rad_index is None:
        return                                   # 互換: rad_index=None なら何もしない
    try:
        from llive.memory.rad.append import append_learning
        domain = self._rad_domain_for(page)        # page_type を [a-z0-9_]+ に
        prov = Provenance(
            source_type="consolidator",
            source_id=page.concept_id,
            derived_from=[e.event_id for e in cluster_events],
            confidence=0.8,
        )
        append_learning(self.rad_index, domain,
                        f"# {page.title}\n\n{page.summary}\n",
                        prov, doc_id=page.concept_id)
    except Exception as exc:
        result.errors.append(f"rad_mirror: {exc}")   # non-fatal

5. Phase C-1 — LLM Backend Abstraction

5.1 抽象化の形

4 つの backend (Mock / Anthropic / OpenAI / Ollama) を 単一の generate(GenerateRequest) -> GenerateResponse に寄せます。 これにより consolidation / MCP tool / HTTP server すべてが backend に依存しない コードで書けます。

@dataclass
class GenerateRequest:
    prompt: str
    system: str | None = None
    max_tokens: int = 1024
    temperature: float = 0.2
    stop: list[str] = field(default_factory=list)
    model: str | None = None                  # backend-specific id
    images: list[bytes | Path | str] = field(default_factory=list)  # VLM (C-1.1)

@dataclass
class GenerateResponse:
    text: str
    finish_reason: str = "stop"
    backend: str = ""
    model: str = ""
    raw: dict = field(default_factory=dict)        # backend native payload (debug)

5.2 backend 解決の優先順位

def resolve_backend(name=None) -> LLMBackend:
    # 1. 引数 name
    # 2. $LLIVE_LLM_BACKEND
    # 3. $ANTHROPIC_API_KEY が在れば anthropic
    # 4. $OPENAI_API_KEY が在れば openai
    # 5. $OLLAMA_HOST が在れば ollama
    # 6. mock (no-network fallback)

5.3 VLM 拡張 (C-1.1)

request.images に bytes | Path | str を渡せるようにし、 各 backend がそれぞれの multimodal フォーマットに変換します。

def _normalise_image(img) -> tuple[str, str]:
    # Path → 拡張子から media_type 推定
    # bytes → magic bytes (PNG/JPEG/GIF/WEBP) 推定、不明は image/png
    # str → 既に base64、media_type=image/png 仮定
    return media_type, base64_str

6. Phase C-2 — MCP Server

6.1 transport 分離設計

MCP の実装で最も価値が高い設計判断は、tool 関数と MCP transport を分けること。

def tool_query_rad(keywords, ...) -> list[dict]:
    # JSON 直列化可能な値だけ受け取り、
    # JSON 直列化可能な値だけ返す
    ...

def dispatch(name, args) -> Any:
    if name == "query_rad":
        return tool_query_rad(**args)
    ...

async def _call_tool(name, arguments):
    try:
        result = dispatch(name, arguments)
    except Exception as e:
        return error_text(e)
    return [TextContent(
        type="text",
        text=json.dumps(result))]

6.2 8 つの tool

6.3 smoke E2E (実 MCP client で round-trip)

async def _call_query_rad(rad_root):
    async with stdio_client(_server_params(rad_root)) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            response = await session.call_tool(
                "query_rad",
                {"keywords": "buffer", "limit": 5})
            return json.loads(response.content[0].text)

公式 mcp client から subprocess 起動 → initialize → list_tools / call_tool までを 本物の stdio プロトコルで実行。モックでなく、Claude Desktop と同じ経路で 動作確認できています。

7. Phase C-3 — OpenAI 互換 HTTP server

7.1 なぜ HTTP も必要か

Ollama / OpenWebUI / その他多くのローカル LLM フロントエンドは MCP client ではなく、OpenAI HTTP 互換クライアントです。 これらから llive を呼ぶには、/v1/chat/completions を喋るサーバが必要です。

7.2 エンドポイント

実装は stdlib http.server.ThreadingHTTPServer + BaseHTTPRequestHandler。 依存ゼロ、起動は py -3.11 -m llive.server.openai_api だけ。

7.3 RAG-on-by-flag 拡張

POST /v1/chat/completions
{
  "model": "llive-rad/llama3.1",
  "messages": [{"role":"user","content":"Explain buffer overflow"}],
  "x_rad_domain": "security_corpus_v2",   # llive 拡張
  "x_rad_hint_limit": 5                      # llive 拡張
}

x_rad_domain が与えられると、サーバ側で最新の user メッセージで query_rad を実行し、上位 N excerpt を system block に prepend してから LLM backend に流します。レスポンスには x_llive_backend / x_llive_model / x_llive_rad_hints を含めてどの doc が使われたか開示します。

8. テスト戦略

9. 利用シナリオ別ガイド

シナリオ A: Claude Desktop / Cursor から RAD を使う

# 1. 取り込み (1 回だけ)
py -3.11 -m pip install -e .[mcp]
py -3.11 scripts/import_rad.py

# 2. MCP host の config に登録 (Claude Desktop の例)
{
  "mcpServers": {
    "llive": {
      "command": "py",
      "args": ["-3.11", "-m", "llive.mcp.server"],
      "env": {"LLIVE_RAD_DIR": "D:/projects/llive/data/rad"}
    }
  }
}

シナリオ B: Ollama / OpenWebUI から llive を呼ぶ

# 1. HTTP server を起動
py -3.11 -m llive.server.openai_api --host 127.0.0.1 --port 8765

# 2. クライアント側
base_url = "http://127.0.0.1:8765/v1"
api_key  = "any"                    # llive 側は認証なし
model    = "llive-rad/llama3.1"      # Ollama の llama3.1 を経由

シナリオ C: 内部コードから直接 query する

from llive.memory.rad import RadCorpusIndex
from llive.memory.rad.query import query
from llive.memory.provenance import Provenance
from llive.memory.rad.append import append_learning

idx = RadCorpusIndex()
hits = query(idx, "buffer overflow", domain="security_corpus_v2", limit=5)

for h in hits:
    print(h.domain, h.doc_path.name, h.score, h.excerpt[:80])

# 学習物を書き戻す
append_learning(
    idx,
    domain="security_concept",
    content="# Heap spray\n\nSummary...",
    provenance=Provenance(source_type="manual", source_id="note-001"),
)

シナリオ D: Consolidator に RAD を繋ぐ

cons = Consolidator(
    episodic=ep,
    structural=sm,
    rad_index=RadCorpusIndex(),         # これだけで biology → RAD パスが完成
)
result = cons.run_once(limit=200)
# ConceptPage が _learned/<page_type>/<concept_id>.md に自動ミラー

10. 拡張ポイント

11. 学習要点まとめ

1. 一時操作は stdlib のみで書く。取り込みのようなセットアップ系コードに 依存を入れると、ユーザの最初の体験を悪くする。本体機能 (C 層) と切り離す。
2. 読み API と書き API は単一のインデックスに統合する。 呼び出し側 (Consolidator) のシンプルさが、長期的に最大の価値を持つ。
3. tool 実装と transport は分ける。MCP / HTTP / 直接呼び出しの 3 経路で 同じロジックを再利用可能になり、テストも transport-free にできる。
4. fallback は常に「動く mock」を用意する。resolve_backend() が 最終的に MockBackend に落ちるおかげで、ネットワーク無し・API キー無しでも 全テストが通り、CI が単純になる。
5. provenance を必ず derived_from で生イベントに繋ぐ。Consolidator が Page を作るとき、その情報源 (event_ids) を持たせる。LLW-AC-01 source-anchored provenance は AI 由来コンテンツのループ汚染防御として効く。
6. 非破壊な拡張フィールドは未知のクライアントが無視するだけで壊れない。 OpenAI HTTP server に x_rad_domain を足したが、知らない既製品 UI (Ollama 純正 chat 等) は単に渡さないだけで正しく動く。
7. ミラー先への書き込み失敗は non-fatal にする。RAD への append_learning が失敗しても consolidation サイクル全体は完了させる。 result.errors に蓄積して可視化、再走で復旧可能にする。
8. MCP の smoke test は実 SDK client で回す。stdio プロトコルや JSON-RPC のシリアライズはモックでは検証しきれない。バージョン更新で 壊れたら即気付ける防御線になる。
9. SemVer はメインフェーズ用に占有しておく。横断エピックは 名前付きマイルストーン (RAD-A / RAD-B / ...) + build メタで識別。 リリース番号の衝突を防ぐ。
10. Path traversal は resolve() + relative_to() で検証する。 文字列レベルでの防御は脆い。Python のパス API を信用する。