scripts/xbrain_recall.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
xbrain_recall.py — XKB hybrid search bridge (XBrain layer)
Provides semantic + keyword hybrid search over the XKB knowledge base
using an embedded vector store (pgvector + RRF + Gemini embedding).
Usage:
python3 scripts/xbrain_recall.py "agent memory workflow"
python3 scripts/xbrain_recall.py "SEO 策略" --limit 5 --json
python3 scripts/xbrain_recall.py "query" --no-semantic # keyword-only
Environment / config (in priority order):
GBRAIN_DIR path to vector store runtime (overrides all)
GEMINI_API_KEY required for semantic (vector) search
--env-file explicit dotenv file for runtime credential injection
XKB_ENV_FILE fallback dotenv file when --env-file is omitted
Process environment variables take precedence over either dotenv file. No
host-specific config or private credential fallback is consulted.
"""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any
sys.path.insert(0, str(Path(__file__).resolve().parent))
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
import xkb_failures
from runtime_config import runtime_env
import xkb_provenance
# ── Config ─────────────────────────────────────────────────────────────────────
def _safe_runtime_env() -> dict[str, str]:
"""讀執行期設定；讀不到就當作空的。
這一行在 import 時就會執行，而 load_env_file 對不存在的路徑會拋
FileNotFoundError。呼叫端只接 ImportError，所以一個過期的 XKB_ENV_FILE
會讓「模組載不進來」變成從 search() 裡拋出的 FileNotFoundError，
而不是安靜地退回關鍵字召回。設定不見是要說的，但不是用讓模組無法載入
的方式說。
"""
try:
return runtime_env()
except Exception as err: # noqa: BLE001
print(f"[xkb] xbrain 讀不到執行期設定，改用行程環境：{type(err).__name__}: {err}",
file=sys.stderr, flush=True)
return {}
_RUNTIME_ENV = _safe_runtime_env()
def _resolve_gbrain_dir(settings: dict[str, str] | None = None) -> Path | None:
"""Find the gbrain runtime directory, in priority order."""
settings = _RUNTIME_ENV if settings is None else settings
candidates = [
os.getenv("GBRAIN_DIR"),
settings.get("GBRAIN_DIR"),
str(Path.home() / "Desktop" / "gbrain"),
str(Path.home() / "gbrain"),
"/opt/gbrain",
]
for c in candidates:
if c and Path(c).joinpath("src", "cli.ts").exists():
return Path(c)
return None
GBRAIN_DIR = _resolve_gbrain_dir()
GBRAIN_AVAILABLE = GBRAIN_DIR is not None
def _resolve_bun() -> str:
"""Find bun executable — checks PATH first, then common install locations."""
import shutil
if shutil.which("bun"):
return "bun"
for candidate in [
str(Path.home() / ".bun" / "bin" / "bun"),
"/usr/local/bin/bun",
"/opt/homebrew/bin/bun",
]:
if Path(candidate).exists():
return candidate
return "bun" # let subprocess give the real error
BUN = _resolve_bun()
GEMINI_API_KEY = _RUNTIME_ENV.get("GEMINI_API_KEY", "")
def _make_subprocess_env(
semantic: bool,
settings: dict[str, str] | None = None,
) -> dict[str, str]:
"""Build subprocess env from the portable runtime contract."""
settings = _RUNTIME_ENV if settings is None else settings
env = {**settings}
gemini_api_key = settings.get("GEMINI_API_KEY", "")
if semantic and gemini_api_key:
env["GEMINI_API_KEY"] = gemini_api_key
# gbrain v0.42+ gateway 的 google recipe 讀這個變數名
env["GOOGLE_GENERATIVE_AI_API_KEY"] = gemini_api_key
elif not semantic:
env.pop("GEMINI_API_KEY", None)
env.pop("GOOGLE_GENERATIVE_AI_API_KEY", None)
env.pop("OPENAI_API_KEY", None)
return env
# ── Core query ─────────────────────────────────────────────────────────────────
def xbrain_query(
query: str,
*,
limit: int = 10,
no_expand: bool = True,
semantic: bool = True,
env_file: str | Path | None = None,
diagnostics: dict | None = None,
) -> list[dict[str, Any]]:
"""
Run hybrid search and return structured results.
Returns [] if XBrain is unavailable or query fails (never raises).
Result keys: slug, title, type, chunk_text, score, source_url, stale
"""
state = diagnostics if diagnostics is not None else {}
state.update(available=False, attempted=False, used=False, status="unavailable")
settings = runtime_env(env_file) if env_file is not None else _RUNTIME_ENV
gbrain_dir = _resolve_gbrain_dir(settings)
if gbrain_dir is None:
return []
state.update(available=True, attempted=True)
cmd = [BUN, "run", str(gbrain_dir / "src" / "cli.ts")]
# The human CLI ignores --json and truncates evidence to 100 characters.
# The operation surface returns the same search as lossless JSON.
cmd += ["call", "query", json.dumps({"query": query, "limit": limit,
"expand": not no_expand})]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
encoding="utf-8",
env=_make_subprocess_env(semantic, settings),
cwd=str(gbrain_dir),
# 召回是互動路徑；連不上後端時，這個秒數就是使用者要等的時間。
timeout=int(os.getenv("XKB_XBRAIN_TIMEOUT", "30")),
)
except subprocess.TimeoutExpired as err:
state["status"] = "timeout"
# 逾時、找不到 bun、非零離開碼——三種都回空陣列，而空陣列的意思是
# 「這個主題我們沒有知識」。它們其實是「問不到」。這正是下面那段
# 註解在講的同一件事，只是原本只有最後一種會留下記錄。
xkb_failures.note("semantic recall (xbrain) 逾時", err)
return []
except FileNotFoundError as err:
state["status"] = "unavailable"
xkb_failures.note("semantic recall (xbrain)：PATH 上沒有 bun", err)
return []
except Exception as err:
state["status"] = "error"
# 語意搜尋的後端掛掉，跟「這個主題我們沒有知識」是兩件事。
# 2026-05-04 的十二週靜默故障，就是因為這兩者長得一模一樣。
xkb_failures.note("semantic recall (xbrain)", err)
return []
if result.returncode != 0:
state["status"] = "error"
# 只留離開碼，不要把 stderr 貼進來：後端的錯誤訊息會回音請求內容，
# 而請求裡有憑證。tests/test_xbrain_recall_runtime.py 的
# test_failed_backend_does_not_echo_credential 就是釘這件事的。
xkb_failures.note("semantic recall (xbrain)", RuntimeError(f"exit {result.returncode}"))
return []
raw = result.stdout.strip()
state["status"] = "empty"
if not raw or raw == "No results.":
return []
truncated_display = False
try:
items = json.loads(raw)
except json.JSONDecodeError:
# gbrain v0.42+ 移除了 query --json，輸出為「[分數] slug -- 內文」行格式
items = _parse_line_format(raw)
truncated_display = True
if not items:
state["status"] = "invalid_response"
return []
if not isinstance(items, list) or not all(isinstance(item, dict) for item in items):
state["status"] = "invalid_response"
return []
results = []
for item in items:
chunk = item.get("chunk_text", "")
slug = item.get("slug", "")
source_url = _source_url(item)
results.append({
"slug": slug,
"title": item.get("title", ""),
"type": item.get("type", ""),
"chunk_text": chunk,
"score": item.get("score", 0.0),
"score_scale": "card_hybrid",
"source_url": source_url,
"stale": item.get("stale", False),
**({"excerpt_truncated": True} if truncated_display else {}),
})
state.update(used=bool(results), status="used" if results else "empty")
return results
def _parse_line_format(raw: str) -> list[dict[str, Any]]:
"""解析 gbrain v0.42+ 的行格式輸出：`[0.9102] slug -- chunk text`。
非結果行（gateway 警告等）自動略過。"""
import re
# gbrain 在最後一筆結果之後還會印東西（警告、計時、棄用提示）。原本的
# 續行判斷是「不是結果、而且已經看過至少一筆」，所以那些行全部被接到
# 最後一張卡的內文裡，然後當成知識顯示給使用者看。
RESULT = re.compile(r"^\[(\d+\.\d+)\]\s+(\S+)\s+--\s?(.*)$")
NOISE = re.compile(
r"^\s*(?:\[|warn|warning|error|deprecat|note:|info:|took\b|elapsed\b|\d+\s*ms\b)",
re.I,
)
MAX_CONTINUATION_CHARS = 4000
items: list[dict[str, Any]] = []
for line in raw.splitlines():
stripped = line.strip()
match = RESULT.match(stripped)
if match:
items.append({
"slug": match.group(2),
"title": "",
"type": "",
"chunk_text": match.group(3),
"score": float(match.group(1)),
"stale": False,
})
continue
if not items or not stripped or NOISE.match(stripped):
continue
# 續行。原本用未 strip 的行判斷開頭，所以縮排過的結果會被當成內文。
last = items[-1]
if len(last["chunk_text"]) < MAX_CONTINUATION_CHARS:
last["chunk_text"] += "\n" + stripped
return items
def _source_url(item: dict) -> str:
"""Prefer exact metadata, then canonical frontmatter; never guess GitHub."""
for metadata in (item, item.get("frontmatter") or {}):
if metadata.get("source_url"):
return str(metadata["source_url"])
slug = item.get("slug", "")
if slug.startswith("xkb-case-"):
try:
slug = bytes.fromhex(slug.removeprefix("xkb-case-")).decode("utf-8")
except (ValueError, UnicodeError):
pass
import xkb_paths
import xkb_frontmatter
roots = (xkb_paths.CARDS_DIR, xkb_paths.BOOKMARKS_DIR / "youtube")
for root in roots:
leaf = slug.removeprefix("youtube-") if root.name == "youtube" else slug
candidate = root / (leaf + ".md")
try:
if candidate.is_file() and candidate.resolve().is_relative_to(root.resolve()):
value = xkb_frontmatter.parse(candidate.read_text(encoding="utf-8")).get("source_url")
if value:
return str(value).strip("\"'")
except (OSError, UnicodeError):
# Local provenance is optional; a damaged local copy must not erase
# otherwise valid remote search hits.
continue
return _extract_source_url(item.get("chunk_text", "")) or _url_from_slug(slug)
def _extract_source_url(text: str) -> str:
"""Extract source URL from card content (multiple patterns)."""
import re
patterns = [
r"Source URL:\s*(https?://\S+)",
r"來源:\s*(https?://\S+)",
r"source_url:\s*(https?://\S+)",
r"- (?:Tweet|原始來源|Source):\s*(https?://\S+)",
]
for pat in patterns:
m = re.search(pat, text)
if m:
return m.group(1).rstrip(".,)")
return ""
def _url_from_slug(slug: str) -> str:
"""Reconstruct source URL from slug naming conventions."""
import re
# Tweet IDs: 15-20 digit slugs → x.com URL
if re.fullmatch(r"\d{15,20}", slug):
return f"https://x.com/i/status/{slug}"
# legacy-* slugs derived from tweet files: extract numeric suffix
m = re.search(r"(\d{15,20})$", slug)
if m:
return f"https://x.com/i/status/{m.group(1)}"
# YouTube slugs: youtube-{video_id}
m = re.match(r"youtube-([A-Za-z0-9_\-]{11})$", slug)
if m:
return f"https://www.youtube.com/watch?v={m.group(1)}"
# GitHub owner/repo both allow hyphens: this encoding is irreversible.
# Never manufacture a provenance URL from it.
return ""
# ── Formatted output ───────────────────────────────────────────────────────────
def format_for_conversation(results: list[dict], query: str) -> str:
"""Format recall results for injecting into conversation context."""
if not results:
return ""
lines = [f"◆ 相關既有知識（query: {query!r}）\n"]
for i, r in enumerate(results[:5], 1):
title = xkb_provenance.strip_markers(r["title"] or r["slug"])
url = r["source_url"]
snippet = xkb_provenance.strip_markers(
r["chunk_text"][:200].replace("\n", " ").strip())
score_pct = f"{r['score'] * 100:.1f}%"
lines.append(f"{i}. **{title}** ({score_pct})")
if snippet:
lines.append(f" {snippet}…")
if url:
lines.append(f" {url}")
lines.append("")
return "\n".join(lines)
# ── Backward compatibility alias ──────────────────────────────────────────────
# Old scripts that import gbrain_recall can import from here instead.
gbrain_query = xbrain_query
# ── CLI ────────────────────────────────────────────────────────────────────────
def main() -> None:
parser = argparse.ArgumentParser(description="XBrain hybrid search")
parser.add_argument("query", help="Search query")
parser.add_argument("--limit", type=int, default=10)
parser.add_argument("--json", action="store_true", dest="output_json")
parser.add_argument("--no-semantic", action="store_true")
parser.add_argument(
"--env-file",
help="Explicit XKB dotenv file (process environment takes precedence)",
)
args = parser.parse_args()
try:
settings = runtime_env(args.env_file) if args.env_file else _RUNTIME_ENV
except (FileNotFoundError, OSError, ValueError) as exc:
print(f"Unable to load XKB runtime configuration: {exc}", file=sys.stderr)
sys.exit(1)
if _resolve_gbrain_dir(settings) is None:
print("XBrain not available. Set GBRAIN_DIR or clone to ~/Desktop/gbrain.",
file=sys.stderr)
sys.exit(1)
if not args.no_semantic and not settings.get("GEMINI_API_KEY"):
print(
"Semantic recall requires GEMINI_API_KEY via the process environment "
"or the selected XKB env file (process environment takes precedence).",
file=sys.stderr,
)
sys.exit(1)
results = xbrain_query(
args.query,
limit=args.limit,
semantic=not args.no_semantic,
env_file=args.env_file,
)
if args.output_json:
print(json.dumps(results, ensure_ascii=False, indent=2))
else:
print(format_for_conversation(results, args.query))
import xkb_usage # noqa: E402 — 量測誰在跑，見 scripts/xkb_usage.py
if __name__ == "__main__":
xkb_usage.record(__file__)
main()