Day 21 已完成平面式檢索增強生成（Flat Retrieval-Augmented Generation, Flat RAG）：九筆公開知識都在同一層，每題直接排序九筆，再把前三筆證據交給生成模型。那條基準的六筆固定檢查全部通過，也留下了一個問題：當知識庫變大時，每個問題是否都要和所有原子規則直接比較？

今天建立階層式檢索增強生成（Hierarchical Retrieval-Augmented Generation, Hierarchical RAG）。它先搜尋較粗的父層（Parent）主題，再展開被選中的子規則（Child），最後才找出真正要交給模型的證據。這很像先找到書架，再從書架抽出書；理論上能縮小候選，但如果一開始走錯書架，正確規則連參加第二輪排序的機會都沒有。

這篇不預設 Hierarchical RAG 一定更好。排序後保留的前 k 筆（Top-k）是實際進入下一步的候選；我們沿用 Day 21 的六筆作者自寫公開知識問題、查詢表示、向量模型、最終 Top-3、生成模型、系統提示詞（System Prompt）與輸出結構，只把檢索候選的組織方式改成「Parent Top-2 → Child Top-3」。

下圖是概念示意。請觀察左側 Flat RAG 直接搜尋九張規則卡，右側 Hierarchical RAG 則先經過三個父層資料夾，再取出三張子規則卡。最右側仍由醫療專業人員判讀，表示系統只是決策支援。

上圖沒有承載精確醫療規則或實驗數字。真正的父子關係、選取數量、相似度、輸出與失敗案例都由設定檔和執行結果固定。

本篇仍只處理公開的韓國急診檢傷與急迫度分級量表（Korean Triage and Acuity Scale, KTAS）摘要，不讀取 Kaggle 病患列、護理師登錄級數或專家重新判定級數。六題結果是小型工程比較，不是病患五級檢傷效能、醫療正確性或臨床安全證據。

今天會完成什麼

完成本篇後，你會得到六項可檢查產物：

1. 一份鎖定 Day 21 共同比較條件與 Day 22 階層路由的 JavaScript 物件表示法（JavaScript Object Notation, JSON）契約。
2. 一個驗證三個 parent、九個 child，以及 child 與 flat 逐筆同源的核心模組。
3. 一條先排 Parent Top-2、展開 children、再排 Child Top-3 的完整 RAG。
4. 十項階層結構單元測試，以及全專案九十項離線測試。
5. 六題 Flat／Hierarchical 同次配對執行的公開結果與執行清單（Run Manifest）。
6. 一筆成功路由和一筆失敗路由，能說明父層錯誤為何會造成不可逆漏失。

今天建立的階層管線稱為 B6-hierarchical。它不是為了取代 B5-flat，而是提供一個可歸因的結構比較。

本篇會用到的名詞

本篇會同時談父子知識圖、兩階段檢索與配對比較，先建立詞彙索引。正文仍會在第一次使用時補充用途。

Hierarchical RAG 不是把檔案放進資料夾就完成

檢索增強生成（Retrieval-Augmented Generation, RAG）先從外部知識庫取回相關內容，再把內容交給大型語言模型（Large Language Model, LLM）產生答案。階層式檢索增強生成的差異在於：檢索不只做一次，而是先做粗粒度路由，再做細粒度取證。

本篇把知識分成兩種角色：

• Parent 是父層主題。它的 context_text 由同群組 child 摘要組成，只參與第一階段 Dense 排名。
• Child 是可引用的原子規則。它保留和 Flat RAG 完全相同的 rule_id、search_text、文字雜湊與 metadata，只有額外的 parent_id 關係。

父層不是新的醫療規則，也不會直接成為答案引用。它只回答「下一步應該到哪幾個群組找」。真正進入最終 Top-3 的仍是 child。

三個 parent 如何收納九個 child

Day 17 已建立以下父子關係。每個 child 恰好屬於一個 parent，不允許同一規則同時被兩個父層重複收納。

韓國 KTAS 官方介紹把制度描述為症狀導向的分類工具，公開流程包含第一印象、感染相關基本詢問與檢查、主要與次要考量，再決定嚴重度與急迫度；同頁也列出五級公開定義：KTAS 官方介紹。Moon 等人的第一手研究則描述生命徵象、疼痛、出血與受傷機轉等主要考量，以及血糖和脫水等次要考量例子：Moon et al., 2019。

這些來源支持本篇九筆摘要的主題，但不等於公開了完整 KTAS 主訴目錄、逐主訴門檻或套用順序。階層分組只能整理現有知識，不能補出不存在的規則。

這次的主要變因是階層結構

公平比較的核心不是讓兩個系統「看起來差不多」，而是明確列出共同條件與唯一主要變因。

最佳匹配 25（Best Matching 25, BM25）是以字詞訊號排序的方法；重排序器（Reranker）會再替初步候選評分；檢索門控（Retrieval Gate）則依額外條件限制候選。本篇三者在兩邊都關閉，避免它們成為混雜因素。

稠密檢索（Dense Retrieval）用向量嵌入（Vector Embedding）表示問題與候選，再以餘弦相似度（Cosine Similarity）排序。餘弦分數較高只代表這個 embedding 模型認為文字語意較接近，不是相關機率、答案正確率或臨床安全分數。

前 k 筆（Top-k）是排序後保留的前 k 筆。本篇固定 Parent Top-2，表示三個父層只展開前兩名；接著固定 Child Top-3，讓兩條 pipeline 最終都交付三筆原子證據。Parent Top-2 是事前寫入設定的工程選擇，不是看完六題結果後調出來的最佳值。

九筆 child 的 search_text 和 Flat 完全相同，執行程式也只建立一組 child 向量，再用 rule_id 映射到階層管線，避免同一文字重複呼叫 embedding 造成不必要差異。需要誠實保留的一個結構差異是：Flat 的 EVIDENCE 是單層清單，Hierarchical 會在三筆 child 外加 parent title 並分組。system prompt、模型與 schema 沒變，但完整 user message 字串並不相同。

下圖由左至右呈現完整資料流。請注意 parent 只用來決定候選集合；生成模型看不到未進入最終 Top-3 的 sibling 文字。

上圖中三個 parent 分別收納 5、2、2 筆 child。任取兩個父層，理論上會展開 4 或 7 筆 child；本次六題實際都選到「五級公開定義」與「定位／高階流程」，所以每題都展開 7 筆。這是執行後的觀察，不是寫死在演算法中的數量。

從上一步輸出到下一步輸入

兩階段流程需要清楚交代中間轉換，否則「階層檢索」容易變成無法追查的黑箱。

