实时音视频理解

# -- coding: utf-8 --
import os
import asyncio
import time
import threading
import queue
import pyaudio
from omni_realtime_client import OmniRealtimeClient, TurnDetectionMode


class AudioPlayer:
  """实时音频播放器"""

  def __init__(self, sample_rate=24000, channels=1, sample_width=2):
    self.sample_rate = sample_rate
    self.channels = channels
    self.sample_width = sample_width  # 16 位占 2 字节
    self.audio_queue = queue.Queue()
    self.is_playing = False
    self.play_thread = None
    self.pyaudio_instance = None
    self.stream = None
    self._lock = threading.Lock()  # 用于同步访问的锁
    self._last_data_time = time.time()  # 最后一次接收数据的时间
    self._response_done = False  # 标记响应是否完成的标志
    self._waiting_for_response = False  # 等待服务器响应的标志
    # 最后一次写入音频流的时间和最近音频块的时长，用于更精确地检测播放结束
    self._last_play_time = time.time()
    self._last_chunk_duration = 0.0

  def start(self):
    """启动音频播放器"""
    with self._lock:
      if self.is_playing:
        return

      self.is_playing = True

      try:
        self.pyaudio_instance = pyaudio.PyAudio()

        # 创建音频输出流
        self.stream = self.pyaudio_instance.open(
          format=pyaudio.paInt16,  # 16 位
          channels=self.channels,
          rate=self.sample_rate,
          output=True,
          frames_per_buffer=1024
        )

        # 启动播放线程
        self.play_thread = threading.Thread(target=self._play_audio)
        self.play_thread.daemon = True
        self.play_thread.start()

        print("音频播放器已启动")
      except Exception as e:
        print(f"启动音频播放器失败: {e}")
        self._cleanup_resources()
        raise

  def stop(self):
    """停止音频播放器"""
    with self._lock:
      if not self.is_playing:
        return

      self.is_playing = False

    # 清空队列
    while not self.audio_queue.empty():
      try:
        self.audio_queue.get_nowait()
      except queue.Empty:
        break

    # 等待播放线程结束（在锁外等待，避免死锁）
    if self.play_thread and self.play_thread.is_alive():
      self.play_thread.join(timeout=2.0)

    # 再次获取锁以清理资源
    with self._lock:
      self._cleanup_resources()

    print("音频播放器已停止")

  def _cleanup_resources(self):
    """清理音频资源（必须在持有锁的情况下调用）"""
    try:
      # 关闭音频流
      if self.stream:
        if not self.stream.is_stopped():
          self.stream.stop_stream()
        self.stream.close()
        self.stream = None
    except Exception as e:
      print(f"关闭音频流出错: {e}")

    try:
      if self.pyaudio_instance:
        self.pyaudio_instance.terminate()
        self.pyaudio_instance = None
    except Exception as e:
      print(f"终止 PyAudio 出错: {e}")

  def add_audio_data(self, audio_data):
    """将音频数据添加到播放队列"""
    if self.is_playing and audio_data:
      self.audio_queue.put(audio_data)
      with self._lock:
        self._last_data_time = time.time()  # 更新最后一次接收数据的时间
        self._waiting_for_response = False  # 收到数据，不再等待

  def stop_receiving_data(self):
    """标记不再有新的音频数据接收"""
    with self._lock:
      self._response_done = True
      self._waiting_for_response = False  # 响应结束，不再等待

  def prepare_for_next_turn(self):
    """重置播放器状态，为下一轮对话做准备。"""
    with self._lock:
      self._response_done = False
      self._last_data_time = time.time()
      self._last_play_time = time.time()
      self._last_chunk_duration = 0.0
      self._waiting_for_response = True  # 开始等待下一个响应

    # 清除上一轮对话的剩余音频数据
    while not self.audio_queue.empty():
      try:
        self.audio_queue.get_nowait()
      except queue.Empty:
        break

  def is_finished_playing(self):
    """检查是否所有音频数据已播放完毕"""
    with self._lock:
      queue_size = self.audio_queue.qsize()
      time_since_last_data = time.time() - self._last_data_time
      time_since_last_play = time.time() - self._last_play_time

      # ---------------------- 智能结束检测 ----------------------
      # 1. 首选：如果服务器已标记完成且播放队列为空。
      #    等待最近一个音频块播放完毕（块时长 + 0.1s 容差）。
      if self._response_done and queue_size == 0:
        min_wait = max(self._last_chunk_duration + 0.1, 0.5)  # 至少等待 0.5s
        if time_since_last_play >= min_wait:
          return True

      # 2. 备选：如果长时间没有收到新数据且播放队列为空。
      #    此逻辑作为服务器未明确发送 response.done 时的兜底方案。
      if not self._waiting_for_response and queue_size == 0 and time_since_last_data > 1.0:
        print("\n（长时间未收到新音频，假设播放已完成）")
        return True

      return False

  def _play_audio(self):
    """播放音频数据的 worker 线程"""
    while True:
      # 检查是否应停止
      with self._lock:
        if not self.is_playing:
          break
        stream_ref = self.stream  # 获取流的引用

      try:
        # 从队列获取音频数据，超时时间 0.1 秒
        audio_data = self.audio_queue.get(timeout=0.1)

        # 再次检查状态和流的有效性
        with self._lock:
          if self.is_playing and stream_ref and not stream_ref.is_stopped():
            try:
              # 播放音频数据
              stream_ref.write(audio_data)
              # 更新最新的播放信息
              self._last_play_time = time.time()
              self._last_chunk_duration = len(audio_data) / (
                    self.channels * self.sample_width) / self.sample_rate
            except Exception as e:
              print(f"写入音频流出错: {e}")
              break

        # 标记此数据块已处理
        self.audio_queue.task_done()

      except queue.Empty:
        # 队列为空，继续等待
        continue
      except Exception as e:
        print(f"播放音频出错: {e}")
        break


