jc/jc-video-recognize
11
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(server): DetectionService 集成 LLM 二次判断完整管道

Browse Source
This commit is contained in:
wuzhuorong
2026-06-16 10:59:16 +08:00
parent 7e59455361
commit dce7014774
1 changed files with 235 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/services/detection_service.py
+235 -2
View File
@@ -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,