# -*- 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': []}