#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ knowledge_points · 知识点嵌套拆分器（v1.16 · 图架构增强 2/3） 设计者方向（2026-08-18）：知识点级嵌套——知识卡 ⊃ 知识点节点， 提高子图检索能力与精确度（问题直接命中知识点，而非整卡扫描）。 拆分：学科卡 content「N. 知识点名: 内容」→ 每知识点一个节点 - 节点：layer=knowledge，tags=[knowledge_point, card:{卡id}, domain, edu] state_attributes={name, kind:"knowledge_point", parent_card, domain, edu_level} - 边：卡 -hierarchical-> 知识点（verified=1：知识点继承已验证卡） - 幂等：按 (parent_card, 知识点名) 查重 检索适配（配合 dex_respond 的 SQL 预过滤 + recursive_item_answer）： 知识点节点带 knowledge_point 标签 → 卡扫描跳过（不拖慢）；知识点级精确命中 由 find_points / recursive_item_answer 优先消费。 规模：75 卡 ≈ 2802 知识点（3 语言卡 + 元学科卡无「N.」格式，跳过）。 纯标准库 · 零外部依赖 """ import json import os import re import sqlite3 import sys import time from typing import Dict, List, Optional # 默认库路径：LINGSHU_DB 环境变量优先，否则取当前用户家目录下的 .dsh 布局 # （不硬编码本机绝对路径——避免把用户名/目录结构带进发布产物） DEFAULT_DB = os.environ.get("LINGSHU_DB") or os.path.join( os.path.expanduser("~"), ".dsh", "profiles", "web", "data", "lingshu.db") HERE = os.path.dirname(os.path.abspath(__file__)) _KP_RE = re.compile(r"^\s*(\d+)\.\s+(.+?)[:：]\s*(.*)$") class KnowledgePointSplitter: """知识卡 content → 知识点节点 + 卡⊃知识点 hierarchical 边 双模（写路径收口 · md 真源裁定）： - `cg` 给定 → **md 真源模式**：写经 `cg.add`（mdcg 唯一写入闸门）+ `cg.append_subgraph_node`（层级边落点=父节点 subgraph.nodes，与迁移 导出形态一致）；读经 `md_access.MdConn`（WB 同款只读访问层）。 - `cg=None` → sqlite 派生库旧路（兜底：无 md_cg 环境仍可运行）。 """ # nodes 表 16 列序（md 行 tuple → dict 的键序，与 md_access 行同构一致） _COLS = ("id", "content", "modality", "spatial_coordinates", "temporal_coordinate", "condition_space", "importance", "confidence", "layer", "access_count", "last_access", "created_at", "tags", "semantic_coordinates", "state_attributes", "entity_id") def __init__(self, db_path: str = DEFAULT_DB, cg=None): """`cg`（MdCG/MdCGSecure）给定走 md 真源，否则 sqlite 连接初始化。""" self.db_path = db_path self.cg = cg self._mdc = None if cg is None: self.conn = sqlite3.connect(db_path, timeout=15) self.conn.row_factory = sqlite3.Row else: self.conn = None def _md_conn(self): """md 模式的只读连接（懒加载，root 取 cg.root）。""" if self._mdc is None: from md_access import MdConn self._mdc = MdConn(self.cg.root) return self._mdc def _query(self, sql: str): """按模式执行只读查询，返回支持 row[列名] 访问的行列表。 md 模式把 SELECT 列表规范化为 *（16 列 dict），谓词不变—— 三条查询（tags LIKE 单/双谓词）均在 md_access 只读文法内。 """ if self.cg is None: return self.conn.execute(sql).fetchall() sql_star = re.sub(r"SELECT\s+.+?\s+FROM", "SELECT * FROM", sql, count=1, flags=re.I | re.S) return [dict(zip(self._COLS, r)) for r in self._md_conn().execute(sql_star).fetchall()] def _write_point(self, pid, pcontent, tags, cs, sa, card_id, card_imp, now, pname): """写一个知识点节点 + 卡⊃知识点层级边（按模式分派）。""" if self.cg is not None: self.cg.add(pid, pcontent, layer="knowledge", tags=tags, condition_space=json.loads(cs), importance=round(card_imp * 0.9, 2), confidence=0.6, modality="text", spatial={}, temporal=now, semantic_coordinates={}, state_attributes=json.loads(sa), verification_basis="textbook", edges=[]) # 卡 -hierarchical-> 知识点：md 落点=父节点 subgraph.nodes self.cg.append_subgraph_node(card_id, pid) else: c = self.conn.cursor() c.execute( "INSERT INTO nodes (id, content, modality, spatial_coordinates, " "temporal_coordinate, condition_space, importance, confidence, layer, " "access_count, last_access, created_at, tags, semantic_coordinates, " "state_attributes) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (pid, pcontent, "text", "{}", now, cs, round(card_imp * 0.9, 2), 0.6, "knowledge", 0, None, now, json.dumps(tags, ensure_ascii=False), "{}", sa)) # 卡 -hierarchical-> 知识点（verified：知识点继承已验证卡） eid = f"edge_kp_{abs(hash(card_id + pname)) % 10**10}_{int(now * 1000)}" c.execute( "INSERT INTO edges (id, source_id, target_id, relation_type, " "condition_space, confidence, weight, verified, created_at, " "last_verified, source_evidence) VALUES (?,?,?,?,?,?,?,?,?,?,?)", (eid, card_id, pid, "hierarchical", cs, 0.9, 0.9, 1, now, now, "extracted")) self.conn.commit() # ---------------- 拆分 ---------------- def parse_points(self, content: str) -> List[tuple]: """content → [(知识点名, 内容)] 学科卡「N. 名: 内容」格式；语言卡「①名——内容；②…」格式（v1.16 补全）。""" points = [] for ln in (content or "").split("\n"): m = _KP_RE.match(ln) if m: points.append((m.group(2).strip(), m.group(3).strip())) if points: return points return self.parse_lang_points(content or "") @staticmethod def parse_lang_points(content: str) -> List[tuple]: """语言卡「语言名 的规律性：①名——内容；②…；适用条件：…」格式""" points = [] segs = re.split(r"([①②③④⑤⑥])", content) for i in range(1, len(segs), 2): seg = segs[i + 1] if i + 1 < len(segs) else "" seg = re.split(r"适用条件", seg)[0].strip().rstrip("。；;") if not seg: continue m = re.match(r"([^——：]{1,20})(?:——|：)(.*)", seg) if m: points.append((m.group(1).strip(), m.group(2).strip())) else: points.append((seg[:20], seg)) m = re.search(r"适用条件[：:]([^。]*。?)", content) if m: points.append(("适用条件", m.group(1).strip())) return points def split_card(self, card_id: str, name: str, domain: str, edu: str, content: str, card_imp: float, dry_run: bool = False) -> Dict: """拆一张卡 → 知识点节点 + hierarchical 边（幂等）""" points = self.parse_points(content) # 已存在的知识点名（幂等）。md 模式走 cg index 直查（不走 MdConn # 快照——本进程刚写入的 kp 必须立即可见），sqlite 模式实时 SQL。 existing = set() if self.cg is not None: pref = f"card:{card_id[:16]}" for nid, e in list(self.cg.index["nodes"].items()): if pref not in (e.get("tags") or []): continue try: node = self.cg.get(nid) or {} fm = node.get("frontmatter") or {} nm = (fm.get("state_attributes") or {}).get("name", "") if nm: existing.add(nm) except Exception: pass else: for row in self.conn.execute( "SELECT state_attributes FROM nodes WHERE tags LIKE ?", ("%card:" + card_id[:16] + "%",)).fetchall(): try: nm = json.loads(row["state_attributes"] or "{}").get( "name", "") if nm: existing.add(nm) except Exception: pass added = 0 skipped = 0 now = time.time() for pname, pcontent in points: if pname in existing: skipped += 1 continue if dry_run: added += 1 continue pid = f"kp_{abs(hash(card_id + pname)) % 10**10}_{int(now * 1000)}" tags = ["knowledge_point", "subject_card", f"card:{card_id[:16]}", f"domain:{domain}"] if edu: tags.append(f"edu:{edu}") sa = json.dumps({"name": pname, "kind": "knowledge_point", "parent_card": card_id, "domain": domain, "edu_level": edu}, ensure_ascii=False) cs = json.dumps({"observation_position": f"{name} 知识点（{pname}）", "observation_tool": "学科知识卡拆分", "time_window": [0.0, 9999999999.0], "existence_constraint": "通用规律/知识（开源非盈利知识库）"}, ensure_ascii=False) self._write_point(pid, pcontent, tags, cs, sa, card_id, card_imp, now, pname) added += 1 return {"card": name, "points": len(points), "added": added, "skipped": skipped} def split_all(self, dry_run: bool = False) -> Dict: """全库拆分（扫描 subject_card 卡）""" report = {"cards": 0, "points_total": 0, "added": 0, "skipped": 0, "no_format": []} cards = self._query( "SELECT id, state_attributes, content, importance FROM nodes " "WHERE tags LIKE '%subject_card%' AND tags NOT LIKE '%knowledge_point%'") for row in cards: sa = json.loads(row["state_attributes"] or "{}") name = sa.get("name", "") content = row["content"] or "" points = self.parse_points(content) if not points: report["no_format"].append(name or row["id"][:16]) continue report["cards"] += 1 report["points_total"] += len(points) r = self.split_card(row["id"], name, sa.get("domain", ""), sa.get("edu_level", ""), content, row["importance"] or 0.7, dry_run=dry_run) report["added"] += r["added"] report["skipped"] += r["skipped"] return report # ---------------- 知识点级精确检索 ---------------- def find_points(self, query: str, limit: int = 8) -> List[Dict]: """问题 → 知识点精确命中（encode 规范词 LIKE 匹配 name/content）""" fp = {} try: sys.path.insert(0, HERE) from semantic_translate import encode fp = encode(query) except Exception: pass scored = [] for row in self._query( "SELECT id, content, state_attributes, importance, tags " "FROM nodes WHERE tags LIKE '%knowledge_point%'"): sa = json.loads(row["state_attributes"] or "{}") name = sa.get("name", "") text = name + " " + (row["content"] or "") s = 0.0 for term, w in fp.items(): if term in text: s += w # v1.16 术语兜底：学科/语言术语不在翻译表（encode 空）， # 用问题窗口直接匹配知识点名（中文 2-4 字 + 英文词） if not fp or s == 0: _hit = False _chars = [ch for ch in query if "\u4e00" <= ch <= "\u9fff"] for i in range(len(_chars)): for L in (4, 3, 2): if i + L <= len(_chars): w = "".join(_chars[i:i + L]) if w in name: # 窗口越长越精确：4字 2.0 / 3字 1.5 / 2字 1.0 s += {4: 2.0, 3: 1.5, 2: 1.0}[L] _hit = True break if _hit: break if not _hit: for _m in re.finditer(r"[A-Za-z]{2,}", query): if _m.group(0).lower() in text.lower(): s += 1.5 break if s > 0: scored.append({"id": row["id"], "name": name, "content": (row["content"] or "")[:200], "score": round(s, 3), "importance": row["importance"], "card": sa.get("parent_card", "")[:16]}) # 同分：短名优先（基础概念优先于长名细分，如「密度」>「密度矩阵与纠缠」） scored.sort(key=lambda x: (-x["score"], len(x["name"]), -x["importance"])) return scored[:limit] def close(self): """收尾：md 模式 flush 持久化索引；sqlite 模式关连接。""" if self.cg is not None: self.cg.flush() else: self.conn.close() if __name__ == "__main__": sys.stdout.reconfigure(encoding="utf-8") _cg = None if (os.environ.get("WB_MD_DIRECT") == "1" and os.environ.get("WB_MD_ROOT")): # md 真源模式：WB_MD_ROOT 即语料根（拆分结果写入 md 语料） sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname( os.path.abspath(__file__))))) from md_cg.mdcg import MdCG _cg = MdCG(root=os.environ["WB_MD_ROOT"]) print(f"[md] 真源根: {_cg.root}") sp = KnowledgePointSplitter(cg=_cg) dry = "--dry-run" in sys.argv rep = sp.split_all(dry_run=dry) print(json.dumps(rep, ensure_ascii=False, indent=1)) if not dry: hits = sp.find_points("水为什么烧开") print("\nfind_points('水为什么烧开'):") for h in hits[:5]: print(" %.2f %s: %s" % (h["score"], h["name"], h["content"][:40])) sp.close()