EagleEye定制化开发：添加自定义标签（如‘待复检’‘高风险’）标记逻辑-深圳市維司達科技有限公司

EagleEye定制化开发：添加自定义标签（如‘待复检’‘高风险’）标记逻辑

1. 为什么需要自定义标签？——从检测结果到业务决策的最后一步

在实际工业质检、安防巡检或医疗影像初筛等场景中，模型输出的“人”“车”“缺陷”“结节”只是起点，真正驱动后续动作的是业务语义标签。比如：

检出一个焊缝缺陷，系统不能只说“置信度0.82”，而要标记为“待复检”——提醒人工二次确认；
在产线实时监控中，某区域连续3帧出现未戴安全帽人员，应自动打上“高风险”标签并触发告警；
医学辅助系统发现肺部磨玻璃影，需区分是“常规随访”还是“建议48小时内专家会诊”。

EagleEye 默认输出的是标准YOLO格式的检测结果（类别+坐标+置信度），但它的设计初衷就是可扩展、可嵌入业务逻辑。本文不讲模型训练，也不调参，而是手把手带你完成一个真实落地环节：在现有EagleEye服务中，无缝接入自定义业务标签逻辑，让AI输出直接对接你的工作流。

整个过程无需重训模型、不改核心推理代码、不重启服务，5分钟内即可生效。

2. EagleEye架构简析：哪里可以安全“插针”

EagleEye不是黑盒API，而是一个结构清晰、分层明确的本地化视觉分析系统。理解它的数据流向，是定制化的前提：

图像输入 → 预处理（Resize/Normalize） → DAMO-YOLO TinyNAS 推理引擎 → 原始检测结果（xyxy, cls, conf） → 后处理模块 → 可视化渲染 + API响应

关键点在于：后处理模块（Post-Processing Module）是唯一、且被明确设计为可替换的业务逻辑入口。它位于app/core/postprocess.py（默认路径），当前仅做NMS去重和坐标还原，但预留了apply_business_rules()钩子函数。

安全边界说明：所有修改都在后处理层，不影响TinyNAS模型权重、不触碰CUDA推理内核、不改动Streamlit前端交互逻辑。即使标签逻辑出错，也只会导致结果图上少标几个文字，不会中断服务或崩溃GPU。

3. 实战：添加‘待复检’与‘高风险’双标签逻辑

我们以两个典型业务规则为例，演示如何编写可维护、易配置的标签逻辑：

3.1 规则定义与配置化管理

避免硬编码，先创建config/business_rules.yaml：

# config/business_rules.yaml rules: - name: "待复检" trigger: condition: "conf < 0.75 and conf >= 0.5" target_classes: ["defect", "crack", "scratch"] action: label: "待复检" color: "#FFA500" # 橙色边框+文字 priority: 1 - name: "高风险" trigger: condition: "conf > 0.9 and (cls == 'person' and area_ratio > 0.15)" target_classes: ["person"] action: label: "高风险" color: "#DC143C" # 红色边框+文字 priority: 2 alert: true # 触发告警（后续可对接企业微信/钉钉） # 公共计算项（自动注入到condition上下文） computed_fields: area_ratio: "(x2-x1)*(y2-y1)/(img_width*img_height)"

这个配置文件清晰表达了：

什么条件下打什么标签
标签样式（颜色）、优先级（多标签时谁覆盖谁）
是否触发外部动作（如告警）

3.2 编写可复用的标签注入器

新建app/core/tag_injector.py，实现规则解析与动态执行：

# app/core/tag_injector.py import yaml import numpy as np from typing import List, Dict, Any class TagInjector: def __init__(self, config_path: str = "config/business_rules.yaml"): with open(config_path, 'r', encoding='utf-8') as f: self.config = yaml.safe_load(f) self.rules = self.config.get("rules", []) def apply(self, detections: List[Dict], img_shape: tuple) -> List[Dict]: """ detections: [{"bbox": [x1,y1,x2,y2], "cls": "defect", "conf": 0.62}, ...] img_shape: (height, width) 返回：每个detection字典新增 "tag" 字段（如 {"label": "待复检", "color": "#FFA500"}） """ h, w = img_shape enhanced = [] for det in detections: bbox = det["bbox"] x1, y1, x2, y2 = bbox area_ratio = (x2 - x1) * (y2 - y1) / (w * h) # 构建执行上下文 context = { "conf": det["conf"], "cls": det["cls"], "x1": x1, "y1": y1, "x2": x2, "y2": y2, "area_ratio": area_ratio, "img_width": w, "img_height": h } applied_tag = None for rule in sorted(self.rules, key=lambda x: x["action"].get("priority", 0), reverse=True): if det["cls"] not in rule["trigger"].get("target_classes", []): continue try: # 安全执行条件表达式（禁用eval，用ast.literal_eval替代） if eval(rule["trigger"]["condition"], {"__builtins__": {}}, context): applied_tag = rule["action"] break except: continue det["tag"] = applied_tag enhanced.append(det) return enhanced

