[LLM] 화상회의 중 STT to TTS 수행하는 시스템 설계 - 3. OpenVidu 기반 화상회의 시스템에서 실시간 STS(STT-번역-TTS) 서비스 구현

드디어 마지막..

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

 

프로젝트 목적

코드 소개에 앞서, 해당 기능이 필요했던 프로젝트의 목적에 대해 먼저 설명하고자 한다. 현지 공장의 문제해결을 더욱 용이하게 만들고, 본사와 현지 공장 간 의사소통 차이를 완화시킬 수 있는 시스템을 설계하였다.

 

현지 공장에서는 만약 공장 내 이슈가 발생하면

 

1단계: 이상현상 모니터링 및 감지

현장에서는 자체적으로 문제 감지 시스템을 구축하여, 이상현상 발생 시 언제 발생한 어떠한 종류의 문제인지 현지 공장에서 확인할 수 있도록 알림을 전송한다.

 

2단계: 챗봇을 통한 문제 해결

발생한 문제상황과 촬영 이미지를 챗봇에 업로드하면 시스템이 문제상황을 분석하고, RAG를 통해 이전에 축적해둔 유사 해결사례, 트러블슈팅 목록, 각종 기술지침 가이드 등을 참고하여 문제 해결 방안을 제공한다.

 

3단계: 본사에 알림 전송

챗봇으로까지 문제 해결이 어렵다면, 지금까지의 문제 발생 상황 및 챗봇 상담 정보를 종합하여 자동으로 본사 및 협력업체에 메일을 전송한다.

 

4단계: 본사와 화상회의

실시간 통역기능을 제공하는 화상회의를 통해 본사와 문제 해결을 진행한다.

 

5단계: 회의록 자동생성

전사 기능을 통해 자동으로 기록되는 스크립트를 종합하여 회의록을 요약 생성한다. 이를 통해 이후에 발생할 수 있는 문제에 대해 대응할 수 있는 또다른 데이터를 추가한다.

 

 

이 5단계 파이프라인을 통해 현지 공장은 언어의 장벽 없이 보다 쉽게 문제 해결 가이드를 제공받고, 본사는 불가피한 파견 근무를 최소화하면서 현장 관리를 편하게 할 수 있는 환경을 만들어주는 것이 본 프로젝트의 목적이었다.

 

나는 이 중 4단계인 본사와 화상회의 진행의 핵심 기능을 담당하여 개발을 진행했다.

 

전체 진행 플로우

본 프로젝트의 전체 구조와 기술적 흐름은 다음과 같다.

 

불펌죄송합니다감사합니다

GitHub Action을 활용하여 코드 변동사항이 있을 때 자동으로 EC2 서버에 반영되도록 했고, 크게 7가지의 서버로 구성하였다.

 

본사와 현지공장의 주요 기능(화상회의 내 들어가는 모델들, 챗봇, 이상현상 탐지 등등)들은 전부 fastAPI에 구성했고,

vue는 프론트

MariaDB는 회의록 및 화상회의의 접속정보를 저장해두는 역할을 하고,

Spring 역시 화상회의 정보를 저장하는 역할을 하도록 구성했다.

 

본 기능 흐름

그 중 내가 설계한 흐름은 위와 같다..

구조가 처음에 프로젝트를 사전 설계하고 그에 맞춰 미니프로젝트를 진행했을 때와 비교해서 많이 바뀌었다.

 

1. 음성 전달 방식: 실시간 → 녹음한 부분만

OpenVidu 구조 상 무료 인프라에서 발생할 수 있는 통신 불안정, 녹음 중 TTS 음성의 개입과 같은 요인으로 인해 현실적으로 좋은 품질의 STT 결과를 가져오기 위해서는 녹음한 부분만 전달해서 진행하는 것이 옳다고 생각했다. 그래서 그냥 녹음버튼을 누르고 정지할 때 까지 녹음한 문장을 넘기는 방식으로 변경했다.

특히 프로젝트를 내주셨던 교수님께서 음성→음성 통신을 강조하셨다. 화상회의 현장에서 자동화 STS를 적용하려면 음성을 입력하는 부분이나 출력하는 부분 둘 중 하나는 반드시 인간의 개입이 필요한데, 왜냐하면 스피커 통신 시 출력 음성이 반드시 입력 음성에 섞일것이라 생각했다. 따라서 팀원들과 논의해본 결과 음성 '출력'이 자동화되어야 한다는 것이 좀 더 맞다는 방향으로 의견이 쏠렸기 때문에, 녹음에 인간의 개입이 들어가는 방향으로 수정했다.

 

