6
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@haystacker(大塚 隼斗)
in

オントロジー × グラフ × LLM の実装と評価（基礎編）

6
Posted at

このチュートリアルでは、プレーン RAG とオントロジー併用 RAG を同じ評価セットで比較し、精度差がどこから生じるのか（特に、正解情報の被覆範囲の違い）を確認します。そのうえで、質問の種類に応じて RAG 方式を切り替えるルーターを構築します。

LLM は OpenAI 互換エンドポイント（Ollama の gemma4:e2b）を使います。

進め方

  1. LLM との接続を確認する
  2. 埋め込みとベクトル検索
  3. 最小 KB（散文だけ）
  4. プレーン RAG を 1 問通す
  5. 多段質問で限界を確認する
  6. 構造を導入（kb.py を改修）
  7. グラフを 2 段階で作る
  8. 併用 RAG で限界のある質問を扱う（何が改善し、何が改善しないか）
  9. 評価を形式化する（正解をグラフから導出）
  10. 全体比較（被覆解析を主根拠に）
  11. ルーティングと総評（router.py を追加）

LLM ・ RAG ・オントロジーの役割

本題に入る前に、この 3 つがそれぞれ何の問題を解くものなのかについて押さえておきます。

結論を先に言うと、3 者はそれぞれ異なる役割を担い、互いの弱点を補完する関係にあります。本チュートリアルでは、一方の仕組みの限界を確認し、それを別の仕組みで補うという流れで説明します。

LLM（大規模言語モデル）— 自然言語を理解し、文章を生成する

LLM は膨大なテキストで学習され、その過程で得た知識がモデルのパラメータに固定されています。

得意なのは、自然言語の理解、要約、言い換え、そして「与えられた情報を読み取り、文章としてまとめる」ことです。

一方で、構造的な弱点が 3 つあります。

  • 知識の凍結：学習後の事実や、社内固有の情報については知りません。
  • 論理の保証がない：確率的に「もっともらしい」次の語を選ぶモデルなので、推移的な依存関係の追跡や、「漏れなくすべて」を列挙する処理を、必ず正しく実行する保証はありません。
  • ハルシネーション：知らないことを、それらしく創作してしまいます。

つまり LLM は「賢い読み手・書き手」ではあっても、「正確な事実データベース」でも「論理エンジン」でもありません。

これらの弱点を補う役割を担うのが、RAG とオントロジーです。

RAG（検索拡張生成）— 外部知識を、回答の直前に注入する

RAG の基本的な考え方はシンプルです。LLM に答えさせる前に、関連しそうな文書を検索して、プロンプトに追加する。

これで「知識の凍結」と「ハルシネーション」を緩和します（モデルの記憶ではなく、目の前の文書を根拠に答えさせられる）。

仕組みは 3 段階です。

  1. 埋め込み（embedding）：各文書を意味を表すベクトルに変換しておく。
  2. 検索（retrieval）：質問もベクトル化し、意味的に近い上位 k 件を取り出す。
  3. 生成（generation）：その k 件をコンテキストに貼って LLM に答えさせる。

ここで決定的に重要なのは、RAG の検索が 「意味の近さ（類似度）」でしか動かないことです。

これは「payment に関係ありそうな文書」を拾うには強力ですが、「payment に依存しているものをすべて求める」といった、関係性に基づく質問には適していません。

類似度は幾何学的な距離であって、グラフ上の経路情報でも論理でもないからです。

——この限界を、5 章で具体的に確認します。

オントロジー＋ナレッジグラフ — 概念と関係を、機械が計算できる形で明示する

オントロジーとは、対象領域の概念（クラス）とその階層、そして概念間の関係を、明確に定義した「概念・語彙・関係の体系」です。本チュートリアルでは次を定義します。

  • クラス（class）と階層（subclass_of / is-a）：RelationalDB ⊂ Datastore ⊂ Component のようなクラス間の包含関係。これを subsumption（包摂） と呼びます。例えば「RelationalDB は Datastore の一種」を機械が知っている状態です。
  • 関係（predicate）とトリプル（triple）：(order-service, depends_on, payment-service) のような主語–述語–目的語の事実。散文で「呼び出す」と書く代わりに、計算できる辺として持ちます。

この「クラス階層＝スキーマ」の層を T-box、「個々の事実＝インスタンス」の層を A-box と呼び分けます（本チュートリアルでは CLASS_HIERARCHY が T-box、ENTITIES と RELATIONS が A-box）。

これを明示することで、確率的な推論やベクトル類似度に依存せず、明示された構造に基づいて決定的に計算できます。

  • 型による網羅：「Datastore に属するものをすべて」→ 定義されたクラス階層をたどって集合を厳密に返す。
  • 推移的な関係の追跡：「payment に依存する全て」→ グラフの到達可能性として厳密に計算。
  • 結合：「RelationalDB に書くサービス」→ 型フィルタ × 関係で厳密に求める。
  • 説明可能性：答えの根拠を「どの辺をたどったか」という経路情報で提示できる。

一方で、人手または自動抽出によってグラフを整備する必要があり、型体系が頻繁に変化する領域では維持コストが高くなります。

3 者の分業表

image.png

RAG は LLM が持たない「固有情報」を、オントロジーは RAG の「関係・網羅」情報の不足を補います。

逆に、単一事実の質問のように不足が無いところにグラフを足すと、後述のとおり過剰になります（10.2 節の Q4）。

能力・性質 LLM 単体 + RAG（ベクトル） + オントロジー + グラフ
自然言語の理解・生成 ◎ ◎ ◎
固有／最新の事実に基づく ✗（知識の凍結） ○（文書を注入） ◎（構造化された事実を保持）
意味的な曖昧検索 — ◎（類似度） △（別名正規化で補助）
多段の関係推論（追跡） △（不確実） ✗（類似度に経路情報なし） ◎（推移閉包で厳密）
網羅・集約（「全部」） △ ✗（top-k による取得件数の制限） ◎（subsumption で厳密）
型×関係の結合 △ ✗（散文から復元不可） ◎（型フィルタ×辺）
根拠の説明可能性 ✗ △（出典文書） ◎（たどった経路情報）

併用 RAG のデータフロー（このチュートリアルが作るもの）

最終的に組む「オントロジー併用 RAG」の一巡は次のとおりです。各矢印は、処理間で受け渡される情報を表します。

ここでは、3 者の役割が明確に分担されていることが分かります。

LLM は、自然言語から構造化された ID・クラスへの変換と、最終的な回答生成を担当し、知識間の関係の探索はグラフが決定的に行い、曖昧な検索候補の絞り込みは、ベクトル検索が担います。

LLM に「グローバルな構造を推論させない」——必要な部分グラフを先に決定的に切り出して読ませるだけにする——のが、精度と説明可能性を高めるうえで重要なポイントです。

0. セットアップ

本チュートリアルでは、パスやモデル名は環境変数で扱い、直書きしないようにします。

$ python3 -m venv .venv
$ source .venv/bin/activate
$ pip install "openai>=1.40" "numpy>=1.26"

$ ollama pull gemma4:e2b
$ ollama pull nomic-embed-text

$ export OPENAI_BASE_URL="http://localhost:11434/v1"
$ export OPENAI_API_KEY="ollama"     # Ollama はキー不要。SDK の形式上ダミーを渡す
$ export CHAT_MODEL="gemma4:e2b"
$ export EMBED_MODEL="nomic-embed-text"

チャットモデルは最後まで gemma4:e2b で一貫させます。以降のコードの既定値もすべて gemma4:e2b です。
base_url を変えれば vLLM や LM Studio でも同じコードが動きます。

用語
Ollama はローカル PC で LLM を動かすランタイムです。
OpenAI 互換エンドポイントとは、OpenAI API と同様の形式（/v1/chat/completions など）でリクエストを受けつけるインターフェースです。OpenAI 向けに書いたコードを base_url の差し替えだけでローカルの Gemma に向けられる——そのため以降も OpenAI の SDK をそのまま使います。

1. まず LLM に疎通する

一度に RAG を組まず、OpenAI 互換で Gemma にリクエストを送信し、正常に応答できることだけを確認します。

llm.py はこの時点では chat の未実装します。

  • llm.py（第1版）

    from __future__ import annotations
import os
from openai import OpenAI

BASE_URL = os.environ.get("OPENAI_BASE_URL", "http://localhost:11434/v1")
API_KEY = os.environ.get("OPENAI_API_KEY", "ollama")
CHAT_MODEL = os.environ.get("CHAT_MODEL", "gemma4:e2b")

_client = OpenAI(base_url=BASE_URL, api_key=API_KEY)

def chat(system: str, user: str, temperature: float = 0.0) -> str:
    resp = _client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[{"role": "system", "content": system},
                  {"role": "user", "content": user}],
        temperature=temperature,
    )
    return resp.choices[0].message.content or ""

実際に LLM リクエストを投げてみましょう。

$ python -c "from llm import chat; print(chat('簡潔に答えて', 'RAG を一文で説明して'))"

RAG（Retrieval-Augmented Generation）とは、大規模言語モデル（LLM）が回答を生成する際に、外部の知識ベースから関連性の高い情報を検索（Retrieval）し、その情報を基に回答を生成（Generation）させることで、より正確で根拠のある応答を可能にする技術です。

一文の回答が返れば、接続確認は完了です。今回はtemperature=0.0 を設定しており、出力のランダム性を抑えています。

2. 埋め込みとベクトル検索をフルスクラッチで

次にベクトル検索の基盤を実装します。まず llm.py に embed を足します。

  • llm.py（第 2 版）

    第 1 版を基にchat はそのまま残しつつ、EMBED_MODEL と embed() を追加します。以下が第 2 版の全文です。

    from __future__ import annotations
import os
from openai import OpenAI

BASE_URL = os.environ.get("OPENAI_BASE_URL", "http://localhost:11434/v1")
API_KEY = os.environ.get("OPENAI_API_KEY", "ollama")
CHAT_MODEL = os.environ.get("CHAT_MODEL", "gemma4:e2b")
EMBED_MODEL = os.environ.get("EMBED_MODEL", "nomic-embed-text")

_client = OpenAI(base_url=BASE_URL, api_key=API_KEY)

def embed(texts: list[str]) -> list[list[float]]:
    resp = _client.embeddings.create(model=EMBED_MODEL, input=texts)
    return [d.embedding for d in resp.data]

def chat(system: str, user: str, temperature: float = 0.0) -> str:
    resp = _client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[{"role": "system", "content": system},
                  {"role": "user", "content": user}],
        temperature=temperature,
    )
    return resp.choices[0].message.content or ""

次に、コサイン類似度による最小構成のベクトルストアを実装します。ここではライブラリに頼らず numpy で書きます。

  • vector_store.py

    from __future__ import annotations
