服务端Python SDK

class MultiModalCallback:
    """
    语音聊天回调类，用于处理语音聊天过程中的各种事件。
    """

    def on_started(self, dialog_id: str) -> None:
        """
        通知对话开始

        :param dialog_id: 回调对话ID
        """
        pass

    def on_stopped(self) -> None:
        """
        通知对话停止
        """
        pass

    def on_state_changed(self, state: 'dialog_state.DialogState') -> None:
        """
        对话状态改变

        :param state: 新的对话状态
        """
        pass

    def on_speech_audio_data(self, data: bytes) -> None:
        """
        合成音频数据回调

        :param data: 音频数据
        """
        pass

    def on_error(self, error) -> None:
        """
        发生错误时调用此方法。

        :param error: 错误信息
        """
        pass

    def on_connected(self) -> None:
        """
        成功连接到服务器后调用此方法。
        """
        pass

    def on_responding_started(self):
        """
        回复开始回调
        """
        pass

    def on_responding_ended(self):
        """
        回复结束
        """
        pass

    def on_speech_content(self, payload):
        """
        语音识别文本

        :param payload: text
        """
        pass

    def on_responding_content(self, payload):
        """
        大模型回复文本。

        :param payload: text
        """
        pass

    def on_request_accepted(self):
        """
        打断请求被接受。
        """
        pass

    def on_close(self, close_status_code, close_msg):
        """
        连接关闭时调用此方法。

        :param close_status_code: 关闭状态码
        :param close_msg: 关闭消息
        """
        pass

# callback visual_qa
def on_responding_content(self, payload: Dict[str, Any]):
        if payload:
            logger.debug(f"Response content: {payload}")
            try:
                commands_str = payload["output"]["extra_info"]["commands"]
                if "visual_qa" in commands_str:
                    if self.vqa_handler_func:
                        self.vqa_handler_func() #send_image_vqa
                    logger.debug("handle visual_qa command>>>>")
            except:
                return
...

# request VQA
def send_image_vqa(self):
    image1 = {"type": "base64",
         "value": CONST_TEST_IMAGE_BASE64}
    # 或者使用下发url方式调用
    image2 = {"type": "url", "value": image_url}

    images = [image1]
    images_params = RequestToRespondParameters(images=images)

    # 使用语音调用VQA，text传""即可
    self.conversation.request_to_respond("prompt", "", parameters=images_params)

# 1. 设置请求模式为AudioAndVideo
up_stream = Upstream(type="AudioAndVideo", mode="duplex", audio_format="pcm")
...

# 2. 发送连接到视频对话Agent请求
# {"action":"connect", "type":"voicechat_video_channel"}
def send_connect_video_command(self):
    """发送切换到视频模式的指令"""
    logger.info("Sending connect video command")
    try:
        video_connect_command = [{"action":"connect", "type":"voicechat_video_channel"}]

        self.conversation.request_to_respond("prompt","", RequestToRespondParameters(biz_params=BizParams(videos=video_connect_command)))
        # 标记视频模式已激活
        self.video_mode_active = True
        logger.info("Video mode activated")

    except Exception as e:
        logger.error(f"Failed to send connect video command: {e}")

 # 3. 间隔500ms 发送一张180KB以下的视频帧图片
 def send_video_frame_data_loop(self):
    """循环发送视频帧数据"""
    logger.info("Starting video frame data loop")
    self.video_thread_running = True

    # 获取示例图片数据
    image_data = self._get_sample_images()

    try:
        while self.video_thread_running and self.video_mode_active:
            # 发送图片数据
            self._send_video_frame(image_data)
            logger.debug(f"Sent video frame, sleeping for {VIDEO_FRAME_INTERVAL}s")

            # 等待500ms
            time.sleep(VIDEO_FRAME_INTERVAL)
...
# 其他调用过程省略，可参考完整示例。