2. 통신 방식: websocket → REST API

1번에 의해 전달방식도 웹소켓을 활용한 무중단 통신 방식 보단 버튼을 눌러 트리거가 발생해야지 음성을 전달하고, 그에 대한 응답으로 전사와 번역 결과를 가져오는 RESTful방식으로 바꿨다.

 

3. 비동기적 구성: 음성을 받아오는 부분에만

REST API로 통신 방식을 바꾸었기 때문에, 웹소켓 구현에서처럼 전사/번역 결과를 개별적으로 전달할 필요성이 없어졌다. 따라서 한 번의 요청으로 전사되는 텍스트와 번역 정보 및 TTS 음성정보를 안정적으로 묶어 반환할 수 있는 구조로 전환했다.

그러나 여전히 다양한 사용자들이 동시에 마이크를 누르고 음성입력을 할 수 있는 회의 상황을 고려해 음성 입력만큼은 스레드를 활용한 비동기 처리 방식을 유지했다.

 

파일 구조

headquater_system/
├── config.py           # 전역 설정 (API 키, 오디오 설정, 전역 클라이언트 등)
├── main.py             # 프로그램의 진입점: 각 모듈을 불러와 스레드 실행
└── modules/
|   ├── __init__.py
|   ├── stt.py          # STT 처리 (Whisper API, VAD, 언어 감지)
|   ├── translation.py  # 번역 처리 (GPT-4o-mini를 사용)
|   ├── tts.py          # TTS 처리 (GPT-4o-mini-tts를 사용)
|   ├── users.py        # 사용자 정보 객체, 업데이트 함수 정의
|   └── utils.py        # 공통 유틸리티 함수 (언어 보정, 로그 파일명 생성, 디스플레이 업데이트 등)
└── routers/
    ├── __init__.py
    └── hq.py        # 라우터 정보 설정

 

대부분의 구조와 함수는 이전과 동일하다. 다만, 여기서는 사용자 정보(이름), 사용 언어, 번역 언어 그리고 사용자 별 발화 정보를 개별적으로 관리하기 위해 users.py를 새로 정의해주었다.

 

---

코드

Router 별 액션 정리

프론트로부터 전달받는 객체는 STTPayload, 전달해주는 객체는 CombineResult의 list 형식으로 정의했다.

class STTPayload(BaseModel):
    type: str
    speakerInfo: dict
    audioData: str
    sampleRate: int
    timestamp: int = None

class CombinedResult(BaseModel):
    speaker: str
    transcription: str
    translation: str
    tts_voice: str

class CombinedResultsResponse(BaseModel):
    results: list[CombinedResult]

 

이렇게 정의된 객체 형태의 POST 요청이 들어오면 어떤 객체를 받고, 어떤 객체를 return 해줄지 까지의 모든 액션을 아래와 같이 작성해주었다. 함수가 길어서 조금씩 잘라보았다........

# 1. STT 관리 함수
@hq_router.post("/stt/audio", response_model=CombinedResultsResponse)
async def stt_audio_endpoint(payload: STTPayload):
    print("[DEBUG] POST 요청")
    global date_log

    # payload의 type 확인
    if payload.type != "live_audio_chunk":
        raise HTTPException(status_code=400, detail="Invalid payload type")

 

