jc/jc-video-recognize

Fork 0

Files

wwh 6819e57d79 docs: 加了两个计划书，后续做实现

2026-06-02 20:28:17 +08:00

74 KiB

Raw Permalink Blame History

事件判断算法架构设计文档

文档信息

项目名称: jc-video-recognize 视频模型检测平台
模块名称: 事件判断算法引擎 (Event Judgment Algorithm Engine)
制定日期: 2026-06-02
文档版本: v1.0

1. 架构概述

1.1 设计目标

事件判断算法引擎的核心目标是：

准确性: 减少误报，确保真实事件被捕获
实时性: 低延迟的事件判断和预警
可配置性: 支持灵活的规则配置和场景适配
可扩展性: 易于添加新的事件类型和判断逻辑

1.2 整体架构图

┌─────────────────────────────────────────────────────────────────────────────┐
│                           事件判断算法引擎架构                                │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                        输入层 (Input Layer)                          │   │
│  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐ │   │
│  │  │ YOLO检测    │  │ 行为检测    │  │ RTSP流      │  │ 外部系统    │ │   │
│  │  │ 结果        │  │ 结果        │  │ 帧数据      │  │ 输入        │ │   │
│  │  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘ │   │
│  │         └─────────────────┴─────────────────┴─────────────────┘      │   │
│  └────────────────────────────────────┬────────────────────────────────┘   │
│                                       │                                      │
│                                       ▼                                      │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      决策层 (Decision Layer)                         │   │
│  │                                                                      │   │
│  │   ┌─────────────────┐      ┌─────────────────┐      ┌─────────────┐ │   │
│  │   │  事件决策引擎    │─────▶│  预警规则引擎    │─────▶│ 事件聚合器  │ │   │
│  │   │                 │      │                 │      │             │ │   │
│  │   │ • 置信度评估    │      │ • 规则匹配      │      │ • 去重合并  │ │   │
│  │   │ • 场景识别      │      │ • 时间窗口      │      │ • 时间窗口  │ │   │
│  │   │ • 初步筛选      │      │ • 区域规则      │      │ • 事件关联  │ │   │
│  │   └─────────────────┘      └─────────────────┘      └──────┬──────┘ │   │
│  │                                                            │        │   │
│  └────────────────────────────────────────────────────────────┼────────┘   │
│                                                               │              │
│                                                               ▼              │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      AI增强层 (AI Enhancement)                       │   │
│  │                                                                      │   │
│  │   ┌─────────────────┐      ┌─────────────────┐      ┌─────────────┐ │   │
│  │   │  大模型触发器    │─────▶│  LLM视觉分析    │─────▶│ 结果融合    │ │   │
│  │   │                 │      │                 │      │             │ │   │
│  │   │ • 智能决策      │      │ • 图像理解      │      │ • 置信融合  │ │   │
│  │   │ • 优先级排序    │      │ • 场景分析      │      │ • 决策输出  │ │   │
│  │   │ • 并发控制      │      │ • 推理验证      │      │             │ │   │
│  │   └─────────────────┘      └─────────────────┘      └──────┬──────┘ │   │
│  │                                                            │        │   │
│  └────────────────────────────────────────────────────────────┼────────┘   │
│                                                               │              │
│                                                               ▼              │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      输出层 (Output Layer)                           │   │
│  │                                                                      │   │
│  │   ┌─────────────────┐      ┌─────────────────┐      ┌─────────────┐ │   │
│  │   │  严重性评估器    │─────▶│  事件格式化     │─────▶│ MQTT发布    │ │   │
│  │   │                 │      │                 │      │             │ │   │
│  │   │ • 风险等级      │      │ • 标准格式      │      │ • 消息发布  │ │   │
│  │   │ • 优先级计算    │      │ • 元数据填充    │      │ • QoS管理   │ │   │
│  │   │ • 响应建议      │      │ • 快照关联      │      │             │ │   │
│  │   └─────────────────┘      └─────────────────┘      └─────────────┘ │   │
│  │                                                                      │   │
│  └─────────────────────────────────────────────────────────────────────┘   │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

1.3 模块职责划分

模块名称	核心职责	输入	输出
事件决策引擎	初步判断检测结果是否构成事件	检测结果	候选事件列表
预警规则引擎	根据配置规则过滤事件	候选事件	符合规则的事件
事件聚合器	合并重复事件，去重	规则事件	聚合后的事件
大模型触发器	决策是否需要LLM二次判断	聚合事件	LLM任务队列
LLM视觉分析	大模型图像理解和验证	图像ROI	分析结果
结果融合	融合YOLO和LLM结果	多源结果	最终决策
严重性评估器	评估事件严重等级	最终事件	分级事件

2. 核心模块设计

2.1 事件决策引擎 (EventDecisionEngine)

2.1.1 功能说明

事件决策引擎是事件判断的第一道关卡，负责：

评估检测结果的置信度
识别检测场景（室内/室外/禁区等）
进行初步的事件筛选

2.1.2 类设计

class EventDecisionEngine:
    """
    事件决策引擎
    
    职责：
    1. 评估单个检测结果是否构成候选事件
    2. 根据场景和置信度进行初步筛选
    3. 为后续规则引擎提供标准化的事件描述
    """
    
    def __init__(self, config: DecisionEngineConfig):
        self.config = config
        self.scene_classifier = SceneClassifier()
        self.confidence_calibrator = ConfidenceCalibrator()
    
    def evaluate(
        self, 
        detection_result: DetectionResult,
        context: DetectionContext
    ) -> List[CandidateEvent]:
        """
        评估检测结果
        
        Args:
            detection_result: YOLO/行为检测的结果
            context: 检测上下文（时间、位置、摄像头信息等）
        
        Returns:
            候选事件列表（可能为空）
        """
        pass
    
    def _assess_confidence(
        self, 
        detection: Detection,
        model_type: str
    ) -> ConfidenceAssessment:
        """评估检测置信度的可靠性"""
        pass
    
    def _classify_scene(
        self, 
        frame: np.ndarray,
        camera_config: CameraConfig
    ) -> SceneType:
        """识别检测场景类型"""
        pass
    
    def _create_candidate_event(
        self,
        detection: Detection,
        scene: SceneType,
        confidence: ConfidenceAssessment
    ) -> CandidateEvent:
        """创建候选事件"""
        pass

2.1.3 决策流程

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  输入检测   │────▶│ 置信度评估  │────▶│ 场景识别    │────▶│ 事件创建    │
│  结果       │     │             │     │             │     │             │
└─────────────┘     └──────┬──────┘     └──────┬──────┘     └──────┬──────┘
                           │                   │                   │
                           ▼                   ▼                   ▼
                    ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
                    │ • 原始置信度 │     │ • 室内/室外  │     │ • 事件ID    │
                    │ • 模型类型  │     │ • 区域类型   │     │ • 事件类型  │
                    │ • 历史校准  │     │ • 时间特征   │     │ • 置信度    │
                    │ • 可信度    │     │ • 光照条件   │     │ • 场景信息  │
                    └─────────────┘     └─────────────┘     └─────────────┘

2.1.4 置信度评估策略

置信度区间	可信度等级	处理策略
0.0 - 0.5	低	直接丢弃
0.5 - 0.7	中	标记为待验证，优先走LLM
0.7 - 0.9	高	直接通过，可选LLM验证
0.9 - 1.0	极高	直接通过，跳过LLM

2.2 预警规则引擎 (AlertRuleEngine)

2.2.1 功能说明

预警规则引擎负责根据配置的规则判断事件是否需要触发预警：

时间规则（工作时间/非工作时间）
区域规则（禁区/普通区域）
频率规则（冷却时间、最大频次）
组合规则（多条件组合）

2.2.2 类设计

