diff --git a/apps/server/services/llm_analysis_service.py b/apps/server/services/llm_analysis_service.py
new file mode 100644
index 0000000..37c7b5e
--- /dev/null
+++ b/apps/server/services/llm_analysis_service.py
@@ -0,0 +1,524 @@
+"""LLM 分析服务 (MVP-3 / D29)
+
+调用多模态大模型 (GPT-4V / Qwen-VL / GLM-4V) 对可疑事件进行二次判断。
+
+设计要点:
+
+1. **provider 抽象**: ``BaseLLMProvider`` 定义统一接口，
+   ``OpenAICompatibleProvider`` 覆盖所有 OpenAI 协议兼容的 vendor，
+   ``MockLLMProvider`` 提供完全离线的可预测行为，便于测试与本地开发。
+2. **并发控制**: 使用 ``asyncio.Semaphore`` 限制同时进行的 LLM 调用数量，
+   避免大量并发请求拖垮 API 配额。
+3. **降级策略**: 上层调用方可通过 ``analyze_with_fallback`` 在 LLM 失败时
+   返回 ``confirmed=None`` 的"未确认"结果，由 ``ResultFusion`` 决定如何处理。
+4. **图像编码**: 自动按最大边长缩放后编码为 base64 JPEG，控制 token 成本。
+5. **结构化输出**: prompt 强制 LLM 输出 JSON，解析失败时降级为
+   ``confirmed=None`` 而不是抛异常。
+
+依赖说明:
+
+- HTTP 客户端使用 ``httpx`` (已在 requirements 中)
+- 图像编码使用 ``opencv-python`` 已存在的 cv2.imencode
+
+无需新增 Python 依赖即可运行 (mock provider)，仅当 ``provider != mock`` 时
+才会真正发起 HTTP 请求。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import json
+import logging
+import re
+import time
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Sequence
+
+import numpy as np
+
+from models.event_schemas import CandidateEvent, EventType
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# 数据结构
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class LLMAnalysisRequest:
+    """LLM 分析请求。
+
+    Attributes:
+        candidate: 触发分析的候选事件 (含事件类型 / bbox)
+        frames: 关键帧序列 (BGR ``np.ndarray``)，
+            通常取累积窗口内的 1-3 张代表帧
+        prompt_extra: 业务侧附加提示词 (可选)
+        request_id: 请求标识 (用于日志关联)
+    """
+
+    candidate: CandidateEvent
+    frames: Sequence[np.ndarray]
+    prompt_extra: Optional[str] = None
+    request_id: Optional[str] = None
+
+
+@dataclass
+class LLMAnalysisResult:
+    """LLM 分析结果。
+
+    Attributes:
+        confirmed: True=LLM 确认事件成立，False=否决，None=未知/调用失败
+        confidence: LLM 给出的置信度 [0, 1]
+        reasoning: LLM 的简短推理说明
+        provider: 实际调用的 provider 名称
+        model: 实际调用的模型名
+        latency_ms: 调用耗时 (毫秒)
+        error: 失败时的错误信息
+        raw: 原始返回 (调试用)
+    """
+
+    confirmed: Optional[bool]
+    confidence: float = 0.0
+    reasoning: str = ""
+    provider: str = ""
+    model: str = ""
+    latency_ms: float = 0.0
+    error: Optional[str] = None
+    raw: Optional[str] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# Provider 抽象
+# ---------------------------------------------------------------------------
+
+
+class BaseLLMProvider:
+    """LLM provider 抽象。"""
+
+    name: str = "base"
+
+    async def analyze(self, request: LLMAnalysisRequest) -> LLMAnalysisResult:
+        raise NotImplementedError
+
+
+class MockLLMProvider(BaseLLMProvider):
+    """离线 Mock provider。
+
+    根据候选事件置信度生成可预测的判定结果，便于测试与无 API Key 环境。
+
+    规则:
+    - confidence >= 0.7  -> confirmed=True
+    - confidence <  0.3  -> confirmed=False
+    - 其余 -> confirmed=None
+    """
+
+    name = "mock"
+
+    def __init__(self, model: str = "mock-vlm") -> None:
+        self.model = model
+
+    async def analyze(self, request: LLMAnalysisRequest) -> LLMAnalysisResult:
+        start = time.time()
+        await asyncio.sleep(0)  # 模拟异步
+        conf = float(request.candidate.confidence)
+        if conf >= 0.7:
+            confirmed: Optional[bool] = True
+        elif conf < 0.3:
+            confirmed = False
+        else:
+            confirmed = None
+
+        reasoning = (
+            f"mock provider: candidate confidence={conf:.3f}, "
+            f"event_type={request.candidate.event_type.value}"
+        )
+        return LLMAnalysisResult(
+            confirmed=confirmed,
+            confidence=conf,
+            reasoning=reasoning,
+            provider=self.name,
+            model=self.model,
+            latency_ms=(time.time() - start) * 1000.0,
+        )
+
+
+class OpenAICompatibleProvider(BaseLLMProvider):
+    """OpenAI 协议兼容 provider (覆盖 GPT-4V / Qwen-VL / GLM-4V 等)。"""
+
+    name = "openai"
+
+    def __init__(
+        self,
+        api_base: str,
+        api_key: str,
+        model: str,
+        timeout: float = 15.0,
+        max_retries: int = 2,
+        max_tokens: int = 512,
+        temperature: float = 0.0,
+        provider_name: Optional[str] = None,
+    ) -> None:
+        if not api_base:
+            raise ValueError("api_base 不能为空")
+        if not api_key:
+            raise ValueError("api_key 不能为空")
+        self.api_base = api_base.rstrip("/")
+        self.api_key = api_key
+        self.model = model
+        self.timeout = timeout
+        self.max_retries = max_retries
+        self.max_tokens = max_tokens
+        self.temperature = temperature
+        if provider_name:
+            self.name = provider_name
+
+    async def analyze(self, request: LLMAnalysisRequest) -> LLMAnalysisResult:
+        try:
+            import httpx  # noqa: WPS433
+        except ImportError as exc:  # pragma: no cover - 依赖在 requirements 已声明
+            return LLMAnalysisResult(
+                confirmed=None,
+                provider=self.name,
+                model=self.model,
+                error=f"httpx 未安装: {exc}",
+            )
+
+        url = f"{self.api_base}/chat/completions"
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+        }
+        payload = {
+            "model": self.model,
+            "messages": _build_messages(request),
+            "max_tokens": self.max_tokens,
+            "temperature": self.temperature,
+        }
+
+        attempt = 0
+        last_error: Optional[str] = None
+        start = time.time()
+        while attempt <= self.max_retries:
+            attempt += 1
+            try:
+                async with httpx.AsyncClient(timeout=self.timeout) as client:
+                    resp = await client.post(url, headers=headers, json=payload)
+                    resp.raise_for_status()
+                    data = resp.json()
+                content = (
+                    data.get("choices", [{}])[0]
+                    .get("message", {})
+                    .get("content", "")
+                )
+                parsed = _parse_llm_json(content)
+                return LLMAnalysisResult(
+                    confirmed=parsed.get("confirmed"),
+                    confidence=float(parsed.get("confidence", 0.0) or 0.0),
+                    reasoning=str(parsed.get("reasoning", "")),
+                    provider=self.name,
+                    model=self.model,
+                    latency_ms=(time.time() - start) * 1000.0,
+                    raw=content,
+                )
+            except Exception as exc:  # noqa: BLE001
+                last_error = f"{type(exc).__name__}: {exc}"
+                logger.warning(
+                    "LLM 调用失败 attempt=%d/%d err=%s",
+                    attempt,
+                    self.max_retries + 1,
+                    last_error,
+                )
+                if attempt > self.max_retries:
+                    break
+                await asyncio.sleep(min(2 ** (attempt - 1), 5))
+
+        return LLMAnalysisResult(
+            confirmed=None,
+            provider=self.name,
+            model=self.model,
+            latency_ms=(time.time() - start) * 1000.0,
+            error=last_error,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Service
+# ---------------------------------------------------------------------------
+
+
+class LLMAnalysisService:
+    """LLM 二次判断服务。
+
+    Args:
+        provider: LLM provider 实例 (Mock / OpenAI 兼容)
+        max_concurrency: 同时进行的最大调用数
+        image_max_side: 编码前图像最大边长 (像素)
+    """
+
+    def __init__(
+        self,
+        provider: BaseLLMProvider,
+        max_concurrency: int = 2,
+        image_max_side: int = 768,
+    ) -> None:
+        if max_concurrency < 1:
+            raise ValueError("max_concurrency 必须 >= 1")
+        if image_max_side < 64:
+            raise ValueError("image_max_side 必须 >= 64")
+        self.provider = provider
+        self._semaphore = asyncio.Semaphore(max_concurrency)
+        self.image_max_side = image_max_side
+
+        # 统计
+        self._call_count = 0
+        self._success_count = 0
+        self._failure_count = 0
+        self._total_latency_ms = 0.0
+
+    async def analyze(self, request: LLMAnalysisRequest) -> LLMAnalysisResult:
+        """调用 LLM 分析单个事件 (并发受信号量限制)。"""
+
+        # 图像缩放预处理 (mock provider 也走一遍，保持行为一致)
+        prepared_frames = [
+            _resize_frame(f, self.image_max_side) for f in request.frames
+        ]
+        if any(p is not o for p, o in zip(prepared_frames, request.frames)):
+            request = LLMAnalysisRequest(
+                candidate=request.candidate,
+                frames=prepared_frames,
+                prompt_extra=request.prompt_extra,
+                request_id=request.request_id,
+            )
+
+        async with self._semaphore:
+            self._call_count += 1
+            try:
+                result = await self.provider.analyze(request)
+            except Exception as exc:  # noqa: BLE001
+                logger.error("LLM provider 异常: %s", exc)
+                result = LLMAnalysisResult(
+                    confirmed=None,
+                    provider=getattr(self.provider, "name", "unknown"),
+                    model=getattr(self.provider, "model", ""),
+                    error=f"{type(exc).__name__}: {exc}",
+                )
+
+        if result.error or result.confirmed is None:
+            self._failure_count += 1
+        else:
+            self._success_count += 1
+        self._total_latency_ms += result.latency_ms
+        return result
+
+    async def analyze_with_fallback(
+        self,
+        request: LLMAnalysisRequest,
+    ) -> LLMAnalysisResult:
+        """带降级的分析: 任意异常都不会向上抛出。"""
+
+        try:
+            return await self.analyze(request)
+        except Exception as exc:  # noqa: BLE001
+            logger.error("LLM 分析降级: %s", exc)
+            return LLMAnalysisResult(
+                confirmed=None,
+                provider=getattr(self.provider, "name", "unknown"),
+                model=getattr(self.provider, "model", ""),
+                error=f"{type(exc).__name__}: {exc}",
+            )
+
+    @property
+    def stats(self) -> Dict[str, Any]:
+        avg_latency = (
+            self._total_latency_ms / self._call_count
+            if self._call_count > 0
+            else 0.0
+        )
+        return {
+            "provider": getattr(self.provider, "name", "unknown"),
+            "model": getattr(self.provider, "model", ""),
+            "call_count": self._call_count,
+            "success_count": self._success_count,
+            "failure_count": self._failure_count,
+            "avg_latency_ms": round(avg_latency, 2),
+        }
+
+
+# ---------------------------------------------------------------------------
+# Factory
+# ---------------------------------------------------------------------------
+
+
+def create_llm_service_from_settings(settings: Any) -> Optional[LLMAnalysisService]:
+    """根据 ``Settings.llm`` 配置构造 LLMAnalysisService。
+
+    返回 None 表示未启用，调用方应跳过 LLM 分析。
+    """
+
+    llm_cfg = getattr(settings, "llm", None)
+    if llm_cfg is None or not getattr(llm_cfg, "enabled", False):
+        return None
+
+    provider_name = (llm_cfg.provider or "mock").lower()
+    if provider_name == "mock":
+        provider: BaseLLMProvider = MockLLMProvider(model=llm_cfg.model or "mock-vlm")
+    else:
+        api_base = llm_cfg.api_base
+        api_key = llm_cfg.api_key
+        if not api_base or not api_key:
+            logger.warning(
+                "LLM provider=%s 缺少 api_base/api_key，回退到 mock", provider_name
+            )
+            provider = MockLLMProvider(model=llm_cfg.model or "mock-vlm")
+        else:
+            provider = OpenAICompatibleProvider(
+                api_base=api_base,
+                api_key=api_key,
+                model=llm_cfg.model,
+                timeout=llm_cfg.timeout,
+                max_retries=llm_cfg.max_retries,
+                max_tokens=llm_cfg.max_tokens,
+                temperature=llm_cfg.temperature,
+                provider_name=provider_name,
+            )
+
+    return LLMAnalysisService(
+        provider=provider,
+        max_concurrency=llm_cfg.max_concurrency,
+        image_max_side=llm_cfg.image_max_side,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+_EVENT_TYPE_DESCRIPTIONS: Dict[EventType, str] = {
+    EventType.FIRE: "fire / open flame",
+    EventType.SMOKE: "smoke / haze",
+    EventType.SMOKING: "a person smoking a cigarette",
+    EventType.FIGHT: "a fight or physical altercation",
+    EventType.LOITERING: "a person loitering / lingering suspiciously",
+    EventType.STATIONARY: "a stationary person (possibly fallen or unconscious)",
+    EventType.INTRUSION: "an intrusion into a restricted area",
+    EventType.ILLEGAL_PARKING: "a vehicle illegally parked",
+    EventType.VEHICLE: "a vehicle of interest",
+    EventType.PERSON: "a person of interest",
+    EventType.UNKNOWN: "an unspecified suspicious event",
+}
+
+
+def _build_messages(request: LLMAnalysisRequest) -> List[Dict[str, Any]]:
+    """构造 OpenAI ChatCompletion messages。"""
+
+    candidate = request.candidate
+    event_desc = _EVENT_TYPE_DESCRIPTIONS.get(candidate.event_type, "an event")
+    bbox = candidate.detection.bbox.to_list()
+
+    system_prompt = (
+        "You are a strict video surveillance auditor. "
+        "Given one or more frames and a candidate event reported by an AI detector, "
+        "decide whether the event is truly present. "
+        "Respond with a single JSON object: "
+        '{"confirmed": true|false, "confidence": 0.0-1.0, "reasoning": "..."} '
+        "and nothing else."
+    )
+
+    user_text_parts: List[str] = [
+        f"Candidate event: {event_desc}.",
+        f"Detector confidence: {candidate.confidence:.3f}.",
+        f"Bounding box (x1,y1,x2,y2): {bbox}.",
+    ]
+    if request.prompt_extra:
+        user_text_parts.append(request.prompt_extra)
+    user_text_parts.append(
+        "Please answer strictly in JSON: "
+        '{"confirmed": <bool>, "confidence": <float>, "reasoning": <str>}'
+    )
+
+    content: List[Dict[str, Any]] = [
+        {"type": "text", "text": "\n".join(user_text_parts)},
+    ]
+    for frame in request.frames:
+        b64 = _encode_frame_base64(frame)
+        if b64:
+            content.append(
+                {
+                    "type": "image_url",
+                    "image_url": {"url": f"data:image/jpeg;base64,{b64}"},
+                }
+            )
+
+    return [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": content},
+    ]
+
+
+def _encode_frame_base64(frame: np.ndarray) -> Optional[str]:
+    """将 BGR 帧编码为 base64 JPEG 字符串。"""
+
+    try:
+        import cv2  # noqa: WPS433
+    except ImportError:  # pragma: no cover
+        return None
+    if frame is None or frame.size == 0:
+        return None
+    ok, buf = cv2.imencode(".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 80])
+    if not ok:
+        return None
+    return base64.b64encode(buf.tobytes()).decode("ascii")
+
+
+def _resize_frame(frame: np.ndarray, max_side: int) -> np.ndarray:
+    """按最大边长等比缩放 (节省 token)。"""
+
+    if frame is None or frame.size == 0:
+        return frame
+    try:
+        import cv2  # noqa: WPS433
+    except ImportError:  # pragma: no cover
+        return frame
+    h, w = frame.shape[:2]
+    longest = max(h, w)
+    if longest <= max_side:
+        return frame
+    scale = max_side / float(longest)
+    new_size = (max(1, int(w * scale)), max(1, int(h * scale)))
+    return cv2.resize(frame, new_size, interpolation=cv2.INTER_AREA)
+
+
+_JSON_PATTERN = re.compile(r"\{.*\}", re.DOTALL)
+
+
+def _parse_llm_json(content: str) -> Dict[str, Any]:
+    """从 LLM 文本输出中尽力解析 JSON。"""
+
+    if not content:
+        return {}
+    try:
+        return json.loads(content)
+    except json.JSONDecodeError:
+        match = _JSON_PATTERN.search(content)
+        if match:
+            try:
+                return json.loads(match.group(0))
+            except json.JSONDecodeError:
+                logger.debug("LLM JSON 解析失败: %s", content[:200])
+    return {}
+
+
+__all__ = [
+    "LLMAnalysisService",
+    "LLMAnalysisRequest",
+    "LLMAnalysisResult",
+    "BaseLLMProvider",
+    "MockLLMProvider",
+    "OpenAICompatibleProvider",
+    "create_llm_service_from_settings",
+]
diff --git a/apps/server/services/result_fusion.py b/apps/server/services/result_fusion.py
new file mode 100644
index 0000000..7f12a75
--- /dev/null
+++ b/apps/server/services/result_fusion.py
@@ -0,0 +1,373 @@
+"""结果融合器 (MVP-3 / D30)
+
+将 YOLO 候选事件 (含规则引擎产出的 ``AlertEvent``) 与 LLM 分析结果
+按指定策略融合，产出最终对外发出的 ``AlertEvent``。
+
+策略说明
+--------
+
+1. ``weighted`` (加权平均):
+   final_conf = yolo_weight * yolo_conf + llm_weight * llm_conf
+   - LLM 明确否决时 (confirmed=False) 同样适用，但置信度向 0 收敛
+   - LLM 不确定 (confirmed=None) 时，若 ``fallback_to_yolo`` 为 True
+     直接保留 YOLO 结果
+
+2. ``conservative`` (保守):
+   - LLM 否决 + ``suppress_on_llm_negative`` -> 抑制预警 (返回 None)
+   - LLM 确认 -> 取 ``min(yolo_conf, llm_conf)`` 与原值的最小值
+   - LLM 不确定 -> 抑制 (除非 ``fallback_to_yolo``)
+
+3. ``llm_priority`` (LLM 优先):
+   - LLM 确认 -> 直接使用 LLM 置信度
+   - LLM 否决 -> 抑制
+   - LLM 不确定 -> 回退到 YOLO 置信度 (若 ``fallback_to_yolo``)
+
+融合后会在 ``AlertEvent.metadata`` 中写入 ``llm`` 子对象，便于审计。
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional
+
+from models.event_schemas import AlertEvent, SeverityLevel
+from services.llm_analysis_service import LLMAnalysisResult
+
+logger = logging.getLogger(__name__)
+
+
+SUPPORTED_STRATEGIES = ("weighted", "conservative", "llm_priority")
+
+
+# ---------------------------------------------------------------------------
+# 融合输出
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class FusionOutcome:
+    """融合结果。"""
+
+    alert: Optional[AlertEvent]
+    suppressed: bool = False
+    reason: str = ""
+    final_confidence: float = 0.0
+    yolo_confidence: float = 0.0
+    llm_confidence: float = 0.0
+    strategy: str = ""
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# ResultFusion
+# ---------------------------------------------------------------------------
+
+
+class ResultFusion:
+    """YOLO + LLM 结果融合器。
+
+    Args:
+        strategy: 融合策略，见模块 docstring
+        yolo_weight: weighted 策略下 YOLO 权重
+        llm_weight: weighted 策略下 LLM 权重
+        suppress_on_llm_negative: LLM 明确否决时是否抑制预警
+        fallback_to_yolo: LLM 未确认 / 不可用时是否回退到 YOLO 结果
+        promote_severity_on_high_confidence: 高融合置信度是否提升严重性
+    """
+
+    def __init__(
+        self,
+        strategy: str = "weighted",
+        yolo_weight: float = 0.4,
+        llm_weight: float = 0.6,
+        suppress_on_llm_negative: bool = True,
+        fallback_to_yolo: bool = True,
+        promote_severity_on_high_confidence: bool = False,
+        high_confidence_threshold: float = 0.85,
+    ) -> None:
+        if strategy not in SUPPORTED_STRATEGIES:
+            raise ValueError(
+                f"不支持的融合策略: {strategy}, 支持: {SUPPORTED_STRATEGIES}"
+            )
+        if not 0.0 <= yolo_weight <= 1.0:
+            raise ValueError("yolo_weight 必须在 [0, 1]")
+        if not 0.0 <= llm_weight <= 1.0:
+            raise ValueError("llm_weight 必须在 [0, 1]")
+        total = yolo_weight + llm_weight
+        if total <= 0:
+            raise ValueError("yolo_weight + llm_weight 必须 > 0")
+
+        self.strategy = strategy
+        self.yolo_weight = yolo_weight / total
+        self.llm_weight = llm_weight / total
+        self.suppress_on_llm_negative = suppress_on_llm_negative
+        self.fallback_to_yolo = fallback_to_yolo
+        self.promote_severity_on_high_confidence = promote_severity_on_high_confidence
+        self.high_confidence_threshold = high_confidence_threshold
+
+    # ------------------------------------------------------------------
+    # 主入口
+    # ------------------------------------------------------------------
+
+    def fuse(
+        self,
+        alert: AlertEvent,
+        llm_result: Optional[LLMAnalysisResult],
+    ) -> FusionOutcome:
+        """融合单条预警事件与 LLM 结果。
+
+        Args:
+            alert: 规则引擎/聚合器产出的预警事件 (尚未发布)
+            llm_result: LLM 分析结果，None 表示未触发 LLM
+        """
+
+        yolo_conf = float(alert.confidence)
+
+        # 未调用 LLM 或 LLM 不可用
+        if llm_result is None:
+            return self._fallback_outcome(
+                alert,
+                yolo_conf=yolo_conf,
+                reason="llm_skipped",
+            )
+
+        if llm_result.error or llm_result.confirmed is None:
+            return self._fallback_outcome(
+                alert,
+                yolo_conf=yolo_conf,
+                llm_conf=llm_result.confidence,
+                reason=f"llm_unavailable:{llm_result.error or 'unknown'}",
+                llm_metadata=self._llm_metadata(llm_result),
+            )
+
+        # 至此 LLM 给出明确判定
+        if self.strategy == "weighted":
+            outcome = self._fuse_weighted(alert, yolo_conf, llm_result)
+        elif self.strategy == "conservative":
+            outcome = self._fuse_conservative(alert, yolo_conf, llm_result)
+        else:  # llm_priority
+            outcome = self._fuse_llm_priority(alert, yolo_conf, llm_result)
+
+        if outcome.alert is not None:
+            self._apply_metadata(outcome.alert, llm_result, outcome)
+            if self.promote_severity_on_high_confidence:
+                self._maybe_promote_severity(outcome.alert, outcome.final_confidence)
+
+        return outcome
+
+    # ------------------------------------------------------------------
+    # 策略实现
+    # ------------------------------------------------------------------
+
+    def _fuse_weighted(
+        self,
+        alert: AlertEvent,
+        yolo_conf: float,
+        llm_result: LLMAnalysisResult,
+    ) -> FusionOutcome:
+        # 否决降权: confirmed=False 时 LLM 置信度按 1-conf 反转
+        llm_conf = (
+            float(llm_result.confidence)
+            if llm_result.confirmed
+            else max(0.0, 1.0 - float(llm_result.confidence))
+        )
+        # confirmed=False 直接将 LLM 端贡献当作"反对票"
+        if not llm_result.confirmed:
+            # 反转后 llm 端置信度越高代表越反对，因此对最终值取 (yolo*w_y) - llm*w_l
+            final = self.yolo_weight * yolo_conf - self.llm_weight * llm_conf
+            final = max(0.0, min(1.0, final))
+        else:
+            final = self.yolo_weight * yolo_conf + self.llm_weight * llm_conf
+            final = max(0.0, min(1.0, final))
+
+        # 反对足够强且策略允许时抑制
+        if (
+            not llm_result.confirmed
+            and self.suppress_on_llm_negative
+            and final < 0.2
+        ):
+            return FusionOutcome(
+                alert=None,
+                suppressed=True,
+                reason="weighted_suppressed_by_llm_negative",
+                final_confidence=final,
+                yolo_confidence=yolo_conf,
+                llm_confidence=float(llm_result.confidence),
+                strategy=self.strategy,
+                metadata=self._llm_metadata(llm_result),
+            )
+
+        alert.confidence = round(final, 4)
+        return FusionOutcome(
+            alert=alert,
+            suppressed=False,
+            reason="weighted",
+            final_confidence=final,
+            yolo_confidence=yolo_conf,
+            llm_confidence=float(llm_result.confidence),
+            strategy=self.strategy,
+            metadata=self._llm_metadata(llm_result),
+        )
+
+    def _fuse_conservative(
+        self,
+        alert: AlertEvent,
+        yolo_conf: float,
+        llm_result: LLMAnalysisResult,
+    ) -> FusionOutcome:
+        if not llm_result.confirmed:
+            if self.suppress_on_llm_negative:
+                return FusionOutcome(
+                    alert=None,
+                    suppressed=True,
+                    reason="conservative_llm_negative",
+                    final_confidence=0.0,
+                    yolo_confidence=yolo_conf,
+                    llm_confidence=float(llm_result.confidence),
+                    strategy=self.strategy,
+                    metadata=self._llm_metadata(llm_result),
+                )
+            # 不抑制时也大幅降权
+            final = min(yolo_conf, 1.0 - float(llm_result.confidence))
+        else:
+            # 双方都确认: 取较小者，体现保守
+            final = min(yolo_conf, float(llm_result.confidence))
+
+        final = max(0.0, min(1.0, final))
+        alert.confidence = round(final, 4)
+        return FusionOutcome(
+            alert=alert,
+            suppressed=False,
+            reason="conservative",
+            final_confidence=final,
+            yolo_confidence=yolo_conf,
+            llm_confidence=float(llm_result.confidence),
+            strategy=self.strategy,
+            metadata=self._llm_metadata(llm_result),
+        )
+
+    def _fuse_llm_priority(
+        self,
+        alert: AlertEvent,
+        yolo_conf: float,
+        llm_result: LLMAnalysisResult,
+    ) -> FusionOutcome:
+        if not llm_result.confirmed:
+            if self.suppress_on_llm_negative:
+                return FusionOutcome(
+                    alert=None,
+                    suppressed=True,
+                    reason="llm_priority_negative",
+                    final_confidence=0.0,
+                    yolo_confidence=yolo_conf,
+                    llm_confidence=float(llm_result.confidence),
+                    strategy=self.strategy,
+                    metadata=self._llm_metadata(llm_result),
+                )
+            final = max(0.0, 1.0 - float(llm_result.confidence))
+        else:
+            final = float(llm_result.confidence)
+
+        final = max(0.0, min(1.0, final))
+        alert.confidence = round(final, 4)
+        return FusionOutcome(
+            alert=alert,
+            suppressed=False,
+            reason="llm_priority",
+            final_confidence=final,
+            yolo_confidence=yolo_conf,
+            llm_confidence=float(llm_result.confidence),
+            strategy=self.strategy,
+            metadata=self._llm_metadata(llm_result),
+        )
+
+    # ------------------------------------------------------------------
+    # 降级
+    # ------------------------------------------------------------------
+
+    def _fallback_outcome(
+        self,
+        alert: AlertEvent,
+        yolo_conf: float,
+        llm_conf: float = 0.0,
+        reason: str = "fallback",
+        llm_metadata: Optional[Dict[str, Any]] = None,
+    ) -> FusionOutcome:
+        if not self.fallback_to_yolo:
+            return FusionOutcome(
+                alert=None,
+                suppressed=True,
+                reason=f"{reason}:fallback_disabled",
+                final_confidence=0.0,
+                yolo_confidence=yolo_conf,
+                llm_confidence=llm_conf,
+                strategy=self.strategy,
+                metadata=llm_metadata or {},
+            )
+
+        alert.metadata.setdefault("llm", {})
+        if llm_metadata:
+            alert.metadata["llm"].update(llm_metadata)
+        alert.metadata["llm"]["fusion_reason"] = reason
+        return FusionOutcome(
+            alert=alert,
+            suppressed=False,
+            reason=reason,
+            final_confidence=yolo_conf,
+            yolo_confidence=yolo_conf,
+            llm_confidence=llm_conf,
+            strategy=self.strategy,
+            metadata=llm_metadata or {},
+        )
+
+    # ------------------------------------------------------------------
+    # 元数据 / 严重性
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _llm_metadata(llm_result: LLMAnalysisResult) -> Dict[str, Any]:
+        return {
+            "provider": llm_result.provider,
+            "model": llm_result.model,
+            "confirmed": llm_result.confirmed,
+            "confidence": round(float(llm_result.confidence), 4),
+            "reasoning": llm_result.reasoning,
+            "latency_ms": round(llm_result.latency_ms, 2),
+            "error": llm_result.error,
+            "evaluated_at": time.time(),
+        }
+
+    def _apply_metadata(
+        self,
+        alert: AlertEvent,
+        llm_result: LLMAnalysisResult,
+        outcome: FusionOutcome,
+    ) -> None:
+        alert.metadata.setdefault("llm", {})
+        alert.metadata["llm"].update(self._llm_metadata(llm_result))
+        alert.metadata["llm"]["fusion_strategy"] = self.strategy
+        alert.metadata["llm"]["fusion_reason"] = outcome.reason
+        alert.metadata["llm"]["final_confidence"] = round(
+            outcome.final_confidence, 4
+        )
+
+    @staticmethod
+    def _maybe_promote_severity(alert: AlertEvent, final_confidence: float) -> None:
+        order = [
+            SeverityLevel.INFO,
+            SeverityLevel.LOW,
+            SeverityLevel.MEDIUM,
+            SeverityLevel.HIGH,
+            SeverityLevel.CRITICAL,
+        ]
+        try:
+            idx = order.index(alert.severity)
+        except ValueError:
+            return
+        if final_confidence >= 0.95 and idx < len(order) - 1:
+            alert.severity = order[idx + 1]
+
+
+__all__ = ["ResultFusion", "FusionOutcome", "SUPPORTED_STRATEGIES"]