部品表を1本に組み立てる——借りる部品と自作部品で最小パイプラインを建て、引用ゲートが「幽霊の出典」を突き返すまで | yatmita テックラボ

前編で並べた『借りる部品（subagent/schema）』と『自作部品（外部ループ/決定論ゲート/state外部化）』を、実際に ingest→extract→compile→query の1本に組み立てる。資料を並列で取り込み、構造化して統合し、出典つきで答える。最後に引用真正性ゲートが、存在しない出典を fail-closed で突き返す。動く最小PoCの全コードを見せる。

前編で部品表を並べた。借りる部品（skill / MCP / subagent / Dynamic Workflows）と、自作する部品（外部ループ / 決定論ゲート / state 外部化）。表を眺めるだけなら「あとは組み合わせるだけ」と言える。だが「組み合わせるだけ」ほど嘘になりやすい言葉もない。だから実際に1本、動くものを建てる。

題材は、汎用基盤の核にある ingest → extract → compile → query ——資料群を読み込み、構造化して統合し、問い合わせに答えるパイプライン。リサーチ系・文書生成系の中身を一段抽象化した、最小の骨格だ。サンプルはドメイン知識の要らない観葉植物のケアノート5枚にした。コードは全部この記事に出すので、読めば手元に再構成できる。

完成図

部品の対応はこうだ。ingest と extract は 借りる部品（subagent と schema）、compile と query のゲート、そして外枠ループと state は 自作部品。

借りる部品：エージェントランタイムを1枚かます

まず「サブエージェント」を1つに定義する。このPoCでは 「サブエージェント＝claude -p を1回呼ぶこと」。Dynamic Workflows の agent() に当たるが、誰でも再現できるよう標準の claude -p で実装する。肝は、バックエンドを環境変数で差し替えられる薄い1枚をかますこと。

class Agent:
    def run(self, prompt, fixture_key=None, passthrough=None):
        if self.mode == "mock":              # API 不要・決定論的（配管の検証用）
            if passthrough is not None:
                return passthrough
            return FIXTURES[fixture_key]
        return self._claude(prompt)          # IECQ_AGENT=claude なら実モデル
 
    def _claude(self, prompt):
        r = subprocess.run(["claude", "-p", prompt], capture_output=True, text=True, timeout=180)
        return r.stdout.strip()

mock は固定応答を返すだけ（API 不要・決定論的）、claude は実際に claude -p を叩く。この1行の切り替えが、前編で固めた設計判断そのものだ——「フォーマット収束は schema に逃がし、配管は決定論で検証できるようにしておく」。パイプライン本体は backend を一切知らない。

fan-out：1資料 = 1ワーカーを並列に

ingest と extract は、資料ごとに独立している。だから 1資料＝1ワーカー で並列に流す。これが Dynamic Workflows の parallel を OS プロセスで再現した形だ。

def ingest_extract(doc_path, agent):
    raw = doc_path.read_text()
    # ① ingest（借りる: subagent）— 正規化。mock では原文を通す
    norm = agent.run(prompt("ingest.txt", doc=raw), passthrough=raw)
    # ② extract（借りる: schema）— 構造化抽出
    fact = parse_json(agent.run(prompt("extract.txt", schema=SCHEMA_JSON, note=norm),
                                fixture_key=f"extract:{doc_path.stem}"))
    fact["source"] = doc_path.stem           # 出典idはモデルに任せず機械的に確定
    ok, errs = validate(fact, SCHEMA)        # 自作: 決定論ゲート（形）
    return (fact, None) if ok else (None, (doc_path.stem, errs))
 
# fan-out
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as ex:
    for fact, gap in ex.map(lambda d: ingest_extract(d, agent), DOCS):
        ...

ひとつ細かいが効く判断がある。fact["source"] を モデルの出力からではなく、ファイル名から機械的に確定 している。出典 id をモデルに書かせると、後段の引用照合が「モデルの自己申告」になって意味を失う。出典は決定論で握る。

走らせると、並列が効いているのがログに出る。

$ python3 pipeline.py
{"event": "extract", "source": "01-pothos",       "ok": true}
{"event": "extract", "source": "05-zz-plant",     "ok": true}   ← 05 が 02 より先
{"event": "extract", "source": "04-calathea",     "ok": true}
{"event": "extract", "source": "02-spider-plant", "ok": true}
{"event": "extract", "source": "03-snake-plant",  "ok": true}

01, 05, 04, 02, 03 と順不同で完了しているのが、5資料が同時に走った証拠だ。投入順（01→05）ではなく、終わった順に記録されている。

自作部品：品質は reduce 点に置く