1. 建立查詢向量：讀取一筆作者自寫問題，加上 Day 18 的固定 query instruction，輸出 2,560 維查詢向量。
2. 排列 parent：把查詢向量和三筆 context_text 向量比較，輸出含 parent_id、cosine 與名次的完整父層排名。
3. 截取 Parent Top-2：保留前兩個 parent_id。這一步的輸出就是下一步 metadata filter 的輸入。
4. 展開 child：依 parent.child_rule_ids 取回兩個父層的所有 child，並保留父層名次與分數作為 trace。
5. 重新排列 child：只在展開集合中，以同一查詢向量和 child 向量計算 Dense 排名。
6. 截取 Child Top-3：保留三筆原子規則，加入 rule_id、來源、完整度、child cosine 與父層 trace。
7. 組成 EVIDENCE：依 parent_id 分組三筆 child，但不加入未入選 sibling 的 context_text。
8. 結構化生成：沿用 Day 21 system prompt、模型、seed、溫度與逐題 JSON Schema，輸出狀態、答案、引用與缺少資訊。
9. 配對檢查：同一題逐一比較 Flat 與 Hierarchical 的必要證據命中、引用、答案狀態及最終規則順序。

中繼資料（Metadata）在第 3 到第 7 步負責接起父子關係。parent_id 是路由鍵，rule_id 是最終引用鍵；兩者用途不能互換。若把 parent title 當成證據，模型就可能引用一個沒有原子主張的群組名稱。

為什麼不把整段 parent context 一起交給模型

如果選到一個含五筆 child 的 parent，卻把整段 context_text 全部交給模型，表面上仍叫 Top-3，實際可見規則卻超過三筆。這會同時改變證據數量、文字長度與干擾內容，失去公平比較。

因此本篇只把 parent title 當分組標籤。模型可見的規則文字永遠恰好三筆，而且 evidence_rule_ids 仍只能引用這三筆 child 的 rule_id。

這個選擇保留了階層脈絡，也帶來歸因限制：當兩邊最終 child 完全相同、答案措辭卻不同時，不能說是檢索排序造成，因為 EVIDENCE 的序列化外形也不同。若要拆開這兩個效果，後續消融應加入「使用階層路由但攤平成 Flat evidence」或「不做父層路由、只把同一 Top-3 依 parent 分組」的控制條件。

專案資料夾與檔案如何分工

Day 22 會沿用 Day 21 已建立的生成與 Flat RAG 檔案，再新增階層契約、核心模組、執行入口與測試。

Day 21 的 Ollama 生成用戶端、Flat RAG 核心與提示詞都是已完成前置產物：它們接收問題與 Top-3 evidence，輸出固定七欄位答案。本篇不修改這些檔案；若你的專案缺少它們，請先完成 Day 21 的完整檔案建立與十四項單元測試。

先在自己的專案資料夾建立本篇完整檔案

接下來不會要求你前往任何程式碼網站。請在自己的電腦開啟專案資料夾，依下列順序建立檔案；每個程式碼區塊都是該檔案的完整內容，不含省略號。

本篇沿用 Day 13 的 Poetry 與重現性工具、Day 17 的九筆 flat chunks／九筆 hierarchical children／三筆 parents、Day 18 的 embedding 設定與本機 qwen3-embedding:4b，以及 Day 21 已完整建立的六題、system prompt、Flat RAG、Ollama 生成用戶端與 qwen3.5:4b-mlx。以下是 Day 22 新增或修改後的完整階層契約、RAG 套件入口、階層核心、配對執行入口與測試；公開摘要、完整配對結果與 run manifest 都由執行入口自動產生，不需要手動建立。

先從專案根目錄建立需要的資料夾：

mkdir -p configs/rag src/triage_rag/rag scripts tests results/public results/runs/day-22

如果指令沒有印出訊息是正常的。可用 test -d 資料夾路徑 && echo "資料夾已建立" 驗證單一資料夾。接著使用你熟悉的文字編輯器新增各檔案，把對應區塊完整貼入後儲存。

檔案 1：建立 configs/rag/day-22-hierarchical-rag.json

鎖定 Day 21 上游雜湊與共同比較條件、三個 parent、九個 child、Parent Top-2、Child Top-3、輸出與知識邊界。

請在文字編輯器建立 configs/rag/day-22-hierarchical-rag.json，貼入以下完整內容並儲存：

{
  "schema_version": 1,
  "experiment_id": "day-22-flat-vs-hierarchical-public-knowledge",
  "scope": "paired_author_written_public_knowledge_rag_structure_comparison_not_patient_triage_or_clinical_evaluation",
  "sources": {
    "day21_config_path": "configs/rag/day-21-flat-basic-rag.json",
    "day21_config_sha256": "d65b510e96f39a05d008a274654d63baa76e2e9308c44f74ce5002797ed4c838",
    "flat_chunks_path": "data/knowledge/ktas-public-v1/day-17/flat-chunks.jsonl",
    "flat_chunks_sha256": "10749728567e36c565d9e2e0fb931c1ece7da448bd1c04f4527b624e8507a1fa",
    "hierarchical_parents_path": "data/knowledge/ktas-public-v1/day-17/hierarchical-parents.jsonl",
    "hierarchical_parents_sha256": "18d26af4e71bf1af682ec5858bd712fa4c3bd6ff8e88a20b2566324e6dc16664",
    "hierarchical_children_path": "data/knowledge/ktas-public-v1/day-17/hierarchical-children.jsonl",
    "hierarchical_children_sha256": "ee2732539bea1b3a21aa26c66089d1cbd5399a529e70307271ddb69bf9115d28",
    "embedding_config_path": "configs/representation/day-18-numeric-semantics.json",
    "embedding_config_sha256": "3d24a3688dbff4e617270f6fa9e953c055c835c4b0b10718eb6fa5a5ef3ada8b",
    "model_lock_path": "configs/models/local-ollama-model-lock.json",
    "model_lock_sha256": "0cd15cb4d7e735a19cbe439fe9a9344f717029694a4441764f67e396033cc7f4",
    "system_prompt_path": "prompts/day-21-flat-basic-rag-system.txt",
    "system_prompt_sha256": "e826351ce89d5a68427b42916b9f17d2f4931a886c71e6322dd5f7d508f55c27"
  },
  "fixed_comparison_contract": {
    "question_source": "reuse_day21_cases_without_modification",
    "expected_case_count": 6,
    "expected_supported_case_count": 4,
    "expected_gap_case_count": 2,
    "query_representation": "author_written_question_with_day18_dense_query_instruction",
    "embedding_model": "qwen3-embedding:4b",
    "final_top_k": 3,
    "generator_model": "qwen3.5:4b-mlx",
    "system_prompt": "reuse_day21_prompt_without_modification",
    "output_schema": "reuse_day21_per_case_json_schema",
    "temperature": 0,
    "seed": 20260811,
    "bm25_enabled": false,
    "reranker_enabled": false,
    "retrieval_gate_enabled": false,
    "patient_rows_used": false,
    "reference_labels_used": false
  },
  "flat_pipeline": {
    "pipeline_id": "B5-flat",
    "candidate_unit": "atomic_child_rule",
    "candidate_count": 9,
    "retrieval_method": "dense_cosine",
    "final_top_k": 3,
    "hierarchy_enabled": false
  },
  "hierarchical_pipeline": {
    "pipeline_id": "B6-hierarchical",
    "parent_candidate_unit": "grouped_public_summary_context",
    "expected_parent_count": 3,
    "parent_retrieval_method": "dense_cosine",
    "parent_top_k": 2,
    "child_expansion": "all_children_of_selected_parents",
    "child_candidate_unit": "same_atomic_rules_as_flat",
    "expected_child_count": 9,
    "child_ranking_method": "dense_cosine",
    "final_top_k": 3,
    "generation_context_policy": "group_final_three_children_by_parent_title_without_exposing_unselected_sibling_text",
    "bm25_enabled": false,
    "reranker_enabled": false,
    "retrieval_gate_enabled": false
  },
  "outputs": {
    "public_summary_path": "results/public/day-22-hierarchical-rag.json",
    "run_output_root": "results/runs/day-22",
    "success_trace_case_id": "supported-workflow",
    "failure_trace_case_id": "gap-complete-thresholds"
  },
  "limitations": [
    "六筆作者自寫問題與九筆公開摘要只能驗證小型端到端結構比較，不是病患五級檢傷評估。",
    "Parent Top-2 是本篇固定的工程選擇，尚未以獨立驗證集調參，也不代表最佳設定。",
    "父層文字由既有 child 摘要組合而成；候選縮減、必要證據命中與生成輸出變化只能歸因到這個特定階層設計。",
    "結構與字串檢查不能取代多位專家對答案語意、臨床正確性與安全性的審查。",
    "知識庫沒有完整主訴目錄、逐主訴數值門檻或官方版本沿革；階層檢索不能補出不存在的知識。"
  ]
}

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 2：建立 src/triage_rag/rag/__init__.py

