From 435fe3073f54260734aed18e72991eda5eaf90cd Mon Sep 17 00:00:00 2001 From: wuzhuorong <973204353@qq.com> Date: Tue, 16 Jun 2026 10:20:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(event):=20=E6=96=B0=E5=A2=9E=20LLM=20?= =?UTF-8?q?=E8=A7=A6=E5=8F=91=E5=86=B3=E7=AD=96=E5=99=A8=20LLMTrigger?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/server/services/event/llm_trigger.py | 244 ++++++++++++++++++++++ 1 file changed, 244 insertions(+) create mode 100644 apps/server/services/event/llm_trigger.py diff --git a/apps/server/services/event/llm_trigger.py b/apps/server/services/event/llm_trigger.py new file mode 100644 index 0000000..4e1ef72 --- /dev/null +++ b/apps/server/services/event/llm_trigger.py @@ -0,0 +1,244 @@ +"""LLM 触发决策器 (MVP-3 / D28) + +基于 ``MultiFrameAccumulator`` 的累积统计，决定哪些目标值得调用 LLM 二次判断。 + +触发策略 (任一满足即触发): + +1. ``severity_bypass``: 累积条目最大严重性命中白名单 (默认 critical) 立即触发， + 无需累积窗口 +2. ``连续命中帧数 >= min_consecutive_hits`` 且 ``avg_confidence >= min_avg_confidence`` + +冷却机制: + +- 触发后记录该目标最近一次的触发时间，``cooldown_seconds`` 内不再重复触发， + 避免对同一可疑目标短时间多次调用 LLM 造成成本浪费。 + +可观测性: + +- ``stats`` 暴露 evaluated / triggered / cooled / bypassed 计数，便于监控 + +线程安全: 与 ``MultiFrameAccumulator`` 一致，单事件循环串行使用即可。 +""" + +from __future__ import annotations + +import logging +import time +from collections import OrderedDict +from dataclasses import dataclass +from typing import Dict, List, Optional, Tuple, TypeAlias + +from models.event_schemas import CandidateEvent, SeverityLevel + +from .frame_accumulator import AccumulationEntry, MultiFrameAccumulator + +logger = logging.getLogger(__name__) + + +_TriggerKey: TypeAlias = Tuple[Optional[str], str, str] + + +# --------------------------------------------------------------------------- +# 触发结果 +# --------------------------------------------------------------------------- + + +@dataclass +class TriggerDecision: + """LLM 触发决策结果。""" + + candidate: CandidateEvent + entry: AccumulationEntry + reason: str + + def to_dict(self) -> Dict[str, object]: + return { + "event_type": self.candidate.event_type.value, + "source_id": self.candidate.source_id, + "confidence": round(self.candidate.confidence, 4), + "consecutive_hits": self.entry.consecutive_hits, + "avg_confidence": round(self.entry.avg_confidence, 4), + "max_severity": self.entry.max_severity.value, + "duration": round(self.entry.duration, 3), + "reason": self.reason, + } + + +# --------------------------------------------------------------------------- +# LLMTrigger +# --------------------------------------------------------------------------- + + +class LLMTrigger: + """LLM 触发器。 + + Args: + accumulator: 多帧累积分析器 (由调用方共享，便于状态一致) + min_consecutive_hits: 触发所需的最小连续命中帧数 + min_avg_confidence: 累积平均置信度下限 + cooldown_seconds: 同目标 LLM 冷却时间 (秒)，0 表示不冷却 + severity_bypass: 立即触发的严重性级别集合 + max_cooldown_entries: 冷却记录最大容量 (LRU 淘汰) + """ + + def __init__( + self, + accumulator: MultiFrameAccumulator, + min_consecutive_hits: int = 3, + min_avg_confidence: float = 0.55, + cooldown_seconds: float = 20.0, + severity_bypass: Optional[List[str]] = None, + max_cooldown_entries: int = 5000, + ) -> None: + if min_consecutive_hits < 1: + raise ValueError("min_consecutive_hits 必须 >= 1") + if not 0.0 <= min_avg_confidence <= 1.0: + raise ValueError("min_avg_confidence 必须在 [0, 1]") + if cooldown_seconds < 0: + raise ValueError("cooldown_seconds 必须 >= 0") + if max_cooldown_entries < 1: + raise ValueError("max_cooldown_entries 必须 >= 1") + + self.accumulator = accumulator + self.min_consecutive_hits = min_consecutive_hits + self.min_avg_confidence = min_avg_confidence + self.cooldown_seconds = cooldown_seconds + self.severity_bypass = { + SeverityLevel(s) for s in (severity_bypass or []) + } if severity_bypass else set() + self.max_cooldown_entries = max_cooldown_entries + + self._cooldowns: "OrderedDict[_TriggerKey, float]" = OrderedDict() + + # 统计 + self._evaluated = 0 + self._triggered = 0 + self._cooled = 0 + self._bypassed = 0 + + # ------------------------------------------------------------------ + # 主入口 + # ------------------------------------------------------------------ + + def evaluate( + self, + candidates: List[CandidateEvent], + now: Optional[float] = None, + ) -> List[TriggerDecision]: + """评估候选事件，返回需要触发 LLM 复审的决策列表。 + + Args: + candidates: 当前批次候选事件 (通常来自规则引擎之前的决策结果) + now: 当前时间戳 (供测试注入) + """ + + if now is None: + now = time.time() + + # 1. 先把候选事件喂给累积器 (统一时间戳，确保统计与触发判定基于相同 now) + self.accumulator.accumulate(candidates, now=now) + + decisions: List[TriggerDecision] = [] + for candidate in candidates: + self._evaluated += 1 + entry = self.accumulator.get(candidate) + if entry is None: + continue + + decision = self._make_decision(candidate, entry, now) + if decision is not None: + decisions.append(decision) + + # 维护冷却表容量 + self._evict_cooldowns(now) + return decisions + + # ------------------------------------------------------------------ + # 内部 + # ------------------------------------------------------------------ + + def _make_decision( + self, + candidate: CandidateEvent, + entry: AccumulationEntry, + now: float, + ) -> Optional[TriggerDecision]: + cooldown_key: _TriggerKey = entry.key + + # 冷却检查 + last_fire = self._cooldowns.get(cooldown_key) + if last_fire is not None and (now - last_fire) < self.cooldown_seconds: + self._cooled += 1 + return None + + # 严重性快速通道 + if entry.max_severity in self.severity_bypass: + self._bypassed += 1 + self._mark_cooldown(cooldown_key, now) + return TriggerDecision( + candidate=candidate, + entry=entry, + reason=f"severity_bypass:{entry.max_severity.value}", + ) + + # 多帧累积阈值 + if ( + entry.consecutive_hits >= self.min_consecutive_hits + and entry.avg_confidence >= self.min_avg_confidence + ): + self._triggered += 1 + self._mark_cooldown(cooldown_key, now) + return TriggerDecision( + candidate=candidate, + entry=entry, + reason=( + f"hits={entry.consecutive_hits}>={self.min_consecutive_hits}," + f"avg_conf={entry.avg_confidence:.3f}>=" + f"{self.min_avg_confidence:.2f}" + ), + ) + + return None + + def _mark_cooldown(self, key: _TriggerKey, now: float) -> None: + self._cooldowns[key] = now + self._cooldowns.move_to_end(key) + while len(self._cooldowns) > self.max_cooldown_entries: + self._cooldowns.popitem(last=False) + + def _evict_cooldowns(self, now: float) -> None: + if self.cooldown_seconds <= 0: + self._cooldowns.clear() + return + expired = [ + key + for key, ts in self._cooldowns.items() + if (now - ts) >= self.cooldown_seconds + ] + for key in expired: + self._cooldowns.pop(key, None) + + # ------------------------------------------------------------------ + # 自省 + # ------------------------------------------------------------------ + + @property + def stats(self) -> Dict[str, int]: + return { + "evaluated": self._evaluated, + "triggered": self._triggered, + "cooled": self._cooled, + "bypassed": self._bypassed, + "active_cooldowns": len(self._cooldowns), + "active_accumulations": self.accumulator.active_count, + } + + def reset(self) -> None: + self._cooldowns.clear() + self._evaluated = 0 + self._triggered = 0 + self._cooled = 0 + self._bypassed = 0 + + +__all__ = ["LLMTrigger", "TriggerDecision"]