요약 (TL;DR)
OpenAI는 다양한 사용 사례를 위한 두 가지 WebSocket API 모드를 제공합니다: Responses API WebSocket 모드는 도구 호출이 많은 에이전트 워크플로우(20개 이상의 도구 호출 시 최대 40% 더 빠름)를 위한 것이고, Realtime API는 저지연 음성 및 오디오 애플리케이션을 위한 것입니다. 두 모드 모두 무상태 HTTP 요청 대신 영구적인 WebSocket 연결을 사용하여 반복적인 연결 오버헤드를 제거하여 지연 시간을 줄이고 이벤트 기반의 상태 저장 상호 작용을 가능하게 합니다.
소개
OpenAI의 API는 단순한 요청-응답 패턴을 넘어 발전했습니다. 빠른 도구 호출이나 실시간 오디오 스트리밍이 필요한 애플리케이션의 경우, 전통적인 HTTP 모델은 불필요한 오버헤드를 발생시킵니다. 새로운 요청마다 연결 설정, 인증 및 상태 전송이 필요하며, 동일한 대화를 계속하는 경우에도 마찬가지입니다.
OpenAI의 WebSocket API는 영구적인 양방향 연결을 유지하여 이 문제를 해결합니다. 20개 이상의 순차적인 도구 호출이 있는 에이전트 워크플로우의 경우, 이는 종단 간 실행을 약 40% 더 빠르게 만듭니다. 음성 애플리케이션의 경우, 500ms 미만의 지연 시간으로 자연스럽고 중단 가능한 대화를 가능하게 합니다.
이 가이드에서는 OpenAI의 두 가지 WebSocket 모드를 모두 다룹니다: 도구 사용이 많은 에이전트 워크플로우를 위한 Responses API와 오디오 스트리밍을 위한 Realtime API입니다. 각 모드를 언제 사용하고, 어떻게 구현하며, 효과적으로 테스트하는 방법을 배울 것입니다.
OpenAI WebSocket API란 무엇인가요?
OpenAI WebSocket API는 OpenAI의 언어 모델과 상호 작용하기 위한 HTTP의 대체 전송 메커니즘을 제공합니다. 각 API 호출마다 새로운 연결을 생성하는 대신, WebSocket은 세션 동안 열려 있는 단일의 장기적인 연결을 설정합니다.
주요 특징
영구 연결: 일단 설정되면 WebSocket 연결은 명시적으로 닫힐 때까지 열려 있어 요청당 연결 오버헤드를 제거합니다.
양방향 통신: 클라이언트와 서버 모두 언제든지 메시지를 보낼 수 있어 진정한 이벤트 기반 아키텍처를 가능하게 합니다.
상태 저장 세션: 서버는 현재 연결에 대한 컨텍스트를 유지하여 전체 대화 기록을 다시 보내지 않고도 이전 응답을 참조할 수 있습니다.
이벤트 기반 모델: 통신은 요청-응답 쌍이 아닌 개별 이벤트(JSON 메시지)를 통해 이루어집니다.
WebSocket 프로토콜 기본 사항
WebSocket 연결은 HTTP 업그레이드 요청으로 시작한 다음 WebSocket 프로토콜로 전환됩니다. OpenAI의 경우 다음과 같은 엔드포인트에 연결합니다.
- Responses API:
wss://api.openai.com/v1/responses - Realtime API:
wss://api.openai.com/v1/realtime?model=gpt-realtime
wss:// 스키마는 보안 WebSocket 연결을 나타냅니다(HTTP의 HTTPS와 유사).
두 가지 WebSocket 모드 설명
OpenAI는 각각 다른 사용 사례에 최적화된 두 가지 고유한 WebSocket 모드를 제공합니다.
Responses API WebSocket 모드
Responses API는 순차적인 도구 호출을 많이 해야 하는 에이전트 워크플로우를 위한 WebSocket 연결을 지원합니다. 이 모드는 복잡한 작업을 수행하기 위해 도구를 반복적으로 호출하는 코딩 도우미, 오케스트레이션 시스템 및 자율 에이전트를 위해 설계되었습니다.
작동 방식:
활성 WebSocket 연결에서 서비스는 연결 로컬 메모리 캐시에 이전 응답 상태(가장 최근 응답) 하나를 유지합니다. 차례를 계속할 때 다음만 전송합니다.
previous_response_id(마지막 응답 참조)- 새로운 입력 항목(사용자 메시지, 도구 결과 등)
서버는 전체 대화 기록을 다시 처리하는 대신 캐시된 상태를 재사용합니다.
성능 이점:
20개 이상의 도구 호출이 있는 워크플로우의 경우, OpenAI는 HTTP에 비해 최대 40% 더 빠른 종단 간 실행을 보고합니다. 이는 다음에서 비롯됩니다.
- 요청당 연결 설정 없음
- 반복적인 인증 오버헤드 없음
- 캐시된 상태로 처리 시간 단축
- 작은 연속 메시지에 대한 낮은 네트워크 지연 시간
호환성:
WebSocket 모드는 Zero Data Retention (ZDR) 및 store=false 옵션 모두와 작동하므로 개인 정보 보호에 민감한 애플리케이션에 적합합니다.
Realtime API WebSocket 모드
Realtime API는 음성 기반 애플리케이션을 위한 저지연 스트리밍 오디오 기능을 제공합니다. 이 API는 모델이 오디오 입력에 오디오 출력으로 응답하고 중단을 자연스럽게 처리하는 음성 대 음성 상호 작용을 가능하게 합니다.
작동 방식:
Realtime API는 WebSocket을 사용하여 상태 저장 이벤트 기반 세션을 생성합니다. 오디오 청크를 API로 스트리밍하면 API는 전사본과 생성된 오디오 응답을 모두 스트리밍합니다. 연결은 다음을 지원합니다.
- 오디오 입력 스트리밍(캡처되는 대로 오디오 청크 전송)
- 오디오 출력 스트리밍(생성된 오디오 실시간 수신)
- 텍스트 입출력(하이브리드 텍스트+음성 상호 작용용)
- 자동 중단 처리(사용자가 말할 때 생성 중지)
주요 기능:
음성 활동 감지(VAD): API에는 사용자가 단순히 멈춘 것과 말을 마친 것을 이해하는 의미론적 VAD가 포함되어 있습니다. 이는 더 자연스러운 대화 흐름을 만듭니다.
멀티모달 기능: GPT-4o의 기본 멀티모달 기능에 직접 접근하여 통합 모델에서 오디오와 텍스트를 모두 처리합니다.
낮은 지연 시간: 음성 상호 작용을 위해 500ms 미만의 지연 시간으로 설계되어 실시간 대화에 적합합니다.
WebSocket vs HTTP: 성능 비교
WebSocket과 HTTP 중 무엇을 선택할지는 애플리케이션의 특성에 따라 다릅니다. 각 프로토콜이 뛰어난 경우는 다음과 같습니다.

