OpenAI Realtime API로 음성 대화 AI 만들기

AI API 활용

OpenAI Realtime API로 음성 대화 AI 만들기 - WebSocket 실전 구현

소개왕 탑백귀 2026. 4. 20. 15:45

OpenAI Realtime API로 음성 대화 AI 만들기 - WebSocket 실전 구현

2026년 4월 기준 | AI API 활용

요약: OpenAI Realtime API를 WebSocket으로 연결해 "말하면 바로 응답하는" 실시간 음성 대화 AI를 만듭니다. 마이크 입력, 실시간 스트리밍, 스피커 출력, 끼어들기(interruption)까지 구현한 파이썬 코드와 주의사항을 정리합니다.

Realtime API가 뭐가 다른가

기존 음성 AI는 "음성 → 텍스트(STT) → LLM → 텍스트 → 음성(TTS)" 3단계 파이프라인이었습니다. 각 단계가 순차적이라 응답까지 2~3초씩 걸리는 경우가 많았습니다.

OpenAI Realtime API는 이 세 단계를 하나의 모델 안에서 처리합니다. 오디오 청크를 받자마자 해석하고, 답변 오디오를 스트리밍으로 돌려줍니다. 사람이 느끼는 지연 시간이 200~400ms 수준이라 실제 대화에 가까운 느낌을 줍니다.

동작 원리

WebSocket 연결을 하나 열고, 양방향으로 이벤트를 주고받는 구조입니다.

클라이언트 → 서버: input_audio_buffer.append 이벤트로 PCM 오디오 청크 전송
서버 → 클라이언트: response.audio.delta 이벤트로 응답 오디오 청크 수신
Voice Activity Detection(VAD): 서버가 자동으로 "사용자 발화 종료"를 감지

즉, 사용자 목소리를 계속 밀어 넣고, 서버가 답변을 쏟아내면 그대로 스피커로 재생하면 끝입니다.

환경 준비

pip install openai websockets sounddevice numpy python-dotenv

.env:

OPENAI_API_KEY=sk-xxxxx

기본 WebSocket 연결

import asyncio, os, json, base64
import websockets
from dotenv import load_dotenv

load_dotenv()
URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime-2026"
HEADERS = {
    "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
    "OpenAI-Beta": "realtime=v1",
}

async def main():
    async with websockets.connect(URL, extra_headers=HEADERS) as ws:
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "instructions": "한국어로 친근하게 대화하는 AI 비서야.",
                "voice": "alloy",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {"type": "server_vad"},
            },
        }))

        async for msg in ws:
            event = json.loads(msg)
            print(event["type"])

asyncio.run(main())

turn_detection: server_vad로 두면 사용자가 말을 멈춘 순간을 서버가 알아서 판단합니다. 대화형 UX엔 이 설정이 가장 편합니다.

마이크 입력 스트리밍

sounddevice로 마이크에서 PCM 16비트 24kHz 오디오를 뽑아, base64로 인코딩해 서버로 스트리밍합니다.

import sounddevice as sd
import numpy as np

SAMPLE_RATE = 24000
CHUNK = 1024

async def send_microphone(ws):
    loop = asyncio.get_event_loop()
    q = asyncio.Queue()

    def callback(indata, frames, time, status):
        loop.call_soon_threadsafe(q.put_nowait, indata.copy())

    with sd.InputStream(samplerate=SAMPLE_RATE, channels=1,
                          dtype="int16", blocksize=CHUNK, callback=callback):
        while True:
            chunk = await q.get()
            b64 = base64.b64encode(chunk.tobytes()).decode()
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": b64,
            }))

블록 크기는 1024 샘플(약 42ms) 정도가 반응성과 네트워크 효율의 균형점입니다.

응답 오디오 재생

서버가 보내는 response.audio.delta 이벤트마다 PCM 청크를 스피커에 밀어 넣습니다.

play_queue = asyncio.Queue()

def output_callback(outdata, frames, time, status):
    try:
        chunk = play_queue.get_nowait()
        data = np.frombuffer(chunk, dtype=np.int16)
        outdata[:len(data)] = data.reshape(-1, 1)
    except asyncio.QueueEmpty:
        outdata[:] = 0

async def receive_audio(ws):
    async for msg in ws:
        event = json.loads(msg)
        if event["type"] == "response.audio.delta":
            audio = base64.b64decode(event["delta"])
            await play_queue.put(audio)

OutputStream을 함께 열어두면 큐에 들어오는 대로 자연스럽게 재생됩니다.

끼어들기 처리

AI가 말하는 도중에 사용자가 말하면 즉시 AI를 멈추고 사용자 발화를 받아야 자연스럽습니다. Realtime API는 서버 VAD가 사용자 발화를 감지하면 input_audio_buffer.speech_started 이벤트를 보냅니다.

if event["type"] == "input_audio_buffer.speech_started":
    # 재생 큐 비우기
    while not play_queue.empty():
        play_queue.get_nowait()
    # 서버에도 응답 취소 요청
    await ws.send(json.dumps({"type": "response.cancel"}))

이 한 줄이 진짜 "대화다운" 느낌을 만듭니다. 없으면 AI가 혼자 2분짜리 설명을 끝까지 쏟아내는, 보통 AI의 답답함이 그대로 나타납니다.

음성에서 툴 호출

Realtime API도 Function Calling을 지원합니다. 세션 초기 설정에 툴 정의를 넘기면, 모델이 말하는 도중 필요할 때 툴 호출 이벤트를 보냅니다.

"tools": [{
    "type": "function",
    "name": "get_weather",
    "description": "특정 도시의 현재 날씨 조회",
    "parameters": {
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
}]

예를 들어 "서울 날씨 어때?"라고 물으면 서버가 response.function_call_arguments.done 이벤트로 {"city": "서울"}을 보냅니다. 로컬에서 날씨 API를 호출하고 결과를 conversation.item.create로 다시 보내면 AI가 자연스럽게 음성으로 답합니다.

비용과 주의사항

가격: 오디오 입력 $100/1M 토큰, 오디오 출력 $200/1M 토큰 수준(2026년 4월 기준). 분당 약 300~500원 소요.
세션 시간: 최대 30분. 길게 쓰려면 재연결 로직 필요.
네트워크: 레이턴시가 중요. 클라이언트 → OpenAI 구간 RTT가 100ms 이하일수록 자연스러움.
프라이버시: 실시간 음성이 OpenAI 서버로 전송됨. 사내 민감 정보는 별도 조치 필요.

마무리

Realtime API는 기존 STT+LLM+TTS 파이프라인을 한 번에 대체합니다. 300줄 정도의 코드로도 실제 사람과 말하는 듯한 음성 AI를 만들 수 있고, 콜센터, 음성 튜터, 차량용 비서 같은 분야에 직접 적용할 수 있습니다.

다음 글에서는 Realtime API와 Twilio를 연결해 실제 전화로 걸면 받는 AI 상담원을 만드는 방법, 그리고 긴 세션을 대화 요약 + 재시작으로 이어가는 패턴을 다뤄보겠습니다.

저작자표시 비영리 변경금지 (새창열림)