conf.txt

@@ -14,6 +14,8 @@
max_tokens = 200
threadnum = 4
detectnum = 1
qwendescription=1
qwenwarning=1
qwenaddr = /home/debian/Qwen2.5-VL-3B-Instruct-GPTQ-Int4
cuda = 1
logaddr=/home/debian/logs

 qwen_detect.py

@@ -9,7 +9,8 @@
from pymilvus import connections, Collection
from logging.handlers import RotatingFileHandler
import get_mem
from multiprocessing import Process
from qwen_thread_description import qwen_thread_description


class ThreadPool:
    def __init__(self):
@@ -60,8 +61,10 @@
        # 加载集合
        self.collection = Collection(name="smartobject")
        self.collection.load()
        #创建qwen线程池
        self.pool = qwen_thread(self.config,self.logger)
        #创建qwen线程池--预警线程池
        self.poolWarning = qwen_thread(self.config,self.logger)
        # 创建qwen线程池--图片描述等
        self.poolDescription= qwen_thread_description(self.config, self.logger)
        #是否更新
        self._isupdate = False

@@ -88,12 +91,69 @@
            self.threads[camera_id] = t
            return t

    # 启动线程任务
    # 启动线程任务--预警
    def workerWarning(self, camera_id):
            while True:
                try:
                    res_a = self.collection.query(
                        expr=f"is_desc == 0 and video_point_id=={camera_id}",
                        output_fields=["id", "zh_desc_class", "text_vector", "bounding_box", "video_point_name",
                                       "task_id",
                                       "task_name", "event_level_id", "event_level_name",
                                       "video_point_id", "detect_num", "is_waning", "is_desc", "waning_value",
                                       "rule_id",
                                       "detect_id", "knowledge_id", "suggestion", "risk_description",
                                       "detect_time", "image_path", "image_desc_path", "video_path"],
                        consistency_level="Strong",
                        order_by_field="id",  # 按id字段排序
                        order_by_type="desc"  # 降序排列
                    )
                    # 读取共享内存中的图片
                    # image_id = get_mem.smem_read_frame_qianwen(camera_id)
                    if len(res_a) > 0:
                        # sorted_results = sorted(res_a, key=itemgetter("id"), reverse=True)
                        # res = sorted_results[0]
                        res = max(res_a, key=itemgetter("id"))
                        self.collection.delete(f"id == {res['id']}")
                        # 数据组
                        data = {
                            "event_level_id": res['event_level_id'],  # event_level_id
                            "event_level_name": res['event_level_name'],  # event_level_id
                            "rule_id": res["rule_id"],
                            "video_point_id": res['video_point_id'],  # video_point_id
                            "video_point_name": res['video_point_name'],
                            "is_waning": 0,
                            "is_desc":4,#预警中状态
                            "zh_desc_class": res['zh_desc_class'],
                            "bounding_box": res['bounding_box'],  # bounding_box
                            "task_id": res['task_id'],  # task_id
                            "task_name": res['task_name'],  # task_id
                            "detect_id": res['detect_id'],  # detect_id
                            "detect_time": res['detect_time'],  # detect_time
                            "detect_num": res['detect_num'],
                            "waning_value": res['waning_value'],
                            "image_path": res['image_path'],  # image_path
                            "image_desc_path": res['image_desc_path'],  # image_desc_path
                            "video_path": res['video_path'],
                            "text_vector": res['text_vector'],
                            "risk_description": res['risk_description'],
                            "suggestion": res['suggestion'],
                            "knowledge_id": res['knowledge_id']
                        }
                        # 保存到milvus
                        image_id = self.collection.insert(data).primary_keys
                        res['id'] = image_id[0]
                        self.poolWarning.submit(res)
                    time_sel.sleep(0.01)
                except Exception as e:
                    logging.info(f"{camera_id}线程错误:{e}")

    # 启动线程任务--生成图片描述等
    def worker(self, camera_id):
        while True:
            try:
                res_a = self.collection.query(
                    expr=f"is_desc == 0 and video_point_id=={camera_id}",
                    expr=f"is_desc == 5 and video_point_id=={camera_id}",
                    output_fields=["id", "zh_desc_class", "text_vector", "bounding_box", "video_point_name", "task_id",
                                   "task_name", "event_level_id", "event_level_name",
                                   "video_point_id", "detect_num", "is_waning", "is_desc", "waning_value", "rule_id",
@@ -117,9 +177,9 @@
                        "rule_id": res["rule_id"],
                        "video_point_id": res['video_point_id'],  # video_point_id
                        "video_point_name": res['video_point_name'],
                        "is_waning": 0,
                        "is_waning": res['is_waning'],
                        "is_desc": 1,
                        "zh_desc_class": "",
                        "zh_desc_class": res['zh_desc_class'],
                        "bounding_box": res['bounding_box'],  # bounding_box
                        "task_id": res['task_id'],  # task_id
                        "task_name": res['task_name'],  # task_id
@@ -138,7 +198,7 @@
                    # 保存到milvus
                    image_id = self.collection.insert(data).primary_keys
                    res['id'] = image_id[0]
                    self.pool.submit(res)
                    self.poolDescription.submit(res)
                time_sel.sleep(0.01)
            except Exception as e:
                logging.info(f"{camera_id}线程错误:{e}")
@@ -181,11 +241,11 @@

    def shutdown_all(self) -> None:
        """清理所有线程"""
        with self.lock:
            for camera_id, thread in list(self.threads.items()):
                if thread.is_alive():
                    thread.join(timeout=1)
                del self.threads[camera_id]
        for camera_id, thread in list(self.threads.items()):
            if thread.is_alive():
                thread.join(timeout=1)
            del self.threads[camera_id]


    #获取任务
    def getTaskconf(self,isupdate):
@@ -227,6 +287,9 @@
                        thread = pool.threads.get(camera.get("camera_id"))
                        if not thread:
                            logging.info(f"开始创建{camera.get('camera_id')}线程")
                            #先对数据进行预警
                            pool.safe_start(pool.workerWarning, camera.get('camera_id'))
                            #在生成图片描述、处理建议等信息
                            pool.safe_start(pool.worker, camera.get('camera_id'))
                            logging.info(f"{camera.get('camera_id')}线程创建完毕")

@@ -238,6 +301,7 @@
                        thread = pool.threads.get(camera.get("camera_id"))
                        if not thread:
                            logging.info(f"开始创建{camera.get('camera_id')}线程")
                            pool.safe_start(pool.workerWarning, camera.get('camera_id'))
                            pool.safe_start(pool.worker, camera.get('camera_id'))
                            logging.info(f"{camera.get('camera_id')}线程创建完毕")


 qwen_thread.py

@@ -1,18 +1,16 @@
import logging
import time
from concurrent.futures import ThreadPoolExecutor
import threading

import torch
from PIL import Image
from pymilvus import connections, Collection
from datetime import datetime
import os
import requests
import asyncio
import logging
import re
from logging.handlers import RotatingFileHandler

from qwen_vl_utils import process_vision_info
from transformers import AutoModelForVision2Seq, AutoProcessor


@@ -29,19 +27,22 @@
        # 加载集合
        self.collection = Collection(name="smartobject")
        self.collection.load()

        if config.get('cuda') is None or config.get('cuda') == '0':
            self.device = f"cuda"
        else:
            self.device = f"cuda:{config.get('cuda')}"
        self.model_pool = []
        self.lock_pool = [threading.Lock() for _ in range(int(config.get("threadnum")))]
        for i in range(int(config.get("threadnum"))):
        self.lock_pool = [threading.Lock() for _ in range(int(config.get("qwenwarning")))]
        for i in range(int(config.get("qwenwarning"))):
            model = AutoModelForVision2Seq.from_pretrained(
                config.get("qwenaddr"),
                device_map=f"cuda:{config.get('cuda')}",
                device_map=self.device,
                trust_remote_code=True,
                use_safetensors=True,
                torch_dtype=torch.float16

            ).eval()
            model = model.to(f"cuda:{config.get('cuda')}")
            model = model.to(self.device)
            self.model_pool.append(model)

        # 共享的处理器 (线程安全)
@@ -70,36 +71,15 @@

    def tark_do(self,res,ragurl,rag_mode,max_tokens):
        try:
            # 1. 从集合A获取向量和元数据
            is_waning = 0
            is_desc = 2

            # 生成图片描述
            ks_time = datetime.now()
            desc = self.image_desc(res)
            desc_time = datetime.now() - ks_time
            current_time = datetime.now()

            risk_description = ""
            suggestion = ""
            # 图片描述生成成功
            if desc:
                is_desc = 2
                # 调用规则匹配方法,判断是否预警
                is_waning = self.image_rule_chat(desc, res['waning_value'], ragurl,rag_mode,max_tokens)
                # 如果预警,则生成隐患描述和处理建议
                if is_waning == 1:
                    # 获取规章制度数据
                    filedata = self.get_filedata(res['waning_value'],res['suggestion'], ragurl)
                    # 生成隐患描述
                    risk_description = self.image_rule_chat_with_detail(filedata, res['waning_value'], ragurl,rag_mode,max_tokens)
                    # 生成处理建议
                    suggestion = self.image_rule_chat_suggestion(filedata, res['waning_value'], ragurl,rag_mode,max_tokens)
                    self.logger.info(
                        f"{res['video_point_id']}执行完毕：{res['id']}:是否预警{is_waning},安全隐患：{risk_description}\n处理建议：{suggestion}")
            else:
                is_desc = 3

            # 数据组
            # 调用规则匹配方法,判断是否预警
            is_waning = self.image_rule(res)
            self.logger.info(f"预警规则规则规则is_waning：{is_waning}")
            #更新数据的预警结果与数据预警状态
            data = {
                "event_level_id": res['event_level_id'],  # event_level_id
                "event_level_name": res['event_level_name'],  # event_level_id
@@ -107,8 +87,8 @@
                "video_point_id": res['video_point_id'],  # video_point_id
                "video_point_name": res['video_point_name'],
                "is_waning": is_waning,
                "is_desc": is_desc,
                "zh_desc_class": desc,  # text_vector
                "is_desc": 5,  #改为已经预警
                "zh_desc_class": res['zh_desc_class'],  # text_vector
                "bounding_box": res['bounding_box'],  # bounding_box
                "task_id": res['task_id'],  # task_id
                "task_name": res['task_name'],  # task_id
@@ -127,231 +107,64 @@
            self.collection.delete(f"id == {res['id']}")
            # 保存到milvus
            image_id = self.collection.insert(data).primary_keys
            data = {
                "id": str(image_id[0]),
                "video_point_id": res['video_point_id'],
                "video_path": res["video_point_name"],
                "zh_desc_class": desc,
                "detect_time": res['detect_time'],
                "image_path": f"{res['image_path']}",
                "task_name": res["task_name"],
                "event_level_name": res["event_level_name"],
                "rtsp_address": f"{res['video_path']}"
            }
            # 调用rag
            asyncio.run(self.insert_json_data(ragurl, data))
            rag_time = datetime.now() - current_time
            self.logger.info(f"{res['video_point_id']}执行完毕：{image_id}运行结束总体用时:{datetime.now() - ks_time},图片描述用时{desc_time}，RAG用时{rag_time}")
            res['id'] = image_id[0]
            self.logger.info(f"{res['video_point_id']}预警执行完毕：{image_id}运行结束总体用时:{datetime.now() - ks_time}")
            return None
        except Exception as e:
            self.logger.info(f"线程：执行模型解析时出错::{e}")
            return 0

    def image_desc(self, res_data):
    def image_rule(self, res_data):
        self.logger.info(f"预警规则规则规则等级分类就是打裂缝多少积分")
        try:
            model, lock = self._acquire_model()

            image = Image.open(res_data['image_desc_path']).convert("RGB").resize((600, 600), Image.Resampling.LANCZOS)

            messages = [
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image",
                        },
                        {"type": "text", "text": "请详细描述图片中的目标信息及特征。返回格式为整段文字描述"},
                        {"type": "image", "image": image},
                        {"type": "text", "text": f"请检测图片中是否{res_data['waning_value']}?请回答yes或no"},
                    ],
                }
            ]

            # Preparation for inference
            text = self.processor.apply_chat_template(
                messages, add_generation_prompt=True
                messages, tokenize=False, add_generation_prompt=True
            )
            image_inputs, video_inputs = process_vision_info(messages)
            inputs = self.processor(
                text=[text],
                images=[image],
                images=image_inputs,
                videos=video_inputs,
                padding=True,
                return_tensors="pt",
            )
            inputs = inputs.to(model.device)
            with torch.inference_mode(),torch.cuda.amp.autocast():
                outputs = model.generate(**inputs,max_new_tokens=200)

            with torch.no_grad():
                outputs = model.generate(**inputs, max_new_tokens=10)
            generated_ids = outputs[:, len(inputs.input_ids[0]):]
            image_text = self.processor.batch_decode(
                generated_ids, skip_special_tokens=True, clean_up_tokenization_spaces=True
            )

            image_des = (image_text[0]).strip()
            return image_des
            upper_text = image_des.upper()
            self.logger.info(f"预警规则规则规则：{upper_text}")
            if "YES" in upper_text:
                return 1
            else:
                return 0
        except Exception as e:
            self.logger.info(f"线程：执行图片描述时出错:{e}")
            return 0
        finally:
            # 4. 释放模型
            self._release_model(model)
            torch.cuda.empty_cache()

    def get_rule(self,ragurl):
        try:
            rule_text = None
            search_data = {
                "collection_name": "smart_rule",
                "query_text": "",
                "search_mode": "hybrid",
                "limit": 100,
                "weight_dense": 0.7,
                "weight_sparse": 0.3,
                "filter_expr": "",
                "output_fields": ["text"]
            }
            response = requests.post(ragurl + "/search", json=search_data)
            results = response.json().get('results')
            rule_text = ""
            ruleid = 1
            for rule in results:
                if rule['score'] >= 0:
                    rule_text = rule_text + str(ruleid) + ". " + rule['entity'].get('text') + ";\n"
                    ruleid = ruleid + 1
            # self.logger.info(len(rule_text))
            else:
                self.logger.info(f"线程：执行获取规则时出错:{response}")
            return rule_text
        except Exception as e:
            self.logger.info(f"线程：执行获取规则时出错:{e}")
            return None

    def image_rule_chat(self, image_des,rule_text, ragurl, rag_mode,max_tokens):
        try:
            content = (
                f"图片描述内容为：\n{image_des}\n规则内容：\n{rule_text}。\n请验证图片描述中是否有不符合规则的内容，不进行推理和think。返回结果格式为[xxx符合的规则id]，如果没有返回[]")
            #self.logger.info(len(content))
            search_data = {
                "prompt": "",
                "messages": [
                    {
                        "role": "user",
                        "content": content
                    }
                ],
                "llm_name": rag_mode,
                "stream": False,
                "gen_conf": {
                    "temperature": 0.7,
                    "max_tokens": max_tokens
                }
            }
            response = requests.post(ragurl + "/chat", json=search_data)
            results = response.json().get('data')
            # self.logger.info(results)
            ret = re.sub(r'<think>.*?</think>', '', results, flags=re.DOTALL)
            ret = ret.replace(" ", "").replace("\t", "").replace("\n", "")
            is_waning = 0
            if len(ret) > 2:
                is_waning = 1
            return is_waning
        except Exception as e:
            self.logger.info(f"线程：执行规则匹配时出错:{image_des, rule_text, ragurl, rag_mode,e}")
            return None

    # 隐患描述
    def image_rule_chat_with_detail(self,filedata, rule_text, ragurl, rag_mode,max_tokens):
        # API调用
        content = (
            f"规章制度为：[{filedata}]\n违反内容为：[{rule_text}]\n请查询违反内容在规章制度中的安全隐患，不进行推理和think，返回简短的文字信息")
        # self.logger.info(len(content))
        search_data = {
            "prompt": "",
            "messages": [
                {
                    "role": "user",
                    "content": content
                }
            ],
            "llm_name": rag_mode,
            "stream": False,
            "gen_conf": {
                "temperature": 0.7,
                "max_tokens": max_tokens
            }
        }
        response = requests.post(ragurl + "/chat", json=search_data)
        # 从json提取data字段内容
        ret = response.json()["data"]
        # 移除<think>标签和内容
        ret = re.sub(r'<think>.*?</think>', '', ret, flags=re.DOTALL)
        # 字符串清理,移除空格,制表符,换行符,星号
        ret = ret.replace(" ", "").replace("\t", "").replace("\n", "").replace("**","")
        #print(f"安全隐患:{ret}")
        return ret
    #处理建议
    def image_rule_chat_suggestion(self,filedata, rule_text, ragurl, rag_mode,max_tokens):
        # 请求内容
        content = (
            f"规章制度为：[{filedata}]\n违反内容为：[{rule_text}]\n请查询违反内容在规章制度中的处理建议，不进行推理和think，返回简短的文字信息")
        response = requests.post(
            # ollama地址
            url=f"{ragurl}/chat",
            json={
                # 指定模型
                "llm_name": rag_mode,
                "messages": [
                    {"role": "user", "content": content}
                ],
                "stream": False,  # 关闭流式输出
                "gen_conf": {
                    "temperature": 0.7,
                    "max_tokens": max_tokens
                }
            }
        )
        # 从json提取data字段内容
        ret = response.json()["data"]
        # 移除<think>标签和内容
        ret = re.sub(r'<think>.*?</think>', '', ret, flags=re.DOTALL)
        # 字符串清理,移除空格,制表符,换行符,星号
        ret = ret.replace(" ", "").replace("\t", "").replace("\n", "").replace("**","")
        #print(f"处理建议:{ret}")
        return ret

    # RAG服务发送请求,获取知识库内容
    def get_filedata(self, searchtext,filter_expr, ragurl):
        search_data = {
            # 知识库集合
            "collection_name": "smart_knowledge",
            # 查询文本
            "query_text": searchtext,
            # 搜索模式
            "search_mode": "hybrid",
            # 最多返回结果
            "limit": 10,
            # 调密向量搜索权重
            "weight_dense": 0.9,
            # 稀疏向量搜索权重
            "weight_sparse": 0.1,
            # 空字符串
            "filter_expr": f"docnm_kwd in {filter_expr}",
            # 只返回 text 字段
            "output_fields": ["text"]
        }
        #print(search_data)
        # 向 ragurl + "/search" 端点发送POST请求
        response = requests.post(ragurl + "/search", json=search_data)
        # 从响应中获取'results'字段
        results = response.json().get('results')
        # 初始化 text
        text = ""
        # 遍历所有结果规则(rule)，将每条规则的'entity'中的'text'字段取出.
        for rule in results:
            text = text + rule['entity'].get('text') + ";\n"
        #print(text)
        return text

    async def insert_json_data(self, ragurl, data):
        try:
            data = {'collection_name': "smartrag", "data": data, "description": ""}
            requests.post(ragurl + "/insert_json_data", json=data, timeout=(0.3, 0.3))
            #self.logger.info(f"调用录像服务:{ragurl, data}")
        except Exception as e:
            #self.logger.info(f"{self._thread_name}线程：调用录像时出错:地址：{ragurl}：{e}")
            return

    def _release_semaphore(self, future):
        self.semaphore.release()
        #self.logger.info(f"释放线程 (剩余空闲: {self.semaphore._value}/{self.max_workers})")
