""" 基于 OpenRouter Qwen3 VL 8B Instruct 模型的病虫害识别 FastAPI 服务 """ import os import base64 import binascii from io import BytesIO from typing import Optional from dotenv import load_dotenv import httpx from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.responses import JSONResponse from pydantic import BaseModel from PIL import Image # 加载环境变量 load_dotenv() app = FastAPI( title="病虫害识别 API", description="基于 OpenRouter Qwen3 VL 8B Instruct 模型的病虫害识别服务", version="1.0.0" ) # OpenRouter API 配置 OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY") OPENROUTER_API_URL = os.getenv("OPENROUTER_API_URL") MODEL_NAME = os.getenv("OPENROUTER_MODEL") # 可通过环境变量配置 if not OPENROUTER_API_KEY: raise ValueError("请设置 OPENROUTER_API_KEY 环境变量") class Base64ImageRequest(BaseModel): """Base64 图片识别请求模型""" image_base64: str question: Optional[str] = None def encode_image_to_base64(image: Image.Image) -> str: """将 PIL Image 转换为 base64 编码字符串""" buffered = BytesIO() # 转换为 RGB 模式（如果不是的话） if image.mode != "RGB": image = image.convert("RGB") image.save(buffered, format="JPEG") # 确保使用 UTF-8 编码解码 base64 img_str = base64.b64encode(buffered.getvalue()).decode('utf-8') return img_str async def call_openrouter_api(image_base64: str, user_message: str) -> str: """调用 OpenRouter API""" print() print(f"调用 OpenRouter API: {MODEL_NAME}") print(f"用户消息: {user_message} {OPENROUTER_API_KEY}") headers = { "Authorization": f"Bearer {OPENROUTER_API_KEY}", "Content-Type": "application/json", # "HTTP-Referer": "https://github.com/your-repo", # 可选：用于跟踪 # "X-Title": "Pest Disease Identification Service", # 可选：应用名称（使用英文避免编码问题） } payload = { "model": MODEL_NAME, "messages": [ { "role": "system", "content": "你是一位专业的农业病虫害识别专家。请仔细分析用户提供的植物图片，识别可能存在的病虫害问题，并提供详细的诊断信息，包括：1. 植物类型，病虫害类型和名称 2. 严重程度 3. 可能的病因 4. 防治建议。请用中文回答。" }, { "role": "user", "content": [ { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{image_base64}" } }, { "type": "text", "text": user_message } ] } ], "max_tokens": 2000, "temperature": 0.7, "chat_template_kwargs": { "enable_thinking": False } } async with httpx.AsyncClient(timeout=60.0) as client: try: response = await client.post(OPENROUTER_API_URL, json=payload, headers=headers) response.raise_for_status() result = response.json() print(result) if "choices" in result and len(result["choices"]) > 0: return result["choices"][0]["message"]["content"] else: raise HTTPException(status_code=500, detail="API 响应格式错误") except httpx.HTTPStatusError as e: error_text = e.response.text try: error_text.encode('utf-8') except (UnicodeEncodeError, AttributeError): error_text = repr(e.response.text) raise HTTPException(status_code=e.response.status_code, detail=f"OpenRouter API 错误: {error_text}") except httpx.RequestError as e: error_msg = str(e) try: error_msg.encode('utf-8') except UnicodeEncodeError: error_msg = repr(e) raise HTTPException(status_code=500, detail=f"请求错误: {error_msg}") @app.get("/") async def root(): """根路径，返回 API 信息""" return { "message": "病虫害识别 API", "version": "1.0.0", "model": MODEL_NAME, "endpoints": { "识别病虫害": "/api/v1/identify", "健康检查": "/health" } } @app.get("/health") async def health_check(): """健康检查端点""" return {"status": "healthy", "model": MODEL_NAME} @app.post("/api/v1/identify") async def identify_pest_disease( file: UploadFile = File(..., description="植物图片文件"), question: Optional[str] = None ): """ 病虫害识别接口 - **file**: 上传的植物图片（支持 jpg, jpeg, png 格式） - **question**: 可选的自定义问题，如果不提供则使用默认提示 """ # 验证文件类型 if not file.content_type or not file.content_type.startswith("image/"): raise HTTPException(status_code=400, detail="请上传图片文件（jpg, jpeg, png）") try: # 读取并验证图片 image_data = await file.read() image = Image.open(BytesIO(image_data)) # 验证图片大小（可选：限制最大尺寸） max_size = (4096, 4096) if image.size[0] > max_size[0] or image.size[1] > max_size[1]: image.thumbnail(max_size, Image.Resampling.LANCZOS) # 转换为 base64 image_base64 = encode_image_to_base64(image) # 构建用户消息 user_message = question or "请识别这张图片中的植物是否存在病虫害问题，如果存在，请详细说明病虫害类型、症状、严重程度和防治建议。" # 调用 OpenRouter API result = await call_openrouter_api(image_base64, user_message) return JSONResponse(content={ "success": True, "result": result, "image_info": { "format": image.format, "size": image.size, "mode": image.mode } }) except UnicodeEncodeError as e: raise HTTPException(status_code=500, detail=f"处理图片时出错: 编码错误 - {repr(e)}") except Exception as e: error_msg = str(e) # 确保错误信息可以正确编码 try: error_msg.encode('utf-8') except UnicodeEncodeError: error_msg = repr(e) raise HTTPException(status_code=500, detail=f"处理图片时出错: {error_msg}") @app.post("/api/v1/identify/base64") async def identify_pest_disease_base64(request: Base64ImageRequest): """ 使用 base64 编码图片的病虫害识别接口 - **image_base64**: base64 编码的图片数据（不包含 data:image 前缀） - **question**: 可选的自定义问题 """ try: # 清理 base64 字符串（移除可能的 data:image 前缀和空白字符） base64_str = request.image_base64.strip() if "," in base64_str: # 如果包含 data:image 前缀，提取 base64 部分 base64_str = base64_str.split(",")[-1] # 确保 base64 字符串只包含有效的 base64 字符 # 移除所有空白字符（包括换行符） base64_str = ''.join(base64_str.split()) # 解码 base64 图片 try: # 确保字符串是有效的 base64 格式 if not base64_str: raise ValueError("Base64 字符串为空") image_data = base64.b64decode(base64_str, validate=True) except binascii.Error: raise ValueError("Base64 解码失败: 无效的 base64 格式") except Exception as decode_error: error_msg = str(decode_error) try: error_msg.encode('utf-8') except UnicodeEncodeError: error_msg = repr(decode_error) raise ValueError(f"Base64 解码失败: {error_msg}") image = Image.open(BytesIO(image_data)) # 验证图片大小 max_size = (4096, 4096) if image.size[0] > max_size[0] or image.size[1] > max_size[1]: image.thumbnail(max_size, Image.Resampling.LANCZOS) # 重新编码 image_base64_encoded = encode_image_to_base64(image) # 构建用户消息 user_message = request.question or "请识别这张图片中的植物是否存在病虫害问题，如果存在，请详细说明病虫害类型、症状、严重程度和防治建议。" # 调用 OpenRouter API result = await call_openrouter_api(image_base64_encoded, user_message) return JSONResponse(content={ "success": True, "result": result, "image_info": { "format": image.format, "size": image.size, "mode": image.mode } }) except UnicodeEncodeError as e: raise HTTPException(status_code=500, detail=f"处理图片时出错: 编码错误 - {repr(e)}") except Exception as e: error_msg = str(e) # 确保错误信息可以正确编码 try: error_msg.encode('utf-8') except UnicodeEncodeError: error_msg = repr(e) raise HTTPException(status_code=500, detail=f"处理图片时出错: {error_msg}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)