class AlertRuleEngine:
    """
    预警规则引擎
    
    职责：
    1. 管理预警规则集合
    2. 根据规则过滤候选事件
    3. 支持动态规则更新
    """
    
    def __init__(self, rule_repository: RuleRepository):
        self.rule_repository = rule_repository
        self.rule_cache = {}
    
    def check(
        self, 
        candidate_event: CandidateEvent
    ) -> RuleCheckResult:
        """
        检查候选事件是否满足预警规则
        
        Args:
            candidate_event: 候选事件
        
        Returns:
            规则检查结果
        """
        pass
    
    def get_applicable_rules(
        self, 
        event_type: str,
        scene: SceneType
    ) -> List[AlertRule]:
        """获取适用于当前事件的所有规则"""
        pass
    
    def evaluate_rule(
        self, 
        rule: AlertRule,
        event: CandidateEvent
    ) -> bool:
        """评估单个规则"""
        pass


@dataclass
class AlertRule:
    """预警规则定义"""
    rule_id: str
    rule_name: str
    event_type: str
    
    # 置信度规则
    min_confidence: float = 0.5
    max_confidence: float = 1.0
    
    # 时间规则
    time_windows: List[TimeWindow] = field(default_factory=list)
    timezone: str = "Asia/Shanghai"
    
    # 区域规则
    applicable_areas: List[str] = field(default_factory=list)
    excluded_areas: List[str] = field(default_factory=list)
    
    # 频率规则
    cooldown_seconds: int = 30
    max_alerts_per_hour: int = 10
    
    # 启用状态
    enabled: bool = True
    priority: int = 1  # 数字越小优先级越高


@dataclass
class TimeWindow:
    """时间窗口定义"""
    days_of_week: List[int]  # 0=周一, 6=周日
    start_time: str  # "HH:MM" 格式
    end_time: str    # "HH:MM" 格式

2.2.3 规则类型

class RuleType(Enum):
    """规则类型枚举"""
    CONFIDENCE = "confidence"      # 置信度规则
    TIME = "time"                  # 时间规则
    AREA = "area"                  # 区域规则
    FREQUENCY = "frequency"        # 频率规则
    COMBINATION = "combination"    # 组合规则
    SCENE = "scene"                # 场景规则

2.2.4 规则配置示例

# rules/fire_detection.yaml
rules:
  - rule_id: "fire_critical"
    rule_name: "火灾紧急预警"
    event_type: "fire"
    min_confidence: 0.8
    time_windows:
      - days_of_week: [0, 1, 2, 3, 4, 5, 6]
        start_time: "00:00"
        end_time: "23:59"
    applicable_areas: ["warehouse", "lab", "office"]
    cooldown_seconds: 10
    max_alerts_per_hour: 100
    enabled: true
    priority: 1

  - rule_id: "fire_normal"
    rule_name: "火灾普通预警"
    event_type: "fire"
    min_confidence: 0.5
    max_confidence: 0.8
    time_windows:
      - days_of_week: [0, 1, 2, 3, 4, 5, 6]
        start_time: "00:00"
        end_time: "23:59"
    applicable_areas: ["warehouse", "lab", "office"]
    cooldown_seconds: 30
    max_alerts_per_hour: 20
    enabled: true
    priority: 2

# rules/smoking_detection.yaml
rules:
  - rule_id: "smoking_indoor"
    rule_name: "室内吸烟预警"
    event_type: "smoking"
    min_confidence: 0.6
    scene_filter:
      location_type: "indoor"
      no_smoking_zone: true
    time_windows:
      - days_of_week: [0, 1, 2, 3, 4]
        start_time: "09:00"
        end_time: "18:00"
    cooldown_seconds: 60
    enabled: true
    priority: 1

2.3 事件聚合器 (EventAggregator)

2.3.1 功能说明

事件聚合器负责：

合并同一目标的连续检测事件
防止短时间内重复预警
关联相似事件

2.3.2 类设计

class EventAggregator:
    """
    事件聚合器
    
    职责：
    1. 去重：合并同一目标的连续事件
    2. 关联：识别相关事件
    3. 窗口：基于时间窗口管理事件
    """
    
    def __init__(self, config: AggregatorConfig):
        self.config = config
        self.active_events: Dict[str, AggregatedEvent] = {}
        self.event_history: deque = deque(maxlen=1000)
    
    def aggregate(
        self, 
        new_event: CandidateEvent
    ) -> Optional[AggregatedEvent]:
        """
        聚合新事件
        
        Args:
            new_event: 新检测到的事件
        
        Returns:
            聚合后的事件（如果是新事件）或None（如果是重复事件）
        """
        pass
    
    def _find_matching_event(
        self, 
        new_event: CandidateEvent
    ) -> Optional[AggregatedEvent]:
        """查找匹配的历史事件"""
        pass
    
    def _calculate_similarity(
        self, 
        event1: CandidateEvent,
        event2: AggregatedEvent
    ) -> float:
        """计算两个事件的相似度"""
        pass
    
    def _merge_events(
        self, 
        existing: AggregatedEvent,
        new_event: CandidateEvent
    ) -> AggregatedEvent:
        """合并两个事件"""
        pass
    
    def cleanup_expired_events(self):
        """清理过期的事件"""
        pass


@dataclass
class AggregatedEvent:
    """聚合事件"""
    event_id: str
    event_type: str
    
    # 时间信息
    first_seen: datetime
    last_seen: datetime
    
    # 位置信息
    camera_id: str
    location: Location
    bbox_history: List[BBox]
    
    # 置信度信息
    confidence_max: float
    confidence_min: float
    confidence_avg: float
    
    # 检测次数
    detection_count: int
    
    # 关联的检测ID
    detection_ids: List[str]
    
    # 状态
    status: EventStatus  # active, closed, confirmed


class EventStatus(Enum):
    """事件状态"""
    ACTIVE = "active"        # 活跃中（持续检测）
    PENDING = "pending"      # 待确认（等待LLM）
    CONFIRMED = "confirmed"  # 已确认
    CLOSED = "closed"        # 已关闭
    FALSE_POSITIVE = "false_positive"  # 误报

2.3.3 聚合策略

策略	说明	适用场景
时间窗口	同一目标在N秒内的事件合并	持续检测场景
空间邻近	位置距离小于阈值的事件合并	移动目标跟踪
类型相同	同一类型的事件合并	同类事件去重
置信度加权	保留置信度最高的检测结果	质量优先

2.3.4 相似度计算

def calculate_event_similarity(
    event1: CandidateEvent,
    event2: AggregatedEvent
) -> float:
    """
    计算事件相似度（0-1）
    
    考虑因素：
    1. 时间 proximity
    2. 空间 proximity
    3. 类型匹配
    4. 置信度差异
    """
    
    # 时间相似度
    time_diff = abs(
        (event1.timestamp - event2.last_seen).total_seconds()
    )
    time_sim = max(0, 1 - time_diff / TIME_WINDOW_SECONDS)
    
    # 空间相似度（IoU或中心点距离）
    spatial_sim = calculate_spatial_similarity(
        event1.bbox, 
        event2.bbox_history[-1]
    )
    
    # 类型匹配
    type_sim = 1.0 if event1.event_type == event2.event_type else 0.0
    
    # 加权综合
    similarity = (
        0.3 * time_sim +
        0.4 * spatial_sim +
        0.3 * type_sim
    )
    
    return similarity

2.4 大模型触发器 (LLMTrigger)

2.4.1 功能说明

大模型触发器负责智能决策是否需要调用大模型进行二次判断：

多帧累积判断: 不是单帧触发，而是基于多帧检测结果的综合判断
置信度趋势分析: 分析置信度变化趋势，避免瞬时波动误触发
基于场景的强制触发: 特定场景下强制触发LLM验证
并发控制和队列管理: 控制LLM调用频率和成本
成本优化策略: 缓存、去重、优先级队列

2.4.2 多帧累积判断机制

