如何在Python异步函数中传参，动态配置Deepgram识别语言

一、修复WebSocket URL携带参数的问题 #

1. 前端传递参数 #

在WebSocket连接时，将选中的语言码拼接到URL中：

// 假设selectedLang是用户选择的语言（如'de'、'en'）
const socket = io.connect(`/ws?language=${selectedLang}`);

2. 后端接收并传递参数 #

如果使用Flask-SocketIO，在连接事件中捕获参数并存入会话，确保异步函数能访问：

from flask import request, session
from flask_socketio import socketio, emit

@socketio.on('connect')
def handle_connect():
    # 从URL查询参数获取语言，默认英语
    language = request.args.get('language', 'en')
    # 存入session，供后续异步处理函数使用
    session['deepgram_language'] = language

# 处理音频流的异步函数，需复制请求上下文确保能访问session
@socketio.on('audio_chunk')
def handle_audio_chunk(chunk):
    @copy_current_request_context
    async def process_audio():
        # 从session获取语言参数
        lang = session.get('deepgram_language', 'en')
        # 初始化Deepgram并传入语言配置
        deepgram = Deepgram(os.getenv('DEEPGRAM_API_KEY'))
        config = {'language': lang, 'punctuate': True}
        
        async with deepgram.transcription.live(config) as dg_session:
            # 音频流处理逻辑
            dg_session.send(chunk)
            # 监听转录结果
            async for result in dg_session:
                emit('transcript', result.channel.alternatives[0].transcript)
    
    asyncio.create_task(process_audio())

如果使用原生WebSocket，通过闭包传递参数：

from flask import request
import websockets

@app.route('/ws')
def ws_endpoint():
    language = request.args.get('language', 'en')
    
    async def ws_handler(websocket):
        deepgram = Deepgram(os.getenv('DEEPGRAM_API_KEY'))
        config = {'language': language}
        async with deepgram.transcription.live(config) as dg_session:
            # 处理WebSocket消息和音频流
            while True:
                audio_data = await websocket.recv()
                dg_session.send(audio_data)
                result = await dg_session.next()
                await websocket.send(result.channel.alternatives[0].transcript)
    
    return serve(ws_handler, host='0.0.0.0', port=5001)

二、更优的动态语言配置方案 #

1. 登录用户默认语言持久化 #

将用户偏好语言存入数据库，连接WebSocket时自动读取：

@socketio.on('connect')
def handle_connect():
    user_id = session.get('user_id')
    if user_id:
        # 从数据库获取用户预设语言
        user = User.query.get(user_id)
        language = user.preferred_language or 'en'
    else:
        language = request.args.get('language', 'en')
    session['deepgram_language'] = language

2. 转录中动态切换语言 #

无需断开WebSocket，通过消息触发语言切换并重启Deepgram会话：

前端发送切换请求

function switchLanguage(newLang) {
    socket.emit('switch_language', { language: newLang });
}

后端处理切换逻辑

@socketio.on('switch_language')
def handle_switch(data):
    new_lang = data.get('language', 'en')
    session['deepgram_language'] = new_lang
    # 通知前端已切换，可重新发送音频流
    emit('language_updated', {'status': 'success', 'language': new_lang})

此时异步处理函数会在下一次音频流处理时，使用更新后的session参数重新创建Deepgram会话。

三、排查session传递失败的常见原因 #

• 确保Flask配置了SECRET_KEY，session才能正常工作
• 异步函数需用copy_current_request_context装饰，复制当前请求上下文以访问session
• 使用Flask-SocketIO时，需正确配置异步模式（如async_mode='eventlet'）

内容的提问来源于stack exchange，提问作者AcastaTeacup