#!/usr/bin/env python """在本机 bmem 镜像里检索历史：语义条目 / 文档章节 / 事件。 bmem 的 CLI 目前只负责写入与同步，没有检索命令，但它把 Notes 全文、章节拆分、 语义条目和事件都存进了本机 SQLite 镜像。这个脚本把 bmem 当库调用，只读地把它们 查出来——不改任何数据，不连中心库，离线可用。 用法： python bmem_recall.py "多径 时延" # 三类都查 python bmem_recall.py "CP 长度" --project amem # 限定项目 python bmem_recall.py "时间戳" --kind decision # 只看某类语义条目 python bmem_recall.py --recent 10 # 不给关键词=看最近发生了什么 中文按 bigram 检索（bmem 写入时做了滑窗补偿），所以两到四个字的词组比整句命中率高。 """ from __future__ import annotations import argparse import os import subprocess import sys from pathlib import Path def _bootstrap() -> None: """bmem 装在 langgraph conda 环境里，当前解释器没有时换一个再跑。""" try: import bmem # noqa: F401 return except ImportError: pass home = Path(os.path.expanduser("~")) candidates = [ home / "miniconda3" / "envs" / "langgraph" / "python.exe", home / ".conda" / "envs" / "langgraph" / "python.exe", home / "miniconda3" / "envs" / "langgraph" / "bin" / "python", home / ".conda" / "envs" / "langgraph" / "bin" / "python", ] for python in candidates: if python.exists() and Path(sys.executable).resolve() != python.resolve(): env = dict(os.environ, PYTHONUTF8="1", PYTHONIOENCODING="utf-8") raise SystemExit( subprocess.run([str(python), __file__, *sys.argv[1:]], env=env).returncode ) raise SystemExit( "找不到 bmem 包。它以 editable 方式装在 langgraph conda 环境里；\n" "确认 `bmem help` 能跑，或查 ~/skills_env.md 里记录的解释器路径。" ) _bootstrap() from bmem import config as config_mod # noqa: E402 from bmem.localdb import Database, fts_query # noqa: E402 def _print_facts(db: Database, query: str, prj_id: str | None, kind: str | None, limit: int) -> int: sql = ( "SELECT f.prj_id, f.rel_path, f.kind, f.status, f.title, f.body, f.event_date, " "f.extraction_gen FROM doc_facts f WHERE 1=1" ) params: list[object] = [] if prj_id: sql += " AND f.prj_id = ?" params.append(prj_id) if kind: sql += " AND f.kind = ?" params.append(kind) if query: sql += " AND (f.title LIKE ? OR f.body LIKE ?)" params += [f"%{query}%", f"%{query}%"] sql += " ORDER BY f.event_date DESC NULLS LAST, f.rel_path, f.fact_no LIMIT ?" params.append(limit) rows = db.rows(sql, params) if not rows: return 0 print("=== 语义条目 (doc_facts) ===") for row in rows: status = f" [{row['status']}]" if row["status"] else "" date = f" {row['event_date']}" if row["event_date"] else "" print(f"\n[{row['kind']}]{status}{date} {row['title']}") print(f" 来源 {row['prj_id']}:{row['rel_path']} (第 {row['extraction_gen']} 代抽取)") body = row["body"].strip().replace("\n", "\n ") print(f" {body[:400]}{'…' if len(row['body']) > 400 else ''}") return len(rows) def _print_sections(db: Database, query: str, prj_id: str | None, limit: int) -> int: """章节级命中：比整篇文档精确，直接告诉你该回读哪一节。""" sql = ( "SELECT s.prj_id, s.rel_path, s.section_no, s.heading, s.level, s.body_md " "FROM note_sections s JOIN (" " SELECT prj_id, rel_path, content_sha256 FROM notes_revisions r1 " " WHERE imported_at = (SELECT max(imported_at) FROM notes_revisions r2 " " WHERE r2.prj_id=r1.prj_id AND r2.rel_path=r1.rel_path)" ") latest ON latest.prj_id=s.prj_id AND latest.rel_path=s.rel_path " " AND latest.content_sha256=s.content_sha256 " "WHERE s.body_md LIKE ?" ) params: list[object] = [f"%{query}%"] if prj_id: sql += " AND s.prj_id = ?" params.append(prj_id) sql += " ORDER BY s.prj_id, s.rel_path, s.section_no LIMIT ?" params.append(limit) rows = db.rows(sql, params) if not rows: return 0 print("\n=== 文档章节 (note_sections，仅最新修订) ===") for row in rows: heading = row["heading"] or "（前言）" print(f"\n{row['prj_id']}:{row['rel_path']} § {row['section_no']} {heading}") body = row["body_md"] pos = body.lower().find(query.lower()) start = max(0, pos - 120) if pos >= 0 else 0 snippet = body[start : start + 400].strip().replace("\n", "\n ") print(f" …{snippet}…") return len(rows) def _print_events(db: Database, query: str, prj_id: str | None, limit: int) -> int: sql = "SELECT ts, machine_id, prj_id, event_type, result, summary FROM events WHERE 1=1" params: list[object] = [] if prj_id: sql += " AND prj_id = ?" params.append(prj_id) if query: sql += " AND summary LIKE ?" params.append(f"%{query}%") sql += " ORDER BY ts DESC LIMIT ?" params.append(limit) rows = db.rows(sql, params) if not rows: return 0 print("\n=== 事件 (events) ===") for row in rows: result = f" -> {row['result']}" if row["result"] else "" print( f" {row['ts'][:16]} {row['machine_id']} {row['prj_id']} " f"[{row['event_type']}{result}] {row['summary'][:100]}" ) return len(rows) def main() -> int: parser = argparse.ArgumentParser(description="在本机 bmem 镜像里只读检索历史") parser.add_argument("query", nargs="?", default="", help="关键词；中文用两到四个字的词组") parser.add_argument("--project", help="限定 prjID") parser.add_argument("--kind", help="限定语义条目类型：decision/issue/requirement/result/param/…") parser.add_argument("--limit", type=int, default=8) parser.add_argument("--recent", type=int, metavar="N", help="不检索，直接看最近 N 条事件") args = parser.parse_args() try: cfg = config_mod.load() except Exception as exc: # noqa: BLE001 print(f"本机还没加入 bmem：{exc}\n先跑 `bmem install`。", file=sys.stderr) return 1 if not cfg.db_path.exists(): print(f"本地镜像不存在：{cfg.db_path}\n跑一次 `bmem sync` 会从中心库重建。", file=sys.stderr) return 1 with Database.open(cfg) as db: if args.recent: _print_events(db, "", args.project, args.recent) return 0 if not args.query and not args.kind: parser.error("给一个关键词，或用 --kind 列举某类条目，或用 --recent N 看最近发生了什么") hits = _print_facts(db, args.query, args.project, args.kind, args.limit) if args.query: # 空关键词配 LIKE '%%' 会把整库倒出来，只在真的有关键词时查这两类 hits += _print_sections(db, args.query, args.project, args.limit) hits += _print_events(db, args.query, args.project, args.limit) if hits == 0: print(f"没有命中「{args.query}」。") print("中文按 bigram 索引，换更短的词组再试；也可能这条历史确实没记过。") print(f"（FTS 等价查询串：{fts_query(args.query)}）") else: print(f"\n共 {hits} 条命中。命中只是线索——回读原文再下结论。") return 0 if __name__ == "__main__": raise SystemExit(main())