如何用Python实时动态处理Twilio通话并实现条件转接？

我正在开发一款具备以下功能的应用：

• 发起外呼通话
• 被叫方接听后启动流媒体传输
• 使用Vosk实时转录流内容
• 检测关键词以区分两种不同的自动应答
• 识别到正确的自动应答后，将通话转接至我的个人号码

我卡在第5步，不知道如何在通话进行中拨打我的个人号码。我需要条件性使用<Dial>动词，但目前不清楚怎么实现。

技术背景 #

我使用的技术栈包括：Twilio Functions、支持WebSocket的Flask服务器，以及ngrok用于将本地主机暴露给Twilio。

模块1：向Twilio发送请求的逻辑 #

from twilio.rest import Client
from twilio.twiml.voice_response import VoiceResponse, Start
from twilio.request_validator import RequestValidator
from server import file_path
import os
import requests
import xml.etree.cElementTree as ET

# 全局变量
client = Client(account_sid, auth_token)
# Twilio函数URL
call_test_url = "https://my_function_url.com"
# 安全验证
validator = RequestValidator(auth_token)

def call_request():
    # 从文件读取ngrok公网URL
    if os.path.exists(file_path):
        with open(file_path, "r") as file:
              public_url = file.read().strip()

        stream_url = public_url.replace("https", "wss") + "/stream"
        # 动态生成Twiml的工具函数
        stream_twiml = generate_stream_twiml(stream_url, "15")

        request_params = {
            "to" : "...",
            "from" : "...",
            "callback_url" : public_url + "/callstatus",
            "twiml" : stream_twiml
        }
        twilio_signature = validator.compute_signature(call_test_url, request_params)
        request_headers = {
            "X-Twilio-Signature" : twilio_signature
        }
        my_call_request = requests.post(call_test_url, headers=request_headers, data=request_params)
        print(my_call_request.content)

        return my_call_request
    
    else:
        print("未找到存储公网端点的文件。")

if __name__ == "__main__":
    call_request()

模块2：Flask服务器代码 #

from flask import Flask, request
from flask_sock import Sock
from vosk import SetLogLevel
import time
import audioop
import base64
import json
import os
import vosk
import logging

port = 1234
file_path = "存储ngrok公网URL的文件路径"

app = Flask(__name__)
sock = Sock(app)
log = logging.getLogger('werkzeug')
log.disabled = True
model = vosk.Model('my-model')

CL = '\x1b[0K'
BS = '\x08'

@app.route('/callstatus', methods=['POST'])
def get_call_status():
    from twilio_script import call_request
    call_status = request.form.get('CallStatus')
    call_sid = request.form.get('CallSID')
    print(f"通话状态: {call_status}")
    if call_status == "busy":
        print("被叫方忙线，重新呼叫...")
        time.sleep(0.5)
        call_request()

    return call_status

@sock.route('/stream')
def stream(ws):
    """接收并转录音频流"""
    rec = vosk.KaldiRecognizer(model, 16000)
    while True:
        message = ws.receive()
        packet = json.loads(message)
        if packet['event'] == 'start':
            print('流媒体传输开始')
        elif packet['event'] == 'stop':
            print('\n流媒体传输结束')
        elif packet['event'] == 'media':
            key_string = "..." # 用于区分自动应答的关键词
            audio = base64.b64decode(packet['media']['payload'])
            audio = audioop.ulaw2lin(audio, 2)
            audio = audioop.ratecv(audio, 2, 1, 8000, 16000, None)[0]
            if rec.AcceptWaveform(audio):
                r = json.loads(rec.Result())
                full_output = CL + r['text'] + ' '
                print("\n 转录结果: ", full_output)

                if key_string in full_output:
                   print("识别到目标应答，开始转接通话...")
                   # 此处需要添加转接逻辑
                   break
            else:
                r = json.loads(rec.PartialResult())
                partial_output = CL + r['partial'] + BS * len(r['partial'])
                print("\n 临时转录结果: ", partial_output)
                
                if key_string in partial_output:
                   print("识别到目标应答，开始转接通话...")
                   # 此处需要添加转接逻辑
                   break
    
if __name__ == "__main__":
    if os.path.exists(file_path):
        print("删除旧文件...")
        os.remove(file_path)

    from pyngrok import ngrok
    public_url = ngrok.connect(port, bind_tls=True).public_url

    with open(file_path, "w") as file:
        file.write(public_url)

    app.run(port=port, debug=True, use_reloader=False)

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