[Whisper|Pyannote] 상담사 통화녹음 화자분리

문제 상황

회사에서는 상담 센터에서 수많은 양의 상담 전화를 받고 있다. 이 상담 전화를 STT whisper로 받아 프로그램을 돌리고 있는데, 사실상 이 서비스가 빛을 발하지 못하고 있다.

• 밤새 돌려놓으면 2000건 까지는 돌릴 수 있으나 그 이상의 데이터(4000건 까지 들어옴)를 돌리지 못하는 어려움이 있음
• whisper 에 화자분리는 지원이 되지 않아 해당 script의 화자 파악에 어려움이 있음✔️
• 통화 음질이 아주 나빠서 정확도가 너무 떨어짐✔️
• 이 STT 자료들을 조금 더 의미있게 사용하기를 원한다. 예를 들어 상담원 A와 상담원 B의 언어적 특징이나, 상담원들이 많이 사용하는 단어들을 추출한다거나 등 등 ..

 

내가 해볼 수 있는 기술적 영역에서 해결해볼 수 있는 것들?

• Kafka 등을 이용해 병렬처리, 멀티 프로세싱을 구축해볼 수 있겠다.
• STT와 목소리 화자인식을 매핑시켜 시간당 누가 이야기하는지 추출해볼 수 있겠다.✔️
• 도메인 특성(금융권) 상 많이 사용되는 단어들이 존재한다. 이걸로 Text 자체를 좀 더 튜닝해볼 수 있겠다.
• 여러 툴들을 이용해 통화 음질을 향상시킬 수 있겠다.(속도 조절, 노이즈 제거, 에코 제거, 음향 증가 등)✔️
• ELK를 이용해 이 데이터들을 가공해서 의미있는 자료를 만들 수 있을 것 같다.

 

위에 생각해본 것들 중에 체크한 내용들을 해결해 보았고 해당 내용을 아래에 담아보았다. !

---

화자분리

우선 화자분리가 가장 필요한 기능이라 이것부터 진행하였다. Whisper에서는 화자분리를 따로 지원하지 않아 whisperX 라이브러리를 사용해서 진행해본 과제가 있었는데 정확도가 현저히 낮아 사용이 제한되고 있었다.

우선 화자분리를 위한 ai 모델이나 api를 먼저 탐색하였다. 내가 사용해볼 수 있는 툴은 UIS_RNN ai 모델과 pyannote라는 api가 있었다. (결론부터 말하면 UIS_RNN 모델은 정확도 측면에서 매우 떨어진다 ㅠㅠ)

---

UIS_RNN 모델

Colab 개발환경에서 실시하였습으며 google/ris-rnn 모델을 이용해 학습시키고 결과값을 도출했습니다.
1. 우선 실습을 위해 uisrnn과 easydict 라이브러리를 설치해줍니다.

2. 다운로드한 데이터에서 데이터를 받아와 sequence와 label을 분리해줍니다.

3. 모델의 argments를 지정해줍니다. github repo 의 uisrnn/arguments.py 를 참조하여 args의 정의에 대해 알아볼 수 있습니다.

4. UIS_model을 생성하고 모델을 훈련합니다.(fit 과정) GPU 없이 iteration 5만 번 돌려서 5시간 걸렸어요..

5. 내가 가진 wav 파일을 train_sequence에 맞추어 저장합니다.
scipy.io 패키지에서 wavfile이라는 모듈을 임포트 해오면 wavfile.read("파일 이름") 함수를 통해 쉽게 .wav 파일을 수치화한 형태로 읽을 수 있습니다.

github 레포에서도 설명되어있는데 train_sequence.shape의 앞단은 상관없지만 뒷단(256) 크기에는 맞춰주어야 하기 때문에 프레임이 frame_size에 맞도록 패딩시켜주어야 합니다. 원하는 프레임크기(256)에 맞추어 만약 더 적으면 그냥 0으로 채워주었습니다.

6. Prediction !!

자 이제 예측을 실시해보면 !!?? 

생각보다 값이 너무 정확하지 않게 나온다... 숫자는 화자를 의미하는데(0번 화자, 11번 화자, 23 번 화자 ..) 두 명이 말한는 음성녹음에 25명이 참여해 버렸다.

