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

# -*- coding: utf-8 -*-
"""assistant.py — 柱3「智能助手」核心: grounded 运维问答 (对结构化 case/finding 层, 非黑箱 RAG).

★差异化 (vs 中能拾贝数字员工/通用 LLM-RAG): 本助手**只从结构化 cases.json/findings.json 答**,
每句带 case_id/finding 出处, 数据无支撑 → INSUFFICIENT (拒答不幻觉)。= 运维Agent转化评估的"诚实基础设施"。
黑箱 RAG(scripts/rag_chat.py) 易幻觉诊断; 本核心确定性、可审计、知道自己不知道。

纯函数 (answer): (query, cases, findings) → {answer, grounded_on[case_id/finding], verdict, intent}。
无 LLM/无网络/无时钟 — 确定性意图路由 (LLM 只该做语言皮层叙事, 判断在结构化层; edge 三连证)。
"""
import re

from src.sop.cases import system_of
from src.sop.schemas import norm_verdict
# 2026-07-09 合并: verdict rank + system 家族匹配 复用共享核 (与 fleet_query 单源)
from src.sop.agent_query_common import rank, system_family_match

# 中文查询词 → 规范 system 前缀 (匹配走 system_of + 家族前缀; 温度=跨传动链热系统)。
# ⚠ specific-first 排序: 长词先 (loop 首命中即返回) — '主轴承'先于'主轴' / '变桨'先于'桨距', 防截胡。
# 无数据的系统(叶片/塔筒/振动...)也列 → 命门差异化: 识别为系统查询后诚实报 INSUFFICIENT(不幻觉),
# 而非落"意图不明"(把明确部件问题当看不懂)。
_SYS_KW = {
    '齿轮箱': ['gearbox'], '齿箱': ['gearbox'],
    '定子': ['gen_stator'], '发电机': ['gen'],
    '主轴承': ['main_brg'], '主轴': ['main_shaft', 'main_brg'],
    '偏航': ['yaw'], '变桨': ['pitch'], '桨距': ['pitch'],
    '叶片': ['blade'], '传动链': ['driveline'], '联轴器': ['coupling'],
    '塔筒': ['tower'], '塔架': ['tower'],
    '变流器': ['converter'], '变频器': ['converter'],
    '安全链': ['safety_chain'], '振动': ['vibration'],
    '温度': ['gearbox', 'gen', 'main_brg'], '限电': ['curtail'],
}


def _turbine_ids(query, cases):
    """从 query 抽台号 (匹配 cases 里真实存在的 turbine, 支持 Dnn/Wnn/含前缀)。"""
    toks = set(re.findall(r'[A-Za-z]?\d{1,3}', query))
    hits = []
    for c in cases or []:
        t = str(c.get('turbine', ''))
        tail = re.sub(r'^\D*', '', t)  # '盂县 D22' → 'D22'? 实际 tail=22; 用后缀数字/字母数字匹配
        for tok in toks:
            if tok and (tok in t or tok.lstrip('DW') == re.sub(r'\D', '', t)):
                hits.append(t)
                break
    return sorted(set(hits))


