Day 20 已經能把九筆公開知識排出順序，但排序第一名仍不是答案。真正的檢索增強生成系統還要把取回內容交給生成模型，限制模型只能依證據作答，最後保存「取回了什麼、回答了什麼、引用了什麼」的完整紀錄。

如果直接把問題交給大型語言模型，回答可能很流暢，讀者卻無法知道內容來自哪一段公開規則。相反地，如果一次把九筆知識全部塞進提示詞，系統雖然看似「有用知識庫」，卻沒有建立可比較的檢索基準。今天先做最克制的一版：九筆知識全部放在同一層，只用語意向量的稠密檢索取前三筆，再交給固定的本機模型輸出結構化答案。

這條基準稱為平面式檢索增強生成（Flat Retrieval-Augmented Generation, Flat RAG）：所有知識區塊位於同一層，不先經過父層路由。本文也會用 Flat Basic RAG 指稱今天這套只保留基本檢索與生成元件的實作。

下圖是概念示意。請先觀察左側九張卡仍在同一層，只有三張被搜尋光束選中；右側工作台代表本機生成步驟，不代表自動臨床判斷。

上圖要建立的直覺是「先找證據，再形成答案」。圖片沒有承載精確醫療規則；真正的文件數、前三筆證據、模型、欄位與驗證條件都由本文的設定檔和程式固定。

本篇只處理作者根據公開資料撰寫的六筆知識問題，不讀取 Kaggle 病患列、護理師登錄級數或專家重新判定級數。這是一個公開知識問答的工程整合測試，不是病患五級檢傷實驗，也不是臨床部署證據。

今天會完成什麼

完成本篇後，你會得到五個可檢查的產物：

1. 一份鎖定九筆平面知識、稠密檢索前三筆、兩個本機模型與六筆問題的實驗契約。
2. 一個只允許本機模型服務、會驗證模型內容指紋與輸出結構的生成用戶端。
3. 一條從查詢向量、平面檢索、前三筆證據到結構化答案的 Flat Basic RAG。
4. 十四項不啟動模型也能執行的單元測試，以及六筆真正的本機端到端結果。
5. 一份公開摘要與一份逐次執行清單，能追查輸入、模型、參數、程式與輸出雜湊。

今天建立的是研究計畫中的 B5 平面稠密 RAG 基準。B4 無檢索生成、Day 20 的混合檢索與 Cohere 重排序，以及其他基準的配對比較，留到後續公平比較與消融實驗處理。

本篇會用到的名詞

本篇同時跨越檢索、生成與輸出驗證，先用詞彙表建立共同語言。正文第一次出現時仍會重新解釋用途。

Flat Basic RAG 到底固定了什麼

檢索增強生成（Retrieval-Augmented Generation, RAG）是一種先從外部知識庫取回相關證據，再讓大型語言模型（Large Language Model, LLM）依證據產生回答的方法。外部證據讓回答可以追查來源，但不保證模型一定正確使用證據。

平面式檢索增強生成（Flat Retrieval-Augmented Generation, Flat RAG）把每一筆知識視為同一層候選。它沒有先選父層主題，也沒有沿父子關係展開規則。今天的九筆文件包含 KTAS 制度目的、高階流程、主要與次要考量摘要，以及第一級到第五級的公開定義。

稠密檢索（Dense Retrieval）先用向量嵌入（Vector Embedding）把問題與文件轉成數字向量，再以餘弦相似度（Cosine Similarity）排序。餘弦相似度越大，代表兩個向量方向越接近；它只能表示模型眼中的語意接近程度，不是答案正確率、機率或醫療安全分數。

前 k 筆（Top-k）是排序後真正留下的前 k 筆候選。本篇固定 k=3，所以生成模型只會看到三筆公開摘要。上一步產生的完整九筆 Dense 排名會先被截成 Top-3；這三筆帶有 rule_id、相似度、來源與完整度，接著才被序列化成 EVIDENCE，成為生成步驟的輸入。

為什麼不用 Day 20 已經完成的 Hybrid 與重排序

因為今天的目的不是把所有已完成元件一次接上，而是建立一條簡單、透明、可歸因的基準。若同時加入最佳匹配 25（Best Matching 25, BM25）、Hybrid、Cohere 重排序、階層結構與門控，之後看到結果改變時就無法知道是哪個元件造成差異。

B5 因此刻意固定：

下圖把每一步的輸入與輸出放在同一條線上。請特別觀察底部四個關閉元件；Day 22 若要比較知識結構，應固定其他條件，只改平面與階層候選的組織方式。

上圖也解釋了為什麼 Basic 不等於隨便：方法可以簡單，但每個輸入、關閉元件、模型與輸出都要明確。缺少這些固定條件的「簡單 RAG」不能成為公平基準。

只用公開知識，先驗證管線而不是病患分類

韓國急診檢傷與急迫度分級量表（Korean Triage and Acuity Scale, KTAS）是五級檢傷制度。本系列目前整理的九筆內容只足以支援部分公開知識問題，沒有完整主訴目錄、各主訴的數值門檻與官方版本沿革。

因此 Day 21 不把病患主訴或生命徵象交給模型，也不要求模型替病患判定級數。六筆問題分成：

四筆支援題都指定必要的 rule_id，讓程式檢查必要證據是否進入 Top-3，以及答案是否真的引用它。兩筆缺口題沒有必要規則；程式只檢查模型是否回報證據不足、級數為 null，並列出缺少資訊。

