cb2c92452e
Some checks failed
kb-lint / lint (push) Has been cancelled
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
"""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()
|