更新 RAG 子套件入口，同時公開 Day 21 flat 與 Day 22 hierarchical 驗證、排序、展開、context 與訊息函式。

請在文字編輯器建立 src/triage_rag/rag/__init__.py，貼入以下完整內容並儲存：

"""Retrieval-augmented generation pipelines."""

from triage_rag.rag.flat import (
    FlatRagContractError,
    build_context,
    build_user_message,
    evaluate_answer,
    validate_flat_contract,
    validate_public_cases,
)
from triage_rag.rag.hierarchical import (
    HierarchicalRagContractError,
    build_hierarchical_context,
    build_hierarchical_user_message,
    expand_selected_children,
    expected_parent_ids,
    group_context_for_generation,
    rank_expanded_children,
    rank_parents,
    validate_hierarchical_contract,
)

__all__ = [
    "FlatRagContractError",
    "build_context",
    "build_user_message",
    "evaluate_answer",
    "validate_flat_contract",
    "validate_public_cases",
    "HierarchicalRagContractError",
    "build_hierarchical_context",
    "build_hierarchical_user_message",
    "expand_selected_children",
    "expected_parent_ids",
    "group_context_for_generation",
    "rank_expanded_children",
    "rank_parents",
    "validate_hierarchical_contract",
]

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 3：建立 src/triage_rag/rag/hierarchical.py

驗證父子圖與 child／flat 同源，實作 parent 排名、Top-2 展開、child 重排、Top-3 context 及不洩漏 sibling 的 evidence 分組。

請在文字編輯器建立 src/triage_rag/rag/hierarchical.py，貼入以下完整內容並儲存：

"""Parent-first hierarchical dense retrieval over the Day 17 public corpus."""

from __future__ import annotations

import json
from collections import Counter
from typing import Any, Dict, List, Mapping, Sequence

from triage_rag.representation.embedding import cosine_similarity


JsonObject = Dict[str, Any]


class HierarchicalRagContractError(ValueError):
    """Raised when the fixed Day 22 hierarchy or comparison contract drifts."""


def validate_hierarchical_contract(
    parents: Sequence[Mapping[str, Any]],
    children: Sequence[Mapping[str, Any]],
    flat_documents: Sequence[Mapping[str, Any]],
    config: Mapping[str, Any],
) -> JsonObject:
    """Validate the parent graph and exact child-to-flat parity before retrieval."""

    pipeline = config["hierarchical_pipeline"]
    fixed = config["fixed_comparison_contract"]
    required = {
        "parent_retrieval_method": "dense_cosine",
        "parent_top_k": 2,
        "child_expansion": "all_children_of_selected_parents",
        "child_candidate_unit": "same_atomic_rules_as_flat",
        "child_ranking_method": "dense_cosine",
        "final_top_k": 3,
        "bm25_enabled": False,
        "reranker_enabled": False,
        "retrieval_gate_enabled": False,
    }
    mismatches = {
        key: {"actual": pipeline.get(key), "expected": expected}
        for key, expected in required.items()
        if pipeline.get(key) != expected
    }
    if fixed.get("final_top_k") != 3:
        mismatches["fixed_final_top_k"] = {
            "actual": fixed.get("final_top_k"),
            "expected": 3,
        }
    if mismatches:
        raise HierarchicalRagContractError(
            f"Day 22 階層比較條件被改動：{mismatches}"
        )

    expected_parent_count = int(pipeline["expected_parent_count"])
    expected_child_count = int(pipeline["expected_child_count"])
    if len(parents) != expected_parent_count:
        raise HierarchicalRagContractError(
            f"parent 應為 {expected_parent_count} 筆，實際為 {len(parents)} 筆"
        )
    if len(children) != expected_child_count or len(flat_documents) != expected_child_count:
        raise HierarchicalRagContractError(
            f"hierarchical child 與 flat 都應為 {expected_child_count} 筆"
        )

    parent_by_id: Dict[str, Mapping[str, Any]] = {}
    declared_child_ids: List[str] = []
    for parent in parents:
        parent_id = parent.get("parent_id")
        if not isinstance(parent_id, str) or not parent_id:
            raise HierarchicalRagContractError("parent 缺少 parent_id")
        if parent_id in parent_by_id:
            raise HierarchicalRagContractError(f"parent_id 重複：{parent_id}")
        if parent.get("store_type") != "hierarchical":
            raise HierarchicalRagContractError("parent store_type 必須是 hierarchical")
        if parent.get("retrieval_role") != "parent_context_only":
            raise HierarchicalRagContractError("parent 只能作為 parent_context_only")
        if not isinstance(parent.get("context_text"), str) or not parent["context_text"].strip():
            raise HierarchicalRagContractError(f"{parent_id} 缺少 context_text")
        child_rule_ids = parent.get("child_rule_ids")
        if not isinstance(child_rule_ids, list) or not child_rule_ids:
            raise HierarchicalRagContractError(f"{parent_id} 缺少 child_rule_ids")
        declared_child_ids.extend(str(rule_id) for rule_id in child_rule_ids)
        parent_by_id[parent_id] = parent

    duplicates = sorted(
        rule_id
        for rule_id, count in Counter(declared_child_ids).items()
        if count > 1
    )
    if duplicates:
        raise HierarchicalRagContractError(
            f"child 不可同時屬於多個 parent：{duplicates}"
        )

    child_by_rule: Dict[str, Mapping[str, Any]] = {}
    for child in children:
        metadata = child.get("metadata")
        rule_id = metadata.get("rule_id") if isinstance(metadata, dict) else None
        parent_id = child.get("parent_id")
        if not isinstance(rule_id, str) or not rule_id:
            raise HierarchicalRagContractError("child 缺少 rule_id")
        if rule_id in child_by_rule:
            raise HierarchicalRagContractError(f"child rule_id 重複：{rule_id}")
        if parent_id not in parent_by_id:
            raise HierarchicalRagContractError(
                f"{rule_id} 指向未知 parent：{parent_id}"
            )
        if child.get("store_type") != "hierarchical":
            raise HierarchicalRagContractError("child store_type 必須是 hierarchical")
        if child.get("retrieval_role") != "search_candidate":
            raise HierarchicalRagContractError("child 必須是 search_candidate")
        if rule_id not in parent_by_id[str(parent_id)]["child_rule_ids"]:
            raise HierarchicalRagContractError(
                f"{rule_id} 未登錄在 parent.child_rule_ids"
            )
        child_by_rule[rule_id] = child

    if set(declared_child_ids) != set(child_by_rule):
        raise HierarchicalRagContractError(
            "parent 宣告的 child 集合與 hierarchical children 不一致"
        )

    flat_by_rule = {
        str(item.get("metadata", {}).get("rule_id")): item
        for item in flat_documents
    }
    if set(flat_by_rule) != set(child_by_rule):
        raise HierarchicalRagContractError("flat 與 hierarchical child 的 rule_id 集合不同")
    parity_fields = ("search_text", "search_text_sha256")
    for rule_id, child in child_by_rule.items():
        flat = flat_by_rule[rule_id]
        for field in parity_fields:
            if child.get(field) != flat.get(field):
                raise HierarchicalRagContractError(
                    f"{rule_id} 的 {field} 與 flat 不一致"
                )
        if child.get("metadata") != flat.get("metadata"):
            raise HierarchicalRagContractError(
                f"{rule_id} 的 metadata 與 flat 不一致"
            )

    return {
        "parent_count": len(parents),
        "child_count": len(children),
        "unique_parent_count": len(parent_by_id),
        "unique_child_rule_id_count": len(child_by_rule),
        "each_child_has_exactly_one_parent": True,
        "child_rule_text_hash_and_metadata_match_flat": True,
        "parent_top_k": 2,
        "final_top_k": 3,
        "disabled_components": ["bm25", "reranker", "retrieval_gate"],
    }


