From f9a863e2711c80835033467f5c3eb169ecc5c2cc Mon Sep 17 00:00:00 2001 From: LittleYellow Date: Tue, 14 Jul 2026 20:58:30 +0800 Subject: [PATCH] =?UTF-8?q?Phase=203:=20=E6=94=9D=E5=85=A5=E7=AE=A1?= =?UTF-8?q?=E7=B7=9A=20=E2=80=94=20kb.py=20=E5=85=B1=E7=94=A8=E6=A8=A1?= =?UTF-8?q?=E7=B5=84=20+=20ingest.py=EF=BC=88=E9=87=8D=E8=A9=A6/fallback/b?= =?UTF-8?q?ranch+PR=EF=BC=89+=20selfcheck?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- tools/ingest.py | 307 ++++++++++++++++++++++++++++++++++++++ tools/kb.py | 186 +++++++++++++++++++++++ tools/selfcheck_ingest.py | 144 ++++++++++++++++++ 3 files changed, 637 insertions(+) create mode 100644 tools/ingest.py create mode 100644 tools/kb.py create mode 100644 tools/selfcheck_ingest.py diff --git a/tools/ingest.py b/tools/ingest.py new file mode 100644 index 0000000..f9ec135 --- /dev/null +++ b/tools/ingest.py @@ -0,0 +1,307 @@ +"""攝入管線（AGENTS.md §6）：讀 raw/converted 的 Markdown，呼叫本地 Ollama +產出/更新 wiki 頁，重建 index.md、追加 log.md，並以 git branch 準備 PR。 + +用法： + python tools/ingest.py raw/converted/xxx.md [...] # 攝入指定來源 + python tools/ingest.py --all-pending # 攝入所有尚無 summary 的來源 + 共用選項：[--force] [--dry-run] [--root DIR] + +注意：--all-pending 以「目前分支」的 wiki 為準；尚未合併的 ingest PR 不會被看到。 +攝入腳本永遠不直接 commit 到 main（AGENTS.md §1.5）。 +""" +import argparse +import datetime +import json +import pathlib +import subprocess +import sys + +sys.path.insert(0, str(pathlib.Path(__file__).parent)) +import kb + +SUMMARY_PROMPT = """你是金融電子支付 QA 部門知識庫的攝入引擎。閱讀以下來源文件，只輸出一個 JSON 物件，不得有任何其他文字。 + +規則： +- 文字內容用繁體中文；slug 用英文小寫 kebab-case。 +- description 為一句話摘要；tags 為 2 至 6 個英文小寫 kebab-case 標籤。 +- summary_markdown：文件重點摘要（目的、關鍵事實、結論、教訓）。不要逐步驟複製測試案例內容——單檔可 grep 的細節不進 wiki。 +- knowledge_items：值得建頁或更新的項目。entity＝系統/模組/API/法規條目/專案；concept＝跨來源的缺陷模式/測試策略/法遵準則。沒有就給空陣列。 +- facts_markdown：本文件中關於該項目的事實要點（markdown bullet），供合併至該項目頁面。 + +JSON 格式： +{{"title": "...", "description": "...", "slug": "...", "tags": ["..."], + "summary_markdown": "...", + "knowledge_items": [{{"type": "entity 或 concept", "slug": "...", "title": "...", + "description": "...", "tags": ["..."], "facts_markdown": "..."}}]}} + +檔名：{filename} +文件內容： +<<< +{content} +>>>""" + +CHUNK_PROMPT = """以下是一份長文件的第 {i}/{n} 段。請以繁體中文摘要此段重點（markdown bullet），只輸出 JSON：{{"summary_markdown": "..."}} + +<<< +{content} +>>>""" + +MERGE_PROMPT = """你是知識庫維護引擎。以下是既有 wiki 頁面內文，以及來自新來源的事實要點。 +請把新事實整合進頁面（就地編輯：合併、去重；若與既有內容矛盾，兩種說法並陳並標註「⚠ 待查證」）。 +只輸出 JSON：{{"updated_markdown": "...", "updated_description": "..."}} +規則：繁體中文；保留原有結構與仍然有效的內容。 + +頁面標題：{title} +既有內文： +<<< +{body} +>>> +新事實（來源 {source_ref}）： +<<< +{facts} +>>>""" + + +def _s(o, k): + return isinstance(o.get(k), str) and o[k].strip() + + +def _tags_ok(o): + return isinstance(o.get("tags"), list) and o["tags"] and all(isinstance(t, str) for t in o["tags"]) + + +def validate_summary(obj): + for k in ("title", "description", "slug", "summary_markdown"): + if not _s(obj, k): + return f"欄位 {k} 缺漏或非字串" + if not _tags_ok(obj): + return "tags 須為非空字串列表" + if not isinstance(obj.get("knowledge_items"), list): + return "knowledge_items 須為列表" + for it in obj["knowledge_items"]: + if not isinstance(it, dict) or it.get("type") not in ("entity", "concept"): + return "knowledge_item 須為物件且 type 為 entity|concept" + for k in ("slug", "title", "description", "facts_markdown"): + if not _s(it, k): + return f"knowledge_item 欄位 {k} 缺漏或非字串" + if not _tags_ok(it): + return "knowledge_item.tags 須為非空字串列表" + return None + + +def validate_chunk(obj): + return None if _s(obj, "summary_markdown") else "欄位 summary_markdown 缺漏" + + +def validate_merge(obj): + for k in ("updated_markdown", "updated_description"): + if not _s(obj, k): + return f"欄位 {k} 缺漏或非字串" + return None + + +def split_chunks(text, limit): + """依段落邊界切塊，每塊不超過 limit 字元（單一超長段落則硬切）。""" + chunks, cur = [], "" + for para in text.split("\n\n"): + while len(para) > limit: # 單段超長：硬切 + chunks.append(para[:limit]) + para = para[limit:] + if len(cur) + len(para) + 2 > limit and cur: + chunks.append(cur) + cur = para + else: + cur = f"{cur}\n\n{para}" if cur else para + if cur: + chunks.append(cur) + return chunks + + +def summarize(cfg, filename, text): + limit = cfg["limits"]["ingest_chunk_max_chars"] + if len(text) > limit: + parts = split_chunks(text, limit) + partials = [] + for i, part in enumerate(parts, 1): + obj, _ = kb.llm_json(cfg, "ingest", + CHUNK_PROMPT.format(i=i, n=len(parts), content=part), + validate_chunk) + partials.append(f"### 分段 {i} 摘要\n{obj['summary_markdown']}") + text = "（以下為長文件的分段摘要，請據此綜合）\n\n" + "\n\n".join(partials) + return kb.llm_json(cfg, "ingest", + SUMMARY_PROMPT.format(filename=filename, content=text), + validate_summary) + + +def _unique_path(d, slug, sha8): + p = d / f"{slug}.md" + return p if not p.exists() else d / f"{slug}-{sha8}.md" + + +def write_summary_page(root, obj, source_ref, sha8, today): + d = root / kb.PAGE_DIRS["summary"] + p = _unique_path(d, kb.safe_slug(obj["slug"], f"doc-{sha8}"), sha8) + meta = {"type": "summary", "title": obj["title"], "description": obj["description"], + "tags": obj["tags"], "timestamp": today, "sources": [source_ref], + "status": "draft"} + p.write_text(kb.dump_page(meta, obj["summary_markdown"]), encoding="utf-8") + return p + + +def upsert_item(root, cfg, item, source_ref, today, changed): + slug = kb.safe_slug(item["slug"], f"item-{source_ref[-8:]}") + existing = None + for t in ("entity", "concept"): # 既有頁優先（就地編輯），不管本次 LLM 判的型別 + p = root / kb.PAGE_DIRS[t] / f"{slug}.md" + if p.exists(): + existing = p + break + if existing: + meta, body = kb.parse_page(existing.read_text(encoding="utf-8")) + obj, _ = kb.llm_json(cfg, "ingest", + MERGE_PROMPT.format(title=meta["title"], body=body, + source_ref=source_ref, + facts=item["facts_markdown"]), + validate_merge) + meta["description"] = obj["updated_description"] + meta["tags"] = sorted(set(meta["tags"]) | set(item["tags"])) + meta["timestamp"] = today + meta["status"] = "draft" # 內容變動須重新人審 + if source_ref not in meta["sources"]: + meta["sources"].append(source_ref) + existing.write_text(kb.dump_page(meta, obj["updated_markdown"]), encoding="utf-8") + changed.append(f"更新 {existing.relative_to(root).as_posix()}") + else: + p = root / kb.PAGE_DIRS[item["type"]] / f"{slug}.md" + meta = {"type": item["type"], "title": item["title"], + "description": item["description"], "tags": item["tags"], + "timestamp": today, "sources": [source_ref], "status": "draft"} + p.write_text(kb.dump_page(meta, item["facts_markdown"]), encoding="utf-8") + changed.append(f"新增 {p.relative_to(root).as_posix()}") + + +# ---------- git ---------- + +def run_git(root, *args, capture=False): + return subprocess.run(["git", *args], cwd=root, check=True, text=True, + encoding="utf-8", capture_output=capture) + + +def git_preflight(root): + branch = run_git(root, "branch", "--show-current", capture=True).stdout.strip() + if branch != "main": + raise SystemExit(f"須在 main 分支執行（目前：{branch}）") + dirty = [l for l in run_git(root, "status", "--porcelain", capture=True) + .stdout.splitlines() if l.strip() and not l[3:].startswith("raw/")] + if dirty: + raise SystemExit("工作區有 raw/ 以外的未提交變更，請先處理：\n" + "\n".join(dirty)) + + +def referenced_sha8s(root): + refs = set() + for p in kb.iter_pages(root): + meta, _ = kb.parse_page(p.read_text(encoding="utf-8")) + for s in meta.get("sources", []): + refs.add(str(s).rsplit("#", 1)[-1]) + return refs + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("paths", nargs="*", help="raw/converted 下的 Markdown 路徑") + ap.add_argument("--all-pending", action="store_true") + ap.add_argument("--force", action="store_true", help="已攝入過（hash 相同）也重跑") + ap.add_argument("--dry-run", action="store_true", help="只列計畫，不呼叫 LLM、不寫檔") + ap.add_argument("--root", default=str(kb.ROOT)) + a = ap.parse_args(argv) + root = pathlib.Path(a.root).resolve() + cfg = kb.load_config(root) + manifest = json.loads((root / "raw" / "manifest.json").read_text(encoding="utf-8")) + conv = {k: v for k, v in manifest["files"].items() if v["kind"] == "converted"} + + if a.paths: + targets = [] + for p in a.paths: + rel = pathlib.Path(p) + rel = (rel.relative_to(root) if rel.is_absolute() else rel).as_posix() + if rel not in conv: + raise SystemExit(f"{rel} 不在 manifest 的 converted 條目中，請先跑 convert.py") + targets.append(rel) + elif a.all_pending: + done = referenced_sha8s(root) + targets = [k for k, v in conv.items() if v["sha256"][:8] not in done] + else: + ap.error("請指定來源路徑或 --all-pending") + if not a.force: + done = referenced_sha8s(root) + skipped = [t for t in targets if conv[t]["sha256"][:8] in done] + targets = [t for t in targets if conv[t]["sha256"][:8] not in done] + for t in skipped: + print(f"skip: {t} 已有 summary 引用（--force 可重跑）") + if not targets: + print("沒有待攝入的來源。") + return + + today = datetime.date.today().isoformat() + stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") + branch = f"ingest/{stamp}-{kb.safe_slug(pathlib.Path(targets[0]).stem, 'batch')[:30]}" + + if a.dry_run: + limit = cfg["limits"]["ingest_chunk_max_chars"] + print(f"[dry-run] 分支：{branch}，模型：{cfg['tasks']['ingest']['model']}") + for t in targets: + n = len((root / t).read_text(encoding="utf-8")) + print(f"[dry-run] {t}: {n} chars，{max(1, -(-n // limit))} 段") + return + + git_preflight(root) + run_git(root, "checkout", "-q", "-b", branch) + ok, failed, changed = [], [], [] + try: + for t in targets: + try: + text = (root / t).read_text(encoding="utf-8") + sha8 = conv[t]["sha256"][:8] + source_ref = f"{t}#{sha8}" + obj, model = summarize(cfg, pathlib.Path(conv[t]["original_path"]).name, text) + p = write_summary_page(root, obj, source_ref, sha8, today) + changed.append(f"新增 {p.relative_to(root).as_posix()}") + for item in obj["knowledge_items"]: + upsert_item(root, cfg, item, source_ref, today, changed) + ok.append((t, model)) + print(f"ingested: {t}（model={model}）") + except Exception as e: + failed.append((t, e)) + print(f"error: {t}: {e}", file=sys.stderr) + if not ok: + run_git(root, "checkout", "-q", "main") + run_git(root, "branch", "-q", "-D", branch) + raise SystemExit("全部來源攝入失敗，已還原至 main。") + kb.rebuild_index(root) + kb.append_log(root, "ingest", f"攝入 {len(ok)} 份來源（branch {branch}）", + changed + [f"失敗：{t}（{e}）" for t, e in failed]) + run_git(root, "add", "wiki", "index.md", "log.md", "raw") + run_git(root, "commit", "-q", "-m", + f"ingest: {len(ok)} 份來源\n\n" + "\n".join(f"- {c}" for c in changed)) + try: + run_git(root, "push", "-q", "-u", "origin", branch) + remote = run_git(root, "remote", "get-url", "origin", + capture=True).stdout.strip().removesuffix(".git") + print(f"PR 建立網址：{remote}/compare/main...{branch}") + except subprocess.CalledProcessError: + print(f"push 失敗（無 remote 或離線）。分支 {branch} 保留於本地，請手動 push 後開 PR。") + run_git(root, "checkout", "-q", "main") + print(f"完成：{len(ok)} 成功、{len(failed)} 失敗。變更在分支 {branch}，經 PR 審核後合併。") + except SystemExit: + raise + except Exception: + print(f"攝入中斷。目前在分支 {branch}，工作區可能有未提交變更，" + "請人工檢查（不自動清除以免遺失資料）。", file=sys.stderr) + raise + if failed: + raise SystemExit(1) + + +if __name__ == "__main__": + main() diff --git a/tools/kb.py b/tools/kb.py new file mode 100644 index 0000000..a88963f --- /dev/null +++ b/tools/kb.py @@ -0,0 +1,186 @@ +"""共用模組：config 載入驗證、本地 Ollama 呼叫（重試 + fallback）、 +wiki 頁 frontmatter 解析/寫出、index.md 重建、log.md 追加。 + +供 ingest.py / lint.py / search.py / mcp/server.py 使用。 +""" +import datetime +import json +import pathlib +import os +import re +import urllib.request + +import yaml + +ROOT = pathlib.Path(__file__).resolve().parents[1] +PAGE_DIRS = {"summary": "wiki/summaries", "entity": "wiki/entities", "concept": "wiki/concepts"} +PAGE_TYPES = set(PAGE_DIRS) +STATUSES = {"draft", "reviewed", "stale"} +REQUIRED_FM = ("type", "title", "description", "tags", "timestamp", "sources", "status") + + +# ---------- config ---------- + +def load_config(root=ROOT): + """讀取 config/models.yaml。缺欄位報錯，不使用隱含預設值（AGENTS.md §1.2）。""" + path = root / "config" / "models.yaml" + cfg = yaml.safe_load(path.read_text(encoding="utf-8")) + try: + cfg["ollama"]["base_url"] + cfg["ollama"]["timeout_seconds"] + for task in ("ingest", "lint"): + for key in ("model", "temperature", "max_retries"): + cfg["tasks"][task][key] + cfg["tasks"]["fallback"]["model"] + cfg["limits"]["mcp_response_max_tokens"] + cfg["limits"]["ingest_chunk_max_chars"] + except (KeyError, TypeError) as e: + raise SystemExit(f"config/models.yaml 缺欄位：{e}") + for task, spec in cfg["tasks"].items(): + if spec["model"] == "CHANGE_ME": + raise SystemExit(f"config/models.yaml：tasks.{task}.model 尚未設定") + if spec["model"].endswith(":cloud"): + raise SystemExit(f"tasks.{task}.model 為 :cloud 模型，違反資料落地（AGENTS.md §1.1）") + # 環境變數是唯一允許的覆寫來源（AGENTS.md §1.2） + cfg["ollama"]["base_url"] = os.environ.get("OLLAMA_BASE_URL", cfg["ollama"]["base_url"]) + return cfg + + +# ---------- Ollama ---------- + +def ollama_chat(cfg, model, prompt, temperature, json_format=False): + """單次呼叫本地 Ollama /api/chat（stdlib urllib，不需額外依賴）。""" + body = { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "stream": False, + "options": {"temperature": temperature}, + } + if json_format: + body["format"] = "json" + req = urllib.request.Request( + cfg["ollama"]["base_url"].rstrip("/") + "/api/chat", + data=json.dumps(body).encode("utf-8"), + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=cfg["ollama"]["timeout_seconds"]) as r: + return json.loads(r.read())["message"]["content"] + + +def llm_json(cfg, task, prompt, validate): + """要求 JSON 輸出的呼叫：主模型重試 max_retries 次，仍失敗切 fallback 模型。 + + 本地模型輸出不穩定是預期情況（AGENTS.md 前提），故每次失敗把錯誤附回 + prompt 再試。validate(obj) 回傳錯誤訊息字串或 None。 + 回傳 (obj, 實際使用的 model)。全部失敗拋 RuntimeError。 + """ + spec = cfg["tasks"][task] + attempts = [] + for model in (spec["model"], cfg["tasks"]["fallback"]["model"]): + hint = "" + for _ in range(spec["max_retries"]): + try: + raw = ollama_chat(cfg, model, prompt + hint, spec["temperature"], + json_format=True) + obj = json.loads(raw) + err = validate(obj) + if err is None: + return obj, model + except Exception as e: + err = f"{type(e).__name__}: {e}" + attempts.append(f"{model}: {err}") + hint = f"\n\n注意：上次輸出無效（{err}）。請只輸出符合規格的 JSON，不得有其他文字。" + raise RuntimeError("LLM 輸出驗證失敗（含 fallback）：" + " | ".join(attempts[-4:])) + + +# ---------- wiki 頁 ---------- + +def parse_page(text): + """回傳 (frontmatter dict, body)。格式不符拋 ValueError。""" + m = re.match(r"^---\r?\n(.*?)\r?\n---\r?\n(.*)$", text, re.S) + if not m: + raise ValueError("缺少 YAML frontmatter") + meta = yaml.safe_load(m.group(1)) + if not isinstance(meta, dict): + raise ValueError("frontmatter 不是 mapping") + return meta, m.group(2).strip() + + +def dump_page(meta, body): + fm = yaml.safe_dump(meta, allow_unicode=True, sort_keys=False) + return f"---\n{fm}---\n\n{body.strip()}\n" + + +def validate_meta(meta): + """依 AGENTS.md §3.2 檢查 frontmatter，回傳錯誤訊息列表。""" + errs = [f"缺欄位 {k}" for k in REQUIRED_FM if k not in meta] + if errs: + return errs + if meta["type"] not in PAGE_TYPES: + errs.append(f"type 不合法：{meta['type']}") + if meta["status"] not in STATUSES: + errs.append(f"status 不合法：{meta['status']}") + if not isinstance(meta["tags"], list) or not meta["tags"]: + errs.append("tags 須為非空列表") + if not isinstance(meta["sources"], list) or not meta["sources"]: + errs.append("sources 須為非空列表") + for s in meta.get("sources") or []: + if not re.match(r"^raw/(converted|originals)/.+#[0-9a-f]{8}$", str(s)): + errs.append(f"source_ref 格式不符：{s}") + if not re.match(r"^\d{4}-\d{2}-\d{2}$", str(meta["timestamp"])): + errs.append(f"timestamp 須為 YYYY-MM-DD：{meta['timestamp']}") + return errs + + +def safe_slug(s, fallback): + s = re.sub(r"[^a-z0-9]+", "-", str(s).lower()).strip("-")[:60] + return s or fallback + + +def iter_pages(root=ROOT): + for sub in PAGE_DIRS.values(): + d = root / sub + if d.is_dir(): + yield from sorted(d.glob("*.md")) + + +# ---------- index.md / log.md ---------- + +def rebuild_index(root=ROOT): + """由全部 wiki 頁 frontmatter 決定性重建 index.md（AGENTS.md §10）。""" + groups = {"summary": [], "entity": [], "concept": []} + for p in iter_pages(root): + meta, _ = parse_page(p.read_text(encoding="utf-8")) + rel = p.relative_to(root).as_posix() + tags = " ".join(f"#{t}" for t in meta["tags"]) + groups[meta["type"]].append( + (str(meta["title"]), f"- [{meta['title']}]({rel}) — {meta['description']} ｜ {tags}")) + def section(key): + lines = [line for _, line in sorted(groups[key])] + return "\n".join(lines) if lines else "（尚無頁面）" + text = f"""# 知識庫索引 + + + +## 摘要頁（summaries） + +{section('summary')} + +## 實體頁（entities） + +{section('entity')} + +## 概念頁（concepts） + +{section('concept')} +""" + (root / "index.md").write_text(text, encoding="utf-8") + + +def append_log(root, op, title, lines): + entry = f"\n## [{datetime.date.today().isoformat()}] {op} | {title}\n\n" + entry += "".join(f"- {l}\n" for l in lines) + with open(root / "log.md", "a", encoding="utf-8") as f: + f.write(entry) diff --git a/tools/selfcheck_ingest.py b/tools/selfcheck_ingest.py new file mode 100644 index 0000000..b2914f2 --- /dev/null +++ b/tools/selfcheck_ingest.py @@ -0,0 +1,144 @@ +"""Phase 3 自檢：以假 LLM + 沙盒 git repo 驗證攝入管線的完整 plumbing—— +分支建立、頁面產出、frontmatter 合規、index/log 更新、冪等跳過、就地編輯合併， +以及 kb.llm_json 的重試與 fallback 切換。不需要 Ollama 在線。 + +執行：.venv/Scripts/python tools/selfcheck_ingest.py +""" +import hashlib +import pathlib +import shutil +import subprocess +import sys +import tempfile + +sys.path.insert(0, str(pathlib.Path(__file__).parent)) +import kb +import ingest + +CFG = """ollama: {base_url: "http://localhost:11434", timeout_seconds: 5} +tasks: + ingest: {model: "fake:test", temperature: 0.2, max_retries: 2} + lint: {model: "fake:test", temperature: 0.1, max_retries: 2} + fallback: {model: "fake:fb"} +limits: {mcp_response_max_tokens: 2000, ingest_chunk_max_chars: 200} +""" + + +def fake_llm(cfg, task, prompt, validate): + if prompt.startswith("以下是一份長文件的第"): + obj = {"summary_markdown": "- 分段重點"} + elif '"updated_markdown"' in prompt: + obj = {"updated_markdown": "- 壓測逾時\n- 新來源事實", + "updated_description": "支付閘道模組（已更新）"} + else: + obj = {"title": "支付閘道壓測報告", "description": "壓測發現逾時缺陷", + "slug": "gateway-load-test", "tags": ["load-test", "gateway"], + "summary_markdown": "- 逾時缺陷於壓測重現", + "knowledge_items": [{"type": "entity", "slug": "payment-gateway", + "title": "支付閘道", "description": "支付閘道模組", + "tags": ["gateway"], "facts_markdown": "- 壓測逾時"}]} + err = validate(obj) # 假輸出也要通過真 validator，確保契約一致 + assert err is None, err + return obj, "fake:test" + + +def git(root, *args): + return subprocess.run(["git", *args], cwd=root, check=True, text=True, + encoding="utf-8", capture_output=True).stdout.strip() + + +def add_doc(root, name, text): + p = root / "raw" / "converted" / name + p.write_text(text, encoding="utf-8") + sha = hashlib.sha256(text.encode("utf-8")).hexdigest() + import json + mp = root / "raw" / "manifest.json" + m = json.loads(mp.read_text(encoding="utf-8")) + m["files"][f"raw/converted/{name}"] = { + "kind": "converted", "sha256": sha, "original_path": f"raw/originals/{name}", + "original_sha256": sha, "converter": "from_docx", "converted_at": "2026-07-14T00:00:00"} + mp.write_text(json.dumps(m, ensure_ascii=False, indent=2), encoding="utf-8") + return sha + + +def main(): + tmp = pathlib.Path(tempfile.mkdtemp(prefix="ppqa-ingest-")) + real_llm, real_chat = kb.llm_json, kb.ollama_chat + try: + root = tmp / "repo" + for sub in ("config", "raw/originals", "raw/converted", + "wiki/summaries", "wiki/entities", "wiki/concepts"): + (root / sub).mkdir(parents=True) + (root / "config" / "models.yaml").write_text(CFG, encoding="utf-8") + (root / "raw" / "manifest.json").write_text('{"version": 1, "files": {}}', + encoding="utf-8") + (root / "index.md").write_text("# 知識庫索引\n", encoding="utf-8") + (root / "log.md").write_text("# 操作日誌\n", encoding="utf-8") + git(root, "init", "-q", "-b", "main") + git(root, "config", "user.name", "selfcheck") + git(root, "config", "user.email", "selfcheck@local") + text = "支付閘道於壓力測試下出現逾時缺陷，交易峰值時重試機制未生效。\n\n" * 12 + add_doc(root, "doc1.md", text) # > 200 chars → 走分段路徑 + git(root, "add", "-A") + git(root, "commit", "-q", "-m", "init") + + # 1) 攝入：建分支、產頁、commit、回到 main + kb.llm_json = fake_llm + ingest.main(["raw/converted/doc1.md", "--root", str(root)]) + assert git(root, "branch", "--show-current") == "main" + branch = git(root, "branch", "--list", "ingest/*").strip("* ").strip() + assert branch, "未建立 ingest 分支" + assert not git(root, "status", "--porcelain"), "工作區不乾淨" + git(root, "checkout", "-q", branch) + summary = list((root / "wiki/summaries").glob("*.md")) + entity = root / "wiki/entities/payment-gateway.md" + assert len(summary) == 1 and entity.is_file() + meta, _ = kb.parse_page(entity.read_text(encoding="utf-8")) + assert kb.validate_meta(meta) == [], kb.validate_meta(meta) + assert "payment-gateway" in (root / "index.md").read_text(encoding="utf-8") + assert "] ingest |" in (root / "log.md").read_text(encoding="utf-8") + print("PASS: 攝入 → 分支 + summary/entity 頁 + index/log") + + # 2) 模擬 PR 合併後：冪等跳過；新來源觸發既有頁就地編輯 + git(root, "checkout", "-q", "main") + git(root, "merge", "-q", branch) + ingest.main(["raw/converted/doc1.md", "--root", str(root)]) # 應 skip + assert git(root, "branch", "--show-current") == "main" + add_doc(root, "doc2.md", "第二份文件：支付閘道逾時已修復並通過回歸測試。") + git(root, "add", "-A") + git(root, "commit", "-q", "-m", "doc2") + ingest.main(["raw/converted/doc2.md", "--root", str(root)]) + b2 = [b.strip("* ").strip() for b in + git(root, "branch", "--list", "ingest/*").splitlines() if branch not in b] + git(root, "checkout", "-q", b2[0]) + meta, body = kb.parse_page(entity.read_text(encoding="utf-8")) + assert len(meta["sources"]) == 2 and "已更新" in meta["description"], meta + assert "新來源事實" in body + git(root, "checkout", "-q", "main") + print("PASS: 冪等跳過 + 既有 entity 頁就地編輯（sources 追加）") + + # 3) kb.llm_json：重試後切 fallback；全失敗拋錯 + kb.llm_json = real_llm + cfg = kb.load_config(root) + calls = {"n": 0} + def flaky(cfg_, model, prompt, temperature, json_format=False): + calls["n"] += 1 + return "not-json" if calls["n"] < 3 else '{"summary_markdown": "ok"}' + kb.ollama_chat = flaky + obj, model = kb.llm_json(cfg, "ingest", "p", ingest.validate_chunk) + assert obj["summary_markdown"] == "ok" and model == "fake:fb" and calls["n"] == 3 + kb.ollama_chat = lambda *a, **k: "junk" + try: + kb.llm_json(cfg, "ingest", "p", ingest.validate_chunk) + raise AssertionError("應拋 RuntimeError") + except RuntimeError as e: + assert "fallback" in str(e) + print("PASS: 重試 + fallback 切換 + 全失敗報錯") + print("ALL PASS") + finally: + kb.llm_json, kb.ollama_chat = real_llm, real_chat + shutil.rmtree(tmp, ignore_errors=True) + + +if __name__ == "__main__": + main()