여기까지가 기본으로 데이터를 받아오는 부분이고, 만약 payload에 'type' 정보가 없거나 live_audio_chunk가 아니라면 더 이상 진행하지 못하게 404 exception을 발생시키게 하였다.

 

    # 사용자 정보 추출
    speaker_info = payload.speakerInfo
    speaker_name = speaker_info.get("name", "Unknown")
    source_lang = speaker_info.get("speakerLang", "ko")
    target_lang = speaker_info.get("targetLang", "en")
    session_id = speaker_info.get("sessionId", None)  # Extract sessionId
    
    print(f"{speaker_name}: src {source_lang}, tar {target_lang}, sessionId: {session_id}")
    # 타임스탬프 처리
    timestamp = payload.timestamp if payload.timestamp is not None else int(time.time() * 1000)
    if not date_log:
        dt = datetime.fromtimestamp(timestamp / 1000)
        date_log = dt.strftime("%Y%m%d_%H%M%S")
    
    # REST 방식이므로 websocket은 None 처리
    user = get_or_create_user(speaker_name, source_lang, target_lang, websocket=None)

    # 사용자별로 STT 처리 스레드 실행 (최초 연결 후 처음 데이터를 수신할 때 실행)
    if not user.processing_started:
        threading.Thread(
            target=stt_processing_thread,
            args=(user,),  # user 내부의 audio_queue 등 사용
            daemon=True
        ).start()

        user.processing_started = True
    
    # audioData 디코딩 및 PCM 데이터 변환 (Int16 -> float32, 정규화, 모노 재배열)
    try:
        raw_bytes = base64.b64decode(payload.audioData)
        print(f"raw_bytes: {len(raw_bytes)}")
        audio_np = np.frombuffer(raw_bytes, dtype=np.int16).astype(np.float32) / 32768.0
        audio_np = audio_np.reshape(-1, 1)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Audio data decoding error: {e}")
    
    # 사용자 객체의 음성 큐에 데이터를 넣음
    user.audio_queue.put((audio_np, payload.sampleRate))

 

이후 payload 내 데이터를 가져와 사용자를 등록하거나 메타정보(사용 언어/번역할 언어)를 업데이트 해주고, 사용자 별 STS 프로세스 스레드를 발생시킨 후, 사용자 별 audio_queue에 음성 정보를 넣어준다.

 

    # 백그라운드 STT 및 번역 처리 스레드가 실행 중이라고 가정하고,
    # 결과가 준비되어 있다면 transcription_queue와 translated_queue에서 꺼내 결합 메시지로 생성
    combined_results = []
    
    # 최대 timeout초동안 결과가 생성될 때까지 기다리는 예시 (polling)
    timeout = 120.0  # 최대 대기 시간
    poll_interval = 0.2  # 200ms 간격으로 폴링
    waited = 0.0

    while timeout > waited:
        try:
            # (stt 결과, 번역 결과, tts 음성(ogg -> base64로 인코딩))
            transcription, translation, tts_voice = user.final_results_queue.get_nowait()
 
            print(f"전사결과: {transcription}\n번역결과:{translation}")
            combined_results.append({
                "speaker": speaker_name,
                "transcription": transcription,
                "translation": translation,
                "tts_voice": tts_voice
            })
            user.final_results_queue.task_done()

            break  # 결과를 받았으므로 종료
        except queue.Empty:
            # 결과가 아직 없다면 잠시 대기
            await asyncio.sleep(poll_interval)
            waited += poll_interval

    return CombinedResultsResponse(results=combined_results)

 

마지막으로 무한루프가 발생되는 것을 방지하기 위해 timeout(120초)을 설정하고 그 시간 이내에 전사/번역이 완료되면 최종 final_results_queue에서 전체 값(stt 결과, 번역 결과, 인코딩된 tts 음성)을 가져와 프론트로 다시 전송해준다.

박본사 님의 화면

나카무라 켄죠 상의 화면

전체 실행 화면은 이렇게 된다👍

User

# modules/user.py

import threading
import time
import queue

# 사용자 정보를 담는 클래스 정의
class User:
    def __init__(self, name: str, source_lang: str = "ko", target_lang: str = "en", session_id: str = None):
        self.name = name
        self.source_lang = source_lang  # 사용자가 말하는 언어
        self.target_lang = target_lang  # 번역할 대상 언어
        self.last_update = time.time()  # 마지막 업데이트 시간 등 추가 정보 기록 가능
        self.session_id = session_id    # 세션 ID 추가
        
        self.detected_language = source_lang  # 개별 감지 언어
        self.processing_started = False  # 처리 스레드 실행 여부

        # 사용자 전용 큐들
        self.audio_queue = queue.Queue()       
        self.final_results_queue = queue.Queue()

        self.websocket = None

    def update(self, name: str = None, source_lang: str = None, target_lang: str = None, websocket=None, session_id: str = None):
        if name:
            self.name = name
        if source_lang:
            self.source_lang = source_lang
        if target_lang:
            self.target_lang = target_lang
        if session_id:
            self.session_id = session_id
        # 새 WebSocket 객체가 제공되면 업데이트
        if websocket is not None:
            self.websocket = websocket
        self.last_update = time.time()

# 전역 사용자 저장소 (동시 접근을 위해 lock 사용)
users_lock = threading.Lock()
users = {}