import numpy as np
from llm import embed

class VectorStore:
    def __init__(self, docs: list[dict]):
        self.docs = docs
        vecs = embed([d["text"] for d in docs])
        self.mat = self._normalize(np.array(vecs, dtype=np.float32))

    @staticmethod
    def _normalize(m: np.ndarray) -> np.ndarray:
        n = np.linalg.norm(m, axis=1, keepdims=True)
        n[n == 0] = 1.0
        return m / n

    def search(self, query: str, k: int = 4) -> list[dict]:
        q = self._normalize(np.array(embed([query]), dtype=np.float32))[0]
        sims = self.mat @ q
        idx = np.argsort(-sims)[:k]
        return [{**self.docs[i], "score": float(sims[i])} for i in idx]

ダミーの文章 3 件を使って、ベクトルストア検索の感触を掴みます。

$ python - << 'PY'
from vector_store import VectorStore
vs = VectorStore([
    {"id": "a", "text": "決済処理と返金を担うサービス"},
    {"id": "b", "text": "商品検索のインデックスを提供する"},
    {"id": "c", "text": "ユーザーのプロフィールを管理する"},
])
for h in vs.search("お金の返金はどこ？", k=2):
    print(h["id"], round(h["score"], 3), h["text"])
PY

a 0.887 決済処理と返金を担うサービス
b 0.591 商品検索のインデックスを提供する

a（決済）が最上位に来れば、埋め込みによる意味検索が機能していることを確認できます。

embedとは何か（なぜ「意味」で検索できるのか）
embedは各文を数百次元のベクトルに変換します。訓練の結果、意味の近い文どうしはベクトルも近くなるように配置されます。
つまり各文章が意味空間上のベクトルとして表現され、質問も同じ空間の一点に落とすと、点と点の近さ（コサイン類似度）によって意味の近さを測れます。これが、ベクトル検索における基本的な考え方です。
キーワードの一致は不要で、「返金」と「payment（決済）」のように語が違っても近くに来ます。

ここに、後で問題となる本質的な限界があります。この空間が測れるのは 「意味的に類似しているか」だけで、「A が B に依存しているか」「A の一種をすべて挙げられるか」といった関係や構造が、明示的な辺として保持されているわけではありません。
近傍検索は「近いものを数件」返す操作であって、辺をたどる操作でも、集合を漏れなく列挙する操作でもないのです。
この 「意味的に近い情報を返す」という性質が、関係性を問う質問ではどのような限界につながるのかを、5 章で確認します。

コサイン類似度とは。 2 つのベクトルの“向き”がどれだけ揃うかを測る指標です。各ベクトルを長さ 1 に正規化して内積を取ると値は −1〜1 に収まり、1＝同じ向き（意味が近い）／0＝無関係（直交）。文書の長短（ベクトルの長さ）ではなく意味の方向だけを見ます。コードの _normalize（長さ 1 化）→ mat @ q（内積）がこれにあたります。

3. 最小のナレッジベース（散文のみ）

マイクロサービスの依存ドメインを題材に最小のナレッジベースを構築します。この段階では型もトリプルも作りません。

依存関係も含めてすべて散文で記述します（＝通常の RAG が扱う素朴なコーパス）。

まず全体像です。以降の質問（5 章の障害波及など）は、すべてこの依存関係の上での探索になります。ただし、この図は読者のためのもので、機械は把握していません——RAG が見るのは、下の kb.py が生成する散文だけです。「人には構造が見えるのに、RAG には辿れない」——この落差が次のステップの主題です。

image.png

  • kb.py（第1版: 散文ドキュメントのみ）

    from __future__ import annotations
from dataclasses import dataclass

@dataclass
class Entity:
    id: str
    label: str
    desc: str   # 依存などもこの段階では散文で書く

ENTITIES: list[Entity] = [
    Entity("api-gateway", "API Gateway",
        "全ての外部リクエストを最初に受ける入口。稼働時に Auth Service・Order Service・"
        "User Service・Search Service を呼び出す。担当は Platform Team。"),
    Entity("auth-service", "Auth Service",
        "トークン発行と検証を行う認証の中核。データを Session Cache に保存する。担当は Platform Team。"),
    Entity("user-service", "User Service",
        "ユーザープロフィールと権限を管理する。データを Users DB に保存する。担当は Growth Team。"),
    Entity("order-service", "Order Service",
        "注文の受付・確定・状態遷移を管理する。稼働時に Payment Service・Inventory Service・"
        "User Service を呼び出す。データを Orders DB に保存し、Event Bus へイベントを送る。担当は Commerce Team。"),
    Entity("payment-service", "Payment Service",
        "決済処理と返金を担う。稼働時に Auth Service を呼び出す。データを Payments DB に保存し、"
        "Event Bus へイベントを送る。担当は Commerce Team。"),
    Entity("inventory-service", "Inventory Service",
        "在庫の引き当てと補充を管理する。担当は Commerce Team。"),
    Entity("search-service", "Search Service",
        "商品検索インデックスを提供する。担当は Growth Team。"),
    Entity("users-db", "Users DB", "ユーザー情報を格納する PostgreSQL クラスタ。担当は Growth Team。"),
    Entity("orders-db", "Orders DB", "注文履歴を格納する PostgreSQL クラスタ。担当は Commerce Team。"),
    Entity("payments-db", "Payments DB", "決済トランザクションを格納する PostgreSQL クラスタ。担当は Commerce Team。"),
    Entity("session-cache", "Session Cache", "セッションとトークンを保持する Redis。担当は Platform Team。"),
    Entity("event-bus", "Event Bus", "ドメインイベントを配送する Kafka トピック群。担当は Platform Team。"),
]

def generate_documents() -> list[dict]:
    return [{"id": e.id, "text": f"{e.label}（{e.id}）: {e.desc}"} for e in ENTITIES]

if __name__ == "__main__":
    for d in generate_documents():
        print(d["id"], "->", d["text"])

先ほど構築したナレッジベースを表示してみます。

$ python kb.py

api-gateway -> API Gateway（api-gateway）: 全ての外部リクエストを最初に受ける入口。稼働時に Auth Service・Order Service・User Service・Search Service を呼び出す。担当は Platform Team。
auth-service -> Auth Service（auth-service）: トークン発行と検証を行う認証の中核。データを Session Cache に保存する。担当は Platform Team。
user-service -> User Service（user-service）: ユーザープロフィールと権限を管理する。データを Users DB に保存する。担当は Growth Team。
order-service -> Order Service（order-service）: 注文の受付・確定・状態遷移を管理する。稼働時に Payment Service・Inventory Service・User Service を呼び出す。データを Orders DB に保存し、Event Bus へイベントを送る。担当は Commerce Team。
payment-service -> Payment Service（payment-service）: 決済処理と返金を担う。稼働時に Auth Service を呼び出す。データを Payments DB に保存し、Event Bus へイベントを送る。担当は Commerce Team。
inventory-service -> Inventory Service（inventory-service）: 在庫の引き当てと補充を管理する。担当は Commerce Team。
search-service -> Search Service（search-service）: 商品検索インデックスを提供する。担当は Growth Team。
users-db -> Users DB（users-db）: ユーザー情報を格納する PostgreSQL クラスタ。担当は Growth Team。
orders-db -> Orders DB（orders-db）: 注文履歴を格納する PostgreSQL クラスタ。担当は Commerce Team。
payments-db -> Payments DB（payments-db）: 決済トランザクションを格納する PostgreSQL クラスタ。担当は Commerce Team。
session-cache -> Session Cache（session-cache）: セッションとトークンを保持する Redis。担当は Platform Team。
event-bus -> Event Bus（event-bus）: ドメインイベントを配送する Kafka トピック群。担当は Platform Team。

依存関係が「Payment Service は稼働時に Auth Service を呼び出す」のように通常の文章の中に埋め込まれている点に注目してください。

ここが後の弱点になります。

4. プレーン RAG を 1 問実行

まずは単純な構成で、top-k 件を取得し、その結果をコンテキストとして回答させます。回答は採点しやすいよう、ID の JSON 配列で出力させます。

  • rag_plain.py（第1版）

    from __future__ import annotations
import json
from vector_store import VectorStore
from llm import chat
from kb import generate_documents

SYSTEM = (
    "あなたはシステム構成に関する質問に答えるアシスタントです。"
    "与えられたコンテキストだけを根拠に答えてください。"
    "最終的な該当コンポーネントの ID を、必ず JSON 配列だけで最後に出力してください。"
    "例: [\"order-service\"]。該当なしは [] とします。"
)

def extract_json_array(text: str) -> list[str]:
    s, e = text.find("["), text.rfind("]")
    if s == -1 or e == -1 or e < s:
        return []
    try:
        return [str(x).strip() for x in json.loads(text[s:e + 1]) if str(x).strip()]
    except json.JSONDecodeError:
        return []

class PlainRAG:
    def __init__(self, k: int = 4):
        self.k = k
        self.store = VectorStore(generate_documents())

    def answer(self, question: str) -> tuple[list[str], str]:
        hits = self.store.search(question, k=self.k)
        context = "\n\n".join(f"[{h['id']}] {h['text']}" for h in hits)
        prompt = f"# コンテキスト\n{context}\n\n# 質問\n{question}\n\nJSON 配列で答えよ。"
        out = chat(SYSTEM, prompt)
        return extract_json_array(out), out

top-k とは：
類似度が高い順に上位 k 件だけを取ること（ここでは k=4）。RAG はこの k 件しかコンテキストに載せないので、k を超える数の正解は原理的に同時に入りません。この取得件数の制約が 10 章の Q2（データストア 5 件）で問題になります。

まずは単一事実となる質問をしてみます。

$ python - << 'PY'
from rag_plain import PlainRAG
rag = PlainRAG(k=4)
ids, raw = rag.answer("auth-service の役割は？ 対象の ID を答えよ。")
print("pred:", ids)
PY

pred: ['auth-service']

1 つのチャンクだけで回答に必要な情報が得られる質問では、素朴な RAG で十分です。

この点は後の全体比較でも（プレーン RAG の得意分野として）重要になります。次に、あえて限界が現れる質問を扱います。

5. 多段の質問で限界を確認する（いちばん大事なステップ）

同じ RAG に「障害の波及」について質問してみます。「payment-service が停止したら、直接的・間接的に影響を受けるサービスをすべて挙げよ。」という質問の場合、正解はpayment-serviceに（推移的に）依存するものなので、order-service（直接）およびapi-gateway（order 経由）となるはずです。

$ python - << 'PY'
from rag_plain import PlainRAG
rag = PlainRAG(k=4)
ids, raw = rag.answer(
    "payment-service が停止したら、直接的・間接的に影響を受けるサービスをすべて挙げよ。")
