https://a.storyblok.com/f/270183/281468/f871591d83/building-a-machine-learning-model-for-answering-machine-detection.png

Criação de um modelo de aprendizado de máquina para a detecção de secretárias eletrônicas

Publicado em May 12, 2021

Tempo de leitura: 6 minutos

Você já precisou de uma maneira de saber quando uma secretária eletrônica estava ocupada em uma chamada Voice? Não? Tudo bem. Eu precisei!

Pré-requisitos

Este artigo pressupõe que você tenha experiência básica em Python, bem como um conhecimento muito básico sobre aprendizado de máquina. Abordaremos alguns conceitos básicos sobre aprendizado de máquina e incluímos links para mais recursos ao longo deste artigo.


Há algumas semanas, recebi uma solicitação de um de nossos engenheiros de vendas sobre um serviço de detecção de secretária eletrônica para um cliente. Eles queriam uma maneira de enviar uma mensagem para a secretária eletrônica quando a chamada fosse encaminhada para o correio de voz.

Pesquisei um pouco sobre isso e parece que é possível, mas não consegui encontrar nada sobre COMO isso foi feito. Então, decidi descobrir por conta própria...

A primeira ideia foi criar um modelo de aprendizado de máquina capaz de detectar quando o beep som em uma secretária eletrônica é ouvido. Nesta postagem, vamos explicar como o modelo foi treinado e implantado em um aplicativo.

Dados de treinamento

Antes de começarmos a construir um modelo de aprendizado de máquina, precisamos ter alguns dados. Para esse problema, precisamos de vários arquivos de áudio com a secretária eletrônica beep , como este:

https://soundcloud.com/user-872225766-984610678/7eaeb600-0202-11e9-bb68-51880c8718e4 ou isto:

https://soundcloud.com/user-872225766-984610678/7eaeb600-0202-11e9-bb68-51880c8718e4

Também precisamos incluir amostras que não contenham o bipe:

https://soundcloud.com/user-872225766-984610678/7eaeb600-0202-11e9-bb68-51880c8718e4

Como esse tipo de dado parece não existir na internet, precisávamos coletar o maior número possível de amostras de bipes e outros sons de chamadas, a fim de treinar nosso modelo. Para isso, criei uma página da web que permite que qualquer pessoa grave sua mensagem de saudação na caixa postal.

Quando você ligar para o número da Vonage, o aplicativo iniciará uma chamada de saída para o mesmo número. Quando a chamada for recebida, basta encaminhá-la diretamente para a caixa postal. A partir daí, gravamos a chamada usando a record ação e salvamos o arquivo em um bucket do Google Cloud Storage. Depois de coletar vários exemplos, podemos começar a analisar os dados.


Em qualquer projeto de aprendizado de máquina, uma das primeiras coisas a fazer é analisar os dados e garantir que sejam adequados para o nosso trabalho.

Como se trata de áudio, não podemos olhar diretamente, mas podemos visualizar os arquivos de áudio usando um espectrograma MEL, que se parece com isto:

Mel Spectogram

Um mel-espectrograma mostra uma faixa de frequências (as mais baixas na parte inferior da tela, as mais altas na parte superior) e indica o nível de intensidade sonora dos eventos em diferentes frequências. Em geral, os eventos mais intensos aparecem claros, enquanto os mais suaves aparecem escuros.

Precisaremos carregar alguns arquivos de ambos os tipos de sons, representá-los graficamente e ver como ficam. Para exibir o espectrograma mel, usaremos um pacote do Python chamado Librosa para carregar a gravação de áudio e, em seguida, plotar o espectrograma mel usando o matplotlib, outro pacote do Python para traçar tabelas e gráficos.

import glob
import librosa
import matplotlib.pyplot as plt
%matplotlib inline

def plot_specgram(file_path):
  y, sr = librosa.load(file_path)
  S = librosa.feature.melspectrogram(y=y, sr=sr, n_mels=128,fmax=8000)
  plt.figure(figsize=(10, 4))
  librosa.display.specshow(librosa.power_to_db(S,ref=np.max),y_axis='mel', fmax=8000,x_axis='time')
  plt.colorbar(format='%+2.0f dB')
  plt.title(file_path.split("/")[-2])
  plt.tight_layout()