그럼 두 명의 스피커로 argument를 지정해 predict 하면 되지 않을까!? 싶어 ISSUE를 찾아봤더니

Issue#78 에서 확인할 수 있는데,, 그런 기능은 지원하지 않는다고 한다. 심지어 이 기능을 PR로 기여한 사람이 있었는데 해당 PR은 모델 학습 측면에서도 정확성이 높아지지 않는다고 판단하여 merge 시켜주지 않았다 ㅠㅠ

 

7. 결론

여기에서 사용하는 training_data.npz가  AI Hub 음성 데이터 와 양식이 유사하기 때문에 우리나라 버전의 더 많은 양의 데이터로 학습하면 좀 나아질지도 모르겠다 !
하지만 내 프로젝트 상 2명보다 많은 화자가 나오는 상황은 거의 존재하지 않아 결국 이 방법은 파기...

---

Pyannote API

UIS_RNN에 실망한 마음을 저버리지 않고 Pyannote라는 툴을 한 번 살펴보았다. 다행히 화자 수 지정이 가능하였고, 모델이 아니라 API여서 더 쉽게 적용할 수 있겠다 ! 정확도도 나쁘지 않았다.

오 나름 정확도 굿!?

Pyannote를 사용하기 전 살펴본 문제 상황 ??

1. Whisper text는 문장별로 split된 형태로 나온다.
2. pyannote는 화자가 겹쳐지는 곳이 있다면 둘 다 해당하도록 나온다.

series 객체에 series 를 매핑.. 가장 오차가 적은 방법이 무엇일까? 에 대해 고민해보았다. 

해결방법 

whisperX 로직과 다르게 ‘시간’상의 기록에 집착하지 않고 ‘화자’에 집중하여 선택할 수 있도록 하기 위해 우선순위큐로 시간 순서대로 화자별 말하는 시간을 꺼내고, 화자별로 말하는 시간에 속하는 영역을 자료구조로 %로 나타냈다.

이후 더 정확도가 높은 whisper 자료의 시간과 비교한 후 speaker00 이 속하는 정도와 speaker01이 속하는 정도를 비교하여 더 크게 부합하는 곳을 찾아서 걔로 선정한다.
그리고 대화 특성 상 밀리 초에 화자가 확확 바뀔 수 있기 때문에 그 자료를 나타내기 위해 리스트 index는 ‘초(second)’로 할당하고 0~(whisper에서 나오는 최대 시간. 많아봤자 20분). index를 한 번 더 2개로 나누어 그 값은 소수점, 즉 ‘밀리초(millisecond)’로 할당하여 시간의 정확도를 기록해두었다.

약간 그림으로 표현하자면 이런 느낌..??

위의 상황을 글로 표현하자면 아래와 같다.
[ 시작점 중 뒤에서부터 할당되는 만큼의 확률 | 0.5 ] [ 0.5 | 끝나는 점 중 앞에서부터 할당되는 만큼의 확률 ]

또다른 예시로는 2.6~5.2 초동안 말한다면 다음과 같이 표현할 수 있다.
[0 |시작점 분수] [0.5|0.5] [0.5|0.5] [끝나는 점 분수| 0]

시작하는 곳이 0.5 초반일 경우/0.5 후반일 경우, 그리고 끝나는 곳이 0.5 초반일 경우/0.5 후반일 경우로 나누어서 각 확률을 채울 수 있도록 하였다. A사람이 말하고 B 사람이 말하고 다시 A사람이 말하는, 즉 나의 차례가 돌아오는 텀이 0.5보다 작은 경우는 거의 없다고 판단하여 이런 식으로 구현해보았고 정확도를 확실히 높일 수 있었다.

 

음성녹음 전처리

파이썬 차원에서 쉽게 노이즈를 제거하고 볼륨을 높일 수 있다.

from pydub import AudioSegment
from pydub.playback import play

# 오디오파일 가져오기
audio = AudioSegment.from_file("./20221118 배달의민족 고객센터 상담 음성.mp3")

# 노이즈 제거
audio = audio.low_pass_filter(3000)  # Example low-pass filter

#-------------------------------------------#

# 볼륨 높이기 + 5 dB
more_volume = audio + 5
more_volume.export("./20221118 배달의민족 고객센터 상담 음성_new.wav", format="wav")

