[LLM] 화상회의 중 STT to TTS 수행하는 시스템 설계 - 2. 실시간 STT와 번역이 가능한 시스템 구현(+ FastAPI 모델서빙)

Special thanks to. 다건🙌

 

 

이전 포스팅: [LLM] 화상회의 중 STT to TTS 수행하는 시스템 설계 - 1. OpenAI API 'Whisper-1' 활용하여 실시간 STT 구현

 

이전에 만들었던 음성 녹음 + 실시간 STT 코드를 가지고

녹음 시작 버튼을 누르면 중단 없이 프로그램이 자체적으로 문장의 끝을 판단하여 한 문장씩 전사와 번역을 수행하는 프로그램을 설계했다.

사실 FE/BE 설정은 거의 다건이가 다 해줬고,

음성 관련 트러블슈팅 부분만 내가 발 얹었다. 감사합니다👍

 

사진은 5초 정도 길이의 문장을 3초만에 전사하고 -> 곧이어 1초 내로 번역이 완료되는 실행 결과 모습이다.

 

 

프론트 못함 + 저게 메인 프로젝트가 아니었음 이슈로 보여주는 완성도를 생각하지 않고 기능에만 집중해서 구현하고자 했기 때문에

이전 콘솔 화면처럼 실시간 전사되는 모습은 확인이 안된다

 

맞다. 변명이다. 암튼 실시간이다.

?? : 실시간의 의미는 자신이 정의함에 따라서 달라.

사용 모델

- STT: whisper

- Translation: gpt-4o-mini

- TTS(여기선 언급 안함): gpt-4o-mini-tts

 

토큰을 최대한 절약하면서 실시간성을 확보하기 위해 gpt-4o나 tts-1같은 모델보단 작은 모델들을 사용하였다. 

엄청 방대한 양의 대화를 녹음하거나, 복잡한 처리가 필요한 과정이 없었기 때문에 mini 모델로도 충분히 잘 돌아갔다.👍

 

전체 진행 플로우

대충그린기린그림

1. 처음 웹에 들어왔을 때, '녹음 시작'을 누르면 웹소켓이 연결되어 2초 단위 청크로 webm blob을 fastAPI로 전송한다.

 

2. fastAPI는 실행 시 스레드를 가동시켜 각 전사 큐 / 번역 큐에 내용이 들어오면 바로 작업을 수행할 수 있도록 하였고,

websocket이 연결되면 전달받은 .webm 파일을 .wav로 변환하고, audio_queue에 값을 넣어준다.

 

3. stt_processing_thread()에서는 audio_queue에 값이 들어오면 전사 작업을 수행해서 전사된 텍스트를 전사 결과 큐에 넣어주고

translation_thread()에서는 전사 결과 큐에 값이 들어오면 번역을 해서 번역 결과 큐에 결과를 넣어준다.

 

4. 동일하게 fastAPI의 메인 라우터에서 비동기적으로 계속 전사 결과 큐와 번역 결과 큐를 확인하며

값이 들어올때마다 websocket으로 웹에 각 결과를 전송하면, 결과가 화면에 보여진다.

 

5. 사용자가 그만 갖고 놀고싶어서 '녹음 중단'을 눌러 웹소켓이 닫힐때까지 반복해서 진행된다.

 

파일 구조

project/
├── config.py           # 전역 설정 (API 키, 오디오 설정, 전역 클라이언트 등)
├── main.py             # 프로그램의 진입점: 각 모듈을 불러와 스레드 실행
└── modules/
|   ├── __init__.py
|   ├── stt.py          # STT 처리 (Whisper-1, VAD, 언어 자동 감지)
|   ├── translation.py  # 번역 처리 (GPT-4o-mini 사용)
|   ├── tts.py          # TTS 처리 (GPT-4o-mini-tts 사용)
|   └── utils.py        # 공통 유틸리티 함수 (언어 보정, 로그 파일명 생성 등)
└── templates/
    └── index.html      # 테스트용 간단한 프론트 코드

---

코드

프론트에서 꼭 해줘야 할 일!!!