sound_file_paths = [
                    "answering-machine/07a3d677-0fdd-4155-a804-37679c039a8e.wav",
                    "answering-machine/26b25bb7-6825-43e7-b8bd-03a3884ed694.wav",
                    "answering-machine/2a685eda-8dd9-4a4d-b00e-4f43715f81a4.wav",
                    "answering-machine/55b654e5-7d9f-4132-bc98-93e576b2d665.wav",
                    "speech-recordings/110ac98e-34fa-42e7-bbc5-450c72851db5.wav",
                    "speech-recordings/3840b850-02e6-11e9-aa3d-ad1a095d8d72.wav",
                    "speech-recordings/55b654e5-7d9f-4132-bc98-93e576b2d665.wav",
                    "speech-recordings/81270a2a-088b-4e3c-9f47-fd927a90b0ab.wav"
                    ]

for file in sound_file_paths:
  plot_specgram(file)

Vamos ver como é cada arquivo de áudio.

Audio files

Dá para perceber claramente qual arquivo de áudio é um beep e qual é apenas speech.


Antes de treinar nosso modelo, vamos utilizar todas as gravações que temos, tanto beeps e os que não são bipes, que estão rotulados como speech, e converteremos cada gravação em um vetor de números, já que nosso modelo aceita apenas números, e não imagens.

Para processar os dados, utilizaremos os coeficientes cepstrais de frequência mel (MFCCs) de cada amostra. Em seguida, salvaremos esse valor em um arquivo CSV para que não precisemos recalcular os MFCCs novamente.

Para cada amostra de áudio, o arquivo CSV conterá o caminho para a amostra de áudio, o rótulo da amostra de áudio (beepou speech), o MFCC e a duração da amostra de áudio (usando a get_duration função do librosa). Também testamos algumas outras características de áudio, incluindo croma, contraste e tonnetz). No entanto, essas características não foram utilizadas na versão mais recente do modelo.

Vamos dar uma olhada agora nas primeiras 5 linhas do arquivo CSV, só para ver como são os dados.

CSV data

Cada linha contém um vetor de 1 dimensão para cada uma das características de áudio. É isso que usaremos para treinar nosso modelo.

Treinamento

Agora vamos usar esses dados para treinar um modelo. Usaremos o pacote Scikit-learn para realizar o treinamento. O Scikit-learn é um ótimo pacote que permite criar modelos simples de aprendizado de máquina sem precisar ser um especialista na área.

Para cada modelo, utilizamos nosso dataframe, que continha o rótulo de cada arquivo de áudio, (beep, speech), com os MFCC de cada amostra, dividimos em um conjunto de dados de treinamento e outro de teste e executamos cada modelo nos dados.

def train(features, model):
  X, y = generateFeaturesLabels(features)
  X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.33, random_state=42)

  model.fit(X_train, y_train)
  print("Score:",model.score(X_test, y_test))

  cross_val_scores = cross_val_score(model, X, y, cv=5, scoring='f1_macro')
  print("cross_val_scores:", cross_val_scores)
  print("Accuracy: %0.2f (+/- %0.2f)" % (cross_val_scores.mean(), cross_val_scores.std() * 2))

  predictions = model.predict(X_test)

  cm = metrics.confusion_matrix(y_test, predictions)
  plot_confusion_matrix(cm, class_names)

  return model

A função train recebe uma lista de características que queremos usar, que são apenas os MFCC da amostra de áudio, bem como o modelo que queremos treinar. Em seguida, exibimos nossa pontuação, que representa o desempenho do modelo. Também exibimos a pontuação da validação cruzada. Isso garante que nosso modelo foi treinado corretamente. A plot_confusion_matrix função traça uma matriz de confusão que mostra exatamente o que o modelo acertou e o que errou.

Confusion Matrix

Em seguida, testamos os modelos a seguir e incluímos sua precisão (pontuação de 0 a 100% que reflete o desempenho do modelo).

