Compare commits

...

2 Commits

Author SHA1 Message Date
LittleYellow
cfd1876359 轉換工具強化:失敗診斷訊息 + 舊版格式升版 + 掃描 PDF vision fallback
- convert.py:轉換失敗改印例外型別、批次尾端彙總、--traceback 旗標;
  空白轉換結果示警(避免純圖片 docx 等假成功靜默入庫)
- from_legacy.py(B):.doc/.xls/.ppt 經本地 LibreOffice headless 升版為
  現代格式後再走既有轉換器;raw/originals 仍保存真正的舊格式原始檔
- to_image.py + from_vision.py(A):needs_ocr 的掃描/圖片型 PDF 逐頁
  render → 本地 vision 模型轉錄(--vision);:cloud/未設定防護,
  模型設定走 config/models.yaml 的 tasks.vision(預設不啟用)
- kb.ollama_chat 加 images 參數(向後相容);models.yaml 加 vision 範例
- selfcheck 補 legacy 路由、to_image 真測、from_vision 注入測、
  --vision 端到端,全數 ALL PASS

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 07:00:32 +08:00
LittleYellow
1d1454e1b9 docs: 新增 README 使用手冊
操作導向的安裝/使用說明(環境、四大工作流、MCP、疑難排解);
schema 與規則仍以 AGENTS.md 為正本,不複製其內容(§11 雙棧規則)。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 07:00:21 +08:00
8 changed files with 635 additions and 37 deletions

207
README.md Normal file
View File

