Add support for image input in model resources and enhance related configurations

- Introduce a new `support_image_input` field in model resources, allowing models to indicate support for image input. - Update the backend models, schemas, and database seed scripts to accommodate the new field. - Enhance the AssistantConfig and related routes to handle image input capabilities, ensuring proper validation and error handling. - Modify the frontend components to include toggles for enabling visual understanding and filtering models based on image input support. - Implement necessary adjustments in the voice preview and pipeline to integrate video stream handling alongside audio functionalities.
2026-07-07 19:41:45 +08:00
parent 32a52d318a
commit c5db918830
18 changed files with 411 additions and 116 deletions
--- a/backend/services/config_resolver.py
+++ b/backend/services/config_resolver.py
@@ -86,6 +86,9 @@ async def resolve_runtime_config(
        realtimeModel=str(_value(realtime_resource, "modelId", "")),
        llm_values=(llm_resource.values or {}) if llm_resource else {},
        llm_secrets=(llm_resource.secrets or {}) if llm_resource else {},
+        llm_support_image_input=(
+            bool(llm_resource.support_image_input) if llm_resource else False
+        ),
        stt_values=(stt_resource.values or {}) if stt_resource else {},
        stt_secrets=(stt_resource.secrets or {}) if stt_resource else {},
        tts_values=(tts_resource.values or {}) if tts_resource else {},
--- a/backend/services/interface_catalog.py
+++ b/backend/services/interface_catalog.py
@@ -138,13 +138,6 @@ INTERFACE_DEFINITIONS: list[dict] = [
            field("textAggregationMode", "Text Aggregation Mode", default="token"),
        ],
    },
-    {
-        "interface_type": "dashscope-llm",
-        "name": "DashScope LLM",
-        "capability": "LLM",
-        "fields": OPENAI_COMMON
-        + [field("temperature", "Temperature", type_="number", default=0.7)],
-    },
    {
        "interface_type": "dashscope-asr",
        "name": "DashScope ASR",
@@ -165,7 +158,6 @@ INTERFACE_DEFINITIONS: list[dict] = [
    },
 ]

-
 def validate_fields(definition: dict, values: dict, secrets: dict) -> None:
    for item in definition["fields"]:
        source = secrets if item["group"] == "secrets" else values
