Python SDK

class MultiModalCallback:
    """
    The callback class for voice chat, used to handle various events during the voice chat process.
    """

    def on_started(self, dialog_id: str) -> None:
        """
        Notifies that the dialog has started.

        :param dialog_id: The callback dialog ID.
        """
        pass

    def on_stopped(self) -> None:
        """
        Notifies that the dialog has stopped.
        """
        pass

    def on_state_changed(self, state: 'dialog_state.DialogState') -> None:
        """
        The dialog state has changed.

        :param state: The new dialog state.
        """
        pass

    def on_speech_audio_data(self, data: bytes) -> None:
        """
        Callback for synthesized audio data.

        :param data: The audio data.
        """
        pass

    def on_error(self, error) -> None:
        """
        This method is called when an error occurs.

        :param error: The error message.
        """
        pass

    def on_connected(self) -> None:
        """
        This method is called after a successful connection to the server.
        """
        pass

    def on_responding_started(self):
        """
        Callback for when the response starts.
        """
        pass

    def on_responding_ended(self):
        """
        The response has ended.
        """
        pass

    def on_speech_content(self, payload):
        """
        The speech recognition text.

        :param payload: text
        """
        pass

    def on_responding_content(self, payload):
        """
        The large model's response text.

        :param payload: text
        """
        pass

    def on_request_accepted(self):
        """
        The interruption request was accepted.
        """
        pass

    def on_close(self, close_status_code, close_msg):
        """
        This method is called when the connection is closed.

        :param close_status_code: The closing status code.
        :param close_msg: The closing message.
        """
        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}
    # Alternatively, call by sending a URL
    image2 = {"type": "url", "value": image_url}

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

    # To call VQA using voice, pass "" for text
    self.conversation.request_to_respond("prompt", "", parameters=images_params)

# 1. Set the request mode to AudioAndVideo
up_stream = Upstream(type="AudioAndVideo", mode="duplex", audio_format="pcm")
...

# 2. Send a request to connect to the video chat agent
# {"action":"connect", "type":"voicechat_video_channel"}
def send_connect_video_command(self):
    """Sends the instruction to switch to video mode"""
    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)))
        # Mark video mode as activated
        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. Send a video frame image (under 180 KB) every 500 ms
 def send_video_frame_data_loop(self):
    """Sends video frame data in a loop"""
    logger.info("Starting video frame data loop")
    self.video_thread_running = True

    # Get sample image data
    image_data = self._get_sample_images()

    try:
        while self.video_thread_running and self.video_mode_active:
            # Send image data
            self._send_video_frame(image_data)
            logger.debug(f"Sent video frame, sleeping for {VIDEO_FRAME_INTERVAL}s")

            # Wait for 500 ms
            time.sleep(VIDEO_FRAME_INTERVAL)
...
# Other call processes are omitted. See the complete example for details.