코드 작성

데이터 가공 과정은 생략하고 메인으로 사용하는 로직만 가져왔습니다.

import math
import heapq
import copy

# 시간 문자열을 초로 변환하는 함수
def time_to_seconds_pyannote(time_str):
    h, m, s = map(float, time_str.split(':'))
    return (h * 3600 + m * 60 + s)
def time_to_seconds_whisper(time_str):
    m,s = map(float, time_str.split(':'))
    return (m * 60 + s)
def parse_time_data(time_str):
    time_str = time_str.strip("[]").split(" --> ")
    start_time = time_str[0]
    end_time = time_str[1]
    return start_time, end_time


ListA = [] #whisper
for i in whisper_intervals:
  start, end = parse_time_data(i)
  ListA.append([time_to_seconds_whisper(start), time_to_seconds_whisper(end)])
ListB = []
for j in speaker_00_intervals:
  start, end = parse_time_data(j)
  ListB.append([time_to_seconds_pyannote(start), time_to_seconds_pyannote(end)])
ListC = []
for k in speaker_01_intervals:
  start, end = parse_time_data(k)
  ListC.append([time_to_seconds_pyannote(start), time_to_seconds_pyannote(end)])

#ListB와 ListC를 우선순위큐로
heapq.heapify(ListB)
heapq.heapify(ListC)

#영역을 저장할 area_B, area_C init
area_B = []
area_C = []

area_B=[[0,0] for _ in range(math.ceil(ListA[-1][1])+1)]
area_C=[[0,0] for _ in range(math.ceil(ListA[-1][1])+1)]

#result = 저장해둘 곳 ListA와 크기 같아야 함
result = ['A or B' for _ in range(len(ListA))]
index_B=0

while ListB :
  start, end = heapq.heappop(ListB)
  area_B[math.floor(start)+1:min(math.floor(end), math.floor(ListA[-1][0]))] = [[0.5,0.5] for _ in range(min(math.floor(end), math.floor(ListA[-1][0]))-math.floor(start)-1)]
  if (start-int(start))>0.5:
    area_B[math.floor(start)][1] = max(0.5-(start-int(start)), area_B[math.floor(start)][1])
  else:
    area_B[math.floor(start)][1] = 0.5
    area_B[math.floor(start)][0] = max((start-int(start)), area_B[math.floor(start)][0])
  if (end-int(end)) < 0.5:
    area_B[math.floor(end)][0] = max((end-int(end)), area_B[math.floor(end)][1])
  else:
    area_B[math.floor(end)][0] = 0.5
    area_B[math.floor(end)][1] = max((end-int(end))-0.5, area_B[math.floor(end)][1])


while ListC :
  start, end = heapq.heappop(ListC)
  area_C[math.floor(start)+1:min(math.floor(end), math.floor(ListA[-1][0]))] = [[0.5,0.5] for _ in range(min(math.floor(end), math.floor(ListA[-1][0]))-math.floor(start)-1)]
  if (start-int(start))>0.5:
    area_C[math.floor(start)][1] = max(0.5-(start-int(start)), area_C[math.floor(start)][1])
  else:
    area_C[math.floor(start)][1] = 0.5
    area_C[math.floor(start)][0] = max((start-int(start)), area_C[math.floor(start)][0])
  if (end-int(end)) < 0.5:
    area_C[math.floor(end)][0] = max((end-int(end)), area_C[math.floor(end)][1])
  else:
    area_C[math.floor(end)][0] = 0.5
    area_C[math.floor(end)][1] = max((end-int(end))-0.5, area_C[math.floor(end)][1])

for j in range(len(ListA)):
  start=ListA[j][0]; end=ListA[j][1]
  # 밀리세컨드까지의 확률이 B가 더 높을 경우
  temp_B=0; temp_C=0;
  for i in range(math.floor(start),math.ceil(end)):
    temp_B += sum(area_B[i])
    temp_C += sum(area_C[i])
  print(temp_B, temp_C)
  if temp_C*0.9 < temp_B : result[j]='A'
  else: result[j]='B'

