摘要：上一篇把端侧应用封装成「检测 → 事件 → 小模型解释」。卡点通常不在解释脚本，而在检测结果仍停在画框上，下游拿不到稳定输入。本文写如何把现有 YOLO / TensorRT 检测输出，自动写成events/*.json：字段约定、阈值过滤、冷却去重、截图保存，以及怎样接到已有推理循环。适合已经跑通 Orin 检测闭环，准备接大模型解释层的人。可与 Orin 上目标检测 + 大模型做端侧应用、Orin 上目标检测最小闭环 一起看。

建议目录（可与上一篇共用）：

mkdir-p~/orin-edge-app/{events,scripts,out,logs,snapshots}cd~/orin-edge-app# 本文新增：event_writer.py / demo_from_detections.py / watch_and_explain.sh

1. 先写成 JSON

很多检测程序到这一步就停了：

• 屏幕上有框
• 终端打印了检出 N 个目标
• 下一帧继续画

对值班系统和大模型解释层来说，这还不够。它们需要的是可消费的记录：什么时间、哪路相机、什么类别、置信度多少、要不要重复报。

图1. 画框是给人看的；events/*.json才是给下游系统看的。

所以本篇目标很具体：

不重写整个检测工程，只在“得到 boxes / scores / cls_ids”之后，多写一步，把结果保存成事件文件。

2. 先说结论

1. 事件层是检测和大模型之间的稳定接口。
2. 先过滤和冷却，再谈解释；否则语言模型会被误报灌爆。
3. 先用假检测数据把事件写入跑通，再挂到真实detect_trt.py循环。

图2. 检测输出 → 过滤 → 冷却 → 写成 JSON。

3. 事件字段与上一篇对齐

继续使用这套字段（与上一篇一致，便于直接喂给解释脚本）：

示例：

{"event_id":"cam01-20260804-093000-001","ts":"2026-08-04T09:30:00","camera_id":"cam01","line_id":"line-A","defect":"scratch","confidence":0.91,"bbox":[100,120,240,260],"repeat_count_1min":2,"snapshot_path":"snapshots/cam01_20260804_093000.jpg","kb_hint":""}

4. 写入事件前先做过滤

图3. 白名单、阈值、冷却，决定事件质量。

没有这三个规则，大模型解释层会变成“误报扩音器”。

5. 核心模块：event_writer.py

保存为scripts/event_writer.py：

#!/usr/bin/env python3"""把检测结果写成 events/*.json。 输入约定： boxes: Nx4, xyxy scores: N cls_ids: N """from__future__importannotationsimportjsonimporttimefromdataclassesimportdataclass,fieldfromdatetimeimportdatetimefrompathlibimportPathfromtypingimportDict,Iterable,List,Optional,Sequence,Tupleimportcv2importnumpyasnp@dataclassclassEventWriterConfig:events_dir:Path=Path("events")snapshots_dir:Path=Path("snapshots")camera_id:str="cam01"line_id:str="line-A"class_names:Dict[int,str]=field(default_factory=dict)allow_classes:Optional[set]=None# 空=全开；建议填缺陷类名集合conf_thres:float=0.5cooldown_sec:float=15.0save_snapshot:bool=Truekb_hint:str=""classEventWriter:def__init__(self,cfg:EventWriterConfig):self.cfg=cfg self.cfg.events_dir.mkdir(parents=True,exist_ok=True)self.cfg.snapshots_dir.mkdir(parents=True,exist_ok=True)self._last_fire:Dict[Tuple[str,str],float]={}self._repeat_1min:Dict[Tuple[str,str],List[float]]={}self._seq=0def_classname(self,cls_id:int)->str:returnself.cfg.class_names.get(int(cls_id),str(int(cls_id)))def_allowed(self,name:str)->bool:ifnotself.cfg.allow_classes:returnTruereturnnameinself.cfg.allow_classesdef_repeat_count(self,key:Tuple[str,str],now:float)->int:arr=[tfortinself._repeat_1min.get(key,[])ifnow-t<=60.0]arr.append(now)self._repeat_1min[key]=arrreturnlen(arr)defwrite_detections(self,boxes:Sequence[Sequence[float]],scores:Sequence[float],cls_ids:Sequence[int],frame_bgr:Optional[np.ndarray]=None,)->List[Path]:"""返回本帧新写入的事件文件路径。"""written:List[Path]=[]now=time.time()ts=datetime.now().strftime("%Y-%m-%dT%H:%M:%S")stamp=datetime.now().strftime("%Y%m%d_%H%M%S")forbox,score,cls_idinzip(boxes,scores,cls_ids):iffloat(score)<self.cfg.conf_thres:continuename=self._classname(int(cls_id))ifnotself._allowed(name):continuekey=(self.cfg.camera_id,name)last=self._last_fire.get(key,0.0)ifnow-last<self.cfg.cooldown_sec:# 冷却期内仍计入 1 分钟重复次数，但先不写新文件self._repeat_count(key,now)continueself._last_fire[key]=now repeat=self._repeat_count(key,now)self._seq+=1event_id=f"{self.cfg.camera_id}-{stamp}-{self._seq:03d}"snapshot_path=""ifself.cfg.save_snapshotandframe_bgrisnotNone:snap=self.cfg.snapshots_dir/f"{self.cfg.camera_id}_{stamp}_{self._seq:03d}.jpg"cv2.imwrite(str(snap),frame_bgr)snapshot_path=str(snap)x1,y1,x2,y2=[int(round(float(v)))forvinbox]event={"event_id":event_id,"ts":ts,"camera_id":self.cfg.camera_id,"line_id":self.cfg.line_id,"defect":name,"confidence":round(float(score),4),"bbox":[x1,y1,x2,y2],"repeat_count_1min":repeat,"snapshot_path":snapshot_path,"kb_hint":self.cfg.kb_hint,}out=self.cfg.events_dir/f"{event_id}.json"out.write_text(json.dumps(event,ensure_ascii=False,indent=2)+"\n",encoding="utf-8")written.append(out)returnwrittendefdemo_boxes():"""无引擎时的本地自测数据。"""boxes=np.array([[100,120,240,260],[300,80,400,180]],dtype=np.float32)scores=np.array([0.91,0.32],dtype=np.float32)cls_ids=np.array([0,2],dtype=np.int32)returnboxes,scores,cls_ids

这段代码保持独立：不绑定某一版 TensorRT 封装，只要提供boxes/scores/cls_ids就能用。

6. 先用假数据把事件写入跑通

保存scripts/demo_from_detections.py：

#!/usr/bin/env python3"""不依赖引擎，先验证 EventWriter 是否正常写出事件。"""frompathlibimportPathimportsys sys.path.append(str(Path(__file__).resolve().parent))fromevent_writerimportEventWriter,EventWriterConfig,demo_boxes# noqa: E402defmain()->None:cfg=EventWriterConfig(events_dir=Path("events"),snapshots_dir=Path("snapshots"),camera_id="cam01",line_id="line-A",class_names={0:"scratch",1:"dent",2:"person"},allow_classes={"scratch","dent"},# person 会被过滤conf_thres=0.5,cooldown_sec=5.0,save_snapshot=False,kb_hint="历史相似：导轨毛刺可能导致周期性划痕",)writer=EventWriter(cfg)boxes,scores,cls_ids=demo_boxes()foriinrange(3):paths=writer.write_detections(boxes,scores,cls_ids,frame_bgr=None)print(f"round={i+1}written={[str(p)forpinpaths]}")if__name__=="__main__":main()

cd~/orin-edge-app python3 scripts/demo_from_detections.pylsevents/catevents/*.json|head

预期：

• 第 1 轮写入scratch（person因白名单被滤，低分目标因阈值被滤）
• 第 2、3 轮若在冷却期内，不再重复写同相机同类文件

确认这一点后，再接到真实检测循环，排障会轻松很多。

7. 接到现有 YOLO / TensorRT 循环

图4. 在得到 boxes 之后插入 EventWriter，不必重写引擎加载。

以 Orin 上目标检测最小闭环 里的流程为例：你已经有

preprocess → infer → postprocess → boxes/scores/cls_ids → draw

只需在draw前后加上事件写入。关键不是改 TensorRT 初始化，而是确认一件事：

引擎负责推理；事件模块负责“哪些结果值得留下”。

7.1 最小插入方式

# 放在检测主程序同目录，或把 scripts 加进 PYTHONPATHfrompathlibimportPathfromevent_writerimportEventWriter,EventWriterConfig writer=EventWriter(EventWriterConfig(events_dir=Path("events"),snapshots_dir=Path("snapshots"),camera_id="cam01",line_id="line-A",# 按你的类别映射改；若是自训缺陷模型，这里写缺陷名class_names={0:"scratch",1:"dent"},allow_classes={"scratch","dent"},conf_thres=0.5,cooldown_sec=15.0,save_snapshot=True,))# ... 推理得到 boxes, scores, cls_ids, frame ...paths=writer.write_detections(boxes,scores,cls_ids,frame_bgr=frame)forpinpaths:print(f"[event]{p}")

7.2 视频循环里还要注意

若你用的是视频或相机循环，下面几条比“会不会写 JSON”更容易踩坑：

1. 每帧都可能有框，但不代表每帧都该写事件
   冷却就是为这个场景准备的。没有冷却，解释队列和磁盘都会先被打满。

2. 截图保存必须限频
   1080p 连续存 JPEG，几小时就能把小容量系统盘写满。事件可以留，截图可先关，或只在成功写出事件时存。

3. 类别映射一定要显式写清
   不要默认把cls_id当缺陷名。自训模型还好办；若临时用 COCO 权重做联调，更要在配置里写明“演示映射”，避免后面误当成业务类别。

4. 告警阈值可以高于可视化阈值
   画面上0.25能看到框，不代表就要写事件。事件阈值建议更严，例如0.5起，先保证下游干净。

5. 先统计“每分钟事件数”
   接大模型前，先看真实运行 10 分钟会写出多少 JSON。若已经上百条，先调白名单和冷却，不要先开解释。

7.3 和detect_trt.py的分工

更稳的结构是两个文件：

不要把事件逻辑塞进 TensorRT 封装类深处。下次换导出方式、换后处理，事件层还能原样复用。

COCO 预训练权重做质检 demo 时，常见做法是：

• 优先改用自训缺陷模型；或
• 临时把某个类映射成演示缺陷名，但配置和日志里必须写清楚这是演示映射

8. JSON 稳定后，再接解释层

图5. 事件接口稳了，上一篇的explain_event.py才能真正进工作流。

可选：监听到新文件就解释（先串行，别一上来高并发）。

保存scripts/watch_and_explain.sh：

#!/usr/bin/env bashset-euopipefailEVENTS_DIR="${1:-events}"MODEL="${MODEL:-qwen2.5:7b}"PROMPT="${PROMPT:-prompts/explain_alarm.txt}"OUT_DIR="${OUT_DIR:-out}"mkdir-p"$OUT_DIR"# 需要 inotifywait：sudo apt-get install -y inotify-toolsinotifywait-m-eclose_write--format'%w%f'"$EVENTS_DIR"|whileread-rf;docase"$f"in*.json)base="$(basename"$f".json)"echo"[explain]$f"python3 scripts/explain_event.py\--event"$f"\--prompt"$PROMPT"\--model"$MODEL"\--out"$OUT_DIR/${base}.txt"||echo"[explain failed]$f";;esacdone

chmod+x scripts/watch_and_explain.sh# 先确保 explain_event.py 与 prompts/explain_alarm.txt 已按上一篇就位# ./scripts/watch_and_explain.sh events

没有inotify-tools时，也可以简单轮询：

python3 -<<'PY' from pathlib import Path import time, subprocess seen=set() while True: for p in sorted(Path('events').glob('*.json')): if p in seen: continue seen.add(p) out=Path('out')/(p.stem+'.txt') print('explain', p) subprocess.run([ 'python3','scripts/explain_event.py', '--event',str(p),'--out',str(out) ], check=False) time.sleep(1) PY

9. 常见坑

9.1 每帧都写 JSON

结果：磁盘爆、解释队列堵、值班端重复告警过多。
处理：冷却 + 更高告警阈值。先看ls events | wc -l，不要凭感觉。

9.2 类别名混乱

cls_id=0直接当缺陷名，解释层会不知所云。
处理：显式class_names映射，并在日志里打印映射后的defect。

9.3 先接大模型，事件还不稳定

模型会把垃圾输入组织成很像样的废话。
处理：先ls events && cat人工看 20 条，确认字段和类别都对，再接解释。

9.4 截图路径写了但文件不存在

解释或复盘时对不上。
处理：检查cv2.imwrite是否成功；磁盘满时先关截图，只留 JSON。

9.5 把 EventWriter 写死进引擎代码深处

后期换 Ultralytics / TensorRT 版本会很痛。
处理：保持“推理后钩子”这种薄接入。

9.6 多目标同一帧全部写成事件

一帧里 8 个同类框，可能对应同一物理缺陷的多次响应。
处理：同一帧内可先按类别取最高分，或提高冷却；不要无脑全写。

9.7 事件目录与检测工作目录不一致

检测在~/orin-detect跑，事件写到别处，解释脚本找不到。
处理：在配置里写绝对路径，或统一到~/orin-edge-app/events。

10. 验收清单

# 快速抽查最近事件ls-ltevents|headpython3 -<<'PY' import json,glob files=sorted(glob.glob('events/*.json')) print('count=',len(files)) if files: print(json.load(open(files[-1],encoding='utf-8'))) PY

11. 小结

把端侧 YOLO 检测结果接到大模型告警，本质是补上检测与大模型之间的接口层：

1. 统一字段
2. 阈值 + 白名单过滤
3. 冷却去重
4. 写成 JSON，并可选取景
5. 再交给解释脚本

先让检测结果变成可消费事件，端侧大模型才接得上；否则再强的模型也只是对着空气聊天。

12. 相关阅读

• Orin 上目标检测 + 大模型做端侧应用
• Orin 上目标检测最小闭环
• Orin 上跑本地大模型，适合什么场景
• 本地大模型跑通了，为什么还是不好用

相关链接：

• Ultralytics YOLOv8
• Ollama OpenAI compatibility

如果这篇对你有帮助，欢迎点赞、收藏，也欢迎关注后续更新。