class MicrophoneRecorder:
  """实时麦克风录音器"""

  def __init__(self, sample_rate=16000, channels=1, chunk_size=3200):
    self.sample_rate = sample_rate
    self.channels = channels
    self.chunk_size = chunk_size
    self.pyaudio_instance = None
    self.stream = None
    self.frames = []
    self._is_recording = False
    self._record_thread = None

  def _recording_thread(self):
    """录音 worker 线程"""
    # 当 _is_recording 为 True 时持续从音频流读取数据
    while self._is_recording:
      try:
        # 使用 exception_on_overflow=False 避免缓冲区溢出导致崩溃
        data = self.stream.read(self.chunk_size, exception_on_overflow=False)
        self.frames.append(data)
      except (IOError, OSError) as e:
        # 当流被关闭时可能会报错
        print(f"读取录音流出错，可能已被关闭: {e}")
        break

  def start(self):
    """开始录音"""
    if self._is_recording:
      print("录音已在进行中。")
      return

    self.frames = []
    self._is_recording = True

    try:
      self.pyaudio_instance = pyaudio.PyAudio()
      self.stream = self.pyaudio_instance.open(
        format=pyaudio.paInt16,
        channels=self.channels,
        rate=self.sample_rate,
        input=True,
        frames_per_buffer=self.chunk_size
      )

      self._record_thread = threading.Thread(target=self._recording_thread)
      self._record_thread.daemon = True
      self._record_thread.start()
      print("麦克风录音已开始...")
    except Exception as e:
      print(f"启动麦克风失败: {e}")
      self._is_recording = False
      self._cleanup()
      raise

  def stop(self):
    """停止录音并返回音频数据"""
    if not self._is_recording:
      return None

    self._is_recording = False

    # 等待录音线程安全退出
    if self._record_thread:
      self._record_thread.join(timeout=1.0)

    self._cleanup()

    print("麦克风录音已停止。")
    return b''.join(self.frames)

  def _cleanup(self):
    """安全清理 PyAudio 资源"""
    if self.stream:
      try:
        if self.stream.is_active():
          self.stream.stop_stream()
        self.stream.close()
      except Exception as e:
        print(f"关闭音频流出错: {e}")

    if self.pyaudio_instance:
      try:
        self.pyaudio_instance.terminate()
      except Exception as e:
        print(f"终止 PyAudio 实例出错: {e}")

    self.stream = None
    self.pyaudio_instance = None