사실 이 프로젝트 하면서 느낀건 프론트가 제일 중요했다.... 정말 너무힘들었다😭

프론트단에서 제대로 된 데이터를 보내주지 않으면, 아무리 파이썬 쪽에서 코드를 기깔나게 구성했어도 아무런 작동 결과를 확인할 수 없다.

 

 

해당 프로젝트에서 주의할 것은 webm파일을 청크단위로 보낼 때 '헤더 정보'를 반드시 포함해서 보내야 하는 것이다.

 

 

그러나, 최초 1회 전송되는 파일에만 헤더 정보가 붙고 그 이후에는 녹음된 데이터만 전송되는데, 그럼 ffmpeg 모듈이 제대로 된 데이터가 아니라고 난 이거 변환 못한다며 미쳐 날뛴다.

이를 해결하기 위한 방법은

 

1. 헤더 정보를 변수에 지정해놓고, 매 blob에 헤더 정보를 붙여 보내준다.

2. 그냥 녹음을 아예 청크마다 새로 한다.

 

 

였는데, webm의 헤더정보는 고정길이가 아니라 매번 크기가 달라질 수 있으므로 2번이 무식해보이지만 생각보다 잘먹혔다.

근데 그럼

어? 그럼 중간중간 녹음이 안될 수 있는거 아닌가?

 

싶기도 했지만, 실제 실행 결과 다시 녹음기를 껐다 키는 과정은 굉장히 빠른 시간안에 이루어졌고,

그 과정에서 대단한 내용 손실이 발생하지도 않았기 때문에

2초마다 녹음파일을 한번씩 보내주는 방식을 선택하게 되었다.

 

function startRecording() {
  const chunks = [];
  
  mediaRecorder = new MediaRecorder(stream, { mimeType: 'audio/webm' });
  
  mediaRecorder.ondataavailable = (event) => {
    if (event.data.size > 0) {
      chunks.push(event.data);
    }
  };
  
  mediaRecorder.onstop = () => {
    if (chunks.length > 0 && ws && ws.readyState === WebSocket.OPEN) {
      const blob = new Blob(chunks, { type: 'audio/webm' });
      ws.send(blob);
      console.log('녹음 데이터 전송됨, 크기:', blob.size);
    }
  };
  
  mediaRecorder.start();
  console.log('새 녹음 세션 시작');
}

 

startRecording()에서는 audio/webm 지정해줘서 MediaRecorder 사용하여 녹음을 수행하고,

아래서 설정해주는 Interval 시간마다 blob에 데이터를 담아 웹소켓으로 전송해준다.

// 마이크 스트림 얻기
stream = await navigator.mediaDevices.getUserMedia({ audio: true });
// WebSocket 연결 (HTTP 환경이면 "ws://", HTTPS이면 "wss://")
ws = new WebSocket(`ws://${window.location.host}/ws/stt`);