print("pred:", ids)   # 例: ['order-service'] のように api-gateway を取りこぼしやすい
PY

pred: ['order-service', 'event-bus']

多くの場合、api-gatewayを取得することができません（実行環境により結果は変動します）。理由は 2 つあり、いずれも仕組みに起因します。

  • 経路情報を持たない：api-gateway → order-service → payment-service の連鎖を追跡するには関係文書を同時に引く必要がある。
    しかしベクトル検索は「payment に意味的に近い数件」を返すだけで、多段の関係追跡は、近傍検索だけでは扱えません。
  • 被覆の限界：後で確かめますが、top-k は k 件しか返せないため、「すべて列挙せよ」という質問では正解の取りこぼしが構造的に起きます。

なぜ「意味的に類似している」では「関係の追跡ができない」のか（RAG の構造的限界）
この失敗は、埋め込みモデルの性能不足ではありません。問題の種類が、ベクトル検索の守備範囲の外なのです。
「payment が停止したら影響を受けるもの」を求めるには、本来こう推論します：

  • 誰が payment を呼ぶ？ → order-service。では誰が order を呼ぶ？ → api-gateway。…という辺の連鎖（推移閉包）

これはグラフ上の到達可能性の計算です。

ところがベクトル空間には「呼ぶ／依存する」という辺そのものが存在しません。あるのは「意味が近い／遠い」だけ。
api-gateway の説明文は「入口・ルーティング」が主で、「payment の障害」とは語彙上の類似性が低いため、top-k の上位に入ってこない。しかも top-k は k 件で頭打ちなので、「すべて列挙せよ」という質問に対しては取得件数の上限もあります。
LLM を賢くしても、渡すコンテキストにapi-gatewayが入っていなければ答えようがない（ハルシネーションで当てるしかない）。

要点：
この構成のベクトル検索の検索精度を向上させるだけでは、この問題を直接解決できません。足りないのは「関係を明示的にたどれる知識表現」＝グラフです。

image.png

つまり、この構成のベクトル検索だけを高精度化しても、関係を明示的にたどる仕組みにはなりません。必要なのは、知識の表現形式に関係構造を導入することです——これが構造導入の動機です。

6. 構造を導入する（kb.py を改修）

通常の文章として記述されていた「型」と「関係」を明示的なオントロジーとトリプルとして取り出します。

そして、generate_documentsをトリプルから散文を生成する形に作り替えます。

同じ事実を、散文ではベクトル検索用の情報として、構造情報ではトリプルとして保持します。これにより、知識そのものではなく、表現形式の違いを比較できるようにします。

オントロジーの語彙

  • クラス（class）：概念のカテゴリ。例：Service, Datastore, RelationalDB。
  • インスタンス（instance）：クラスに属する具体物。例：payment-service は CoreService のインスタンス。
  • subclass_of（is-a / 包摂＝subsumption）：クラス間の階層。RelationalDB ⊂ Datastore ⊂ Component。
    これにより「RelationalDB を 1 つ挙げれば、それは自動的に Datastore でもある」という型推論が可能になります。
  • 述語（predicate）とトリプル（triple）：事実を (主語, 述語, 目的語) で表す最小単位。
    例：(order-service, depends_on, payment-service)。散文の「呼び出す」を、計算可能な辺に置き換えたものです。

スキーマの層（クラス階層＝CLASS_HIERARCHY）を T-box、事実の層（ENTITIES＋RELATIONS）を A-box と呼びます。

散文「Order Service は Payment Service を呼び出す」は、人間には読めますが、機械にとっては構造化されていないテキストで、「では Payment に依存するものを全部たどれ」に答える手掛かりになりません。
同じ事実を (order-service, depends_on, payment-service) という辺にしておけば、グラフ探索で決定的にたどることができるようになります。この「LLM の確率にもベクトルの類似度にも頼らず、計算だけで答えが決まる」性質が、以降のステップで recall を保証するための基盤になります。

T-box / A-box をもう一段かみ砕くと：T-box＝型の定義（スキーマ）、A-box＝個々の事実（データ）となります。DB の「テーブル定義」と「レコード」の関係に近いです。例：payments-db は RelationalDB のインスタンス。RelationalDB ⊂ Datastore なので payments-db は自動的に Datastore でもある——これを一件ずつ書かずに階層から導けることが、型推論（subsumption） の重要な点です。

このステップで作るドメインの全体像です。第1版（3 章）に notification-service が加わっただけで、依存・永続化・イベントの骨格は同じ。以降の質問（障害波及・型網羅・結合）は、すべてこの上での探索になります。

image.png

そのうえで、各コンポーネントに型を与え、型どうしの階層（クラス階層＝subsumption）を新設します。これが、型に基づく網羅的な検索を可能にする基盤となります。

image.png

Entity に type（型）を追加し、notification-service と 3 チーム（platform / commerce / growth）を追加。CLASS_HIERARCHY / CLASS_SYNONYMS / RELATIONS / owner_of() を新設し、generate_documents を「トリプルから散文を生成」する形に差し替えます。本質的な違いは、同じ題材に型と関係トリプルを持たせたことです（構成要素は第1版とほぼ同じ）。

以下が第 2 版の全文です（kb.py をこの内容に丸ごと置き換えてください）。

  • kb.py（第2版: オントロジー + トリプルを追加し、散文はそこから生成）

    # kb.py  (第2版: オントロジー + トリプルを追加し、散文はそこから生成)
from __future__ import annotations
from dataclasses import dataclass

# --- オントロジー: クラス階層 (subclass_of) ---
CLASS_HIERARCHY: dict[str, str | None] = {
    "Component": None,
    "Service": "Component",
    "EdgeService": "Service",
    "CoreService": "Service",
    "SupportingService": "Service",
    "Datastore": "Component",
    "RelationalDB": "Datastore",
    "CacheStore": "Datastore",
    "MessageQueue": "Datastore",
    "Team": None,
}

CLASS_SYNONYMS: dict[str, list[str]] = {
    "Datastore": ["データストア", "datastore", "永続化層", "storage"],
    "RelationalDB": ["リレーショナルDB", "relational database", "RDB", "SQL DB"],
    "CacheStore": ["キャッシュ", "cache"],
    "MessageQueue": ["メッセージキュー", "message queue", "MQ", "イベントバス"],
    "Service": ["サービス", "service"],
    "Team": ["チーム", "team"],
}

@dataclass
class Entity:
    id: str
    type: str          # 所属するリーフクラス
    label: str
    desc: str          # 役割の説明(単一事実)

ENTITIES: list[Entity] = [
    Entity("api-gateway", "EdgeService", "API Gateway",
           "全ての外部リクエストを最初に受ける入口。認証委譲とルーティングを担う。"),
    Entity("auth-service", "CoreService", "Auth Service",
           "トークン発行と検証を行う認証の中核。停止すると全体のログインが不能になる。"),
    Entity("user-service", "CoreService", "User Service",
           "ユーザープロフィールと権限を管理する。"),
    Entity("order-service", "CoreService", "Order Service",
           "注文の受付・確定・状態遷移を管理する。"),
    Entity("payment-service", "CoreService", "Payment Service",
           "決済処理と返金を担う。外部決済ゲートウェイと連携する。"),
    Entity("inventory-service", "CoreService", "Inventory Service",
           "在庫の引き当てと補充を管理する。"),
    Entity("notification-service", "SupportingService", "Notification Service",
           "メールとプッシュ通知を送る。イベント駆動で動作する。"),
    Entity("search-service", "SupportingService", "Search Service",
           "商品検索インデックスを提供する。"),
    Entity("users-db", "RelationalDB", "Users DB",
           "ユーザー情報を格納する PostgreSQL クラスタ。"),
    Entity("orders-db", "RelationalDB", "Orders DB",
           "注文履歴を格納する PostgreSQL クラスタ。"),
    Entity("payments-db", "RelationalDB", "Payments DB",
           "決済トランザクションを格納する PostgreSQL クラスタ。"),
    Entity("session-cache", "CacheStore", "Session Cache",
           "セッションとトークンを保持する Redis。"),
    Entity("event-bus", "MessageQueue", "Event Bus",
           "ドメインイベントを配送する Kafka トピック群。"),
    Entity("platform-team", "Team", "Platform Team", "基盤・認証・配信を担当。"),
    Entity("commerce-team", "Team", "Commerce Team", "取引・決済・在庫を担当。"),
    Entity("growth-team", "Team", "Growth Team", "ユーザー・通知・検索を担当。"),
]

# --- 関係トリプル (subject, predicate, object) ---
RELATIONS: list[tuple[str, str, str]] = [
    ("api-gateway", "depends_on", "auth-service"),
    ("api-gateway", "depends_on", "order-service"),
    ("api-gateway", "depends_on", "user-service"),
    ("api-gateway", "depends_on", "search-service"),
    ("order-service", "depends_on", "payment-service"),
    ("order-service", "depends_on", "inventory-service"),
    ("order-service", "depends_on", "user-service"),
    ("payment-service", "depends_on", "auth-service"),
    ("user-service", "persists_to", "users-db"),
    ("order-service", "persists_to", "orders-db"),
    ("payment-service", "persists_to", "payments-db"),
    ("auth-service", "persists_to", "session-cache"),
    ("order-service", "publishes_to", "event-bus"),
    ("payment-service", "publishes_to", "event-bus"),
    ("notification-service", "subscribes_to", "event-bus"),
    ("api-gateway", "owned_by", "platform-team"),
    ("auth-service", "owned_by", "platform-team"),
    ("event-bus", "owned_by", "platform-team"),
    ("session-cache", "owned_by", "platform-team"),
    ("order-service", "owned_by", "commerce-team"),
    ("payment-service", "owned_by", "commerce-team"),
    ("inventory-service", "owned_by", "commerce-team"),
    ("orders-db", "owned_by", "commerce-team"),
    ("payments-db", "owned_by", "commerce-team"),
    ("user-service", "owned_by", "growth-team"),
    ("notification-service", "owned_by", "growth-team"),
    ("search-service", "owned_by", "growth-team"),
    ("users-db", "owned_by", "growth-team"),
]

TRANSITIVE_PREDICATES = {"depends_on"}
ENTITY_BY_ID = {e.id: e for e in ENTITIES}

def owner_of(entity_id: str) -> str | None:
    """entity_id の owned_by 先を返す。無ければ None (堅牢化のため例外にしない)。"""
    for s, p, o in RELATIONS:
        if s == entity_id and p == "owned_by":
            return o
    return None