@@ -0,0 +1,207 @@
# README — QA 知識庫使用手冊
金融電子支付 QA 部門知識庫,採 **Karpathy LLM Wiki 模式**:知識在文件**攝入時**由本地
LLM 編譯成 wiki 頁,查詢時只讀已編譯頁面(**非 RAG**,檢索用 BM25
> **本檔是操作手冊(怎麼裝、怎麼跑)。** 所有 schema、工作流細節、硬性約束與行為紀律的
> **唯一正本**是 [AGENTS.md](AGENTS.md)——動手前請先讀它。本檔只做操作說明,不複製其內容。
---
## 1. 環境需求
| 項目 | 需求 |
|------|------|
| Python | 3.11+ |
| 本地 Ollama | 執行於 `http://localhost:11434`(見 [config/models.yaml](config/models.yaml) |
| Ollama 模型 | 已 `ollama pull` 且與 `config/models.yaml` 的 tag 一致的**合規本地模型** |
| Git remote選用 | 攝入自動開 PR 用;無 remote 時分支保留於本地 |
**資料落地是硬性約束([AGENTS.md](AGENTS.md) §1.1**:所有文件內容的 LLM 處理一律走本地
Ollama**禁止**任何雲端 API、雲端轉換服務及任何 `:cloud` 後綴模型。供應鏈限制§1.3
禁用中國關聯模型bge / Qwen / GLM / MiniMax / Jina 等)。
---
## 2. 安裝
Windows PowerShell本專案主要環境
```powershell
python -m venv .venv
.venv\Scripts\pip install -r requirements.txt
```
Linux / Gitea runner
```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements.txt
```
接著確認 Ollama 已拉好 `config/models.yaml` 內指定的模型:
```powershell
ollama list # 對照 tasks.ingest.model / tasks.lint.model / tasks.fallback.model
```
**換模型只改 [config/models.yaml](config/models.yaml),不動任何程式碼**§1.2)。設定缺欄位時程式直接
報錯,不使用隱含預設值。唯一允許的覆寫來源是環境變數 `OLLAMA_BASE_URL`
> 下文指令以 Windows 的 `.venv\Scripts\python` 為例Linux 換成 `.venv/bin/python`。
---
## 3. 四大工作流
資料流:**原始檔 → `convert``raw/converted/` Markdown → `ingest``wiki/` 頁面 →
`search` / MCP 查詢**`lint` 定期健檢整庫。
### 3.1 轉換 `convert` — 原始檔轉 Markdown
把 docx / xlsx / pdf / pptx / html 或網頁 URL 轉成 Markdown並登記 SHA-256 至
`raw/manifest.json``raw/` 只讀不可變§1.4)。
```powershell
# 本機檔案(可多個)
.venv\Scripts\python tools\convert\convert.py 報告.docx 測項.xlsx 規格.pdf
# 網頁快照(保存完整 HTML因來源可能消失
.venv\Scripts\python tools\convert\convert.py https://example.com/guideline
# 只看計畫、不寫檔
.venv\Scripts\python tools\convert\convert.py 報告.docx --dry-run
```
- 支援:`.docx .xlsx .pdf .pptx .html/.htm``http(s)://` URL。
- hash 相同的檔會自動 `skip`(去重)。
- 掃描件PDF 無文字層)標記 `needs_ocr` 並跳過,**不擅自 OCR**,留待人工決定。
- 單檔失敗不中斷批次,最後彙總並以結束碼 1 回報。
### 3.2 攝入 `ingest` — 編譯成 wiki 頁
`raw/converted/` 的 Markdown呼叫本地 Ollama 產出 / 更新 wiki 頁,重建
`index.md`、追加 `log.md`,並**開新 git 分支 + PR**HITL 閘門§1.5——永不直接 commit 到 main
```powershell
# 攝入指定來源
.venv\Scripts\python tools\ingest.py raw\converted\xxx.md
# 攝入所有「尚無 summary 頁引用」的來源
.venv\Scripts\python tools\ingest.py --all-pending
# 只列計畫(分支名、模型、分段數),不呼叫 LLM、不寫檔
.venv\Scripts\python tools\ingest.py --all-pending --dry-run
# 已攝入過hash 相同)也強制重跑
.venv\Scripts\python tools\ingest.py raw\converted\xxx.md --force
```
**前置條件**:須在 `main` 分支、且工作區除 `raw/` 外無未提交變更,否則會報錯中止。
執行後:變更落在 `ingest/<時間戳>-<slug>` 分支。有 remote 時自動 push 並印出 PR 建立網址;
無 remote / 離線時分支保留本地,請手動 push 後開 PR。**經人工審核合併後才進 main。**
### 3.3 健檢 `lint` — 全庫品質檢查
檢查 schema 完整性、過期主張、覆蓋缺口、孤兒頁,並用 LLM 偵測重複 / 矛盾頁。
**絕不修改或刪除任何檔案**一律產出報告列為「待人工核准」§1.5、§7
```powershell
# 完整健檢,報告寫入 reports/
.venv\Scripts\python tools\lint.py --output reports
# 跳過 LLM 重複/矛盾比對(較快,純 schema/結構檢查)
.venv\Scripts\python tools\lint.py --no-llm
```
- 報告路徑印在 stdout例如 `reports/lint-YYYYMMDD-HHMMSS.md`
- **結束碼**`0` = 無發現;`1` = 有待人工核准項目(供 CI 判斷 / 標紅通知)。
- 排程:[.gitea/workflows/lint.yaml](.gitea/workflows/lint.yaml)(每週一台北時間 05:00草稿部署 Gitea 時啟用)。
### 3.4 查詢 `search` — BM25 檢索
```powershell
.venv\Scripts\python tools\search.py "AML 測試準則" -k 10
```
輸出 JSON 陣列(`path / title / description / tags / score`),供人工或 agent 定位候選頁。
`-k` 上限 10查詢工作流 §8先讀 `index.md` 定位、最多 10 頁、只讀命中頁、回答必附
`source_ref`、禁止全庫掃描)。
---
## 4. MCP server — 供內部 AI agent / 團隊查詢
[mcp/server.py](mcp/server.py) 是 FastMCP 薄殼,只暴露兩個唯讀 tool`search_wiki`BM25 檢索)與
`read_page`(讀單頁全文,依 `models.yaml` 的 token 上限截斷)。只讀已編譯 wiki 頁,不觸碰
`raw/` 層、不做全庫掃描。
```powershell
# stdio 啟動
.venv\Scripts\python mcp\server.py
```
測試時可用環境變數 `PP_QA_ROOT` 指定專案根目錄。`read_page` 有路徑跳脫防護,只接受
`wiki/` 下的 `.md`
---
## 5. Copilot Custom Agents雙棧
同一套 [AGENTS.md](AGENTS.md) 規範同時服務 Claude Code 與 GitHub Copilot§11 雙棧規則)。
[.github/agents/](.github/agents/) 提供三個 Copilot 編排 agent——它們**只編排、不處理文件內容**
(成本紀律 §1.6LLM 處理一律 shell out 給本地腳本agent 只讀精簡結果):
| Agent | 職責 |
|-------|------|
| [kb-ingest](.github/agents/kb-ingest.agent.md) | 轉換 → 攝入 → 開 PR |
| [kb-lint](.github/agents/kb-lint.agent.md) | 執行健檢並摘要報告重點 |
| [kb-query](.github/agents/kb-query.agent.md) | 檢索 → 只讀命中頁 → 附 source_ref 回答 |
---
## 6. 目錄速查
```
config/models.yaml # Ollama 端點與模型 tag唯一設定來源
raw/originals/ # 原始檔,只讀
raw/converted/ # 轉換後 Markdown攝入的實際輸入只讀
raw/manifest.json # SHA-256 登記 + 原始檔↔轉換檔對應
wiki/summaries/ # 每份來源一頁摘要
wiki/entities/ # 實體頁(系統/模組/API/法規/專案)
wiki/concepts/ # 概念頁(跨來源綜合)
index.md # 目錄式導航(由 ingest 自動重建)
log.md # append-only 操作日誌
tools/ # convert/ ingest.py lint.py search.py kb.py共用模組
mcp/server.py # MCP 薄殼
```
完整結構與各檔用途見 [AGENTS.md](AGENTS.md) §2。
---
## 7. 疑難排解
| 訊息 / 現象 | 原因與處置 |
|-------------|-----------|
| `config/models.yaml 缺欄位` | 設定不完整補齊欄位不要用預設值繞過§1.2)。 |
| `tasks.X.model 為 :cloud 模型` | 違反資料落地;改用本地模型 tag§1.1)。 |
| `須在 main 分支執行` | 攝入前先切回 `main`。 |
| `工作區有 raw/ 以外的未提交變更` | 先 commit / stash 非 `raw/` 的變更再攝入。 |
| `... 不在 manifest 的 converted 條目中` | 該來源尚未轉換;先跑 `convert.py`。 |
| `LLM 輸出驗證失敗(含 fallback` | 本地模型連主模型 + fallback 都無法產出合規 JSON檢查 Ollama 是否在線、模型是否拉好。 |
| `push 失敗(無 remote 或離線)` | 正常;分支已保留本地,手動 push 後開 PR。 |
---
## 8. 開發者自檢
各工具附 assert 式自檢腳本(免框架、免 fixtures§13.4
```powershell
.venv\Scripts\python tools\convert\selfcheck.py
.venv\Scripts\python tools\selfcheck_ingest.py
.venv\Scripts\python tools\selfcheck_lint.py
.venv\Scripts\python tools\selfcheck_search.py
```

View File

@@ -20,6 +20,13 @@ tasks: # 依任務指定模型,全部可調
max_retries: 3 max_retries: 3
fallback: fallback:
model: "hermes3:8b" # 主模型格式驗證連續失敗時切換 model: "hermes3:8b" # 主模型格式驗證連續失敗時切換
# visionconvert --vision 的掃描/圖片型 PDF fallback 才會用到):預設不啟用。
# 要啟用時取消下方註解並填入「合規本地 vision 模型」——須自行確認非中國關聯、
# 非 :cloud§1.1/§1.3)。留成註解可避免 load_config 對未設定值報錯。
# vision:
# model: "CHANGE_ME" # 例Google Gemma 視覺版 / IBM Granite Vision 等合規本地模型
# temperature: 0.1
# max_retries: 2
limits: limits:
mcp_response_max_tokens: 2000 mcp_response_max_tokens: 2000

View File

@@ -4,7 +4,9 @@
原始檔存入 raw/originals/ 並登記 hash → 轉換 → Markdown 存入 raw/converted/ 原始檔存入 raw/originals/ 並登記 hash → 轉換 → Markdown 存入 raw/converted/
並登記 hash 與原始檔對應 → 之後攝入管線只讀 converted 層。 並登記 hash 與原始檔對應 → 之後攝入管線只讀 converted 層。
用法python tools/convert/convert.py <檔案路徑或URL>... [--dry-run] [--root DIR] 用法python tools/convert/convert.py <檔案路徑或URL>... [--vision] [--dry-run] [--root DIR]
支援 .docx/.xlsx/.pdf/.pptx/.html 與 URL舊版 .doc/.xls/.ppt 經 LibreOffice 升版;
--vision 讓掃描/圖片型 PDF 走本地 vision 模型轉錄(見 from_vision.py
""" """
import argparse import argparse
import datetime import datetime
@@ -12,11 +14,15 @@ import hashlib
import json import json
import pathlib import pathlib
import re import re
import shutil
import sys import sys
import tempfile
import traceback
import urllib.parse import urllib.parse
sys.path.insert(0, str(pathlib.Path(__file__).parent)) sys.path.insert(0, str(pathlib.Path(__file__).parent))
import from_docx import from_docx
import from_legacy
import from_pdf import from_pdf
import from_pptx import from_pptx
import from_web import from_web
@@ -29,6 +35,10 @@ CONVERTERS = {
".pdf": (from_pdf.convert, "from_pdf", "pdf"), ".pdf": (from_pdf.convert, "from_pdf", "pdf"),
".pptx": (from_pptx.convert, "from_pptx", "pptx"), ".pptx": (from_pptx.convert, "from_pptx", "pptx"),
} }
# 舊版 OLE 二進位 → 現代格式(經 LibreOffice 升版後走上表對應轉換器)。
# <!-- ASSUMPTION: media_type 記真實舊格式doc/xls/ppt擴充 §9 的列舉;
# converter 記為 "from_docx+libreoffice" 等,保留兩段式 provenance。 -->
LEGACY_MAP = {".doc": ".docx", ".xls": ".xlsx", ".ppt": ".pptx"}
def _now(): def _now():
@@ -72,6 +82,9 @@ def _dest_name(dir_, name, sha):
def _register_pair(m, root, orig_rel, orig_entry, md_rel, md_text, converter, dry): def _register_pair(m, root, orig_rel, orig_entry, md_rel, md_text, converter, dry):
if not md_text.strip():
# 空轉換結果幾乎都是失敗(如純圖片 docx登記前示警避免空頁靜默入庫
print(f"warning: {orig_rel} 轉換結果為空白,請人工檢查來源", file=sys.stderr)
if not dry: if not dry:
md_path = root / md_rel md_path = root / md_rel
md_path.write_text(md_text, encoding="utf-8") md_path.write_text(md_text, encoding="utf-8")
@@ -87,49 +100,96 @@ def _register_pair(m, root, orig_rel, orig_entry, md_rel, md_text, converter, dr
print(f"converted: {orig_rel} -> {md_rel}" + (" [dry-run]" if dry else "")) print(f"converted: {orig_rel} -> {md_rel}" + (" [dry-run]" if dry else ""))
def process_local(root, m, path, dry): def _vision_pdf(root, pdf_path, assets, conv_name):
"""--vision fallbackAGENTS.md §3 稽核原則needs_ocr 的 PDF 逐頁 render →
本地 vision 模型轉錄。回傳 (markdown, status, converter)。
成功 → ('...', 'ok', 'from_vision')vision 執行期失敗 → 降級回 needs_ocr不遺失檔案
設定錯誤(未設 vision / :cloud由 from_vision 拋 SystemExit向上傳播不靜默降級。
"""
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) # tools/ 供 import kb
import kb
import to_image
import from_vision
cfg = kb.load_config(root)
md_dir = root / "raw" / "converted"
try:
images = to_image.render_pdf_pages(pdf_path, assets)
texts, model = from_vision.transcribe(cfg, images)
except SystemExit:
raise
except Exception as e:
print(f"warning: {pathlib.Path(pdf_path).name} vision 轉錄失敗,降級為 needs_ocr"
f"{type(e).__name__}: {e}", file=sys.stderr)
return "", "needs_ocr", conv_name
blocks = []
for i, (img, text) in enumerate(zip(images, texts), 1):
rel = img.relative_to(md_dir).as_posix()
blocks.append(f"<!-- page {i} (vision: {model}) -->\n\n![page {i}]({rel})\n\n{text}")
return "\n\n".join(blocks), "ok", "from_vision"
def process_local(root, m, path, dry, vision=False):
src = pathlib.Path(path).resolve() src = pathlib.Path(path).resolve()
if not src.is_file(): if not src.is_file():
raise FileNotFoundError(src) raise FileNotFoundError(src)
ext = src.suffix.lower() ext = src.suffix.lower()
if ext in (".html", ".htm"):
conv, conv_name, media = (
lambda p, assets_dir=None, md_dir=None: from_web.extract(
pathlib.Path(p).read_text(encoding="utf-8", errors="replace")),
"from_web", "html")
elif ext in CONVERTERS:
conv, conv_name, media = CONVERTERS[ext]
else:
raise ValueError(f"不支援的格式:{src.name}")
sha = hashlib.sha256(src.read_bytes()).hexdigest() sha = hashlib.sha256(src.read_bytes()).hexdigest()
if _find_by_sha(m, sha): if _find_by_sha(m, sha):
print(f"skip: {src.name} 已登記hash 相同)") print(f"skip: {src.name} 已登記hash 相同)")
return return
# raw/originals 一律保存「真正的原始檔」(舊格式也是),維持 provenance§1.4
originals = root / "raw" / "originals" originals = root / "raw" / "originals"
if src.parent == originals.resolve(): dest = src if src.parent == originals.resolve() else originals / _dest_name(originals, src.name, sha)
dest = src
else:
dest = originals / _dest_name(originals, src.name, sha)
if not dry and not dest.exists():
dest.write_bytes(src.read_bytes())
orig_rel = f"raw/originals/{dest.name}" orig_rel = f"raw/originals/{dest.name}"
orig_entry = {"kind": "original", "sha256": sha, "media_type": media,
"added_at": _now(), "status": "converted"}
md_name = f"{dest.stem}-{sha[:8]}.md" tmpdir = None
assets = None if dry else root / "raw" / "converted" / "assets" / f"{dest.stem}-{sha[:8]}" try:
md_text, status = conv(str(src), assets_dir=assets, md_dir=root / "raw" / "converted") if ext in LEGACY_MAP:
if status == "needs_ocr": if dry:
orig_entry["status"] = "needs_ocr" print(f"legacy: {src.name} 需 LibreOffice 升版為 {LEGACY_MAP[ext]}dry-run 不執行)")
if not dry: return
m["files"][orig_rel] = orig_entry tmpdir = tempfile.mkdtemp(prefix="ppqa-legacy-")
print(f"needs_ocr: {orig_rel} 無文字層,已登記並跳過(不擅自 OCR" upgraded = from_legacy.upgrade(src, tmpdir) # 找不到 soffice / 升版失敗 → RuntimeError
+ (" [dry-run]" if dry else "")) conv, base_name, _ = CONVERTERS[upgraded.suffix.lower()]
return conv_name, media, conv_src = f"{base_name}+libreoffice", ext.lstrip("."), upgraded
_register_pair(m, root, orig_rel, orig_entry, f"raw/converted/{md_name}", elif ext in (".html", ".htm"):
md_text, conv_name, dry) conv = (lambda p, assets_dir=None, md_dir=None: from_web.extract(
pathlib.Path(p).read_text(encoding="utf-8", errors="replace")))
conv_name, media, conv_src = "from_web", "html", src
elif ext in CONVERTERS:
conv, conv_name, media = CONVERTERS[ext]
conv_src = src
else:
raise ValueError(f"不支援的格式:{src.name}")
if not dry and dest is not src and not dest.exists():
dest.write_bytes(src.read_bytes())
orig_entry = {"kind": "original", "sha256": sha, "media_type": media,
"added_at": _now(), "status": "converted"}
md_name = f"{dest.stem}-{sha[:8]}.md"
assets = None if dry else root / "raw" / "converted" / "assets" / f"{dest.stem}-{sha[:8]}"
md_text, status = conv(str(conv_src), assets_dir=assets,
md_dir=root / "raw" / "converted")
if status == "needs_ocr":
if vision and not dry:
md_text, status, conv_name = _vision_pdf(root, conv_src, assets, conv_name)
if status == "needs_ocr":
orig_entry["status"] = "needs_ocr"
if not dry:
m["files"][orig_rel] = orig_entry
print(f"needs_ocr: {orig_rel} 無文字層,已登記並跳過"
+ ("--vision 轉錄未成功)" if vision and not dry else "(不擅自 OCR")
+ (" [dry-run]" if dry else ""))
return
_register_pair(m, root, orig_rel, orig_entry, f"raw/converted/{md_name}",
md_text, conv_name, dry)
finally:
if tmpdir:
shutil.rmtree(tmpdir, ignore_errors=True)
def process_url(root, m, url, dry): def process_url(root, m, url, dry):
@@ -164,6 +224,11 @@ def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__) ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("inputs", nargs="+", help="檔案路徑或 http(s) URL") ap.add_argument("inputs", nargs="+", help="檔案路徑或 http(s) URL")
ap.add_argument("--dry-run", action="store_true") ap.add_argument("--dry-run", action="store_true")
ap.add_argument("--vision", action="store_true",
help="掃描/圖片型 PDFneeds_ocr改走本地 vision 模型轉錄,"
"而非跳過(需 config/models.yaml 設定 tasks.vision")
ap.add_argument("--traceback", action="store_true",
help="轉換失敗時額外印出完整堆疊,供定位除錯")
ap.add_argument("--root", default=str(ROOT), help="專案根目錄(測試用)") ap.add_argument("--root", default=str(ROOT), help="專案根目錄(測試用)")
a = ap.parse_args(argv) a = ap.parse_args(argv)
root = pathlib.Path(a.root).resolve() root = pathlib.Path(a.root).resolve()
@@ -174,13 +239,22 @@ def main(argv=None):
if item.startswith(("http://", "https://")): if item.startswith(("http://", "https://")):
process_url(root, m, item, a.dry_run) process_url(root, m, item, a.dry_run)
else: else:
process_local(root, m, item, a.dry_run) process_local(root, m, item, a.dry_run, vision=a.vision)
except Exception as e: # 單檔失敗不中斷批次,最後彙總報錯 except Exception as e: # 單檔失敗不中斷批次,最後彙總報錯
failed.append((item, e)) failed.append((item, e))
print(f"error: {item}: {e}", file=sys.stderr) # 帶例外型別:只印訊息字串常無法判斷成因(如 KeyError 只印 key
print(f"error: {item}: {type(e).__name__}: {e}", file=sys.stderr)
if a.traceback:
traceback.print_exc()
if not a.dry_run: if not a.dry_run:
save_manifest(root, m) save_manifest(root, m)
if failed: if failed:
# 尾端彙總,避免失敗行淹沒在大批次輸出中而漏看
print(f"\n{len(failed)} 個項目轉換失敗:", file=sys.stderr)
for item, e in failed:
print(f" - {item}: {type(e).__name__}: {e}", file=sys.stderr)
if not a.traceback:
print("(加 --traceback 可印完整堆疊定位)", file=sys.stderr)
raise SystemExit(1) raise SystemExit(1)

