YOLO+OpenClaw农业检测:三道坎与三招破局,Qwen-VL多模态大模型让果园真智慧

# YOLO+Qwen-VL协同检测流程def enhanced_detection_pipeline(image_path):    # Step1: YOLO快速定位疑似区域    yolo_results = yolo_model(image_path)    # Step2: 对每个疑似区域，裁剪后送Qwen-VL二次校验    for box in yolo_results:        crop = crop_image(image_path, box)        vlm_result = qwen_vl_analyze(crop)        # Step3: 融合判断        if vlm_result["has_pest"] and vlm_result["confidence"] > 0.8:            confirm_pest(box, vlm_result)  # 确诊        elif vlm_result["has_pest"] and vlm_result["confidence"] > 0.5:            mark_suspicious(box, vlm_result)  # 可疑，标记待人工确认        else:            ignore(box)  # YOLO误报，丢弃

# 安装YOLO依赖pip install ultralytics opencv-python pillow requests# 下载YOLOv8农业专用模型mkdir -p /root/yolo_modelscd /root/yolo_modelswget https://github.com/ultralytics/assets/releases/download/v8.2.0/yolov8n.pt

#!/usr/bin/env python3"""农业病虫害检测Skill - 基于YOLOv8+Qwen-VL支持：单图检测、批量统计、Qwen-VL多模态二次校验作者：AI小怪兽"""import osimport sysimport jsonimport cv2import base64from datetime import datetimefrom pathlib import Pathfrom ultralytics import YOLOimport requests# 全局配置MODEL_PATH = "/root/yolo_models/yolov8n.pt"PEST_CLASSES = ["aphid", "red_spider", "leaf_roller", "powdery_mildew", "正常叶片"]STATS_FILE = "/tmp/pest_stats.json"QWEN_API_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation"QWEN_API_KEY = "your_dashscope_api_key"  # 需替换为实际API Keyclass PestDetector:    def __init__(self):        """初始化YOLO模型"""        self.model = YOLO(MODEL_PATH)        self.stats = self.load_stats()    def load_stats(self):        """加载统计数据"""        if os.path.exists(STATS_FILE):            with open(STATS_FILE, 'r') as f:                return json.load(f)        return {            "total_images": 0,            "pest_regions": 0,            "pest_types": {cls: 0 for cls in PEST_CLASSES},            "daily": {},            "confirmed_by_vlm": 0  # 大模型确认次数        }    def save_stats(self):        """保存统计数据"""        with open(STATS_FILE, 'w') as f:            json.dump(self.stats, f, indent=2)    def call_qwen_vl(self, image_path, bbox):        """        调用Qwen-VL多模态大模型进行二次校验        """        try:            # 裁剪检测区域            img = cv2.imread(image_path)            x1, y1, x2, y2 = map(int, bbox)            crop = img[y1:y2, x1:x2]            # 保存临时裁剪图            crop_path = f"/tmp/crop_{Path(image_path).stem}_{datetime.now().timestamp()}.jpg"            cv2.imwrite(crop_path, crop)            # 读取并编码图像            with open(crop_path, 'rb') as f:                img_base64 = base64.b64encode(f.read()).decode('utf-8')            # 构造Qwen-VL请求            headers = {                "Authorization": f"Bearer {QWEN_API_KEY}",                "Content-Type": "application/json"            }            payload = {                "model": "qwen-vl-plus",                "input": {                    "messages": [                        {                            "role": "user",                            "content": [                                {                                    "image": f"data:image/jpeg;base64,{img_base64}"                                },                                {                                    "text": "请判断这张图片是否包含病虫害症状？如果是，请说明具体是什么病虫害，置信度如何。只输出JSON格式：{\"has_pest\": true/false, \"pest_type\": \"病害名称\", \"confidence\": 0-100, \"severity\": \"轻度/中度/重度\"}"                                }                            ]                        }                    ]                }            }            response = requests.post(QWEN_API_URL, headers=headers, json=payload)            result = response.json()            # 解析返回结果            if response.status_code == 200:                content = result.get("output", {}).get("choices", [{}])[0].get("message", {}).get("content", "{}")                # 提取JSON部分                import re                json_match = re.search(r'\{.*\}', content, re.DOTALL)                if json_match:                    vlm_result = json.loads(json_match.group())                    return vlm_result            return {"has_pest": False, "error": "调用失败"}        except Exception as e:            print(f"Qwen-VL调用异常: {e}")            return {"has_pest": False, "error": str(e)}        finally:            # 清理临时文件            if os.path.exists(crop_path):                os.remove(crop_path)    def detect_single(self, image_path, use_vlm=True):        """        单张图片检测，可选择是否使用Qwen-VL二次校验        """        # 执行YOLO推理        results = self.model(image_path, conf=0.25)[0]        # 解析检测结果        detections = []        for box in results.boxes:            cls_id = int(box.cls)            conf = float(box.conf)            x1, y1, x2, y2 = box.xyxy[0].tolist()            # 类别映射            if cls_id < len(PEST_CLASSES):                pest_type = PEST_CLASSES[cls_id]            else:                pest_type = f"unknown_{cls_id}"            detection = {                "type": pest_type,                "confidence": round(conf, 2),                "bbox": [int(x1), int(y1), int(x2), int(y2)]            }            # 如果启用VLM二次校验，且不是"正常叶片"            if use_vlm and pest_type != "正常叶片":                vlm_result = self.call_qwen_vl(image_path, [x1, y1, x2, y2])                detection["vlm_verified"] = vlm_result                if vlm_result.get("has_pest", False):                    self.stats["confirmed_by_vlm"] += 1                    # 如果VLM给出了更精确的类型，可以更新                    if vlm_result.get("pest_type"):                        detection["type"] = vlm_result["pest_type"]                        detection["severity"] = vlm_result.get("severity", "未知")                else:                    # VLM认为不是病虫害，可能是误报                    continue  # 跳过这个检测框            detections.append(detection)            # 更新统计            if detection["type"] in self.stats["pest_types"]:                self.stats["pest_types"][detection["type"]] += 1        # 更新统计        self.stats["total_images"] += 1        if detections:            self.stats["pest_regions"] += len(detections)        # 记录日统计        today = datetime.now().strftime("%Y-%m-%d")        if today not in self.stats["daily"]:            self.stats["daily"][today] = {"images": 0, "pests": 0}        self.stats["daily"][today]["images"] += 1        if detections:            self.stats["daily"][today]["pests"] += len(detections)        self.save_stats()        # 生成标注图像        img = cv2.imread(image_path)        for d in detections:            x1, y1, x2, y2 = d["bbox"]            color = (0, 0, 255) if d.get("vlm_verified", {}).get("has_pest", True) else (255, 0, 0)            cv2.rectangle(img, (x1, y1), (x2, y2), color, 2)            label = f"{d['type']}:{d['confidence']}"            if "severity" in d:                label += f"[{d['severity']}]"            cv2.putText(img, label, (x1, y1-10),                        cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)        output_path = f"/tmp/detected_{Path(image_path).name}"        cv2.imwrite(output_path, img)        return {            "detections": detections,            "count": len(detections),            "image_path": output_path        }    def get_today_report(self):        """生成今日虫情统计报表"""        today = datetime.now().strftime("%Y-%m-%d")        daily = self.stats["daily"].get(today, {"images": 0, "pests": 0})        if daily["images"] == 0:            return "今日暂无检测数据"        report = f"""📊 今日虫情统计报告 ({today}){'='*40}总检测照片：{daily['images']}张发现虫害区域：{daily['pests']}处大模型确认次数：{self.stats.get('confirmed_by_vlm', 0)}次📋 虫害类型分布："""        for pest, count in self.stats["pest_types"].items():            if pest != "正常叶片" and count > 0:                report += f"- {pest}：{count}次\n"        # 添加预警信息（如果某类虫害超过阈值）        if self.stats["pest_types"].get("aphid", 0) > 50:            report += f"\n⚠️ 预警：蚜虫密度超标（>50处），建议立即防治！\n"        report += f"{'='*40}\n生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"        return reportdef main():    """命令行入口"""    if len(sys.argv) < 2:        print("Usage: pest_detector.py <command> [args]")        print("Commands:")        print("  detect <image_path>          - 单张图片检测")        print("  batch <directory>             - 批量检测目录")        print("  report                        - 生成今日报表")        sys.exit(1)    detector = PestDetector()    command = sys.argv[1]    if command == "detect" and len(sys.argv) >= 3:        use_vlm = len(sys.argv) >= 4 and sys.argv[3] == "--use-vlm"        result = detector.detect_single(sys.argv[2], use_vlm=use_vlm)        print(json.dumps(result, indent=2, ensure_ascii=False))    elif command == "batch" and len(sys.argv) >= 3:        results = []        image_exts = ['.jpg', '.jpeg', '.png']        for file in Path(sys.argv[2]).iterdir():            if file.suffix.lower() in image_exts:                result = detector.detect_single(str(file), use_vlm=True)                results.append({"file": file.name, "result": result})        print(json.dumps(results, indent=2, ensure_ascii=False))    elif command == "report":        print(detector.get_today_report())    else:        print("Invalid command")if __name__ == "__main__":    main()