def answer(query, cases, findings=None, farm='?'):
    """grounded 问答核心。query 中文自然语言; cases/findings = 结构化层。返回带出处的答案 dict。"""
    q = str(query or '')
    cases = cases or []
    findings = findings or []
    if not cases and not findings:
        return {'intent': 'no_data', 'verdict': 'INSUFFICIENT',
                'answer': f'{farm} 无 cases/findings 数据 → 判不了 (须先跑 onboard→分析→cases)。',
                'grounded_on': []}

    # ── 意图1: 具体台号 → 该台档案 ──
    tids = _turbine_ids(q, cases)
    if tids and not re.search(r'哪些|所有|全部|列', q):
        lines, gon = [], []
        for t in tids:
            tcs = [c for c in cases if str(c.get('turbine')) == t]
            if not tcs:
                lines.append(f'{t}: 无 case (该台未被任何判据 flag = 相对健康或无数据)')
                continue
            for c in tcs:
                ev = (c.get('evidence') or [{}])[0]
                lines.append(f"{t} · {c.get('system')}: 状态={c.get('state')} verdict={c.get('verdict')} "
                             f"(判据 {ev.get('discriminator_id')} 值={ev.get('value')}); case={c.get('case_id')}")
                # 深挖层: Phase-2 机制归因 + 独立审 (人工分析层, 比自动 fleet-z 更深, 不埋没)
                an = c.get('analysis')
                if isinstance(an, dict):
                    lines.append(f"  ▸ 深挖[{an.get('verdict', '?')}]: {an.get('mechanism', '')}"
                                 + (f" | 独立审: {an.get('review')}" if an.get('review') else '')
                                 + (f" | 现场需求: {an.get('field_needs')}" if an.get('field_needs') else '')
                                 + (f" (据 {an.get('ref')})" if an.get('ref') else ''))
                gon.append(c.get('case_id'))
        return {'intent': 'turbine_detail', 'verdict': '参考' if gon else 'INSUFFICIENT',
                'answer': '\n'.join(lines), 'grounded_on': gon}

    # ── 意图2: 现场需求/怎么办 → findings.recommendation ──
    if re.search(r'怎么办|现场|需求|建议|措施|下一步|要做', q):
        recs, gon = [], []
        for f in findings:
            r = f.get('recommendation')
            if r and norm_verdict(f.get('verdict')) in ('候选', '准定论·预警', '定论'):
                recs.append(f"[{f.get('verdict')}] {f.get('title')}: {r}")
                gon.append(f.get('title'))
        if not recs:
            return {'intent': 'recommendation', 'verdict': 'INSUFFICIENT',
                    'answer': f'{farm} 无候选级以上 finding 带现场建议 (健康清场或仅参考级)。', 'grounded_on': []}
        return {'intent': 'recommendation', 'verdict': '参考',
                'answer': '现场建议 (仅候选级以上):\n' + '\n'.join(recs), 'grounded_on': gon}

    # ── 意图3: 系统查询 (齿轮箱/温度/偏航...) ──
    for kw, prefixes in _SYS_KW.items():
        if kw in q:
            hits, gon = [], []
            for c in cases:
                if any(system_family_match(c.get('system'), p) for p in prefixes):
                    hits.append(c)
            hits.sort(key=lambda c: -rank(c.get('verdict')))
            if not hits:
                # 查 findings 里该系统是否有但未 flag (走 system_of 单源)
                fhit = [f for f in findings if any(system_family_match(system_of(f), p) for p in prefixes)]
                if fhit:
                    vds = {norm_verdict(f.get('verdict')) for f in fhit}
                    return {'intent': 'system_query', 'verdict': '参考',
                            'answer': f'{farm} {kw}相关 finding {len(fhit)}条但无 case (最高 verdict={max(vds, key=rank)} 未过开案门/健康)。',
                            'grounded_on': [f.get('title') for f in fhit]}
                return {'intent': 'system_query', 'verdict': 'INSUFFICIENT',
                        'answer': f'{farm} 无 {kw} 相关 case/finding → 判不了 (未分析该系统或数据无此通道)。', 'grounded_on': []}
            lines = [f"{c.get('turbine')} · {c.get('system')}: {c.get('state')}/{c.get('verdict')} "
                     f"(z={(c.get('evidence') or [{}])[0].get('value')}); {c.get('case_id')}" for c in hits]
            gon = [c.get('case_id') for c in hits]
            return {'intent': 'system_query', 'verdict': '参考',
                    'answer': f'{farm} {kw} 相关 {len(hits)} 台:\n' + '\n'.join(lines), 'grounded_on': gon}

    # ── 意图4: 哪些台异常/嫌疑/候选 → 全场按 verdict 排 ──
    if re.search(r'哪些|异常|问题|候选|嫌疑|偏热|发热|健康', q):
        by_vd = {}
        for c in cases:
            by_vd.setdefault(norm_verdict(c.get('verdict')), []).append(c)
        order = sorted(by_vd, key=lambda v: -rank(v))
        lines, gon = [], []
        for vd in order:
            cs = by_vd[vd]
            tag = ', '.join(f"{c.get('turbine')}({c.get('system')})" for c in cs)
            lines.append(f"【{vd}】{len(cs)}台: {tag}")
            gon += [c.get('case_id') for c in cs]
        return {'intent': 'list_anomalies', 'verdict': '参考',
                'answer': f'{farm} 共 {len(cases)} 个 case, 按证据强度:\n' + '\n'.join(lines), 'grounded_on': gon}

    # ── 兜底: 意图不明 → 不猜 ──
    return {'intent': 'unknown', 'verdict': 'INSUFFICIENT',
            'answer': f'问题意图不明 (支持: 某台Dnn详情 / 某系统 / 哪些台异常 / 现场需求)。'
                      f'{farm} 现有 {len(cases)} case + {len(findings)} finding 可查。', 'grounded_on': []}