┌─────────────────────────────────────────────────────────────────────────────┐
│                         多帧累积判断流程                                      │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│   帧N-3        帧N-2        帧N-1        帧N(当前)                           │
│     │            │            │            │                                 │
│     ▼            ▼            ▼            ▼                                 │
│  ┌──────┐    ┌──────┐    ┌──────┐    ┌──────┐                              │
│  │检测  │    │检测  │    │检测  │    │检测  │                              │
│  │conf: │    │conf: │    │conf: │    │conf: │                              │
│  │ 0.55 │    │ 0.62 │    │ 0.58 │    │ 0.65 │                              │
│  └──┬───┘    └──┬───┘    └──┬───┘    └──┬───┘                              │
│     │            │            │            │                                 │
│     └────────────┴────────────┴────────────┘                                 │
│                    │                                                         │
│                    ▼                                                         │
│         ┌─────────────────────┐                                              │
│         │   多帧累积分析器     │                                              │
│         │                     │                                              │
│         │ 分析维度：           │                                              │
│         │ • 检测频次: 4/4帧   │  ← 连续检测次数                              │
│         │ • 平均置信度: 0.60  │  ← 累积置信度                                │
│         │ • 置信度趋势: 上升  │  ← 趋势判断                                  │
│         │ • 位置稳定性: 稳定  │  ← 空间一致性                                │
│         │ • 持续时间: 1.2秒   │  ← 时间累积                                  │
│         └──────────┬──────────┘                                              │
│                    │                                                         │
│         ┌──────────┴──────────┐                                              │
│         │   触发决策判断       │                                              │
│         │                     │                                              │
│         │ 触发条件检查：       │                                              │
│         │ ✓ 连续帧数 >= 3     │  ← 最小帧数阈值                              │
│         │ ✓ 平均置信度 >= 0.55│  ← 累积置信度阈值                            │
│         │ ✓ 位置变化 < 50px   │  ← 空间稳定性阈值                            │
│         │ ✓ 持续时间 >= 1s    │  ← 时间阈值                                  │
│         │                     │                                              │
│         │ 决策结果: 触发LLM   │                                              │
│         └─────────────────────┘                                              │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

2.4.3 类设计

class LLMTrigger:
    """
    大模型触发器
    
    职责：
    1. 基于多帧累积的触发决策
    2. 管理帧历史缓冲区
    3. 分析检测趋势和稳定性
    4. 管理LLM调用队列和成本
    """
    
    def __init__(self, config: LLMTriggerConfig):
        self.config = config
        
        # 多帧累积分析器
        self.frame_accumulator = MultiFrameAccumulator(
            buffer_size=config.frame_buffer_size,
            time_window=config.frame_time_window
        )
        
        # 队列和缓存
        self.task_queue = PriorityQueue()
        self.result_cache = LRUCache(maxsize=1000)
        
        # 成本控制
        self.concurrent_calls = 0
        self.cost_tracker = CostTracker()
        
        # 触发状态追踪
        self.trigger_state: Dict[str, TriggerState] = {}
    
    def should_trigger(
        self, 
        event: AggregatedEvent,
        frame_context: FrameContext
    ) -> TriggerDecision:
        """
        决策是否触发LLM分析
        
        基于多帧累积判断，不是单帧决策
        
        Args:
            event: 当前帧的聚合事件
            frame_context: 帧上下文信息
        
        Returns:
            触发决策
        """
        # 1. 将当前帧加入累积器
        self.frame_accumulator.add_frame(event, frame_context)
        
        # 2. 获取多帧分析结果
        frame_analysis = self.frame_accumulator.analyze(event.event_id)
        
        # 3. 基于多帧结果做触发决策
        return self._make_trigger_decision(event, frame_analysis)
    
    def _make_trigger_decision(
        self,
        event: AggregatedEvent,
        frame_analysis: FrameAnalysisResult
    ) -> TriggerDecision:
        """基于多帧分析结果做触发决策"""
        
        # 检查多帧触发条件
        trigger_conditions = self._evaluate_trigger_conditions(frame_analysis)
        
        if not trigger_conditions.all_met:
            return TriggerDecision(
                should_trigger=False,
                priority=0.0,
                reason=trigger_conditions.reason,
                strategy="multi_frame_accumulation",
                frame_stats=frame_analysis
            )
        
        # 计算优先级
        priority = self._calculate_priority_with_frames(event, frame_analysis)
        
        return TriggerDecision(
            should_trigger=True,
            priority=priority,
            reason=f"多帧累积满足条件: {trigger_conditions.summary}",
            strategy="multi_frame_accumulation",
            frame_stats=frame_analysis
        )
    
    def _evaluate_trigger_conditions(
        self,
        frame_analysis: FrameAnalysisResult
    ) -> TriggerConditions:
        """评估多帧触发条件"""
        conditions = TriggerConditions()
        
        # 条件1: 最小连续帧数
        if frame_analysis.consecutive_frames < self.config.min_consecutive_frames:
            conditions.add_failure(
                "consecutive_frames",
                f"连续帧数不足: {frame_analysis.consecutive_frames} < {self.config.min_consecutive_frames}"
            )
        
        # 条件2: 累积置信度阈值
        if frame_analysis.avg_confidence < self.config.min_accumulated_confidence:
            conditions.add_failure(
                "avg_confidence",
                f"平均置信度不足: {frame_analysis.avg_confidence:.2f} < {self.config.min_accumulated_confidence}"
            )
        
        # 条件3: 空间稳定性
        if frame_analysis.position_variance > self.config.max_position_variance:
            conditions.add_failure(
                "position_stability",
                f"位置变化过大: {frame_analysis.position_variance:.1f}px"
            )
        
        # 条件4: 最小持续时间
        if frame_analysis.duration_seconds < self.config.min_duration_seconds:
            conditions.add_failure(
                "duration",
                f"持续时间不足: {frame_analysis.duration_seconds:.1f}s < {self.config.min_duration_seconds}s"
            )
        
        # 条件5: 置信度趋势（可选）
        if self.config.require_confidence_trend:
            if frame_analysis.confidence_trend < 0:  # 下降趋势
                conditions.add_failure(
                    "trend",
                    "置信度呈下降趋势"
                )
        
        return conditions
    
    def _calculate_priority_with_frames(
        self,
        event: AggregatedEvent,
        frame_analysis: FrameAnalysisResult
    ) -> float:
        """基于多帧信息计算优先级"""
        base_priority = self.get_priority_score(event)
        
        # 多帧因子调整
        frame_factors = {
            # 连续帧数越多，优先级越高
            "consecutive_bonus": min(0.2, frame_analysis.consecutive_frames * 0.05),
            
            # 置信度趋势向上，增加优先级
            "trend_bonus": 0.1 if frame_analysis.confidence_trend > 0 else 0,
            
            # 位置越稳定，优先级越高
            "stability_bonus": 0.1 * (1 - min(1, frame_analysis.position_variance / 100))
        }
        
        adjusted_priority = base_priority + sum(frame_factors.values())
        return min(1.0, adjusted_priority)
    
    def get_priority_score(self, event: AggregatedEvent) -> float:
        """计算基础优先级分数"""
        pass
    
    def enqueue_analysis(self, event: AggregatedEvent) -> str:
        """将事件加入LLM分析队列"""
        pass
    
    def get_analysis_result(
        self, 
        task_id: str,
        timeout: float = 30.0
    ) -> Optional[LLMResult]:
        """获取LLM分析结果"""
        pass


