From 5bc4e24adb40c65d0a27e7b14ebff539eb92df97 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Wed, 8 Jul 2026 10:33:44 +0800 Subject: [PATCH] Add vision model enhancements and image processing capabilities - Update `requirements.txt` to include Pillow for image handling. - Refactor vision model validation logic in `voice_webrtc.py` to improve error handling for unsupported image input. - Introduce new functions in `pipeline.py` for image data processing and analysis using vision models. - Implement `VisionCaptureProcessor` to manage video frame requests for auxiliary vision model analysis. - Enhance the pipeline to support image input requests and integrate vision model responses into the processing flow. --- backend/requirements.txt | 1 + backend/routes/voice_webrtc.py | 22 ++-- backend/services/pipecat/pipeline.py | 170 +++++++++++++++++++++++++-- 3 files changed, 169 insertions(+), 24 deletions(-) diff --git a/backend/requirements.txt b/backend/requirements.txt index 0296877..436435d 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -3,6 +3,7 @@ # silero -> 本地 VAD(判断用户说话起止),语音必备 # openai -> OpenAI 兼容的 LLM/STT/TTS 客户端(DeepSeek、SenseVoice、CosyVoice 都走它) pipecat-ai[webrtc,websocket,silero,openai]==1.3.0 +Pillow>=11.1.0,<13 # FastGPT 类型助手:本地 SDK(包 /api/v1/chat/completions 流式 + chatId 会话) fastgpt-client @ file:///Users/wangx/Code/AI-VideoAssistant-Project/fastgpt-python-sdk diff --git a/backend/routes/voice_webrtc.py b/backend/routes/voice_webrtc.py index b2b03e3..45d72c0 100644 --- a/backend/routes/voice_webrtc.py +++ b/backend/routes/voice_webrtc.py @@ -74,16 +74,6 @@ async def _resolve_config(offer: SignalingOffer) -> AssistantConfig: raise ValueError("offer 缺少 assistant_id 或 inline_config") -def _apply_vision_model(cfg: AssistantConfig) -> None: - cfg.model = cfg.vision_model - cfg.llm_interface_type = cfg.vision_llm_interface_type - cfg.llm_values = cfg.vision_llm_values - cfg.llm_secrets = cfg.vision_llm_secrets - cfg.llm_support_image_input = cfg.vision_llm_support_image_input - cfg.llm_api_key = cfg.vision_llm_api_key - cfg.llm_base_url = cfg.vision_llm_base_url - - async def _handle_offer(websocket, payload, peers): from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection from services.pipecat.pipeline import run_pipeline @@ -99,11 +89,17 @@ async def _handle_offer(websocket, payload, peers): cfg = await _resolve_config(offer) # 解析放在建连前,配置错就别建连 vision_enabled = offer.vision_enabled or cfg.vision_enabled if vision_enabled: - if not cfg.vision_llm_support_image_input: + has_native_vision = ( + not cfg.vision_model_resource_id and cfg.llm_support_image_input + ) + has_aux_vision_model = ( + bool(cfg.vision_model_resource_id) + and cfg.vision_llm_support_image_input + ) + if not (has_native_vision or has_aux_vision_model): raise ValueError( - "当前视觉模型不支持图片输入,请在模型资源中选择支持图片输入的 LLM" + "当前模型不支持图片输入,请在模型资源中选择支持图片输入的视觉模型" ) - _apply_vision_model(cfg) pc = SmallWebRTCConnection(ice_servers=aiortc_ice_servers()) if pc_id: pc._pc_id = pc_id diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 95b92e4..841e0d6 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -6,11 +6,16 @@ 对应 dograh 的 pipeline_builder.py + run_pipeline.py(已砍掉 workflow 引擎/DB/录音/指标)。 """ +import asyncio +import base64 +from io import BytesIO from uuid import uuid4 import config from loguru import logger from models import AssistantConfig +from openai import AsyncOpenAI +from PIL import Image from services.brains import build_brain from services.pipecat.service_factory import ( create_realtime_service, @@ -35,6 +40,7 @@ from pipecat.frames.frames import ( OutputTransportMessageUrgentFrame, TextFrame, TTSSpeakFrame, + UserImageRawFrame, UserImageRequestFrame, ) from pipecat.pipeline.pipeline import Pipeline @@ -65,6 +71,67 @@ VISION_SYSTEM_HINT = ( "当前会话打开了视觉理解。用户询问当前画面、摄像头里有什么、人物/物品/" "环境状态或需要你看一眼时,调用 fetch_user_image 获取当前视频帧,再基于画面回答。" ) +VISION_ANALYSIS_SYSTEM_PROMPT = ( + "你是一个视觉理解模型。请只根据图片内容和用户问题给出准确、简洁的中文观察结果。" + "如果画面不足以判断,请明确说明不确定。" +) + + +def _vision_uses_main_llm(cfg: AssistantConfig) -> bool: + """模型自己支持图片时,沿用 Pipecat 的同上下文视觉工具路径。""" + return not cfg.vision_model_resource_id and cfg.llm_support_image_input + + +def _image_data_uri(frame: UserImageRawFrame) -> str: + if not frame.format: + raise ValueError("摄像头图片帧缺少 format,无法编码给视觉模型") + buffer = BytesIO() + Image.frombytes(frame.format, frame.size, frame.image).save( + buffer, + format="JPEG", + quality=85, + ) + encoded = base64.b64encode(buffer.getvalue()).decode("utf-8") + return f"data:image/jpeg;base64,{encoded}" + + +async def _analyze_image_with_vision_model( + cfg: AssistantConfig, + frame: UserImageRawFrame, + question: str, +) -> str: + if cfg.vision_llm_interface_type not in {"openai-llm", "dashscope-llm"}: + raise ValueError(f"不支持的视觉 LLM 接口类型: {cfg.vision_llm_interface_type}") + + data_uri = await asyncio.to_thread(_image_data_uri, frame) + extra_body = cfg.vision_llm_values.get("extraBody") + extra = {"extra_body": extra_body} if isinstance(extra_body, dict) else {} + client = AsyncOpenAI( + api_key=cfg.vision_llm_api_key or config.LLM_API_KEY, + base_url=cfg.vision_llm_base_url or config.LLM_BASE_URL, + ) + try: + response = await client.chat.completions.create( + model=cfg.vision_model or config.LLM_MODEL, + messages=[ + {"role": "system", "content": VISION_ANALYSIS_SYSTEM_PROMPT}, + { + "role": "user", + "content": [ + {"type": "text", "text": question}, + {"type": "image_url", "image_url": {"url": data_uri}}, + ], + }, + ], + **extra, + ) + finally: + await client.close() + + content = response.choices[0].message.content if response.choices else "" + if isinstance(content, str): + return content.strip() + return str(content or "").strip() def _text_input(message) -> tuple[str, bool] | None: @@ -130,6 +197,50 @@ class TextInputProcessor(FrameProcessor): await self._call_event_handler("on_text_append", text) +class VisionCaptureProcessor(FrameProcessor): + """Capture one requested video frame for auxiliary vision-model analysis.""" + + def __init__(self, timeout_s: float = 3.0): + super().__init__() + self._timeout_s = timeout_s + self._pending: dict[str, asyncio.Future[UserImageRawFrame]] = {} + + async def request_image( + self, + requester: FrameProcessor, + request: UserImageRequestFrame, + ) -> UserImageRawFrame: + key = request.tool_call_id or str(uuid4()) + request.tool_call_id = key + request.append_to_context = False + request.result_callback = None + + loop = asyncio.get_running_loop() + future: asyncio.Future[UserImageRawFrame] = loop.create_future() + self._pending[key] = future + await requester.push_frame(request, FrameDirection.UPSTREAM) + try: + return await asyncio.wait_for(future, timeout=self._timeout_s) + finally: + self._pending.pop(key, None) + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if ( + isinstance(frame, UserImageRawFrame) + and frame.request + and frame.request.tool_call_id + and frame.request.tool_call_id in self._pending + ): + future = self._pending[frame.request.tool_call_id] + if not future.done(): + future.set_result(frame) + return + + await self.push_frame(frame, direction) + + class RealtimeTextInputProcessor(FrameProcessor): """Route text input directly to a realtime service without cascade semantics.""" @@ -314,6 +425,8 @@ async def run_pipeline( ) assistant_aggregator = PassthroughLLMAssistantAggregator(context) text_input = TextInputProcessor() + vision_capture = VisionCaptureProcessor() + vision_native_mode = vision_enabled and _vision_uses_main_llm(cfg) vision_state: dict[str, str | None] = {"client_id": None} vision_schema = FunctionSchema( name=VISION_TOOL_NAME, @@ -342,17 +455,51 @@ async def run_pipeline( ) return - logger.debug(f"请求当前视频帧: user_id={user_id}, question={question}") - await params.llm.push_frame( - UserImageRequestFrame( - user_id=user_id, - text=question, - append_to_context=True, - function_name=params.function_name, - tool_call_id=params.tool_call_id, - result_callback=params.result_callback, - ), - FrameDirection.UPSTREAM, + request = UserImageRequestFrame( + user_id=user_id, + text=question, + append_to_context=vision_native_mode, + function_name=params.function_name, + tool_call_id=params.tool_call_id, + result_callback=params.result_callback if vision_native_mode else None, + ) + if vision_native_mode: + logger.debug( + f"请求当前视频帧进入主 LLM 上下文: user_id={user_id}, question={question}" + ) + await params.llm.push_frame(request, FrameDirection.UPSTREAM) + return + + logger.debug( + f"请求当前视频帧给单独视觉模型分析: user_id={user_id}, question={question}" + ) + try: + frame = await vision_capture.request_image(params.llm, request) + observation = await _analyze_image_with_vision_model(cfg, frame, question) + except asyncio.TimeoutError: + await params.result_callback( + { + "status": "timeout", + "message": "等待摄像头视频帧超时。", + } + ) + return + except Exception as e: + logger.exception(f"视觉模型分析失败: {e}") + await params.result_callback( + { + "status": "error", + "message": f"视觉理解失败: {type(e).__name__}", + } + ) + return + + await params.result_callback( + { + "status": "ok", + "question": question, + "observation": observation or "视觉模型没有返回有效观察结果。", + } ) if vision_enabled: @@ -399,6 +546,7 @@ async def run_pipeline( pipeline = Pipeline( [ transport.input(), + vision_capture, text_input, stt, user_aggregator,