這個設計不能評估病患檢傷能力，卻能先回答三個工程問題：

1. 上游平面知識能否成為 Dense 檢索輸入？
2. Top-3 能否成為本機生成模型唯一可見的證據？
3. 答案能否用固定欄位留下狀態、級數、引用與缺口？

生成模型不是拿到一段文字就算完成

應用程式介面（Application Programming Interface, API）是程式交換請求與回應的規格。本篇使用本機模型執行環境 Ollama 的 /api/chat 端點。Ollama 預設在 http://localhost:11434/api 提供本機介面；本篇程式只接受 127.0.0.1、localhost 或 ::1 的超文字傳輸協定（Hypertext Transfer Protocol, HTTP）endpoint，不允許把內容改送到遠端網址：Ollama API 簡介。

Ollama 的 Chat API 接收 model、messages、format、stream、think 與執行選項；format 可以直接放入 JSON Schema：Ollama Chat API。官方也建議除了把 schema 放在 format，同時在提示詞中提供 schema，並以較低溫度增加結構化輸出的穩定性：Ollama Structured Outputs。

JSON 結構描述（JSON Schema）在本篇負責限制七個欄位：

題型 schema 只限制「級數題必須輸出某個 1 到 5」，不會把正確級數 2 或 4 注入模型。正確級數仍由模型根據 Top-3 證據產生，再由執行程式和預先固定的案例契約比較。

schema 失敗要明確處理

即使使用 JSON Schema，本機模型仍可能漏欄位或輸出不符合題型的值。本篇最多嘗試兩次：第一次失敗時，附加一段固定的七欄位提醒；第二次仍失敗就停止，不寫出一份假裝成功的公開結果。

這個修復流程只修正格式，不新增證據、不改問題，也不告訴模型正確答案。每一題的 attempt_count、是否啟動修復與第一次錯誤都會保存在結果中。

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

開始建立檔案前，先理解每個路徑負責什麼：

若只閱讀本篇但缺少前置產物，請先完成 Day 13 建立 Poetry 環境與重現性工具、Day 17 建立九筆 flat-chunks.jsonl，以及 Day 18 安裝 qwen3-embedding:4b 和 embedding 用戶端。以下完整檔案不需要從其他程式碼網站取得。

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

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

本篇沿用 Day 13 的 Poetry 與重現性工具、Day 17 的九筆 flat chunks、Day 18 的 qwen3-embedding:4b 與 Ollama embedding 用戶端，以及本機已下載的 qwen3.5:4b-mlx。六筆問題全部由作者根據公開知識與已登錄缺口撰寫，不讀取 Kaggle 病患列或參考標籤。以下是 Day 21 新增的完整契約、提示詞、本機生成用戶端、Flat RAG 核心、執行入口與測試；公開摘要、完整 trace 與 run manifest 都由執行入口自動產生，不需要手動建立。

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

mkdir -p configs/rag prompts src/triage_rag/generation src/triage_rag/rag scripts tests results/public results/runs/day-21

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

檔案 1：建立 configs/rag/day-21-flat-basic-rag.json

鎖定九筆平面候選、Dense Top-3、兩個本機模型 digest、生成參數、六筆公開問題、輸出與知識邊界。

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