@@ -378,29 +191,4 @@
                self.lock_pool[i].release()
                break


    def remove_duplicate_lines(self,text):
        seen = set()
        result = []
        for line in text.split('。'):  # 按句号分割
            if line.strip() and line not in seen:
                seen.add(line)
                result.append(line)
        return '。'.join(result)
    def remove_duplicate_lines_d(self,text):
        seen = set()
        result = []
        for line in text.split(','):  # 按句号分割
            if line.strip() and line not in seen:
                seen.add(line)
                result.append(line)
        return '。'.join(result)
    def remove_duplicate_lines_n(self,text):
        seen = set()
        result = []
        for line in text.split('\n'):  # 按句号分割
            if line.strip() and line not in seen:
                seen.add(line)
                result.append(line)
        return '。'.join(result)


 qwen_thread_description.py

New file
@@ -0,0 +1,464 @@
import logging
import time
from concurrent.futures import ThreadPoolExecutor
import threading
import torch
from PIL import Image
from pymilvus import connections, Collection
from datetime import datetime
import requests
import asyncio
import re

from qwen_vl_utils import process_vision_info
from transformers import AutoModelForVision2Seq, AutoProcessor


class qwen_thread_description:
    def __init__(self, config,logger):
        self.config = config
        self.max_workers = int(config.get("threadnum"))
        self.executor = ThreadPoolExecutor(max_workers=int(config.get("threadnum")))
        self.semaphore = threading.Semaphore(int(config.get("threadnum")))
        self.logger = logger

        # 初始化Milvus集合
        connections.connect("default", host=config.get("milvusurl"), port=config.get("milvusport"))
        # 加载集合
        self.collection = Collection(name="smartobject")
        self.collection.load()
        if config.get('cuda') is None or config.get('cuda') == '0':
            self.device = f"cuda"
        else:
            self.device = f"cuda:{config.get('cuda')}"
        self.model_pool = []
        self.lock_pool = [threading.Lock() for _ in range(int(config.get("qwendescription")))]
        for i in range(int(config.get("qwendescription"))):
            model = AutoModelForVision2Seq.from_pretrained(
                config.get("qwenaddr"),
                device_map=self.device,
                trust_remote_code=True,
                use_safetensors=True,
                torch_dtype=torch.float16

            ).eval()
            model = model.to(self.device)
            self.model_pool.append(model)

        # 共享的处理器 (线程安全)
        self.processor = AutoProcessor.from_pretrained(config.get("qwenaddr"), use_fast=True)


    def submit(self,res_a):
        # 尝试获取信号量（非阻塞）
        acquired = self.semaphore.acquire(blocking=False)

        if not acquired:
            #self.logger.info(f"线程池已满，等待空闲线程... (当前活跃: {self.max_workers - self.semaphore._value}/{self.max_workers})")
            # 阻塞等待直到有可用线程
            self.semaphore.acquire(blocking=True)

        future = self.executor.submit(self._wrap_task, res_a)
        future.add_done_callback(self._release_semaphore)
        return future

    def _wrap_task(self, res_a):
        try:
            self.tark_do(res_a, self.config.get("ragurl"), self.config.get("ragmode"), self.config.get("max_tokens"))
        except Exception as e:
            self.logger.info(f"处理出错: {e}")
            raise

    def tark_do(self,res,ragurl,rag_mode,max_tokens):
        try:
            # 1. 从集合A获取向量和元数据
            is_waning = 0
            is_desc = 2
            # 生成图片描述
            ks_time = datetime.now()

            risk_description = ""
            suggestion = ""
            # 调用规则匹配方法,判断是否预警
            is_waning = res['is_waning']
            self.logger.info(f"预警规则规则规则is_waning：{is_waning}")
            # 如果预警,则生成隐患描述和处理建议
            if is_waning == 1:
                # 获取规章制度数据
                filedata = self.get_filedata(res['waning_value'],res['suggestion'], ragurl)
                # 生成隐患描述
                risk_description = self.image_rule_chat_with_detail(filedata, res['waning_value'], ragurl,rag_mode,max_tokens)
                # 生成处理建议
                suggestion = self.image_rule_chat_suggestion(filedata, res['waning_value'], ragurl,rag_mode,max_tokens)
                #self.logger.info(f"{res['video_point_id']}执行完毕：{res['id']}:是否预警{is_waning},安全隐患：{risk_description}\n处理建议：{suggestion}")
                # 数据组

            desc_time = datetime.now() - ks_time
            current_time = datetime.now()
            # 图片描述生成成功
            desc = self.image_desc(res)
            if desc:
                is_desc = 2
            else:
                is_desc = 3
            # 数据组
            data = {
                "event_level_id": res['event_level_id'],  # event_level_id
                "event_level_name": res['event_level_name'],  # event_level_id
                "rule_id": res["rule_id"],
                "video_point_id": res['video_point_id'],  # video_point_id
                "video_point_name": res['video_point_name'],
                "is_waning": is_waning,
                "is_desc": is_desc,
                "zh_desc_class": desc,  # text_vector
                "bounding_box": res['bounding_box'],  # bounding_box
                "task_id": res['task_id'],  # task_id
                "task_name": res['task_name'],  # task_id
                "detect_id": res['detect_id'],  # detect_id
                "detect_time": res['detect_time'],  # detect_time
                "detect_num": res['detect_num'],
                "waning_value": res['waning_value'],
                "image_path": res['image_path'],  # image_path
                "image_desc_path": res['image_desc_path'],  # image_desc_path
                "video_path": res['video_path'],
                "text_vector": res['text_vector'],
                "risk_description": risk_description,
                "suggestion": suggestion,
                "knowledge_id": res['knowledge_id']
            }
            self.collection.delete(f"id == {res['id']}")
            # 保存到milvus
            image_id = self.collection.insert(data).primary_keys
            data = {
                "id": str(image_id[0]),
                "video_point_id": res['video_point_id'],
                "video_path": res["video_point_name"],
                "zh_desc_class": desc,
                "detect_time": res['detect_time'],
                "image_path": f"{res['image_path']}",
                "task_name": res["task_name"],
                "event_level_name": res["event_level_name"],
                "rtsp_address": f"{res['video_path']}"
            }
            # 调用rag
            asyncio.run(self.insert_json_data(ragurl, data))
            rag_time = datetime.now() - current_time
            self.logger.info(f"{res['video_point_id']}执行完毕：{image_id}运行结束总体用时:{datetime.now() - ks_time},RAG用时{desc_time},图片描述用时{rag_time}")
            if is_waning == 1:
                self.logger.info(f"{res['video_point_id']}执行完毕：{image_id},图片描述：{desc}\n隐患：{risk_description}\n建议：{suggestion}")
        except Exception as e:
            self.logger.info(f"线程：执行模型解析时出错::{e}")
            return 0

    def image_desc(self, res_data):
        try:
            model, lock = self._acquire_model()

            image = Image.open(res_data['image_desc_path']).convert("RGB").resize((600, 600), Image.Resampling.LANCZOS)
            messages = [
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image",
                        },
                        {"type": "text", "text": "请详细描述图片中的目标信息及特征。返回格式为整段文字描述"},
                    ],
                }
            ]
            # Preparation for inference
            text = self.processor.apply_chat_template(
                messages, add_generation_prompt=True
            )
            inputs = self.processor(
                text=[text],
                images=[image],
                padding=True,
                return_tensors="pt",
            )
            inputs = inputs.to(model.device)
            with torch.inference_mode(), torch.amp.autocast(device_type=self.device, dtype=torch.float16):
                outputs = model.generate(**inputs,max_new_tokens=200,do_sample=False,num_beams=1,temperature=0.1)
            generated_ids = outputs[:, len(inputs.input_ids[0]):]
            image_text = self.processor.batch_decode(
                generated_ids, skip_special_tokens=True, clean_up_tokenization_spaces=True
            )
            image_des = (image_text[0]).strip()
            #self.logger.info(f"{res_data['video_point_id']}:{res_data['id']}:{res_data['detect_time']}:{image_des}")
            return image_des
        except Exception as e:
            self.logger.info(f"线程：执行图片描述时出错:{e}")
        finally:
            # 4. 释放模型
            self._release_model(model)
            torch.cuda.empty_cache()

    def image_rule(self, res_data):
        self.logger.info(f"预警规则规则规则等级分类就是打裂缝多少积分")
        try:
            model, lock = self._acquire_model()
            image = Image.open(res_data['image_desc_path']).convert("RGB").resize((600, 600), Image.Resampling.LANCZOS)

            messages = [
                {
                    "role": "user",
                    "content": [
                        {"type": "image", "image": image},
                        {"type": "text", "text": f"请检测图片中{res_data['waning_value']}?请回答yes或no"},
                    ],
                }
            ]

            # Preparation for inference
            text = self.processor.apply_chat_template(
                messages, tokenize=False, add_generation_prompt=True
            )
            image_inputs, video_inputs = process_vision_info(messages)
            inputs = self.processor(
                text=[text],
                images=image_inputs,
                videos=video_inputs,
                padding=True,
                return_tensors="pt",
            )
            inputs = inputs.to(model.device)

            with torch.no_grad():
                outputs = model.generate(**inputs, max_new_tokens=10)
            generated_ids = outputs[:, len(inputs.input_ids[0]):]
            image_text = self.processor.batch_decode(
                generated_ids, skip_special_tokens=True, clean_up_tokenization_spaces=True
            )

            image_des = (image_text[0]).strip()
            upper_text = image_des.upper()
            self.logger.info(f"预警规则规则规则：{upper_text}")
            if "YES" in upper_text:
                return 1
            else:
                return 0
        except Exception as e:
            self.logger.info(f"线程：执行图片描述时出错:{e}")
            return 0
        finally:
            # 4. 释放模型
            self._release_model(model)
            torch.cuda.empty_cache()

    def get_rule(self,ragurl):
        try:
            rule_text = None
            search_data = {
                "collection_name": "smart_rule",
                "query_text": "",
                "search_mode": "hybrid",
                "limit": 100,
                "weight_dense": 0.7,
                "weight_sparse": 0.3,
                "filter_expr": "",
                "output_fields": ["text"]
            }
            response = requests.post(ragurl + "/search", json=search_data)
            results = response.json().get('results')
            rule_text = ""
            ruleid = 1
            for rule in results:
                if rule['score'] >= 0:
                    rule_text = rule_text + str(ruleid) + ". " + rule['entity'].get('text') + ";\n"
                    ruleid = ruleid + 1
            # self.logger.info(len(rule_text))
            else:
                self.logger.info(f"线程：执行获取规则时出错:{response}")
            return rule_text
        except Exception as e:
            self.logger.info(f"线程：执行获取规则时出错:{e}")
            return None

    def image_rule_chat(self, image_des,rule_text, ragurl, rag_mode,max_tokens):
        try:
            content = (
                f"图片描述内容为：\n{image_des}\n规则内容：\n{rule_text}。\n请验证图片描述中是否有不符合规则的内容，不进行推理和think。返回结果格式为[xxx符合的规则id]，如果没有返回[]")
            #self.logger.info(len(content))
            search_data = {
                "prompt": "",
                "messages": [
                    {
                        "role": "user",
                        "content": content
                    }
                ],
                "llm_name": rag_mode,
                "stream": False,
                "gen_conf": {
                    "temperature": 0.7,
                    "max_tokens": max_tokens
                }
            }
            response = requests.post(ragurl + "/chat", json=search_data)
            results = response.json().get('data')
            ret = re.sub(r'<think>.*?</think>', '', results, flags=re.DOTALL)
            ret = ret.replace(" ", "").replace("\t", "").replace("\n", "")
            #self.logger.info(f"{rule_text}:{ret}")
            is_waning = 0
            if len(ret) > 2:
                is_waning = 1
            return is_waning
        except Exception as e:
            self.logger.info(f"线程：执行规则匹配时出错:{image_des, rule_text, ragurl, rag_mode,e}")
            return None

    # 隐患描述
    def image_rule_chat_with_detail(self,filedata, rule_text, ragurl, rag_mode,max_tokens):
        # API调用
        content = (
            f"规章制度为：[{filedata}]\n违反内容为：[{rule_text}]\n请查询违反内容在规章制度中的安全隐患，不进行推理和think，返回简短的文字信息")
        # self.logger.info(len(content))
        search_data = {
            "prompt": "",
            "messages": [
                {
                    "role": "user",
                    "content": content
                }
            ],
            "llm_name": rag_mode,
            "stream": False,
            "gen_conf": {
                "temperature": 0.7,
                "max_tokens": max_tokens
            }
        }
        #self.logger.info(content)
        response = requests.post(ragurl + "/chat", json=search_data)
        # 从json提取data字段内容
        ret = response.json()["data"]
        # 移除<think>标签和内容
        ret = re.sub(r'<think>.*?</think>', '', ret, flags=re.DOTALL)
        # 字符串清理,移除空格,制表符,换行符,星号
        ret = ret.replace(" ", "").replace("\t", "").replace("\n", "").replace("**","")
        #print(f"安全隐患:{ret}")
        return ret
    #处理建议
    def image_rule_chat_suggestion(self,filedata, rule_text, ragurl, rag_mode,max_tokens):
        # 请求内容
        content = (
            f"规章制度为：[{filedata}]\n违反内容为：[{rule_text}]\n请查询违反内容在规章制度中的处理建议，不进行推理和think，返回简短的文字信息")
        response = requests.post(
            # ollama地址
            url=f"{ragurl}/chat",
            json={
                # 指定模型
                "llm_name": rag_mode,
                "messages": [
                    {"role": "user", "content": content}
                ],
                "stream": False,  # 关闭流式输出
                "gen_conf": {
                    "temperature": 0.7,
                    "max_tokens": max_tokens
                }
            }
        )
        # 从json提取data字段内容
        ret = response.json()["data"]
        # 移除<think>标签和内容
        ret = re.sub(r'<think>.*?</think>', '', ret, flags=re.DOTALL)
        # 字符串清理,移除空格,制表符,换行符,星号
        ret = ret.replace(" ", "").replace("\t", "").replace("\n", "").replace("**","")
        #print(f"处理建议:{ret}")
        return ret

    # RAG服务发送请求,获取知识库内容
    def get_filedata(self, searchtext,filter_expr, ragurl):
        search_data = {
            # 知识库集合
            "collection_name": "smart_knowledge",
            # 查询文本
            "query_text": searchtext,
            # 搜索模式
            "search_mode": "hybrid",
            # 最多返回结果
            "limit": 10,
            # 调密向量搜索权重
            "weight_dense": 0.9,
            # 稀疏向量搜索权重
            "weight_sparse": 0.1,
            # 空字符串
            "filter_expr": f"docnm_kwd in {filter_expr}",
            # 只返回 text 字段
            "output_fields": ["text"]
        }
        #print(search_data)
        # 向 ragurl + "/search" 端点发送POST请求
        response = requests.post(ragurl + "/search", json=search_data)
        # 从响应中获取'results'字段
        results = response.json().get('results')
        # 初始化 text
        text = ""
        # 遍历所有结果规则(rule)，将每条规则的'entity'中的'text'字段取出.
        for rule in results:
            text = text + rule['entity'].get('text') + ";\n"
        #print(text)
        return text

    async def insert_json_data(self, ragurl, data):
        try:
            data = {'collection_name': "smartrag", "data": data, "description": ""}
            requests.post(ragurl + "/insert_json_data", json=data, timeout=(0.3, 0.3))
            #self.logger.info(f"调用录像服务:{ragurl, data}")
        except Exception as e:
            #self.logger.info(f"{self._thread_name}线程：调用录像时出错:地址：{ragurl}：{e}")
            return

    def _release_semaphore(self, future):
        self.semaphore.release()
        #self.logger.info(f"释放线程 (剩余空闲: {self.semaphore._value}/{self.max_workers})")

    def shutdown(self):
        """安全关闭"""
        self.executor.shutdown(wait=False)
        for model in self.model_pool:
            del model
        torch.cuda.empty_cache()

    def _acquire_model(self):
        """从池中获取一个空闲模型 (简单轮询)"""
        while True:
            for i, (model, lock) in enumerate(zip(self.model_pool, self.lock_pool)):
                if lock.acquire(blocking=False):
                    return model, lock
            time.sleep(0.1)  # 避免CPU空转

    def _release_model(self, model):
        """释放模型回池"""
        for i, m in enumerate(self.model_pool):
            if m == model:
                self.lock_pool[i].release()
                break


    def remove_duplicate_lines(self,text):
        seen = set()
        result = []
        for line in text.split('。'):  # 按句号分割
            if line.strip() and line not in seen:
                seen.add(line)
                result.append(line)
        return '。'.join(result)
    def remove_duplicate_lines_d(self,text):
        seen = set()
        result = []
        for line in text.split(','):  # 按句号分割
            if line.strip() and line not in seen:
                seen.add(line)
                result.append(line)
        return '。'.join(result)
    def remove_duplicate_lines_n(self,text):
        seen = set()
        result = []
        for line in text.split('\n'):  # 按句号分割
            if line.strip() and line not in seen:
                seen.add(line)
                result.append(line)
        return '。'.join(result)