def rank_parents(
    query_vector: Sequence[float],
    parents: Sequence[Mapping[str, Any]],
    parent_vectors: Sequence[Sequence[float]],
) -> List[JsonObject]:
    if len(parents) != len(parent_vectors):
        raise HierarchicalRagContractError("parent 與向量數量不同")
    ranked = [
        {
            "parent_id": parent["parent_id"],
            "title": parent["title"],
            "child_count": len(parent["child_rule_ids"]),
            "cosine_similarity": cosine_similarity(query_vector, vector),
        }
        for parent, vector in zip(parents, parent_vectors)
    ]
    ranked.sort(key=lambda item: (-item["cosine_similarity"], item["parent_id"]))
    for rank, item in enumerate(ranked, start=1):
        item["rank"] = rank
    return ranked


def expand_selected_children(
    parent_ranking: Sequence[Mapping[str, Any]],
    parent_by_id: Mapping[str, Mapping[str, Any]],
    child_by_rule: Mapping[str, Mapping[str, Any]],
    *,
    parent_top_k: int,
) -> List[JsonObject]:
    if parent_top_k != 2:
        raise HierarchicalRagContractError("Day 22 固定使用 Parent Top-2")
    if len(parent_ranking) < parent_top_k:
        raise HierarchicalRagContractError("parent 排名少於 Top-2")
    expanded = []
    for parent_hit in parent_ranking[:parent_top_k]:
        parent_id = str(parent_hit["parent_id"])
        if parent_id not in parent_by_id:
            raise HierarchicalRagContractError(f"未知 parent：{parent_id}")
        parent = parent_by_id[parent_id]
        for rule_id in parent["child_rule_ids"]:
            if rule_id not in child_by_rule:
                raise HierarchicalRagContractError(f"未知 child：{rule_id}")
            expanded.append(
                {
                    "rule_id": rule_id,
                    "parent_id": parent_id,
                    "parent_title": parent["title"],
                    "parent_rank": int(parent_hit["rank"]),
                    "parent_cosine_similarity": float(
                        parent_hit["cosine_similarity"]
                    ),
                }
            )
    return expanded


def rank_expanded_children(
    query_vector: Sequence[float],
    expanded_children: Sequence[Mapping[str, Any]],
    child_vector_by_rule: Mapping[str, Sequence[float]],
) -> List[JsonObject]:
    ranked = []
    for expanded in expanded_children:
        rule_id = str(expanded["rule_id"])
        if rule_id not in child_vector_by_rule:
            raise HierarchicalRagContractError(f"{rule_id} 缺少 child vector")
        ranked.append(
            {
                **dict(expanded),
                "cosine_similarity": cosine_similarity(
                    query_vector, child_vector_by_rule[rule_id]
                ),
            }
        )
    ranked.sort(key=lambda item: (-item["cosine_similarity"], item["rule_id"]))
    for rank, item in enumerate(ranked, start=1):
        item["rank"] = rank
    return ranked


def build_hierarchical_context(
    child_ranking: Sequence[Mapping[str, Any]],
    child_by_rule: Mapping[str, Mapping[str, Any]],
    *,
    top_k: int,
) -> List[JsonObject]:
    if top_k != 3:
        raise HierarchicalRagContractError("Day 22 固定使用 Child Top-3")
    if len(child_ranking) < top_k:
        raise HierarchicalRagContractError("展開後 child 少於 Top-3")
    context = []
    for hit in child_ranking[:top_k]:
        rule_id = str(hit["rule_id"])
        if rule_id not in child_by_rule:
            raise HierarchicalRagContractError(f"未知 child：{rule_id}")
        child = child_by_rule[rule_id]
        metadata = child["metadata"]
        context.append(
            {
                "rank": len(context) + 1,
                "rule_id": rule_id,
                "parent_id": hit["parent_id"],
                "parent_title": hit["parent_title"],
                "parent_rank": int(hit["parent_rank"]),
                "parent_cosine_similarity": round(
                    float(hit["parent_cosine_similarity"]), 8
                ),
                "topic": metadata["topic"],
                "completeness": metadata["completeness"],
                "source_id": metadata["source_id"],
                "source_url": metadata["source_url"],
                "cosine_similarity": round(float(hit["cosine_similarity"]), 8),
                "text": child["search_text"],
            }
        )
    return context


def group_context_for_generation(
    context: Sequence[Mapping[str, Any]],
) -> List[JsonObject]:
    """Keep parent membership while exposing exactly the final three child texts."""

    if len(context) != 3:
        raise HierarchicalRagContractError("生成 context 必須恰好包含三個 child")
    grouped: Dict[str, JsonObject] = {}
    order: List[str] = []
    for item in context:
        parent_id = str(item["parent_id"])
        if parent_id not in grouped:
            order.append(parent_id)
            grouped[parent_id] = {
                "parent_id": parent_id,
                "parent_title": item["parent_title"],
                "parent_rank": item["parent_rank"],
                "children": [],
            }
        grouped[parent_id]["children"].append(
            {
                "rank": item["rank"],
                "rule_id": item["rule_id"],
                "topic": item["topic"],
                "completeness": item["completeness"],
                "source_id": item["source_id"],
                "source_url": item["source_url"],
                "cosine_similarity": item["cosine_similarity"],
                "text": item["text"],
            }
        )
    return [grouped[parent_id] for parent_id in order]