ws.onopen = () => {
  console.log('WebSocket 연결됨');
  status.textContent = '상태: WebSocket 연결됨';

  // 녹음 시작
  startRecording();
  
  // 3초마다 새로운 녹음 세션 시작
  recordingInterval = setInterval(() => {
    if (isRecording) {
      stopRecording();
      startRecording();
    }
  }, 2000);
// 이후 코드

 

main

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    # stt, 번역 스레드 시작
    threading.Thread(
        target=stt_processing_thread,
        args=(audio_queue, sentence_queue, transcription_queue, recording_active, "ko"),
        daemon=True
    ).start()
    threading.Thread(
        target=translation_thread,
        args=(sentence_queue, translation_queue, translated_queue, "en"),
        daemon=True
    ).start()

    asyncio.create_task(result_sender_task())

 

on_event 부분에서 fastAPI 실행 시 어떤 것들이 실행되어야 하는지 정의해주었다.

 

"startup" 이벤트 실행이 시작되면, 

두 개의 스레드 실행

- stt_processing_thread 호출

- translation_thread 호출

결과 전송해주는 asyncio 코루틴 정의

가 수행된다.

 

참고로 on_event는 deprecated 된 기능이기 때문에, FastAPI 0.95+ 버전부터는 lifespan 이벤트 처리 방식을 사용하는걸 권장한다. 보통은 lifespan 함수를 정의해서 쓰지만, 그냥 if name=='__main__' 안에 정의해줘도 상관은 없다. 근데 나는 그냥 실행에 문제없어서 저대로 썼다.

 

# STT WebSocket 엔드포인트
@app.websocket("/ws/stt")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    print("[DEBUG] WebSocket 연결됨")
    websocket_clients.append(websocket)

    try:
        while True:
            data = await websocket.receive_bytes()
            # 임시 파일로 저장
            with tempfile.NamedTemporaryFile(suffix=".webm", delete=False) as temp_file:
                temp_file_path = temp_file.name
                temp_file.write(data)
            print(f"[DEBUG] 임시 파일 생성됨: {temp_file_path}")
            
            try:
                # ffmpeg로 .webm → .wav 변환 (BytesIO 형태)
                wav_buffer = convert_webm_to_wav_bytes(temp_file_path)
                os.unlink(temp_file_path)  # 임시 파일 삭제

                if wav_buffer is None:
                    print("[DEBUG] 변환 실패한 청크 건너뜀")
                    continue

                # BytesIO 버퍼에서 오디오 데이터 읽기
                wav_buffer.seek(0)
                audio_data, sample_rate = sf.read(wav_buffer, dtype='float32')
                print(f"[DEBUG] 수신된 오디오 데이터: shape {audio_data.shape}, sample_rate {sample_rate}")

                # 스테레오면 모노로 변환
                if len(audio_data.shape) > 1:
                    audio_data = audio_data.mean(axis=1).reshape(-1, 1)
                else:
                    audio_data = audio_data.reshape(-1, 1)

                # audio_queue에 추가 (STT 처리 스레드로 전달)
                audio_queue.put(audio_data)
                print(f"[DEBUG] audio_queue에 데이터 추가됨. 현재 queue 크기: {audio_queue.qsize()}")
            except Exception as e:
                print(f"[DEBUG] 오디오 처리 오류: {e}")
                if os.path.exists(temp_file_path):
                    os.unlink(temp_file_path)
    except WebSocketDisconnect:
        print("[DEBUG] WebSocket 연결 종료됨")
        if websocket in websocket_clients:
            websocket_clients.remove(websocket)

 

가장 중요한 데이터 받아오는 부분..

웹소켓 사용 + 동시다발적으로 음성을 받아오면 받아오는 대로 큐에 넣는 작업을 수행하기 위해 비동기 함수로 위와 같이 코드를 짰다.

 

프론트단에서 음성 데이터를 웹소켓으로 FastAPI에 쏴주면, 이를 stt모델인 whisper에 전달해주기 위해 변환을 수행한다.

whisper 모델은 .wav를 입력으로 받기 때문에 webm으로 녹음된 파일을 한번 변환해주는 작업이 필요하다.

때문에 ffmpeg 설치가 필수이며, 반드시 파이썬과 로컬에 각각 설치를 해줘야 한다.

 

pip install ffmpeg-python
brew install ffmpeg

 

stt

def is_speech(buffer, sample_rate=16000, frame_duration_ms=30, speech_threshold=0.3):
    audio_int16 = np.int16(buffer * 32767)
    audio_bytes = audio_int16.tobytes()
    frame_size = int(sample_rate * (frame_duration_ms / 1000.0))
    num_frames = len(audio_int16) // frame_size
    if num_frames == 0:
        return False
    speech_frames = 0
    for i in range(num_frames):
        start = i * frame_size * 2  # 2바이트 per 샘플
        frame = audio_bytes[start: start + frame_size * 2]
        if len(frame) < frame_size * 2:
            break
        if vad.is_speech(frame, sample_rate):
            speech_frames += 1
    fraction = speech_frames / num_frames
    return fraction >= speech_threshold

 

stt 함수에서는 우선 현재 들어오는 청크 정보가 제대로 된 음성 정보인지, 너무 짧은 단위의 청크가 들어온 것은 아닌지를 구별하기 위해 is_speech라는 함수를 정의했다.

버퍼에 포함된 음성정보가 발화로 인식하기 힘든 지나치게 짧은(0.3초 이하)지, 말하고있긴 한것인지를 구분한다.

 

def detect_language(audio_path):
    try:
        with open(audio_path, "rb") as audio_file:
            response = CLIENT.audio.transcriptions.create(
                model="whisper-1",
                file=audio_file,
                response_format="verbose_json"
            )
        detected_lang = response.language
        sanitized = sanitize_language_code(detected_lang)
        print(f"[DEBUG] 감지된 언어 (보정됨): {sanitized}")
        return sanitized
    except Exception as e:
        print(f"언어 감지 오류: {e}", file=sys.stderr)
        return DEFAULT_LANGUAGE

 

돌아가는 프로그램이 단순히 한국어 -> 영어로 번역하는 것이 아니고, 이후 화상회의에서 다국어 영역에 활용될 것을 고려하여 언어를 직접 감지하게 해보고 싶었다. 그래서 detect_language를 만들어 버퍼에 저장된 오디오 파일을 공유하며 언어를 자동으로 탐지하는 함수를 만들어줬다.

 

그래서 비록 UI는 수정이 귀찮은 관계로 원본 텍스트(한국어) 로 적혀있긴 하지만..

 

이렇게 일본어랑 프랑스어도 잘 된다👍

원래 강남이라고 발음하고싶었는데 발음이슈로 코난된거만 빼고

 

def stt_processing_thread(audio_queue, sentence_queue, transcription_queue, recording_active, target_language):
    global detected_language
    buffer = np.zeros((0, 1), dtype=np.float32)
    silence_threshold = 0.02
    silence_duration_threshold = 0.5
    silence_start = None
    language_detected_once = False

    while True:
        try:
            data = audio_queue.get(timeout=1)
            buffer = np.concatenate((buffer, data), axis=0)
            current_time = time.time()
            amplitude = np.mean(np.abs(data))
            if amplitude < silence_threshold:
                if silence_start is None:
                    silence_start = current_time
                elif current_time - silence_start >= silence_duration_threshold:
                    if len(buffer) > int(SAMPLE_RATE * 0.5):
                        if not is_speech(buffer):
                            buffer = np.zeros((0, 1), dtype=np.float32)
                            silence_start = None
                            audio_queue.task_done()
                            continue
                        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
                            sf.write(f.name, buffer, SAMPLE_RATE, format='WAV', subtype='PCM_16')  
                            if not language_detected_once:
                                lang_code = detect_language(f.name)
                                if lang_code:
                                    with language_lock:
                                        detected_language = lang_code
                                    language_detected_once = True

                            with language_lock:
                                current_lang = detected_language if detected_language is not None else DEFAULT_LANGUAGE
                            with open(f.name, "rb") as audio_file:
                                response = CLIENT.audio.transcriptions.create(
                                    model="whisper-1",
                                    file=audio_file,
                                    language=current_lang,
                                    prompt="We're now on meeting. Please transcribe exactly what you hear."
                                )
                            os.unlink(f.name)
                        text = response.text.strip()
                        if text:
                            print(f"[DEBUG] STT 결과: {text}")
                            source_log, _ = get_log_filenames(detected_language, target_language)
                            with open(source_log, "a", encoding="utf-8") as f:
                                f.write(text + "\n")
                            with language_lock:
                                src_lang = detected_language if detected_language is not None else DEFAULT_LANGUAGE
                            sentence_queue.put((text, src_lang))
                            transcription_queue.put((text, src_lang))
                    buffer = np.zeros((0, 1), dtype=np.float32)
                    silence_start = None
            else:
                silence_start = None
            audio_queue.task_done()
        except queue.Empty:
            continue

 

stt를 실제로 수행하는 함수에서는 이전과 크게 달라진 부분은 없고,

노이즈를 최대한 반영하지 않기 위해 진폭을 감지하여 silence_threshold(0.02)값을 추가하여 특정 진폭보다 낮은 값이 들어오면 침묵이라고 가정하도록 했고,

이러한 침묵시간이 silence_duration_threshold(0.5초)보다 길어지면 한 문장이 끝났다고 간주하여 data에 누적해놓은 버퍼들의 전사를 진행한다.

 

silence_threshold값은 너무 낮으면 침묵에 섞인 노이즈도 발화라고 간주되어 구독해주셔서 감사하다는 문구로 인식하고,

너무 높으면 내가 실제로 말한 것도 인식이 되지 않으니 적당한 값을 경험적으로 설정해주는 것이 좋을 것 같당.

 

전사가 완료된 문장은 sentence_queue와 transciption_queue로 전달되는데,

각각 translation_thread와 웹소켓으로 전송되는 역할을 한다.

큐에서 데이터 처리 시 get(pop)을 통해 값을 꺼내오기 때문에, 이중으로 꺼내오는 것을 방지하기 위해 두개의 큐를 정의하여 처리를 진행하게 했다.

 

translation

def translation_thread(sentence_queue, translation_queue, translated_queue, target_language):
    while True:
        try:
            sentence_data = sentence_queue.get(timeout=1)
            if isinstance(sentence_data, tuple):
                sentence, source_lang = sentence_data
            else:
                sentence = sentence_data
                source_lang = "ko"
            print(f"[DEBUG] 번역할 문장: {sentence} (소스 언어: {source_lang})")
            if source_lang == target_language:
                print("[DEBUG] 소스 언어와 타겟 언어 동일, 번역 건너뜀")
                translation_queue.put(sentence)
                sentence_queue.task_done()
                continue
            try:
                source_name = language_map.get(source_lang, "감지된 언어")
                target_name = language_map.get(target_language, "영어")
                response = CLIENT.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[
                        {"role": "system", "content": f"Translate the following text from {source_name} to {target_name}. Only provide the translation without any additional explanation."},
                        {"role": "user", "content": sentence}
                    ]
                )
                translation = response.choices[0].message.content.strip()
                print(f"[DEBUG] 번역 결과: {translation}")
            except Exception as e:
                print(f"번역 오류: {e}", file=sys.stderr)
                translation = ""
            if translation:
                _, target_log = get_log_filenames(source_lang, target_language)
                with open(target_log, "a", encoding="utf-8") as f:
                    f.write(translation + "\n")
                translation_queue.put(translation)
                translated_queue.put(translation)
            sentence_queue.task_done()
        except queue.Empty:
            continue

 