def get_or_create_user(name: str, default_source: str = "ko", default_target: str = "en", websocket=None, session_id: str = None) -> User:
    with users_lock:
        if name in users:
            user = users[name]
            # 사용자 정보 업데이트 (WebSocket도 함께 업데이트)
            user.update(name=name, source_lang=default_source, target_lang=default_target, websocket=websocket, session_id=session_id)
            return user
        else:
            user = User(name, default_source, default_target, session_id)
            user.websocket = websocket
            users[name] = user
            print(f"[DEBUG] 새로운 사용자가 추가되었습니다: {user.name}, 세션 ID: {session_id}, 참여인원: {len(users)}")
            return user

def get_user_by_connection(connection_id: str) -> User:
    with users_lock:
        return users.get(connection_id)

 

마지막까지 사용자 관련 함수를 계속 건드려서 코드 정리가 잘 안된 상태지만..😅

이전에 on_event("startup")과 기타 전역 변수로 선언해주었던 부분을 사용자 별로 구성하기 위해 class User에 전부 정의해둔 코드이다.

 

영어 / 일본어 / 중국어 / 튀르키예어 / 한국어

테스트를 위해 사용자 계정 5개를 미리 정의하였고, 사람마다 사용 언어를 다르게 정의해주었다.

 

우선 참여 계정은 두개라는 전제 하에 프론트 레벨에서 미팅 참여자들을 감지하고, 내가 아닌 상대 참여자의 사용 언어를 참조하여 번역할 언어를 payload에 같이 보내주기 때문에,

get_or_create_user 함수에서 이미 존재하는 사용자라면 번역 언어(target language)를 update할 수 있게 코드를 구성해주었다.

할수있는 중국어가 이거밖에 없다.

그래서 !!별도의 언어 선택 과정 없이!!(나름 자랑👍) 이렇게 참여하는 참가자 별로 상대를 인식하여 번역되는 언어가 달라지는 것을 확인할 수 있다.

 

사실 전사의 경우 whisper 자체가 어느정도 언어를 자동으로 감지하는 기능이 존재하기도 하고, 이전에 사용했던 언어 자동 감지 코드를 활용하면 사용자의 사용 언어(source language)는 굳이 프론트단에서 전달해주지 않아도 된다ㅎ

번역할 언어를 설정하는게 더 어려운 작업이기 때문에, 추후 3명 이상이 회의에 참여할 수 있다고 가정하고, 회의 참여자들의 언어권이 3개 이상일 경우를 고려한 수정 방안을 생각중이다.

 

STS pipeline

def stt_processing_thread(user):
    while True:
        try:
            # user.audio_queue에 (audio_np, sample_rate) 형태의 데이터를 넣었다고 가정
            data_tuple = user.audio_queue.get(timeout=1)
            try:
                if isinstance(data_tuple, tuple):
                    data, sample_rate = data_tuple
                else:
                    data = data_tuple
                    sample_rate = 16000 # 기본값은 16000

                # Optional: 음성 체크 (is_speech) – 파일 전체에 대해서 음성의 유무를 판단
                if len(data) < int(sample_rate * 0.5) or not is_speech(data):
                    print(f"[DEBUG] {user.name} - 음성 없음 또는 너무 짧은 발화")
                    continue
                    
                text = stt_processing(user, data, sample_rate)
                if not text:
                    print(f"[DEBUG] {user.name} - STT 결과 없음")
                    continue
            
                print(f"[DEBUG] STT 결과: {text}")

                # “Please transcribe exactly what you hear.” 은 에러 유도 메시지이므로 스킵
                if text.strip().lower().startswith("please transcribe exactly what you hear"):
                    print(f"[DEBUG] {user.name} - 에러 프롬프트 감지, 스킵")
                    continue

                try:
                    translation = translation_process(user, text)
                except Exception as te:
                    print(f"[DEBUG] {user.name} 번역 호출 중 오류: {te}", file=sys.stderr)
                    translation = ""
                
                try:
                    tts_voice = tts_process(translation)
                except Exception as te:
                    print(f"[DEBUG] {user.name} tts 호출 중 오류: {te}", file=sys.stderr)
                    tts_voice = ""

                user.final_results_queue.put((text, translation, tts_voice))
            finally:
                user.audio_queue.task_done()
            
        except queue.Empty:
                continue
        except Exception as e:
            print(f"STT 처리 중 오류 발생: {e}", file=sys.stderr)

 

 

코드 구조가 다소 단순해졌다.