def build_hierarchical_user_message(
    case_id: str,
    question: str,
    context: Sequence[Mapping[str, Any]],
) -> str:
    if not case_id.strip() or not question.strip():
        raise HierarchicalRagContractError("case_id 與 question 都不可為空")
    evidence = json.dumps(
        group_context_for_generation(context),
        ensure_ascii=False,
        sort_keys=True,
        indent=2,
    )
    return (
        f"CASE_ID:\n{case_id}\n\n"
        f"QUESTION:\n{question}\n\n"
        f"EVIDENCE:\n{evidence}"
    )


def expected_parent_ids(
    required_rule_ids: Sequence[str],
    child_by_rule: Mapping[str, Mapping[str, Any]],
) -> List[str]:
    parent_ids = []
    for rule_id in required_rule_ids:
        if rule_id not in child_by_rule:
            raise HierarchicalRagContractError(f"未知必要證據：{rule_id}")
        parent_id = str(child_by_rule[rule_id]["parent_id"])
        if parent_id not in parent_ids:
            parent_ids.append(parent_id)
    return parent_ids

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 4：建立 scripts/run_day22_hierarchical_rag.py

在同一次執行中配對跑 Flat 與 Hierarchical，驗證來源和模型、保存父層與 child trace、計算聚合差異並寫出公開摘要與 manifest。

請在文字編輯器建立 scripts/run_day22_hierarchical_rag.py，貼入以下完整內容並儲存：

#!/usr/bin/env python3
"""Run the paired Day 22 flat versus parent-first hierarchical RAG comparison."""

from __future__ import annotations

import argparse
import platform
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Mapping, Sequence

from triage_rag.generation import (
    OllamaChatError,
    build_response_schema,
    chat_structured,
    get_chat_runtime_metadata,
)
from triage_rag.knowledge_base.builder import read_jsonl
from triage_rag.rag import (
    build_context,
    build_hierarchical_context,
    build_hierarchical_user_message,
    build_user_message,
    evaluate_answer,
    expand_selected_children,
    expected_parent_ids,
    rank_expanded_children,
    rank_parents,
    validate_flat_contract,
    validate_hierarchical_contract,
    validate_public_cases,
)
from triage_rag.representation.embedding import (
    embed_texts,
    get_runtime_metadata,
    rank_documents,
    vector_sha256,
)
from triage_rag.reproducibility import (
    file_record,
    git_state,
    load_json,
    resolve_project_path,
    sha256_file,
    write_json,
)


PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONFIG = "configs/rag/day-22-hierarchical-rag.json"
EXPECTED_SCOPE = (
    "paired_author_written_public_knowledge_rag_structure_comparison_"
    "not_patient_triage_or_clinical_evaluation"
)
IMPLEMENTATION_PATHS = (
    "scripts/run_day22_hierarchical_rag.py",
    "src/triage_rag/generation/__init__.py",
    "src/triage_rag/generation/ollama_chat.py",
    "src/triage_rag/rag/__init__.py",
    "src/triage_rag/rag/flat.py",
    "src/triage_rag/rag/hierarchical.py",
    "src/triage_rag/representation/embedding.py",
    "src/triage_rag/reproducibility.py",
)
JsonObject = Dict[str, Any]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "以同一組六筆公開問題配對比較 Flat Dense Top-3 與 "
            "Parent Top-2 → Child Dense Top-3。"
        )
    )
    parser.add_argument("--config", default=DEFAULT_CONFIG)
    return parser.parse_args()


def _verify_sha256(path: Path, expected: str, label: str) -> None:
    actual = sha256_file(path)
    if actual != expected:
        raise ValueError(
            f"{label} SHA-256 不同；預期 {expected}，實際 {actual}。"
            "請先重建上游產物，或建立新版 Day 22 契約。"
        )


def _verify_sources(config: Mapping[str, Any]) -> List[Path]:
    sources = config["sources"]
    pairs = [
        ("day21_config_path", "day21_config_sha256", "Day 21 比較契約"),
        ("flat_chunks_path", "flat_chunks_sha256", "Day 17 flat chunks"),
        (
            "hierarchical_parents_path",
            "hierarchical_parents_sha256",
            "Day 17 hierarchical parents",
        ),
        (
            "hierarchical_children_path",
            "hierarchical_children_sha256",
            "Day 17 hierarchical children",
        ),
        (
            "embedding_config_path",
            "embedding_config_sha256",
            "Day 18 embedding 設定",
        ),
        ("model_lock_path", "model_lock_sha256", "本機模型鎖定"),
        ("system_prompt_path", "system_prompt_sha256", "Day 21 system prompt"),
    ]
    paths = []
    for path_key, hash_key, label in pairs:
        path = resolve_project_path(PROJECT_ROOT, str(sources[path_key]))
        _verify_sha256(path, str(sources[hash_key]), label)
        paths.append(path)
    return paths


def _verify_fixed_comparison(
    config: Mapping[str, Any],
    day21: Mapping[str, Any],
    embedding: Mapping[str, Any],
) -> None:
    fixed = config["fixed_comparison_contract"]
    generation = day21["generation"]
    actual = {
        "expected_case_count": len(day21["cases"]),
        "expected_supported_case_count": sum(
            case["expected_status"] == "supported" for case in day21["cases"]
        ),
        "expected_gap_case_count": sum(
            case["expected_status"] == "insufficient_evidence"
            for case in day21["cases"]
        ),
        "embedding_model": embedding["model"],
        "final_top_k": day21["baseline"]["top_k"],
        "generator_model": generation["model"],
        "temperature": generation["temperature"],
        "seed": generation["seed"],
        "bm25_enabled": day21["baseline"]["bm25_enabled"],
        "reranker_enabled": day21["baseline"]["reranker_enabled"],
        "retrieval_gate_enabled": day21["baseline"]["retrieval_gate_enabled"],
    }
    mismatches = {
        key: {"actual": value, "expected": fixed.get(key)}
        for key, value in actual.items()
        if value != fixed.get(key)
    }
    fixed_literals = {
        "question_source": "reuse_day21_cases_without_modification",
        "query_representation": (
            "author_written_question_with_day18_dense_query_instruction"
        ),
        "system_prompt": "reuse_day21_prompt_without_modification",
        "output_schema": "reuse_day21_per_case_json_schema",
        "patient_rows_used": False,
        "reference_labels_used": False,
    }
    for key, expected in fixed_literals.items():
        if fixed.get(key) != expected:
            mismatches[key] = {"actual": fixed.get(key), "expected": expected}
    if mismatches:
        raise ValueError(f"Day 22 單一變因比較契約不一致：{mismatches}")