WebSocket이 HTTP보다 뛰어난 경우
높은 도구 호출량:
워크플로우에서 10개 이상의 순차적인 도구 호출을 수행하는 경우, WebSocket의 영구 연결은 반복적인 설정 오버헤드를 제거합니다. 각 HTTP 요청에는 다음이 필요합니다.
- DNS 조회(캐시되지 않은 경우)
- TCP 핸드셰이크(3방향)
- TLS 핸드셰이크(TLS 1.3의 경우 2회 왕복)
- HTTP 요청/응답 헤더
WebSocket은 이를 한 번 수행한 다음 연결을 재사용합니다.
지연 시간에 민감한 애플리케이션:
모든 밀리초가 중요한 실시간 음성 또는 채팅 애플리케이션의 경우, WebSocket의 영구 연결 및 스트리밍 기능은 인지되는 지연 시간을 크게 줄여줍니다.
서버 시작 업데이트:
WebSocket은 서버가 폴링 없이 클라이언트로 데이터를 푸시할 수 있도록 합니다. 장기 실행 에이전트 작업의 경우, 서버는 이벤트가 발생하면 진행 상황 업데이트를 보낼 수 있습니다.
HTTP로 충분한 경우
간단한 요청-응답:
일회성 API 호출 또는 1-2개의 도구 호출이 있는 워크플로우의 경우 HTTP가 구현 및 디버깅이 더 간단합니다. 대부분의 개발자는 HTTP 클라이언라이언트에 익숙하며, 인프라(로드 밸런서, 프록시)는 HTTP를 잘 처리합니다.
무상태 작업:
요청 간에 세션 상태를 유지할 필요가 없는 경우, HTTP의 무상태 특성은 실제로 이점입니다. 연결 관리가 필요하지 않습니다.
인프라 제약:
일부 배포 환경(서버리스 함수, 특정 프록시)은 장기적인 WebSocket 연결을 지원하지 않습니다. HTTP는 보편적으로 작동합니다.
성능 지표
OpenAI의 문서 및 커뮤니티 테스트를 기반으로:
| 측정 항목 | HTTP | WebSocket (Responses API) | WebSocket (Realtime API) |
|---|---|---|---|
| 연결 설정 | 요청당 (~100-300ms) | 한 번 (~100-300ms) | 한 번 (~100-300ms) |
| 20개 이상의 도구 호출 워크플로우 | 기준 | ~40% 더 빠름 | 해당 없음 |
| 음성 왕복 지연 시간 | 해당 없음 (이 용도로 설계되지 않음) | 해당 없음 | <500ms |
| 메모리 오버헤드 | 낮음 (무상태) | 중간 (캐시된 상태) | 중간-높음 (세션 상태) |
| 구현 복잡도 | 낮음 | 중간 | 중간-높음 |
Responses API WebSocket 모드 사용 방법
Responses API에 대한 WebSocket 연결을 에이전트 워크플로우용으로 구현해 봅시다.
전제 조건
- Responses API 접근 권한이 있는 OpenAI API 키
- WebSocket 클라이언트 라이브러리(Node.js의 경우
ws, Python의 경우websocket-client) - OpenAI API의 도구 호출에 대한 이해
연결 설정
Node.js 예시:
const WebSocket = require('ws');
// Connect to Responses API WebSocket endpoint
const ws = new WebSocket('wss://api.openai.com/v1/responses', {
headers: {
'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
'OpenAI-Beta': 'responses-api=v1'
}
});
ws.on('open', () => {
console.log('Connected to OpenAI Responses API');
// Send initial request
const initialMessage = {
model: 'gpt-4o',
messages: [
{ role: 'user', content: 'Help me analyze this codebase and suggest improvements.' }
],
tools: [
{
type: 'function',
function: {
name: 'read_file',
description: 'Read contents of a file',
parameters: {
type: 'object',
properties: {
path: { type: 'string', description: 'File path to read' }
},
required: ['path']
}
}
},
{
type: 'function',
function: {
name: 'search_code',
description: 'Search for code patterns',
parameters: {
type: 'object',
properties: {
pattern: { type: 'string', description: 'Regex pattern to search' }
},
required: ['pattern']
}
}
}
]
};
ws.send(JSON.stringify(initialMessage));
});
ws.on('message', (data) => {
const response = JSON.parse(data);
console.log('Received:', response);
// Check if model wants to call tools
if (response.choices[0].finish_reason === 'tool_calls') {
const toolCalls = response.choices[0].message.tool_calls;
// Execute tools (simplified)
const toolResults = toolCalls.map(call => ({
tool_call_id: call.id,
output: executeToolLocally(call.function.name, call.function.arguments)
}));
// Continue the conversation with tool results
const continuation = {
previous_response_id: response.id, // Reference previous response
input: toolResults
};
ws.send(JSON.stringify(continuation));
}
});
ws.on('error', (error) => {
console.error('WebSocket error:', error);
});
ws.on('close', () => {
console.log('Connection closed');
});
function executeToolLocally(name, args) {
// Your tool execution logic
if (name === 'read_file') {
const { path } = JSON.parse(args);
return fs.readFileSync(path, 'utf-8');
}
// ... other tools
}
Python 예시:
import websocket
import json
import os
def on_message(ws, message):
response = json.loads(message)
print(f"Received: {response}")
# Handle tool calls
if response['choices'][0]['finish_reason'] == 'tool_calls':
tool_calls = response['choices'][0']['message']['tool_calls']
# Execute tools
tool_results = []
for call in tool_calls:
result = execute_tool(call['function']['name'],
json.loads(call['function']['arguments']))
tool_results.append({
'tool_call_id': call['id'],
'output': result
})
# Send continuation with only new input + previous_response_id
continuation = {
'previous_response_id': response['id'],
'input': tool_results
}
ws.send(json.dumps(continuation))
def on_error(ws, error):
print(f"Error: {error}")
def on_close(ws, close_status_code, close_msg):
print("Connection closed")
def on_open(ws):
print("Connected to OpenAI Responses API")
# Send initial request
initial_message = {
'model': 'gpt-4o',
'messages': [
{'role': 'user', 'content': 'Analyze this codebase and suggest improvements.'}
],
'tools': [
{
'type': 'function',
'function': {
'name': 'read_file',
'description': 'Read file contents',
'parameters': {
'type': 'object',
'properties': {
'path': {'type': 'string'}
},
'required': ['path']
}
}
}
]
}
ws.send(json.dumps(initial_message))
def execute_tool(name, args):
if name == 'read_file':
with open(args['path'], 'r') as f:
return f.read()
# ... other tools
# Create WebSocket connection
ws = websocket.WebSocketApp(
"wss://api.openai.com/v1/responses",
header={
"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
"OpenAI-Beta": "responses-api=v1"
},
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close
)
ws.run_forever()
주요 구현 세부 사항
상태 관리:
HTTP와의 결정적인 차이점은 연속에서 previous_response_id를 사용하는 것입니다. 이는 API에게 마지막 응답의 캐시된 상태를 재사용하도록 지시합니다.
입력 전용 연속:
차례를 계속할 때 다음만 전송합니다.
previous_response_id: 캐시된 응답 참조input: 새로운 데이터(도구 결과, 사용자 메시지 등)
전체 messages 배열을 다시 보내지 마세요. 서버는 이미 해당 컨텍스트를 가지고 있습니다.
제로 데이터 보존:
WebSocket 모드에서 ZDR을 사용하려면 초기 요청에 store: false를 포함하세요.
Realtime API WebSocket 모드 사용 방법
Realtime API는 저지연 음성 상호 작용을 가능하게 합니다. 구현 방법은 다음과 같습니다.
전제 조건
- Realtime API 접근 권한이 있는 OpenAI API 키
- 오디오 캡처/재생 기능
- WebSocket 클라이언트 라이브러리
- 오디오 인코딩(최적의 결과를 위해 24kHz, 16비트, 모노 PCM)
연결 설정
JavaScript (브라우저) 예시:
// Connect to Realtime API
const ws = new WebSocket(
`wss://api.openai.com/v1/realtime?model=gpt-realtime`,
['realtime', 'openai-insecure-api-key.' + process.env.OPENAI_API_KEY]
);
ws.addEventListener('open', () => {
console.log('Connected to Realtime API');
// Configure session
ws.send(JSON.stringify({
type: 'session.update',
session: {
modalities: ['text', 'audio'],
voice: 'alloy',
input_audio_format: 'pcm16',
output_audio_format: 'pcm16',
turn_detection: {
type: 'server_vad', // or 'semantic_vad' for smarter detection
threshold: 0.5,
prefix_padding_ms: 300,
silence_duration_ms: 500
}
}
}));
});
ws.addEventListener('message', (event) => {
const message = JSON.parse(event.data);
switch (message.type) {
case 'session.created':
console.log('Session created:', message.session);
break;
case 'conversation.item.created':
console.log('New item:', message.item);
break;
case 'response.audio.delta':
// Received audio chunk - play it
const audioChunk = base64ToArrayBuffer(message.delta);
playAudioChunk(audioChunk);
break;
case 'response.audio_transcript.delta':
// Received transcript chunk
console.log('Transcript:', message.delta);
break;
case 'input_audio_buffer.speech_started':
console.log('User started speaking');
break;
case 'input_audio_buffer.speech_stopped':
console.log('User stopped speaking');
break;
case 'error':
console.error('API error:', message.error);
break;
}
});
// Send audio from microphone
async function streamMicrophoneToAPI() {
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
const audioContext = new AudioContext({ sampleRate: 24000 });
const source = audioContext.createMediaStreamSource(stream);
const processor = audioContext.createScriptProcessor(4096, 1, 1);
processor.onaudioprocess = (e) => {
const inputData = e.inputBuffer.getChannelData(0);
// Convert Float32 to Int16 PCM
const pcmData = new Int16Array(inputData.length);
for (let i = 0; i < inputData.length; i++) {
pcmData[i] = Math.max(-32768, Math.min(32767, inputData[i] * 32768));
}
// Send to API
ws.send(JSON.stringify({
type: 'input_audio_buffer.append',
audio: arrayBufferToBase64(pcmData.buffer)
}));
};
source.connect(processor);
processor.connect(audioContext.destination);
}
// Send text input
function sendTextMessage(text) {
ws.send(JSON.stringify({
type: 'conversation.item.create',
item: {
type: 'message',
role: 'user',
content: [
{ type: 'input_text', text: text }
]
}
}));
// Request response generation
ws.send(JSON.stringify({
type: 'response.create'
}));
}
function playAudioChunk(arrayBuffer) {
const audioContext = new AudioContext({ sampleRate: 24000 });
audioContext.decodeAudioData(arrayBuffer, (buffer) => {
const source = audioContext.createBufferSource();
source.buffer = buffer;
source.connect(audioContext.destination);
source.start();
});
}
Python 예시:
import websocket
import json
import base64
import pyaudio
# Audio configuration
RATE = 24000
CHUNK = 4096
FORMAT = pyaudio.paInt16
CHANNELS = 1
audio = pyaudio.PyAudio()
def on_open(ws):
print("Connected to Realtime API")
# Configure session
ws.send(json.dumps({
'type': 'session.update',
'session': {
'modalities': ['text', 'audio'],
'voice': 'alloy',
'input_audio_format': 'pcm16',
'output_audio_format': 'pcm16',
'turn_detection': {
'type': 'server_vad',
'threshold': 0.5,
'silence_duration_ms': 500
}
}
}))
# Start streaming microphone
stream_microphone(ws)
def on_message(ws, message):
data = json.loads(message)
if data['type'] == 'response.audio.delta':
# Decode and play audio
audio_chunk = base64.b64decode(data['delta'])
play_audio(audio_chunk)
elif data['type'] == 'response.audio_transcript.delta':
print(f"Transcript: {data['delta']}", end='', flush=True)
elif data['type'] == 'input_audio_buffer.speech_started':
print("\n[User speaking...]")
elif data['type'] == 'error':
print(f"Error: {data['error']}")
def stream_microphone(ws):
stream = audio.open(
format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK
)
def audio_thread():
while True:
audio_data = stream.read(CHUNK)
ws.send(json.dumps({
'type': 'input_audio_buffer.append',
'audio': base64.b64encode(audio_data).decode('utf-8')
}))
import threading
threading.Thread(target=audio_thread, daemon=True).start()
def play_audio(audio_chunk):
stream = audio.open(
format=FORMAT,
channels=CHANNELS,
rate=RATE,
output=True
)
stream.write(audio_chunk)
stream.stop_stream()
stream.close()
# Create WebSocket connection
ws = websocket.WebSocketApp(
f"wss://api.openai.com/v1/realtime?model=gpt-realtime",
header={
"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"
},
on_open=on_open,
on_message=on_message
)
ws.run_forever()
주요 구현 세부 사항
이벤트 유형:
Realtime API는 이벤트 기반 통신을 사용합니다. 일반적인 이벤트:
클라이언트 → 서버:
session.update- 세션 매개변수 구성input_audio_buffer.append- 오디오 청크 전송conversation.item.create- 텍스트 메시지 추가response.create- AI 응답 요청
서버 → 클라이언트:
session.created- 세션 설정 확인response.audio.delta- AI의 오디오 청크response.audio_transcript.delta- AI 오디오의 전사본input_audio_buffer.speech_started/stopped- VAD 이벤트error- 오류 알림
음성 활동 감지:
두 가지 VAD 모드 중 선택하세요.
server_vad: 오디오 볼륨 및 무음 지속 시간에 기반한 기본 음성 활동 감지.
semantic_vad: 자연스러운 일시 정지와 차례 완료를 이해하는 더 스마트한 감지. 사용자가 생각을 멈출 수 있는 더 자연스러운 대화에 이것을 사용하세요.
Apidog로 WebSocket 연결 테스트하기
WebSocket API 테스트는 HTTP 테스트와 다릅니다. 연결을 유지하고, 이벤트를 보내고, 양방향 메시지 흐름을 모니터링해야 합니다. Apidog는 전문적인 WebSocket 테스트 기능을 제공합니다.

Apidog에서 WebSocket 테스트 설정하기
1단계: WebSocket 요청 생성
Apidog에서 새 요청을 생성하고 프로토콜로 "WebSocket"을 선택합니다. 연결 URL을 입력합니다.
wss://api.openai.com/v1/responses
2단계: 헤더 구성
인증 헤더를 추가합니다.
Authorization: Bearer YOUR_OPENAI_API_KEY
OpenAI-Beta: responses-api=v1
Realtime API의 경우 URL 기반 인증도 사용할 수 있습니다.
wss://api.openai.com/v1/realtime?model=gpt-realtime
서브프로토콜 헤더에 API 키를 포함합니다.
3단계: 연결 설정
"연결"을 클릭하여 WebSocket 연결을 설정합니다. Apidog는 다음을 보여줍니다.
- 연결 상태(연결됨/연결 끊김)
- 지연 시간 지표
- 연결 지속 시간
4단계: 이벤트 전송
Apidog의 메시지 컴포저를 사용하여 JSON 이벤트를 전송합니다. Responses API의 경우:
{
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": "What's the weather in San Francisco?"
}
],
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather",
"parameters": {
"type": "object",
"properties": {
"location": { "type": "string" }
}
}
}
}
]
}
5단계: 응답 모니터링
Apidog는 다음을 표시합니다.
- 수신된 모든 메시지를 시간 순서대로
- 메시지 타임스탬프 및 크기
- JSON 형식 지정 및 구문 강조
- 디버깅을 위한 복사/내보내기 기능
연속 테스트하기
previous_response_id를 사용하여 연속 패턴을 테스트하려면:
- 초기 메시지를 보내고, 응답에서
response.id를 기록합니다. - 새로운 입력만 포함하는 연속을 보냅니다:
{
"previous_response_id": "resp_abc123",
"input": [
{
"tool_call_id": "call_xyz789",
"output": "{\"temperature\": 72, \"conditions\": \"sunny\"}"
}
]
}
- 전체 컨텍스트를 다시 보내는 것에 비해 줄어든 지연 시간을 관찰합니다.
Realtime API 테스트하기
Realtime API의 경우 Apidog는 다음을 허용합니다.
- Base64로 인코딩된 오디오 청크 전송
- session.update 이벤트 모니터링
- VAD 이벤트 추적(음성 시작/중지)
- 실시간으로 전사본 델타 보기
이는 음성 비서가 사용자를 차단하거나 음성을 제대로 감지하지 못하는 이유를 디버깅하는 데 특히 유용합니다.
환경 변수
Apidog의 환경 변수를 사용하여 API 키를 안전하게 저장하세요.
{{OPENAI_API_KEY}}
이를 통해 요청을 편집하지 않고도 개발 및 프로덕션 키 간에 전환할 수 있습니다.
실제 사용 사례
OpenAI의 WebSocket 모드가 탁월한 실제 시나리오를 살펴보겠습니다.
사용 사례 1: 자율 코딩 에이전트
시나리오: 코드베이스를 분석하고, 문제를 식별하며, 자율적으로 개선하는 코딩 도우미.
Responses API WebSocket을 사용하는 이유:
- 일반적인 워크플로우: 파일 읽기 → 분석 → 유사 패턴 검색 → 추가 파일 읽기 → 변경 제안
- 이는 작업당 15-30개의 도구 호출을 생성합니다.
- WebSocket 모드는 총 실행 시간을 약 40% 단축합니다.
- 영구 연결은 모든 도구 호출에 걸쳐 컨텍스트를 유지합니다.
구현 패턴:
// Initial task
ws.send({ messages: [{ role: 'user', content: 'Audit security vulnerabilities' }], tools: [...] })
// First response: model calls read_file
ws.on('message', (resp1) => {
ws.send({ previous_response_id: resp1.id, input: [tool_result_1] })
})
// Second response: model calls search_code
ws.on('message', (resp2) => {
ws.send({ previous_response_id: resp2.id, input: [tool_result_2] })
})
// Continue for 20+ iterations...
사용 사례 2: 음성 고객 서비스 봇
시나리오: 자연스러운 대화 흐름으로 고객 문의를 처리하는 전화 지원 봇.
Realtime API WebSocket을 사용하는 이유:
- 낮은 지연 시간 중요(자연스러운 대화에 <500ms)
- 중단 처리 필요(고객이 봇을 가로채 말하는 경우)
- 별도의 전사 과정 없이 음성 입력을 직접 처리
- 응답을 실시간으로 스트리밍(완전한 문장을 기다리지 않음)
구현 패턴:
// Stream phone audio to API
phoneSystem.on('audio', (chunk) => {
ws.send({
type: 'input_audio_buffer.append',
audio: base64Encode(chunk)
})
})
// Play AI responses immediately
ws.on('message', (event) => {
if (event.type === 'response.audio.delta') {
phoneSystem.playAudio(base64Decode(event.delta))
}
})
일반적인 문제 해결
연결 설정 실패
증상: WebSocket 연결이 열리지 않고 즉시 닫힘 이벤트가 발생합니다.
일반적인 원인:
- 잘못된 API 키 -
Authorization헤더를 다시 확인하세요. - 베타 헤더 누락 - Responses API에는
OpenAI-Beta: responses-api=v1이 필요합니다. - 네트워크 제한 - 일부 기업 네트워크는 WebSocket을 차단합니다.
- 잘못된 URL -
wss://(ws://아님) 및 엔드포인트 경로를 확인하세요.
해결책:
자세한 오류 메시지와 함께 Apidog를 사용하여 연결을 테스트하세요. 요청 검사기는 어떤 헤더가 전송되었는지 정확히 보여주므로 누락되거나 잘못된 API 키를 쉽게 찾을 수 있습니다.
디버깅 코드:
ws.on('error', (error) => {
console.error('Connection error:', error);
});
ws.on('close', (code, reason) => {
console.log(`Closed with code ${code}: ${reason}`);
// Common codes:
// 1006: Abnormal closure (often auth issues)
// 1008: Policy violation (invalid headers)
});
WebSocket임에도 불구하고 높은 지연 시간
증상: WebSocket 연결은 작동하지만 HTTP보다 빠르지 않습니다.
일반적인 원인:
previous_response_id를 사용하지 않음 - 전체 컨텍스트를 다시 보내고 있습니다.- 콜드 스타트 - 새 연결의 첫 요청이 더 느립니다.
- 네트워크 지연 시간 - API 서버까지의 지리적 거리
- 큰 페이로드 - 연속에서 불필요한 데이터를 보냅니다.
해결책:
연속에서 새로운 입력만 보내고 있는지 확인하세요.
// WRONG - sends full context every time
ws.send({
messages: [...allPreviousMessages, newMessage],
tools: [...]
})
// RIGHT - references cached state
ws.send({
previous_response_id: lastResponse.id,
input: [newMessage]
})
장기 실행 연결의 메모리 누수
증상: 영구 연결에서 시간이 지남에 따라 애플리케이션 메모리가 증가합니다.
일반적인 원인:
- 이벤트 리스너가 제거되지 않음 - 재연결 시 리스너가 누적됨
- 오디오 버퍼가 해제되지 않음 - 재생된 오디오에 대한 참조 유지
- 메시지 기록 증가 - 수신된 모든 메시지 저장
해결책:
// Clean up event listeners on reconnection
function cleanupAndReconnect(ws) {
ws.removeAllListeners();
ws.close();
const newWs = createConnection();
return newWs;
}
// Release audio buffers after playing
function playAndRelease(audioBuffer) {
const source = audioContext.createBufferSource();
source.buffer = audioBuffer;
source.connect(audioContext.destination);
source.start();
source.onended = () => {
source.disconnect();
// Buffer will be garbage collected
};
}
// Limit message history
const messageHistory = [];
const MAX_HISTORY = 100;
ws.on('message', (data) => {
messageHistory.push(data);
if (messageHistory.length > MAX_HISTORY) {
messageHistory.shift(); // Remove oldest
}
});
결론
OpenAI의 WebSocket API 모드는 AI 애플리케이션을 위한 새로운 가능성을 열어줍니다. Responses API WebSocket 모드는 도구 호출이 많은 에이전트 워크플로우에 대해 최대 40% 더 빠른 실행을 제공하여 자율 코딩 도우미 및 오케스트레이션 시스템에 이상적입니다. Realtime API는 음성 애플리케이션에 500ms 미만의 지연 시간을 제공하여 자연스럽고 중단 가능한 대화를 가능하게 합니다.
올바른 모드를 선택하는 것은 사용 사례에 따라 다릅니다.
- Responses API WebSocket: 도구 사용이 많은 에이전트, 코딩 도우미, 연구 도구(10개 이상의 도구 호출)
- Realtime API WebSocket: 음성 비서, 전화 봇, 언어 튜터(오디오 스트리밍)
- HTTP: 간단한 요청, 서버리스 환경, 1-5개의 API 호출
WebSocket 연결의 영구적이고 이벤트 기반 특성은 HTTP와 다른 테스트 접근 방식을 요구합니다. Apidog의 실시간 WebSocket 클라이언트를 사용하여 OpenAI의 WebSocket API를 테스트하세요. API 키를 가져오고, 연결을 설정하고, 이벤트를 보내고, 상세한 로깅으로 응답을 모니터링하세요. 프로덕션 배포 전에 통합을 검증하기 위해 무료로 사용해보세요.
