# ============================================================================= # 企微IT智能服务台 — 视觉理解服务（D5 / P1-3） # ============================================================================= # 说明：封装 Qwen-VL 截图理解能力，通过 Dify vision workflow 调用本地 # Qwen3-VL-8B-Instruct 模型，将员工截屏转换为结构化描述文本， # 注入会话上下文参与后续 AI 推理。 # # 核心能力： # 1. analyze_screenshot: 接收图片字节流 → Dify vision workflow → 结构化描述 # 2. _preprocess_image: 图片预处理（resize/compress） # 3. inject_to_conversation_context: 将视觉描述注入会话消息上下文 # # 设计决策： # - 视觉理解经 Dify 后端 → Qwen-VL 本地推理（D5 硬约束） # - 图片预处理：Pillow resize max 1024px + JPEG quality=85 # - 视觉模型可配置（settings.qwen_vl_model，默认 Qwen3-VL-8B-Instruct） # ============================================================================= import base64 import io import logging from typing import Any, Dict, Optional import httpx from PIL import Image from app.config import settings logger = logging.getLogger(__name__) class VisionService: """视觉理解服务 — 截图→结构化描述。 调用本地 Qwen-VL（经 Dify vision workflow）分析员工截图， 生成结构化文本描述并注入会话上下文。 使用方式： service = VisionService() result = await service.analyze_screenshot(image_bytes, conversation_id) await service.inject_to_conversation_context(result["description"], conversation_id) Attributes: dify_vision_api_url: Dify Vision Workflow API 端点 dify_vision_api_key: Dify Vision Workflow API Key model: 视觉模型名称（默认 Qwen3-VL-8B-Instruct） """ # 图片预处理参数 _MAX_DIMENSION: int = 1024 # 最大边长（像素） _JPEG_QUALITY: int = 85 # JPEG 压缩质量 def __init__( self, dify_vision_api_url: Optional[str] = None, dify_vision_api_key: Optional[str] = None, model: Optional[str] = None, ): """初始化视觉理解服务。 Args: dify_vision_api_url: Dify Vision Workflow API 端点 dify_vision_api_key: Dify Vision Workflow API Key model: 视觉模型名称 """ self.dify_vision_api_url: str = ( dify_vision_api_url or settings.dify_vision_api_url ) self.dify_vision_api_key: str = ( dify_vision_api_key or settings.dify_vision_api_key ) self.model: str = model or settings.qwen_vl_model self._client: Optional[httpx.AsyncClient] = None async def _get_client(self) -> httpx.AsyncClient: """获取或创建 httpx 异步客户端。""" if self._client is None or self._client.is_closed: self._client = httpx.AsyncClient( timeout=httpx.Timeout(60.0), # 视觉推理可能需要更长时间 headers={ "Authorization": f"Bearer {self.dify_vision_api_key}", "Content-Type": "application/json", }, ) return self._client async def close(self): """关闭 httpx 客户端。""" if self._client and not self._client.is_closed: await self._client.aclose() self._client = None # -------------------------------------------------------------------------- # 核心方法 # -------------------------------------------------------------------------- async def analyze_screenshot( self, image_bytes: bytes, conversation_id: str ) -> Dict[str, Any]: """分析截图，返回结构化视觉描述。 Args: image_bytes: 图片字节流 conversation_id: 会话ID（用于上下文关联） Returns: Dict: { "description": str, # 结构化的视觉描述文本 "confidence": float, # 视觉理解置信度 "metadata": dict, # 元数据（detected_ui_elements, error_codes等） } """ # 默认降级响应 default_response: Dict[str, Any] = { "description": "", "confidence": 0.0, "metadata": {}, } if not self.dify_vision_api_url: logger.warning("Dify Vision API 未配置，跳过视觉分析") return default_response try: # 1. 预处理图片 processed_image = await self._preprocess_image(image_bytes) # 2. 调用 Dify vision workflow result = await self._call_vision_workflow(processed_image, conversation_id) if result is None: return default_response return { "description": result.get("description", ""), "confidence": float(result.get("confidence", 0.0)), "metadata": result.get("metadata", {}), } except Exception as e: logger.error(f"截图视觉分析失败: {e}") return default_response async def inject_to_conversation_context( self, description: str, conversation_id: str ) -> bool: """将视觉描述注入会话消息上下文。 以 system 消息形式将视觉理解结果写入会话消息表， 后续 AI 推理时可读取此描述作为上下文。 Args: description: 视觉理解描述文本 conversation_id: 会话ID Returns: bool: 注入成功返回 True """ if not description: logger.debug("视觉描述为空，跳过上下文注入") return False try: from app.database import _get_session_factory from app.models.message import Message session_factory = _get_session_factory() async with session_factory() as db: msg = Message( conversation_id=conversation_id, sender_type="system", content=f"[视觉理解] {description}", ) db.add(msg) await db.commit() logger.info( f"视觉描述已注入会话 {conversation_id}: " f"description_length={len(description)}" ) return True except ImportError: logger.warning("Message 模型不可用，无法注入视觉描述") return False except Exception as e: logger.error(f"注入视觉描述失败: {e}") return False # -------------------------------------------------------------------------- # 内部方法 # -------------------------------------------------------------------------- async def _preprocess_image(self, image_bytes: bytes) -> bytes: """预处理图片：resize + compress。 使用 Pillow 将图片缩小到最大 1024px，压缩为 JPEG quality=85， 减少传输大小和视觉模型推理开销。 Args: image_bytes: 原始图片字节流 Returns: bytes: 预处理后的图片字节流 """ try: img = Image.open(io.BytesIO(image_bytes)) # 转换为 RGB（处理 RGBA/PNG 等格式） if img.mode in ("RGBA", "P", "LA"): img = img.convert("RGB") # 按最大边长等比缩放 w, h = img.size max_dim = max(w, h) if max_dim > self._MAX_DIMENSION: ratio = self._MAX_DIMENSION / max_dim new_w, new_h = int(w * ratio), int(h * ratio) img = img.resize((new_w, new_h), Image.LANCZOS) logger.debug(f"图片缩放: {w}x{h} → {new_w}x{new_h}") # 输出为 JPEG buffer = io.BytesIO() img.save(buffer, format="JPEG", quality=self._JPEG_QUALITY) result = buffer.getvalue() logger.debug( f"图片预处理完成: input_size={len(image_bytes)}, " f"output_size={len(result)}" ) return result except Exception as e: logger.warning(f"图片预处理失败，使用原始图片: {e}") return image_bytes async def _call_vision_workflow( self, processed_image: bytes, conversation_id: str ) -> Optional[Dict[str, Any]]: """调用 Dify Vision Workflow 进行视觉理解。 将预处理后的图片以 base64 格式发送到 Dify vision workflow。 Args: processed_image: 预处理后的图片字节流 conversation_id: 会话ID Returns: Optional[Dict]: 视觉理解结果，失败返回 None """ try: # Base64 编码图片 image_base64 = base64.b64encode(processed_image).decode("utf-8") payload: Dict[str, Any] = { "inputs": { "image_base64": image_base64, "conversation_id": conversation_id, }, "response_mode": "blocking", "user": f"vision-{conversation_id[:8]}", } client = await self._get_client() logger.info( f"调用 Dify Vision Workflow: conversation_id={conversation_id}, " f"model={self.model}" ) response = await client.post(self.dify_vision_api_url, json=payload) response.raise_for_status() data = response.json() # 解析 Dify workflow 返回 outputs = data.get("data", {}).get("outputs", {}) if not outputs: logger.warning("Dify Vision Workflow 返回空 outputs") return None return { "description": outputs.get("description", ""), "confidence": float(outputs.get("confidence", 0.0)), "metadata": outputs.get("metadata", {}), } except httpx.TimeoutException: logger.error("Dify Vision Workflow 超时") return None except httpx.HTTPStatusError as e: logger.error(f"Dify Vision Workflow HTTP 错误: status={e.response.status_code}") return None except Exception as e: logger.error(f"Dify Vision Workflow 调用失败: {e}") return None