Phase 5: MCP 薄殼(search_wiki/read_page)+ BM25 檢索 + 三個 Copilot Custom Agents
Some checks failed
kb-lint / lint (push) Has been cancelled

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
LittleYellow
2026-07-14 21:05:01 +08:00
parent 156c6db2f8
commit cb2c92452e
8 changed files with 302 additions and 0 deletions

69
tools/search.py Normal file
View File

@@ -0,0 +1,69 @@
"""wiki BM25 全文檢索第一版不用向量庫AGENTS.md §0
用法python tools/search.py "查詢詞" [-k 10] [--root DIR]
輸出JSON 陣列path、title、description、tags、score供 agent 與 MCP 使用。
"""
import argparse
import json
import pathlib
import re
import sys
sys.path.insert(0, str(pathlib.Path(__file__).parent))
import kb
from rank_bm25 import BM25Okapi
def tokenize(text):
"""英數字詞 + CJK bigram。
ponytail: bigram 斷詞無語意理解(同義詞不匹配),天花板是字面重疊;
升級路徑Phase 6 後評估 EmbeddingGemma / snowflake-arctic-embed 向量檢索。
"""
text = text.lower()
tokens = re.findall(r"[a-z0-9]+", text)
for run in re.findall(r"[一-鿿]+", text):
tokens += [run] if len(run) == 1 else [run[i:i + 2] for i in range(len(run) - 1)]
return tokens
def build_corpus(root):
docs = []
for p in kb.iter_pages(root):
try:
meta, body = kb.parse_page(p.read_text(encoding="utf-8"))
except ValueError:
continue # 壞頁由 lint 報,檢索直接略過
text = (f"{meta['title']} " * 3 + f"{meta['description']} " * 2
+ " ".join(meta.get("tags") or []) + " " + body)
docs.append({"path": p.relative_to(root).as_posix(), "title": str(meta["title"]),
"description": str(meta["description"]),
"tags": [str(t) for t in meta.get("tags") or []],
"tokens": tokenize(text)})
return docs
def search(root, query, k=10):
docs = build_corpus(pathlib.Path(root))
if not docs:
return []
bm = BM25Okapi([d["tokens"] for d in docs])
scores = bm.get_scores(tokenize(query))
ranked = sorted(zip(docs, scores), key=lambda x: -x[1])[:k]
return [{"path": d["path"], "title": d["title"], "description": d["description"],
"tags": d["tags"], "score": round(float(s), 4)}
for d, s in ranked if s > 0]
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("query")
ap.add_argument("-k", type=int, default=10, help="回傳筆數上限(查詢工作流上限 10")
ap.add_argument("--root", default=str(kb.ROOT))
a = ap.parse_args(argv)
hits = search(a.root, a.query, max(1, min(a.k, 10)))
print(json.dumps(hits, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()

88
tools/selfcheck_search.py Normal file
View File

@@ -0,0 +1,88 @@
"""Phase 5 自檢BM25 檢索排序、MCP search_wiki / read_page、
路徑跳脫防護信任邊界、token 上限截斷。
執行:.venv/Scripts/python tools/selfcheck_search.py
"""
import os
import pathlib
import shutil
import sys
import tempfile
sys.path.insert(0, str(pathlib.Path(__file__).parent))
import kb
import search
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: 100, ingest_chunk_max_chars: 12000}
"""
def page(root, rel, title, desc, tags, body):
meta = {"type": {"summaries": "summary", "entities": "entity",
"concepts": "concept"}[rel.split("/")[1]],
"title": title, "description": desc, "tags": tags,
"timestamp": "2026-07-14", "sources": ["raw/converted/x.md#abcd1234"],
"status": "draft"}
(root / rel).write_text(kb.dump_page(meta, body), encoding="utf-8")
def main():
tmp = pathlib.Path(tempfile.mkdtemp(prefix="ppqa-search-"))
try:
root = tmp / "repo"
for sub in ("config", "wiki/summaries", "wiki/entities", "wiki/concepts"):
(root / sub).mkdir(parents=True)
(root / "config/models.yaml").write_text(CFG, encoding="utf-8")
page(root, "wiki/entities/payment-gateway.md", "支付閘道", "支付閘道模組",
["gateway"], "壓力測試時出現逾時缺陷,重試機制未生效。")
page(root, "wiki/concepts/aml-testing.md", "AML 測試準則", "可疑交易監控測試綜合",
["aml", "compliance"], "大額交易與分散交易樣態的監控測試要求。")
page(root, "wiki/summaries/long-doc.md", "長文件摘要", "截斷測試用",
["test"], "逾時。" * 500)
hits = search.search(root, "支付閘道 逾時")
assert hits and hits[0]["path"] == "wiki/entities/payment-gateway.md", hits
assert all(h["score"] > 0 for h in hits)
hits2 = search.search(root, "可疑交易監控")
assert hits2[0]["path"] == "wiki/concepts/aml-testing.md", hits2
assert search.search(root, "zzz-nonexistent-term") == []
print("PASS: BM25 檢索排序(繁中 bigram")
os.environ["PP_QA_ROOT"] = str(root)
import importlib
sys.path.insert(0, str(pathlib.Path(__file__).parents[1] / "mcp"))
server = importlib.import_module("server")
res = server.search_wiki("支付閘道 逾時", top_k=99) # top_k 超限 → 收斂為 10
assert res[0]["path"] == "wiki/entities/payment-gateway.md"
pg = server.read_page("wiki/entities/payment-gateway.md")
assert pg["frontmatter"]["title"] == "支付閘道" and not pg["truncated"]
assert pg["source_refs"] == ["raw/converted/x.md#abcd1234"]
long = server.read_page("wiki/summaries/long-doc.md")
assert long["truncated"] and len(long["body"]) <= 150 and long["source_refs"]
print("PASS: MCP search_wiki + read_page含 token 截斷、source_refs 保留)")
for bad in ("../AGENTS.md", "wiki/../config/models.yaml", "wiki/entities/x.txt"):
try:
server.read_page(bad)
raise AssertionError(f"應拒絕:{bad}")
except (ValueError, FileNotFoundError) as e:
assert isinstance(e, ValueError), f"{bad} 應為 ValueError"
try:
server.read_page("wiki/entities/no-such.md")
raise AssertionError("應拋 FileNotFoundError")
except FileNotFoundError:
pass
print("PASS: 路徑跳脫防護(信任邊界)")
print("ALL PASS")
finally:
os.environ.pop("PP_QA_ROOT", None)
shutil.rmtree(tmp, ignore_errors=True)
if __name__ == "__main__":
main()