#!/usr/bin/env python3 """對話主動召回：根據當前對話 query，從 X 書籤索引找出最相關的知識卡。""" from __future__ import annotations import argparse import json import math import re import os import sys from pathlib import Path from typing import List, Dict, Any sys.path.insert(0, str(Path(__file__).resolve().parent)) import xkb_paths import xkb_text from xkb_mmr import mmr_select from xkb_provenance import is_self_derived WORKSPACE_DIR = xkb_paths.WORKSPACE BOOKMARKS_DIR = xkb_paths.BOOKMARKS_DIR INDEX_FILE = xkb_paths.INDEX_FILE VECTOR_FILE = xkb_paths.VECTOR_FILE TOPIC_PROFILE_FILE = xkb_paths.TOPIC_PROFILE_FILE _SKILL_DIR = xkb_paths.SKILL_DIR WIKI_TOPICS_DIR = xkb_paths.WIKI_TOPICS_DIR GENERIC_CATEGORIES = {"general", "99-general", "other", "misc", "uncategorized"} LOW_SIGNAL_SUMMARIES = {"（待整理）", "待整理", "todo", "tbd", "n/a"} LOW_SIGNAL_TITLES = {"(untitled)", "untitled", "tweet"} LOW_SIGNAL_SOURCES = {"x", "twitter"} STOPWORDS = { "的", "了", "是", "我", "你", "他", "她", "它", "在", "有", "和", "與", "就", "也", "都", "很", "想", "要", "用", "讓", "把", "跟", "對", "中", "上", "下", "嗎", "呢", "啊", "吧", "這", "那", "this", "that", "with", "from", "have", "will", "about", "into", "your", "their", "they", "them", "what", "when", "where", "which", "how", "why", "for", "and", "the", "are", "was", "were", "been", } def load_index(index_file: Path) -> Dict[str, Any]: if not index_file.exists(): raise FileNotFoundError(f"search index not found: {index_file}") return json.loads(index_file.read_text(encoding="utf-8")) def tokenize(text: str) -> List[str]: return xkb_text.tokenize(text, STOPWORDS) def clean_summary(text: str) -> str: text = (text or "").strip() text = re.sub(r"^#+\s*", "", text) text = re.sub(r"^一句話摘要\s*", "", text) text = re.sub(r"^[-•]\s*", "", text) text = re.sub(r"\s+", " ", text) return text.strip() def load_topic_profile(topic_profile_file: Path = TOPIC_PROFILE_FILE) -> Dict[str, Any]: if not topic_profile_file.exists(): return {} try: return json.loads(topic_profile_file.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): return {} def get_topic_profile_matches(query_tokens: List[str], topic_profile: Dict[str, Any]) -> Dict[str, Any]: categories = {item.get("name", ""): item.get("weight", 0.0) for item in topic_profile.get("top_categories", [])} tags = {item.get("name", ""): item.get("weight", 0.0) for item in topic_profile.get("top_tags", [])} matched_categories = [name for name in categories if name and any(token in name or name in token for token in query_tokens)] matched_tags = [name for name in tags if name and any(token == name or token in name or name in token for token in query_tokens)] cat_boost = max((categories[name] for name in matched_categories), default=0.0) tag_boost = max((tags[name] for name in matched_tags), default=0.0) combined = max(cat_boost, tag_boost) return { "matched_categories": matched_categories[:3], "matched_tags": matched_tags[:5], "topic_boost": round(combined, 4), } def _is_valid_source_url(url: str) -> bool: url = (url or "").strip() if not url.startswith(("http://", "https://")): return False x_status = re.search(r"https?://(?:x|twitter)\.com/[^\s/]+/status/(\d{15,20})(?:\b|/|\?)", url) x_i_status = re.search(r"https?://x\.com/i/status/(\d{15,20})(?:\b|/|\?)", url) if ("x.com" in url or "twitter.com" in url) and not (x_status or x_i_status): return False return True def _normalize_source_url(url: str) -> str: url = (url or "").strip().strip('"') if not _is_valid_source_url(url): return "" return url def extract_source_url(md_path: Path) -> str: if not md_path.exists(): return "" text = md_path.read_text(encoding="utf-8", errors="ignore") patterns = [ r"^source_url:\s*\"?([^\"\n]+)\"?\s*$", r"^original_url:\s*\"?([^\"\n]+)\"?\s*$", r"\*\*原始連結\*\*：\s*(\S+)", r"https://x\.com/\S+", r"https://twitter\.com/\S+", ] for pattern in patterns: m = re.search(pattern, text, re.MULTILINE) if m: url = m.group(1).strip() if m.groups() else m.group(0).strip() normalized = _normalize_source_url(url) if normalized: return normalized return "" def build_relevance_reason(item: Dict[str, Any], query_tokens: List[str], topic_matches: Dict[str, Any] | None = None) -> str: title = (item.get("title") or "").lower() category = (item.get("category") or "").lower() tags = " ".join(item.get("tags") or []).lower() summary = (item.get("summary") or "").lower() reasons = [] title_hits = [t for t in query_tokens if t in title] tag_hits = [t for t in query_tokens if t in tags] category_hits = [t for t in query_tokens if t in category] summary_hits = [t for t in query_tokens if t in summary] if title_hits: reasons.append(f"標題命中：{'、'.join(title_hits[:3])}") if tag_hits: reasons.append(f"標籤接近：{'、'.join(tag_hits[:3])}") if category_hits: reasons.append(f"分類相關：{'、'.join(category_hits[:3])}") if summary_hits and not reasons: reasons.append(f"摘要語意接近：{'、'.join(summary_hits[:3])}") if topic_matches: if topic_matches.get("matched_categories"): reasons.append(f"命中使用者高頻分類：{'、'.join(topic_matches['matched_categories'][:2])}") elif topic_matches.get("matched_tags"): reasons.append(f"命中使用者高頻標籤：{'、'.join(topic_matches['matched_tags'][:3])}") return "；".join(reasons[:2]) or "主題與當前對話高度相關" def score_item(item: Dict[str, Any], query_tokens: List[str], query_text: str) -> int: title = (item.get("title") or "").lower() category = (item.get("category") or "").lower() tags = " ".join(item.get("tags") or []).lower() summary = (item.get("summary") or "").lower() blob = (item.get("searchable") or "").lower() score = 0 for token in query_tokens: if token in title: score += 8 if token in tags: score += 6 if token in category: score += 4 if token in summary: score += 3 if token in blob: score += 1 if query_text and query_text in blob: score += 8 # 偏好有摘要、有 tags 的卡片，較適合直接對話回用 if item.get("summary"): score += 2 if item.get("tags"): score += 1 return score def _keyword_score(query: str, item: dict) -> float: """Fraction of query tokens found in title + tags + summary.""" tokens = tokenize(query) if not tokens: return 0.0 text = " ".join([ (item.get("title") or "").lower(), " ".join(item.get("tags") or []).lower(), (item.get("summary") or "").lower(), ]) return xkb_text.overlap_score(tokens, text) def _cosine_similarity(a: list, b: list) -> float: dot = sum(x * y for x, y in zip(a, b)) norm_a = math.sqrt(sum(x * x for x in a)) norm_b = math.sqrt(sum(y * y for y in b)) denom = norm_a * norm_b return dot / denom if denom > 1e-9 else 0.0 def _normalize_vector(vec: Any) -> list[float] | None: if not isinstance(vec, list) or not vec: return None normalized = [] for value in vec: if not isinstance(value, (int, float)) or isinstance(value, bool): return None normalized.append(float(value)) return normalized def _source_quality(item: Dict[str, Any]) -> float: source = (item.get("source") or "").strip().lower() source_url = (item.get("source_url") or "").strip().lower() title = (item.get("title") or "").strip().lower() score = 0.0 if source == "youtube": score += 0.05 if source_url.startswith("https://"): score += 0.05 if "github.com" in source_url: score += 0.05 if any(host in source_url for host in ["x.com/", "twitter.com/"]): score += 0.02 if re.match(r"^\d{4}-\d{2}-\d{2}-", title): score -= 0.08 if source in LOW_SIGNAL_SOURCES and not source_url: score -= 0.03 return score def _summary_quality(summary: str) -> float: summary = clean_summary(summary) if not summary: return -0.08 if summary in LOW_SIGNAL_SUMMARIES: return -0.12 if len(summary) < 12: return -0.05 if len(summary) >= 40: return 0.05 return 0.02 def _title_quality(title: str) -> float: title = (title or "").strip() lowered = title.lower() if not title: return -0.12 if lowered in LOW_SIGNAL_TITLES: return -0.15 if re.fullmatch(r"tweet\s+\d{15,20}", lowered): return -0.14 if re.fullmatch(r"\d{15,20}", title): return -0.16 if re.match(r"^\d{4}-\d{2}-\d{2}-", title) and len(title) < 42: return -0.08 if len(title) < 8: return -0.05 return 0.03 def _category_penalty(item: Dict[str, Any]) -> float: category = (item.get("category") or "").strip().lower() return -0.08 if category in GENERIC_CATEGORIES else 0.0 # 回音室防護（TODOS 2026-07-13）：對話蒸餾而來的知識（self-derived）召回時降權， # 避免「收藏 → 召回 → 討論 → 再入庫」閉環讓既有觀點自我強化。 SELF_DERIVED_PENALTY = float(os.getenv("XKB_SELF_DERIVED_PENALTY", "-0.15")) # 索引項目實際有的欄位：category, enriched, mtime, path, relative_path, # searchable, size, source_type, source_url, summary, tags, title。 # 原本這裡讀的是 provenance / content / source_file / section_text——四個 # 都不存在，所以這道防護從第一天起就沒有觸發過一次。 PROVENANCE_FIELDS = ("summary", "searchable", "relative_path", "path", "title") def _provenance_only(item: Dict[str, Any]) -> Dict[str, float]: """只帶出處降權的調整包。gbrain 路徑沒有主題加權可算。""" penalty = _provenance_penalty(item) return {"provenance": penalty, "total_adjustment": penalty} def _provenance_penalty(item: Dict[str, Any]) -> float: if (item.get("provenance") or "").strip().lower() == "self-derived": return SELF_DERIVED_PENALTY # 路徑要一起看：self-derived 的標記常常在 memory/YYYY-MM-DD 這種路徑上， # 而不是在摘要文字裡。 text = " ".join(str(item.get(k) or "") for k in PROVENANCE_FIELDS) if is_self_derived(text): return SELF_DERIVED_PENALTY return 0.0 def _should_filter_result(item: Dict[str, Any], source_url: str) -> bool: title = (item.get("title") or "").strip() summary = clean_summary(item.get("summary") or "") category = (item.get("category") or "").strip().lower() if item.get("excluded"): return True if summary in LOW_SIGNAL_SUMMARIES and not source_url: return True if re.fullmatch(r"\d{15,20}", title): return True if re.fullmatch(r"tweet\s+\d{15,20}", title.lower()) and summary in LOW_SIGNAL_SUMMARIES: return True if category in GENERIC_CATEGORIES and summary in LOW_SIGNAL_SUMMARIES: return True return False def _ranking_adjustments(item: Dict[str, Any], topic_matches: Dict[str, Any]) -> Dict[str, float]: topic_boost = topic_matches.get("topic_boost", 0.0) matched_categories = set(topic_matches.get("matched_categories", [])) matched_tags = set(topic_matches.get("matched_tags", [])) item_category = (item.get("category") or "").lower() item_tags = {str(tag).lower() for tag in (item.get("tags") or [])} topic_bonus = 0.0 if topic_boost > 0: if item_category in matched_categories: topic_bonus += round(topic_boost * 0.12, 4) elif item_tags & matched_tags: topic_bonus += round(topic_boost * 0.08, 4) summary_bonus = _summary_quality(item.get("summary") or "") title_bonus = _title_quality(item.get("title") or "") source_bonus = _source_quality(item) category_penalty = _category_penalty(item) provenance_penalty = _provenance_penalty(item) return { "topic_bonus": topic_bonus, "summary_bonus": summary_bonus, "title_bonus": title_bonus, "source_bonus": source_bonus, "category_penalty": category_penalty, "provenance_penalty": provenance_penalty, "total_adjustment": round(topic_bonus + summary_bonus + title_bonus + source_bonus + category_penalty + provenance_penalty, 4), } def _display_title(item: dict) -> str: """Return a human-readable title. If title is just a tweet ID, use summary snippet.""" import re as _re title = (item.get("title") or "").strip() if title and not _re.match(r"^\d{10,}$", title): return title summary = (item.get("summary") or "").strip() if summary and summary not in ("待整理", "待補充") and not summary.startswith("###"): first = summary.split("。")[0].split(".")[0][:60].strip() if first: return first + "…" return title or "(untitled)" def gbrain_semantic_recall(query: str, limit: int = 5) -> List[Dict[str, Any]]: """ Recall via gbrain hybrid search (pgvector RRF + Gemini). Returns same format as semantic_recall() for drop-in replacement. Falls back silently if gbrain is not available. """ try: sys.path.insert(0, str(Path(__file__).resolve().parent)) from xbrain_recall import xbrain_query as gbrain_query except ImportError: return [] try: raw = gbrain_query(query, limit=limit) except RuntimeError: return [] results = [] for item in raw: chunk = item.get("chunk_text", "") # Extract title from first markdown heading title_m = re.search(r"^#\s+(.+)$", chunk, re.MULTILINE) title = title_m.group(1).strip() if title_m else item.get("title") or item.get("slug", "") # Extract summary from 雙語摘要 ZH line zh_m = re.search(r"ZH[:\s]+(.+?)(?:\n|EN:|$)", chunk, re.IGNORECASE) summary = zh_m.group(1).strip() if zh_m else chunk[:120].replace("\n", " ").strip() source_url = item.get("source_url") or "" score = item.get("score", 0.0) slug = item.get("slug", "") # 出處懲罰算一次就好。原本同一個 dict literal 建了兩次， # 一次進分數、一次進說明——同一個量兩個來源，正是這個專案的老毛病。 _adjustments = _provenance_only({ "summary": summary, "relative_path": f"cards/{slug}.md", "title": title, }) results.append({ "title": title, "summary": summary, "category": item.get("type", "knowledge-card"), "tags": [], "relative_path": f"cards/{slug}.md", "source_url": source_url, # 懲罰要進分數。原本只放進 ranking_adjustments 給人看，score # 還是原始 RRF——於是回音室防護在這台機器實際會走的路徑上， # 算得出來、顯示得出來、就是不會影響排序。 "score": round(score + _adjustments["total_adjustment"], 4), "topic_boost": 0.0, "matched_categories": [], "matched_tags": [], # gbrain 是這台機器上實際會走的模式，而它原本完全不算調整—— # 於是回音室防護在主要路徑上等於不存在。 # 要看組好的欄位（摘要、路徑），不是 gbrain 回傳的原始 item： # self-derived 的線索在 relative_path 與 summary 上。 "ranking_adjustments": _adjustments, "relevance_reason": f"gbrain 語意+關鍵字混合 RRF ({score:.4f})", }) # 另外兩條 semantic_recall 都是 sort 之後才 [:limit]，只有這條直接回 # gbrain 的原始順序——而這條正是這台機器實際會走的。同一個函式契約、 # 三個實作、兩個排序，於是任何加減分在這裡都不會改變輸出。 results.sort(key=lambda r: r["score"], reverse=True) return results[:limit] def wiki_recall(query: str, limit: int = 2) -> List[Dict[str, Any]]: """第一層召回：從 wiki/topics/*.md 找合成知識段落。""" if not WIKI_TOPICS_DIR.exists(): return [] query_tokens = tokenize(query) if not query_tokens: return [] results = [] for path in WIKI_TOPICS_DIR.glob("*.md"): try: content = path.read_text(encoding="utf-8", errors="ignore") except OSError: continue # 從 frontmatter 取 title fm_match = re.match(r"^---\n(.*?)\n---\n", content, re.DOTALL) topic_title = path.stem if fm_match: title_m = re.search(r"^title:\s*(.+)$", fm_match.group(1), re.MULTILINE) if title_m: topic_title = title_m.group(1).strip().strip('"') # 移除 frontmatter，切成 ## 段落 body = re.sub(r"^---\n.*?\n---\n", "", content, flags=re.DOTALL).strip() sections = re.split(r"\n(?=##+ )", body) for section in sections: section = section.strip() if not section: continue first_line = section.split("\n")[0] section_title = re.sub(r"^#+\s*", "", first_line).strip() section_body = "\n".join(section.split("\n")[1:]).strip() if not section_body: continue text_lower = (section_title + " " + section_body).lower() score = 0 for token in query_tokens: if token in section_title.lower(): score += 6 elif token in text_lower: score += 2 if score < 4: continue # 擷取有意義的摘錄（跳過純分隔線） excerpt_lines = [l for l in section_body.split("\n") if l.strip() and not re.fullmatch(r"-{3,}", l.strip())] excerpt = " ".join(excerpt_lines)[:240].strip() if len(excerpt) >= 240: excerpt = excerpt.rsplit(" ", 1)[0] + "…" results.append({ "topic_title": topic_title, "section_title": section_title, "excerpt": excerpt, "path": f"wiki/topics/{path.name}", "score": score, }) results.sort(key=lambda x: x["score"], reverse=True) # 原本是「每個 topic 只留最高分的一段」。那條規則兩邊都鈍：同一頁的第二段 # 就算講的是完全不同的事也會被丟掉，而兩段幾乎一樣的內容只要分屬不同頁就 # 都會留下。治理開始讓每則升級的論點各自成段之後，一頁可以有好幾百段， # 這條規則等於讓一頁的內容在每次召回時互相淘汰，只出得來一段。 # # 改成 MMR：一段要「比已選的多說了些什麼」才留。分數接近時它會挑不重複的， # 分數差很多時不會為了多樣性硬塞——後者靠函式內的正規化保證。 return mmr_select( results, limit, text=lambda item: f"{item['section_title']} {item['excerpt']}", ) def semantic_recall(query: str, limit: int, vector_file: Path = VECTOR_FILE, index_file: Path = INDEX_FILE, topic_profile_file: Path = TOPIC_PROFILE_FILE) -> List[Dict[str, Any]]: """Semantic recall using vector similarity. Falls back to keyword if index missing.""" if not vector_file.exists(): print(f"⚠️ Vector index not found: {vector_file}", file=sys.stderr) print(" Falling back to keyword search. Run: python3 scripts/build_vector_index.py", file=sys.stderr) return [] # Load vector index import json as _json vdata = _json.loads(vector_file.read_text(encoding="utf-8")) vectors = vdata.get("vectors", {}) if not vectors: return [] normalized_vectors: dict[str, list[float]] = {} expected_dim = None for rel_path, raw_vec in vectors.items(): norm_vec = _normalize_vector(raw_vec) if norm_vec is None: print(f"⚠️ Invalid vector payload for: {rel_path}", file=sys.stderr) continue if expected_dim is None: expected_dim = len(norm_vec) elif len(norm_vec) != expected_dim: print( f"⚠️ Vector dimension mismatch for: {rel_path} " f"(expected {expected_dim}, got {len(norm_vec)})", file=sys.stderr, ) continue normalized_vectors[rel_path] = norm_vec if not normalized_vectors: print("⚠️ No valid vectors found in vector index. Falling back to keyword search.", file=sys.stderr) return [] # Embed the query try: import sys as _sys, os as _os _skill_dir = Path(__file__).resolve().parent.parent if str(_skill_dir) not in _sys.path: _sys.path.insert(0, str(_skill_dir)) from tools.embedding_providers import get_provider provider = get_provider() query_vec = provider.embed(query) query_vec = _normalize_vector(query_vec) if query_vec is None: print("⚠️ Query embedding payload is invalid. Falling back to keyword search.", file=sys.stderr) return [] if expected_dim is not None and len(query_vec) != expected_dim: print( f"⚠️ Query embedding dimension mismatch (expected {expected_dim}, got {len(query_vec)}). " "Falling back to keyword search.", file=sys.stderr, ) return [] except EnvironmentError as e: print(f"⚠️ {e}", file=sys.stderr) print(" Falling back to keyword search.", file=sys.stderr) return [] except Exception as e: print(f"⚠️ Embedding failed: {e}", file=sys.stderr) return [] # Compute cosine similarity, then aggregate to card level # （論點級向量鍵為 relpath#kpN，同卡取最高分 — TODOS 2026-07-13） best: dict[str, float] = {} for key, vec in normalized_vectors.items(): sim = _cosine_similarity(query_vec, vec) base = key.split("#", 1)[0] if sim > best.get(base, -1.0): best[base] = sim scored = sorted(best.items(), key=lambda x: x[1], reverse=True) top = scored[:limit * 2] # fetch extra to allow filtering # Load search index for metadata data = load_index(index_file) items = data.get("items", []) index_by_path = { (item.get("relative_path") or item.get("path") or ""): item for item in items } query_tokens = tokenize(query) topic_profile = load_topic_profile(topic_profile_file) topic_matches = get_topic_profile_matches(query_tokens, topic_profile) if topic_profile else {} results = [] for rel_path, sim in top: if sim < 0.25: break item = index_by_path.get(rel_path) if not item: continue source_url = _normalize_source_url(item.get("source_url") or "") if not source_url: md_path = BOOKMARKS_DIR / rel_path if not rel_path.startswith("/") else Path(rel_path) source_url = extract_source_url(md_path) kw = _keyword_score(query, item) hybrid = 0.65 * sim + 0.35 * kw adjustments = _ranking_adjustments(item, topic_matches) final_score = round(hybrid + adjustments["total_adjustment"], 4) if _should_filter_result(item, source_url): continue reason = [f"語意 {sim:.0%}", f"關鍵字 {kw:.0%}"] if adjustments["topic_bonus"] > 0: reason.append(f"主題加權 +{adjustments['topic_bonus']:.2f}") if adjustments["summary_bonus"] != 0: reason.append(f"摘要調整 {adjustments['summary_bonus']:+.2f}") if adjustments["title_bonus"] != 0: reason.append(f"標題調整 {adjustments['title_bonus']:+.2f}") if adjustments["source_bonus"] != 0: reason.append(f"來源調整 {adjustments['source_bonus']:+.2f}") if adjustments["category_penalty"] != 0: reason.append(f"泛分類調整 {adjustments['category_penalty']:+.2f}") results.append({ "title": _display_title(item), "summary": clean_summary(item.get("summary") or ""), "category": item.get("category") or "general", "tags": item.get("tags") or [], "relative_path": rel_path, "source_url": source_url, "score": final_score, "topic_boost": topic_matches.get("topic_boost", 0.0), "matched_categories": topic_matches.get("matched_categories", []), "matched_tags": topic_matches.get("matched_tags", []), "ranking_adjustments": adjustments, "relevance_reason": " + ".join(reason), }) # Re-sort by hybrid score and trim to limit results.sort(key=lambda x: x["score"], reverse=True) return results[:limit] def recall(query: str, limit: int, min_score: int, index_file: Path = INDEX_FILE, topic_profile_file: Path = TOPIC_PROFILE_FILE) -> List[Dict[str, Any]]: data = load_index(index_file) items = data.get("items", []) query = query.strip() query_tokens = tokenize(query) if not query_tokens and not query: return [] topic_profile = load_topic_profile(topic_profile_file) topic_matches = get_topic_profile_matches(query_tokens, topic_profile) if topic_profile else {} topic_boost = topic_matches.get("topic_boost", 0.0) results = [] for item in items: base_score = float(score_item(item, query_tokens, query.lower())) adjustments = _ranking_adjustments(item, topic_matches) score = base_score + adjustments["total_adjustment"] * 10 if score < min_score: continue rel_path = item.get("relative_path") or item.get("path") or "" # Use pre-indexed source_url first; fall back to file scan only if missing source_url = _normalize_source_url(item.get("source_url") or "") if not source_url: md_path = BOOKMARKS_DIR / rel_path if rel_path and not rel_path.startswith("/") else Path(rel_path) source_url = extract_source_url(md_path) if _should_filter_result(item, source_url): continue results.append({ "title": _display_title(item), "summary": clean_summary(item.get("summary") or ""), "category": item.get("category") or "general", "tags": item.get("tags") or [], "relative_path": rel_path, "source_url": source_url, "score": round(score, 4), "topic_boost": topic_boost, "matched_categories": topic_matches.get("matched_categories", []), "matched_tags": topic_matches.get("matched_tags", []), "ranking_adjustments": adjustments, "relevance_reason": build_relevance_reason(item, query_tokens, topic_matches), }) results.sort(key=lambda x: x["score"], reverse=True) return results[:limit] def print_markdown(wiki_hits: List[Dict[str, Any]], results: List[Dict[str, Any]], query: str) -> None: if not wiki_hits and not results: print("沒有找到適合主動召回的知識。") return print(f"# 對話主動召回結果\n") print(f"查詢：{query}\n") if wiki_hits: print("## Wiki 知識層\n") for hit in wiki_hits: print(f"### {hit['topic_title']} — {hit['section_title']}") print(f"{hit['excerpt']}") print(f"- 來源：{hit['path']}\n") if results: print("## 原始卡片層\n") for idx, item in enumerate(results, start=1): print(f"### {idx}. {item['title']}") print(f"- 分類：{item['category']}") if item.get("tags"): print(f"- 標籤：{', '.join(item['tags'][:8])}") print(f"- 相關原因：{item['relevance_reason']}") print(f"- 分數：{item['score']}") if item.get("summary"): print(f"- 一句話摘要：{item['summary'][:180]}") if item.get("source_url"): print(f"- 原文連結：{item['source_url']}") print(f"- 檔案：{item['relative_path']}\n") def print_prompt(wiki_hits: List[Dict[str, Any]], results: List[Dict[str, Any]], query: str) -> None: if not wiki_hits and not results: print("NO_RECALL") return if wiki_hits: print("根據你的知識庫，這個主題有以下整理：") for hit in wiki_hits: print(f"- [{hit['topic_title']}] {hit['section_title']}：{hit['excerpt'][:120]}") if results: print("相關原始書籤：") for item in results: print(f"- {item['title']}") if item.get("summary"): print(f" 摘要：{item['summary'][:120]}") print(f" 為什麼相關：{item['relevance_reason']}") if item.get("source_url"): print(f" 原文：{item['source_url']}") else: print(f" 檔案：{item['relative_path']}") def print_chat(wiki_hits: List[Dict[str, Any]], results: List[Dict[str, Any]], query: str) -> None: if not wiki_hits and not results: print("NO_RECALL") return if wiki_hits: hit = wiki_hits[0] print(f"[知識庫] {hit['topic_title']}：{hit['excerpt'][:120]}") if results: top = results[:2] print("相關書籤：") for item in top: summary = item.get("summary") or "這篇和你現在聊的主題接近。" reason = item.get("relevance_reason") or "主題接近" print(f"- {item['title']}：{summary[:90]}") print(f" 為什麼相關：{reason}") if item.get("source_url"): print(f" 原文：{item['source_url']}") MODE_LABELS = { "gbrain": "gbrain 混合搜尋（RRF + Gemini）", "semantic": "語意向量搜尋", "keyword": "關鍵字搜尋", "keyword_fallback": "關鍵字搜尋（語意降級）", } def _fts_leg(query: str, limit: int) -> List[Dict[str, Any]]: """BM25 腿。fts_recall 載不進來或索引沒建就回空——不是「查無資料」的訊號， 但也不該讓整條召回中止。""" try: from fts_recall import fts_search except ImportError: return [] try: return fts_search(query, limit) except Exception as e: # noqa: BLE001 try: import xkb_failures xkb_failures.note("bm25 leg", e) except Exception: # noqa: BLE001 pass return [] def search( query: str, limit: int = 3, min_score: int = 6, *, index_file: Path | None = None, vector_file: Path | None = None, topic_profile_file: Path | None = None, no_wiki: bool = False, no_semantic: bool = False, force_semantic: bool = False, force_gbrain: bool = False, fts_only: bool = False, with_bm25: bool = False, # 概念查詢會被 n-gram 撞到，呼叫端明確開 ) -> dict: """兩層召回：先查合成過的 wiki 知識，再查卡片細節。 這段原本長在 main() 裡，所以想用它的人只能另外開一個 Python 行程—— 父行程已經為了語意召回載入過向量索引，子行程再載入一次，一次 hard 召回 因此多付 1.3 秒，而 hook 只給六秒。搬出來之後兩邊呼叫同一份程式， main() 回歸它本來的角色：命令列的殼。 """ index_path = Path(index_file) if index_file else INDEX_FILE vector_path = Path(vector_file) if vector_file else VECTOR_FILE profile_path = Path(topic_profile_file) if topic_profile_file else TOPIC_PROFILE_FILE # 第一層：wiki 知識層（先查合成知識）。fts_only 是 light 掃描的純 BM25 # 模式，呼叫端已經自己查過 wiki，這裡再查一次只是丟掉——跳過。 wiki_hits: List[Dict[str, Any]] = ([] if (no_wiki or fts_only) else wiki_recall(query, limit=2)) # 第二層：cards 細節層（gbrain > semantic > keyword） try: sys.path.insert(0, str(Path(__file__).resolve().parent)) from xbrain_recall import GBRAIN_AVAILABLE as _gbrain_available, GEMINI_API_KEY as _gemini_key except ImportError: _gbrain_available = False _gemini_key = "" use_gbrain = force_gbrain or (not no_semantic and _gbrain_available and bool(_gemini_key)) search_mode = "keyword" if fts_only: # light 掃描專用：跳過所有語意後端，只跑 BM25 腿——一次 SQLite MATCH， # 不打 embedding，light 幾乎每句都跑得起。 results = _fts_leg(query, limit) search_mode = "bm25" if results else "bm25_empty" elif use_gbrain: results = gbrain_semantic_recall(query, limit) if results: search_mode = "gbrain" else: results = recall(query, limit, min_score, index_path, profile_path) search_mode = "keyword_fallback" elif (not no_semantic) and (force_semantic or vector_path.exists()): results = semantic_recall(query, limit, vector_path, index_path, profile_path) if results: search_mode = "semantic" else: results = recall(query, limit, min_score, index_path, profile_path) search_mode = "keyword_fallback" else: results = recall(query, limit, min_score, index_path, profile_path) # BM25 腿：任何語意後端跑完後都併一次（fts_only 模式上面已經跑過，跳過）。 # 這條腿跟語意腿對等——語意查詢語意腿扛，字面查詢 BM25 扛，xkb_score.rank() # 用 RRF 融合。BM25 結果帶 score_scale=card_bm25 / wiki_bm25，自成一條腿。 if not fts_only and with_bm25: # 不在這裡去重。同一張卡被語意腿和 BM25 都撈到是最強訊號——兩筆都要 # 進 xkb_score.rank()，它按 _key 合併、把兩條腿的 RRF 貢獻都算進去。 # with_bm25 由呼叫端按查詢形狀決定：概念查詢不跑 BM25（n-gram 會撞到 # 只共用一個 2-gram 的無關卡片，而下游的餘弦門檻對 *_bm25 是放行的）。 results = results + _fts_leg(query, limit) return { "query": query, "wiki_hits": wiki_hits, "results": results, "search_mode": search_mode, } def main() -> int: parser = argparse.ArgumentParser(description="Recall relevant X bookmarks for conversation use") parser.add_argument("query", nargs="?", help="當前對話查詢") parser.add_argument("--query-file", help="從檔案讀取 query") parser.add_argument("--limit", type=int, default=3) parser.add_argument("--min-score", type=int, default=6) parser.add_argument("--json", action="store_true") parser.add_argument("--format", choices=["markdown", "prompt", "chat"], default="markdown") parser.add_argument("--index-file", default=str(INDEX_FILE)) parser.add_argument("--semantic", action="store_true", help="Force semantic search (default: auto-detect)") parser.add_argument("--no-semantic", action="store_true", help="Force keyword search even if vector index exists") parser.add_argument("--gbrain", action="store_true", help="Use gbrain hybrid search backend (pgvector + RRF + Gemini)") parser.add_argument("--vector-file", default=str(VECTOR_FILE)) parser.add_argument("--topic-profile-file", default=str(TOPIC_PROFILE_FILE)) parser.add_argument("--no-wiki", action="store_true", help="Skip wiki layer, search cards only") args = parser.parse_args() query = args.query or "" if args.query_file: query = Path(args.query_file).read_text(encoding="utf-8").strip() if not query.strip(): print("請提供 query", file=sys.stderr) return 1 # 檢索本身在 search() 裡。原本這裡有一份一模一樣的副本，於是 CLI 與 # hook 走的是兩份會各自漂移的程式——而 commit message 宣稱這裡已經只是 # 命令列的殼了。 found = search( query, limit=args.limit, min_score=args.min_score, index_file=Path(args.index_file), vector_file=Path(args.vector_file), topic_profile_file=Path(args.topic_profile_file), no_wiki=args.no_wiki, no_semantic=args.no_semantic, force_semantic=args.semantic, force_gbrain=args.gbrain, with_bm25=True, # CLI：全部都秀出來 ) wiki_hits = found["wiki_hits"] results = found["results"] search_mode = found["search_mode"] if not args.json: wiki_label = "" if args.no_wiki else f" + Wiki({'有' if wiki_hits else '無'}命中)" print(f"[搜尋模式：{MODE_LABELS.get(search_mode, search_mode)}{wiki_label}]") if args.json: print(json.dumps({ "query": query, "wiki_hits": wiki_hits, "results": results, "search_mode": search_mode, }, ensure_ascii=False, indent=2)) elif args.format == "prompt": print_prompt(wiki_hits, results, query) elif args.format == "chat": print_chat(wiki_hits, results, query) else: print_markdown(wiki_hits, results, query) return 0 import xkb_usage # noqa: E402 — 量測誰在跑，見 scripts/xkb_usage.py if __name__ == "__main__": xkb_usage.record(__file__) raise SystemExit(main())