def generate_documents() -> list[dict]:
    """トリプルから散文を生成(プレーン RAG 用)。同じ事実を散文側にも持たせる。"""
    docs = []
    for e in ENTITIES:
        lines = [f"{e.label}（{e.id}）: {e.desc}"]
        for s, p, o in RELATIONS:
            if s != e.id:
                continue
            ol = ENTITY_BY_ID[o].label
            if p == "depends_on":
                lines.append(f"{e.label} は稼働時に {ol} を呼び出す。")
            elif p == "persists_to":
                lines.append(f"{e.label} はデータを {ol} に保存する。")
            elif p == "publishes_to":
                lines.append(f"{e.label} はイベントを {ol} へ送る。")
            elif p == "subscribes_to":
                lines.append(f"{e.label} は {ol} を購読して動く。")
            elif p == "owned_by":
                lines.append(f"{e.label} の担当は {ol} である。")
        docs.append({"id": e.id, "text": " ".join(lines)})
    return docs

if __name__ == "__main__":
    for d in generate_documents():
        print(d["id"], "->", d["text"])

生成した散文を表示してみましょう。散文の見た目は第 1 版とほぼ同じでも、裏に型とトリプルを持っているのが大きな違いです。

$ python kb.py

api-gateway -> API Gateway（api-gateway）: 全ての外部リクエストを最初に受ける入口。認証委譲とルーティングを担う。 API Gateway は稼働時に Auth Service を呼び出す。 API Gateway は稼働時に Order Service を呼び出す。 API Gateway は稼働時に User Service を呼び出す。 API Gateway は稼働時に Search Service を呼び出す。 API Gateway の担当は Platform Team である。
auth-service -> Auth Service（auth-service）: トークン発行と検証を行う認証の中核。停止すると全体のログインが不能になる。 Auth Service はデータを Session Cache に保存する。 Auth Service の担当は Platform Team である。
user-service -> User Service（user-service）: ユーザープロフィールと権限を管理する。 User Service はデータを Users DB に保存する。 User Service の担当は Growth Team である。
order-service -> Order Service（order-service）: 注文の受付・確定・状態遷移を管理する。 Order Service は稼働時に Payment Service を呼び出す。 Order Service は稼働時に Inventory Service を呼び出す。 Order Service は稼働時に User Service を呼び出す。 Order Service はデータを Orders DB に保存する。 Order Service はイベントを Event Bus へ送る。 Order Service の担当は Commerce Team である。
payment-service -> Payment Service（payment-service）: 決済処理と返金を担う。外部決済ゲートウェイと連携する。 Payment Service は稼働時に Auth Service を呼び出す。 Payment Service はデータを Payments DB に保存する。 Payment Service はイベントを Event Bus へ送る。 Payment Service の担当は Commerce Team である。
inventory-service -> Inventory Service（inventory-service）: 在庫の引き当てと補充を管理する。 Inventory Service の担当は Commerce Team である。
notification-service -> Notification Service（notification-service）: メールとプッシュ通知を送る。イベント駆動で動作する。 Notification Service は Event Bus を購読して動く。 Notification Service の担当は Growth Team である。
search-service -> Search Service（search-service）: 商品検索インデックスを提供する。 Search Service の担当は Growth Team である。
users-db -> Users DB（users-db）: ユーザー情報を格納する PostgreSQL クラスタ。 Users DB の担当は Growth Team である。
orders-db -> Orders DB（orders-db）: 注文履歴を格納する PostgreSQL クラスタ。 Orders DB の担当は Commerce Team である。
payments-db -> Payments DB（payments-db）: 決済トランザクションを格納する PostgreSQL クラスタ。 Payments DB の担当は Commerce Team である。
session-cache -> Session Cache（session-cache）: セッションとトークンを保持する Redis。 Session Cache の担当は Platform Team である。
event-bus -> Event Bus（event-bus）: ドメインイベントを配送する Kafka トピック群。 Event Bus の担当は Platform Team である。
platform-team -> Platform Team（platform-team）: 基盤・認証・配信を担当。
commerce-team -> Commerce Team（commerce-team）: 取引・決済・在庫を担当。
growth-team -> Growth Team（growth-team）: ユーザー・通知・検索を担当。

7. グラフを 2 段階で作る

一度にすべてを実装するのではなく、まず型（subsumption）だけで「網羅的な検索」を可能にし、そのあとで近傍＋推移閉包を足して「多段」を解けるようにします。

7.1 まずは subsumption だけ

  • graph_store.py（第1版: クラス階層と型網羅のみ）

    from __future__ import annotations
from collections import defaultdict
from kb import CLASS_HIERARCHY, ENTITIES

class OntologyGraph:
    def __init__(self):
        self.parent = dict(CLASS_HIERARCHY)
        self.children: dict[str, list[str]] = defaultdict(list)
        for c, p in self.parent.items():
            if p is not None:
                self.children[p].append(c)
        self.instances_of_leaf: dict[str, list[str]] = defaultdict(list)
        for e in ENTITIES:
            self.instances_of_leaf[e.type].append(e.id)

    def descendant_classes(self, cls: str) -> set[str]:
        seen, stack = set(), [cls]
        while stack:
            c = stack.pop()
            if c in seen:
                continue
            seen.add(c)
            stack.extend(self.children.get(c, []))
        return seen

    def instances_of(self, cls: str) -> list[str]:
        """クラス cls とその下位クラスに属する全インスタンス(型推論)。"""
        result = []
        for c in self.descendant_classes(cls):
            result.extend(self.instances_of_leaf.get(c, []))
        return sorted(set(result))

Datastore クラス（とその下位クラス）に属するインスタンスを表示してみます。

$ python -c "from graph_store import OntologyGraph as G; g=G(); print(g.instances_of('Datastore'))"

['event-bus', 'orders-db', 'payments-db', 'session-cache', 'users-db']

top-k では k 件しか返せなかった網羅が、subsumption なら全件取得できます。これが型情報を持つことによる 1 つ目の効果です。

なぜ subsumption だと漏れないのか
これは検索ではなく集合の計算だからです。
instances_of("Datastore") は「Datastore の下位クラス（RelationalDB ・ CacheStore ・ MessageQueue）を全部追跡し、それぞれのインスタンスを集める」という決定的な列挙を行います。
類似度で上位 k 件を近似的に拾うベクトル検索と違い、件数の上限も、意味的な遠近も関係ありません。
「該当するものは、定義上すべて含まれる」—— RAG では扱いにくかった問題を、型情報によって構造的に解決できる理由がここにあります。

image.png

7.2 近傍＋推移閉包＋直列化を足す

「多段の依存」を解くための探索と、部分グラフを型注釈つきトリプルにする直列化を足します。

推移閉包/ホップ/直列化

用語解説：

  • 推移閉包（transitive closure）：「A→B、B→C なら A→C」という間接関係を、間接的な関係を含め、到達可能な関係をすべて含む集合。api-gateway→order-service→payment-service なら、api-gatewayはpayment-serviceに間接的に依存していると言えます。これによって、対象への影響範囲を求めることができます。
  • ホップ（hop）：辺を 1 本たどること（2 ホップ＝辺 2 本ぶん）。
  • 直列化（serialize）：グラフ（ノードと辺）を、LLM が処理できるテキスト形式に変換すること（ここでは「型／エンティティ／関係トリプル」の 3 節に整形）。

3 つの新メソッドが、それぞれどの限界を補うのか：

  • reverse_transitive：推移閉包。「payment に依存するもの」を辺の連鎖で到達可能なノードをすべて探索する＝5 章でベクトルが解けなかった多段の追跡そのもの。グラフ上の到達可能性計算です。
  • neighborhood：質問の焦点（seed）の周辺だけを切り出す。グラフ全体を LLM に渡すのは無駄でノイズ源なので、関係する部分グラフに絞ります（depends_on は推移的なので閉包まで、他の辺は指定ホップまで）。
  • serialize_subgraph：切り出した部分グラフを、型注釈つきの明示的トリプルという LLM が読める文字列にする。ここで T-box（クラス階層）と A-box（トリプル）が 1 つのコンテキストに合流します。

つまりこのステップで、「たどる（reverse_transitive）→ 対象範囲を絞り込む（neighborhood）→ LLM が処理可能な形式に変換する（serialize）」 という、
グラフ側の決定的パイプラインが揃います。LLM はこの後、出来上がった部分グラフを読むことで、多段の質問に必要な構造情報を利用できるようになります。

descendant_classes / instances_of は第 1 版のまま残しつつ、import を拡張し、__init__ に隣接リスト（out / inc）を追加、メソッド superclasses / link / reverse_transitive / neighborhood / serialize_subgraph / retrieve_context を新設します。

以下がgraph_store.py（第 2 版）の全文です。中でも核心はreverse_transitive（推移閉包）・neighborhood（近傍抽出）・retrieve_context（型網羅＋近傍の直列化）の 3 つです。

  • graph_store.py（第2版: 近傍探索・推移閉包・リンキング・直列化を追加）

    from __future__ import annotations
from collections import defaultdict, deque
from kb import (CLASS_HIERARCHY, CLASS_SYNONYMS, ENTITIES, RELATIONS,
                TRANSITIVE_PREDICATES, ENTITY_BY_ID)