기존에 thread처럼 구성되어 있던 각 프로세스를 일반 함수로 바꿔주고,

음성 데이터가 들어오면 각 단계별로 함수를 실행하고, 텍스트 값이 변환되면 하나로 묶어 final_results_queue에 데이터를 넣어주면 끝.

 

STT

def stt_processing(user, data, sample_rate):
    try:
        # 임시 파일에 저장하여 STT 처리 (Whisper API 호출)
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
            sf.write(f.name, data, sample_rate, format='WAV', subtype='PCM_16')

            # Whisper API 호출
            with open(f.name, "rb") as audio_file:
                response = CLIENT.audio.transcriptions.create(
                    model="whisper-1",
                    file=audio_file,
                    language=user.source_lang,
                    prompt="We're now on meeting. Please transcribe exactly what you hear."
                )
                    
        text = response.text.strip()

        return text
        
    except Exception as e:
        print(f"STT 처리 중 오류 발생: {e}", file=sys.stderr)
        return ""
    finally:
        # 예외 발생 여부와 상관없이 임시 파일이 있다면 삭제합니다.
        if f.name and os.path.exists(f.name):
            try:
                os.unlink(f.name)
            except Exception as del_e:
                print(f"임시 파일 삭제 오류: {del_e}", file=sys.stderr)

 

이전보다 훨씬 간단하게 사용자 정보, 입력 음성, 음성의 rate(!!매우 중요!!)만 전달하여 전사된 텍스트를 반환하는 코드로 바꾸어주었다.

 

기존에는 openvidu에서 음성을 받아오고자 하였으나, 프리티어 환경에서는 아무래도 8GB 이상의 최소 사양을 요구하는 OpenVidu의 특성상 상대방의 음성/영상 스트리밍이 제대로 진행이 되지 않는 문제가 존재했다.

따라서 좀 더 안정된 환경에서 보장된 품질의 음성 데이터를 사용하기 위해 로컬 마이크를 사용하여 녹음을 진행하도록 되어있는데,

그렇게 될 경우 로컬 기기마다 마이크의 정보가 다를 수 있기 때문에 오디오의 샘플레이트를 전달해주는 것이 매우 중요하다.. 이걸 사용해서 인코딩 디코딩을 진행하는데 이 값이 맞지 않으면 제대로 된 오디오가 전달되지 않는다..🥲

 

Translation

def translation_process(user, text):
    try:
        # 만약 메시지의 원본 언어와 자신의 대상 언어가 같다면 번역하지 않고 그대로 전달
        if user.source_lang == user.target_lang:
            print(f"[DEBUG] {user.name} 소스 언어와 타겟 언어가 동일하여 번역 없이 전송")
            return text
            
        # 번역 API 호출 (예시: GPT-4o-mini 번역 요청)
        try:
            source_name = language_map.get(user.source_lang, "감지된 언어")
            target_name = language_map.get(user.target_lang)
            response = CLIENT.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": f"""You are a professional interpreter. When translating from {source_name} to {target_name},
