diff --git a/apps/server/services/detection_service.py b/apps/server/services/detection_service.py index 4789081..cdfe91b 100644 --- a/apps/server/services/detection_service.py +++ b/apps/server/services/detection_service.py @@ -12,6 +12,15 @@ import torch from .loitering_service import get_loitering_service from .adapters import DetectionAdapter from .event import AlertRuleEngine, EventAggregator, EventDecisionEngine +from .event.frame_accumulator import MultiFrameAccumulator +from .event.llm_trigger import LLMTrigger +from .llm_analysis_service import ( + LLMAnalysisService, + LLMAnalysisRequest, + create_llm_service_from_settings, +) +from .llm_cost_tracker import LLMCostTracker, get_global_tracker +from .result_fusion import ResultFusion from core.settings import get_settings from models.event_schemas import DetectionSource @@ -41,6 +50,41 @@ class DetectionService: dedup_window_seconds=settings.event_engine.dedup_window_seconds, max_active_events=settings.event_engine.max_active_events, ) + + # LLM 二次判断管道 (MVP-3 / D26-D30) + self._llm_enabled = settings.llm.enabled + if self._llm_enabled: + self._frame_accumulator = MultiFrameAccumulator( + window_seconds=settings.llm_trigger.window_seconds, + ) + self._llm_trigger = LLMTrigger( + accumulator=self._frame_accumulator, + min_consecutive_hits=settings.llm_trigger.min_consecutive_hits, + min_avg_confidence=settings.llm_trigger.min_avg_confidence, + cooldown_seconds=settings.llm_trigger.cooldown_seconds, + severity_bypass=settings.llm_trigger.severity_bypass, + ) + self._llm_service = create_llm_service_from_settings(settings) + self._cost_tracker = get_global_tracker() + self._result_fusion = ResultFusion( + strategy=settings.fusion.strategy, + yolo_weight=settings.fusion.yolo_weight, + llm_weight=settings.fusion.llm_weight, + suppress_on_llm_negative=settings.fusion.suppress_on_llm_negative, + fallback_to_yolo=settings.fusion.fallback_to_yolo, + ) + logger.info( + "LLM 管道已初始化: provider=%s model=%s strategy=%s", + settings.llm.provider, + settings.llm.model, + settings.fusion.strategy, + ) + else: + self._frame_accumulator = None + self._llm_trigger = None + self._llm_service = None + self._cost_tracker = None + self._result_fusion = None async def detect_image( self, @@ -94,6 +138,7 @@ class DetectionService: } } result_data = self._apply_event_pipeline(result_data, model_id) + result_data = await self._apply_llm_pipeline(result_data) return result_data results = model(image, conf=confidence, iou=iou, verbose=False) @@ -185,6 +230,8 @@ class DetectionService: # 事件管道 (MVP-1): 决策 → 规则 → 聚合 result_data = self._apply_event_pipeline(result_data, model_id) + # LLM 二次判断管道 (MVP-3) + result_data = await self._apply_llm_pipeline(result_data, frame=image) return result_data except Exception as e: @@ -268,6 +315,7 @@ class DetectionService: } } result_data = self._apply_event_pipeline(result_data, model_id) + result_data = await self._apply_llm_pipeline(result_data, frame=frame) if draw: frame = self.draw_detections(frame, detections, fps) @@ -388,6 +436,8 @@ class DetectionService: # 事件管道 (MVP-1): 决策 → 规则 → 聚合 result_data = self._apply_event_pipeline(result_data, model_id=model_id) + # LLM 二次判断管道 (MVP-3) + result_data = await self._apply_llm_pipeline(result_data, frame=frame) return frame, result_data except Exception as e: @@ -502,6 +552,8 @@ class DetectionService: model_id='fire_composite', source=DetectionSource.COMPOSITE, ) + # LLM 二次判断管道 (MVP-3) + result_data = await self._apply_llm_pipeline(result_data, frame=image) return result_data except Exception as e: @@ -579,13 +631,15 @@ class DetectionService: model_id: Optional[str] = None, source_id: Optional[str] = None, source: DetectionSource = DetectionSource.YOLO, + frame: Optional[np.ndarray] = None, ) -> Dict: - """对检测结果执行 决策 → 规则 → 聚合 三段管道。 + """对检测结果执行 决策 → 规则 → 聚合 → LLM 二次判断 → 融合 管道。 - 在 ``result_data`` 中追加两个字段: + 在 ``result_data`` 中追加以下字段: - ``candidate_events``: List[dict] 决策引擎产出的候选事件 - ``alert_events``: List[dict] 规则命中后经聚合的预警事件 + - ``llm_results``: List[dict] LLM 分析结果 (仅 LLM 启用时) """ if not result_data.get('success') or not result_data.get('detections'): @@ -615,6 +669,185 @@ class DetectionService: return result_data + async def _apply_llm_pipeline( + self, + result_data: Dict, + frame: Optional[np.ndarray] = None, + ) -> Dict: + """LLM 二次判断管道 (MVP-3 / D35)。 + + 在已有 ``alert_events`` 的基础上: + 1. 将候选事件喂给 LLMTrigger 判断是否需要 LLM 复审 + 2. 对触发的决策调用 LLMAnalysisService + 3. 用 ResultFusion 融合 YOLO + LLM 结果 + 4. 用 LLMCostTracker 记录成本 + + 追加字段: + - ``llm_results``: List[dict] LLM 分析结果列表 + """ + + if not self._llm_enabled: + result_data['llm_results'] = [] + return result_data + + alert_events = result_data.get('alert_events', []) + if not alert_events: + result_data['llm_results'] = [] + return result_data + + logger.info( + "[LLM管道] 开始评估 | alert_events=%d | 有frame=%s", + len(alert_events), + frame is not None and frame.size > 0, + ) + + try: + # 1. 从 alert_events 重建 CandidateEvent 列表供 LLMTrigger 评估 + from models.event_schemas import CandidateEvent, AlertEvent, EventType, SeverityLevel, UnifiedDetection + candidates_for_llm = [] + for evt_dict in alert_events: + try: + # AlertEvent 用 detections(列表), CandidateEvent 用 detection(单个) → 取首个 + dets = evt_dict.get('detections', []) + det = dets[0] if dets else None + + candidate = CandidateEvent( + event_type=EventType(evt_dict['event_type']), + severity=SeverityLevel(evt_dict.get('severity', 'high')), + confidence=evt_dict.get('confidence', 0.5), + detection=det, + source_id=evt_dict.get('source_id'), + timestamp=evt_dict.get('timestamp', evt_dict.get('first_seen')), + triggered_rules=[evt_dict.get('rule_name')] if evt_dict.get('rule_name') else [], + ) + candidates_for_llm.append(candidate) + except Exception as exc: + logger.debug("[LLM管道] 重建候选事件失败: %s | 原始数据: %s", exc, {k: v for k, v in evt_dict.items() if k != 'detections'}) + continue + + if not candidates_for_llm: + logger.info("[LLM管道] 无有效候选事件,跳过") + result_data['llm_results'] = [] + return result_data + + # 2. LLMTrigger 评估 + trigger_decisions = self._llm_trigger.evaluate(candidates_for_llm) + + if not trigger_decisions: + logger.info("[LLM管道] 触发器评估: 未达到触发条件 (需连续%d帧或critical严重性)", self._llm_trigger.min_consecutive_hits) + result_data['llm_results'] = [] + return result_data + + logger.info( + "[LLM管道] 触发器命中 %d 个决策 | 详情: %s", + len(trigger_decisions), + [{d.candidate.event_type.value: d.reason} for d in trigger_decisions], + ) + + # 3. 逐个触发决策调用 LLM + 融合 + llm_results = [] + fused_alert_events = [] + for decision in trigger_decisions: + llm_result = None + + # 检查成本追踪器是否允许调用 + if self._cost_tracker and not self._cost_tracker.can_call(): + logger.info( + "[LLM管道] 调用被降级拦截: %s", + self._cost_tracker.reason_for_block(), + ) + elif self._llm_service: + # 构造 LLM 请求 + frames = [frame] if frame is not None and frame.size > 0 else [] + logger.info( + "[LLM管道] 调用 LLM 分析 | 事件=%s | 置信度=%.2f | 图片=%d张", + decision.candidate.event_type.value, + decision.candidate.confidence, + len(frames), + ) + request = LLMAnalysisRequest( + candidate=decision.candidate, + frames=frames, + ) + llm_result = await self._llm_service.analyze_with_fallback(request) + + if llm_result: + logger.info( + "[LLM管道] LLM 返回 | confirmed=%s | confidence=%.2f | reasoning=%.80s", + llm_result.confirmed, + llm_result.confidence, + (llm_result.reasoning or ''), + ) + else: + logger.warning("[LLM管道] LLM 返回空结果 (可能降级到 mock)") + + # 记录成本 + if self._cost_tracker and llm_result: + self._cost_tracker.record( + llm_result, + image_count=len(frames) or 1, + ) + + # 4. 找到对应的 AlertEvent 进行融合 + alert_dict = None + for evt_dict in alert_events: + if ( + evt_dict.get('event_type') == decision.candidate.event_type.value + and evt_dict.get('source_id') == decision.candidate.source_id + ): + alert_dict = evt_dict + break + + if alert_dict is not None and self._result_fusion: + try: + alert_event = AlertEvent(**alert_dict) + outcome = self._result_fusion.fuse(alert_event, llm_result) + if outcome.alert is not None and not outcome.suppressed: + fused_alert_events.append(outcome.alert.model_dump(mode='json')) + llm_results.append({ + 'trigger_reason': decision.reason, + 'llm_confirmed': llm_result.confirmed if llm_result else None, + 'llm_confidence': llm_result.confidence if llm_result else None, + 'llm_reasoning': llm_result.reasoning if llm_result else None, + 'fusion_strategy': outcome.strategy, + 'fusion_reason': outcome.reason, + 'final_confidence': outcome.final_confidence, + 'suppressed': outcome.suppressed, + }) + logger.info( + "[LLM管道] 融合完成 | 策略=%s | 最终置信度=%.2f | 抑制=%s | 原因=%s", + outcome.strategy, + outcome.final_confidence, + outcome.suppressed, + outcome.reason, + ) + except Exception as exc: + logger.error("LLM 融合失败: %s", exc) + llm_results.append({ + 'trigger_reason': decision.reason, + 'llm_confirmed': None, + 'error': str(exc), + }) + + # 用融合后的结果替换原始 alert_events + if fused_alert_events: + result_data['alert_events'] = fused_alert_events + + result_data['llm_results'] = llm_results + logger.info( + "[LLM管道] 完成 | 总决策=%d | LLM结果=%d | 融合后事件=%d", + len(trigger_decisions), + len(llm_results), + len(fused_alert_events), + ) + + except Exception as e: # noqa: BLE001 + logger.error("LLM 管道执行失败: %s", e) + result_data['llm_results'] = [] + result_data['llm_pipeline_error'] = str(e) + + return result_data + def draw_detections( self,