class OntologyGraph:
    def __init__(self):
        self.parent = dict(CLASS_HIERARCHY)
        self.children: dict[str, list[str]] = defaultdict(list)
        for c, p in self.parent.items():
            if p is not None:
                self.children[p].append(c)
        self.out: dict[str, list[tuple[str, str]]] = defaultdict(list)  # s -> [(p,o)]
        self.inc: dict[str, list[tuple[str, str]]] = defaultdict(list)  # o -> [(p,s)]
        for s, p, o in RELATIONS:
            self.out[s].append((p, o))
            self.inc[o].append((p, s))
        self.instances_of_leaf: dict[str, list[str]] = defaultdict(list)
        for e in ENTITIES:
            self.instances_of_leaf[e.type].append(e.id)

    # --- subsumption ---
    def descendant_classes(self, cls: str) -> set[str]:
        seen, stack = set(), [cls]
        while stack:
            c = stack.pop()
            if c in seen:
                continue
            seen.add(c)
            stack.extend(self.children.get(c, []))
        return seen

    def instances_of(self, cls: str) -> list[str]:
        result = []
        for c in self.descendant_classes(cls):
            result.extend(self.instances_of_leaf.get(c, []))
        return sorted(set(result))

    def superclasses(self, cls: str) -> list[str]:
        chain, c = [], cls
        while c is not None:
            chain.append(c)
            c = self.parent.get(c)
        return chain

    # --- エンティティ/クラス リンキング(文字列マッチ) ---
    def link(self, mentions: list[str]) -> tuple[set[str], set[str]]:
        ent_ids, cls_ids = set(), set()
        norm = {m.strip().lower() for m in mentions if m.strip()}
        for m in norm:
            for e in ENTITIES:
                if m == e.id.lower() or m == e.label.lower() or m in e.label.lower():
                    ent_ids.add(e.id)
            for cls in self.parent:
                if m == cls.lower():
                    cls_ids.add(cls)
                for syn in CLASS_SYNONYMS.get(cls, []):
                    if m == syn.lower():
                        cls_ids.add(cls)
        return ent_ids, cls_ids

    # --- 部分グラフ抽出 ---
    def reverse_transitive(self, target: str, pred: str) -> set[str]:
        """target に(推移的に) pred で依存する主体の集合。障害波及の計算に使う。"""
        result, stack = set(), [target]
        while stack:
            cur = stack.pop()
            for p, s in self.inc.get(cur, []):
                if p == pred and s not in result:
                    result.add(s)
                    stack.append(s)
        return result

    def neighborhood(self, seeds: set[str], hops: int = 2) -> set[str]:
        """seed から hops ホップ以内(両方向)。推移的関係は閉包まで辿る。"""
        included = set(seeds)
        frontier = deque((s, 0) for s in seeds)
        while frontier:
            node, d = frontier.popleft()
            for p, o in self.out.get(node, []):
                nd = d if p in TRANSITIVE_PREDICATES else d + 1
                if o not in included and nd <= hops:
                    included.add(o); frontier.append((o, nd))
            for p, s in self.inc.get(node, []):
                nd = d if p in TRANSITIVE_PREDICATES else d + 1
                if s not in included and nd <= hops:
                    included.add(s); frontier.append((s, nd))
        return included

    def serialize_subgraph(self, nodes: set[str]) -> str:
        """部分グラフを型注釈付きの明示的トリプルとして文字列化。"""
        lines = ["# 型(オントロジー)"]
        shown = set()
        for nid in sorted(nodes):
            e = ENTITY_BY_ID.get(nid)
            if e and e.type not in shown:
                lines.append(f"- クラス階層: {' ⊂ '.join(self.superclasses(e.type))}")
                shown.add(e.type)
        lines.append("# エンティティ")
        for nid in sorted(nodes):
            e = ENTITY_BY_ID.get(nid)
            if e:
                lines.append(f"- {e.id} (型: {e.type}) : {e.desc}")
        lines.append("# 関係トリプル")
        for s, p, o in RELATIONS:
            if s in nodes and o in nodes:
                lines.append(f"- {s} --{p}--> {o}")
        return "\n".join(lines)

    def retrieve_context(self, ent_ids: set[str], cls_ids: set[str],
                         hops: int = 2) -> str:
        """クラス言及は subsumption で全インスタンス(+1 ホップ近傍で結合先も可視化)、
        エンティティ言及は近傍部分グラフを抽出して直列化する。"""
        nodes: set[str] = set()
        class_seeds: set[str] = set()
        for cls in cls_ids:
            class_seeds.update(self.instances_of(cls))
        nodes.update(class_seeds)
        if class_seeds:
            nodes.update(self.neighborhood(class_seeds, hops=1))
        if ent_ids:
            nodes.update(self.neighborhood(ent_ids, hops=hops))
        if not nodes:
            return ""
        return self.serialize_subgraph(nodes)

payment-serviceの部分グラフをhops=2で表示してみます。

$ python - << 'PY'
from graph_store import OntologyGraph
g = OntologyGraph()
ents, clss = g.link(["payment-service"])
print(g.retrieve_context(ents, clss, hops=2))
PY

# 型(オントロジー)
- クラス階層: EdgeService ⊂ Service ⊂ Component
- クラス階層: CoreService ⊂ Service ⊂ Component
- クラス階層: Team
- クラス階層: MessageQueue ⊂ Datastore ⊂ Component
- クラス階層: SupportingService ⊂ Service ⊂ Component
- クラス階層: RelationalDB ⊂ Datastore ⊂ Component
- クラス階層: CacheStore ⊂ Datastore ⊂ Component
# エンティティ
- api-gateway (型: EdgeService) : 全ての外部リクエストを最初に受ける入口。認証委譲とルーティングを担う。
- auth-service (型: CoreService) : トークン発行と検証を行う認証の中核。停止すると全体のログインが不能になる。
- commerce-team (型: Team) : 取引・決済・在庫を担当。
- event-bus (型: MessageQueue) : ドメインイベントを配送する Kafka トピック群。
- growth-team (型: Team) : ユーザー・通知・検索を担当。
- inventory-service (型: CoreService) : 在庫の引き当てと補充を管理する。
- notification-service (型: SupportingService) : メールとプッシュ通知を送る。イベント駆動で動作する。
- order-service (型: CoreService) : 注文の受付・確定・状態遷移を管理する。
- orders-db (型: RelationalDB) : 注文履歴を格納する PostgreSQL クラスタ。
- payment-service (型: CoreService) : 決済処理と返金を担う。外部決済ゲートウェイと連携する。
- payments-db (型: RelationalDB) : 決済トランザクションを格納する PostgreSQL クラスタ。
- platform-team (型: Team) : 基盤・認証・配信を担当。
- search-service (型: SupportingService) : 商品検索インデックスを提供する。
- session-cache (型: CacheStore) : セッションとトークンを保持する Redis。
- user-service (型: CoreService) : ユーザープロフィールと権限を管理する。
- users-db (型: RelationalDB) : ユーザー情報を格納する PostgreSQL クラスタ。
# 関係トリプル
- api-gateway --depends_on--> auth-service
- api-gateway --depends_on--> order-service
- api-gateway --depends_on--> user-service
- api-gateway --depends_on--> search-service
- order-service --depends_on--> payment-service
- order-service --depends_on--> inventory-service
- order-service --depends_on--> user-service
- payment-service --depends_on--> auth-service
- user-service --persists_to--> users-db
- order-service --persists_to--> orders-db
- payment-service --persists_to--> payments-db
- auth-service --persists_to--> session-cache
- order-service --publishes_to--> event-bus
- payment-service --publishes_to--> event-bus
- notification-service --subscribes_to--> event-bus
- api-gateway --owned_by--> platform-team
- auth-service --owned_by--> platform-team
- event-bus --owned_by--> platform-team
- session-cache --owned_by--> platform-team
- order-service --owned_by--> commerce-team
- payment-service --owned_by--> commerce-team
- inventory-service --owned_by--> commerce-team
- orders-db --owned_by--> commerce-team
- payments-db --owned_by--> commerce-team
- user-service --owned_by--> growth-team
- notification-service --owned_by--> growth-team
- search-service --owned_by--> growth-team
- users-db --owned_by--> growth-team

出力の # 関係トリプル に order-service --depends_on--> payment-service とapi-gateway --depends_on--> order-service が明示的に並ぶのがポイントです。

LLM は、この明示された関係を参照することで、api-gatewayまでの依存関係を読み取るための情報を得られます。散文では埋もれていた経路情報が、トリプルとして明示的に取り出されました。

ここで重要なのは、このretrieve_context(payment-service, hops=2)が 16 ノードすべてを含んでいることです。これはhops=2 の両方向探索が、推移的 depends_on を経由してグラフ全体に届くためです。
正解（order-service, api-gateway）は含まれますが、payments-db / event-bus / 各 team まで一緒に入ります。
この「正解は必ず入るが、周辺ノイズも多い」性質が、10 章で precision を下げる原因になります。この点は、10 章の評価で重要になります。

8. 限界が確認された質問をオントロジー併用 RAG で処理する

「言及抽出＝LLM、グラフ探索＝決定的、最終回答＝構造を主根拠に LLM」という役割分担で実装していきます。

OntologyRAG.answer() の一巡（3 者の分業を時系列で）

image.png

肝は LLM に構造推論をさせないことです。多段の追跡や網羅はグラフが決定的に済ませ、LLM には「①言葉を記号に直す（言及抽出）」と「②出来上がった部分グラフを読んで答える（最終回答）」の言語処理だけを任せます。ベクトル検索は、曖昧な候補の絞り込みを補助します。

正しい言及抽出・リンキングを前提とすれば、正解がコンテキストに含まれることはグラフ探索によって保証されます。残る誤りは、主に LLM によるコンテキストの「解釈」に起因します。

リンキング（entity linking）とは
質問から抜き出した語（「payment-service」「データストア」等）を、グラフ上の正式な ID ／クラスに対応づける処理です。表記ゆれは CLASS_SYNONYMS が吸収します（例：「リレーショナル DB」「RDB」→ RelationalDB）。ここが「自然言語→構造化された ID・クラス」の入口で、以降のグラフ探索はこの記号だけで決定的に進みます。

まず extract_json_array を複数箇所で使うので llm.py に移します。

  • llm.py（第 3 版＝最終）
    chat / embed はそのままにしてextract_json_array を追加します。以下が全文です。

    from __future__ import annotations
import os, json
from openai import OpenAI

BASE_URL = os.environ.get("OPENAI_BASE_URL", "http://localhost:11434/v1")
API_KEY = os.environ.get("OPENAI_API_KEY", "ollama")
CHAT_MODEL = os.environ.get("CHAT_MODEL", "gemma4:e2b")
EMBED_MODEL = os.environ.get("EMBED_MODEL", "nomic-embed-text")

_client = OpenAI(base_url=BASE_URL, api_key=API_KEY)

def embed(texts: list[str]) -> list[list[float]]:
    resp = _client.embeddings.create(model=EMBED_MODEL, input=texts)
    return [d.embedding for d in resp.data]

def chat(system: str, user: str, temperature: float = 0.0) -> str:
    resp = _client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[{"role": "system", "content": system},
                  {"role": "user", "content": user}],
        temperature=temperature,
    )
    return resp.choices[0].message.content or ""

def extract_json_array(text: str) -> list[str]:
    s, e = text.find("["), text.rfind("]")
    if s == -1 or e == -1 or e < s:
        return []
    try:
        return [str(x).strip() for x in json.loads(text[s:e + 1]) if str(x).strip()]
    except json.JSONDecodeError:
        return []

  • rag_plain.py（第 2 版）
    自前の extract_json_array を削除し、llm から import します。以下が全文です。

    from __future__ import annotations
from vector_store import VectorStore
from llm import chat, extract_json_array
from kb import generate_documents

SYSTEM = (
    "あなたはシステム構成に関する質問に答えるアシスタントです。"
    "与えられたコンテキストだけを根拠に答えてください。"
    "最終的な該当コンポーネントの ID を、必ず JSON 配列だけで最後に出力してください。"
    "例: [\"order-service\", \"api-gateway\"]。該当なしは [] とします。"
)

