guanlan-src/app_ontology/app_ontology_guanlan/sop/assistant_llm.py

# -*- coding: utf-8 -*-
"""assistant_llm.py — 柱3 助手 LLM 语言皮层 (叙事层; 判断留结构化, 反幻觉接地闸).

★edge三连证铁律: **LLM 只做语言皮层(叙事/措辞), 判断在结构化层(assistant.answer)**。
本层 = 把 assistant.answer 的 grounded dict 用自然语言复述给现场工程师, 但:
- **反幻觉接地闸**: LLM 叙事若引入结构化答案里没有的**台号/case_id**(高危捏造) → 拒绝, 回退结构化。
- **优雅回退**: 无 llm_fn / LLM 失败 / 接地闸不过 → 返回确定性结构化答案 (LLM 纯附加, 不可破)。
- **INSUFFICIENT 透传**: 结构化判 INSUFFICIENT → 叙事必说"判不了", LLM 不许"圆"成有结论。

= 即便加 LLM, "可审计 + 知道自己不知道" thesis 不破 (结构化答案永远是权威版, 随 nl 一起返回可核)。
llm_fn(prompt)->str|None 可插拔 (ollama_llm_fn 尽力+失败None; 也可接任意 provider)。
"""

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 re

from src.sop.assistant import answer as _structured_answer

# 台号 token (高危捏造): A/B/D/W/F + 1-3位数字, 或 'N号风机'
_TID_RE = re.compile(r'[ABDWF]\d{1,3}|\d{1,3}号风机')


def _tid_tokens(text):
    return set(_TID_RE.findall(str(text)))


def _grounding_ok(nl, structured):
    """接地闸: nl 叙事的台号 token 必 ⊆ 结构化答案(answer+grounded_on)的台号 token。
    引入新台号 = 捏造 → False。数字/措辞不检(低危, LLM 复述会动)。"""
    allowed = _tid_tokens(structured.get('answer', '')) | _tid_tokens(' '.join(map(str, structured.get('grounded_on', []))))
    return _tid_tokens(nl) <= allowed


def _prompt(structured, query, farm):
    return (f"你是风电运维助手, 回答现场工程师。用户问: {query}\n\n"
            f"下面是结构化诊断结果 = **唯一事实来源, 不许增删任何事实/台号/数字/结论**:\n"
            f"  判级(verdict): {structured.get('verdict')}\n"
            f"  内容: {structured.get('answer')}\n"
            f"  出处(case_id/finding): {structured.get('grounded_on')}\n\n"
            f"要求: ①只用上面的事实复述, 不编造任何台号/数字/机制 ②带出处 "
            f"③若 verdict=INSUFFICIENT 必须明说'判不了/数据不支持', 不许圆成有结论 "
            f"④简洁, 给现场能用的话。直接输出复述, 不要前言。")


def narrate(structured, query, llm_fn=None, farm='?'):
    """结构化答案 → 自然语言叙事 (LLM 皮层 + 接地闸 + 回退)。
    返回 {nl, source, grounding_ok, verdict, grounded_on, structured}。
    source ∈ {structured(无LLM/失败), llm(过闸), llm_blocked(接地闸拦截,回退结构化)}。"""
    base = {'verdict': structured.get('verdict'), 'grounded_on': structured.get('grounded_on', []),
            'structured': structured.get('answer', '')}
    if llm_fn is None:
        return {**base, 'nl': structured.get('answer', ''), 'source': 'structured', 'grounding_ok': True}
    try:
        nl = llm_fn(_prompt(structured, query, farm))
    except Exception:
        nl = None
    if not nl or not str(nl).strip():
        return {**base, 'nl': structured.get('answer', ''), 'source': 'structured', 'grounding_ok': True}
    if not _grounding_ok(str(nl), structured):
        # 接地闸拦截: LLM 引入了结构化里没有的台号 → 判定捏造, 回退权威结构化版
        return {**base, 'nl': structured.get('answer', ''), 'source': 'llm_blocked', 'grounding_ok': False}
    return {**base, 'nl': str(nl).strip(), 'source': 'llm', 'grounding_ok': True}


def answer_nl(query, cases, findings=None, farm='?', llm_fn=None):
    """柱3 一站: 结构化判断(assistant.answer) + LLM 叙事皮层。判断永远来自结构化层。"""
    structured = _structured_answer(query, cases, findings, farm)
    return {**narrate(structured, query, llm_fn, farm), 'intent': structured.get('intent')}


def ollama_llm_fn(model='qwen2.5:7b', url='http://localhost:11434/api/generate', timeout=30):
    """Ollama 适配 (尽力; 未运行/超时/任何错 → 返回 None 触发回退)。返回 llm_fn 闭包。"""
    import json
    import urllib.request

    def _fn(prompt):
        try:
            req = urllib.request.Request(
                url, data=json.dumps({'model': model, 'prompt': prompt, 'stream': False}).encode('utf-8'),
                headers={'Content-Type': 'application/json'})
            with urllib.request.urlopen(req, timeout=timeout) as r:
                return json.loads(r.read().decode('utf-8')).get('response')
        except Exception:
            return None
    return _fn


# ---------------------------------------------------------------------------
if __name__ == '__main__':
    # 控制台可能是 GBK(中文 Windows 代码页 936): 正文里的 ✔ ✗ ✅ ⚠ 这类字符编不出来会抛
    # UnicodeEncodeError, 脚本干成了事却以退出码 1 结束(同类坑见 src/console.py)。降级为 '?' 而不是崩;
    # 不用 import 是为了兼顾 python -m 与直接当脚本跑两种启动方式。
    import sys as _sys
    for _s in (_sys.stdout, _sys.stderr):
        try: _s.reconfigure(errors='replace')
        except Exception: pass
    # CLI: python -m src.sop.assistant_llm <farm> "问题"  (有Ollama则LLM叙事, 否则回退结构化)
    import argparse
    import json
    from pathlib import Path
    ROOT = _install_root(__file__)
    ap = argparse.ArgumentParser()
    ap.add_argument('farm')
    ap.add_argument('query')
    ap.add_argument('--no-llm', action='store_true', help='强制结构化(不调LLM)')
    a = ap.parse_args()
    sopd = P.sop(a.farm)
    cases = json.loads((sopd / 'cases.json').read_text(encoding='utf-8')) if (sopd / 'cases.json').exists() else []
    findings = (json.loads((sopd / 'findings.json').read_text(encoding='utf-8')).get('findings')
                if (sopd / 'findings.json').exists() else [])
    llm = None if a.no_llm else ollama_llm_fn()
    r = answer_nl(a.query, cases, findings, farm=a.farm, llm_fn=llm)
    print(f"[{r['intent']}/{r['verdict']}/source={r['source']}]")
    print(r['nl'])
    print(f"\n出处: {r['grounded_on'][:5]}")
    if r['source'] == 'llm_blocked':
        print("⚠ LLM 叙事引入结构化外台号(疑捏造) → 已拦截回退结构化权威版")