class MultiFrameAccumulator:
    """
    多帧累积分析器
    
    管理多帧检测历史，分析检测趋势和稳定性
    """
    
    def __init__(self, buffer_size: int = 30, time_window: float = 5.0):
        self.buffer_size = buffer_size
        self.time_window = time_window
        
        # 帧历史缓冲区: {event_id: deque[FrameRecord]}
        self.frame_buffers: Dict[str, deque] = {}
    
    def add_frame(
        self, 
        event: AggregatedEvent,
        frame_context: FrameContext
    ):
        """添加一帧检测记录"""
        event_id = event.event_id
        
        if event_id not in self.frame_buffers:
            self.frame_buffers[event_id] = deque(maxlen=self.buffer_size)
        
        record = FrameRecord(
            timestamp=frame_context.timestamp,
            confidence=event.confidence_avg,
            bbox=event.bbox,
            frame_id=frame_context.frame_id
        )
        
        self.frame_buffers[event_id].append(record)
        
        # 清理过期缓冲区
        self._cleanup_expired_buffers()
    
    def analyze(self, event_id: str) -> FrameAnalysisResult:
        """分析多帧检测数据"""
        buffer = self.frame_buffers.get(event_id, deque())
        
        if len(buffer) < 2:
            return FrameAnalysisResult(
                consecutive_frames=len(buffer),
                avg_confidence=buffer[0].confidence if buffer else 0,
                position_variance=0,
                duration_seconds=0,
                confidence_trend=0
            )
        
        # 计算各项指标
        confidences = [r.confidence for r in buffer]
        positions = [self._bbox_center(r.bbox) for r in buffer]
        timestamps = [r.timestamp for r in buffer]
        
        return FrameAnalysisResult(
            consecutive_frames=len(buffer),
            avg_confidence=np.mean(confidences),
            max_confidence=np.max(confidences),
            min_confidence=np.min(confidences),
            position_variance=self._calculate_position_variance(positions),
            duration_seconds=(timestamps[-1] - timestamps[0]).total_seconds(),
            confidence_trend=self._calculate_trend(confidences),
            frames_per_second=len(buffer) / max(0.001, (timestamps[-1] - timestamps[0]).total_seconds())
        )
    
    def _calculate_position_variance(
        self, 
        positions: List[Tuple[int, int]]
    ) -> float:
        """计算位置方差（稳定性指标）"""
        if len(positions) < 2:
            return 0.0
        
        x_coords = [p[0] for p in positions]
        y_coords = [p[1] for p in positions]
        
        x_var = np.var(x_coords)
        y_var = np.var(y_coords)
        
        return np.sqrt(x_var + y_var)
    
    def _calculate_trend(self, values: List[float]) -> float:
        """计算趋势（线性回归斜率）"""
        if len(values) < 2:
            return 0.0
        
        x = np.arange(len(values))
        slope, _, _, _, _ = linregress(x, values)
        return slope
    
    def _bbox_center(self, bbox: Tuple[int, int, int, int]) -> Tuple[int, int]:
        """计算边界框中心点"""
        x1, y1, x2, y2 = bbox
        return ((x1 + x2) // 2, (y1 + y2) // 2)
    
    def _cleanup_expired_buffers(self):
        """清理过期的帧缓冲区"""
        current_time = datetime.now()
        expired_events = []
        
        for event_id, buffer in self.frame_buffers.items():
            if buffer:
                last_time = buffer[-1].timestamp
                if (current_time - last_time).total_seconds() > self.time_window * 2:
                    expired_events.append(event_id)
        
        for event_id in expired_events:
            del self.frame_buffers[event_id]


@dataclass
class FrameRecord:
    """单帧检测记录"""
    timestamp: datetime
    confidence: float
    bbox: Tuple[int, int, int, int]
    frame_id: int


@dataclass
class FrameAnalysisResult:
    """多帧分析结果"""
    consecutive_frames: int          # 连续检测帧数
    avg_confidence: float            # 平均置信度
    max_confidence: float            # 最大置信度
    min_confidence: float            # 最小置信度
    position_variance: float         # 位置方差（像素）
    duration_seconds: float          # 持续时间（秒）
    confidence_trend: float          # 置信度趋势（斜率）
    frames_per_second: float = 0.0   # 检测帧率


@dataclass
class TriggerDecision:
    """触发决策结果"""
    should_trigger: bool
    priority: float                  # 优先级分数
    reason: str                      # 决策原因
    strategy: str                    # 使用的策略
    frame_stats: Optional[FrameAnalysisResult] = None  # 多帧统计


@dataclass
class TriggerConditions:
    """触发条件评估结果"""
    failures: List[Dict] = field(default_factory=list)
    
    def add_failure(self, condition: str, reason: str):
        self.failures.append({"condition": condition, "reason": reason})
    
    @property
    def all_met(self) -> bool:
        return len(self.failures) == 0
    
    @property
    def reason(self) -> str:
        if self.all_met:
            return "所有条件满足"
        return "; ".join([f["reason"] for f in self.failures])
    
    @property
    def summary(self) -> str:
        return f"通过{len(self.failures)}项, 失败{len(self.failures)}项"


@dataclass
class LLMTriggerConfig:
    """LLM触发器配置"""
    
    # ========== 多帧累积配置 ==========
    # 帧缓冲区大小
    frame_buffer_size: int = 30
    
    # 帧时间窗口（秒）
    frame_time_window: float = 5.0
    
    # 最小连续帧数（触发LLM的最小帧数）
    min_consecutive_frames: int = 3
    
    # 最小累积置信度
    min_accumulated_confidence: float = 0.55
    
    # 最大位置方差（像素平方）
    max_position_variance: float = 2500  # 50px * 50px
    
    # 最小持续时间（秒）
    min_duration_seconds: float = 1.0
    
    # 是否要求置信度趋势向上
    require_confidence_trend: bool = False
    
    # ========== 置信度触发区间 ==========
    confidence_trigger_range: Tuple[float, float] = (0.5, 0.85)
    
    # ========== 强制触发场景 ==========
    force_trigger_scenes: List[str] = field(
        default_factory=lambda: ["indoor_no_smoking_area", "chemical_lab"]
    )
    
    # ========== 并发控制 ==========
    max_concurrent_calls: int = 5
    max_queue_size: int = 100
    
    # ========== 成本限制 ==========
    max_calls_per_minute: int = 30
    max_cost_per_hour: float = 10.0
    
    # ========== 缓存配置 ==========
    cache_ttl_seconds: int = 300
    similarity_threshold: float = 0.9

2.4.4 触发策略（含多帧判断）

class TriggerStrategy:
    """触发策略集合"""
    
    @staticmethod
    def multi_frame_strategy(
        event: AggregatedEvent,
        frame_analysis: FrameAnalysisResult,
        config: LLMTriggerConfig
    ) -> TriggerDecision:
        """
        多帧累积触发策略（默认策略）
        
        核心逻辑：
        1. 不是单帧触发，而是基于多帧的综合判断
        2. 要求检测在时间和空间上都是稳定的
        3. 避免瞬时误检导致的误触发
        """
        # 快速拒绝：如果帧数太少
        if frame_analysis.consecutive_frames < config.min_consecutive_frames:
            return TriggerDecision(
                should_trigger=False,
                priority=0.0,
                reason=f"连续帧数不足: {frame_analysis.consecutive_frames}/{config.min_consecutive_frames}",
                strategy="multi_frame",
                frame_stats=frame_analysis
            )
        
        # 快速拒绝：如果平均置信度太低
        if frame_analysis.avg_confidence < config.min_accumulated_confidence:
            return TriggerDecision(
                should_trigger=False,
                priority=0.0,
                reason=f"平均置信度不足: {frame_analysis.avg_confidence:.2f}/{config.min_accumulated_confidence}",
                strategy="multi_frame",
                frame_stats=frame_analysis
            )
        
        # 检查空间稳定性
        if frame_analysis.position_variance > config.max_position_variance:
            return TriggerDecision(
                should_trigger=False,
                priority=0.0,
                reason=f"位置不稳定: 方差={frame_analysis.position_variance:.1f}px²",
                strategy="multi_frame",
                frame_stats=frame_analysis
            )
        
        # 检查持续时间
        if frame_analysis.duration_seconds < config.min_duration_seconds:
            return TriggerDecision(
                should_trigger=False,
                priority=0.0,
                reason=f"持续时间不足: {frame_analysis.duration_seconds:.1f}s/{config.min_duration_seconds}s",
                strategy="multi_frame",
                frame_stats=frame_analysis
            )
        
        # 所有条件满足，计算优先级
        priority = TriggerStrategy._calculate_frame_based_priority(
            frame_analysis, config
        )
        
        return TriggerDecision(
            should_trigger=True,
            priority=priority,
            reason=f"多帧累积满足条件: {frame_analysis.consecutive_frames}帧, "
                   f"平均置信度{frame_analysis.avg_confidence:.2f}, "
                   f"持续{frame_analysis.duration_seconds:.1f}s",
            strategy="multi_frame",
            frame_stats=frame_analysis
        )
    
    @staticmethod
    def _calculate_frame_based_priority(
        frame_analysis: FrameAnalysisResult,
        config: LLMTriggerConfig
    ) -> float:
        """基于多帧信息计算优先级"""
        
        # 基础分：平均置信度
        base_score = frame_analysis.avg_confidence
        
        # 连续帧数奖励（帧数越多越可靠）
        frame_bonus = min(0.15, (frame_analysis.consecutive_frames - config.min_consecutive_frames) * 0.03)
        
        # 稳定性奖励（方差越小越稳定）
        stability_ratio = 1 - min(1, frame_analysis.position_variance / config.max_position_variance)
        stability_bonus = 0.1 * stability_ratio
        
        # 趋势奖励（置信度上升）
        trend_bonus = 0.05 if frame_analysis.confidence_trend > 0 else 0
        
        total_score = base_score + frame_bonus + stability_bonus + trend_bonus
        return min(1.0, total_score)
    
    @staticmethod
    def scene_based_strategy(
        event: AggregatedEvent,
        config: LLMTriggerConfig
    ) -> TriggerDecision:
        """
        基于场景的策略
        
        某些场景强制触发LLM验证（不依赖多帧累积）
        """
        scene = event.location.scene_type
        
        if scene in config.force_trigger_scenes:
            return TriggerDecision(
                should_trigger=True,
                priority=1.0,
                reason=f"强制触发场景: {scene}",
                strategy="scene_based"
            )
        
        return TriggerDecision(
            should_trigger=False,
            priority=0.0,
            reason="非强制触发场景",
            strategy="scene_based"
        )
    
    @staticmethod
    def single_frame_high_confidence_strategy(
        event: AggregatedEvent,
        config: LLMTriggerConfig
    ) -> TriggerDecision:
        """
        单帧高置信度策略
        
        对于极高置信度的检测，可以跳过多帧累积直接触发
        （用于紧急情况）
        """
        if event.confidence_avg > 0.95:
            return TriggerDecision(
                should_trigger=True,
                priority=0.9,
                reason=f"单帧极高置信度: {event.confidence_avg:.2f}",
                strategy="single_frame_high_conf"
            )
        
        return TriggerDecision(
            should_trigger=False,
            priority=0.0,
            reason="置信度未达到单帧触发阈值",
            strategy="single_frame_high_conf"
        )

2.4.5 多帧累积配置示例

# config/llm_trigger.yaml

llm_trigger:
  # ========== 多帧累积配置 ==========
  multi_frame:
    # 帧缓冲区大小（保存最近N帧）
    buffer_size: 30
    
    # 帧时间窗口（秒）- 只分析最近5秒内的帧
    time_window: 5.0
    
    # 最小连续帧数 - 至少检测到3帧才考虑触发
    min_consecutive_frames: 3
    
    # 最小累积置信度 - 多帧平均置信度阈值
    min_accumulated_confidence: 0.55
    
    # 最大位置方差（像素平方）- 位置变化不能超过50px
    max_position_variance: 2500
    
    # 最小持续时间（秒）- 检测至少持续1秒
    min_duration_seconds: 1.0
    
    # 是否要求置信度趋势向上
    require_confidence_trend: false
    
    # 帧率要求（fps）- 检测帧率不能太低
    min_detection_fps: 2.0
  
  # ========== 触发策略优先级 ==========
  strategy_priority:
    - "scene_based"           # 场景强制触发（最高优先级）
    - "single_frame_high_conf" # 单帧极高置信度
    - "multi_frame"           # 多帧累积（默认）
  
  # ========== 不同事件类型的多帧配置 ==========
  event_type_overrides:
    fire:
      # 火灾检测更敏感，减少帧数要求
      min_consecutive_frames: 2
      min_duration_seconds: 0.5
      min_accumulated_confidence: 0.50
    
    smoking:
      # 抽烟检测需要更稳定
      min_consecutive_frames: 5
      min_duration_seconds: 2.0
      min_accumulated_confidence: 0.60
      require_confidence_trend: true
    
    loitering:
      # 徘徊检测本来就需要时间
      min_consecutive_frames: 10
      min_duration_seconds: 5.0
      min_accumulated_confidence: 0.55

2.5 结果融合 (ResultFusion)

2.5.1 功能说明

结果融合模块负责将YOLO检测结果和LLM分析结果进行融合，得出最终决策：

置信度加权融合
冲突解决策略
决策解释生成

2.5.2 类设计

class ResultFusion:
    """
    结果融合器
    
    职责：
    1. 融合多源检测结果
    2. 解决结果冲突
    3. 生成决策解释
    """
    
    def __init__(self, config: FusionConfig):
        self.config = config
        self.fusion_strategies = {
            "weighted": WeightedFusionStrategy(),
            "conservative": ConservativeFusionStrategy(),
            "optimistic": OptimisticFusionStrategy()
        }
    
    def fuse(
        self,
        yolo_result: DetectionResult,
        llm_result: Optional[LLMResult],
        strategy: str = "weighted"
    ) -> FusionResult:
        """
        融合检测结果
        
        Args:
            yolo_result: YOLO检测结果
            llm_result: LLM分析结果（可能为None）
            strategy: 融合策略
        
        Returns:
            融合后的结果
        """
        pass
    
    def _calculate_fused_confidence(
        self,
        yolo_conf: float,
        llm_conf: float,
        weights: Tuple[float, float]
    ) -> float:
        """计算融合后的置信度"""
        pass
    
    def _generate_explanation(
        self,
        yolo_result: DetectionResult,
        llm_result: Optional[LLMResult],
        fused_result: FusionResult
    ) -> str:
        """生成决策解释"""
        pass


@dataclass
class FusionResult:
    """融合结果"""
    is_positive: bool           # 是否为真实事件
    confidence: float           # 融合后的置信度
    yolo_contribution: float    # YOLO贡献度
    llm_contribution: float     # LLM贡献度
    explanation: str            # 决策解释
    metadata: Dict              # 元数据

2.5.3 融合策略

策略	说明	适用场景
加权融合	按权重合并YOLO和LLM置信度	通用场景
保守策略	两者都高才算高置信度	高风险场景
乐观策略	任一高就算高置信度	宁可误报不可漏报
LLM优先	LLM结果覆盖YOLO	LLM可靠性高时

2.6 严重性评估器 (SeverityAssessor)

2.6.1 功能说明

严重性评估器负责评估事件的严重程度：

基于事件类型的基础等级
基于场景的调整
基于持续时间的升级
生成响应建议

2.6.2 类设计

class SeverityAssessor:
    """
    严重性评估器
    
    职责：
    1. 评估事件严重程度
    2. 计算响应优先级
    3. 生成响应建议
    """
    
    def __init__(self, config: SeverityConfig):
        self.config = config
        self.severity_matrix = self._load_severity_matrix()
    
    def assess(
        self,
        event: FusionResult,
        context: EventContext
    ) -> SeverityAssessment:
        """
        评估事件严重性
        
        Args:
            event: 融合后的事件结果
            context: 事件上下文
        
        Returns:
            严重性评估结果
        """
        pass
    
    def _calculate_base_severity(
        self,
        event_type: str
    ) -> SeverityLevel:
        """计算基础严重等级"""
        pass
    
    def _apply_scene_modifier(
        self,
        base_severity: SeverityLevel,
        scene: SceneType
    ) -> SeverityLevel:
        """应用场景调整"""
        pass
    
    def _apply_duration_modifier(
        self,
        severity: SeverityLevel,
        duration_seconds: float
    ) -> SeverityLevel:
        """应用持续时间调整"""
        pass
    
    def generate_response_recommendation(
        self,
        assessment: SeverityAssessment
    ) -> ResponseRecommendation:
        """生成响应建议"""
        pass


class SeverityLevel(Enum):
    """严重等级枚举"""
    LOW = "low"              # 低
    MEDIUM = "medium"        # 中
    HIGH = "high"            # 高
    CRITICAL = "critical"    # 紧急


@dataclass
class SeverityAssessment:
    """严重性评估结果"""
    level: SeverityLevel
    score: float  # 0-100
    factors: List[str]  # 影响因素
    escalation_recommended: bool  # 建议升级
    response_time_target: int  # 目标响应时间（秒）


@dataclass
class ResponseRecommendation:
    """响应建议"""
    immediate_actions: List[str]
    notification_targets: List[str]
    escalation_path: Optional[str]
    documentation_required: bool

2.6.3 严重性矩阵

事件类型	基础等级	室内场景	夜间场景	持续>5分钟
火灾	CRITICAL	+0	+0	+0
抽烟	MEDIUM	HIGH	MEDIUM	HIGH
徘徊	LOW	MEDIUM	HIGH	MEDIUM
违停	LOW	+0	LOW	MEDIUM

3. 数据流设计

3.1 完整数据流图

┌─────────────────────────────────────────────────────────────────────────────┐
│                              完整事件处理数据流                               │
└─────────────────────────────────────────────────────────────────────────────┘

[视频帧输入]
     │
     ▼
┌─────────────────┐
│  YOLO检测       │
│  • 目标检测     │
│  • 行为分析     │
└────────┬────────┘
         │ DetectionResult
         ▼
┌─────────────────┐     ┌─────────────────┐
│ 事件决策引擎    │────▶│ 置信度评估      │
│                 │     │ • 原始置信度    │
│ • 初步筛选      │     │ • 模型可信度    │
│ • 场景识别      │     │ • 历史校准      │
└────────┬────────┘     └─────────────────┘
         │ CandidateEvent
         ▼
┌─────────────────┐     ┌─────────────────┐
│ 预警规则引擎    │────▶│ 规则匹配        │
│                 │     │ • 时间规则      │
│ • 规则过滤      │     │ • 区域规则      │
│ • 频次检查      │     │ • 置信度规则    │
└────────┬────────┘     └─────────────────┘
         │ RuleCheckResult
         ▼
┌─────────────────┐     ┌─────────────────┐
│ 事件聚合器      │────▶│ 去重策略        │
│                 │     │ • 时间窗口      │
│ • 事件合并      │     │ • 空间邻近      │
│ • 关联分析      │     │ • 相似度计算    │
└────────┬────────┘     └─────────────────┘
         │ AggregatedEvent
         ▼
┌─────────────────┐     ┌─────────────────┐
│ 大模型触发器    │────▶│ 触发决策        │
│                 │     │ • 置信度区间    │
│ • 智能决策      │     │ • 场景匹配      │
│ • 队列管理      │     │ • 并发控制      │
└────────┬────────┘     └─────────────────┘
         │ (可选分支)
    ┌────┴────┐
    │         │
    ▼         ▼
┌──────┐  ┌─────────────────┐
│ 跳过 │  │ LLM视觉分析     │
│ LLM  │  │                 │
│      │  │ • 图像理解      │
│      │  │ • 场景分析      │
│      │  │ • 推理验证      │
└──┬───┘  └────────┬────────┘
   │               │ LLMResult
   └───────┬───────┘
           ▼
┌─────────────────┐     ┌─────────────────┐
│ 结果融合        │────▶│ 融合策略        │
│                 │     │ • 加权融合      │
│ • 置信度合并    │     │ • 冲突解决      │
│ • 决策生成      │     │ • 解释生成      │
└────────┬────────┘     └─────────────────┘
         │ FusionResult
         ▼
┌─────────────────┐     ┌─────────────────┐
│ 严重性评估器    │────▶│ 等级计算        │
│                 │     │ • 基础等级      │
│ • 风险评估      │     │ • 场景调整      │
│ • 响应建议      │     │ • 持续升级      │
└────────┬────────┘     └─────────────────┘
         │ SeverityAssessment
         ▼
┌─────────────────┐     ┌─────────────────┐
│ 事件格式化      │────▶│ 标准格式        │
│                 │     │ • 事件ID        │
│ • 元数据填充    │     │ • 时间戳        │
│ • 快照关联      │     │ • 位置信息      │
└────────┬────────┘     └─────────────────┘
         │ FinalEvent
         ▼
┌─────────────────┐
│ MQTT发布        │
│                 │
│ • 消息序列化    │
│ • QoS管理       │
│ • 发布确认      │
└─────────────────┘

3.2 状态流转

                    ┌─────────────┐
                    │   DETECTED  │
                    │   (检测到)   │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
       ┌──────────┐ ┌──────────┐ ┌──────────┐
       │ REJECTED │ │ PENDING  │ │ CONFIRMED│
       │ (已拒绝)  │ │ (待确认) │ │ (已确认) │
       └──────────┘ └────┬─────┘ └────┬─────┘
                         │            │
                         ▼            │
                  ┌──────────┐        │
                  │ LLM_ANAL │        │
                  │ (LLM分析)│        │
                  └────┬─────┘        │
                       │              │
                       ▼              │
                  ┌──────────┐        │
                  │ FUSION   │◀───────┘
                  │ (融合)   │
                  └────┬─────┘
                       │
                       ▼
                  ┌──────────┐
                  │ SEVERITY │
                  │ (严重性) │
                  └────┬─────┘
                       │
                       ▼
                  ┌──────────┐
                  │ PUBLISHED│
                  │ (已发布) │
                  └──────────┘

4. 算法详细设计

4.1 置信度校准算法

class ConfidenceCalibrator:
    """
    置信度校准器
    
    使用温度缩放（Temperature Scaling）校准模型置信度
    """
    
    def __init__(self, temperature: float = 1.0):
        self.temperature = temperature
        self.calibration_history = []
    
    def calibrate(
        self, 
        raw_confidence: float,
        model_type: str,
        scene: SceneType
    ) -> float:
        """
        校准置信度
        
        公式: calibrated = sigmoid(logit(raw) / temperature)
        """
        # 防止除零
        eps = 1e-10
        
        # 转换为logit
        raw_clipped = np.clip(raw_confidence, eps, 1 - eps)
        logit = np.log(raw_clipped / (1 - raw_clipped))
        
        # 温度缩放
        scaled_logit = logit / self.temperature
        
        # 转回概率
        calibrated = 1 / (1 + np.exp(-scaled_logit))
        
        # 场景调整
        scene_factor = self._get_scene_factor(scene)
        
        return min(1.0, calibrated * scene_factor)
    
    def _get_scene_factor(self, scene: SceneType) -> float:
        """获取场景调整因子"""
        factors = {
            SceneType.INDOOR: 0.95,
            SceneType.OUTDOOR: 1.0,
            SceneType.LOW_LIGHT: 0.85,
            SceneType.CROWDED: 0.9
        }
        return factors.get(scene, 1.0)

4.2 事件相似度算法

class EventSimilarityCalculator:
    """事件相似度计算器"""
    
    @staticmethod
    def calculate_iou(
        bbox1: Tuple[int, int, int, int],
        bbox2: Tuple[int, int, int, int]
    ) -> float:
        """计算边界框IoU"""
        x1_1, y1_1, x2_1, y2_1 = bbox1
        x1_2, y1_2, x2_2, y2_2 = bbox2
        
        # 计算交集
        xi1 = max(x1_1, x1_2)
        yi1 = max(y1_1, y1_2)
        xi2 = min(x2_1, x2_2)
        yi2 = min(y2_1, y2_2)
        
        inter_width = max(0, xi2 - xi1)
        inter_height = max(0, yi2 - yi1)
        inter_area = inter_width * inter_height
        
        # 计算并集
        box1_area = (x2_1 - x1_1) * (y2_1 - y1_1)
        box2_area = (x2_2 - x1_2) * (y2_2 - y1_2)
        union_area = box1_area + box2_area - inter_area
        
        return inter_area / union_area if union_area > 0 else 0.0
    
    @staticmethod
    def calculate_temporal_proximity(
        time1: datetime,
        time2: datetime,
        max_window: int = 30
    ) -> float:
        """计算时间邻近度"""
        diff_seconds = abs((time1 - time2).total_seconds())
        return max(0, 1 - diff_seconds / max_window)
    
    @staticmethod
    def calculate_feature_similarity(
        event1: CandidateEvent,
        event2: CandidateEvent
    ) -> float:
        """计算特征相似度"""
        similarities = []
        
        # 类型相似度
        type_sim = 1.0 if event1.event_type == event2.event_type else 0.0
        similarities.append(type_sim)
        
        # 置信度相似度
        conf_sim = 1.0 - abs(
            event1.confidence - event2.confidence
        )
        similarities.append(conf_sim)
        
        # 空间相似度
        spatial_sim = EventSimilarityCalculator.calculate_iou(
            event1.bbox, event2.bbox
        )
        similarities.append(spatial_sim)
        
        return np.mean(similarities)

4.3 优先级队列算法

class PriorityQueueManager:
    """优先级队列管理器"""
    
    def __init__(self):
        self.queue = []
        self.counter = 0
    
    def enqueue(
        self, 
        event: AggregatedEvent,
        priority_score: float
    ) -> str:
        """
        将事件加入优先级队列
        
        优先级计算考虑：
        1. 事件严重性（40%）
        2. 置信度不确定性（30%）
        3. 时间紧迫性（20%）
        4. 历史误报率（10%）
        """
        task_id = f"llm_task_{self.counter}"
        self.counter += 1
        
        # 使用堆实现优先级队列
        # Python heapq是最小堆，所以用负数优先级
        heapq.heappush(
            self.queue, 
            (-priority_score, self.counter, task_id, event)
        )
        
        return task_id
    
    def dequeue(self) -> Optional[Tuple[str, AggregatedEvent]]:
        """取出最高优先级的事件"""
        if not self.queue:
            return None
        
        _, _, task_id, event = heapq.heappop(self.queue)
        return task_id, event
    
    def calculate_priority(
        self,
        event: AggregatedEvent,
        severity: SeverityLevel,
        fp_rate: float
    ) -> float:
        """计算优先级分数"""
        # 严重性权重 (0-1)
        severity_weights = {
            SeverityLevel.CRITICAL: 1.0,
            SeverityLevel.HIGH: 0.8,
            SeverityLevel.MEDIUM: 0.5,
            SeverityLevel.LOW: 0.2
        }
        severity_score = severity_weights.get(severity, 0.5)
        
        # 置信度不确定性 (0-1, 越高越不确定)
        uncertainty = 1.0 - event.confidence_avg
        
        # 时间紧迫性 (0-1)
        time_urgency = self._calculate_time_urgency(event)
        
        # 综合计算
        priority = (
            0.4 * severity_score +
            0.3 * uncertainty +
            0.2 * time_urgency +
            0.1 * min(1.0, fp_rate * 2)  # 误报率越高优先级越高（需要验证）
        )
        
        return priority

5. 配置管理

5.1 配置文件结构

# config/event_engine.yaml

event_decision_engine:
  confidence_thresholds:
    discard: 0.5
    llm_required: 0.7
    direct_pass: 0.85
  
  scene_classification:
    enabled: true
    model_path: "models/scene_classifier.onnx"
  
  confidence_calibration:
    enabled: true
    temperature: 1.2
    history_window: 1000

alert_rule_engine:
  rule_files:
    - "rules/fire_detection.yaml"
    - "rules/smoking_detection.yaml"
    - "rules/loitering_detection.yaml"
  
  default_rules:
    cooldown_seconds: 30
    max_alerts_per_hour: 50
  
  dynamic_update:
    enabled: true
    check_interval: 60

event_aggregator:
  dedup_window_seconds: 30
  spatial_tolerance_pixels: 50
  iou_threshold: 0.3
  max_events_in_memory: 1000
  cleanup_interval: 300

llm_trigger:
  confidence_trigger_range: [0.5, 0.85]
  
  force_trigger_scenes:
    - "indoor_no_smoking_area"
    - "chemical_lab"
    - "server_room"
  
  concurrency_control:
    max_concurrent_calls: 5
    max_queue_size: 100
    timeout_seconds: 30
  
  cost_control:
    max_calls_per_minute: 30
    max_cost_per_hour: 10.0
    cache_enabled: true
    cache_ttl_seconds: 300
  
  priority_weights:
    severity: 0.4
    uncertainty: 0.3
    urgency: 0.2
    false_positive_rate: 0.1

result_fusion:
  default_strategy: "weighted"
  weights:
    yolo: 0.6
    llm: 0.4
  
  strategies:
    weighted:
      yolo_weight: 0.6
      llm_weight: 0.4
    conservative:
      mode: "both_high"
    optimistic:
      mode: "either_high"

severity_assessor:
  base_severity:
    fire: "critical"
    smoking: "medium"
    loitering: "low"
    illegal_parking: "low"
  
  scene_modifiers:
    indoor:
      smoking: +1  # 升级一级
    night_time:
      loitering: +1
    crowded:
      fire: +0  # 已经是最高
  
  duration_escalation:
    enabled: true
    thresholds:
      - duration: 300  # 5分钟
        escalation: +1
      - duration: 600  # 10分钟
        escalation: +2
  
  response_time_targets:
    critical: 10  # 秒
    high: 30
    medium: 120
    low: 300

5.2 规则文件示例

# rules/fire_detection.yaml
rules:
  - rule_id: "fire_critical"
    rule_name: "火灾紧急预警"
    event_type: "fire"
    description: "高置信度火灾检测，立即预警"
    
    conditions:
      confidence:
        min: 0.8
        max: 1.0
      time:
        - days: [0, 1, 2, 3, 4, 5, 6]  # 每天
          start: "00:00"
          end: "23:59"
      areas:
        include: ["warehouse", "lab", "office", "server_room"]
        exclude: []
    
    actions:
      alert_level: "critical"
      cooldown_seconds: 10
      max_frequency:
        per_hour: 100
        per_day: 500
      notifications:
        - type: "mqtt"
          topic: "jc-video/alerts/fire/critical"
          qos: 2
        - type: "webhook"
          url: "https://api.company.com/alerts"
      
    enabled: true
    priority: 1

  - rule_id: "fire_normal"
    rule_name: "火灾普通预警"
    event_type: "fire"
    description: "中等置信度火灾检测，需要LLM验证"
    
    conditions:
      confidence:
        min: 0.5
        max: 0.8
      time:
        - days: [0, 1, 2, 3, 4, 5, 6]
          start: "00:00"
          end: "23:59"
    
    actions:
      alert_level: "high"
      llm_verification: true
      cooldown_seconds: 30
      max_frequency:
        per_hour: 20
      notifications:
        - type: "mqtt"
          topic: "jc-video/alerts/fire/high"
          qos: 2
    
    enabled: true
    priority: 2

6. 接口定义

6.1 内部接口

# 事件决策引擎接口
class IEventDecisionEngine(ABC):
    @abstractmethod
    def evaluate(
        self, 
        detection_result: DetectionResult,
        context: DetectionContext
    ) -> List[CandidateEvent]:
        pass

# 预警规则引擎接口
class IAlertRuleEngine(ABC):
    @abstractmethod
    def check(
        self, 
        candidate_event: CandidateEvent
    ) -> RuleCheckResult:
        pass
    
    @abstractmethod
    def reload_rules(self) -> bool:
        """热加载规则"""
        pass

# 事件聚合器接口
class IEventAggregator(ABC):
    @abstractmethod
    def aggregate(
        self, 
        new_event: CandidateEvent
    ) -> Optional[AggregatedEvent]:
        pass
    
    @abstractmethod
    def get_active_events(self) -> List[AggregatedEvent]:
        pass

# LLM触发器接口
class ILLMTrigger(ABC):
    @abstractmethod
    def should_trigger(
        self, 
        event: AggregatedEvent
    ) -> TriggerDecision:
        pass
    
    @abstractmethod
    def submit_analysis(
        self, 
        event: AggregatedEvent
    ) -> str:
        """提交分析任务，返回任务ID"""
        pass
    
    @abstractmethod
    def get_result(
        self, 
        task_id: str
    ) -> Optional[LLMResult]:
        pass

# 严重性评估器接口
class ISeverityAssessor(ABC):
    @abstractmethod
    def assess(
        self,
        event: FusionResult,
        context: EventContext
    ) -> SeverityAssessment:
        pass

6.2 外部接口

# 主入口接口
class IEventJudgmentEngine(ABC):
    """
    事件判断引擎主接口
    
    这是外部系统调用的统一入口
    """
    
    @abstractmethod
    async def process_detection(
        self,
        detection_result: DetectionResult,
        context: DetectionContext
    ) -> Optional[FinalEvent]:
        """
        处理检测结果
        
        Args:
            detection_result: YOLO/行为检测结果
            context: 检测上下文
        
        Returns:
            最终事件（如果需要预警）或None
        """
        pass
    
    @abstractmethod
    async def process_frame(
        self,
        frame: np.ndarray,
        camera_id: str,
        timestamp: datetime
    ) -> List[FinalEvent]:
        """
        处理视频帧（完整流程）
        
        Args:
            frame: 视频帧
            camera_id: 摄像头ID
            timestamp: 时间戳
        
        Returns:
            检测到的事件列表
        """
        pass
    
    @abstractmethod
    def get_statistics(self) -> EngineStatistics:
        """获取引擎统计信息"""
        pass
    
    @abstractmethod
    def reload_configuration(self) -> bool:
        """重新加载配置"""
        pass

7. 性能考虑

7.1 性能指标目标

指标	目标值	说明
事件决策延迟	< 10ms	单事件决策时间
规则匹配延迟	< 5ms	单规则匹配时间
事件聚合延迟	< 20ms	包括相似度计算
LLM触发决策	< 5ms	决策时间（不含LLM调用）
严重性评估	< 5ms	评估时间
总处理延迟	< 50ms	不含LLM的完整流程
并发处理能力	> 100事件/秒	单实例
内存占用	< 500MB	事件缓存和队列

7.2 优化策略

7.2.1 计算优化

优化点	策略	预期收益
规则匹配	使用倒排索引	减少90%的比较次数
相似度计算	空间哈希网格	O(n) -> O(1)
置信度校准	预计算查找表	避免重复计算
场景分类	轻量级模型	< 5ms推理时间

7.2.2 缓存策略

class EventEngineCache:
    """事件引擎缓存管理"""
    
    def __init__(self):
        # 规则缓存
        self.rule_cache = LRUCache(maxsize=1000)
        
        # 场景分类缓存（帧级别）
        self.scene_cache = LRUCache(maxsize=100)
        
        # LLM结果缓存（图像特征级别）
        self.llm_cache = LRUCache(maxsize=1000)
        
        # 置信度校准缓存
        self.calibration_cache = LRUCache(maxsize=5000)
    
    def get_cache_key(
        self, 
        frame_hash: str,
        detection_bbox: Tuple[int, int, int, int]
    ) -> str:
        """生成缓存键"""
        return f"{frame_hash}_{bbox_hash(detection_bbox)}"

7.2.3 异步处理

class AsyncEventProcessor:
    """异步事件处理器"""
    
    def __init__(self):
        self.processing_queue = asyncio.Queue()
        self.llm_semaphore = asyncio.Semaphore(5)
        self.worker_tasks = []
    
    async def start(self, num_workers: int = 4):
        """启动处理工作线程"""
        self.worker_tasks = [
            asyncio.create_task(self._worker())
            for _ in range(num_workers)
        ]
    
    async def _worker(self):
        """工作线程"""
        while True:
            event = await self.processing_queue.get()
            try:
                await self._process_event(event)
            finally:
                self.processing_queue.task_done()
    
    async def _process_event(self, event: AggregatedEvent):
        """处理单个事件"""
        # 决策是否需要LLM
        trigger_decision = await self.llm_trigger.should_trigger(event)
        
        if trigger_decision.should_trigger:
            async with self.llm_semaphore:
                llm_result = await self.llm_client.analyze(event)
        else:
            llm_result = None
        
        # 融合结果
        fused_result = self.result_fusion.fuse(
            event.yolo_result,
            llm_result
        )
        
        # 评估严重性
        severity = self.severity_assessor.assess(fused_result)
        
        # 发布事件
        await self._publish_event(fused_result, severity)

7.3 监控指标

@dataclass
class EngineMetrics:
    """引擎监控指标"""
    
    # 处理指标
    events_processed: int = 0
    events_accepted: int = 0
    events_rejected: int = 0
    events_aggregated: int = 0
    
    # 延迟指标（毫秒）
    decision_latency_ms: float = 0.0
    rule_match_latency_ms: float = 0.0
    aggregation_latency_ms: float = 0.0
    llm_decision_latency_ms: float = 0.0
    severity_latency_ms: float = 0.0
    total_latency_ms: float = 0.0
    
    # LLM指标
    llm_calls_total: int = 0
    llm_calls_cached: int = 0
    llm_queue_size: int = 0
    llm_cost_usd: float = 0.0
    
    # 质量指标
    false_positive_rate: float = 0.0
    precision: float = 0.0
    recall: float = 0.0
    
    # 资源指标
    memory_usage_mb: float = 0.0
    cpu_usage_percent: float = 0.0

附录

A. 术语表

术语	英文	说明
候选事件	Candidate Event	通过初步筛选的潜在事件
聚合事件	Aggregated Event	合并去重后的事件
最终事件	Final Event	经过完整流程处理后的事件
触发决策	Trigger Decision	是否触发LLM分析的决策
融合结果	Fusion Result	YOLO和LLM结果融合后的结果
严重等级	Severity Level	事件的严重程度分级
冷却时间	Cooldown	同一类型事件的重复预警间隔

B. 数据模型关系图

DetectionResult
    │
    ├──> CandidateEvent (1:N)
    │       │
    │       └──> RuleCheckResult
    │               │
    │               └──> AggregatedEvent (N:1)
    │                       │
    │                       ├──> TriggerDecision
    │                       │       │
    │                       │       └──> LLMResult (Optional)
    │                       │
    │                       └──> FusionResult
    │                               │
    │                               └──> SeverityAssessment
    │                                       │
    │                                       └──> FinalEvent
    │
    └──> BehaviorAlert (if behavior detected)

文档结束

74 KiB Raw Permalink Blame History Unescape Escape

事件判断算法架构设计文档

文档信息

目录

1. 架构概述

1.1 设计目标

1.2 整体架构图

1.3 模块职责划分

2. 核心模块设计

2.1 事件决策引擎 (EventDecisionEngine)

2.1.1 功能说明

2.1.2 类设计

2.1.3 决策流程

2.1.4 置信度评估策略

2.2 预警规则引擎 (AlertRuleEngine)

2.2.1 功能说明

2.2.2 类设计

2.2.3 规则类型

2.2.4 规则配置示例

2.3 事件聚合器 (EventAggregator)

2.3.1 功能说明

2.3.2 类设计

2.3.3 聚合策略

2.3.4 相似度计算

2.4 大模型触发器 (LLMTrigger)

2.4.1 功能说明

2.4.2 多帧累积判断机制

2.4.3 类设计

2.4.4 触发策略（含多帧判断）

2.4.5 多帧累积配置示例

2.5 结果融合 (ResultFusion)

2.5.1 功能说明

2.5.2 类设计

2.5.3 融合策略

2.6 严重性评估器 (SeverityAssessor)

2.6.1 功能说明

2.6.2 类设计

2.6.3 严重性矩阵

3. 数据流设计

3.1 完整数据流图

3.2 状态流转

4. 算法详细设计

4.1 置信度校准算法

4.2 事件相似度算法

4.3 优先级队列算法

5. 配置管理

5.1 配置文件结构

5.2 规则文件示例

6. 接口定义

6.1 内部接口

6.2 外部接口

7. 性能考虑

7.1 性能指标目标

7.2 优化策略

7.2.1 计算优化

7.2.2 缓存策略

7.2.3 异步处理

7.3 监控指标

附录

A. 术语表

B. 数据模型关系图

74 KiB

Raw Permalink Blame History