class PlainRAG:
    def __init__(self, k: int = 4):
        self.k = k
        self.store = VectorStore(generate_documents())

    def answer(self, question: str) -> tuple[list[str], str]:
        hits = self.store.search(question, k=self.k)
        context = "\n\n".join(f"[{h['id']}] {h['text']}" for h in hits)
        prompt = f"# コンテキスト\n{context}\n\n# 質問\n{question}\n\nJSON 配列で答えよ。"
        out = chat(SYSTEM, prompt)
        return extract_json_array(out), out

そして本命の併用 RAGの実装に入ります。

  • rag_onto.py

    from __future__ import annotations
from vector_store import VectorStore
from graph_store import OntologyGraph
from llm import chat, extract_json_array
from kb import generate_documents, ENTITIES, CLASS_HIERARCHY, CLASS_SYNONYMS

MENTION_SYS = (
    "質問文から、既知のコンポーネント名・型名・チーム名に相当する語だけを抜き出し、"
    "JSON 配列で返してください。説明は不要です。"
)

def _mention_hint() -> str:
    ents = ", ".join(e.id for e in ENTITIES)
    clss = ", ".join(c for c in CLASS_HIERARCHY)
    syns = "; ".join(f"{k}={'/'.join(v)}" for k, v in CLASS_SYNONYMS.items())
    return f"既知エンティティ: {ents}\n既知クラス: {clss}\nクラス別名: {syns}"

ANSWER_SYS = (
    "あなたはナレッジグラフを根拠に推論するアシスタントです。"
    "型(クラス階層)と関係トリプルを厳密にたどって答えてください。"
    "推移的な依存や型の包含 (subsumption) を正しく適用すること。"
    "コンテキストには回答対象でない周辺ノード(チーム・データストア等)も文脈として含まれる。"
    "質問が問う関係・型を満たす ID だけを出力し、周辺ノードを安易に列挙に含めないこと。"
    "最終的な該当コンポーネントの ID を、必ず JSON 配列だけで最後に出力してください。"
    "該当なしは [] とします。"
)

class OntologyRAG:
    def __init__(self, k: int = 4, hops: int = 2):
        self.k, self.hops = k, hops
        self.store = VectorStore(generate_documents())
        self.graph = OntologyGraph()

    def _mentions(self, question: str) -> list[str]:
        out = chat(MENTION_SYS, f"{_mention_hint()}\n\n質問: {question}\n\nJSON 配列:")
        return extract_json_array(out)

    def answer(self, question: str) -> tuple[list[str], str]:
        hits = self.store.search(question, k=self.k)
        text_ctx = "\n".join(f"[{h['id']}] {h['text']}" for h in hits)
        mentions = self._mentions(question)
        ents, clss = self.graph.link(mentions)
        graph_ctx = self.graph.retrieve_context(ents, clss, hops=self.hops)
        prompt = (
            f"# 文書コンテキスト\n{text_ctx}\n\n"
            f"# 構造化ナレッジ(オントロジー+トリプル)\n{graph_ctx}\n\n"
            f"# 質問\n{question}\n\n"
            "構造化ナレッジを主たる根拠として、JSON 配列で答えよ。"
        )
        out = chat(ANSWER_SYS, prompt)
        return extract_json_array(out), out

5章では取りこぼした質問を、今度は併用 RAG を使って問い合わせてみます。

この質問について確実に言えるのは、次の決定的な事実のみです。

  • 部分グラフは正解を必ず含む：retrieve_context(["payment-service"]) の出力にorder-service と api-gateway（および両者を結ぶ depends_on トリプル）は必ず現れます（7.2 節で確認済み）。つまり「答えに必要な材料はコンテキストに載っている」ことは仕組みから保証されます。

一方、その材料から最終的にどの ID を拾うかは LLM 次第です。ある 1 回の実行例：

$ python - << 'PY'
from rag_onto import OntologyRAG
rag = OntologyRAG(k=4, hops=2)
ids, raw = rag.answer(
    "payment-service が停止したら、直接的・間接的に影響を受けるサービスをすべて挙げよ。")
print("pred:", ids)   # 期待 (gold): ['api-gateway', 'order-service']
PY

pred: ['order-service', 'auth-service', 'payments-db']

この回は api-gateway を正しく抽出できず、当たったのは order-service だけでした。ここで重要なのは、取得は成功しているのに外しています。7.2 節のとおり部分グラフには api-gateway も
api-gateway --depends_on--> order-service の辺も入っている——必要な情報はコンテキストに含まれていたにもかかわらず、LLM がapi-gateway → order-service → payment-service の2 ホップの逆向きの連鎖を最後までたどれなかったわけです。

さらに、混入した 2 つが症状を的確に語っています。auth-service は payment-service が依存する先（payment-service --depends_on--> auth-service）、payments-db は payment-service が書き込む先（persists_to）。どちらも「payment-service が停止して影響を受ける側」ではなく、逆方向です。

つまり小さいモデルが、16 ノード・両方向混在の部分グラフを渡されて 「影響を受ける（逆向き）」と「依存する／使う（順向き）」を取り違えた、というのが、この失敗の原因です。

これは失敗例ですが、本チュートリアルで示している「被覆と最終回答は別の問題である」という点を具体的に示す例です。

被覆（recall の土台）は仕組みで保証できるが、その先の“読み”は LLM 依存で外れる。

だから 10 章では結論を LLM の数値ではなく被覆に置き、11 章では質問種別でルーティング（＋取得の絞り込み）へ進みます。

とくに障害波及の質問は、両方向 2 ホップの塊ではなく逆推移閉包の部分グラフだけを渡せば、auth-service / payments-db のような逆方向ノードがそもそも文脈に入らず、この誤りは構造的に消えます（発展で言及）。

実行のたびに結果は変わります。api-gateway を拾う回も、別のノイズが増える回もあります。
たとえば 10 章の集計表でも Q1 は api-gateway を落としています（別のノイズが乗って F1 は 0.40）。実行ごとにノイズの中身は変わりますが、api-gateway を落とす傾向は共通です。
実行ごとに結果は変動しますが、変わらないのは「被覆が成立している＝正解は必ずコンテキストに載っている」という決定的事実だけです。

9. 評価を形式化する（正解をグラフから導出）

質問を 1 問ずつ目視で評価する方法には限界があります。正解を人手で個別に定義するのではなく、グラフから決定的に計算して採点します。

これで質問を足しても正解が自動的に導出され、再現性も出ます。指標は集合ベースの Precision / Recall / F1 とします。

  • evaluate.py

    from __future__ import annotations
from dataclasses import dataclass
from graph_store import OntologyGraph
from kb import RELATIONS, owner_of

G = OntologyGraph()

def reverse_transitive(target: str, pred: str = "depends_on") -> set[str]:
    """target に(推移的に)依存する主体 = 障害の影響範囲。graph_store に委譲。"""
    return G.reverse_transitive(target, pred)

def instances(cls: str) -> set[str]:
    return set(G.instances_of(cls))

def persists_to_relational() -> set[str]:
    rdb = instances("RelationalDB")
    return {s for s, p, o in RELATIONS if p == "persists_to" and o in rdb}

def cross_team_deps(team: str) -> set[str]:
    owned = {s for s, p, o in RELATIONS if p == "owned_by" and o == team}
    deps: set[str] = set()
    for s, p, o in RELATIONS:
        if p == "depends_on" and s in owned:
            owner = owner_of(o)              # 堅牢化: 所有者不明なら None
            if owner is not None and owner != team:
                deps.add(o)
    return deps

@dataclass
class Question:
    qid: str
    text: str
    gold: set[str]
    kind: str   # multihop / type / join / single

QUESTIONS: list[Question] = [
    Question("Q1",
        "payment-service が障害で停止した場合、直接的または間接的に影響を受けるサービスをすべて挙げよ。",
        reverse_transitive("payment-service"), "multihop"),
    Question("Q2",
        "このシステムのデータストアに該当するコンポーネントをすべて列挙せよ。",
        instances("Datastore"), "type"),
    Question("Q3",
        "リレーショナルDBに永続化しているサービスをすべて挙げよ。",
        persists_to_relational(), "join"),
    Question("Q4",
        "auth-service の役割を説明し、対象コンポーネントの ID を答えよ。",
        {"auth-service"}, "single"),
    Question("Q5",
        "commerce-team が所有するサービスが依存している、他チーム所有のコンポーネントをすべて挙げよ。",
        cross_team_deps("commerce-team"), "join"),
]

def prf1(pred: set[str], gold: set[str]) -> tuple[float, float, float]:
    if not gold:
        return (1.0, 1.0, 1.0) if not pred else (0.0, 1.0, 0.0)
    tp = len(pred & gold)
    prec = tp / len(pred) if pred else 0.0
    rec = tp / len(gold)
    f1 = 2 * prec * rec / (prec + rec) if (prec + rec) else 0.0
    return prec, rec, f1

if __name__ == "__main__":
    for q in QUESTIONS:
        print(f"{q.qid} [{q.kind}] gold = {sorted(q.gold)}")

Q1～Q5 までの質問を投げてみましょう。

$ python evaluate.py

Q1 [multihop] gold = ['api-gateway', 'order-service']
Q2 [type] gold = ['event-bus', 'orders-db', 'payments-db', 'session-cache', 'users-db']
Q3 [join] gold = ['order-service', 'payment-service', 'user-service']
Q4 [single] gold = ['auth-service']
Q5 [join] gold = ['auth-service', 'user-service']

10. 全体比較

ここが本チュートリアルの心臓部です。結論を LLM の数値ではなく、被覆（正解が取得範囲に入るか） に置きます。

Recall と Precision とは

  • Recall（再現率）：正解のうち、何割を拾えたか。 「正解をどれだけ漏れなく取得できたか」。取りこぼすと下がる。
  • Precision（適合率）：答えた ID のうち、何割が正解か。 「誤った情報をどれだけ含めずに済んだか」。ノイズを混ぜると下がる。
  • F1 は両者の調和平均で、片方だけ高くても伸びません。

このチュートリアルの分業は、この 2 つに対応します。
Recall を決めるのは「検索・取得」の層（正解がコンテキストに載るか）なので、被覆が保証されれば recall は取り戻すことができます。
Precision を最後に左右するのは「LLM の読み」なので、 部分グラフに周辺ノイズが同居すると、LLM が余計な ID を混ぜて下がります。
以下ではまず recall の基盤となる情報の被覆を決定的に確認し、その上で precision の変動を LLM 側の問題として観察します。

10.1 被覆解析

各質問について「正解ノードが検索器の取得範囲に入るか」を、LLM を介さず判定します。

  • coverage_check.py（被覆の検証: Ollama 不要）

    from graph_store import OntologyGraph
