sync: WIN11-BOWEI 2026-09-02 新增 bmem 记忆 skill；imx6-flow/plan-review 文档与脚本更新
远端 pack 服务仍故障（git-upload-pack 返回 0008NAK 后截断），push 待服务器恢复。 未跟踪的 5 个归档临时副本（rtl-ip-replace / sim-rtl-flow / simulink_rtl_dev / simulink-rtl-redev / zynq-imx6-flow）刻意不入库，待 pull 后由远端提供。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0135YszNe4pwwJxwaLqKTCKG
This commit is contained in:
Myfro9
@@ -0,0 +1,193 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user