feat: 准备用上 BaiLianVoiceRecogniseWebsocket

2025-11-02 18:29:20 +08:00
parent dd4422ff8b
commit d3d6e1afc6
1 changed files with 1 additions and 1 deletions
@@ -0,0 +1,156 @@
+import json
+from typing import Callable, TypedDict
+
+import websockets
+from websockets.client import ClientConnection
+
+from app.config.env import env
+
+
+class BaiLianVoiceRecogniseSocket:
+  """
+  通义千问——实时语音识别
+  官方文档：https://help.aliyun.com/zh/model-studio/qwen-real-time-speech-recognition?spm=a2c4g.11186623.0.0.620e5a26eUNz80#0d1b30e0a1glp
+  客户端事件：https://help.aliyun.com/zh/model-studio/qwen-asr-realtime-client-events
+  服务端事件：https://help.aliyun.com/zh/model-studio/qwen-asr-realtime-server-events?spm=a2c4g.11186623.help-menu-2400256.d_2_6_8_1.620e5acdS0URfG
+  """
+
+  def __init__(
+    self,
+    # 百炼平台的api key，https://bailian.console.aliyun.com/?spm=5176.28326591.0.0.4d0555e3pzewCB&tab=model#/api-key
+    api_key: str = env.llm_key_bailian,
+    # 模型名称，支持“qwen3-asr-flash-realtime”与“qwen3-asr-flash-realtime-2025-10-27”
+    qwen_model: str = "qwen3-asr-flash-realtime",
+    # websocket接口地址
+    base_url: str = "wss://dashscope.aliyuncs.com/api-ws/v1/realtime",
+    # 可选的附加头信息
+    additional_headers: dict | None = None,
+    # 是否启用vad模式，否则启用manual模式
+    # vad模式：服务端自动检测语音的起点和终点（断句）。开发者只需持续发送音频流，服务端会在检测到一句话结束时自动返回最终识别结果。此模式适用于实时对话、会议记录等场景。
+    # vad文档：https://help.aliyun.com/zh/model-studio/qwen-asr-realtime-interaction-process?spm=a2c4g.11186623.0.0.16fe44488GkDZB#9b49887720jcw
+    # manual模式：由客户端控制断句。客户端需要发送完一整句话的音频后，再发送一个input_audio_buffer.commit事件来通知服务端。此模式适用于客户端能明确判断语句边界的场景，如聊天软件中的发送语音。
+    # manual文档：https://help.aliyun.com/zh/model-studio/qwen-asr-realtime-interaction-process?spm=a2c4g.11186623.0.0.16fe44488GkDZB#ee09a3493fsuc
+    enable_server_vad: bool = True,
+
+    # 监听开始说话动作，参数为item_id
+    on_speech_started: Callable[[str], None] | None = None,
+    # 监听正在说话的内容，参数为item_id以及本次说话叠加的完整内容
+    on_speech_content: Callable[[str, str], None] | None = None,
+    # 监听说话结束动作，参数为item_id
+    on_speech_stopped: Callable[[str], None] | None = None,
+    # 监听说话完毕动作，参数为item_id以及本次说话万恒内容
+    on_speech_completed: Callable[[str, str], None] | None = None
+  ):
+    """
+    构造函数，会自动设置连接百炼实时语音识别websocket接口所需要的秘钥，模型名称，websocket接口地址，以及可选的附加头信息
+    """
+    self.api_key = api_key
+    self.qwen_model = qwen_model
+    self.base_url = base_url
+    self.additional_headers = additional_headers or {}
+    self.enable_server_vad = enable_server_vad
+
+    self.on_speech_started = on_speech_started
+    self.on_speech_content = on_speech_content
+    self.on_speech_stopped = on_speech_stopped
+    self.on_speech_completed = on_speech_completed
+
+    socket: ClientConnection | None = None
+    self.socket = socket
+
+  @staticmethod
+  def log(message):
+    print("BaiLianVoiceRecognise: " + message)
+
+  async def connect(self):
+    """
+    连接websocket接口，得到socket对象
+    """
+    url = f"{self.base_url}?model={self.qwen_model}"
+    headers = {
+      "Authorization": f"Bearer {self.api_key}",
+      "OpenAI-Beta": "realtime=v1",
+      **self.additional_headers,
+    }
+    self.socket = await websockets.connect(url, additional_headers=headers)
+
+    await self.send_session_update_event()
+
+  async def send_session_update_event(self):
+    """
+    发送session.update事件，通知百炼接口启用vad模式还是manual模式
+    """
+    if not self.socket:
+      raise Exception("请先调用connect方法初始化websocket连接")
+    # 会话更新事件
+    event_manual = {
+      "event_id": "event_123",
+      "type": "session.update",
+      "session": {
+        "modalities": ["text"],
+        "input_audio_format": "pcm",
+        "sample_rate": 16000,
+        "input_audio_transcription": {
+          # 语种标识，可选，如果有明确的语种信息，建议设置
+          "language": "zh"
+          # 语料，可选，如果有语料，建议设置以增强识别效果
+          # "corpus": {
+          #     "text": ""
+          # }
+        },
+        "turn_detection": None
+      }
+    }
+    event_vad = {
+      "event_id": "event_123",
+      "type": "session.update",
+      "session": {
+        "modalities": ["text"],
+        "input_audio_format": "pcm",
+        "sample_rate": 16000,
+        "input_audio_transcription": {
+          "language": "zh"
+        },
+        "turn_detection": {
+          "type": "server_vad",
+          "threshold": 0.2,
+          "silence_duration_ms": 800
+        }
+      }
+    }
+    if self.enable_server_vad:
+      self.log(f"send session.update event: {json.dumps(event_vad, indent=2)}")
+      self.socket.send(json.dumps(event_vad))
+    else:
+      self.log(f"send session.update event: {json.dumps(event_manual, indent=2)}")
+      self.socket.send(json.dumps(event_manual))
+
+  async def waiting_message_from_socket(self):
+    """将百炼识别结果转发到客户端"""
+    try:
+      async for message in self.socket:
+        data = json.loads(message)
+
+        message_data_type = data.get('type', None)
+        message_data_item_id = data.get('item_id', None)
+
+        if message_data_type == "input_audio_buffer.speech_started":
+          self.on_speech_started(message_data_item_id)
+        if message_data_type == "conversation.item.input_audio_transcription.text":
+          self.on_speech_content(message_data_item_id, data.get('stash', ""))
+        if message_data_type == "input_audio_buffer.speech_stopped":
+          self.on_speech_stopped(message_data_item_id)
+        if message_data_type == "conversation.item.input_audio_transcription.completed":
+          self.on_speech_completed(message_data_item_id, data.get('transcript', ""))
+
+    except Exception as e:
+      print(f"等待百炼的消息出错: {e}")
+
+
+class BaiLianRecogniseMessage(TypedDict):
+  event_id: str
+  type: str
+  item_id: str
+  content_index: int
+  text: str
+  stash: str