from evaluate import QUESTIONS

g = OntologyGraph()

# 各問で「言及抽出が理想的に動けば渡るべき語」を固定し、
# retrieve_context のノード集合に gold が含まれるか(被覆)を決定的に確認する。
IDEAL_MENTIONS = {
    "Q1": ["payment-service"], "Q2": ["データストア"], "Q3": ["リレーショナルDB"],
    "Q4": ["auth-service"],    "Q5": ["commerce-team"],
}

print(f"{'QID':<4}{'kind':<10}{'|gold|':>7}  被覆(gold⊆取得) 取得ノード数")
for q in QUESTIONS:
    ents, clss = g.link(IDEAL_MENTIONS[q.qid])
    ctx = g.retrieve_context(ents, clss, hops=2)
    nodes = {ln[2:].split(" (型:")[0] for ln in ctx.splitlines()
             if ln.startswith("- ") and "(型:" in ln}
    print(f"{q.qid:<4}{q.kind:<10}{len(q.gold):>7}  {str(q.gold <= nodes):>10}      {len(nodes)}")

K = 4
print("\nプレーン RAG の被覆上限(件数の観点): top-k=4 で gold 全件が同時に入り得るか")
for q in QUESTIONS:
    print(f"  {q.qid} [{q.kind}] |gold|={len(q.gold)} vs k={K} -> "
          f"{'不可 (件数超過)' if len(q.gold) > K else '件数の上では取得可能'}")

それでは被覆解析をしてみましょう。

$ python coverage_check.py

QID kind       |gold|  被覆(gold⊆取得) 取得ノード数
Q1  multihop        2        True      16
Q2  type            5        True      16
Q3  join            3        True      12
Q4  single          1        True      16
Q5  join            2        True      15

プレーン RAG の被覆上限(件数の観点): top-k=4 で gold 全件が同時に入り得るか
  Q1 [multihop] |gold|=2 vs k=4 -> 件数の上では取得可能
  Q2 [type] |gold|=5 vs k=4 -> 不可 (件数超過)
  Q3 [join] |gold|=3 vs k=4 -> 件数の上では取得可能
  Q4 [single] |gold|=1 vs k=4 -> 件数の上では取得可能
  Q5 [join] |gold|=2 vs k=4 -> 件数の上では取得可能

この表から、LLM を一切動かさない状態で以下のような状況になっていると言えます。

  • 併用 RAG は全 5 問で「gold ⊆ 取得部分グラフ」が成立します（被覆はすべて True）。
    Q1 は推移閉包、Q2 は subsumption、Q3 は型網羅＋1 ホップ結合先、Q5 は近傍で、それぞれ正解を部分グラフに含めます。
    つまり、「正解集合が取得部分グラフに含まれること」 は、探索方式によって保証されます。ただし、そこから正しい ID を選択して回答できるかどうかは LLM に依存します。

  • プレーン RAG の限界は 2 種類です。

    1. 件数：Q2 は正解が 5 件で、k=4 の top-k にはそもそも同時に入りません（recall 上限が 4/5 未満に固定）。
    2. 意味的近接：Q1/Q3/Q5 は件数的には収まり得ますが、正解群が互いに意味的に近いとは限らない（例：「payment の障害」と api-gateway は語彙が遠い）。

    ベクトルは経路情報も結合条件も持たないため、top-k がこれらを同時に引く保証はありません。

10.2 エンドツーエンドの実測（1 回分）

上の被覆が保証された上で、実際に LLM が最終回答をどう出すか。まず Plain と Onto の 2 方式を run.py で回します（Hybrid は 11 章で router.py を作ってから足します）。

  • run.py（第1版: Plain と Onto を比較。Hybrid は 11 章で router を作ってから追加）

    from __future__ import annotations
from evaluate import QUESTIONS, prf1
from rag_plain import PlainRAG
from rag_onto import OntologyRAG

def run_system(name, system):
    rows, macro = [], []
    for q in QUESTIONS:
        pred, raw = system.answer(q.text)
        pred_set = {p.strip().lower() for p in pred}
        prec, rec, f1 = prf1(pred_set, q.gold)
        rows.append((q.qid, q.kind, sorted(pred_set), prec, rec, f1))
        macro.append(f1)
    return rows, sum(macro) / len(macro)

def print_table(name, rows, macro_f1):
    print(f"\n===== {name} =====")
    print(f"{'QID':<4}{'kind':<10}{'P':>6}{'R':>6}{'F1':>6}  pred")
    for qid, kind, pred, p, r, f1 in rows:
        print(f"{qid:<4}{kind:<10}{p:>6.2f}{r:>6.2f}{f1:>6.2f}  {pred}")
    print(f"{'':<20}{'':>6}{'macro-F1':>12} = {macro_f1:.3f}")

if __name__ == "__main__":
    r1, m1 = run_system("Plain RAG", PlainRAG(k=4))
    r2, m2 = run_system("Ontology + RAG", OntologyRAG(k=4, hops=2))
    print_table("Plain RAG", r1, m1)
    print_table("Ontology + RAG", r2, m2)
    print(f"\nΔ macro-F1 (onto - plain) = {m2 - m1:+.3f}")

macro-F1 とは
各質問の F1 をそのまま平均した値（問ごとに同じ重み）。件数の多い問いに引っ張られない代わりに、1 問の当たり外れが 1/5 ぶん効きます。だから小さいモデルでは値が動きやすく、順位の目安として読むのが安全です。

適合率を測定してみましょう。

gemma4:e2b は確率的で、モデル・バージョン・シードにより数値は変わるため、下表は参考出力となります。

$ python run.py

===== Plain RAG =====
QID kind           P     R    F1  pred
Q1  multihop    0.00  0.00  0.00  []
Q2  type        0.33  0.20  0.25  ['inventory-service', 'search-service', 'session-cache']
Q3  join        1.00  0.33  0.50  ['payment-service']
Q4  single      0.00  0.00  0.00  []
Q5  join        0.00  0.00  0.00  []
                              macro-F1 = 0.150

===== Ontology + RAG =====
QID kind           P     R    F1  pred
Q1  multihop    0.33  0.50  0.40  ['commerce-team', 'notification-service', 'order-service']
Q2  type        1.00  1.00  1.00  ['event-bus', 'orders-db', 'payments-db', 'session-cache', 'users-db']
Q3  join        1.00  1.00  1.00  ['order-service', 'payment-service', 'user-service']
Q4  single      0.00  0.00  0.00  ['platform-team', 'session-cache']
Q5  join        0.00  0.00  0.00  ['event-bus']
                              macro-F1 = 0.480

Δ macro-F1 (onto - plain) = +0.330

解説：

  • type / join（Q2・Q3・Q5）は、構造情報が正しく取得できていれば、Ontology RAG が高い精度を出しやすい質問種別です。 この回も Q2/Q3 は 1.00 でした。被覆が保証され、問いが「該当する型・関係の要素を挙げよ」と明快なので、小規模なモデルでも比較的誤答しにくいと考えられます。一方、Q5 のように結合の向きを取り違えて落とす回もあります。
  • multihop（Q1）は当たっても部分的。 この実行では Ontology RAG が正解の片方 order-service を拾いつつ、commerce-team / notification-service（無関係＝ノイズ）を混ぜ、api-gateway を落としました。被覆は成立している（api-gateway は文脈に居る）のに、2 ホップの逆向きの連鎖を最後までたどれない——8 章と同じ現象です。recall は上げやすい一方、precision と“最終的な関係追跡”は LLM 依存で揺れます。
  • single（Q4）はこの回、両方式とも外しました。 plain は []、onto は ['platform-team', 'session-cache']（無関係）で、ともに 0.00。plain は 4 章では単一事実を正答しますが、常に安定とは限りません。確実に言えるのは構造だけ：単一事実に 16 ノードの部分グラフを渡す onto は無関係ノードを足すだけで過剰——そのため11章では、単一事実の質問をプレーンRAGに振り分けます。
  • 総じて F1 はモデルの出力変動の影響を受けます。ゆえに結論は数値ではなく 10.1 節の被覆——「正解は必ず取得部分グラフに載る」という決定的事実——に置きます。macro-F1 の差（onto−plain＝+0.330）は「関係系で recall を取り戻した」ことの要約にすぎません。

つまり、「recall を改善するための土台」は、正解情報の被覆によって決定的に保証されます。一方、「precision と単一事実の劣化」は、不要な情報まで含めた過剰な取得＝ノイズによって説明できます。 これは、「併用すれば一律に性能が上がる」という単純な説明よりも、今回の実測結果と整合した捉え方です。

11. ルーティングと総評

10 章の観察はそのまま設計指針になります。単一事実はプレーン、関係・網羅・結合はグラフ併用に振り分けることで、Q4 のような単一事実質問での過剰取得を避けつつ、関係系で正解情報の被覆を改善する土台を作れます。これを実装します。

  • router.py（10 章の観察を実装: 質問種別でRAGを切り替える）

    from __future__ import annotations
from rag_plain import PlainRAG
from rag_onto import OntologyRAG

# 関係性・網羅性・結合条件を示唆する語。含まれている場合はグラフ併用 RAG に振り分ける。
# 素朴なキーワード方式。本番では分類器や LLM 判定に差し替え可能。
RELATIONAL_HINTS = ("影響", "依存", "波及", "停止", "すべて", "全て", "列挙",
                    "永続化", "所有", "他チーム", "データストア", "リレーショナル")

def route(question: str) -> str:
    """'onto' か 'plain' を返す。単一事実の質問はプレーン RAG、それ以外の関係性を含む質問は Ontology RAG に振り分ける"""
    return "onto" if any(h in question for h in RELATIONAL_HINTS) else "plain"

class HybridRAG:
    def __init__(self, k: int = 4, hops: int = 2):
        self.plain = PlainRAG(k=k)
        self.onto = OntologyRAG(k=k, hops=hops)

    def answer(self, question: str) -> tuple[list[str], str]:
        if route(question) == "onto":
            return self.onto.answer(question)
        return self.plain.answer(question)

LLM を介さないキーワード判定したルーティングの結果を確認してみます。

$ python -c "from router import route; from evaluate import QUESTIONS; \
[print(q.qid, q.kind, '->', route(q.text)) for q in QUESTIONS]"

Q1 multihop -> onto
Q2 type -> onto
Q3 join -> onto
Q4 single -> plain
Q5 join -> onto

単一事実の Q4 だけがプレーンに回り、残りはグラフ併用となっていることが確認できます。目的は、プレーン RAG が得意とする単一事実の質問と、グラフ併用 RAG が得意とする関係性を含む質問を適切に使い分けることです。

