5장: 음성 AI — STT, TTS, 실시간 음성 대화

import whisper
 
model = whisper.load_model("large-v3")
 
result = model.transcribe(
    "meeting_recording.mp3",
    language="ko",
    task="transcribe",
)
 
print(result["text"])
 
# 타임스탬프 포함 세그먼트
for segment in result["segments"]:
    start = segment["start"]
    end = segment["end"]
    text = segment["text"]
    print(f"[{start:.1f}s - {end:.1f}s] {text}")

from openai import OpenAI
 
client = OpenAI()
 
with open("audio.mp3", "rb") as audio_file:
    transcription = client.audio.transcriptions.create(
        model="whisper-1",
        file=audio_file,
        language="ko",
        response_format="verbose_json",
        timestamp_granularities=["word", "segment"],
    )
 
for segment in transcription.segments:
    print(f"[{segment.start:.1f}s] {segment.text}")

from pydub import AudioSegment
 
def split_audio(file_path: str, chunk_duration_ms: int = 30000):
    """긴 오디오를 적정 크기로 분할"""
    audio = AudioSegment.from_file(file_path)
    chunks = []
 
    for i in range(0, len(audio), chunk_duration_ms):
        chunk = audio[i:i + chunk_duration_ms]
        chunks.append(chunk)
 
    return chunks
 
def transcribe_long_audio(file_path: str) -> str:
    """긴 오디오를 청크별로 처리"""
    chunks = split_audio(file_path)
    full_text = []
 
    for i, chunk in enumerate(chunks):
        # 청크를 임시 파일로 저장
        chunk_path = f"/tmp/chunk_{i}.mp3"
        chunk.export(chunk_path, format="mp3")
 
        # 이전 텍스트를 프롬프트로 제공하여 연속성 유지
        prev_text = full_text[-1] if full_text else ""
        result = transcribe_with_context(chunk_path, prev_text)
        full_text.append(result)
 
    return " ".join(full_text)

from openai import OpenAI
from pathlib import Path
 
client = OpenAI()
 
speech_response = client.audio.speech.create(
    model="tts-1-hd",
    voice="alloy",  # alloy, echo, fable, onyx, nova, shimmer
    input="안녕하세요. 이것은 텍스트를 음성으로 변환하는 예시입니다.",
    speed=1.0,
)
 
speech_response.stream_to_file(Path("output.mp3"))

from openai import OpenAI
 
client = OpenAI()
 
# LLM 응답을 스트리밍으로 받으면서 TTS에 전달
def stream_speech(text_chunks):
    """텍스트 청크를 스트리밍으로 음성 변환"""
    buffer = ""
 
    for chunk in text_chunks:
        buffer += chunk
 
        # 문장 단위로 TTS 변환
        while "." in buffer or "?" in buffer or "!" in buffer:
            # 문장 경계 찾기
            for sep in [".", "?", "!"]:
                idx = buffer.find(sep)
                if idx != -1:
                    sentence = buffer[:idx + 1]
                    buffer = buffer[idx + 1:].lstrip()
 
                    # 문장 단위 TTS
                    response = client.audio.speech.create(
                        model="tts-1",
                        voice="nova",
                        input=sentence,
                    )
                    yield response.content
                    break

import torch
 
model, utils = torch.hub.load(
    repo_or_dir="snakers4/silero-vad",
    model="silero_vad",
)
(get_speech_timestamps, _, read_audio, *_) = utils
 
audio = read_audio("recording.wav")
speech_timestamps = get_speech_timestamps(audio, model)
 
for ts in speech_timestamps:
    print(f"Speech: {ts['start']/16000:.1f}s - {ts['end']/16000:.1f}s")

import asyncio
import websockets
import json
 
async def realtime_voice_chat():
    """GPT-4o 실시간 음성 대화"""
    url = "wss://api.openai.com/v1/realtime"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "OpenAI-Beta": "realtime=v1",
    }
 
    async with websockets.connect(url, extra_headers=headers) as ws:
        # 세션 설정
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": "한국어로 친근하게 대화해주세요.",
                "voice": "nova",
                "input_audio_transcription": {"model": "whisper-1"},
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                },
            },
        }))
 
        # 오디오 스트리밍 시작
        asyncio.create_task(send_audio_stream(ws))
 
        # 응답 수신
        async for message in ws:
            event = json.loads(message)
            if event["type"] == "response.audio.delta":
                # 오디오 응답 청크 재생
                play_audio_chunk(event["delta"])
            elif event["type"] == "response.text.delta":
                # 텍스트 응답 (자막용)
                print(event["delta"], end="", flush=True)

import noisereduce as nr
import numpy as np
import soundfile as sf
 
# 오디오 로드
audio, sr = sf.read("noisy_audio.wav")
 
# 노이즈 프로파일 추출 (첫 1초를 노이즈로 가정)
noise_sample = audio[:sr]
 
# 노이즈 제거
cleaned_audio = nr.reduce_noise(
    y=audio,
    sr=sr,
    y_noise=noise_sample,
    prop_decrease=0.8,
)
 
sf.write("cleaned_audio.wav", cleaned_audio, sr)

# pyannote.audio를 활용한 화자 분리
from pyannote.audio import Pipeline
 
pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization-3.1")
diarization = pipeline("meeting.wav")
 
for turn, _, speaker in diarization.itertracks(yield_label=True):
    print(f"[{turn.start:.1f}s - {turn.end:.1f}s] {speaker}")

def generate_meeting_minutes(audio_path: str, llm_client) -> str:
    """오디오에서 회의록 자동 생성"""
    # 1. 화자 분리
    diarization = diarize(audio_path)
 
    # 2. STT (화자별)
    transcript = transcribe_with_speakers(audio_path, diarization)
 
    # 3. LLM으로 회의록 작성
    response = llm_client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"""다음 회의 녹취록을 기반으로 공식 회의록을 작성해주세요.
 
녹취록:
{transcript}
 
회의록에 포함할 내용:
1. 참석자
2. 안건 요약
3. 주요 논의 사항
4. 결정 사항
5. 액션 아이템 (담당자, 기한)""",
        }],
    )
    return response.content[0].text