def _verify_embedding_runtime(
    runtime: Mapping[str, Any],
    embedding: Mapping[str, Any],
    model_lock: Mapping[str, Any],
) -> None:
    dense_lock = model_lock["models"]["dense_embedding"]
    checks = {
        "lock_name": (dense_lock.get("name"), embedding["model"]),
        "lock_id": (dense_lock.get("model_id"), embedding["expected_model_id_prefix"]),
        "runtime_name": (runtime.get("model"), embedding["model"]),
        "runtime_digest": (runtime.get("digest"), embedding["expected_model_digest"]),
        "parameter_size": (
            runtime.get("parameter_size"),
            embedding["expected_parameter_size"],
        ),
        "quantization_level": (
            runtime.get("quantization_level"),
            embedding["expected_quantization_level"],
        ),
    }
    mismatches = {
        name: {"actual": actual, "expected": expected}
        for name, (actual, expected) in checks.items()
        if actual != expected
    }
    if mismatches:
        raise ValueError(f"本機 embedding 模型與固定契約不同：{mismatches}")


def _verify_generator_runtime(
    runtime: Mapping[str, Any],
    generation: Mapping[str, Any],
    model_lock: Mapping[str, Any],
) -> None:
    generator_lock = model_lock["models"]["primary_generator"]
    checks = {
        "lock_name": (generator_lock.get("name"), generation["model"]),
        "lock_id": (
            generator_lock.get("model_id"),
            generation["expected_model_id_prefix"],
        ),
        "runtime_name": (runtime.get("model"), generation["model"]),
        "runtime_digest": (
            runtime.get("digest"),
            generation["expected_model_digest"],
        ),
        "format": (runtime.get("format"), generation["expected_format"]),
        "family": (runtime.get("family"), generation["expected_family"]),
        "parameter_size": (
            runtime.get("parameter_size"),
            generation["expected_parameter_size"],
        ),
        "quantization_level": (
            runtime.get("quantization_level"),
            generation["expected_quantization_level"],
        ),
    }
    mismatches = {
        name: {"actual": actual, "expected": expected}
        for name, (actual, expected) in checks.items()
        if actual != expected
    }
    required = set(generation["required_capabilities"])
    capabilities = set(runtime.get("capabilities") or [])
    if not required.issubset(capabilities):
        mismatches["capabilities"] = {
            "actual": sorted(capabilities),
            "expected_at_least": sorted(required),
        }
    if mismatches:
        raise ValueError(f"本機生成模型與固定契約不同：{mismatches}")


def _round_floats(value: Any) -> Any:
    if isinstance(value, float):
        return round(value, 8)
    if isinstance(value, dict):
        return {key: _round_floats(child) for key, child in value.items()}
    if isinstance(value, list):
        return [_round_floats(child) for child in value]
    return value


def _generate_with_contract_repair(
    *,
    system_prompt: str,
    user_message: str,
    schema: Mapping[str, Any],
    generation: Mapping[str, Any],
) -> tuple[JsonObject, JsonObject]:
    maximum_attempts = int(generation["maximum_attempts"])
    if maximum_attempts != 2:
        raise ValueError("Day 22 沿用 Day 21：最多兩次結構化生成嘗試")
    repair_note = (
        "\n\n前一次輸出未通過固定 JSON Schema。請重新檢查，且只輸出包含 "
        "case_id、status、answer_level、answer_zh、evidence_rule_ids、"
        "missing_information、safety_notice 七個欄位的 JSON 物件。"
    )
    first_error = None
    for attempt in range(1, maximum_attempts + 1):
        message = user_message if attempt == 1 else user_message + repair_note
        try:
            answer, metrics = chat_structured(
                system_prompt=system_prompt,
                user_message=message,
                schema=schema,
                config=generation,
            )
        except OllamaChatError as exc:
            if attempt == maximum_attempts:
                raise
            first_error = str(exc)
            continue
        return answer, {
            **metrics,
            "attempt_count": attempt,
            "contract_repair_triggered": attempt > 1,
            "first_attempt_error": first_error,
        }
    raise AssertionError("unreachable")


def _pipeline_aggregate(
    case_results: Sequence[Mapping[str, Any]], pipeline_key: str
) -> JsonObject:
    supported = [
        item for item in case_results if item["expected_status"] == "supported"
    ]
    gaps = [
        item
        for item in case_results
        if item["expected_status"] == "insufficient_evidence"
    ]
    checks = [item[pipeline_key]["evaluation"]["checks"] for item in case_results]
    supported_checks = [
        item[pipeline_key]["evaluation"]["checks"] for item in supported
    ]
    return {
        "case_count": len(case_results),
        "passed_case_count": sum(
            item[pipeline_key]["evaluation"]["passed"] for item in case_results
        ),
        "supported_case_count": len(supported),
        "supported_status_match_count": sum(
            item[pipeline_key]["answer"]["status"] == "supported"
            for item in supported
        ),
        "gap_case_count": len(gaps),
        "gap_refusal_count": sum(
            item[pipeline_key]["answer"]["status"] == "insufficient_evidence"
            for item in gaps
        ),
        "required_evidence_retrieval_hit_count": sum(
            check["required_evidence_was_retrieved"] for check in supported_checks
        ),
        "required_evidence_citation_hit_count": sum(
            check["required_evidence_was_cited"] for check in supported_checks
        ),
        "citation_subset_pass_count": sum(
            check["citations_are_retrieved"] for check in checks
        ),
        "all_cases_passed": all(
            item[pipeline_key]["evaluation"]["passed"] for item in case_results
        ),
    }


def _paired_aggregate(case_results: Sequence[Mapping[str, Any]]) -> JsonObject:
    supported = [
        item for item in case_results if item["expected_status"] == "supported"
    ]
    expanded_counts = [
        item["hierarchical"]["expanded_child_count"] for item in case_results
    ]
    return {
        "supported_case_count": len(supported),
        "required_evidence_hit_by_both_count": sum(
            item["flat"]["evaluation"]["checks"]["required_evidence_was_retrieved"]
            and item["hierarchical"]["evaluation"]["checks"]["required_evidence_was_retrieved"]
            for item in supported
        ),
        "flat_only_required_evidence_hit_count": sum(
            item["flat"]["evaluation"]["checks"]["required_evidence_was_retrieved"]
            and not item["hierarchical"]["evaluation"]["checks"]["required_evidence_was_retrieved"]
            for item in supported
        ),
        "hierarchical_only_required_evidence_hit_count": sum(
            not item["flat"]["evaluation"]["checks"]["required_evidence_was_retrieved"]
            and item["hierarchical"]["evaluation"]["checks"]["required_evidence_was_retrieved"]
            for item in supported
        ),
        "hierarchical_required_parent_selected_count": sum(
            item["comparison"]["required_parent_selected"] is True
            for item in supported
        ),
        "identical_final_rule_order_count": sum(
            item["comparison"]["identical_final_rule_order"]
            for item in case_results
        ),
        "identical_answer_object_count": sum(
            item["comparison"]["identical_answer_object"]
            for item in case_results
        ),
        "expanded_child_count_min": min(expanded_counts),
        "expanded_child_count_max": max(expanded_counts),
        "expanded_child_count_mean": sum(expanded_counts) / len(expanded_counts),
        "flat_candidate_count": 9,
        "parent_candidate_count": 3,
    }