이전 프로젝트와 달라진 것은 여기서는 번역이 추가되었다는 점!

 

그러나 번역은 사실 쉽다..

입출력 데이터가 모두 텍스트이기 때문에 중간 결과들 로그 찍어서 보기도 너무 좋고

번역도 GPT-4o-mini 모델을 썼기 때문에 stt로 들어갔던 문장중 맥락에 부자연스러운 단어가 포함되어있어도 알아서 개선된 번역을 내놓기 때문에 결과물에 대한 걱정도 없다.

 

교수님죄송합니다..

수업하실때 몰래 테스트해본적이 있는데 여러개의 꽤 긴 문장의 발화도 잘되는 걸 볼 수 있다.

 

위의 전사-번역 결과 중 첫번째 문장에서

원문이

데이터 웨어 하우스가 갖고 있는 데이터 사전에 맞게 데이터를 변환시키고 그 다음에 이제 데이터 웨어 하우스에서 울려넣는거다.

로 인식되었으나,

 

번역에서는

The data is transformed to match the data dictionary of the data warehouse, and then it is loaded into the data warehouse.

 

이렇게 알아서 올려 넣다 / 적재의 의미인 load로 잘 수정된 것을 볼 수 있다.

 

sentence_data에서 꺼내온 데이터에서부터 번역을 진행하는데, 여기서부턴 '한 문장 단위'의 전사 결과가 전달되었을것이므로, 문장의 길이가 짧고 말고를 판단해 번역 진행을 결정하진 않는다. 그래서 예외처리 해줄 것도 적었다.

 

아래에 stt와 마찬가지로 번역된 텍스트 결과가 반환되었을 경우 tts 전송 단계를 고려하여 translation_queue와 translated_queue에 각각 번역 결과를 넣어준다.

 

---

실행 결과