#!/usr/bin/env python3 """大模型告警测试脚本（OpenAI 兼容 API 格式）。 与 C++ 端 PostProcessor::algDescription / buildPrompt / buildDetectPrompt / parseVerdict 行为保持一致，用来对单张图片或文件夹下所有图片调用兼容 OpenAI 接口的大模型服务 （如 Ollama / vLLM / ModelScope 等），看大模型对某个算法的检测/复核效果。 用法示例： python3 test_openai_af.py --alg smoking_detection --input one.jpg python3 test_openai_af.py --alg cv_sewage_tank_overflow_detection --input ./imgs --mode detect python3 test_openai_af.py --alg helmet_detection --input ./imgs --model qwen3.5:2b --save-json out.json 文件夹模式下，文件名以 alarm_ / pos_ / 1_ 开头视为正样本， 以 noalarm_ / neg_ / 0_ 开头视为负样本，便于打印混淆矩阵；其余样本归 unknown。 你是一名严格的安防视觉判定员，请仔细观察图像完成 yes/no 判定。判断任务：图中是否存在以下情况？ 确认图中是否有工程车辆出现在施工现场，仅识别工程车辆，小轿车、普通卡车不计入判定目标。 人员明显手持施工工具器械（镐/铲/锹/钻/线缆等），注意避免日常用品的误报 输出仅返回单一 JSON 对象，禁止附加多余文字、Markdown 代码块，字段固定为 is_alarm、confidence、reason，不可修改、缺失； 存在目标示例 {"is_alarm": true, "confidence": 0.92, "reason": "画面中有清晰可辨的人体在画面中央活动"}，无目标示例 {"is_alarm": false, "confidence": 0.88, "reason": "画面无人，仅有静止背景"}。 判定遵循三条规则：1. 严格依照任务描述判定，不额外延伸场景与常识；2. confidence 置信度分级标准：大于 0.85 为几乎确定，0.6 至 0.85 为较有把握，低于 0.6 仅作疑似；3. reason 字段简洁客观描述画面可见内容，不写空洞无依据表述。 """ from __future__ import annotations import argparse import base64 import json import os import sys import time from pathlib import Path import requests # ───────────────────────────────────────────────────────────────────────────── # 与 C++ algDescription() 同步的算法描述表（修改时两边一起改） # ───────────────────────────────────────────────────────────────────────────── ALG_DESC_MAP: dict[str, str] = { "smoking_detection": "人员吸烟（手中或嘴边明显的香烟、烟头，或正在吐烟），注意目标小也要检测，注意避免笔/牙签/吸管/糖果棒等的误报", "phone_detection": "人员在打电话或手中持有手机查看，注意目标小也要检测，注意避免笔/遥控器/对讲机/小工具等的误报", "play_phone_detection": "人员长时间低头玩手机（手中可见手机屏幕），注意目标小也要检测，注意避免低头看书/写字/吃饭/操作设备等的误报", "helmet_detection": "施工/作业现场未佩戴安全帽的人员（头顶应有硬质安全帽却没有），注意避免鸭舌帽/草帽/头巾/发型被误判为已戴安全帽", "reflective_clothing_detection": "现场未穿着反光衣的人员，注意区分普通亮色衣物与带反光条纹的工作反光衣，避免亮色T恤误判为已穿", "mask_detection": "未规范佩戴口罩的人员（口鼻露出），注意避免胡须/手挡脸/低头/侧脸等遮挡导致的误报", "chef_hat_detection": "厨房人员未佩戴厨师帽，注意区分厨师帽与普通帽子/头巾/发网，避免戴普通帽子被判为已戴厨师帽", "work_clothes_detection": "未穿着规定工作服的人员，注意避免颜色相近的便装/外套被误判为工作服", "cv_gas_mask": "应防护场所未佩戴防毒面具的人员，注意区分普通口罩/眼镜与带滤罐的专业防毒面具", "cv_goggles": "应防护场所未佩戴护目镜的人员，注意区分普通近视眼镜/太阳镜与防护用护目镜", "cv_school_uniforms_detection": "校园内未穿着校服的学生，注意避免与校服颜色相近的便装误判", "fire_detection": "明显的明火/火焰（跳动的红黄色火苗，伴随发光），注意避免红色衣物/暖色灯光/夕阳/广告灯/反光等的误报", "smoke_detection": "明显的连续上升的烟雾（柱状或弥漫，伴有色泽），注意避免雾气/水汽/扬尘/云/灯光光晕等自然/光学现象的误报", "fire_and_smoke_detection": "明显的火焰或烟雾，注意避免红色物体/暖光/雾气/水汽/夕阳等的误报", "intrusion_detection": "图中是否出现人（含全身/半身/可识别的局部肢体，只要有人就是要报警的），注意目标小也要检测，注意避免衣服/海报/雕像等人形物的误报", "cv_perimeter_invasion": "周界禁区内出现的入侵人员，注意避免围栏外正常通行/远处路人的误报", "climbing_recognition": "人员明显的攀爬行为（双手抓握栏杆/墙体，身体明显离地或上抬），注意避免蹲下/弯腰/正常上下台阶/靠墙站立的误报", "over_the_wall_detection": "人员翻越围墙（一侧腿跨越墙体或骑跨在墙上），注意避免墙边正常站立/经过的误报", "lane_pedestrian_detection": "机动车道内出现的行人（在车流通行区域内步行），注意避免人行道/路肩/斑马线上的行人误报", "mouse_detection": "出现的老鼠（小型啮齿动物，灰褐色尖嘴长尾），注意避免阴影/污渍/小杂物/猫/玩具等的误报", "garbage_exposure": "暴露在外、未入桶的成堆/散落生活垃圾，注意避免树叶/正常摆放物品/施工材料/落地工具的误报", "trash_can_full": "垃圾桶已装满或溢出（垃圾明显高出桶口），注意避免桶盖打开但内部未满的误报", "cv_pour_bag_detection": "倾倒袋装物或往地面/绿化带乱丢垃圾的行为（手臂下抛动作 + 落地物），注意避免正常拎袋/搬运/递物的误报", "cv_large_item_inspection": "通道或区域内堆放的大件物品（沙发/床垫/纸箱/废旧家电等），注意避免行人/正常推车/小型行李箱的误报", "cv_aisle_cargo_detection": "通道内违规堆放、占用通道的货物（无人看管的成堆货物或托盘），注意避免短暂停留的推车/正在装卸的货物的误报", "people_gathering_detection": "明显的人员聚集（多人围拢停留并有交流），注意避免排队/路过/分散行走/同向移动的人群误报", "traffic_jam_detection": "明显的车辆拥堵（多辆车贴近停滞/缓行，长时间不动），注意避免红灯短时排队/正常停车场停车的误报", "illegal_parking_detection": "违规停放的车辆（占用消防通道/盲道/禁停区/人行道），注意避免上下客/装卸货/临时短停的误报", "charging_pile_occupied_by_oil_car": "燃油车占用充电车位（车位内是燃油车且无充电枪连接），注意避免新能源车正在充电的误报", "loitering_detection": "人员在区域内长时间徘徊（来回走动或反复出现），注意避免正常路过/短暂驻足看手机/等候朋友的误报", "cv_wandering": "人员在区域内徘徊（无明确目的地来回走），注意避免正常通行/短暂停留的误报", "leaving_post": "应在岗位（值班台/操作台/工位）的人员离岗（岗位空无一人），注意避免人员短暂偏离镜头但仍在岗位附近的误报", "sleeping_post": "在岗人员趴桌/闭眼睡觉（头部明显趴下或后仰且静止），注意避免低头看资料/打电话/调试设备等正常工作姿态的误报", "student_absent_detection": "学生座位空缺/缺席（应坐位置无人），注意避免学生短暂起立/俯身写字/被遮挡的误报", "wash_hands_detection": "人员洗手时长不足规范要求（用于水池/洗手台区域），主要看图中是否有手在洗手区", "cv_site_personnel_limit": "区域内人数明显超过设定上限，注意避免重复计数/远处路人误计", "cv_double_banknote_detection": "点钞过程中出现重复点钞或异常的钞票数量（双张/缺张），注意避免钞票自然弯曲/光线干扰的误报", "road_ponding": "路面明显的积水区域（有水面反光/连片水迹），注意避免阴影/路面颜色不均/油渍/未干洒水的误报", "water_gauge_reading_detection": "水位刻度尺上的水位读数超过预警线，注意避免水面倒影/光线变化/水尺被遮挡的误报", # "cv_sewage_tank_overflow_detection": "污水池/罐体液位溢出（液面接近或超过容器边缘，可能溢流），注意避免水面平静的正常液位/雨水积存/光线反射等的误报", # // {"cv_sewage_tank_overflow_detection", "污水池/罐体液位溢出（液面接近或超过容器边缘，可能溢流），注意避免水面平静的正常液位/雨水积存/光线反射等的误报"}, "cv_sewage_tank_overflow_detection":"污水池中有污水、浑浊水、沉淀水或者淤泥水，判断是否液位溢出（液面超过边缘），忽略平静的正常液位/雨水积存/光线反射的情况", # // {"cv_sewage_pit_warn_detection", "污水池中有污水、浑浊水、沉淀水或者淤泥水，红线是那面墙上的预警线，不是横跨在池体中上部的，这样看这面墙的水位/淤泥的高度是否超过了预警线。如果没有红线则判断污水池中水位/淤泥是否溢出容器边缘，注意避免水面雨水积存/光线反射等的误报"}, "cv_sewage_pit_warn_detection": "污水池中有污水、浑浊水、沉淀水或者淤泥水，红线是那面墙上的预警线，不是横跨在池体中上部的，这样看这面墙的水位/淤泥的高度是否超过了预警线。", # // {"cv_dunghill_exceed_detection", "粪堆超出限定范围或超量堆积（明显成堆的褐色/黑色粪便），注意避免土堆/砂石堆/落叶/正常少量堆放等的误报"}, "cv_dunghill_exceed_detection": "粪堆超出限定范围或超量堆积（明显成堆的褐色/黑色粪便），注意避免土堆/砂石堆/落叶/正常少量堆放等的误报", "cv_aeration_detection": "曝气池表面气泡缺失/曝气异常（水面平静、不见翻滚气泡），注意避免风浪/光线反射造成的误判", "cv_machine_room_property_detection": "机房资产位置发生变化（设备被搬动/缺失），注意避免光线变化/拍摄角度差异/小物件移动的误报", "cv_asset_deficiency_detection": "资产缺失/不在规定位置（应有物品消失），注意避免被人/其他物体部分遮挡造成的误报", "cv_cold_storage_goods_detection": "冷库内出现叉车作业或货物状态异常，注意避免正常通行人员/静态货架的误报", "cabinet_indicator_light_detection": "机柜指示灯异常（应绿灯亮起处出现红灯/熄灭/闪烁），注意避免反光/拍摄角度造成的颜色偏差", "cv_iden_minors": "出现明显的未成年人（学龄前/学龄儿童身高、体型、面部特征），注意避免身高较矮的成年人/远景人物误判", "cv_personnel_carrying_tools": "人员明显手持违规工具器械（管制刀具/棍棒/铁器等），注意避免日常用品/手机/雨伞的误报", "cv_sales_terminal_out_location": "销售终端（POS机/扫码设备）不在规定位置（指定位置缺失或被搬移），注意避免被手部/物品短暂遮挡的误报", "pointer_instrument_detection": "指针仪表读数超出告警阈值（指针指向红色/异常区域），注意避免反光/角度造成的读数偏差", "empty_parking_space_detection": "停车位空闲数量统计（用于盘点空位，需识别车位线内是否有车）", } IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".bmp", ".webp"} POSITIVE_PREFIXES = ("alarm_", "pos_", "1_", "true_") NEGATIVE_PREFIXES = ("noalarm_", "neg_", "0_", "false_") def alg_description(alg: str) -> str: return ALG_DESC_MAP.get(alg, alg) def build_prompt(alg: str, mode: str, roi_cropped: bool, custom: str | None = None) -> str: """对齐 C++ buildPrompt 文案。mode='verify' 二次复核 / 'detect' 直接检测。""" if custom: return custom desc = alg_description(alg) roi_line = ( "本图已从原始画面裁剪出算法关注的 ROI 区域（含少量周边 padding），仅在此范围内判断。" if roi_cropped else "本图为相机完整画面。" ) bg_block = ( "# 背景\n前端 AI 算法已触发告警，本次为独立复核，不必相信前端结论。\n\n" if mode == "verify" else "" ) return ( "你是一名严格的安防视觉判定员。请仔细观察图像，回答下面的 yes/no 判定。\n" "\n" "# 判断任务\n" "图中是否存在以下情况？\n" f"- {desc}\n" "\n" "# 图像说明\n" f"{roi_line}\n" "\n" f"{bg_block}" "# 输出格式\n" "仅输出一个 JSON 对象，不要任何额外文字、不要 markdown 代码块。\n" "字段名必须严格使用 is_alarm / confidence / reason 这三个，不能改成 alarm / result / detected 等其他名字，不能漏字段。\n" "\n" "正例（存在目标）：\n" "{\"is_alarm\": true, \"confidence\": 0.92, \"reason\": \"画面中有清晰可辨的人体在画面中央活动\"}\n" "正例（不存在目标）：\n" "{\"is_alarm\": false, \"confidence\": 0.88, \"reason\": \"画面无人，仅有静止背景\"}\n" "\n" "# 判定原则\n" "1. 严格按上面\"判断任务\"的描述判断，不要因算法英文名或常识引申到其他场景。\n" "2. confidence 含义：>0.85 几乎确定；0.6-0.85 较有把握；<0.6 仅疑似。\n" "3. reason 简明陈述看到了什么 / 没看到什么，避免空话。\n" ) def encode_image_b64(path: Path) -> str: return base64.b64encode(path.read_bytes()).decode("ascii") def call_openai( url: str, model: str, img_b64: str, prompt: str, timeout: float, ) -> tuple[str, dict]: """返回 (response_text, raw_json)；HTTP/解析异常向上抛。 使用 OpenAI 兼容的 /v1/chat/completions 接口（Ollama / vLLM / ModelScope 等均支持）。 """ body = { "model": model, "messages": [ { "role": "user", "content": [ {"type": "text", "text": prompt}, { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{img_b64}" } } ] } ], "temperature": 0.0, "max_tokens": 1024, "stream": False, "response_format": {"type": "json_object"}, } base = url if url.startswith(("http://", "https://")) else f"http://{url}" headers = {"Content-Type": "application/json"} # api_key = os.environ.get("OLLAMA_API_KEY", "").strip() api_key = "sk-MZn22xhtGtUERr1Mcb4uGAn3C7sraurI4bxicpEuh91u3Q3w" if api_key: headers["Authorization"] = f"Bearer {api_key}" resp = requests.post(f"{base}/v1/chat/completions", json=body, headers=headers, timeout=timeout) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"] or "", data def parse_verdict(text: str) -> tuple[bool | None, str]: """对齐 C++ parseVerdict。 返回 (is_alarm, reason)；is_alarm=None 表示无法判定（字段全缺失或解析失败）。 关键：is_alarm 字段缺失时返回 None，不再静默当 false。 """ if not text: return None, "" ALARM_KEYS = ("is_alarm", "alarm", "is_detected", "detected", "result", "alert") def _extract_alarm(v: dict) -> bool | None: for key in ALARM_KEYS: if key not in v: continue val = v[key] if isinstance(val, bool): return val if isinstance(val, (int, float)): return val != 0 if isinstance(val, str): s = val.strip().lower() if s in ("true", "yes", "y", "1"): return True if s in ("false", "no", "n", "0"): return False return None def _try(s: str): try: v = json.loads(s) except (json.JSONDecodeError, ValueError): return None if not isinstance(v, dict): return None is_alarm = _extract_alarm(v) reason = str(v.get("reason", "")) if v.get("reason") is not None else "" return is_alarm, reason r = _try(text) if r is not None and r[0] is not None: return r # 整体 parse 失败 / 字段缺失，尝试提取 {...} 子串 lb, rb = text.find("{"), text.rfind("}") if 0 <= lb < rb: r2 = _try(text[lb : rb + 1]) if r2 is not None and r2[0] is not None: return r2 if r is None and r2 is not None: r = r2 if r is not None: # 至少 reason 提到了，但 alarm 判定失败 return r return None, text.strip() def label_from_filename(name: str) -> str | None: lower = name.lower() if any(lower.startswith(p) for p in POSITIVE_PREFIXES): return "pos" if any(lower.startswith(p) for p in NEGATIVE_PREFIXES): return "neg" return None def collect_inputs(path: Path) -> list[Path]: if path.is_file(): return [path] if path.is_dir(): files = [p for p in path.rglob("*") if p.is_file() and p.suffix.lower() in IMAGE_EXTS] files.sort() return files raise FileNotFoundError(f"input not found: {path}") def fmt_bool(b: bool | None) -> str: if b is True: return "ALARM" if b is False: return "no " return "? " def main() -> int: ap = argparse.ArgumentParser(description="大模型告警测试（OpenAI 兼容 API）") ap.add_argument("--alg", required=True, help="算法名称（与 C++ alg_desc_map key 一致）") ap.add_argument("--input", required=True, help="单张图片文件，或图片所在目录") # ap.add_argument("--url", default="192.168.3.113:11434", help="API 服务地址（host:port，兼容 OpenAI 接口）") ap.add_argument("--url", default="https://newapi.cvvii.com", help="API 服务地址（host:port，兼容 OpenAI 接口）") ap.add_argument("--model", default="qwen3.5:9b", help="模型名") ap.add_argument("--mode", choices=["verify", "detect"], default="detect", help="prompt 模式：verify=二次校验/detect=直接检测（默认 detect）") ap.add_argument("--prompt", default=None, help="自定义 prompt 覆盖（非空则忽略 --mode）") ap.add_argument("--roi-cropped", action="store_true", help="prompt 中标注本图为 ROI 裁剪后的（仅影响 prompt 文案，不真做裁剪）") ap.add_argument("--timeout", type=float, default=300.0, help="单次请求超时秒") ap.add_argument("--save-json", default=None, help="把每张图的结果写入此 JSON 文件") ap.add_argument("--quiet", action="store_true", help="只打印汇总，不打印每张图") args = ap.parse_args() if args.alg not in ALG_DESC_MAP: print(f"[WARN] '{args.alg}' 不在 ALG_DESC_MAP 中，将用算法名作为目标描述") try: files = collect_inputs(Path(args.input)) except FileNotFoundError as e: print(e, file=sys.stderr) return 1 if not files: print("[ERROR] no images found", file=sys.stderr) return 1 prompt = build_prompt(args.alg, args.mode, args.roi_cropped, args.prompt) print("=" * 70) print(f"alg : {args.alg}") print(f"desc : {alg_description(args.alg)}") print(f"mode : {args.mode}{' (custom prompt)' if args.prompt else ''}") print(f"model : {args.model}") print(f"server : {args.url}") print(f"images : {len(files)} 张") print("-" * 70) print("Prompt:") print(prompt) print("=" * 70) counts = {"ALARM": 0, "no": 0, "unparsed": 0, "error": 0} confusion = {"TP": 0, "FP": 0, "TN": 0, "FN": 0, "unknown": 0} total_time = 0.0 results: list[dict] = [] for i, f in enumerate(files, 1): try: img_b64 = encode_image_b64(f) except OSError as e: print(f"[{i}/{len(files)}] {f.name} READ-ERROR: {e}") counts["error"] += 1 continue t0 = time.time() try: resp_text, raw = call_openai(args.url, args.model, img_b64, prompt, args.timeout) print("========================= resp_text ==========================") print(resp_text) print("==============================================================") except Exception as e: elapsed = time.time() - t0 counts["error"] += 1 results.append({"file": str(f), "error": str(e), "latency": elapsed}) print(f"[{i}/{len(files)}] {f.name} HTTP-ERROR ({elapsed:.1f}s): {e}") continue elapsed = time.time() - t0 total_time += elapsed is_alarm, reason = parse_verdict(resp_text) if is_alarm is True: counts["ALARM"] += 1 elif is_alarm is False: counts["no"] += 1 else: counts["unparsed"] += 1 gt = label_from_filename(f.name) if gt == "pos": if is_alarm is True: confusion["TP"] += 1 elif is_alarm is False: confusion["FN"] += 1 elif gt == "neg": if is_alarm is True: confusion["FP"] += 1 elif is_alarm is False: confusion["TN"] += 1 else: confusion["unknown"] += 1 results.append({ "file": str(f), "label": gt, "is_alarm": is_alarm, "reason": reason, "raw_response": resp_text, "latency": elapsed, }) if not args.quiet: tag = fmt_bool(is_alarm) gt_tag = f" gt={gt}" if gt else "" short_reason = (reason or "").replace("\n", " ")[:80] print(f"[{i:>3}/{len(files)}] {tag} ({elapsed:5.1f}s){gt_tag} {f.name} | {short_reason}") print("=" * 70) print("汇总：") n = len(files) n_done = n - counts["error"] print(f" 总数 {n} | ALARM {counts['ALARM']} no {counts['no']} unparsed {counts['unparsed']} error {counts['error']}") if n_done > 0: print(f" 平均耗时 {total_time / n_done:.2f}s/张") n_labelled = confusion["TP"] + confusion["FP"] + confusion["TN"] + confusion["FN"] if n_labelled > 0: tp, fp, tn, fn = confusion["TP"], confusion["FP"], confusion["TN"], confusion["FN"] precision = tp / (tp + fp) if (tp + fp) else 0.0 recall = tp / (tp + fn) if (tp + fn) else 0.0 accuracy = (tp + tn) / n_labelled f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0 print(f" ground-truth 样本 {n_labelled} 张（unknown {confusion['unknown']}）：") print(f" TP={tp} FP={fp} TN={tn} FN={fn}") print(f" 准确率 {accuracy:.3f} 精确率 {precision:.3f} 召回率 {recall:.3f} F1 {f1:.3f}") else: print(" （文件名未带 alarm_/noalarm_ 等前缀，跳过 TP/FP 统计）") if args.save_json: out = { "alg": args.alg, "model": args.model, "mode": args.mode, "url": args.url, "prompt": prompt, "counts": counts, "confusion": confusion, "avg_latency": (total_time / n_done) if n_done else None, "results": results, } Path(args.save_json).write_text(json.dumps(out, ensure_ascii=False, indent=2)) print(f" → 详细结果写入 {args.save_json}") return 0 if __name__ == "__main__": sys.exit(main()) # python3 test_openai_af.py --alg intrusion_detection --input /media/ps/data3/civi24/数据集/人/人/test_llm/12.jpg # python3 test_openai_af.py --alg cv_sewage_tank_overflow_detection --input /media/ps/data1/liuym/data/5-environment/污水池/train_coco/1125/images/ScreenShot_2025-11-24_112451_421.png