View File

@@ -0,0 +1,91 @@
"""舊版 OLE 二進位格式(.doc/.xls/.ppt→ 現代格式(.docx/.xlsx/.pptx升版。
以 LibreOffice headless 在本機升版deterministic、內容不外流AGENTS.md §1.1
升版後由 convert.py 回頭走既有 from_docx / from_xlsx / from_pptx 轉換器。
升版產物是暫時中介檔,不進 raw/raw/originals 仍保存真正的舊格式原始檔§1.4)。
可獨立執行python tools/convert/from_legacy.py <file.doc> [--out DIR]
"""
import argparse
import os
import pathlib
import shutil
import subprocess
import sys
import tempfile
LEGACY_TARGET = {".doc": "docx", ".xls": "xlsx", ".ppt": "pptx"}
# 常見 Windows 安裝路徑;其他平台靠 PATHsoffice/libreoffice或 SOFFICE_BIN 覆寫。
_WIN_CANDIDATES = (
r"C:\Program Files\LibreOffice\program\soffice.exe",
r"C:\Program Files (x86)\LibreOffice\program\soffice.exe",
)
def find_soffice():
"""定位 LibreOffice 執行檔。SOFFICE_BIN 環境變數優先(唯一允許的覆寫來源)。"""
env = os.environ.get("SOFFICE_BIN")
if env and pathlib.Path(env).is_file():
return env
for c in _WIN_CANDIDATES:
if pathlib.Path(c).is_file():
return c
return shutil.which("soffice") or shutil.which("libreoffice")
def upgrade(src, out_dir, timeout=180):
"""升版 src 至 out_dir回傳升版後檔案路徑。
找不到 LibreOffice、格式不支援、或升版未產出檔案時拋 RuntimeError
convert.py 逐檔捕捉並回報,不中斷批次)。
"""
src = pathlib.Path(src)
ext = src.suffix.lower()
if ext not in LEGACY_TARGET:
raise RuntimeError(f"非舊版格式:{src.name}")
soffice = find_soffice()
if not soffice:
raise RuntimeError(
"找不到 LibreOfficesoffice.doc/.xls/.ppt 升版需要它。"
"請安裝 LibreOffice或以 SOFFICE_BIN 環境變數指定執行檔路徑。")
out_dir = pathlib.Path(out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
target = LEGACY_TARGET[ext]
# 每次呼叫用獨立 user profile避免 headless 並行/殘留 profile 造成的鎖定失敗
# LibreOffice headless 的已知痛點§13.3:真實環境不是規格書上的理想值)。
profile = out_dir / ".lo-profile"
try:
subprocess.run(
[soffice, f"-env:UserInstallation=file:///{profile.as_posix().lstrip('/')}",
"--headless", "--convert-to", target, "--outdir", str(out_dir), str(src)],
check=True, capture_output=True, text=True, timeout=timeout)
except subprocess.CalledProcessError as e:
raise RuntimeError(f"LibreOffice 升版失敗exit {e.returncode}{e.stderr.strip()}")
except subprocess.TimeoutExpired:
raise RuntimeError(f"LibreOffice 升版逾時(>{timeout}s{src.name}")
finally:
shutil.rmtree(profile, ignore_errors=True)
out = out_dir / f"{src.stem}.{target}"
if not out.is_file():
raise RuntimeError(f"LibreOffice 未產出預期檔案:{out.name}")
return out
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("src")
ap.add_argument("--out", help="升版產物輸出目錄(預設暫存目錄)")
a = ap.parse_args(argv)
out_dir = pathlib.Path(a.out) if a.out else pathlib.Path(tempfile.mkdtemp(prefix="ppqa-legacy-"))
try:
print(upgrade(a.src, out_dir))
except RuntimeError as e:
print(f"error: {e}", file=sys.stderr)
raise SystemExit(1)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,61 @@
"""影像 → Markdown 轉錄(本地 vision 模型AGENTS.md §1.1 資料落地)。
用於 vision fallback文字抽取失敗的頁面 render 成影像後,逐頁交本地 vision
模型忠實轉錄為 Markdown。模型設定讀 config/models.yaml 的 tasks.vision§1.2
未設定即報錯、不使用預設值;:cloud 模型一律拒絕(影像不得離開本機)。
"""
import base64
import pathlib
import sys
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) # tools/ 供 import kb
import kb
VISION_PROMPT = (
"你是文件轉錄引擎。忠實轉錄此頁影像中的所有文字與表格內容為 Markdown"
"表格用 Markdown table保留標題階層。只輸出轉錄內容本身不要加任何說明或臆測。"
"影像中沒有可辨識內容時,只回覆「(本頁無可辨識文字)」。")
def _require_vision(cfg):
"""取得並驗證 vision 模型設定。缺設定 / :cloud 皆報錯信任邊界§1.1/§1.2)。"""
spec = (cfg.get("tasks") or {}).get("vision")
if not isinstance(spec, dict) or not spec.get("model") or spec["model"] == "CHANGE_ME":
raise SystemExit(
"models.yaml 未設定 tasks.vision.model--vision 需要合規本地 vision 模型"
"(請自行確認非中國關聯、非 :cloudAGENTS.md §1.1/§1.3)。")
if str(spec["model"]).endswith(":cloud"):
raise SystemExit("tasks.vision.model 為 :cloud 模型禁止文件影像不得離開本機§1.1)。")
return spec
def _b64(path):
return base64.b64encode(pathlib.Path(path).read_bytes()).decode("ascii")
def transcribe(cfg, image_paths):
"""逐頁轉錄影像,回傳 (每頁 markdown 列表, 實際使用的 model)。
單頁重試 max_retries 次仍失敗即拋 RuntimeErrorvision 無文字 fallback 可用——
fallback 模型非 vision不切換
"""
spec = _require_vision(cfg)
model = spec["model"]
retries = int(spec.get("max_retries", 2))
temperature = float(spec.get("temperature", 0.1))
pages = []
for idx, img in enumerate(image_paths, 1):
b64 = _b64(img)
last = None
for _ in range(max(1, retries)):
try:
text = kb.ollama_chat(cfg, model, VISION_PROMPT, temperature, images=[b64])
if text and text.strip():
pages.append(text.strip())
break
last = "空回應"
except Exception as e:
last = f"{type(e).__name__}: {e}"
else:
raise RuntimeError(f"vision 轉錄失敗(第 {idx}model={model}{last}")
return pages, model

