极端天气下的目标识别并不只是“换一个更大的模型”。雨雪会遮挡轮廓，雾霾会降低对比度，积水和反光会改变局部纹理，夜间补光还可能造成过曝。即便检测器在常规数据集上表现稳定，进入这些分布偏移场景后，也可能出现漏检、类别混淆和置信度失真。
更现实的问题是，业务通常不能把所有图片都交给人工审核，也不适合把每一帧都发送到外部视觉模型。前者成本高且响应慢，后者会引入延迟、费用、数据合规和外部依赖风险。因此，一个可落地的方案应当把计算分层：本地模型处理多数确定样本，只把低置信度、规则冲突或高风险样本送入复核链路，最终仍无法确定的任务再进入人工队列。
本文使用预训练目标检测器演示工程结构，不把它视为极端天气专用模型。示例模型是否适合真实道路、工业巡检或应急场景，必须通过业务数据验证；如果类别体系与预训练数据集不一致，还需要重新标注和训练。
分层识别的基本原理
整条链路可以拆成四层：
- 本地检测层：输出边界框、类别和置信度，承担主要吞吐量。
- 规则分流层：根据置信度、目标数量、关键类别和图像质量决定是否复核。
- 视觉模型复核层：查看完整图片与本地检测摘要，返回受约束的结构化判断。
- 人工处理层：接收模型意见不一致、字段缺失或高风险事件，并保存最终标签。
这里最重要的不是让两个模型简单“投票”，而是明确各自职责。本地检测器擅长定位，可以提供候选框；视觉模型更适合作为语义复核器，判断场景中是否存在明显遗漏、类别是否合理。复核结果不能无条件覆盖本地输出，而应经过 JSON Schema、类别白名单和风险规则校验。
分流阈值也不应凭直觉永久固定。可以先在标注验证集上记录每个阈值对应的漏检、误检与复核比例，再结合业务对不同错误的代价进行选择。若“漏掉道路障碍物”的代价高于多产生一次人工审核，就应为该类别设置更保守的阈值。
第一步：准备本地检测器
下面以 Python 3、PyTorch 和 torchvision 为例。实际安装命令要根据操作系统以及 CPU、CUDA 环境选择，尤其应确保 PyTorch 与 torchvision 的构建版本匹配。
python -m venv .venv
source .venv/bin/activate
pip install torch torchvision pillow requests pydantic
创建 local_detector.py：
from pathlib import Path
import torch
from PIL import Image
from torchvision.models.detection import (
FasterRCNN_ResNet50_FPN_Weights,
fasterrcnn_resnet50_fpn,
)
WEIGHTS = FasterRCNN_ResNet50_FPN_Weights.DEFAULT
MODEL = fasterrcnn_resnet50_fpn(weights=WEIGHTS).eval()
PREPROCESS = WEIGHTS.transforms()
CATEGORIES = WEIGHTS.meta["categories"]
def detect(image_path: str, score_floor: float = 0.20) -> list[dict]:
image = Image.open(Path(image_path)).convert("RGB")
tensor = PREPROCESS(image)
with torch.inference_mode():
output = MODEL([tensor])[0]
results = []
for box, label, score in zip(
output["boxes"], output["labels"], output["scores"]
):
confidence = float(score)
if confidence < score_floor:
continue
results.append(
{
"label": CATEGORIES[int(label)],
"confidence": round(confidence, 4),
"box": [round(float(v), 1) for v in box],
}
)
return results
score_floor 只是过滤极低分候选框的下限，不是生产环境的最终阈值。示例权重会在首次运行时下载；离线环境应提前缓存权重，并通过制品管理记录其来源、校验值和部署版本。
第二步：建立可解释的分流规则
不要只使用单一平均置信度。平均值可能掩盖关键目标的低分结果，建议同时检查最低置信度、候选数量和关键类别。
REVIEW_THRESHOLD = 0.55
CRITICAL_LABELS = {
"person", "car", "truck", "bus"}
def should_review(detections: list[dict]) -> tuple[bool, list[str]]:
reasons = []
if not detections:
reasons.append("no_detection")
low_confidence = [
item for item in detections
if item["confidence"] < REVIEW_THRESHOLD
]
if low_confidence:
reasons.append("low_confidence")
critical = [
item for item in detections
if item["label"] in CRITICAL_LABELS
]
if critical and min(x["confidence"] for x in critical) < 0.70:
reasons.append("uncertain_critical_object")
if len(detections) > 50:
reasons.append("abnormal_object_count")
return bool(reasons), reasons
这些数值仅用于展示代码结构，不能直接当作通用生产参数。上线前应按摄像头位置、天气类型和业务类别分别评估；如果数据分布差异明显，可以采用分场景阈值，而不是强求一个全局值。
第三步：接入可替换的视觉复核接口
复核层最好依赖内部统一协议，使业务代码不绑定某一家服务。以下示例假设目标服务明确支持与示例一致的请求格式，并且所选模型能够接收图像。模型名称、上下文限制、图片格式、计费方式和数据处理政策均应以服务方当前文档为准。
可将 HaerAPI 作为候选接入端进行文档核验和隔离环境联调，也可以替换为企业自建网关或其他合规服务。
先通过环境变量提供配置，密钥不得写进代码或提交到仓库：
export VISION_API_BASE="https://your-provider.example/v1"
export VISION_API_KEY="replace-with-runtime-secret"
export VISION_MODEL="replace-with-a-vision-model"
创建 vision_reviewer.py：
import base64
import json
import mimetypes
import os
from pathlib import Path
import requests
from pydantic import BaseModel, Field, ValidationError
class ReviewResult(BaseModel):
needs_human: bool
weather: str
missing_labels: list[str] = Field(default_factory=list)
suspicious_labels: list[str] = Field(default_factory=list)
reason: str
def as_data_url(image_path: str) -> str:
path = Path(image_path)
mime = mimetypes.guess_type(path.name)[0] or "image/jpeg"
encoded = base64.b64encode(path.read_bytes()).decode("ascii")
return f"data:{mime};base64,{encoded}"
def review(image_path: str, detections: list[dict]) -> ReviewResult:
base_url = os.environ["VISION_API_BASE"].rstrip("/")
api_key = os.environ["VISION_API_KEY"]
model = os.environ["VISION_MODEL"]
instruction = {
"task": "复核本地目标检测结果，不生成边界框",
"allowed_weather": ["clear", "rain", "snow", "fog", "low_light", "unknown"],
"local_detections": detections,
"output_schema": {
"needs_human": "boolean",
"weather": "string",
"missing_labels": "string[]",
"suspicious_labels": "string[]",
"reason": "string"
}
}
response = requests.post(
f"{base_url}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
json={
"model": model,
"temperature": 0,
"messages": [{
"role": "user",
"content": [
{
"type": "text", "text": json.dumps(instruction, ensure_ascii=False)},
{
"type": "image_url", "image_url": {
"url": as_data_url(image_path)}},
],
}],
},
timeout=(5, 45),
)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"]
try:
return ReviewResult.model_validate_json(content)
except ValidationError as exc:
raise ValueError("视觉复核返回内容不符合约定结构") from exc
部分接口可能不接受 Data URL，或者使用不同的多模态消息结构；此时应按照已核验的接口文档修改适配器。不要在未知情况下假定接口完全兼容。对于大图片，还应在本地去除无关元数据、限制分辨率和文件大小，但压缩参数要通过识别效果评估后确定。
第四步：编排失败降级和人工闭环
外部复核失败不能拖垮主检测服务。调用方应区分业务不确定、响应格式错误、超时和限流，并让高风险任务进入人工队列。
from local_detector import detect
from routing import should_review
from vision_reviewer import review
def process(image_path: str) -> dict:
detections = detect(image_path)
required, reasons = should_review(detections)
result = {
"detections": detections,
"review_reasons": reasons,
"review": None,
"status": "accepted",
}
if not required:
return result
try:
checked = review(image_path, detections)
result["review"] = checked.model_dump()
result["status"] = "human_pending" if checked.needs_human else "reviewed"
except Exception as exc:
result["status"] = "human_pending"
result["review_error"] = type(exc).__name__
return result
生产实现不要把完整异常、密钥或原始图片写入普通日志。建议记录任务 ID、本地模型版本、规则版本、复核模型标识、耗时区间、错误类型和最终人工标签。图片本身应放在权限受控的对象存储中，并配置保留期限。
人工修正结果不是只用于事后统计。经过脱敏和质量抽查后，它可以形成困难样本集，用于重新训练检测器、校准阈值以及评估复核提示词。评测集需要按摄像头或采集批次合理划分，避免相邻帧同时进入训练集和验证集造成数据泄漏。
常见问题
为什么不直接让视觉大模型完成全部检测？
视觉模型可以描述场景，但描述正确不等于稳定输出精确边界框。持续上传全部图片还会增加成本、时延和数据治理压力。只有在经过专项评测、接口满足定位要求且业务能够接受相关约束时，才适合扩大其职责。
低置信度一定意味着检测错误吗？
不一定。置信度是模型输出，不是天然准确率。不同模型、类别和场景的分数不可直接横向比较，应在独立标注集上做校准，并结合误报与漏报代价设定阈值。
两个模型结论冲突时听谁的？
不要预设某个模型永远正确。关键类别冲突、复核输出越界或图像质量不足时，应进入人工队列。对低风险类别，可以根据离线评测形成明确的优先级规则，并对规则做版本化管理。
如何控制外部接口故障的影响？
设置连接与读取超时、并发上限和熔断策略；只对可恢复错误做有限次数重试，并加入退避。复核服务不可用时，本地检测仍应工作，高风险结果转入待处理队列。是否允许低风险结果直接通过，需要由业务风险等级决定。
如何处理隐私与合规？
发送前先判断图片是否包含人脸、车牌、地理位置或内部设施等敏感信息。根据场景进行裁剪、遮挡或禁止外发，并确认服务的数据保存、训练使用、跨境传输和删除机制。仅隐藏 API 密钥并不能解决数据合规问题。
总结
极端天气目标识别的工程重点，是承认单一模型存在边界，并建立可观察、可降级、可复核的处理链路。本地检测器负责吞吐和定位，规则层负责控制复核范围，视觉模型补充语义判断，人工环节处理真正困难或高风险的样本。
落地时应优先完成三件事：用真实业务标注集校准阈值；把视觉接口封装为可替换适配器；保存模型、规则与人工结论之间的可追溯关系。这样即使模型或服务供应方发生变化，识别系统仍能通过明确契约和评测流程持续演进。