Todos esses modelos tiveram um desempenho muito bom, exceto as Máquinas de Vetores de Suporte. O melhor foi o Naive Bayes gaussiano, então vamos usar esse modelo. Em nossa Matriz de Confusão apresentada acima, dos 67 exemplos, 40 amostras que foram classificadas como beep eram, na verdade, beeps, e 22 amostras que foram previstas como speech eram, de fato, speech exemplos. No entanto, 1 exemplo que foi previsto como um beep era, na verdade, speech.

Depois de criarmos nosso modelo, precisamos salvá-lo em um arquivo e, em seguida, importá-lo para nosso aplicativo VAPI.

import pickle
filename = "model.pkl"
pickle.dump(model, open(filename, 'wb'))

Criação do aplicativo

A última etapa consiste agora em integrar nosso modelo a um aplicativo VAPI.

Vamos criar um aplicativo que permita ao usuário discar um número da Vonage. Em seguida, pediremos ao usuário que insira um número de telefone para ligar. Assim que esse número for inserido, vamos conectar essa chamada à conversa atual e nos conectar ao nosso websocket. Usando websockets da Vonage, podemos transmitir a chamada de áudio para o nosso aplicativo.

Primeiro, precisamos carregar nosso modelo no nosso aplicativo.

loaded_model = pickle.load(open("models/model.pkl", "rb"))

Quando o usuário liga pela primeira vez para o número da Vonage, nós retornamos um NCCO com as seguintes informações:

class EnterPhoneNumberHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def get(self):
        ncco = [
              {
                "action": "talk",
                "text": "Please enter a phone number to dial"
              },
              {
                "action": "input",
                "eventUrl": ["https://3c66cdfa.ngrok.io/ivr"],
                "timeOut":10,
                "maxDigits":12,
                "submitOnHash":True
              }

            ]
        self.write(json.dumps(ncco))
        self.set_header("Content-Type", 'application/json; charset="utf-8"')
        self.finish()

Primeiro, enviamos uma ação “Text-To-Speech” na chamada, solicitando que o usuário digite um número de telefone. Quando o número de telefone é digitado, obtemos esses dígitos a partir da https://3c66cdfa.ngrok.io/ivr URL.

class AcceptNumberHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def post(self):
        data = json.loads(self.request.body)
        ncco = [
             {
             "action": "connect",
              "eventUrl": ["https://3c66cdfa.ngrok.io"/event"],
               "from": NEXMO_NUMBER,
               "endpoint": [
                 {
                   "type": "phone",
                   "number": data["dtmf"]
                 }
               ]
             },
              {
                 "action": "connect",
                 "eventUrl": ["https://3c66cdfa.ngrok.io/event"],
                 "from": NEXMO_NUMBER,
                 "endpoint": [
                     {
                        "type": "websocket",
                        "uri" : "ws://3c66cdfa.ngrok.io/socket",
                        "content-type": "audio/l16;rate=16000"

                     }
                 ]
               }
            ]
        self.write(json.dumps(ncco))
        self.set_header("Content-Type", 'application/json; charset="utf-8"')
        self.finish()

Após a inserção do número de telefone, receberemos uma resposta do https://3c66cdfa.ngrok.io/ivr URL. Aqui, pegamos o número de telefone que o usuário inseriu a partir de data["dtmf"] e executamos uma ação de conexão para esse número de telefone e, em seguida, realizamos outra ação de conexão em nosso websocket. Agora, nosso websocket está apto a escutar a chamada.

À medida que a chamada é transmitida pelo WebSocket, precisamos capturar trechos da fala por meio da Detecção de Atividade de Voz, salvá-los em um arquivo WAV e fazer nossas previsões com base nesse arquivo WAV usando nosso modelo treinado.

class AudioProcessor(object):
    def __init__(self, path, rate, clip_min, uuid):
        self.rate = rate
        self.bytes_per_frame = rate/25
        self._path = path
        self.clip_min_frames = clip_min // MS_PER_FRAME
        self.uuid = uuid
    def process(self, count, payload, id):
        if count > self.clip_min_frames:  # If the buffer is less than CLIP_MIN_MS, ignore it
            fn = "{}rec-{}-{}.wav".format('', id, datetime.datetime.now().strftime("%Y%m%dT%H%M%S"))
            output = wave.open(fn, 'wb')
            output.setparams((1, 2, self.rate, 0, 'NONE', 'not compressed'))
            output.writeframes(payload)
            output.close()
            self.process_file(fn)
            self.removeFile(fn)
        else:
            info('Discarding {} frames'.format(str(count)))
    def process_file(self, wav_file):
        if loaded_model != None:
            X, sample_rate = librosa.load(wav_file, res_type='kaiser_fast')
            mfccs = np.mean(librosa.feature.mfcc(y=X, sr=sample_rate, n_mfcc=40).T,axis=0)
            X = [mfccs]
            prediction = loaded_model.predict(X)
            if prediction[0] == 0:
                beep_captured = True
                print("beep detected")
            else:
                beep_captured = False

            for client in clients:
                client.write_message({"uuids":uuids, "beep_detected":beep_captured})

        else:
            print("model not loaded")
    def removeFile(self, wav_file):
         os.remove(wav_file)

Assim que tivermos um arquivo wav, usamos librosa.load para carregar o arquivo e, em seguida, usamos a librosa.feature.mfcc função para gerar o MFCC da amostra. Em seguida, chamamos loaded_model.predict([mfccs]) para fazer nossa previsão. Se a saída dessa função for 0, um beep foi detectado. Se o resultado for 1, então é speech. Em seguida, geramos uma carga JSON indicando se um beep foi detectado e os UUIDs da conversa. Dessa forma, nosso aplicativo cliente pode enviar um TTS para a chamada, usando os UUIDs.

Cliente WebSocket

A etapa final consiste em criar um cliente que se conecte ao WebSocket, monitore quando um bipe for detectado e envie uma mensagem de TTS para a chamada, assim que a mensagem de voz for detectada.

Fonte

Primeiro, precisamos nos conectar ao WebSocket.

ws = websocket.WebSocketApp("ws://3c66cdfa.ngrok.io/socket",
on_message = on_message,
on_error = on_error,
on_close = on_close)
ws.on_open = on_open

ws.run_forever()

Em seguida, basta aguardarmos qualquer mensagem recebida pelo nosso WebSocket.

def on_message(ws, message):
    data = json.loads(message)
    if data["beep_detected"] == True:
        for id in data["uuids"]:
            response = client.send_speech(id, text='Answering Machine Detected')

        time.sleep(4)
        for id in data["uuids"]:
            try:
                client.update_call(id, action='hangup')
            except:
                pass<a href="https://www.nexmo.com/wp-content/uploads/2019/02/amd-confusion-matrix.png"><img src="https://www.nexmo.com/wp-content/uploads/2019/02/amd-confusion-matrix-600x300.png" alt="" width="300" height="150" class="alignnone size-medium wp-image-28012" /></a>

<a href="https://www.nexmo.com/wp-content/uploads/2019/02/amd-df.png"><img src="https://www.nexmo.com/wp-content/uploads/2019/02/amd-df-600x300.png" alt="" width="300" height="150" class="alignnone size-medium wp-image-28015" /></a>

<a href="https://www.nexmo.com/wp-content/uploads/2019/02/amd-eda.jpg"><img src="https://www.nexmo.com/wp-content/uploads/2019/02/amd-eda-600x300.jpg" alt="" width="300" height="150" class="alignnone size-medium wp-image-28018" /></a>

Vamos analisar a mensagem recebida como JSON e, em seguida, verificar a beep_detected propriedade True. Se for, então uma beep foi detectada. Em seguida, enviaremos uma mensagem de TTS para a chamada dizendo “Secretária eletrônica detectada” e, depois, executaremos uma hangup ação na chamada.


Conclusão

Mostramos como criamos um modelo de detecção de secretária eletrônica com 96% de precisão, utilizando algumas amostras de áudio de beeps e speech para treinar nosso modelo. Esperamos ter demonstrado como você pode usar o aprendizado de máquina em seus projetos. Aproveite!

Compartilhar:

https://a.storyblok.com/f/270183/150x150/a3d03a85fd/placeholder.svg
Tony HungEx-funcionários da Vonage

Desenvolvedor iOS que se tornou entusiasta de Ciência de Dados e Aprendizado de Máquina. Quero que as pessoas entendam o que é o aprendizado de máquina e como podemos usá-lo em nossas Applications.