View File

@@ -3,19 +3,25 @@
執行:.venv/Scripts/python tools/convert/selfcheck.py 執行:.venv/Scripts/python tools/convert/selfcheck.py
邏輯壞掉時 assert 會失敗AGENTS.md §13.4:一個可執行的檢查,不用框架)。 邏輯壞掉時 assert 會失敗AGENTS.md §13.4:一個可執行的檢查,不用框架)。
""" """
import contextlib
import io
import json import json
import pathlib import pathlib
import shutil import shutil
import sys import sys
import tempfile import tempfile
sys.path.insert(0, str(pathlib.Path(__file__).parent)) sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) # tools/ 供 import kb
sys.path.insert(0, str(pathlib.Path(__file__).parent)) # tools/convert
import convert import convert
import from_docx import from_docx
import from_pdf import from_pdf
import from_pptx import from_pptx
import from_vision
import from_web import from_web
import from_xlsx import from_xlsx
import kb
import to_image
def make_fixtures(d): def make_fixtures(d):
@@ -118,6 +124,104 @@ def main():
assert m2["files"] == {} and not list((root2 / "raw/originals").iterdir()) assert m2["files"] == {} and not list((root2 / "raw/originals").iterdir())
print("PASS: --dry-run 不落地") print("PASS: --dry-run 不落地")
# 失敗回報:壞輸入不中斷批次,錯誤帶例外型別 + 尾端彙總,好檔仍轉換
root3 = tmp / "root3"
for sub in ("raw/originals", "raw/converted"):
(root3 / sub).mkdir(parents=True)
(root3 / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
(fx / "notes.txt").write_text("不支援的格式", encoding="utf-8") # → ValueError
err = io.StringIO()
code = None
with contextlib.redirect_stderr(err):
try:
convert.main([str(fx / "report.docx"), str(fx / "notes.txt"),
str(fx / "missing.docx"), # 不存在 → FileNotFoundError
"--root", str(root3)])
except SystemExit as e:
code = e.code
log = err.getvalue()
assert code == 1, code
assert "ValueError" in log and "FileNotFoundError" in log, log # 錯誤帶例外型別
assert "2 個項目轉換失敗" in log, log # 尾端彙總
m3 = json.loads((root3 / "raw/manifest.json").read_text(encoding="utf-8"))
assert (root3 / "raw/originals/report.docx").is_file() # 好檔仍轉換並落地
assert any(v["kind"] == "converted" for v in m3["files"].values()), m3
print("PASS: 失敗回報(例外型別 + 尾端彙總,批次不中斷)")
# B: 舊格式路由 —— .doc 應走 legacyLibreOffice而非被當「不支援格式」拒收
root4 = tmp / "root4"
for sub in ("raw/originals", "raw/converted"):
(root4 / sub).mkdir(parents=True)
(root4 / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
(fx / "old.doc").write_bytes(b"\xd0\xcf\x11\xe0legacy-stub") # 非有效 .doc僅測路由/錯誤
err = io.StringIO()
code = None
with contextlib.redirect_stderr(err):
try:
convert.main([str(fx / "old.doc"), "--root", str(root4)])
except SystemExit as e:
code = e.code
log = err.getvalue()
assert code == 1 and "old.doc" in log, log
assert "不支援的格式" not in log and "RuntimeError" in log, log # 已路由 legacy乾淨報錯
print("PASS: 舊格式路由至 legacysoffice 缺失/升版失敗皆乾淨報錯)")
# A: to_image renderdeterministic真跑
imgs = to_image.render_pdf_pages(fx / "text.pdf", tmp / "imgs")
assert imgs and all(p.is_file() and p.stat().st_size > 0 for p in imgs), imgs
print(f"PASS: to_image render{len(imgs)} 頁 PNG")
# A: from_vision 轉錄邏輯(注入假 Ollama不需真模型+ 合規防護
vis_cfg = {"tasks": {"vision": {"model": "fake-vision", "temperature": 0.1,
"max_retries": 1}}}
orig_chat = kb.ollama_chat
calls = {}
def fake_chat(cfg, model, prompt, temperature, json_format=False, images=None):
calls["images"] = images # 記錄 base64 是否傳入
return "可疑交易監控:大額 12,000 元"
kb.ollama_chat = fake_chat
try:
texts, model = from_vision.transcribe(vis_cfg, imgs[:1])
finally:
kb.ollama_chat = orig_chat
assert model == "fake-vision" and texts and "可疑交易監控" in texts[0], texts
assert calls.get("images"), "vision 呼叫必須帶影像"
for bad in ({"tasks": {}}, {"tasks": {"vision": {"model": "foo:cloud"}}}):
try:
from_vision._require_vision(bad)
assert False, "應拒絕未設定 / :cloud"
except SystemExit:
pass
print("PASS: from_vision 轉錄 + :cloud/未設定防護")
# A: convert.py --vision 端到端(沙盒 config + 注入假 vision
root_v = tmp / "root_v"
(root_v / "config").mkdir(parents=True)
for sub in ("raw/originals", "raw/converted"):
(root_v / sub).mkdir(parents=True)
(root_v / "config/models.yaml").write_text(
'ollama:\n base_url: "http://localhost:11434"\n timeout_seconds: 300\n'
'tasks:\n ingest: {model: "x", temperature: 0.2, max_retries: 1}\n'
' lint: {model: "x", temperature: 0.1, max_retries: 1}\n'
' fallback: {model: "x"}\n'
' vision: {model: "fake-vision", temperature: 0.1, max_retries: 1}\n'
'limits:\n mcp_response_max_tokens: 2000\n ingest_chunk_max_chars: 12000\n',
encoding="utf-8")
(root_v / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
kb.ollama_chat = fake_chat
try:
convert.main([str(fx / "scanned.pdf"), "--vision", "--root", str(root_v)])
finally:
kb.ollama_chat = orig_chat
mv = json.loads((root_v / "raw/manifest.json").read_text(encoding="utf-8"))
conv_rel = [k for k, v in mv["files"].items() if v["kind"] == "converted"]
assert len(conv_rel) == 1 and mv["files"][conv_rel[0]]["converter"] == "from_vision", mv
md_v = (root_v / conv_rel[0]).read_text(encoding="utf-8")
assert "可疑交易監控" in md_v and "(vision" in md_v and "![page 1]" in md_v, md_v
print("PASS: convert.py --vision 端到端needs_ocr → 本地 vision 轉錄)")
print("ALL PASS") print("ALL PASS")
finally: finally:
shutil.rmtree(tmp, ignore_errors=True) shutil.rmtree(tmp, ignore_errors=True)

48
tools/convert/to_image.py Normal file
View File

@@ -0,0 +1,48 @@
"""PDF 頁 → PNG 影像pymupdfdeterministic、全程本地
供 vision fallback 使用文字抽取失敗needs_ocr的掃描/圖片型 PDF 逐頁 render
成影像後,交給 from_vision.py 以本地 vision 模型判讀。render 本身不含任何 LLM
維持「deterministic 工具做結構、LLM 只做語意」的分工AGENTS.md §3 稽核原則)。
可獨立執行python tools/convert/to_image.py <file.pdf> --out DIR [--dpi 150]
"""
import argparse
import pathlib
import fitz # pymupdf
# ponytail: 預設 150 DPI 是「可讀性 vs 影像大小/推論成本」的折衷;掃描件文字偏小時
# 可調高。升級路徑:依頁面尺寸/內容自適應 DPI。
DEFAULT_DPI = 150
def render_pdf_pages(src, out_dir, dpi=DEFAULT_DPI, prefix="page"):
"""將 src 每頁 render 成 PNG回傳影像路徑列表依頁序"""
src = pathlib.Path(src)
out_dir = pathlib.Path(out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
paths = []
doc = fitz.open(str(src))
try:
for i, page in enumerate(doc, 1):
pix = page.get_pixmap(dpi=dpi)
p = out_dir / f"{prefix}{i}.png"
pix.save(str(p))
paths.append(p)
finally:
doc.close()
return paths
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("src")
ap.add_argument("--out", required=True, help="影像輸出目錄")
ap.add_argument("--dpi", type=int, default=DEFAULT_DPI)
a = ap.parse_args(argv)
paths = render_pdf_pages(a.src, a.out, a.dpi)
print(f"{len(paths)} 頁 → {a.out}")
if __name__ == "__main__":
main()

View File

@@ -48,11 +48,17 @@ def load_config(root=ROOT):
# ---------- Ollama ---------- # ---------- Ollama ----------
def ollama_chat(cfg, model, prompt, temperature, json_format=False): def ollama_chat(cfg, model, prompt, temperature, json_format=False, images=None):
"""單次呼叫本地 Ollama /api/chatstdlib urllib不需額外依賴""" """單次呼叫本地 Ollama /api/chatstdlib urllib不需額外依賴
imagesbase64 影像字串列表(供 vision 模型判讀);附在 user 訊息上。
"""
msg = {"role": "user", "content": prompt}
if images:
msg["images"] = list(images)
body = { body = {
"model": model, "model": model,
"messages": [{"role": "user", "content": prompt}], "messages": [msg],
"stream": False, "stream": False,
"options": {"temperature": temperature}, "options": {"temperature": temperature},
} }