extract が返すのは、資料ごとにバラバラの構造化ファクトだ。これを compile（reduce）で1つに統合する。map-reduce 型パイプラインの弱点は 統合点に出る（前編で触れたとおり）ので、ゲートはここに置く。

def gate_reduce(facts, expected):
    errs = []
    if len(facts) != expected:                       # 全件揃ったか
        errs.append(f"fact count {len(facts)} != {expected}")
    seen = set()
    for f in facts:
        if f["source"] in seen:                      # 出典の重複がないか
            errs.append(f"duplicate source: {f['source']}")
        seen.add(f["source"])
    return len(errs) == 0, errs

地味だが、「5枚入れたのに4ファクトしか無い」「同じ出典が二重に入った」をここで止めないと、後段は静かに壊れたデータで答えてしまう。

目玉：引用真正性ゲート（照合）

最後の query。統合済みファクトだけを使って質問に答えさせる。質問は「ペットに安全で、かつ低光量に耐える植物は？」。答えはファクトを組み合わせないと出ない（pet_safe と light の2軸）。

ここに、このPoC最大の自作部品を置く。答えが挙げた出典が、本当に存在するソースを指しているかを照合するゲートだ。

def gate_query(ans, known_sources):
    cites = ans.get("citations")
    if not isinstance(cites, list) or not cites:
        return False, ["citations missing or empty"]          # 引用なし → fail-closed
    errs = []
    for c in cites:
        if c not in known_sources:                            # 実在ソース集合と照合
            errs.append(f"hallucinated citation: {c}")
    return len(errs) == 0, errs

正常系はこうなる。

answer: Calathea is both pet-safe and tolerates low to medium light. ...
citations: ['04-calathea']
--> PASS

Calathea（04）は pet_safe かつ low-medium。正しく1件に絞り、出典も実在する。では、答えが存在しない出典を引いたらどうなるか。ゲートに直接食わせて確かめる。

実在する出典のみ ['04-calathea']        → PASS
存在しない出典   ['99-ghost-orchid']     → FAIL  hallucinated citation: 99-ghost-orchid
引用なし         []                       → FAIL  citations missing or empty

幽霊の出典 99-ghost-orchid は突き返される。 引用が無い答えも突き返される。これが fail-closed の意味だ——「もっともらしいが存在しない出典」を、内容の良し悪しを判定する前に、機械的に止める。出典が根拠になるドメインでは、この継ぎ目1本の有無が品質を分ける。borrowed な schema は「citations が文字列配列である」ことまでは保証してくれるが、「その中身が実在する」ことは保証しない。そこは自作するしかない。

外枠：state とループ

ここまでの各ステップは、結果を state/ のファイルに外部化している。log.jsonl（イベント列）、compiled.json（統合結果）、gap.md（ゲートに突き返された箇所）。エージェントの文脈内に記憶を持たせず、外に出す。これがあるから途中で落ちても再開でき、どの入力からどう出たかの監査証跡が残る。

そして全体を包む外枠は、最小版では「1周だけ走り、ゲートが落ちたら非ゼロ終了する」fail-closed の薄いループだ。

set -euo pipefail
python3 pipeline.py        # ゲート失敗で exit 1 → ここが ralph のフック点

本番では、この外枠を ralph-loop に置き換える。gap.md（ゲート失敗）を次イテレーションの入力にして、DONE まで反復させる。Dynamic Workflows が「幅（並列）」を編み、この外枠が「深さ（反復して良くする）」を司る——前編の対応表が、そのままコードの分担になっている。

組み上がって見えたこと

部品表は、動かすと表のままではいられない。組み立てて初めて分かったことが3つある。

出典は決定論で握る。source をモデルに書かせず、ファイル名から確定したからこそ、引用照合が意味を持った。借りる部品（schema）と自作部品（照合ゲート）の継ぎ目は、決定論側が握らないと緩む。
mock/claude の切り替えが設計判断の実装だった。「配管は決定論で検証、収束は schema に逃がす」が、Agent クラス1枚に落ちた。
品質を分けたのは派手な部分ではなかった。fan-out でも LLM の賢さでもなく、reduce 点の件数チェックと、query の引用照合という、地味な自作ゲート2本だった。

借りる部品は強力だが、借り物が供給しない穴——「何を検査するか」「出典は実在するか」「ループの記憶」——を自作で埋めて初めて、無人で回る基盤になる。前編の部品表は、こうして1本の動く実物になった。便利ツールの寄せ集めではなく、世界を上書きするための部品表だと示せるのは、こうやって組み上げて、幽霊の出典が突き返される瞬間を見せられる場所だ。

（このキットは当面リポジトリとしては公開しない。コードは上にすべて出したので、5ファイルほどで手元に再構成できる。）