{
  "schema_version": 1,
  "experiment_id": "day-21-flat-basic-rag-public-knowledge",
  "scope": "author_written_public_knowledge_rag_integration_not_patient_triage_or_clinical_evaluation",
  "sources": {
    "flat_chunks_path": "data/knowledge/ktas-public-v1/day-17/flat-chunks.jsonl",
    "flat_chunks_sha256": "10749728567e36c565d9e2e0fb931c1ece7da448bd1c04f4527b624e8507a1fa",
    "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"
  },
  "baseline": {
    "baseline_id": "B5",
    "store_type": "flat",
    "retrieval_method": "dense_cosine",
    "top_k": 3,
    "expected_document_count": 9,
    "hierarchy_enabled": false,
    "bm25_enabled": false,
    "reranker_enabled": false,
    "retrieval_gate_enabled": false,
    "fixed_query_representation": "author_written_question_with_day18_dense_query_instruction",
    "comparison_policy": "Day 22 若比較階層結構，生成模型、問題、Top-3、提示詞與輸出 schema 必須固定。"
  },
  "generation": {
    "runtime": "Ollama",
    "endpoint": "http://127.0.0.1:11434",
    "model": "qwen3.5:4b-mlx",
    "expected_model_id_prefix": "61aa3858e9d3",
    "expected_model_digest": "61aa3858e9d3022e8fca725550089addd9289c4446f33bd09dfd12f95f2a6792",
    "expected_format": "safetensors",
    "expected_family": "qwen3_5",
    "expected_parameter_size": "4.5B",
    "expected_quantization_level": "nvfp4",
    "required_capabilities": ["completion"],
    "temperature": 0,
    "seed": 20260811,
    "num_predict": 512,
    "think": false,
    "stream": false,
    "keep_alive": "5m",
    "timeout_seconds": 300,
    "maximum_attempts": 2,
    "contract_repair_policy": "若第一次回應未通過 JSON Schema，附上固定的七欄位提醒後只重試一次；第二次仍失敗就停止且不寫公開結果。",
    "safety_notice": "研究與教學用途的決策支援示範，不提供個人醫療建議，也不取代醫療專業人員。"
  },
  "input_contract": {
    "expected_case_count": 6,
    "expected_supported_case_count": 4,
    "expected_gap_case_count": 2,
    "forbidden_fields": [
      "record_index",
      "KTAS_RN",
      "KTAS_expert",
      "Group",
      "diagnosis",
      "disposition",
      "length_of_stay",
      "Error_group",
      "mistriage"
    ]
  },
  "cases": [
    {
      "case_id": "supported-level-2",
      "question": "KTAS 哪一級僅次於第一級，代表生命、肢體或身體功能可能受到威脅，需要快速治療？",
      "answer_kind": "level",
      "expected_status": "supported",
      "expected_answer_level": 2,
      "required_evidence_rule_ids": ["ktas-public-level-2-001"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "case_id": "supported-level-4",
      "question": "KTAS 哪一級會依年齡、疼痛、惡化或併發症可能性，在一至兩小時內接受處置或重新評估？",
      "answer_kind": "level",
      "expected_status": "supported",
      "expected_answer_level": 4,
      "required_evidence_rule_ids": ["ktas-public-level-4-001"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "case_id": "supported-workflow",
      "question": "公開介紹中的 KTAS 高階檢傷流程依序包含哪些步驟？",
      "answer_kind": "text",
      "expected_status": "supported",
      "expected_answer_level": null,
      "required_evidence_rule_ids": ["ktas-public-workflow-001"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "case_id": "supported-primary-considerations",
      "question": "公開研究列出的 KTAS 共通主要考量類型有哪些？",
      "answer_kind": "text",
      "expected_status": "supported",
      "expected_answer_level": null,
      "required_evidence_rule_ids": ["ktas-public-primary-considerations-001"],
      "origin": "author_written_from_public_knowledge_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "case_id": "gap-complete-thresholds",
      "question": "請提供每一個 KTAS 主訴可直接套用的完整數值門檻。",
      "answer_kind": "gap",
      "expected_status": "insufficient_evidence",
      "expected_answer_level": null,
      "required_evidence_rule_ids": [],
      "origin": "author_written_from_documented_knowledge_gap_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    },
    {
      "case_id": "gap-version-history",
      "question": "請根據目前知識庫列出完整官方版本編號、發布日期與逐版變更內容。",
      "answer_kind": "gap",
      "expected_status": "insufficient_evidence",
      "expected_answer_level": null,
      "required_evidence_rule_ids": [],
      "origin": "author_written_from_documented_knowledge_gap_not_patient_record",
      "uses_patient_record": false,
      "uses_reference_label": false
    }
  ],
  "outputs": {
    "public_summary_path": "results/public/day-21-flat-basic-rag.json",
    "run_output_root": "results/runs/day-21",
    "trace_case_id": "supported-level-2"
  },
  "limitations": [
    "六筆問題只驗證公開知識的端到端整合，不是病患五級檢傷評估。",
    "目前知識庫只有九筆公開摘要，沒有完整主訴目錄、逐主訴門檻或官方版本沿革。",
    "結構與字串檢查不能取代多位專家對答案語意、臨床正確性與安全性的審查。",
    "本篇只建立 B5 平面稠密 RAG；B4 無檢索生成與其他基準的配對比較留到 Day 26。"
  ]
}

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

檔案 2：建立 prompts/day-21-flat-basic-rag-system.txt

要求生成模型只能依 EVIDENCE 回答，使用固定狀態、引用、缺口與安全說明欄位。

請在文字編輯器建立 prompts/day-21-flat-basic-rag-system.txt，貼入以下完整內容並儲存：

你是「公開 KTAS 知識問答」的研究與教學助理。KTAS 是韓國急診檢傷與急迫度分級量表（Korean Triage and Acuity Scale）。

你的唯一知識來源是使用者訊息中的 EVIDENCE。即使你記得其他資料，也不得使用模型既有知識補齊缺少的規則、數值門檻、主訴路徑或級數。

請遵守以下規則：

1. 先判斷 EVIDENCE 是否足以直接回答 QUESTION。
2. 若足以回答，status 使用 supported；answer_zh 只整理 EVIDENCE 已明確支持的內容，missing_information 必須是空陣列。
3. 若不足，status 使用 insufficient_evidence；answer_level 必須是 null，answer_zh 必須清楚指出目前證據不足，missing_information 必須列出缺少的知識或資訊。
4. 只有 QUESTION 明確詢問級數，而且 EVIDENCE 直接支持該級數時，answer_level 才能填入 1 到 5；一般制度、流程或考量問題一律填 null。
5. evidence_rule_ids 只能填入 EVIDENCE 中實際支持回答的 rule_id，不得虛構識別碼。沒有足以支持答案的證據時可使用空陣列。
6. 不得把一般級數定義套到個別案例，不得根據生命徵象自行猜測級數，也不得把部分公開摘要說成完整 KTAS 規則。級數的公開定義不等於主訴特定的數值門檻；若問題要求完整門檻，即使 EVIDENCE 出現部分級數定義，仍要指出缺少的是完整主訴目錄與各主訴的數值門檻，不能暗示已取得任何級數的完整門檻。
7. 這是研究與教學用途的決策支援示範，不提供個人醫療建議，也不取代醫療專業人員。
8. 每次都必須完整輸出 case_id、status、answer_level、answer_zh、evidence_rule_ids、missing_information 與 safety_notice 七個欄位；只輸出符合指定 JSON Schema 的 JSON 物件，不要加入 Markdown、程式碼圍欄或額外說明。

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

檔案 3：建立 src/triage_rag/generation/__init__.py

建立本機文字生成子套件入口，公開 Ollama 錯誤、runtime 檢查、JSON Schema 與結構化生成函式。

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

"""Local text generation clients used by the triage RAG experiments."""

from triage_rag.generation.ollama_chat import (
    OllamaChatError,
    build_response_schema,
    chat_structured,
    get_chat_runtime_metadata,
)

__all__ = [
    "OllamaChatError",
    "build_response_schema",
    "chat_structured",
    "get_chat_runtime_metadata",
]

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

檔案 4：建立 src/triage_rag/generation/ollama_chat.py

實作只允許本機 HTTP endpoint 的 Ollama Chat 用戶端、模型資訊核對、逐題 JSON Schema 與回應驗證。

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

"""Minimal local Ollama chat client with JSON Schema validation."""

from __future__ import annotations

import json
from typing import Any, Dict, Mapping, Tuple
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen

from triage_rag.reproducibility import canonical_json_bytes


JsonObject = Dict[str, Any]


class OllamaChatError(RuntimeError):
    """Raised when the local Ollama chat service breaks the Day 21 contract."""


def _local_endpoint(endpoint: str) -> str:
    parsed = urlparse(endpoint)
    if parsed.scheme != "http" or parsed.hostname not in {
        "127.0.0.1",
        "localhost",
        "::1",
    }:
        raise OllamaChatError("Day 21 只允許本機 HTTP Ollama endpoint")
    return endpoint.rstrip("/")


def _request_json(
    endpoint: str,
    path: str,
    *,
    payload: Mapping[str, Any] | None = None,
    timeout_seconds: int = 300,
) -> JsonObject:
    base = _local_endpoint(endpoint)
    data = None if payload is None else canonical_json_bytes(payload)
    request = Request(
        base + path,
        data=data,
        method="GET" if data is None else "POST",
        headers={"Content-Type": "application/json"},
    )
    try:
        with urlopen(request, timeout=timeout_seconds) as response:
            body = response.read().decode("utf-8")
    except HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise OllamaChatError(
            f"Ollama 回傳 HTTP {exc.code}：{detail[:300]}"
        ) from exc
    except URLError as exc:
        raise OllamaChatError(
            "無法連線本機 Ollama；請先啟動 Ollama 並確認 11434 連接埠。"
        ) from exc
    try:
        value = json.loads(body)
    except json.JSONDecodeError as exc:
        raise OllamaChatError("Ollama 回應不是合法 JSON") from exc
    if not isinstance(value, dict):
        raise OllamaChatError("Ollama 回應必須是 JSON 物件")
    if value.get("error"):
        raise OllamaChatError(f"Ollama 錯誤：{value['error']}")
    return value


def get_chat_runtime_metadata(
    endpoint: str, model: str, *, timeout_seconds: int = 60
) -> JsonObject:
    """Read the installed tag digest and model capabilities without generating."""

    version = _request_json(
        endpoint, "/api/version", timeout_seconds=timeout_seconds
    ).get("version")
    if not isinstance(version, str) or not version.strip():
        raise OllamaChatError("/api/version 缺少有效的 version")
    tags = _request_json(endpoint, "/api/tags", timeout_seconds=timeout_seconds)
    models = tags.get("models")
    if not isinstance(models, list):
        raise OllamaChatError("/api/tags 缺少 models 陣列")
    matches = [
        item
        for item in models
        if item.get("name") == model or item.get("model") == model
    ]
    if len(matches) != 1:
        raise OllamaChatError(
            f"本機必須恰好存在一個 {model}，目前找到 {len(matches)} 個"
        )
    tag = matches[0]
    show = _request_json(
        endpoint,
        "/api/show",
        payload={"model": model, "verbose": False},
        timeout_seconds=timeout_seconds,
    )
    details = show.get("details") or {}
    capabilities = show.get("capabilities")
    if not isinstance(capabilities, list) or "completion" not in capabilities:
        raise OllamaChatError(f"{model} 沒有回報 completion 能力")
    return {
        "ollama_version": version,
        "model": tag.get("name") or tag.get("model"),
        "digest": tag.get("digest"),
        "size_bytes": tag.get("size"),
        "format": details.get("format"),
        "family": details.get("family"),
        "parameter_size": details.get("parameter_size"),
        "quantization_level": details.get("quantization_level"),
        "capabilities": capabilities,
    }


def build_response_schema(
    case_id: str,
    safety_notice: str,
    *,
    answer_kind: str = "any",
) -> JsonObject:
    """Build a per-case schema so identifiers and the safety notice cannot drift."""

    answer_levels = {
        "any": [None, 1, 2, 3, 4, 5],
        "level": [1, 2, 3, 4, 5],
        "text": [None],
        "gap": [None],
    }
    if answer_kind not in answer_levels:
        raise OllamaChatError(f"不支援的 answer_kind：{answer_kind}")

    return {
        "type": "object",
        "additionalProperties": False,
        "required": [
            "case_id",
            "status",
            "answer_level",
            "answer_zh",
            "evidence_rule_ids",
            "missing_information",
            "safety_notice",
        ],
        "properties": {
            "case_id": {"type": "string", "enum": [case_id]},
            "status": {
                "type": "string",
                "enum": ["supported", "insufficient_evidence"],
            },
            "answer_level": {"enum": answer_levels[answer_kind]},
            "answer_zh": {"type": "string", "minLength": 1, "maxLength": 500},
            "evidence_rule_ids": {
                "type": "array",
                "items": {"type": "string", "minLength": 1},
                "uniqueItems": True,
                "maxItems": 3,
            },
            "missing_information": {
                "type": "array",
                "items": {"type": "string", "minLength": 1},
                "uniqueItems": True,
                "maxItems": 5,
            },
            "safety_notice": {"type": "string", "enum": [safety_notice]},
        },
    }


def _validate_structured_answer(
    answer: Mapping[str, Any], schema: Mapping[str, Any]
) -> None:
    properties = schema["properties"]
    required = set(schema["required"])
    if set(answer) != required:
        raise OllamaChatError(
            f"結構化答案欄位不同；預期 {sorted(required)}，實際 {sorted(answer)}"
        )
    case_ids = properties["case_id"]["enum"]
    if answer["case_id"] not in case_ids:
        raise OllamaChatError("結構化答案 case_id 不符合本次問題")
    if answer["status"] not in properties["status"]["enum"]:
        raise OllamaChatError("結構化答案 status 不合法")
    if answer["answer_level"] not in properties["answer_level"]["enum"]:
        raise OllamaChatError(
            "answer_level 必須符合本題 schema："
            f"{properties['answer_level']['enum']}"
        )
    if not isinstance(answer["answer_zh"], str) or not answer["answer_zh"].strip():
        raise OllamaChatError("answer_zh 必須是非空字串")
    for field, maximum in (("evidence_rule_ids", 3), ("missing_information", 5)):
        values = answer[field]
        if (
            not isinstance(values, list)
            or len(values) > maximum
            or len(values) != len(set(values))
            or any(not isinstance(value, str) or not value.strip() for value in values)
        ):
            raise OllamaChatError(f"{field} 必須是未重複的非空字串陣列")
    if answer["safety_notice"] not in properties["safety_notice"]["enum"]:
        raise OllamaChatError("safety_notice 不符合固定文字")


def chat_structured(
    *,
    system_prompt: str,
    user_message: str,
    schema: Mapping[str, Any],
    config: Mapping[str, Any],
) -> Tuple[JsonObject, JsonObject]:
    """Call non-streaming local chat and parse its schema-constrained JSON content."""

    if not system_prompt.strip() or not user_message.strip():
        raise OllamaChatError("system prompt 與 user message 都不可為空")
    schema_text = json.dumps(schema, ensure_ascii=False, sort_keys=True)
    payload = {
        "model": config["model"],
        "messages": [
            {
                "role": "system",
                "content": system_prompt + "\n\n本次輸出 JSON Schema：\n" + schema_text,
            },
            {"role": "user", "content": user_message},
        ],
        "stream": False,
        "format": dict(schema),
        "think": False,
        "keep_alive": config["keep_alive"],
        "options": {
            "temperature": config["temperature"],
            "seed": config["seed"],
            "num_predict": config["num_predict"],
        },
    }
    response = _request_json(
        config["endpoint"],
        "/api/chat",
        payload=payload,
        timeout_seconds=int(config["timeout_seconds"]),
    )
    if response.get("model") != config["model"]:
        raise OllamaChatError("Ollama 回應模型與設定不同")
    if response.get("done") is not True:
        raise OllamaChatError("非串流回應缺少 done=true")
    message = response.get("message")
    if not isinstance(message, dict) or not isinstance(message.get("content"), str):
        raise OllamaChatError("Ollama 回應缺少 message.content")
    if isinstance(message.get("thinking"), str) and message["thinking"].strip():
        raise OllamaChatError("Day 21 已設定 think=false，但回應仍包含 thinking")
    try:
        answer = json.loads(message["content"])
    except json.JSONDecodeError as exc:
        raise OllamaChatError("message.content 不是合法 JSON") from exc
    if not isinstance(answer, dict):
        raise OllamaChatError("message.content 必須解析為 JSON 物件")
    _validate_structured_answer(answer, schema)
    metrics = {
        "total_duration_ns": response.get("total_duration"),
        "load_duration_ns": response.get("load_duration"),
        "prompt_eval_count": response.get("prompt_eval_count"),
        "prompt_eval_duration_ns": response.get("prompt_eval_duration"),
        "eval_count": response.get("eval_count"),
        "eval_duration_ns": response.get("eval_duration"),
        "done_reason": response.get("done_reason"),
        "thinking_disabled": True,
        "streaming_disabled": True,
    }
    return dict(answer), metrics

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

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

建立 RAG 子套件入口，公開 Flat baseline 的輸入、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,
)

__all__ = [
    "FlatRagContractError",
    "build_context",
    "build_user_message",
    "evaluate_answer",
    "validate_flat_contract",
    "validate_public_cases",
]

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

檔案 6：建立 src/triage_rag/rag/flat.py

鎖定平面 Dense Top-3，阻擋病患欄位，建立可引用 context，並檢查狀態、級數、必要證據與引用集合。

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

"""Flat dense RAG baseline with explicit evidence and refusal checks."""

from __future__ import annotations

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


JsonObject = Dict[str, Any]


class FlatRagContractError(ValueError):
    """Raised when inputs violate the fixed Day 21 baseline."""


def _all_keys(value: Any) -> Iterable[str]:
    if isinstance(value, Mapping):
        for key, child in value.items():
            yield str(key)
            yield from _all_keys(child)
    elif isinstance(value, list):
        for child in value:
            yield from _all_keys(child)


def validate_flat_contract(
    documents: Sequence[Mapping[str, Any]], config: Mapping[str, Any]
) -> JsonObject:
    baseline = config["baseline"]
    required = {
        "store_type": "flat",
        "retrieval_method": "dense_cosine",
        "top_k": 3,
        "hierarchy_enabled": False,
        "bm25_enabled": False,
        "reranker_enabled": False,
        "retrieval_gate_enabled": False,
    }
    mismatches = {
        key: {"actual": baseline.get(key), "expected": expected}
        for key, expected in required.items()
        if baseline.get(key) != expected
    }
    if mismatches:
        raise FlatRagContractError(f"Day 21 基線條件被改動：{mismatches}")
    expected_count = int(baseline["expected_document_count"])
    if len(documents) != expected_count:
        raise FlatRagContractError(
            f"平面知識候選應為 {expected_count} 筆，實際為 {len(documents)} 筆"
        )
    rule_ids = []
    for document in documents:
        if document.get("store_type") != "flat":
            raise FlatRagContractError("Day 21 只能讀取 flat chunks")
        if document.get("retrieval_role") != "search_candidate":
            raise FlatRagContractError("所有文件都必須是 search_candidate")
        metadata = document.get("metadata")
        if not isinstance(metadata, dict):
            raise FlatRagContractError("知識文件缺少 metadata")
        rule_id = metadata.get("rule_id")
        if not isinstance(rule_id, str) or not rule_id:
            raise FlatRagContractError("知識文件缺少 rule_id")
        if (
            not isinstance(document.get("search_text"), str)
            or not document["search_text"].strip()
        ):
            raise FlatRagContractError("知識文件缺少 search_text")
        rule_ids.append(rule_id)
    duplicates = sorted(
        rule_id for rule_id, count in Counter(rule_ids).items() if count > 1
    )
    if duplicates:
        raise FlatRagContractError(f"rule_id 不可重複：{duplicates}")
    return {
        "document_count": len(documents),
        "unique_rule_id_count": len(set(rule_ids)),
        "store_type": "flat",
        "top_k": 3,
        "disabled_components": [
            "hierarchy",
            "bm25",
            "reranker",
            "retrieval_gate",
        ],
    }


def validate_public_cases(
    cases: Sequence[Mapping[str, Any]],
    documents: Sequence[Mapping[str, Any]],
    config: Mapping[str, Any],
) -> JsonObject:
    """Reject patient fields and validate the fixed public integration cases."""

    contract = config["input_contract"]
    if not isinstance(cases, list):
        raise FlatRagContractError("Day 21 cases 必須是 JSON 陣列")
    if len(cases) != int(contract["expected_case_count"]):
        raise FlatRagContractError("Day 21 case 數量與契約不同")
    forbidden = set(contract["forbidden_fields"])
    present_forbidden = sorted(forbidden.intersection(_all_keys(cases)))
    if present_forbidden:
        raise FlatRagContractError(f"公開問題出現禁止欄位：{present_forbidden}")

    document_rule_ids = {
        str(document.get("metadata", {}).get("rule_id")) for document in documents
    }
    case_ids = []
    supported_count = 0
    gap_count = 0
    for case in cases:
        case_id = case.get("case_id")
        question = case.get("question")
        if not isinstance(case_id, str) or not case_id.strip():
            raise FlatRagContractError("case_id 必須是非空字串")
        if not isinstance(question, str) or not question.strip():
            raise FlatRagContractError(f"{case_id}.question 必須是非空字串")
        case_ids.append(case_id)
        if case.get("uses_patient_record") is not False:
            raise FlatRagContractError(f"{case_id} 不得使用病患紀錄")
        if case.get("uses_reference_label") is not False:
            raise FlatRagContractError(f"{case_id} 不得使用參考標籤")
        origin = case.get("origin")
        if not isinstance(origin, str) or "not_patient_record" not in origin:
            raise FlatRagContractError(f"{case_id}.origin 必須標示非病患紀錄")

        status = case.get("expected_status")
        answer_kind = case.get("answer_kind")
        level = case.get("expected_answer_level")
        required = case.get("required_evidence_rule_ids")
        if status not in {"supported", "insufficient_evidence"}:
            raise FlatRagContractError(f"{case_id}.expected_status 不支援")
        if answer_kind not in {"level", "text", "gap"}:
            raise FlatRagContractError(f"{case_id}.answer_kind 不支援")
        if level not in {None, 1, 2, 3, 4, 5}:
            raise FlatRagContractError(f"{case_id}.expected_answer_level 不合法")
        if (
            not isinstance(required, list)
            or len(required) != len(set(required))
            or any(not isinstance(rule_id, str) or not rule_id for rule_id in required)
        ):
            raise FlatRagContractError(
                f"{case_id}.required_evidence_rule_ids 格式錯誤"
            )
        unknown = sorted(set(required) - document_rule_ids)
        if unknown:
            raise FlatRagContractError(f"{case_id} 引用未知規則：{unknown}")
        if status == "supported":
            supported_count += 1
            if not required:
                raise FlatRagContractError(f"{case_id} 支援案例必須指定證據")
            if (answer_kind == "level") != (level is not None):
                raise FlatRagContractError(f"{case_id} 回答型別與預期級數不一致")
        else:
            gap_count += 1
            if answer_kind != "gap" or level is not None or required:
                raise FlatRagContractError(f"{case_id} 知識缺口契約不一致")

    duplicates = sorted(
        case_id
        for case_id, count in Counter(case_ids).items()
        if count > 1
    )
    if duplicates:
        raise FlatRagContractError(f"case_id 不可重複：{duplicates}")
    if supported_count != int(contract["expected_supported_case_count"]):
        raise FlatRagContractError("supported case 數量與契約不同")
    if gap_count != int(contract["expected_gap_case_count"]):
        raise FlatRagContractError("gap case 數量與契約不同")
    return {
        "case_count": len(cases),
        "supported_case_count": supported_count,
        "gap_case_count": gap_count,
        "forbidden_fields_absent": True,
        "patient_records_used": False,
        "reference_labels_used": False,
    }


def build_context(
    ranking: Sequence[Mapping[str, Any]],
    document_by_rule: Mapping[str, Mapping[str, Any]],
    *,
    top_k: int,
) -> List[JsonObject]:
    """Turn top-ranked flat chunks into the only evidence visible to the model."""

    if top_k != 3:
        raise FlatRagContractError("Day 21 固定使用 Top-3")
    if len(ranking) < top_k:
        raise FlatRagContractError("排名筆數少於 Top-3")
    context = []
    for item in ranking[:top_k]:
        rule_id = item.get("rule_id")
        if rule_id not in document_by_rule:
            raise FlatRagContractError(f"排名出現未知 rule_id：{rule_id}")
        document = document_by_rule[str(rule_id)]
        metadata = document["metadata"]
        context.append(
            {
                "rank": len(context) + 1,
                "rule_id": rule_id,
                "topic": metadata["topic"],
                "completeness": metadata["completeness"],
                "source_id": metadata["source_id"],
                "source_url": metadata["source_url"],
                "cosine_similarity": round(float(item["cosine_similarity"]), 8),
                "text": document["search_text"],
            }
        )
    return context


def build_user_message(
    case_id: str,
    question: str,
    context: Sequence[Mapping[str, Any]],
) -> str:
    if not case_id.strip() or not question.strip():
        raise FlatRagContractError("case_id 與 question 都不可為空")
    evidence = json.dumps(list(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 evaluate_answer(
    answer: Mapping[str, Any],
    case_contract: Mapping[str, Any],
    context: Sequence[Mapping[str, Any]],
    *,
    safety_notice: str,
) -> JsonObject:
    """Evaluate observable grounding behavior without pretending to judge all semantics."""

    retrieved = [str(item["rule_id"]) for item in context]
    cited = answer.get("evidence_rule_ids")
    cited_list = cited if isinstance(cited, list) else []
    required = list(case_contract.get("required_evidence_rule_ids", []))
    expected_level = case_contract.get("expected_answer_level")
    checks = {
        "case_id_matches": answer.get("case_id") == case_contract["case_id"],
        "status_matches_expected": answer.get("status")
        == case_contract["expected_status"],
        "answer_level_matches_expected": answer.get("answer_level")
        == expected_level,
        "citations_are_retrieved": set(cited_list).issubset(retrieved),
        "required_evidence_was_retrieved": set(required).issubset(retrieved),
        "required_evidence_was_cited": set(required).issubset(cited_list),
        "insufficient_answer_has_no_level": not (
            answer.get("status") == "insufficient_evidence"
            and answer.get("answer_level") is not None
        ),
        "insufficient_answer_lists_missing_information": not (
            answer.get("status") == "insufficient_evidence"
            and not answer.get("missing_information")
        ),
        "supported_answer_cites_evidence": not (
            answer.get("status") == "supported" and not cited_list
        ),
        "supported_answer_has_no_missing_information": not (
            answer.get("status") == "supported"
            and bool(answer.get("missing_information"))
        ),
        "safety_notice_matches": answer.get("safety_notice") == safety_notice,
    }
    return {
        "checks": checks,
        "passed": all(checks.values()),
        "validation_boundary": (
            "只驗證結構、預期狀態、級數、檢索命中與引用集合；"
            "不把字串檢查宣稱為完整醫療正確性驗證。"
        ),
    }

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

檔案 7：建立 scripts/run_day21_flat_basic_rag.py

驗證來源與模型、建立向量、逐題檢索與生成、執行固定一次的 schema 修復，最後寫出公開摘要與 run manifest。

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

#!/usr/bin/env python3
"""Run the Day 21 flat dense RAG baseline on public KTAS knowledge."""

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

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_user_message,
    evaluate_answer,
    validate_flat_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-21-flat-basic-rag.json"
IMPLEMENTATION_PATHS = (
    "scripts/run_day21_flat_basic_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/representation/embedding.py",
    "src/triage_rag/reproducibility.py",
)
JsonObject = Dict[str, Any]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="以九筆公開 KTAS 平面知識執行 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 21 契約。"
        )


def _verify_sources(config: Mapping[str, Any]) -> List[Path]:
    sources = config["sources"]
    pairs = [
        ("flat_chunks_path", "flat_chunks_sha256", "Day 17 flat chunks"),
        (
            "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_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 模型與 Day 21 契約不同：{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"]
    required_capabilities = set(generation["required_capabilities"])
    runtime_capabilities = set(runtime.get("capabilities") or [])
    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
    }
    if not required_capabilities.issubset(runtime_capabilities):
        mismatches["capabilities"] = {
            "actual": sorted(runtime_capabilities),
            "expected_at_least": sorted(required_capabilities),
        }
    if mismatches:
        raise ValueError(f"本機生成模型與 Day 21 契約不同：{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 _aggregate(case_results: List[JsonObject]) -> 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["evaluation"]["checks"] for item in case_results]
    supported_checks = [item["evaluation"]["checks"] for item in supported]
    return {
        "case_count": len(case_results),
        "passed_case_count": sum(item["evaluation"]["passed"] for item in case_results),
        "supported_case_count": len(supported),
        "supported_status_match_count": sum(
            item["answer"]["status"] == "supported" for item in supported
        ),
        "gap_case_count": len(gaps),
        "gap_refusal_count": sum(
            item["answer"]["status"] == "insufficient_evidence" for item in gaps
        ),
        "required_evidence_case_count": len(supported),
        "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["evaluation"]["passed"] for item in case_results),
    }


def _generate_with_contract_repair(
    *,
    system_prompt: str,
    user_message: str,
    schema: Mapping[str, Any],
    generation: Mapping[str, Any],
) -> tuple[JsonObject, JsonObject]:
    """Retry once with a fixed reminder when local output breaks the schema."""

    maximum_attempts = int(generation["maximum_attempts"])
    if maximum_attempts != 2:
        raise ValueError("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 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") != (
        "author_written_public_knowledge_rag_integration_not_patient_triage_or_clinical_evaluation"
    ):
        raise ValueError("Day 21 scope 不正確")
    source_paths = _verify_sources(config)

    sources = config["sources"]
    documents = read_jsonl(
        resolve_project_path(PROJECT_ROOT, sources["flat_chunks_path"])
    )
    flat_contract = validate_flat_contract(documents, config)
    cases = list(config["cases"])
    case_contract = validate_public_cases(cases, documents, config)

    embedding_config = load_json(
        resolve_project_path(PROJECT_ROOT, sources["embedding_config_path"])
    )
    embedding = embedding_config["embedding"]
    generation = config["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/4：驗證平面知識、六筆公開問題與兩個本機模型。", flush=True)
    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/4：建立九筆文件向量與六筆查詢向量。", flush=True)
    document_vectors, document_embedding_metrics = embed_texts(
        [item["search_text"] for item in documents], 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)
    document_by_rule = {
        item["metadata"]["rule_id"]: item for item in documents
    }

    print("步驟 3/4：逐題執行 Dense Top-3、結構化生成與可觀察檢查。", flush=True)
    case_results = []
    for index, (case, query_vector) in enumerate(
        zip(cases, query_vectors), start=1
    ):
        ranking = rank_documents(query_vector, documents, document_vectors)
        context = build_context(
            ranking,
            document_by_rule,
            top_k=int(config["baseline"]["top_k"]),
        )
        schema = build_response_schema(
            case["case_id"],
            generation["safety_notice"],
            answer_kind=case["answer_kind"],
        )
        answer, generation_metrics = _generate_with_contract_repair(
            system_prompt=system_prompt,
            user_message=build_user_message(
                case["case_id"], case["question"], context
            ),
            schema=schema,
            generation=generation,
        )
        evaluation = evaluate_answer(
            answer,
            case,
            context,
            safety_notice=generation["safety_notice"],
        )
        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": case["required_evidence_rule_ids"],
                "query_vector_sha256": vector_sha256(query_vector),
                "retrieved_context": context,
                "answer": answer,
                "evaluation": evaluation,
                "generation_metrics": generation_metrics,
            }
        )
        state = "通過" if evaluation["passed"] else "未通過"
        print(f"  {index:02d}/06 {case['case_id']}：{state}", flush=True)

    aggregate = _aggregate(case_results)
    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-basic-rag-results.json"

    full_results = {
        "schema_version": 1,
        "experiment_id": config["experiment_id"],
        "scope": config["scope"],
        "aggregate": aggregate,
        "case_results": _round_floats(case_results),
    }
    write_json(full_results_path, full_results)

    trace_case_id = config["outputs"]["trace_case_id"]
    trace = next(
        item for item in case_results if item["case_id"] == trace_case_id
    )
    public_summary = {
        "schema_version": 1,
        "experiment_id": config["experiment_id"],
        "scope": config["scope"],
        "baseline_contract": config["baseline"],
        "input_contract": {**flat_contract, **case_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"],
            "structured_output": "per_case_json_schema",
        },
        "aggregate": aggregate,
        "case_results": _round_floats(case_results),
        "trace_case": _round_floats(trace),
        "checks": {
            "source_hashes_match": True,
            "flat_nine_document_candidate_space": True,
            "dense_top3_only": True,
            "hierarchy_bm25_reranker_and_gate_disabled": True,
            "embedding_and_generator_locks_match_runtime": True,
            "only_author_written_public_knowledge_questions_used": True,
            "patient_rows_and_reference_labels_not_read": True,
            "structured_answers_validated_before_writing": True,
        },
        "limitations": config["limitations"],
        "interpretation": (
            "六筆公開知識整合案例的可觀察檢查；不是病患五級分類、"
            "模型效能比較、臨床正確性或部署安全證據。"
        ),
    }
    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": {
            "documents": document_embedding_metrics,
            "queries": query_embedding_metrics,
            "document_vector_sha256": {
                document["metadata"]["rule_id"]: vector_sha256(vector)
                for document, vector in zip(documents, document_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(manife