# 한 번 더 가공
for i in range(1,len(ListA)-1):
  if ListA[i][1]-ListA[i][0]<=1.0 :
    standard = result[i]
    if result[i-1]==result[i] and result[i+1]==result[i]:
      if result[i]=='A': result[i]='B'
      else: result[i]='A'

result_data = copy.deepcopy(text_data)
temp=0
for i in range(len(result_data)):
  result_data[i]=list(result_data[i])
  if result[i]==temp: print(result_data[i][1], end = ' ')
  else: print('\n\n',result[i], result_data[i][1], end = '')
  temp = result[i]

+ 확률적으로 상담사가 먼저 말을 시작하는 경우가 많고 말을 더 많이 하는 경우가 많은 점, 그리고 짧게 말하는 시간의 경우 pyannote에서 화자구분을 잘 못하는 점을 반영하여 후처리하였습니다.

---

결과값

아래 데이터는 공개된 배달의 민족 데이터로 실습한 내용입니다.

기존 whisperX 

[0.308 --> 9.931] SPEAKER_01: 드릴 상담사는 누군가의 소중한 가족입니다. 욕설 또는 폭언을 할 경우 상담이 종료될 수 있습니다. 상담사를 연결해 드리겠습니다.
[10.131 --> 12.872] SPEAKER_02: 네, 반갑습니다. 배달의민족입니다.
[13.112 --> 20.694] SPEAKER_00: 그러니까 이거를 임시 조치를 30일 대기실이 아니라 지금 당장 대기실에 할 수 없는가 해서요. 문제가
[20.914 --> 22.975] SPEAKER_02: 없는데 왜 30일
[23.155 --> 23.495] SPEAKER_00: 동안에...
[23.971 --> 26.332] SPEAKER_02: 지금 바로 재개시는 불가합니다, 고객님.
[27.153 --> 27.353] SPEAKER_01: 네.
[27.373 --> 27.413] SPEAKER_00: 그
[27.453 --> 30.535] SPEAKER_02: 내용으로 그러면 추가적으로 설명을 좀 드려도 괜찮으실까요?
[30.695 --> 31.755] SPEAKER_00: 예예, 설명해주세요. 네,
[32.436 --> 133.9] SPEAKER_02: 확인 감사합니다. 말씀을 해주셨던 30년 불만의 각괄이라고 하는 가계도 우선 확인은 되어서 설명을 드려볼 텐데 우선 리뷰 작성에 대한 부분으로는 이용을 하시는 내용에 자유롭게 고객님 개인적인 의견을 작성하시는 거는 맞습니다.

파인튜닝 이후 

B 연결해드릴 상담사는 누군가의 소중한 가족입니다.욕설 또는 폭언을 할 경우 상담이 종료될 수 있습니다. 상담사를 연결해드리겠습니다. 네 반갑습니다. 배달의 민족 김재우입니다. 

 A 그러니까 이거를 임시 조치를 30일에 재개시가 아니라지금 당장 재개시할 수 없는가 해서요. 문제가 없는데 왜 30일 동안... 

 B 내용을 설명을 드리려고 한 거였는데요.

 A 아 예.

 B 지금 바로 재개시는 불가합니다. 고객님.

 A 네.

 B 내용으로 그러면 추가적으로 설명을 좀 드려도 괜찮으실까요?

 A 예예. 설명해주세요.

 B 네. 확인 감사합니다.말씀을 해주셨던 30년 불만의 각괄이라고 하는 가계도 우선 확인이 되어서 설명을 드려볼 텐데. 

 A 네네.

 B 우선 리뷰 작성에 대한 부분으로는 이용을 하시는 내용에자유롭게 고객님 개인적인 의견을 작성하시는 거는 맞습니다.

---

결론

애초에 이 과제는 정확도를 더 높이는 데에 중점을 둔 것이었어서, 사수 분께서도 ‘그냥 대략적으로 화자 분리 할 수 있게끔 구현해 봐! ’ 라고 부탁하신 내용이었다. 물론 내가 구현한 내용보다도 더 정확도가 높은 방법이 있을 수 있겠지만 그래도 정확도도 잘 나오고 빠르게 결과값이 도출되어서 이 과제로 전체 발표도 진행하여 다른 부서들 분께 공유한다고 하셨다 Vv 아주굿 !