guanlan-src/app_ontology/app_ontology_guanlan/ontology/retrieval.py

# -*- coding: utf-8 -*-
"""本体检索层 (2026-08-28 用户令: ontology 检索层优化)。

问答慢的根因不是模型慢 (裸响应 1.1s), 是它要多轮试探才找到该调哪个工具、该取哪个对象。
预取路由实测只对"有唯一确定入口"的问法有效 (报警码), 对汇总类问法无效 —— 差在
"按语义找到相关对象"这一步没有工具能做。本模块补上这一步。

两路混合, 各补各的短板:
  BM25 (字符 2-gram)  — 术语精确匹配。风电问句里的关键信息多是标识符与专名:
                        3225 / WTG29 / BPFI / 集中润滑泵 — 这类词向量反而会"想歪"。
                        中文用字符 n-gram 免分词, 5,617 对象的索引在毫秒级。
  向量 (bge-m3)       — 语义召回。"哪些机组该重点盯"与"风险在升"字面零重合, 只有向量能连上。
                        模型不在时自动降级为纯 BM25, 不让检索层因为缺模型而整个不可用。

索引落 outputs/rudong/ontology/retrieval_index.npz + .json, 随 objects.json 的 mtime 失效重建。
"""

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]
from src import paths as P
import json, math, pathlib, re, collections
import numpy as np

ROOT = _install_root(__file__)
OBJ = P.objects_json()
IDX = P.ont() / 'retrieval_index.json'
VEC = P.ont() / 'retrieval_vec.npy'
OLLAMA_EMB = 'http://127.0.0.1:11434/api/embed'
EMB_MODEL = 'bge-m3'
K1, B = 1.5, 0.75

# 类型先验 (2026-08-28 实测): 问"哪些机组风险在上升"时 BM25 把说明书与工单排在最前 —
# 它们文本长、词多, 但人要的是结论对象。工单/文档是**证据**不是**结论**, 除非问句里
# 明确要工单或图纸, 否则该往后排。权重只调排序不改召回, 命中仍在结果集里。
TYPE_PRIOR = {
    'Verdict': 1.35, 'FailureMechanism': 1.30, 'FailureMode': 1.25, 'Decision': 1.25,
    'Evidence': 1.15, 'AlarmCode': 1.15, 'WorkInstruction': 1.10, 'MaintTask': 1.05,
    'Component': 1.05, 'Turbine': 1.05, 'OilSample': 0.95,
    'WorkOrder': 0.80, 'Doc': 0.75,
}
# 问句显式点名这些时, 对应类型的降权取消
ASK_TYPE = [(r'工单|检修记录|做过什么|台账', 'WorkOrder'),
            (r'图纸|手册|说明书|资料|文档|原理', 'Doc'),
            (r'油样|油液|化验', 'OilSample')]


# 不进检索文本的字段: 文件路径与数据源指针 — 它们很长且全场雷同, 会把 BM25 的文档长度
# 归一化搅乱, 且没人会用路径来找对象。
SKIP_KEYS = {'path', 'source', 'src', 'file', 'channel', 'rel', 'size_kb', 'id'}


def obj_text(o):
    """对象 → 可检索文本。

    ★取所有标量 props 而非固定字段白名单 (2026-08-28 实测): 白名单漏掉了工单的
    turbine/loc2/part/action、油样的 turbine/comp —— 于是"29号做过什么工单"召回不到任何工单。
    字段名会随摄入器演进, 白名单必然滞后; 改为全取 + 排除路径类噪声。
    """
    p = o.get('props') or {}
    parts = [str(o.get('id', '')), str(o.get('type', ''))]
    for k, v in p.items():
        if k in SKIP_KEYS or v is None:
            continue
        if isinstance(v, (str, int, float, bool)) and str(v).strip():
            parts.append(str(v))
        elif isinstance(v, list) and v and all(isinstance(x, (str, int, float)) for x in v):
            parts.append(' '.join(str(x) for x in v[:8]))
    return ' '.join(parts)[:600]


def _grams(s, n=2):
    """字符 n-gram + 数字/英文整词 — 中文免分词, 同时保住 3225 / WTG29 / BPFI 这类标识符。"""
    s = s.lower()
    toks = re.findall(r'[a-z]{2,}|\d{2,}', s)
    cn = re.sub(r'[^一-鿿]', ' ', s)
    for seg in cn.split():
        toks += [seg[i:i + n] for i in range(max(len(seg) - n + 1, 1))]
    return toks


def build(use_vec=True, verbose=False):
    d = json.loads(OBJ.read_text(encoding='utf-8'))
    ids = list(d)
    texts = [obj_text(d[i]) for i in ids]
    # ── BM25 索引 ──
    df, docs = collections.Counter(), []
    for t in texts:
        tf = collections.Counter(_grams(t))
        docs.append(tf)
        df.update(tf.keys())
    N = len(docs)
    avgdl = sum(sum(x.values()) for x in docs) / max(N, 1)
    inv = collections.defaultdict(list)          # term -> [(doc_i, tf)]
    for i, tf in enumerate(docs):
        for w, c in tf.items():
            inv[w].append((i, c))
    idf = {w: math.log(1 + (N - n + 0.5) / (n + 0.5)) for w, n in df.items()}
    dl = [sum(x.values()) for x in docs]
    IDX.write_text(json.dumps(dict(ids=ids, inv={k: v for k, v in inv.items()},
                                   idf=idf, dl=dl, avgdl=avgdl,
                                   src_mtime=OBJ.stat().st_mtime), ensure_ascii=False),
                   # 索引里全是中文词条: 不给 encoding 会按系统 locale(cp936) 写, 中文 Windows 上直接抛
                   # UnicodeEncodeError (与 src/ontology/store.py 同一类坑, 2026-09-11 逮)
                   encoding='utf-8')
    n_vec = 0
    if use_vec:
        try:
            v = _embed_all(texts, verbose)
            np.save(VEC, v)
            n_vec = len(v)
        except Exception as e:
            if verbose:
                print(f'  向量层跳过 ({str(e)[:70]}) — 降级为纯 BM25')
            if VEC.exists():
                VEC.unlink()
    return dict(objects=N, terms=len(inv), vectors=n_vec)