router ができたので、run.py に Hybrid を足して 3 方式で比べます。10 章の第 1 版に HybridRAG の行を加えるだけです。

  • run.py（第2版: router 追加後。Plain / Onto / Hybrid を比較）

    from __future__ import annotations
from evaluate import QUESTIONS, prf1
from rag_plain import PlainRAG
from rag_onto import OntologyRAG
from router import HybridRAG

def run_system(name, system):
    rows, macro = [], []
    for q in QUESTIONS:
        pred, raw = system.answer(q.text)
        pred_set = {p.strip().lower() for p in pred}
        prec, rec, f1 = prf1(pred_set, q.gold)
        rows.append((q.qid, q.kind, sorted(pred_set), prec, rec, f1))
        macro.append(f1)
    return rows, sum(macro) / len(macro)

def print_table(name, rows, macro_f1):
    print(f"\n===== {name} =====")
    print(f"{'QID':<4}{'kind':<10}{'P':>6}{'R':>6}{'F1':>6}  pred")
    for qid, kind, pred, p, r, f1 in rows:
        print(f"{qid:<4}{kind:<10}{p:>6.2f}{r:>6.2f}{f1:>6.2f}  {pred}")
    print(f"{'':<20}{'':>6}{'macro-F1':>12} = {macro_f1:.3f}")

if __name__ == "__main__":
    r1, m1 = run_system("Plain RAG", PlainRAG(k=4))
    r2, m2 = run_system("Ontology + RAG", OntologyRAG(k=4, hops=2))
    r3, m3 = run_system("Hybrid (router)", HybridRAG(k=4, hops=2))
    print_table("Plain RAG", r1, m1)
    print_table("Ontology + RAG", r2, m2)
    print_table("Hybrid (router)", r3, m3)
    print(f"\nΔ macro-F1 (onto - plain)   = {m2 - m1:+.3f}")
    print(f"Δ macro-F1 (hybrid - plain) = {m3 - m1:+.3f}")

Hybrid を加えた適合率を測定してみます。

$ python run.py

===== Plain RAG =====
QID kind           P     R    F1  pred
Q1  multihop    0.00  0.00  0.00  []
Q2  type        0.33  0.20  0.25  ['inventory-service', 'search-service', 'session-cache']
Q3  join        1.00  0.33  0.50  ['payment-service']
Q4  single      0.00  0.00  0.00  []
Q5  join        0.00  0.00  0.00  []
                              macro-F1 = 0.150

===== Ontology + RAG =====
QID kind           P     R    F1  pred
Q1  multihop    0.33  0.50  0.40  ['commerce-team', 'notification-service', 'order-service']
Q2  type        1.00  1.00  1.00  ['event-bus', 'orders-db', 'payments-db', 'session-cache', 'users-db']
Q3  join        1.00  1.00  1.00  ['order-service', 'payment-service', 'user-service']
Q4  single      0.00  0.00  0.00  ['platform-team', 'session-cache']
Q5  join        0.00  0.00  0.00  ['event-bus']
                              macro-F1 = 0.480

===== Hybrid (router) =====
QID kind           P     R    F1  pred
Q1  multihop    0.33  0.50  0.40  ['commerce-team', 'notification-service', 'order-service']
Q2  type        1.00  1.00  1.00  ['event-bus', 'orders-db', 'payments-db', 'session-cache', 'users-db']
Q3  join        1.00  1.00  1.00  ['order-service', 'payment-service', 'user-service']
Q4  single      0.00  0.00  0.00  []
Q5  join        0.00  0.00  0.00  ['event-bus']
                              macro-F1 = 0.480

Δ macro-F1 (onto - plain)   = +0.330
Δ macro-F1 (hybrid - plain) = +0.330

Hybrid は Q4 を plain に回します。ただしこの回は plain 自身も Q4 を外したため、数値上の改善は出ていません（Hybrid の macro は Onto と同じ 0.480）。ここで重要なのは、「改善する／改善しない」という数値だけではなく、このルーティング設計の考え方です。単一事実に 16 ノードの部分グラフ（＝ノイズ源）を渡さない、という判断は、モデルが弱い場合やグラフが大きい場合に、より効果が現れる可能性があります。gemma4:e2b では出力の変動が大きく改善効果が数値として現れにくいので、より強いモデルや、障害波及を逆推移閉包だけに絞る取得（発展） と組み合わせると、この設計の効果がはっきり出ます。

総評：オントロジー併用が「有効なケース／不要なケース」

判断基準は一つ、「答えが単一チャンクに閉じるか、複数の事実間の関係に依存するか」 です。

オントロジー併用が有効になりやすいケース

  • 推移的・多段の関係推論（障害波及、依存連鎖、到達可能性）。ベクトルは経路情報を持たない（Q1）。
  • 型による網羅・集約（「全ての◯◯」）。subsumption によって、top-k による取得件数の制約を回避できる。（Q2）。
  • 型 × 関係の結合（「◯◯型に□□しているもの」）。散文だけでは、型と関係を組み合わせた検索条件を明示的に扱うことが困難。（Q3 ・ Q5）。
  • 監査可能性・説明性が求められる領域では、明示トリプルを根拠となる経路情報として辿れる点が利点になる。
  • 同義語や語彙の揺らぎが多い領域。CLASS_SYNONYMS で正規化できる。

不要または過剰になりやすいケース（プレーン RAG で十分）

  • 単一事実の検索・定義・要約（Q4）。追加しても精度が向上せず、むしろ部分グラフのノイズで下がり得る。加えて、言及抽出のための LLM 呼び出しがレイテンシの増加要因になる。
  • 文書が互いに独立なコーパス（FAQ ・規約・記事検索）。
  • オントロジー整備が追いつかない／型体系が頻繁に変わる領域。誤った型付けはノイズになる。

実務上は、まずプレーン RAG から始め、関係性・網羅性・結合条件を扱う必要が生じた段階でグラフを導入するのが現実的です。

そして全質問へ一律適用せず、質問の種類でルーティングする（router.py）のが精度と計算コストのバランスを取りやすくなります。

本文の評価軸（multihop / type / join / single に分けて F1 を見る）は、「どの質問種別で効果が現れるか」を可視化する最小の枠組みでもあります。

LLM モデル比較

gpt-5.6 は、temperature パラメータに対応していないため、llm.pyの該当箇所を削除しています。

小型ローカルのgemma4:e2bとgpt-5.6の結果を突き合わせると、かなりきれいな構造が見えます。まず表にまとめます。

macro-F1 サマリ

モデル Plain Onto Hybrid Δ(onto−plain)
gemma4:e2b + nomic-embed 0.150 0.480 0.480 +0.330
gpt-5.6 + te3-small 0.548 1.000 1.000 +0.452

質問種別ごとの F1（Plain → Onto）

QID kind e2b Plain e2b Onto gpt Plain gpt Onto 構造で解けたか
Q1 multihop 0.00 0.40 0.67 1.00 モデル依存
Q2 type 0.25 1.00 0.57 1.00 ◎ 両モデルで満点
Q3 join 0.50 1.00 0.50 1.00 ◎ 両モデルで満点
Q4 single 0.00 0.00 1.00 1.00 小型は素の能力で失敗
Q5 join(2段) 0.00 0.00 0.00 1.00 構造＋強モデルの両方が必要

※ Hybrid は両モデルとも Onto と同一スコアのため列を省略（後述）。

考察

1. 「構造化の効果」はモデル非依存に成立した。 これが最重要の観察です。Q2(型網羅)とQ3(型×関係の結合)は、2B級の e2b でも gpt-5.6 でも等しく 0.00〜0.57 → 1.00 に跳ねました。プレーンでは macro 0.150 しか出ない非力なモデルですら、被覆さえコード側で保証すれば type/join を満点で解く。これは「LLMの賢さで殴る」のではなく「知識の表現形式を変える」という設計思想の、いちばんクリーンな実証です。

2. ただし『到達可能』と『綺麗に到達する』は別物。 被覆を解決しても、Q1・Q4・Q5では e2b が崩れます。中身を見ると失敗の質が示唆的です。

  • Q1(e2b onto=0.40): pred が [commerce-team, notification-service, order-service]。order-service は正解だが、Team を混入(型制約違反) し、近傍展開で入り込んだ notification-service という distractor に釣られ、api-gateway を落としている。トリプルを渡しても、逆向き推移閉包を辿りつつ「サービスだけ」に絞る指示追従が2Bには重い。
  • Q4(e2b): 最も簡単な単一事実で plain=、onto=platform-team, session-cache。構造化コンテキストが、弱いモデルにとっては auth-service の"隣"(所有チーム・永続化先)を拾わせるノイズとして働いた。
  • Q5(2段join): 強モデルでのみ 1.00。owned_by＋depends_on＋他チーム判定の多段結合は、被覆があっても e2b の合成能力を超える。

つまり結果は2軸に分解できます。検索軸(構造=被覆)はモデル非依存に解決／推論軸(回答合成・指示追従)はモデル依存。 標語化すれば「オントロジーは答えを到達可能にする。モデルが綺麗に到達できるかを決める。」Δが強モデルで大きい(+0.452)のは、gpt-5.6が構造を取りこぼしなく使い切って天井(1.000)に届く一方、e2bはオントロジー側の天井自体が推論力で 0.480 に引き下げられているためです。

3. プレーンの底は埋め込み/モデルで動くが、天井は構造で決まる。 plain macro は 0.150(nomic+e2b)→0.548(te3-small+gpt)と大きく改善しますが、改善の中身は Q1/Q2 の recall がわずかに上がった程度で、Q5 は両モデルとも 0.00、Q2/Q3 の recall も 0.33〜0.40 で頭打ち。top-k の被覆限界は、モデルを強くしても外れない構造的な天井だと裏付けられました。

4. Hybrid(router)は今回は効果ゼロだった — これは正直に書くべき点です。 両モデルで Hybrid ≡ Onto。router の利得は本来「onto が弱点化する single-fact を plain に逃がす」ことで出ますが、今回その条件が発生していません。gpt-5.6 では onto が Q4 も満点なので逃がす必要がなく、e2b では plain も Q4 を失敗([]) しているため逃がしても 0 のまま。router が価値を持つのは「plain が single-fact で成功し、かつ onto がそこで劣化する」中間帯のモデル/質問で、その帯が今回のサンプルには無かった、という結果です。

5. 記事の例示表との差(正直な補足)。 私がチュートリアルに載せた例示 plain(macro 0.681)は、実測(gpt 0.548 / e2b 0.150)より楽観的でした。特に Q5 を例示では 0.67 としましたが、実測は plain 両モデルで 0.00。例示は「plain がある程度取れる」前提でしたが、2段joinは実際にはもっと厳しい。記事側は実測値に差し替えるのが誠実です。

6
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
6
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?