基礎編では、プレーン RAG とオントロジー併用 RAG を同じ評価セットで比較し、精度差の源泉が「正解の被覆」にあることを確認しました。また、質問種別に応じて処理方式を切り替える Hybrid も作りました。到達した結論は以下の通りです。
オントロジーによって、答えに必要な情報を取得しやすくなります。
ただし、その情報を正しく読み取り、最終的な答えとして取り出せるかどうかはモデルに左右されます。
被覆（recall の土台）は、探索方式によって決定的に保証できます。しかし、最終回答の precision と多段の追跡は LLM に依存して揺れます。実際に、弱い gemma4:e2b の場合は Q1 / Q4 / Q5 が崩れました。
応用編では、この「揺れ」を設計で潰し切るまでを追います。全体の流れは次のとおりです。
進め方
- 取得を厳密化する — 改修A（型フィルタ）/ 改修B（方向付き検索）と、その落とし穴
- 生成を外部化する① — Text2Cypher（内蔵ミニ Cypher エンジン）
- 生成を外部化する② — Text2SQL（SQLite ＋ 型閉包）
- 非構造テキストからグラフを作る — GraphRAG
- システム統合① — 3方向ルーター（vector / graph / SQL）と「統合＝正義ではない」
- システム統合② — フォールバック・カスケード
- あるべき姿 — 検証可能な型付き IR（Intermediate Representation：中間表現）と決定的実行
- 総括 — 「失敗の圧縮」という視点
今回の検証にあたっては基礎編のコードを引き継ぎます。LLM も基礎編と同様に OpenAI 互換エンドポイントを使います。主に Ollama の gemma4:e2b を使い、比較用に gpt-5.6 も扱います。
壊れるのは、常に「LLM が最終回答やフリーテキストのクエリ（SQL / Cypher）を書く」箇所です。 応用編では、その責務を段階的に LLM から剥がし、決定的に計算できる部分は決定的に処理する 設計へ寄せていきます。
1. 取得を厳密化する — 改修A / 改修B
基礎編 8 章の失敗を思い返してみましょう。「payment-serviceの停止で影響を受けるものは？」という質問に対し、e2b は向きを取り違え、auth-service（依存先）や payments-db（書き込み先）を答えに含めてしまい、api-gateway を取りこぼしていました。原因は、retrieve_context が両方向 2 ホップで 16 ノードを一括で渡し、その中に「順方向の隣接ノード」という distractor が混ざっていたことです。
ここでは 2 つの改修を入れます。
- 改修A（型フィルタ）： 要求される型（例：影響を受ける「サービス」）で出力を機械的に絞ります。
-
改修B（方向付き検索）：「影響を受ける＝逆向き
depends_onの推移閉包」と分かる場合に向きを固定し、distractor を検索段階で入れないようにします。
まず graph_store.py に 2 つのメソッドを追加します。
def is_a(self, entity_id: str, cls: str) -> bool:
"""entity_id が cls(またはその下位)のインスタンスか。"""
e = ENTITY_BY_ID.get(entity_id)
return bool(e) and cls in self.superclasses(e.type)
def directional_closure(self, seeds: set[str], predicate: str,
direction: str) -> set[str]:
"""単一述語を、指定方向にだけ推移的に辿る(seed自身は含めない)。
direction: 'forward'(s->o) / 'reverse'(o<-s) / 'both'。
「影響を受ける= reverse depends_on」のような向きを厳密化し、
両方向広域展開で混入する distractor を排除する。"""
result: set[str] = set()
stack = list(seeds)
while stack:
cur = stack.pop()
if direction in ("forward", "both"):
for p, o in self.out.get(cur, []):
if p == predicate and o not in result:
result.add(o); stack.append(o)
if direction in ("reverse", "both"):
for p, s in self.inc.get(cur, []):
if p == predicate and s not in result:
result.add(s); stack.append(s)
return result
retrieve_context は focus=(述語, 向き) を受け取り、指定があれば方向付き閉包を、なければ従来の両方向近傍を使います（空なら近傍へフォールバックする非破壊化も後述のとおり織り込み済み）。
def retrieve_context(self, ent_ids: set[str], cls_ids: set[str],
hops: int = 2,
focus: tuple[str, str] | None = None) -> str:
"""リンク結果から構造化コンテキストを組み立てる。
- クラス言及: subsumption で全インスタンス(+1ホップ近傍で結合先も可視化)
- エンティティ言及:
focus=(predicate, direction) があれば方向付き閉包で厳密に、
なければ両方向 hops 近傍(フォールバック)で抽出する。"""
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:
focus_nodes: set[str] = set()
if focus is not None and focus[1] in ("forward", "reverse", "both"):
focus_nodes = self.directional_closure(ent_ids, focus[0], focus[1])
if focus_nodes:
nodes.update(ent_ids) # アンカー自身も文脈として残す
nodes.update(focus_nodes) # 方向付き閉包が非空 → 絞り込みを採用
else:
# 焦点が空(Team起点など、その述語では辿れない)= 近傍にフォールバック。
# これを怠るとQ5のような結合質問で文脈がアンカー1点に潰れる。
nodes.update(self.neighborhood(ent_ids, hops=hops))
if not nodes:
return ""
return self.serialize_subgraph(nodes)
方向付き閉包の効果は LLM を介さず確認できます。
$ python -c "from graph_store import OntologyGraph as G; print(sorted(G().directional_closure({'payment-service'},'depends_on','reverse')))"
['api-gateway', 'order-service']
両方向 16 ノードだった Q1 の文脈は、3 ノードに絞られます。
このため、notification-service や各 team はそもそも文脈に存在しなくなります。
1.1 破壊的改修の落とし穴 — 非破壊化ガード
ところが、改修 A/B をそのまま入れて再評価すると、回帰（regression） が出ました。以下は手元の一例です。
| Q1 | Q2 | Q3 | Q4 | Q5 | macro | |
|---|---|---|---|---|---|---|
| gpt Onto 改修前 | 1.00 | 1.00 | 1.00 | 1.00 | 1.00 | 1.000 |
| gpt Onto 改修後（ガード無） | 1.00 | 1.00 | 1.00 | 1.00 | 0.00 | 0.800 |
| e2b Onto 改修前 | 0.40 | 1.00 | 1.00 | 0.00 | 0.00 | 0.480 |
| e2b Onto 改修後（ガード無） | 0.00 | 1.00 | 0.00 | 0.50 | 1.00 | 0.500 |
原因は、改修 A/B がいずれも 「追加はできても復元できない」破壊的操作だったことです。
-
方向付き検索の過剰適用：Q5 のアンカー
commerce-team（Team）を起点にdepends_onを辿ると閉包は空になり、文脈がアンカー 1 点に潰れて結合が消える。 -
型フィルタの誤爆：
expected_typeを取り違えると正解が全消し（サービスを DB 型で濾すと 1.00→0.00）。
対策は、両方を非破壊化することです。「方向付き閉包が空なら近傍へフォールバックする」「型フィルタで全消しになるなら適用しない」という形にします。これにより、Q1 の絞り込みは維持したまま、Q5 の崩壊を回避できます。この点は決定的に確認できます。
この節の要点は、後段フィルタやクエリプランナは failure surface を増やすということです。導入するなら、必ず非破壊（空集合を作らない・元集合より悪化させない） にし、「効かないときは何もしない」を既定にする必要があります。そうでなければ、クエリ解析が外れる弱いモデルほど、改修によって悪化し得ます。
この改修 A/B は、「LLM に構造推論をさせる」路線の延長でした。次章からは発想を変え、構造推論そのものを LLM から剥がします。
2. 生成を外部化する① — Text2Cypher
併用 RAG では、最後の「部分グラフを読んで ID を選ぶ」処理を LLM に委ねていました。ここが揺れの源です。そこで、LLM には『自然言語 → クエリ』の翻訳だけをさせ、答えの計算はエンジンに任せることにします。本番なら Neo4j に任せるのが良いですが、ここでは外部 DB を立てず、対応部分集合の Cypher を実行する最小エンジンを内蔵します。execute_cypher を driver 呼び出しに差し替えれば、実 Neo4j に接続できます。可変長パス -[:depends_on*]->（推移閉包）と、型ラベルの下位包含に対応させます。
"""発展1: Text2Cypher(自己完結版)。
自然言語 → Cypher(部分集合) → 内蔵の最小グラフエンジンで実行。
外部DB不要。実Neo4jへ向ける場合は execute_cypher を neo4j ドライバ呼び出しに差し替える。
対応するCypher部分集合(このチュートリアル用):
MATCH <pattern> [MATCH <pattern> ...] [WHERE <cond> [AND <cond> ...]] RETURN [DISTINCT] var.prop
node : (var) | (var:Label) Label は型(下位型を含む subsumption 照合)
rel : -[:PRED]-> <-[:PRED]- -[:PRED]-(無向) および 可変長 -[:PRED*]-> <-[:PRED*]-
cond : var.id = 'x' | var.id <> 'x' | var.type = 'X'
return : var.id / var.type (複数可、DISTINCT可)
"""
from __future__ import annotations
import re, json
from kb import ENTITIES, RELATIONS, ENTITY_BY_ID
from graph_store import OntologyGraph
from llm import chat
_G = OntologyGraph()
# ---- グラフ(隣接) ----
_OUT: dict[str, list[tuple[str, str]]] = {}
_IN: dict[str, list[tuple[str, str]]] = {}
for _s, _p, _o in RELATIONS:
_OUT.setdefault(_s, []).append((_p, _o))
_IN.setdefault(_o, []).append((_p, _s))
def _label_ok(node_id: str, label: str | None) -> bool:
if label is None:
return True
e = ENTITY_BY_ID.get(node_id)
return bool(e) and label in _G.superclasses(e.type) # 下位型も一致(subsumption)
# ---- パターン解析 ----
_NODE = re.compile(r"\((\w+)(?::(\w+))?\)")
_REL = re.compile(r"(<-|-)\[:(\w+)(\*)?\]-(>?)")
def _parse_pattern(pat: str):
"""(node, rel, node, rel, ...) を [nodes],[rels] に分解。"""
nodes, rels, pos = [], [], 0
m = _NODE.match(pat, pos)
if not m:
raise ValueError(f"pattern must start with node:{pat}")
nodes.append((m.group(1), m.group(2))); pos = m.end()
while pos < len(pat):
rm = _REL.match(pat, pos)
if not rm:
break
left, pred, star, right = rm.groups()
direction = "->" if right == ">" else ("<-" if left == "<-" else "--")
rels.append((pred, direction, bool(star))); pos = rm.end()
nm = _NODE.match(pat, pos)
if not nm:
raise ValueError(f"expected node after rel:{pat[pos:]}")
nodes.append((nm.group(1), nm.group(2))); pos = nm.end()
return nodes, rels
def _neighbors(node_id: str, pred: str, direction: str):
res = []
if direction in ("->", "--"):
res += [o for (p, o) in _OUT.get(node_id, []) if p == pred]
if direction in ("<-", "--"):
res += [s for (p, s) in _IN.get(node_id, []) if p == pred]
return res
def _reachable(node_id: str, pred: str, direction: str) -> list[str]:
"""可変長(*): 1ホップ以上の到達集合(自分自身は含めない)。"""
seen, stack = set(), [node_id]
while stack:
cur = stack.pop()
for nxt in _neighbors(cur, pred, direction):
if nxt not in seen:
seen.add(nxt); stack.append(nxt)
return list(seen)
def _match_one(nodes, rels, binding: dict) -> list[dict]:
"""1つのMATCHパターンを既存bindingに対して展開し、全解を返す(バックトラック)。"""
results = []
def rec(i: int, b: dict):
if i == 0:
var, lbl = nodes[0]
cands = [b[var]] if var in b else [e.id for e in ENTITIES]
for c in cands:
if _label_ok(c, lbl) and (var not in b or b[var] == c):
rec(1, {**b, var: c})
return
if i > len(rels):
results.append(b); return
pred, direction, star = rels[i - 1]
prev_var = nodes[i - 1][0]
cur_var, cur_lbl = nodes[i]
src = b[prev_var]
cands = _reachable(src, pred, direction) if star else _neighbors(src, pred, direction)
for c in cands:
if not _label_ok(c, cur_lbl):
continue
if cur_var in b and b[cur_var] != c:
continue
rec(i + 1, {**b, cur_var: c})
rec(0, binding)
return results
_COND = re.compile(r"(\w+)\.(\w+)\s*(=|<>)\s*'([^']*)'")
def _apply_where(bindings: list[dict], where: str) -> list[dict]:
if not where:
return bindings
conds = [(m.group(1), m.group(2), m.group(3), m.group(4))
for m in _COND.finditer(where)]
out = []
for b in bindings:
ok = True
for var, prop, op, val in conds:
nid = b.get(var)
actual = nid if prop == "id" else (ENTITY_BY_ID[nid].type if nid else None)
if op == "=" and actual != val:
ok = False; break
if op == "<>" and actual == val:
ok = False; break
if ok:
out.append(b)
return out
def execute_cypher(query: str) -> list[str]:
"""対応部分集合のCypherを実行し、RETURNの1列目(id)の重複なしリストを返す。"""
q = query.strip().rstrip(";")
m_ret = re.search(r"\bRETURN\b(.*)$", q, re.IGNORECASE | re.DOTALL)
if not m_ret:
return []
ret_body = m_ret.group(1).strip()
distinct = bool(re.match(r"(?i)DISTINCT\b", ret_body))
ret_body = re.sub(r"(?i)^DISTINCT\b", "", ret_body).strip()
ret_items = [x.strip() for x in ret_body.split(",") if x.strip()]
ret_var, ret_prop = ret_items[0].split(".")
body = q[:m_ret.start()]
where = ""
mw = re.search(r"\bWHERE\b(.*)$", body, re.IGNORECASE | re.DOTALL)
if mw:
where = mw.group(1); body = body[:mw.start()]
match_clauses = re.findall(r"\bMATCH\b\s*(.+?)(?=\bMATCH\b|$)",
body, re.IGNORECASE | re.DOTALL)
bindings = [{}]
for pat in match_clauses:
nodes, rels = _parse_pattern(pat.strip())
nxt = []
for b in bindings:
nxt += _match_one(nodes, rels, b)
bindings = nxt
bindings = _apply_where(bindings, where)
seen, out = set(), []
for b in bindings:
nid = b.get(ret_var)
val = nid if ret_prop == "id" else (ENTITY_BY_ID[nid].type if nid else None)
if val is not None and (not distinct or val not in seen):
seen.add(val); out.append(val)
return out
# ============ 自然言語 → Cypher（LLM生成 + 実行）============
def _schema_hint() -> str:
classes = ("Component>Service>{EdgeService,CoreService,SupportingService}, "
"Component>Datastore>{RelationalDB,CacheStore,MessageQueue}, Team")
preds = "depends_on, persists_to, publishes_to, subscribes_to, owned_by"
ids = ", ".join(e.id for e in ENTITIES)
return (f"ノードラベル(型階層):{classes}\n"
f"リレーション(述語):{preds}\n"
f"ノードのプロパティ: id, type\n"
f"既知のid:{ids}")
_CYPHER_SYS = (
"あなたは自然言語をCypherに翻訳する専門家です。次の部分集合のみ使用可能:\n"
"MATCH <pattern> [MATCH ...] [WHERE cond [AND ...]] RETURN [DISTINCT] var.id\n"
"node=(v) or (v:Label)。Labelは型で下位型も一致する。"
"rel=-[:PRED]-> / <-[:PRED]- / 可変長 -[:PRED*]->。"
"cond= v.id = 'x' / v.id <> 'x' / v.type = 'X'。\n"
"『影響を受ける=停止したノードに(推移的に)依存する主体』は "
"MATCH (a)-[:depends_on*]->(t) WHERE t.id='...' RETURN DISTINCT a.id の形。\n"
"Cypher文のみを1行で返し、説明やコードフェンスは付けない。"
)
_FEWSHOT = (
"例1 Q:データストアを全部挙げて → MATCH (d:Datastore) RETURN DISTINCT d.id\n"
"例2 Q:注文サービスが依存する先は → MATCH (s)-[:depends_on]->(o) "
"WHERE s.id='order-service' RETURN DISTINCT o.id\n"
"例3 Q:リレーショナルDBに保存するサービスは → "
"MATCH (s:Service)-[:persists_to]->(d:RelationalDB) RETURN DISTINCT s.id"
)
def nl_to_cypher(question: str) -> str:
raw = chat(_CYPHER_SYS, f"{_schema_hint()}\n\n{_FEWSHOT}\n\n質問:{question}\n\nCypher:")
line = raw.strip().splitlines()[0] if raw.strip() else ""
return re.sub(r"^```\w*|```$", "", line).strip().rstrip(";")
class Text2CypherRAG:
"""発展1: NL→Cypher→内蔵エンジン実行。retry で軽い自己修復を行う。"""
def __init__(self, retries: int = 1):
self.retries = retries
def answer(self, question: str) -> tuple[list[str], str]:
last_cypher = ""
for attempt in range(self.retries + 1):
cypher = nl_to_cypher(question if attempt == 0
else f"{question}\n(前回のCypher『{last_cypher}』は"
f"結果が空/不正。修正して。)")
last_cypher = cypher
try:
got = execute_cypher(cypher)
except Exception:
got = []
if got:
return got, cypher
return [], last_cypher
手書き Cypher では、5 問すべてが gold と一致します。これはエンジンの正しさを確認するための結果です。LLM は、この Cypher を出せればよいことになります。
実測（run_advanced.py） では、gpt-5.6 の Text2Cypher が macro-F1 = 1.000 になりました。Q5 の 2 段結合まで全問正解です。決定的実行なので、正しい Cypher さえ出れば、型名の綴り違いや取りこぼしは起きません。一方、e2b は Cypher 生成自体が不安定で、0.400 でした。
3. 生成を外部化する② — Text2SQL
同じ発想を SQL でも試します。KB を SQLite に載せ、型の下位包含を class_closure テーブルで表現します。SQL は、集約・件数・結合が得意な経路です。一方、推移的依存（Q1）は素の SQL では書きにくいため、後段のルーターで graph へ回します。
"""発展3の一部: Text2SQL。KBをSQLiteに載せ、NL→SQL→実行。
集約・数え上げ・結合が得意な経路。型の下位包含は class_closure テーブルで表現する。"""
from __future__ import annotations
import re, sqlite3
from kb import ENTITIES, RELATIONS, CLASS_HIERARCHY
from graph_store import OntologyGraph
from llm import chat
_G = OntologyGraph()
def build_db() -> sqlite3.Connection:
con = sqlite3.connect(":memory:")
c = con.cursor()
c.execute("CREATE TABLE entity(id TEXT PRIMARY KEY, type TEXT, label TEXT, descr TEXT)")
c.execute("CREATE TABLE relation(subj TEXT, pred TEXT, obj TEXT)")
# class_closure(sub, super): sub は super の下位(自身も含む)。型包含をSQLで扱うため。
c.execute("CREATE TABLE class_closure(sub TEXT, super TEXT)")
c.executemany("INSERT INTO entity VALUES(?,?,?,?)",
[(e.id, e.type, e.label, e.desc) for e in ENTITIES])
c.executemany("INSERT INTO relation VALUES(?,?,?)", RELATIONS)
closure = []
for cls in CLASS_HIERARCHY:
for sup in _G.superclasses(cls):
closure.append((cls, sup))
c.executemany("INSERT INTO class_closure VALUES(?,?)", closure)
con.commit()
return con
_SCHEMA = """テーブル定義:
entity(id, type, label, descr) -- type はリーフクラス
relation(subj, pred, obj) -- pred: depends_on/persists_to/publishes_to/subscribes_to/owned_by
class_closure(sub, super) -- sub型は super型の下位(自身含む)。型の包含判定に使う
型の包含を使う例: あるエンティティが Datastore かは
entity e JOIN class_closure cc ON e.type=cc.sub WHERE cc.super='Datastore'
推移的依存(影響波及)は素のSQLでは書きにくいので、その種の質問は避けるか単純化する。"""
_SQL_SYS = (
"あなたは自然言語をSQLite用SQLに翻訳する専門家です。"
"回答対象の entity.id を1列だけ SELECT すること(重複はDISTINCT)。"
"型の下位包含が要るときは class_closure を JOIN する。"
"SQL文のみを返し、説明やコードフェンスは付けない。"
)
def nl_to_sql(question: str) -> str:
raw = chat(_SQL_SYS, f"{_SCHEMA}\n\n質問:{question}\n\nSQL:")
line = " ".join(raw.strip().splitlines())
return re.sub(r"```\w*|```", "", line).strip().rstrip(";")
class Text2SQLRAG:
def __init__(self, retries: int = 1):
self.con = build_db()
self.retries = retries
def _run(self, sql: str) -> list[str]:
if not re.match(r"(?is)^\s*select\b", sql): # SELECT以外は実行しない
return []
cur = self.con.cursor()
cur.execute(sql)
return [str(r[0]) for r in cur.fetchall()]
def answer(self, question: str) -> tuple[list[str], str]:
last = ""
for attempt in range(self.retries + 1):
sql = nl_to_sql(question if attempt == 0
else f"{question}\n(前回SQL『{last}』は空/不正。修正して。)")
last = sql
try:
got = self._run(sql)
except Exception:
got = []
if got:
# 重複除去(順序維持)
seen, out = set(), []
for x in got:
if x not in seen:
seen.add(x); out.append(x)
return out, sql
return [], last
実測では、gpt-5.6 の Text2SQL は macro-F1 = 0.800 でした。Q1 の再帰 CTE が label / id を取り違えて空振りし、それ以外は 1.00 でした。class_closure により、型網羅（Q2）と型 × 関係の結合（Q3 / Q5）は SQL で厳密に解けます。
4. 非構造テキストからグラフを作る — GraphRAG
これまではトリプルを手書きしてきました。しかし、実務では非構造テキストしかないことも多いのが実情です。GraphRAG（Microsoft, 2024）は、LLM でテキストからグラフを抽出し、コミュニティ検出で要約を作り、局所質問と大域質問の双方に答える方式です。ここでは、その最小自己完結版を作ります。コミュニティ検出は、ラベル伝播をフルスクラッチで実装します。
"""発展2: GraphRAG(Microsoft, 2024 の最小自己完結版)。
非構造テキスト → LLMでトリプル抽出 → グラフ構築 → コミュニティ検出(ラベル伝播,
フルスクラッチ) → コミュニティ要約(LLM) → 局所検索(local)/大域検索(global)。
- local : 特定エンティティ近傍＋所属コミュニティ要約で答える(「Xは何に依存?」等)
- global : 全コミュニティ要約へ map-reduce で答える(「全体像」「◯◯を全部」等)
"""
from __future__ import annotations
import json, re
from collections import defaultdict, Counter
from llm import chat, embed, extract_json_array
# ---------- 1. 抽出 ----------
_EXTRACT_SYS = (
"次の文から知識グラフを抽出し、JSONだけ返す(説明不要):\n"
'{"entities":[{"name":"...","type":"..."}],'
'"relations":[{"source":"...","relation":"...","target":"..."}]}\n'
"name は文中の固有名、type は概念クラス(例: Service, Datastore, Team)。"
)
def extract_graph(docs: list[dict]) -> dict:
"""docs=[{id,text}] からトリプルとエンティティ型を抽出してマージ。"""
ent_type: dict[str, str] = {}
edges: set[tuple[str, str, str]] = set()
for d in docs:
raw = chat(_EXTRACT_SYS, d["text"])
s, e = raw.find("{"), raw.rfind("}")
if s == -1 or e < s:
continue
try:
obj = json.loads(raw[s:e + 1])
except json.JSONDecodeError:
continue
for ent in obj.get("entities", []):
name = str(ent.get("name", "")).strip()
if name:
ent_type.setdefault(name, str(ent.get("type", "Unknown")))
for r in obj.get("relations", []):
a, rel, b = (str(r.get("source", "")).strip(),
str(r.get("relation", "")).strip(),
str(r.get("target", "")).strip())
if a and b and rel:
edges.add((a, rel, b))
ent_type.setdefault(a, "Unknown")
ent_type.setdefault(b, "Unknown")
return {"types": ent_type, "edges": sorted(edges)}
# ---------- 2. コミュニティ検出(ラベル伝播: フルスクラッチ・決定的) ----------
def detect_communities(graph: dict) -> list[list[str]]:
nodes = sorted(graph["types"])
adj: dict[str, set[str]] = defaultdict(set)
for a, _r, b in graph["edges"]: # 無向として扱う
adj[a].add(b); adj[b].add(a)
label = {n: n for n in nodes}
for _ in range(100):
changed = False
for n in nodes: # 決定的順序
neigh = adj.get(n, set())
if not neigh:
continue
cnt = Counter(label[m] for m in neigh)
top = max(cnt.values())
best = min(l for l, c in cnt.items() if c == top) # 同数は辞書順で決定的に
if best != label[n]:
label[n] = best; changed = True
if not changed:
break
comms: dict[str, list[str]] = defaultdict(list)
for n, l in label.items():
comms[l].append(n)
return [sorted(v) for v in comms.values()]
# ---------- 3. コミュニティ要約(LLM) ----------
_SUMMARY_SYS = ("与えられたエンティティ群と関係から、このコミュニティが何のまとまりかを"
"3文以内で要約してください。")
def summarize_communities(graph: dict, comms: list[list[str]]) -> list[dict]:
summaries = []
for i, members in enumerate(comms):
mset = set(members)
rels = [f"{a} -{r}->{b}" for a, r, b in graph["edges"]
if a in mset and b in mset]
body = ("エンティティ: " + ", ".join(members) + "\n関係: " + "; ".join(rels))
summaries.append({"id": i, "members": members,
"summary": chat(_SUMMARY_SYS, body), "rels": rels})
return summaries
# ---------- 4. 検索 ----------
_LOCAL_SYS = ("あなたはグラフを根拠に答えるアシスタント。近傍の関係と要約を使い、"
"該当エンティティのidをJSON配列で最後に出力。該当なしは []。")
_MAP_SYS = ("次のコミュニティ要約に、質問に該当するエンティティidがあればJSON配列で挙げよ。"
"無ければ []。要約に無い情報は出さない。")
_REDUCE_SYS = ("複数の部分回答(idの配列)を統合し、重複を除いた最終idのJSON配列だけを返す。")
class GraphRAG:
def __init__(self, docs: list[dict]):
self.graph = extract_graph(docs)
self.comms = detect_communities(self.graph)
self.summaries = summarize_communities(self.graph, self.comms)
self._adj_out = defaultdict(list)
self._adj_in = defaultdict(list)
for a, r, b in self.graph["edges"]:
self._adj_out[a].append((r, b)); self._adj_in[b].append((r, a))
# エンティティ名の埋め込み(局所検索のリンク用)
self._names = sorted(self.graph["types"])
self._vecs = embed(self._names) if self._names else []
def _link(self, question: str, top: int = 3) -> list[str]:
if not self._vecs:
return []
import numpy as np
M = np.array(self._vecs, dtype=np.float32)
M /= (np.linalg.norm(M, axis=1, keepdims=True) + 1e-9)
q = np.array(embed([question])[0], dtype=np.float32)
q /= (np.linalg.norm(q) + 1e-9)
idx = np.argsort(-(M @ q))[:top]
return [self._names[i] for i in idx]
def local(self, question: str) -> tuple[list[str], str]:
seeds = self._link(question)
ctx_lines = []
for s in seeds:
for r, b in self._adj_out.get(s, []):
ctx_lines.append(f"{s} -{r}->{b}")
for r, a in self._adj_in.get(s, []):
ctx_lines.append(f"{a} -{r}->{s}")
prompt = ("# 近傍関係\n" + "\n".join(sorted(set(ctx_lines))) +
f"\n\n# 質問\n{question}\n\nJSON配列で答えよ。")
return extract_json_array(chat(_LOCAL_SYS, prompt)), "local"
def global_(self, question: str) -> tuple[list[str], str]:
partials: list[str] = []
for s in self.summaries: # map
body = (f"要約:{s['summary']}\n関係:{'; '.join(s['rels'])}\n\n質問:{question}")
partials += extract_json_array(chat(_MAP_SYS, body))
merged = extract_json_array(chat( # reduce
_REDUCE_SYS, f"部分回答:{json.dumps(partials, ensure_ascii=False)}"))
return (merged or sorted(set(partials))), "global"
def answer(self, question: str) -> tuple[list[str], str]:
# 「全部/一覧/全体像」等は大域、それ以外は局所(素朴な振り分け)
g = any(w in question for w in ("すべて", "全て", "全部", "一覧", "全体", "列挙"))
return self.global_(question) if g else self.local(question)
実 KB のトリプルを抽出結果として与えると、ラベル伝播が 2 つの妥当なコミュニティに分割し、局所近傍・大域 map-reduce の文脈組み立ても機能します。
正直な注意として、抽出された固有名（“Payment Service”）が id 空間（“payment-service”）と食い違うことがあります。また、IR を厳密に実行できても、その正しさは元グラフの品質で頭打ちになります。抽出が不確実になるぶん、これまでの厳密性は弱まります。詰めるなら、抽出に証拠スパンを付与する検証層が次の課題になります。
5. システム統合① — 3方向ルーター
基礎編の router.py は plain / onto の 2 択でした。ここでは、ベクトル / グラフ（Cypher）/ SQL を使い分ける、本番寄りの定石に拡張します。
"""発展3: ベクトル / グラフ(Cypher) / SQL を使い分ける3方向ルーター。
- vector : 単一事実・定義・説明 → PlainRAG
- graph : 経路・推移・到達可能性 → Text2CypherRAG(可変長パスが得意)
- sql : 集約・件数・結合・型網羅 → Text2SQLRAG
LLM分類 + ヒューリスティックのフォールバック。分類器は本番では学習分類器に差し替え可。"""
from __future__ import annotations
from rag_plain import PlainRAG
from text2cypher import Text2CypherRAG
from text2sql import Text2SQLRAG
from llm import chat
_ROUTE_SYS = (
"質問を次のいずれか1語で分類し、その語だけ返す:\n"
"vector = 単一の事実/定義/説明の検索\n"
"graph = 推移的な依存・経路・到達可能性(例: 影響波及, 連鎖)\n"
"sql = 集約・件数・結合・型による網羅(例: 全ての◯◯, ◯◯型で□□するもの)\n"
"語のみを返す。"
)
_GRAPH_HINTS = ("影響", "波及", "依存", "停止", "到達", "連鎖")
_SQL_HINTS = ("すべて", "全て", "全部", "列挙", "何個", "件数", "永続化", "所有", "他チーム",
"データストア", "リレーショナル")
def route(question: str) -> str:
"""LLM分類 → 不明ならヒューリスティック。"""
ans = chat(_ROUTE_SYS, f"質問:{question}\n分類:").strip().lower()
for key in ("graph", "sql", "vector"):
if key in ans:
return key
if any(h in question for h in _GRAPH_HINTS):
return "graph"
if any(h in question for h in _SQL_HINTS):
return "sql"
return "vector"
class AdvancedRouterRAG:
def __init__(self, k: int = 4):
self.backends = {
"vector": PlainRAG(k=k),
"graph": Text2CypherRAG(),
"sql": Text2SQLRAG(),
}
def answer(self, question: str) -> tuple[list[str], str]:
r = route(question)
pred, detail = self.backends[r].answer(question)
return pred, f"{r}:{detail}"
ここで、実測が重要な教訓を与えてくれます。 以下は gpt-5.6 の結果で、決定的なので信頼できます。
| システム | macro-F1 |
|---|---|
| Onto（基礎編） | 1.000 |
| Text2Cypher | 1.000 |
| Text2SQL | 0.800 |
| Advanced Router | 0.800 |
3 方向ルーターは、単体最強の Cypher / Onto（1.000）より悪くなりました。 Q3 を sql に回したところ、SQL が型名を RelationalDatabase と誤生成して空振りし、そこで確定して 0 点になりました。単一選択ルーターは、誤ルーティングや生成失敗を復旧できないため、単体最良より劣化し得ます。統合＝正義ではない、というのが実測の結論です。
6. システム統合② — フォールバック・カスケード
単一選択の弱点は、「1 回選んで終わり」になることです。そこで、主バックエンドを選び、空またはエラーなら次へ落ちるカスケードにします。verify フックを持たせ、既定は非空チェックにします。将来的には、型整合や件数サニティも挿入できます。
"""実運用向けの統合版(1つの出荷システム)。
ルーターで主バックエンドを選び、空/エラーなら他バックエンドへフォールバック(カスケード)。
単一選択ルーターの『誤ルーティング＋生成失敗で0点』を、経路の多重化で吸収する。
注意(正直な限界): カスケードは「空・実行エラー」でしか次へ落ちない。
非空だが誤り(過剰包含など)は主経路で確定してしまう。そこを詰めるには
型整合や件数上限などの verifier を挟む(下の verify フックで拡張可能)。"""
from __future__ import annotations
from rag_plain import PlainRAG
from rag_onto import OntologyRAG
from text2cypher import Text2CypherRAG
from text2sql import Text2SQLRAG
from router_advanced import route
# 主経路の既定カスケード順(構造クエリに強い順 → 最後に単一事実のvector)
DEFAULT_ORDER = ["graph", "sql", "onto", "vector"]
class UnifiedRAG:
def __init__(self, k: int = 4, verify=None):
self.by_name = {
"vector": PlainRAG(k=k),
"onto": OntologyRAG(k=k),
"graph": Text2CypherRAG(),
"sql": Text2SQLRAG(),
}
# verify(question, pred) -> bool。Falseなら不採用として次へ落ちる。既定は非空チェックのみ。
self.verify = verify or (lambda q, pred: bool(pred))
def answer(self, question: str) -> tuple[list[str], str]:
primary = route(question)
order = [primary] + [b for b in DEFAULT_ORDER if b != primary]
trace = []
for name in order:
try:
pred, detail = self.by_name[name].answer(question)
except Exception:
pred, detail = [], "error"
ok = self.verify(question, pred)
trace.append(name + ("✓" if ok else "×"))
if ok:
return pred, f"[{' > '.join(trace)}]{name}:{detail}"
return [], f"[{' > '.join(trace)}] all empty"
これで、Q3 が sql で空振りしても graph に落ちて回収されます（sql× > graph✓）。gpt-5.6 のクエリ列を再現した検証では、単一選択ルーターが失った 0.2 を取り戻し、macro-F1 = 1.000 になりました。
正直な限界として、カスケードが救えるのは空・実行エラーのみです。非空だが誤り（過剰包含など）は、主経路で確定してしまいます。そこを詰めるには、「有効か」ではなく「正しいか」を検証する層が必要です。これが次章の核心です。
7. あるべき姿 — 検証可能な型付き IR と決定的実行
ここまでの失敗の原因は、一貫して LLM がフリーテキスト（最終回答・SQL・Cypher）を書く箇所にあります。型名の綴り違い、label / id の取り違え、過剰包含、逆向きの解釈は、すべて「自由に書けてしまう」ことに起因します。
そこで、発想を最後まで推し進めます。LLM の仕事を『自然言語 → 検証可能な型付き IR』の意味解析だけに限定し、実行はグラフ上で決定的・厳密に行います。 グラフは既知で信頼できるため、これで「型 / 述語 / id の綴り違い」や「過剰包含」は原理的に消えます。検証を通った IR の実行結果は、定義上正しいからです。
7.1 IR 代数と検証・実行
IR は、「id の集合を返す小さな代数」です。const / instances（型網羅）/ step（1 ホップ）/ closure（推移閉包）/ of_type（型で絞る）/ union・intersect・difference / vector（構造で表せない合図）を用意します。検証器が未知の型・述語・id・不正な向きを弾き、実行器が決定的に評価します。
"""あるべき姿の中核: 型付きクエリ中間表現(IR)の検証 + 決定的実行。
LLMはNL→この安全なIRへの意味解析だけを行い、実行はグラフ上で厳密に行う。
検証を通ったIRは、未知の型/述語/id・不正な向きを含み得ない → 実行結果は定義上正しい。
IRの代数(各ノードは『entity idの集合』を返す):
{"op":"const", "ids":[...]} リテラル集合
{"op":"instances", "type":"Datastore"} 型(下位型含む)の全インスタンス
{"op":"step", "seed":<ir>, "predicate":"persists_to", "direction":"forward|reverse|both"} 1ホップ
{"op":"closure", "seed":<ir>, "predicate":"depends_on", "direction":"reverse"} 推移閉包
{"op":"of_type", "seed":<ir>, "type":"Service"} 集合を型で絞る(subsumption)
{"op":"union"|"intersect"|"difference", "args":[<ir>, <ir>, ...]}
{"op":"vector"} 構造で表せない → ベクトルRAGへ委譲の合図
"""
from __future__ import annotations
from kb import RELATIONS, ENTITY_BY_ID, CLASS_HIERARCHY
from graph_store import OntologyGraph
_G = OntologyGraph()
_PREDICATES = {p for _s, p, _o in RELATIONS}
_OUT: dict[str, list[tuple[str, str]]] = {}
_IN: dict[str, list[tuple[str, str]]] = {}
for _s, _p, _o in RELATIONS:
_OUT.setdefault(_s, []).append((_p, _o))
_IN.setdefault(_o, []).append((_p, _s))
_SET_OPS = {"union", "intersect", "difference"}
_ALL_OPS = {"const", "instances", "step", "closure", "of_type", "vector", "ref"} | _SET_OPS
class PlanError(ValueError):
pass
# ---------------- 検証(スキーマ整合) ----------------
def validate(ir: dict, path: str = "$", known_names: set | None = None) -> None:
known_names = known_names or set()
if not isinstance(ir, dict) or "op" not in ir:
raise PlanError(f"{path}: opを持つオブジェクトが必要")
op = ir["op"]
if op not in _ALL_OPS:
raise PlanError(f"{path}: 未知のop '{op}'。使用可能:{sorted(_ALL_OPS)}")
if op == "vector":
return
if op == "ref":
if ir.get("name") not in known_names:
raise PlanError(f"{path}.ref: 未定義の中間変数 '{ir.get('name')}'。"
f"定義済み:{sorted(known_names)}")
return
if op == "const":
ids = ir.get("ids")
if not isinstance(ids, list) or not ids:
raise PlanError(f"{path}.ids: 非空の配列が必要")
for i in ids:
if i not in ENTITY_BY_ID:
raise PlanError(f"{path}.ids: 未知のid '{i}'")
return
if op == "instances" or op == "of_type":
t = ir.get("type")
if t not in CLASS_HIERARCHY:
raise PlanError(f"{path}.type: 未知の型 '{t}'。使用可能:{sorted(CLASS_HIERARCHY)}")
if op == "of_type":
validate(ir.get("seed", {}), path + ".seed", known_names)
return
if op in ("step", "closure"):
if ir.get("predicate") not in _PREDICATES:
raise PlanError(f"{path}.predicate: 未知の述語 '{ir.get('predicate')}'。"
f"使用可能:{sorted(_PREDICATES)}")
if ir.get("direction") not in ("forward", "reverse", "both"):
raise PlanError(f"{path}.direction: forward/reverse/both のいずれか")
validate(ir.get("seed", {}), path + ".seed", known_names)
return
if op in _SET_OPS:
args = ir.get("args")
if not isinstance(args, list) or len(args) < 2:
raise PlanError(f"{path}.args: 2つ以上のIRが必要")
for j, a in enumerate(args):
validate(a, f"{path}.args[{j}]", known_names)
return
# ---------------- 実行(決定的) ----------------
def _step(seed: set[str], pred: str, direction: str) -> set[str]:
out: set[str] = set()
for n in seed:
if direction in ("forward", "both"):
out |= {o for (p, o) in _OUT.get(n, []) if p == pred}
if direction in ("reverse", "both"):
out |= {s for (p, s) in _IN.get(n, []) if p == pred}
return out
def _closure(seed: set[str], pred: str, direction: str) -> set[str]:
result: set[str] = set()
stack = list(seed)
while stack:
cur = stack.pop()
for nxt in _step({cur}, pred, direction):
if nxt not in result:
result.add(nxt); stack.append(nxt)
return result
def execute(ir: dict, env: dict | None = None) -> set[str]:
env = env or {}
op = ir["op"]
if op == "vector":
raise PlanError("vector: 構造では実行不可(ベクトルRAGへ委譲する合図)")
if op == "ref":
return set(env[ir["name"]])
if op == "const":
return set(ir["ids"])
if op == "instances":
return set(_G.instances_of(ir["type"]))
if op == "of_type":
return {x for x in execute(ir["seed"], env) if _G.is_a(x, ir["type"])}
if op == "step":
return _step(execute(ir["seed"], env), ir["predicate"], ir["direction"])
if op == "closure":
return _closure(execute(ir["seed"], env), ir["predicate"], ir["direction"])
if op in _SET_OPS:
sets = [execute(a, env) for a in ir["args"]]
acc = set(sets[0])
for s in sets[1:]:
if op == "union":
acc |= s
elif op == "intersect":
acc &= s
else:
acc -= s
return acc
raise PlanError(f"未対応のop:{op}")
def run_plan(ir: dict) -> list[str]:
validate(ir)
return sorted(execute(ir))
def run_steps(steps: list[dict], result: dict) -> list[str]:
"""スケッチ誘導: 名前付きの浅い部分集合を順に構築し、最後に result を評価する。
steps=[{"name":..., "ir":<浅いIR>}...]。各irは先行stepを {"op":"ref","name":...} で参照可。"""
env: dict[str, set] = {}
for st in steps:
name, ir = st.get("name"), st.get("ir", {})
if not name:
raise PlanError("stepにnameが必要")
validate(ir, f"$.{name}", known_names=set(env))
env[name] = execute(ir, env)
validate(result, "$.result", known_names=set(env))
return sorted(execute(result, env))
5 問すべてについて、手書き IR は gold と厳密に一致します。
$ python - << 'PY'
from plan_ir import run_plan
print(run_plan({"op":"closure","seed":{"op":"const","ids":["payment-service"]},
"predicate":"depends_on","direction":"reverse"})) # Q1
PY
['api-gateway', 'order-service']
7.2 5 層の堅牢化（検証・修復・自己無矛盾・vector・スケッチ誘導）
検証を通っても、「有効だが意味が誤った IR」（Q5 で結合の向きを取り違えるなど）は残ります。これは、検証では防げない意味解析の残余誤差です。完成システム PerfectRAG では、次の 5 層で堅牢化します。
- スキーマ検証 — 未知の型/述語/id・不正な向きを弾く
- 修復ループ — 検証エラー文言を LLM に返して自己修正
-
自己無矛盾（多数決） —
n_samples回サンプリングし、実行結果集合の最頻値を採用 -
ベクトルフォールバック — 構造で表せない純粋な説明（
op:vector）は PlainRAG へ委譲 -
スケッチ誘導 — 単発 IR が空なら、浅い部分集合の列に分解して決定的に合成（
plan_ir.run_stepsとref）。弱いモデルに深い入れ子を一発で書かせない
"""あるべき姿の完成版(出荷する1システム)。
設計思想: LLMは『NL→検証可能な型付きIR』への意味解析だけを担い、
実行はグラフ上で決定的・厳密。堅牢性は次の4層で担保する。
1) スキーマ検証 … 未知の型/述語/id・不正な向きを含むIRを弾く
2) 修復ループ … 検証エラー文言をLLMへ返して自己修正させる(最大 repairs 回)
3) 自己無矛盾(多数決) … n_samples 回サンプリングし、実行結果集合の最頻値を採用
4) ベクトルфォールバック … 構造で表せない質問(op=vector や失敗継続)は PlainRAG へ委譲
検証を通ったIRの実行結果は正しいので、型名綴り違い・label/id取り違え・過剰包含は原理的に生じない。
"""
from __future__ import annotations
import json
from collections import Counter
from kb import ENTITY_BY_ID, CLASS_HIERARCHY, RELATIONS, ENTITIES
from plan_ir import validate, execute, run_steps, PlanError
from llm import chat
from rag_plain import PlainRAG
_PREDS = sorted({p for _s, p, _o in RELATIONS})
def _schema() -> str:
types = ", ".join(CLASS_HIERARCHY)
ids = ", ".join(e.id for e in ENTITIES)
return (f"型(下位含む):{types}\n述語:{', '.join(_PREDS)}\n"
f"id:{ids}\n"
"型階層: Component⊃Service⊃{EdgeService,CoreService,SupportingService}, "
"Component⊃Datastore⊃{RelationalDB,CacheStore,MessageQueue}, Team")
_GRAMMAR = (
"IRの代数(各ノードはidの集合を返す。JSONのみ返す):\n"
'{"op":"const","ids":[...]} リテラル集合\n'
'{"op":"instances","type":"型"} その型(下位含む)の全インスタンス\n'
'{"op":"step","seed":IR,"predicate":"述語","direction":"forward|reverse|both"} 1ホップ\n'
'{"op":"closure","seed":IR,"predicate":"述語","direction":"..."} 推移閉包(影響波及等)\n'
'{"op":"of_type","seed":IR,"type":"型"} 集合を型で絞る\n'
'{"op":"union|intersect|difference","args":[IR,IR,...]}\n'
'{"op":"vector"} 構造で表せない(定義/説明)ときの委譲合図\n'
"向きの定義: forward=(seed)-[:述語]->(結果), reverse=(結果)-[:述語]->(seed)。"
"『Xの停止で影響を受ける=Xに推移的に依存する主体』は "
'closure(const[X], depends_on, reverse)。'
"特定の既知idについて役割やidを問う照会は const[その id] で表す(vectorにしない)。"
)
_FEWSHOT = (
"例1『データストアを全部』→ {\"op\":\"instances\",\"type\":\"Datastore\"}\n"
"例2『order-service が停止したら影響を受けるものは』→ "
"{\"op\":\"closure\",\"seed\":{\"op\":\"const\",\"ids\":[\"order-service\"]},"
"\"predicate\":\"depends_on\",\"direction\":\"reverse\"}\n"
"例3『キャッシュに保存しているサービスは』→ "
"{\"op\":\"of_type\",\"type\":\"Service\",\"seed\":{\"op\":\"step\","
"\"predicate\":\"persists_to\",\"direction\":\"reverse\","
"\"seed\":{\"op\":\"instances\",\"type\":\"CacheStore\"}}}\n"
"例4『auth-service の役割とidを答えて』→ {\"op\":\"const\",\"ids\":[\"auth-service\"]}\n"
"例5『Aチームが所有するサービスが依存する、A以外が所有するもの』→ "
"{\"op\":\"difference\",\"args\":["
"{\"op\":\"step\",\"predicate\":\"depends_on\",\"direction\":\"forward\","
"\"seed\":{\"op\":\"of_type\",\"type\":\"Service\","
"\"seed\":{\"op\":\"step\",\"predicate\":\"owned_by\",\"direction\":\"reverse\","
"\"seed\":{\"op\":\"const\",\"ids\":[\"A-team\"]}}}},"
"{\"op\":\"step\",\"predicate\":\"owned_by\",\"direction\":\"reverse\","
"\"seed\":{\"op\":\"const\",\"ids\":[\"A-team\"]}}]}\n"
"例6『RAGとは何かを説明して(idの答えなし)』→ {\"op\":\"vector\"}"
)
_SYS = ("あなたは自然言語を、指定された型付きIR(JSON)へ変換する意味解析器です。"
"スキーマに無い型/述語/idは使わないこと。JSONのオブジェクトだけを返し、"
"説明やコードフェンスは付けない。"
"重要: 答えが『特定エンティティのid集合』になるなら(単一idの照会も含めて)必ず構造化プランを使う"
"(単一idの照会は const で表す)。vector は、答えがidにならない純粋な説明・定義のときだけ。")
def _extract_json(text: str) -> dict | None:
s, e = text.find("{"), text.rfind("}")
if s == -1 or e < s:
return None
try:
return json.loads(text[s:e + 1])
except json.JSONDecodeError:
return None
_STAGED_SYS = ("複雑な結合/差集合の質問を、浅い部分集合の列に分解してJSONで返す(説明不要)。"
"各stepは name と ir を持ち、irは const/instances/step/closure/of_type と "
'{"op":"ref","name":先行step名} だけで構成する(深い入れ子にしない)。'
"最後に result(union/intersect/difference と ref の組み合わせ、または単一ref)を書く。"
"スキーマ外の型/述語/idは使わない。JSONのみ。")
_STAGED_FEWSHOT = (
'例『Aチーム所有サービスが依存する、A以外所有のもの』→\n'
'{"steps":['
'{"name":"owned","ir":{"op":"step","predicate":"owned_by","direction":"reverse",'
'"seed":{"op":"const","ids":["A-team"]}}},'
'{"name":"svc","ir":{"op":"of_type","type":"Service","seed":{"op":"ref","name":"owned"}}},'
'{"name":"deps","ir":{"op":"step","predicate":"depends_on","direction":"forward",'
'"seed":{"op":"ref","name":"svc"}}}],'
'"result":{"op":"difference","args":[{"op":"ref","name":"deps"},{"op":"ref","name":"owned"}]}}'
)
def _extract_obj(text: str):
s, e = text.find("{"), text.rfind("}")
if s == -1 or e < s:
return None
try:
return json.loads(text[s:e + 1])
except json.JSONDecodeError:
return None
class PerfectRAG:
def __init__(self, k: int = 4, n_samples: int = 1, repairs: int = 2):
self.vector = PlainRAG(k=k)
self.n_samples = n_samples
self.repairs = repairs
def _one_plan(self, question: str, temperature: float) -> tuple[dict | None, str]:
"""1サンプル: 生成→検証→(失敗なら)修復ループ。valid planか None を返す。"""
err = None
for _ in range(self.repairs + 1):
hint = f"{_schema()}\n\n{_GRAMMAR}\n\n{_FEWSHOT}\n\n質問:{question}\n"
if err:
hint += f"\n前回のIRは無効:{err}\n修正したJSONを返す。"
ir = _extract_json(chat(_SYS, hint + "\nJSON:", temperature=temperature))
if ir is None:
err = "JSONとして解析不能"; continue
try:
validate(ir)
return ir, ""
except PlanError as pe:
err = str(pe)
return None, err or "unknown"
def _staged(self, question: str) -> list[str] | None:
"""スケッチ誘導: 浅い部分集合の列へ分解して段階的に実行。失敗なら None。"""
err = None
for _ in range(self.repairs + 1):
hint = f"{_schema()}\n\n{_STAGED_FEWSHOT}\n\n質問:{question}\n"
if err:
hint += f"\n前回は無効:{err}\n修正したJSONを返す。"
obj = _extract_obj(chat(_STAGED_SYS, hint + "\nJSON:", temperature=0.0))
if not obj or "steps" not in obj or "result" not in obj:
err = "steps/result を持つJSONが必要"; continue
try:
got = run_steps(obj["steps"], obj["result"])
if got:
return got
err = "結果が空"
except PlanError as pe:
err = str(pe)
return None
def answer(self, question: str) -> tuple[list[str], str]:
results: list[frozenset] = []
vector_votes = 0
traces: list[str] = []
for i in range(self.n_samples):
temp = 0.0 if self.n_samples == 1 else 0.4
ir, err = self._one_plan(question, temp)
if ir is None:
traces.append(f"s{i}:invalid({err})"); continue
if ir.get("op") == "vector":
vector_votes += 1; traces.append(f"s{i}:vector"); continue
try:
results.append(frozenset(execute(ir)))
traces.append(f"s{i}:ok")
except PlanError as pe:
traces.append(f"s{i}:execfail({pe})")
# 1) 単発IRの自己無矛盾: 非空の最頻結果があれば採用
nonempty = [r for r in results if r]
if nonempty:
best, _ = Counter(nonempty).most_common(1)[0]
return sorted(best), f"[plan{'|'.join(traces)}] → structured"
# 2) 純粋な説明(vector優勢)ならベクトルへ
if vector_votes > len(results):
pred, _ = self.vector.answer(question)
return pred, f"[plan{'|'.join(traces)}] → vector fallback"
# 3) 構造化できるが単発では空 → スケッチ誘導(段階分解)を試す
staged = self._staged(question)
if staged:
return staged, f"[plan{'|'.join(traces)}] → staged"
# 4) それでも無理 → ベクトル
pred, _ = self.vector.answer(question)
return pred, f"[plan{'|'.join(traces)}] → vector fallback"
7.3 実測
gpt-5.6：macro-F1 = 1.000（全問。Q4 は const['auth-service'] として構造化）。
gemma4:e2b：macro-F1 = 0.800 — これまで作った全システムの弱モデル最高値（Onto 0.480 / Router 0.275 / Cypher 0.400 を上回る）。
===== PerfectRAG =====
QID kind P R F1 pred
Q1 multihop 1.00 1.00 1.00 ['api-gateway', 'order-service'] (structured)
Q2 type 1.00 1.00 1.00 [5 datastores] (structured)
Q3 join 1.00 1.00 1.00 ['order-service','payment-service','user-service']
Q4 single 1.00 1.00 1.00 ['auth-service'] (structured/const)
Q5 join 0.00 0.00 0.00 [] (e2b のみ)
macro-F1 = 0.800 (e2b) / 1.000 (gpt-5.6)
e2b に残る Q5 は、検証を通る有効 IR だが意味が誤っているという既約な残余です。複雑な差集合結合を組めず、空になっています。スケッチ誘導（浅いステップへの分解）は合成負荷を大きく下げます。決定的検証では「単発空 → 段階分解 → gold」を確認済みです。ただし、弱いモデルが各ステップの浅い IR を正しく書けるかは依然としてモデル依存であり、保証はできません。--samples を増やすと、確率的に改善します。
数値の扱い：
gemma4:e2bは非決定的で、単発評価は ±0.2〜0.3 程度は普通に振れます。run_perfect.py --samples 3 --runs 5のように、自己無矛盾＋反復平均で見るのが安全です。gpt-5.6はtemperature=0で決定的なので、数値を素直に読めます。
8. 総括 — 「失敗の圧縮」という視点
応用編で行ったことは、一言で言えば、失敗クラスを段階的に潰し込むことでした。
| 段階 | 失敗の主因 | 対処 |
|---|---|---|
| プレーン RAG | 型・経路・結合が構造的に解けない | — |
| オントロジー併用（基礎編） | 到達可能化したが precision と追跡が LLM 依存で揺れる | 被覆の保証 |
| 改修A/B | 後段処理が破壊的で回帰 | 非破壊ガード |
| Text2Cypher / SQL | フリーテキストのクエリ生成が壊れる | 決定的実行エンジン |
| 単一選択ルーター | 誤ルーティングで単体最良より劣化 | フォールバック・カスケード |
| 検証可能 IR（あるべき姿） | 失敗を「意味解析の誤り一種類」に圧縮 | 検証・修復・自己無矛盾・スケッチ誘導 |
到達点は次のとおりです。「LLM に自由に書かせる」箇所をゼロに近づけ、決定的に計算できる部分を決定的な処理へ寄せると、失敗は「意味解析で IR を取り違える」一種類まで縮みます。強いモデルならほぼ天井に届きます（gpt-5.6 で 1.000）。弱いモデルでも従来最良を上回ります（e2b で 0.800）。
残った一種類の誤差を厳密に 0 にするのは、モデル・デコード層の仕事です。制約付きデコードで IR / JSON を文法拘束する、より強いモデルを使う、複数経路の相互検証を挟む、といった対応が考えられます。いずれも「あるべき姿（アーキテクチャ）」の外側であり、アーキテクチャとしては本稿で完成しています。
基礎編の標語を、言い直して締めます。
オントロジーは答えを到達可能にします。検証可能な IR と決定的実行は、その答えを“確実に取り出し”ます。残るのは、質問を IR に写し取る意味解析だけです。