feat: 将前端录音转发给百炼websocket接口

app/controller/add_websocket_voice_recognise_route.py

@@ -1,8 +1,15 @@
 import json
+import base64
+import os
+import struct
 
+import websockets
+import time
 from fastapi import FastAPI, WebSocket, Query
 from starlette.websockets import WebSocketState
 
+from app.config.env import env
+
 
 def add_websocket_voice_recognise_route(app: FastAPI):
   @app.websocket("/ws_voice_recognise")
@@ -10,17 +17,112 @@ def add_websocket_voice_recognise_route(app: FastAPI):
     print("新用户连接", user_id)
     # 接收连接
     await receive_socket.accept()
+
+    # 百炼WebSocket配置
+    API_KEY = env.llm_key_bailian
+    QWEN_MODEL = "qwen3-asr-flash-realtime"
+    baseUrl = "wss://dashscope.aliyuncs.com/api-ws/v1/realtime"
+    url = f"{baseUrl}?model={QWEN_MODEL}"
+
+    headers = {
+      "Authorization": f"Bearer {API_KEY}",
+      "OpenAI-Beta": "realtime=v1"
+    }
+
+    bailing_socket = None
+
     try:
-      while True:
+      # 连接百炼WebSocket
-        # 将接收得到消息发送供其他用户
+      bailing_socket = await websockets.connect(url, additional_headers=headers)
-        pcm_data_string = await receive_socket.receive_text()
+      print("已连接到百炼WebSocket")
-        pcm_data = json.loads(pcm_data_string)
+
-        print("pcm_data", pcm_data)
+      # 发送会话更新事件
-        # await person_socket.send_text(data)
+      session_event = {
+        "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
+          }
+        }
+      }
+      await bailing_socket.send(json.dumps(session_event))
+
+      # 创建双向转发任务
+      async def forward_to_bailing():
+        """将前端PCM数据转发到百炼"""
+        try:
+          while True:
+            # 接收前端PCM数据（二进制格式）
+            pcm_data_string = await receive_socket.receive_text()
+            pcm_data_list = json.loads(pcm_data_string)
+            # print("pcm_data", pcm_data_list)
+            # print(f"接收到PCM数组数据，长度: {len(pcm_data_list)}")
+
+            # 将 int16 列表打包成 bytes
+            pcm_bytes = struct.pack('<' + 'h' * len(pcm_data_list), *pcm_data_list)
+            # 将二进制PCM数据转换为base64编码
+            encoded_data = base64.b64encode(pcm_bytes).decode('utf-8')
+
+            # 构造发送到百炼的事件
+            audio_event = {
+              "event_id": f"event_{int(time.time() * 1000)}",
+              "type": "input_audio_buffer.append",
+              "audio": encoded_data
+            }
+            await bailing_socket.send(json.dumps(audio_event))
+            # print("已转发音频数据到百炼", f"PCM二进制数据，长度: {len(pcm_bytes)} 字节")
+
+        except Exception as e:
+          print(f"转发到百炼时出错: {e}")
+
+      async def forward_to_client():
+        """将百炼识别结果转发到客户端"""
+        try:
+          async for message in bailing_socket:
+            data = json.loads(message)
+            print("收到百炼消息", data.get("type"))
+
+            # 将识别结果发送给前端
+            await receive_socket.send_text(json.dumps(data, ensure_ascii=False))
+
+            # 如果收到最终转录结果，可以单独处理
+            if data.get("type") == "conversation.item.input_audio_transcription.completed":
+              transcript = data.get("transcript", "")
+              print(f"最终转录结果: {transcript}")
+              # 可以发送专门的转录消息给前端
+              await receive_socket.send_text(json.dumps({
+                "type": "final_transcript",
+                "text": transcript
+              }, ensure_ascii=False))
+
+        except Exception as e:
+          print(f"从百炼转发到客户端时出错: {e}")
+
+      # 并发运行两个任务
+      import asyncio
+      await asyncio.gather(
+        forward_to_bailing(),
+        forward_to_client(),
+        return_exceptions=True
+      )
+
     except Exception as e:
-      print(f"WebSocket error: {e}")
+      print(f"WebSocket连接错误: {e}")
     finally:
-      if receive_socket.application_state == WebSocketState.CONNECTING:
+      # 清理连接
+      if bailing_socket and not bailing_socket.closed:
+        await bailing_socket.close()
+
+      if receive_socket.application_state != WebSocketState.DISCONNECTED:
         print("关闭websocket")
         await receive_socket.close()
       print("websocket已经关闭", user_id)