def _rule_rank(
    ranking: Sequence[Mapping[str, Any]], required: Sequence[str]
) -> Dict[str, int | None]:
    rank_by_rule = {str(item["rule_id"]): int(item["rank"]) for item in ranking}
    return {rule_id: rank_by_rule.get(rule_id) for rule_id in required}


def main() -> int:
    args = parse_args()
    started_at = datetime.now(timezone.utc)
    config_path = resolve_project_path(PROJECT_ROOT, args.config)
    config = load_json(config_path)
    if config.get("schema_version") != 1:
        raise ValueError("目前只支援 schema_version=1")
    if config.get("scope") != EXPECTED_SCOPE:
        raise ValueError("Day 22 scope 不正確")
    source_paths = _verify_sources(config)
    sources = config["sources"]

    day21 = load_json(resolve_project_path(PROJECT_ROOT, sources["day21_config_path"]))
    flat_documents = read_jsonl(
        resolve_project_path(PROJECT_ROOT, sources["flat_chunks_path"])
    )
    parents = read_jsonl(
        resolve_project_path(PROJECT_ROOT, sources["hierarchical_parents_path"])
    )
    children = read_jsonl(
        resolve_project_path(PROJECT_ROOT, sources["hierarchical_children_path"])
    )
    embedding_config = load_json(
        resolve_project_path(PROJECT_ROOT, sources["embedding_config_path"])
    )
    embedding = embedding_config["embedding"]
    generation = day21["generation"]
    model_lock = load_json(
        resolve_project_path(PROJECT_ROOT, sources["model_lock_path"])
    )
    system_prompt = resolve_project_path(
        PROJECT_ROOT, sources["system_prompt_path"]
    ).read_text(encoding="utf-8")

    print("步驟 1/5：驗證單一變因、父子圖、child/flat 同源與模型鎖定。", flush=True)
    _verify_fixed_comparison(config, day21, embedding)
    flat_contract = validate_flat_contract(flat_documents, day21)
    cases = list(day21["cases"])
    case_contract = validate_public_cases(cases, flat_documents, day21)
    hierarchy_contract = validate_hierarchical_contract(
        parents, children, flat_documents, config
    )
    embedding_runtime = get_runtime_metadata(
        embedding["endpoint"],
        embedding["model"],
        timeout_seconds=int(embedding["timeout_seconds"]),
    )
    generator_runtime = get_chat_runtime_metadata(
        generation["endpoint"],
        generation["model"],
        timeout_seconds=int(generation["timeout_seconds"]),
    )
    _verify_embedding_runtime(embedding_runtime, embedding, model_lock)
    _verify_generator_runtime(generator_runtime, generation, model_lock)

    print("步驟 2/5：建立九筆 child、三筆 parent 與六筆 query 向量。", flush=True)
    flat_vectors, flat_embedding_metrics = embed_texts(
        [item["search_text"] for item in flat_documents], embedding
    )
    parent_vectors, parent_embedding_metrics = embed_texts(
        [item["context_text"] for item in parents], embedding
    )
    query_inputs = [
        f"Instruct: {embedding['query_instruction']}\nQuery: {case['question']}"
        for case in cases
    ]
    query_vectors, query_embedding_metrics = embed_texts(query_inputs, embedding)
    flat_by_rule = {
        item["metadata"]["rule_id"]: item for item in flat_documents
    }
    parent_by_id = {item["parent_id"]: item for item in parents}
    child_by_rule = {item["metadata"]["rule_id"]: item for item in children}
    child_vector_by_rule = {
        document["metadata"]["rule_id"]: vector
        for document, vector in zip(flat_documents, flat_vectors)
    }

    print("步驟 3/5：逐題配對執行 Flat 與 Parent Top-2 → Child Top-3。", flush=True)
    case_results = []
    parent_top_k = int(config["hierarchical_pipeline"]["parent_top_k"])
    final_top_k = int(config["fixed_comparison_contract"]["final_top_k"])
    for index, (case, query_vector) in enumerate(zip(cases, query_vectors), start=1):
        flat_ranking = rank_documents(
            query_vector, flat_documents, flat_vectors
        )
        flat_context = build_context(
            flat_ranking, flat_by_rule, top_k=final_top_k
        )

        parent_ranking = rank_parents(query_vector, parents, parent_vectors)
        expanded = expand_selected_children(
            parent_ranking,
            parent_by_id,
            child_by_rule,
            parent_top_k=parent_top_k,
        )
        hierarchical_child_ranking = rank_expanded_children(
            query_vector, expanded, child_vector_by_rule
        )
        hierarchical_context = build_hierarchical_context(
            hierarchical_child_ranking,
            child_by_rule,
            top_k=final_top_k,
        )

        schema = build_response_schema(
            case["case_id"],
            generation["safety_notice"],
            answer_kind=case["answer_kind"],
        )
        flat_answer, flat_generation_metrics = _generate_with_contract_repair(
            system_prompt=system_prompt,
            user_message=build_user_message(
                case["case_id"], case["question"], flat_context
            ),
            schema=schema,
            generation=generation,
        )
        hierarchical_answer, hierarchical_generation_metrics = (
            _generate_with_contract_repair(
                system_prompt=system_prompt,
                user_message=build_hierarchical_user_message(
                    case["case_id"], case["question"], hierarchical_context
                ),
                schema=schema,
                generation=generation,
            )
        )
        flat_evaluation = evaluate_answer(
            flat_answer,
            case,
            flat_context,
            safety_notice=generation["safety_notice"],
        )
        hierarchical_evaluation = evaluate_answer(
            hierarchical_answer,
            case,
            hierarchical_context,
            safety_notice=generation["safety_notice"],
        )

        required = list(case["required_evidence_rule_ids"])
        required_parents = expected_parent_ids(required, child_by_rule)
        selected_parent_ids = [
            str(item["parent_id"]) for item in parent_ranking[:parent_top_k]
        ]
        flat_rule_order = [item["rule_id"] for item in flat_context]
        hierarchical_rule_order = [
            item["rule_id"] for item in hierarchical_context
        ]
        comparison = {
            "required_parent_ids": required_parents,
            "selected_parent_ids": selected_parent_ids,
            "required_parent_selected": (
                None
                if not required_parents
                else set(required_parents).issubset(selected_parent_ids)
            ),
            "flat_required_rule_full_ranks": _rule_rank(flat_ranking, required),
            "hierarchical_required_rule_expanded_ranks": _rule_rank(
                hierarchical_child_ranking, required
            ),
            "flat_final_rule_ids": flat_rule_order,
            "hierarchical_final_rule_ids": hierarchical_rule_order,
            "final_rule_overlap_count": len(
                set(flat_rule_order).intersection(hierarchical_rule_order)
            ),
            "identical_final_rule_order": flat_rule_order == hierarchical_rule_order,
            "identical_answer_object": flat_answer == hierarchical_answer,
        }
        case_results.append(
            {
                "case_id": case["case_id"],
                "question": case["question"],
                "answer_kind": case["answer_kind"],
                "expected_status": case["expected_status"],
                "expected_answer_level": case["expected_answer_level"],
                "required_evidence_rule_ids": required,
                "query_vector_sha256": vector_sha256(query_vector),
                "flat": {
                    "candidate_count": len(flat_documents),
                    "retrieved_context": flat_context,
                    "answer": flat_answer,
                    "evaluation": flat_evaluation,
                    "generation_metrics": flat_generation_metrics,
                },
                "hierarchical": {
                    "parent_ranking": parent_ranking,
                    "selected_parent_count": parent_top_k,
                    "expanded_child_count": len(expanded),
                    "expanded_child_ranking": hierarchical_child_ranking,
                    "retrieved_context": hierarchical_context,
                    "answer": hierarchical_answer,
                    "evaluation": hierarchical_evaluation,
                    "generation_metrics": hierarchical_generation_metrics,
                },
                "comparison": comparison,
            }
        )
        state = (
            "兩邊通過"
            if flat_evaluation["passed"] and hierarchical_evaluation["passed"]
            else "至少一邊未通過"
        )
        print(f"  {index:02d}/06 {case['case_id']}：{state}", flush=True)

    print("步驟 4/5：計算配對聚合、候選縮減與 trace 差異。", flush=True)
    flat_aggregate = _pipeline_aggregate(case_results, "flat")
    hierarchical_aggregate = _pipeline_aggregate(case_results, "hierarchical")
    paired_aggregate = _paired_aggregate(case_results)
    aggregate = {
        "flat": flat_aggregate,
        "hierarchical": hierarchical_aggregate,
        "paired": _round_floats(paired_aggregate),
        "both_pipelines_all_cases_passed": (
            flat_aggregate["all_cases_passed"]
            and hierarchical_aggregate["all_cases_passed"]
        ),
    }

    run_id = (
        f"{started_at.strftime('%Y%m%dT%H%M%S%fZ')}-"
        f"{sha256_file(config_path)[:8]}"
    )
    run_root = resolve_project_path(PROJECT_ROOT, config["outputs"]["run_output_root"])
    run_directory = run_root / run_id
    run_directory.mkdir(parents=True, exist_ok=False)
    full_results_path = run_directory / "flat-vs-hierarchical-results.json"
    rounded_case_results = _round_floats(case_results)
    full_results = {
        "schema_version": 1,
        "experiment_id": config["experiment_id"],
        "scope": config["scope"],
        "aggregate": aggregate,
        "case_results": rounded_case_results,
    }
    write_json(full_results_path, full_results)

    success_id = config["outputs"]["success_trace_case_id"]
    failure_id = config["outputs"]["failure_trace_case_id"]
    public_summary = {
        "schema_version": 1,
        "experiment_id": config["experiment_id"],
        "scope": config["scope"],
        "fixed_comparison_contract": config["fixed_comparison_contract"],
        "flat_pipeline": config["flat_pipeline"],
        "hierarchical_pipeline": config["hierarchical_pipeline"],
        "input_contract": {
            **flat_contract,
            **case_contract,
            **hierarchy_contract,
        },
        "runtime": {
            "embedding": embedding_runtime,
            "generator": generator_runtime,
        },
        "parameters": {
            "embedding_model": embedding["model"],
            "embedding_dimensions": embedding["dimensions"],
            "query_instruction": embedding["query_instruction"],
            "generator_model": generation["model"],
            "temperature": generation["temperature"],
            "seed": generation["seed"],
            "think": generation["think"],
            "stream": generation["stream"],
            "system_prompt_sha256": sources["system_prompt_sha256"],
            "structured_output": "same_day21_per_case_json_schema",
        },
        "aggregate": aggregate,
        "case_results": rounded_case_results,
        "success_trace_case": next(
            item for item in rounded_case_results if item["case_id"] == success_id
        ),
        "failure_trace_case": next(
            item for item in rounded_case_results if item["case_id"] == failure_id
        ),
        "checks": {
            "source_hashes_match": True,
            "day21_questions_generation_prompt_and_schema_reused": True,
            "child_rule_text_hash_and_metadata_match_flat": True,
            "parent_top2_then_child_dense_top3_only": True,
            "bm25_reranker_and_gate_disabled_for_both": True,
            "only_final_three_child_texts_exposed_to_generator": True,
            "embedding_and_generator_locks_match_runtime": True,
            "only_author_written_public_knowledge_questions_used": True,
            "patient_rows_and_reference_labels_not_read": True,
        },
        "limitations": config["limitations"],
        "interpretation": (
            "六筆公開知識問題的配對工程比較；只描述這個 Parent Top-2 → "
            "Child Top-3 設計的觀察，不是病患五級分類、臨床正確性或部署安全證據。"
        ),
    }
    public_path = resolve_project_path(
        PROJECT_ROOT, config["outputs"]["public_summary_path"]
    )
    write_json(public_path, public_summary)

    manifest = {
        "manifest_schema_version": 1,
        "run_id": run_id,
        "experiment_id": config["experiment_id"],
        "started_at_utc": started_at.isoformat().replace("+00:00", "Z"),
        "command": [
            sys.executable,
            str(Path(__file__).relative_to(PROJECT_ROOT)),
            "--config",
            args.config,
        ],
        "git": git_state(PROJECT_ROOT),
        "runtime": {
            "python": platform.python_version(),
            "embedding": embedding_runtime,
            "generator": generator_runtime,
        },
        "parameters": public_summary["parameters"],
        "inputs": [
            file_record(PROJECT_ROOT, args.config),
            *[
                file_record(PROJECT_ROOT, str(path.relative_to(PROJECT_ROOT)))
                for path in source_paths
            ],
            *[file_record(PROJECT_ROOT, path) for path in IMPLEMENTATION_PATHS],
        ],
        "embedding_metrics": {
            "flat_children": flat_embedding_metrics,
            "parents": parent_embedding_metrics,
            "queries": query_embedding_metrics,
            "flat_child_vector_sha256": {
                document["metadata"]["rule_id"]: vector_sha256(vector)
                for document, vector in zip(flat_documents, flat_vectors)
            },
            "parent_vector_sha256": {
                parent["parent_id"]: vector_sha256(vector)
                for parent, vector in zip(parents, parent_vectors)
            },
        },
        "outputs": [
            file_record(
                PROJECT_ROOT, str(full_results_path.relative_to(PROJECT_ROOT))
            ),
            file_record(PROJECT_ROOT, str(public_path.relative_to(PROJECT_ROOT))),
        ],
        "scope": config["scope"],
    }
    manifest_path = run_directory / "run-manifest.json"
    write_json(manifest_path, manifest)

    print("步驟 5/5：寫出公開摘要、完整配對結果與 run manifest。", flush=True)
    print(f"公開摘要：{public_path.relative_to(PROJECT_ROOT)}", flush=True)
    print(f"完整結果：{full_results_path.relative_to(PROJECT_ROOT)}", flush=True)
    print(f"執行紀錄：{manifest_path.relative_to(PROJECT_ROOT)}", flush=True)
    if not aggregate["both_pipelines_all_cases_passed"]:
        print(
            "固定比較已完成；至少一個 pipeline 未通過逐題檢查，"
            "這是保留在結果中的實驗觀察，不是執行錯誤。",
            flush=True,
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

儲存後先確認檔名與相對路徑完全一致，再繼