follow these rules:
1. 애매하거나 오해 소지가 있으면 → 명확한 용어로 고쳐 번역한다.
2. 잘못된 근거 정보일 경우 → 부드럽게 재확인하도록 번역한다.
3. 편협·공격적 발언일 경우 → 건설적인 논의 기회로 전환되도록 톤을 조절한다.
4. 반복·딜레이 중인 논의일 경우 → 핵심을 요약해 주제에 이끌어낸다.
5. 감정이 격해진 발언일 경우 → 중립적 완충 역할을 하며 번역한다.
Translate exactly what they say, without any extra commentary."""},
                    {"role": "user", "content": text}
                ]
            )
            translation = response.choices[0].message.content.strip()
            print(f"[DEBUG] {user.name} 번역 결과: {translation}")
        except Exception as e:
            print(f"[DEBUG] {user.name} 번역 오류: {e}", file=sys.stderr)
            translation = text
        return translation
        
    except Exception as e:
        print(f"[DEBUG] {user.name} 번역 처리 중 오류 발생: {e}", file=sys.stderr)
        return text

 

본 함수의 프롬프트가 조금 더 길어졌다.

 

이전에는 단순히 번역만 수행하라고 요청하였으나 해당 프로그램을 실제 업무에서 사용한다고 가정하고, 그 상황에서 비즈니스 통역가가 지켜야 하는 비즈니스 매너와 룰을 고려하여 프롬프트를 위와 같이 수정해주었다.

 

참고자료: https://www.linkedin.com/posts/xissy_%EC%BF%A0%ED%8C%A1%EC%9D%84-%EB%96%A0%EB%82%98-%EB%8B%A4%EB%A5%B8-%ED%95%9C%EA%B5%AD-%EC%8A%A4%ED%83%80%ED%8A%B8%EC%97%85%EB%93%A4%EC%97%90%EC%84%9C-%EC%9D%BC%ED%95%98%EB%A9%B4%EC%84%9C-%EB%8A%98-%ED%92%88%EA%B3%A0-%EC%9E%88%EB%8D%98-%EA%B6%81%EA%B8%88%EC%A6%9D%EC%9D%B4-%ED%95%98%EB%82%98-activity-7278644686020296704-Bg7l?utm_source=share&utm_medium=member_desktop&rcm=ACoAAC6WNvQB0_t0vhm3zjf2As1MOEsY_MwkSws

쿠팡을 떠나 다른 한국 스타트업들에서 일하면서 늘 품고 있던 궁금증이 하나 있었다. "회사가

 

TTS

def tts_process(translation):
    print(f"[TTS] 합성할 텍스트: {translation}")
    try:
        with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as temp_file:
            temp_audio_path = Path(temp_file.name)
            print(f"[DEBUG] TTS 임시 파일 경로: {temp_audio_path}")

       # TTS API 호출 (model="tts-1") – CLIENT.audio.speech.with_streaming_response.create 사용
        with CLIENT.audio.speech.with_streaming_response.create(
            model="tts-1",      # TTS 처리 모델: tts-1
            voice="nova",       # 선택 옵션 (원하는 목소리로 설정)
            input=translation,
            response_format="opus"
            # instructions="Optional additional instructions"  # 필요 시 추가 지침
        ) as response:
            response.stream_to_file(temp_audio_path)
        
        # 생성된 음성 파일을 binary 모드로 읽은 후 base64로 인코딩
        with open(str(temp_audio_path), "rb") as audio_file:
            audio_bytes = audio_file.read()
        base64_audio = base64.b64encode(audio_bytes).decode('utf-8')
        print(f"[DEBUG] TTS 음성이 base64로 인코딩됨 (길이: {len(base64_audio)} 문자)")
        
        # 임시 파일 삭제
        os.unlink(str(temp_audio_path))
        
        # tts_result_queue에 base64 인코딩 음성 데이터를 저장
        return base64_audio
    except Exception as e:
        print(f"TTS 오류: {e}", file=sys.stderr)
        return ""

 

 

전송 음성 타입을 지정할 수가 있다. 근데 여기서 중요한 것은 mp3는 base64 인코딩 시 5초 가량의 발화가 무려 70KB를 넘는다 .. 그래서 사실 데이터가 전송될 때 자꾸 로딩이 걸린 후 메시지 자체가 씹히는 문제가 발생했다 ㅜㅜ

찾아보니 네트워크 성능 문제인거같아.. 더 적은 용량의 음성 데이터를 사용하고자 웹소켓 스트리밍 및 통신용에 최적화된 저지연·고압축 코덱인 opus 포맷을 사용해준다.

 

이를 openAI API에서 설정해줄땐 response_format에 지정해주면 된다. 기본 설정이 mp3니까, 제외하고 지정해주면 된다.

옵션은 다음과 같다.:

• MP3: The default response format for general use cases.
• Opus: For internet streaming and communication, low latency.
• AAC: For digital audio compression, preferred by YouTube, Android, iOS.
• FLAC: For lossless audio compression, favored by audio enthusiasts for archiving.
• WAV: Uncompressed WAV audio, suitable for low-latency applications to avoid decoding overhead.
• PCM: Similar to WAV but contains the raw samples in 24kHz (16-bit signed, low-endian), without the header.

🔗 참고: https://platform.openai.com/docs/guides/text-to-speech

 

Opus(.ogg)를 사용하면 mp3 대비 토큰이 90% 가량 압축되는 것을 확인할 수 있었다.

그리고 실제로 화상회의에서 들을 때 크게 음질 차이도 느끼지 못한다. 야호🙌

 

그러나 만약에 확장성을 생각한다면 임시 파일을 업로드했다가 미팅 종료 후 삭제하는 방식을 고려해도 괜찮을 것 같다.

---

결과

 

하울링이 살짝 있씁니다..