Create-knowledge-find-decode/scripts/scope_link.py

"""作用域定位（scope-link）：候选值 → 本地树 embedding 余弦最近邻 → top-K。

对得上（高分）→ 复用现有节点原名，ingest 时按名挂靠；对不上 → 保留为新建值（丰富树）。
本地内存 + numpy 暴力最近邻，无需向量数据库。

编程接口：
    from scripts.scope_link import ScopeLinker
    ScopeLinker().link("撕裂共识", source_type="作用", top_k=5)

自测：python scripts/scope_link.py
"""
from __future__ import annotations

import json
from pathlib import Path

import numpy as np

from core.embedding import ArkEmbedConfig, embed_text

OUT = Path("scope_trees")


class ScopeLinker:
    def __init__(self, env_file: str = ".env") -> None:
        self.index = json.loads((OUT / "trees_index.json").read_text(encoding="utf-8"))
        emb = np.load(OUT / "trees_embeddings.npy")
        self.norm = emb / (np.linalg.norm(emb, axis=1, keepdims=True) + 1e-9)
        self.src = np.array([it["source_type"] for it in self.index])
        self.cfg = ArkEmbedConfig.from_env(env_file)

    def link(self, candidate: str, source_type: str | None = None, top_k: int = 5) -> list[dict]:
        q = np.asarray(embed_text(candidate, self.cfg), dtype=np.float32)
        q = q / (np.linalg.norm(q) + 1e-9)
        sims = self.norm @ q
        idxs = (np.where(self.src == source_type)[0] if source_type
                else np.arange(len(sims)))
        order = idxs[np.argsort(-sims[idxs])][:top_k]
        return [{
            "name": self.index[i]["name"], "path": self.index[i]["path"],
            "source_type": self.index[i]["source_type"],
            "score": round(float(sims[i]), 4),
        } for i in order]


def _selftest() -> None:
    sl = ScopeLinker()
    cases = [
        ("撕裂共识", "作用"), ("撕裂共识", "意图"), ("三幕结构", "形式"),
        ("角色代入共鸣", "感受"), ("反转", "形式"), ("引发讨论", "作用"),
    ]
    for cand, st in cases:
        print(f"\n[{cand}] @ {st}:")
        for hit in sl.link(cand, st, 3):
            print(f"  {hit['score']:.3f}  {hit['name']}   ({hit['path']})")


if __name__ == "__main__":
    _selftest()