525 lines
17 KiB
Python
525 lines
17 KiB
Python
"""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",
|
||
]
|