async def interactive_test():
  """
  交互式测试脚本：支持多轮对话，每轮可发送音频和图像。
  """
  # ------------------- 1. 初始化并建立连接（一次性） -------------------
  api_key = os.environ.get("DASHSCOPE_API_KEY")
  if not api_key:
    print("请设置 DASHSCOPE_API_KEY 环境变量。")
    return

  print("--- 实时音视频多模态对话客户端 ---")
  print("正在初始化音频播放器和客户端...")

  audio_player = AudioPlayer()
  audio_player.start()

  def on_audio_received(audio_data):
    audio_player.add_audio_data(audio_data)

  transcription_done = threading.Event()

  def on_transcription_completed(transcript):
    transcription_done.set()

  def on_response_done(event):
    print("\n（收到响应结束标记）")
    audio_player.stop_receiving_data()

  realtime_client = OmniRealtimeClient(
    base_url="wss://maas.qianwenaiapi.com/api-ws/v1/realtime",
    api_key=api_key,
    model="qwen3.5-omni-plus-realtime",
    voice="Tina",
    instructions="你是小云，一个私人助手。请准确、友好地回答用户的问题，始终保持乐于助人的态度。", # 设置模型角色
    on_text_delta=lambda text: print(f"助手回复: {text}", end="", flush=True),
    on_audio_delta=on_audio_received,
    on_input_transcript=on_transcription_completed,
    turn_detection_mode=TurnDetectionMode.MANUAL,
    extra_event_handlers={"response.done": on_response_done}
  )

  message_handler_task = None
  try:
    await realtime_client.connect()
    print("已连接到服务器。随时输入 'q' 或 'quit' 退出。")
    message_handler_task = asyncio.create_task(realtime_client.handle_messages())
    await asyncio.sleep(0.5)

    turn_counter = 1
    # ------------------- 2. 多轮对话循环 -------------------
    while True:
      print(f"\n--- 第 {turn_counter} 轮 ---")
      audio_player.prepare_for_next_turn()

      recorded_audio = None
      image_paths = []

      # --- 获取用户输入：从麦克风录音 ---
      loop = asyncio.get_event_loop()
      recorder = MicrophoneRecorder(sample_rate=16000)  # 推荐 16k 采样率用于语音识别

      print("准备录音。按 Enter 开始录音（或输入 'q' 退出）...")
      user_input = await loop.run_in_executor(None, input)
      if user_input.strip().lower() in ['q', 'quit']:
        print("用户请求退出...")
        return

      try:
        recorder.start()
      except Exception:
        print("无法启动录音。请检查麦克风权限和设备。跳过本轮。")
        continue

      print("录音中...再次按 Enter 停止录音。")
      await loop.run_in_executor(None, input)

      recorded_audio = recorder.stop()

      if not recorded_audio or len(recorded_audio) == 0:
        print("未录制到有效音频。请重新开始本轮。")
        continue

      # --- 3. 发送数据并获取响应 ---
      print("\n--- 输入确认 ---")
      print(f"待处理音频: 1（来自麦克风），图像: {len(image_paths)}")
      print("------------------")

      # 3.1 发送音频数据
      try:
        print(f"发送麦克风录音（{len(recorded_audio)} 字节）")
        await realtime_client.stream_audio(recorded_audio)
        await asyncio.sleep(0.1)
      except Exception as e:
        print(f"发送麦克风录音失败: {e}")
        continue

      # 3.3 提交并等待响应
      print("录音结束，等待模型回复...")
      await realtime_client.commit_audio_buffer()
      # 等待转录完成后再触发模型回复，避免输出交错
      await asyncio.to_thread(transcription_done.wait, 10)
      transcription_done.clear()
      await realtime_client.create_response()

      print("等待并播放服务器响应音频...")
      start_time = time.time()
      max_wait_time = 60
      while not audio_player.is_finished_playing():
        if time.time() - start_time > max_wait_time:
          print(f"\n等待超时（{max_wait_time} 秒）。进入下一轮。")
          break
        await asyncio.sleep(0.2)

      print("\n本轮音频播放完成！")
      turn_counter += 1

  except (asyncio.CancelledError, KeyboardInterrupt):
    print("\n程序被中断。")
  except Exception as e:
    print(f"发生未处理的错误: {e}")
  finally:
    # ------------------- 4. 清理资源 -------------------
    print("\n正在关闭连接并清理资源...")
    if message_handler_task and not message_handler_task.done():
      message_handler_task.cancel()

    if 'realtime_client' in locals() and realtime_client.ws and not realtime_client.ws.closed:
      await realtime_client.close()
      print("连接已关闭。")

    audio_player.stop()
    print("程序已退出。")


if __name__ == "__main__":
  try:
    asyncio.run(interactive_test())
  except KeyboardInterrupt:
    print("\n程序被用户强制退出。")