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_models
cd /root/yolo_models
wget https://github.com/ultralytics/assets/releases/download/v8.2.0/yolov8n.pt

#!/usr/bin/env python3
"""
农业病虫害检测Skill - 基于YOLOv8+Qwen-VL
支持：单图检测、批量统计、Qwen-VL多模态二次校验
作者：AI小怪兽
"""

import os
import sys
import json
import cv2
import base64
from datetime import datetime
from pathlib import Path
from ultralytics import YOLO
import 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 Key

class 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 report

def 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()

用户：每天凌晨1点，用pest-detector技能检测 /data/orchard/ 目录下所有图片，统计病虫害分布，早上7点推送报表
OpenClaw：已创建定时任务，每天1:00执行批量检测，7:00推送报表

📊 今日虫情统计报告 (2026-03-20)
========================================
总检测照片：1,245张
发现虫害区域：37处
大模型确认次数：42次

📋 虫害类型分布：
- aphid：18处（东区严重）
- red_spider：12处（西区新发）
- powdery_mildew：7处（南区零星）

⚠️ 预警：蚜虫密度超标（>50处），建议立即防治！
========================================

用户：监听 /data/vineyard/ 目录，每新增100张照片就批量检测一次，发现白粉病立即调用Qwen-VL分析严重程度，推送企业微信
OpenClaw：已创建文件监听任务，检测到病害将调用Qwen-VL分析并推送

⚠️ 紧急预警：葡萄园白粉病
发现时间：2026-03-20 14:23:17
照片编号：DJI_20240320_1423.jpg
位置坐标：东经118.23°，北纬34.56°
YOLO置信度：0.92
Qwen-VL确认：已确诊为白粉病，严重程度：初期
防治建议：25%嘧菌酯1500倍液，傍晚喷施

请立即安排喷药防治！

用户：统计过去一个月苹果的生长速度，结合Qwen-VL成熟度分析，预测最佳采摘日期
OpenClaw：正在分析历史数据...根据果实大小、颜色变化和Qwen-VL多模态分析，预计最佳采摘窗口为4月5日-4月12日，预估产量约8.5吨