def _embed_all(texts, verbose=False, batch=64):
    import urllib.request
    out = []
    for i in range(0, len(texts), batch):
        chunk = texts[i:i + batch]
        req = urllib.request.Request(OLLAMA_EMB,
                                     data=json.dumps(dict(model=EMB_MODEL, input=chunk)).encode(),
                                     headers={'Content-Type': 'application/json'})
        with urllib.request.urlopen(req, timeout=600) as r:
            e = json.loads(r.read())['embeddings']
        out.extend(e)
        if verbose:
            print(f'  向量 {len(out)}/{len(texts)}', flush=True)
    a = np.asarray(out, dtype='float32')
    return a / (np.linalg.norm(a, axis=1, keepdims=True) + 1e-9)


_M = {}


def _load():
    if _M.get('ids') and _M.get('mtime') == OBJ.stat().st_mtime:
        return _M
    if not IDX.exists() or json.loads(IDX.read_text(encoding='utf-8'))['src_mtime'] != OBJ.stat().st_mtime:
        build(verbose=False)
    j = json.loads(IDX.read_text(encoding='utf-8'))
    d = json.loads(OBJ.read_text(encoding='utf-8'))
    types = [d.get(i, {}).get('type', '') for i in j['ids']]
    has_turbine = [bool((d.get(i, {}).get('props') or {}).get('turbine')) for i in j['ids']]
    _M.update(j, mtime=OBJ.stat().st_mtime, types=types, has_turbine=has_turbine,
              vec=(np.load(VEC) if VEC.exists() else None))
    return _M


def search(q, top=8, alpha=0.6):
    """混合检索: alpha·BM25 + (1−alpha)·向量。回 [(id, score, 命中来源)]。"""
    m = _load()
    ids, inv, idf, dl, avgdl = m['ids'], m['inv'], m['idf'], m['dl'], m['avgdl']
    sc = collections.defaultdict(float)
    for w in _grams(q):
        if w not in inv:
            continue
        iw = idf.get(w, 0)
        for i, tf in inv[w]:
            sc[i] += iw * tf * (K1 + 1) / (tf + K1 * (1 - B + B * dl[i] / max(avgdl, 1e-9)))
    bm = np.zeros(len(ids), dtype='float32')
    for i, v in sc.items():
        bm[i] = v
    if bm.max() > 0:
        bm = bm / bm.max()
    src = 'BM25'
    tot = bm
    if m.get('vec') is not None:
        try:
            qv = _embed_all([q])[0]
            cos = m['vec'] @ qv
            cos = (cos - cos.min()) / (cos.max() - cos.min() + 1e-9)
            tot = alpha * bm + (1 - alpha) * cos
            src = 'BM25+向量'
        except Exception:
            pass
    # 问"哪些机组"时抬升带台号的对象 (2026-08-28): 问句字面"上升"会精确命中
    # alarmcode/8459「高速轴振动快速上升」这类码 —— 命中不算错, 但提问要的是机组不是码。
    # 带 props.turbine 的对象才回答得了"哪些机组"。
    want_t = bool(re.search(r'哪些机组|哪几台|哪台|逐台|机组.*(有|在|需)', q))
    # 问"结果如何"时抬升结论性实证 (2026-08-28 实测): 问"换了齿轮箱后振动降下来了吗",
    # "齿轮箱"字面命中 48 条 verdict 且 Verdict 先验高, 把闭环证据全挤出前排 —
    # 模型于是拿"某台齿轮箱预警"回答"换件效果", 得出与事实相反的结论。
    # 这类问法要的是 evidence/loop(闭环)与 evidence/trend(在升), 不是状态判级。
    want_outcome = bool(re.search(r'降下来|降下去|恢复|好转|闭环|换件后|修好|效果|有没有用|在升|恶化', q))

    # 类型先验加权 — 只调排序不改召回
    pri = np.ones(len(ids), dtype='float32')
    lifted = {t for pat, t in ASK_TYPE if re.search(pat, q)}
    has_t = m.get('has_turbine') or []
    for i, t in enumerate(m.get('types') or []):
        w = TYPE_PRIOR.get(t, 1.0)
        w = 1.0 if (t in lifted and w < 1.0) else w
        if want_t and i < len(has_t) and has_t[i]:
            w *= 1.30
        if want_outcome and str(ids[i]).startswith(('evidence/loop/', 'evidence/trend/')):
            w *= 1.60
        pri[i] = w
    tot = tot * pri
    order = np.argsort(-tot)[:top]
    return [(ids[i], round(float(tot[i]), 4), src) for i in order if tot[i] > 0]