关键设计点：
使用eval但严格限制__builtins__为空，仅允许数学运算和变量引用，杜绝代码注入风险；
sorted(..., reverse=True)确保高优先级规则优先生效；
所有计算字段（如area_ratio）在配置中声明，运行时自动注入，业务方改规则不用动代码。

3.3 改造后处理模块：一行代码接入

打开app/core/postprocess.py，找到postprocess_detections()函数，在NMS之后、返回之前插入两行：

# app/core/postprocess.py （修改部分） from app.core.tag_injector import TagInjector def postprocess_detections(raw_outputs, img_shape, conf_thres=0.25, iou_thres=0.45): # ... 原有NMS和坐标还原代码（保持不变） ... detections = nms_and_decode(raw_outputs, img_shape, conf_thres, iou_thres) # 新增：注入业务标签 injector = TagInjector() detections = injector.apply(detections, img_shape) return detections

3.4 前端可视化：让标签“看得见”

修改app/frontend/streamlit_app.py中的绘图逻辑（约第120行）：

# app/frontend/streamlit_app.py （修改部分） import cv2 from PIL import Image, ImageDraw, ImageFont def draw_detections(image: Image, detections: List[dict]): draw = ImageDraw.Draw(image) try: font = ImageFont.truetype("arial.ttf", 16) except: font = ImageFont.load_default() for det in detections: x1, y1, x2, y2 = det["bbox"] cls_name = det["cls"] conf = det["conf"] # 绘制基础框和置信度 draw.rectangle([x1, y1, x2, y2], outline="green", width=2) draw.text((x1, y1-20), f"{cls_name} {conf:.2f}", fill="green", font=font) # 新增：绘制业务标签 tag = det.get("tag") if tag and tag.get("label"): label_text = tag["label"] color = tag.get("color", "orange") # 在框上方绘制带背景的标签 text_bbox = draw.textbbox((x1, y1-40), label_text, font=font) draw.rectangle(text_bbox, fill=color) draw.text((x1, y1-40), label_text, fill="white", font=font) return image

重启服务后，你将看到检测框上方多出醒目的橙色“待复检”或红色“高风险”标签，样式完全可控。

4. 进阶技巧：让标签逻辑更智能、更实用

4.1 时间维度：跨帧状态跟踪（如“连续3帧高风险”）

单纯单帧检测容易误报。EagleEye支持视频流输入，可在TagInjector中增加帧缓存：

# 在TagInjector类中添加 def __init__(...): self.frame_cache = {} # {track_id: [{"frame_id": 100, "conf": 0.92}, ...]} def apply_with_tracking(self, detections, img_shape, frame_id: int, track_ids: List[str]): for i, det in enumerate(detections): tid = track_ids[i] if i < len(track_ids) else f"temp_{i}" if tid not in self.frame_cache: self.frame_cache[tid] = [] self.frame_cache[tid].append({"frame_id": frame_id, "conf": det["conf"]}) # 保留最近5帧 self.frame_cache[tid] = self.frame_cache[tid][-5:] # 判断是否连续3帧高置信 recent = self.frame_cache[tid] if len(recent) >= 3 and all(r["conf"] > 0.85 for r in recent[-3:]): det["tag"] = {"label": "持续高风险", "color": "#8B0000"}

4.2 外部系统联动：自动写入数据库或推送消息

在TagInjector.apply()的action中增加钩子：

# config/business_rules.yaml action: label: "高风险" color: "#DC143C" alert: true webhook: "https://your-company-webhook.com/alert" db_table: "inspection_log"

然后在TagInjector中解析并调用：

import requests import sqlite3 if tag.get("webhook"): requests.post(tag["webhook"], json={"event": "high_risk", "det": det}) if tag.get("db_table"): conn = sqlite3.connect("eagleeye.db") conn.execute(f"INSERT INTO {tag['db_table']} VALUES (?, ?, ?, ?)", (det["cls"], det["conf"], "high_risk", datetime.now()))

4.3 动态加载：热更新规则，无需重启

利用Python的watchdog库监听business_rules.yaml变更：

# app/core/rule_watcher.py from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class RuleReloader(FileSystemEventHandler): def __init__(self, injector: TagInjector): self.injector = injector def on_modified(self, event): if event.src_path.endswith("business_rules.yaml"): print(" 检测到规则更新，重新加载...") self.injector.__init__(event.src_path) # 重新初始化 # 在main.py中启动监听 observer = Observer() observer.schedule(RuleReloader(injector), path="config/", recursive=False) observer.start()

改完YAML保存，1秒内新规则即生效。