--- a/backend/services/pipecat/pipeline.py
+++ b/backend/services/pipecat/pipeline.py
@@ -35,6 +35,7 @@ from pipecat.frames.frames import (
    OutputTransportMessageUrgentFrame,
    TextFrame,
    TTSSpeakFrame,
+    UserImageRequestFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.worker import PipelineParams, PipelineWorker
@@ -45,6 +46,11 @@ from pipecat.processors.aggregators.llm_response_universal import (
    LLMUserAggregatorParams,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.runner.utils import (
+    get_transport_client_id,
+    maybe_capture_participant_camera,
+)
+from pipecat.services.llm_service import FunctionCallParams
 from pipecat.turns.user_start import (
    TranscriptionUserTurnStartStrategy,
    VADUserTurnStartStrategy,
@@ -54,6 +60,13 @@ from pipecat.utils.time import time_now_iso8601
 from pipecat.workers.runner import WorkerRunner


+VISION_TOOL_NAME = "fetch_user_image"
+VISION_SYSTEM_HINT = (
+    "当前会话打开了视觉理解。用户询问当前画面、摄像头里有什么、人物/物品/"
+    "环境状态或需要你看一眼时,调用 fetch_user_image 获取当前视频帧,再基于画面回答。"
+)
+
+
 def _text_input(message) -> tuple[str, bool] | None:
    """解析现有 user-text 与 RTVI send-text 两种前端文字消息。"""
    if not isinstance(message, dict):
@@ -204,7 +217,12 @@ class PassthroughLLMAssistantAggregator(LLMAssistantAggregator):
        self._stream_text = ""


-async def run_pipeline(transport, cfg: AssistantConfig) -> None:
+async def run_pipeline(
+    transport,
+    cfg: AssistantConfig,
+    *,
+    vision_enabled: bool = False,
+) -> None:
    """在给定 transport 上构建并运行管线,直到连接结束。

    Args:
@@ -212,7 +230,10 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
                   只要有 .input() / .output() / event_handler 即可。
        cfg:       助手配置(随请求内联传入)。
    """
-    logger.info(f"启动管线: assistant={cfg.name} type={cfg.type} mode={cfg.runtimeMode}")
+    logger.info(
+        f"启动管线: assistant={cfg.name} type={cfg.type} "
+        f"mode={cfg.runtimeMode} vision={vision_enabled}"
+    )

    # 大脑:按类型决定 LLM 槽/开场白/上下文归属。每通电话一个实例(可持会话状态)。
    brain = build_brain(cfg)
@@ -224,6 +245,8 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
        cfg.runtimeMode = "pipeline"

    if cfg.runtimeMode == "realtime":
+        if vision_enabled:
+            logger.warning("Realtime 模式暂未接入视频帧工具,本次仅启用语音通话")
        await run_realtime_pipeline(transport, cfg)
        return

@@ -263,7 +286,16 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
        greeting = await brain.greeting(cfg)
        system_content = ""

-    context = LLMContext(messages=[{"role": "system", "content": system_content}])
+    def with_vision_hint(text: str) -> str:
+        if not vision_enabled:
+            return text
+        if not text:
+            return VISION_SYSTEM_HINT
+        return f"{text}\n\n{VISION_SYSTEM_HINT}"
+
+    context = LLMContext(
+        messages=[{"role": "system", "content": with_vision_hint(system_content)}]
+    )
    # LLM 槽由大脑提供:内部类型=OpenAI 兼容服务;fastgpt=包 SDK 的伪 LLM。
    llm = brain.build_llm(cfg, context)
    user_aggregator = LLMUserAggregator(
@@ -282,6 +314,58 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
    )
    assistant_aggregator = PassthroughLLMAssistantAggregator(context)
    text_input = TextInputProcessor()
+    vision_state: dict[str, str | None] = {"client_id": None}
+    vision_schema = FunctionSchema(
+        name=VISION_TOOL_NAME,
+        description=(
+            "获取用户当前摄像头画面。当用户询问当前画面、看到了什么、"
+            "人/物品/环境状态或需要视觉判断时调用。"
+        ),
+        properties={
+            "question": {
+                "type": "string",
+                "description": "用户关于当前视频画面的具体问题。",
+            }
+        },
+        required=["question"],
+    )
+
+    async def fetch_user_image(params: FunctionCallParams):
+        question = str(params.arguments.get("question") or "请描述当前画面。")
+        user_id = vision_state.get("client_id")
+        if not user_id:
+            await params.result_callback(
+                {
+                    "status": "no_video_client",
+                    "message": "当前还没有可用的摄像头视频流。",
+                }
+            )
+            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,
+        )
+
+    if vision_enabled:
+        llm.register_function(VISION_TOOL_NAME, fetch_user_image)
+
+    def set_visible_tools(schemas: list[FunctionSchema] | None = None) -> None:
+        tools = list(schemas or [])
+        if vision_enabled:
+            tools.append(vision_schema)
+        if tools:
+            context.set_tools(ToolsSchema(standard_tools=tools))
+        else:
+            context.set_tools()

    # 结束节点:等结束语「说完」(BotStoppedSpeakingFrame)再挂断,确保结束语的
    # 文字(走 data channel)与音频都已下发,避免前端只听到声音、看不到文字。
@@ -366,16 +450,17 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
    def set_system_prompt(text: str) -> None:
        """替换上下文里的系统提示(节点切换时整体替换,而非追加)。"""
        messages = context.get_messages()
+        content = with_vision_hint(text)
        if messages and messages[0].get("role") == "system":
-            messages[0] = {"role": "system", "content": text}
+            messages[0] = {"role": "system", "content": content}
        else:
-            messages.insert(0, {"role": "system", "content": text})
+            messages.insert(0, {"role": "system", "content": content})

    def apply_node(node_id: str | None) -> None:
        """进入节点:设置系统提示 + 把出边注册为可调用的转移工具。"""
        set_system_prompt(engine.system_prompt_for(node_id))
        if engine.is_end(node_id):
-            context.set_tools()  # 终止节点无工具
+            set_visible_tools([])  # 终止节点不展示转移工具,但保留视觉工具
            return
        schemas = [
            FunctionSchema(
@@ -386,10 +471,7 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
            )
            for edge in engine.outgoing(node_id)
        ]
-        if schemas:
-            context.set_tools(ToolsSchema(standard_tools=schemas))
-        else:
-            context.set_tools()  # 无出边:清空工具
+        set_visible_tools(schemas)

    async def go_to_node(target: str) -> None:
        """执行转移:切当前节点、重置计数、点亮画布、设置提示/工具。
@@ -455,6 +537,8 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:
                    engine.edge_fn_name(edge), make_transition_handler(edge)
                )
        apply_node(wf_state["current"])  # 设初始节点的提示与工具
+    elif vision_enabled:
+        set_visible_tools([])

    async def append_user_text_to_context(text: str, *, run_llm: bool) -> None:
        await worker.queue_frame(
@@ -559,6 +643,16 @@ async def run_pipeline(transport, cfg: AssistantConfig) -> None:

    @transport.event_handler("on_client_connected")
    async def on_client_connected(_transport, _client):
+        if vision_enabled:
+            try:
+                vision_state["client_id"] = get_transport_client_id(
+                    _transport,
+                    _client,
+                )
+                await maybe_capture_participant_camera(_transport, _client)
+                logger.info(f"视觉理解已接入视频客户端: {vision_state['client_id']}")
+            except Exception as e:
+                logger.warning(f"视觉理解摄像头捕获初始化失败: {e}")
        if greeting:
            # 外部托管类型的上下文由对方服务端维护,开场白不写入本地 context
            if brain.spec.owns_context:
--- a/backend/services/pipecat/service_factory.py
+++ b/backend/services/pipecat/service_factory.py
@@ -72,10 +72,15 @@ def create_llm(cfg: AssistantConfig):
    """DeepSeek 等,走 OpenAI 兼容的 /v1/chat/completions。"""
    if cfg.llm_interface_type not in {"openai-llm", "dashscope-llm"}:
        raise ValueError(f"不支持的 LLM 接口类型: {cfg.llm_interface_type}")
+    extra_body = cfg.llm_values.get("extraBody")
+    extra = {"extra_body": extra_body} if isinstance(extra_body, dict) else {}
    return OpenAILLMService(
        api_key=cfg.llm_api_key or config.LLM_API_KEY,
        base_url=cfg.llm_base_url or config.LLM_BASE_URL,
-        settings=OpenAILLMService.Settings(model=cfg.model or config.LLM_MODEL),
+        settings=OpenAILLMService.Settings(
+            model=cfg.model or config.LLM_MODEL,
+            extra=extra,
+        ),
    )


--- a/backend/services/pipecat/transports.py
+++ b/backend/services/pipecat/transports.py
@@ -24,21 +24,26 @@ from pipecat.transports.websocket.fastapi import (
 from pipecat.serializers.protobuf import ProtobufFrameSerializer


-def _base_params() -> dict:
+def _base_params(*, video_in_enabled: bool = False) -> dict:
    """两种 transport 共享的音频参数。"""
    return dict(
        audio_in_enabled=True,
        audio_out_enabled=True,
+        video_in_enabled=video_in_enabled,
        # EndFrame 后默认补 2s 静音(防止收尾被截断)。我们的挂断已等到机器人
        # 说完才触发,这段静音纯属空等,置 0 让结束语播完立即挂断。
        audio_out_end_silence_secs=0,
    )


-def build_webrtc_transport(connection: SmallWebRTCConnection) -> SmallWebRTCTransport:
+def build_webrtc_transport(
+    connection: SmallWebRTCConnection,
+    *,
+    video_in_enabled: bool = False,
+) -> SmallWebRTCTransport:
    return SmallWebRTCTransport(
        webrtc_connection=connection,
-        params=TransportParams(**_base_params()),
+        params=TransportParams(**_base_params(video_in_enabled=video_in_enabled)),
    )