OpenAI Realtime API

​

실시간 트레이스 통합하기

import weave
from weave.integrations import patch_openai_realtime

weave.init("your-team-name/your-project-name")
patch_openai_realtime()

# 애플리케이션 로직

import * as weave from 'weave';

await weave.init('your-team-name/your-project-name');

// 애플리케이션 로직

import { patchRealtimeSession } from 'weave';

await patchRealtimeSession();

​

OpenAI Agents SDK로 실시간 에이전트 실행하기

​

WebSocket으로 실시간 음성 어시스턴트 실행하기

1. Python 환경을 실행하고 다음 라이브러리를 설치하세요:
   uv

   pip

   uv add weave websockets pyaudio numpy
   
2. tool_definitions.py라는 이름의 파일을 만들고 다음 도구 정의를 추가하세요. 메인 애플리케이션은 이 모듈에서 필요한 항목을 임포트합니다.
   tool_definitions.py

   import json
   import subprocess
   import tempfile
   from pathlib import Path
   import weave
   
   
   # @function_tool
   @weave.op
   def get_weather(city: str) -> str:
       """Get the current weather for a city.
   
       Args:
           city: The city name to get weather for.
       """
       return json.dumps({"city": city, "temperature": "72°F", "condition": "sunny"})
   
   
   @weave.op
   def calculate(expression: str) -> str:
       """Evaluate a math expression and return the result.
   
       Args:
           expression: A math expression to evaluate, e.g. '2 + 2'.
       """
       try:
           result = eval(expression)
           return str(result)
       except Exception as e:
           return f"Error: {e}"
   
   
   @weave.op
   def run_python_code(code: str) -> str:
       """Write and execute a Python script, returning its stdout/stderr.
   
       Args:
           code: The Python source code to execute.
       """
       with tempfile.NamedTemporaryFile(
           mode="w", suffix=".py", dir=tempfile.gettempdir(), delete=False
       ) as f:
           f.write(code)
           script_path = Path(f.name)
   
       try:
           result = subprocess.run(
               ["python", str(script_path)],
               capture_output=True,
               text=True,
               timeout=30,
           )
           output = result.stdout
           if result.stderr:
               output += f"\nSTDERR:\n{result.stderr}"
           if result.returncode != 0:
               output += f"\n(exit code {result.returncode})"
           return output or "(no output)"
       except subprocess.TimeoutExpired:
           return "Error: script timed out after 30 seconds."
       finally:
           script_path.unlink(missing_ok=True)
   
   
   @weave.op
   async def write_file(file_path: str, content: str) -> str:
       """Write content to a file on disk.
   
       Args:
           file_path: The path to write the file to.
           content: The content to write into the file.
       """
       try:
           path = Path(file_path)
           path.parent.mkdir(parents=True, exist_ok=True)
           path.write_text(content)
           return f"Wrote {len(content)} bytes to {file_path}"
       except Exception as e:
           return f"Error writing file: {e}"
   
3. 같은 디렉터리에 weave_ws_voice_assistant.py 파일을 만들고 다음 코드를 추가하세요.
   weave_ws_voice_assistant.py

   import asyncio
   import base64
   import json
   import os
   import queue
   import threading
   from typing import Any, Callable
   
   import numpy as np
   import pyaudio
   import websockets
   
   import weave
   weave.init("<your-team-name/your-project-name>")
   from weave.integrations import patch_openai_realtime
   patch_openai_realtime()
   
   from tool_definitions import (
       calculate,
       get_weather,
       run_python_code,
       write_file,
   )
   
   # 오디오 형식(Realtime API에서는 반드시 PCM16이어야 함).
   FORMAT = pyaudio.paInt16
   RATE = 24000
   CHUNK = 1024
   MAX_INPUT_CHANNELS = 2
   MAX_OUTPUT_CHANNELS = 2
   
   OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
   REALTIME_URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime"
   
   DEBUG_WRITE_LOG = False
   
   # 함수 호출을 디스패치할 수 있도록 도구 이름을 callable에 매핑합니다.
   TOOL_REGISTRY: dict[str, Callable[..., Any]] = {
       "get_weather": get_weather,
       "calculate": calculate,
       "run_python_code": run_python_code,
       "write_file": write_file,
   }
   
   # Realtime API 세션 설정에 사용할 원시 도구 정의입니다.
   TOOL_DEFINITIONS = [
       {
           "type": "function",
           "name": "get_weather",
           "description": "Get the current weather for a city.",
           "parameters": {
               "type": "object",
               "properties": {
                   "city": {
                       "type": "string",
                       "description": "The city name to get weather for.",
                   }
               },
               "required": ["city"],
           },
       },
       {
           "type": "function",
           "name": "calculate",
           "description": "Evaluate a math expression and return the result.",
           "parameters": {
               "type": "object",
               "properties": {
                   "expression": {
                       "type": "string",
                       "description": "A math expression to evaluate, e.g. '2 + 2'.",
                   }
               },
               "required": ["expression"],
           },
       },
       {
           "type": "function",
           "name": "run_python_code",
           "description": "Write and execute a Python script, returning its stdout/stderr.",
           "parameters": {
               "type": "object",
               "properties": {
                   "code": {
                       "type": "string",
                       "description": "The Python source code to execute.",
                   }
               },
               "required": ["code"],
           },
       },
       {
           "type": "function",
           "name": "write_file",
           "description": "Write content to a file on disk.",
           "parameters": {
               "type": "object",
               "properties": {
                   "file_path": {
                       "type": "string",
                       "description": "The path to write the file to.",
                   },
                   "content": {
                       "type": "string",
                       "description": "The content to write into the file.",
                   },
               },
               "required": ["file_path", "content"],
           },
       },
   ]
   
   
   async def send_event(ws, event: dict) -> None:
       await ws.send(json.dumps(event))
   
   
   async def configure_session(ws) -> None:
       event = {
           "type": "session.update",
           "session": {
               "type": "realtime",
               "model": "gpt-realtime",
               "output_modalities": ["audio"],
               "instructions": (
                   "You are a helpful AI assistant with access to tools. "
                   "Use tools to accomplish tasks whenever possible. "
                   "Speak clearly and briefly."
               ),
               "tools": TOOL_DEFINITIONS,
               "tool_choice": "auto",
               "audio": {
                   "input": {
                       "format": {"type": "audio/pcm", "rate": 24000},
                       "transcription": {"model": "gpt-4o-transcribe"},
                       "turn_detection": {
                           "type": "server_vad",
                           "threshold": 0.5,
                           "prefix_padding_ms": 300,
                           "silence_duration_ms": 500,
                       },
                   },
                   "output": {
                       "format": {"type": "audio/pcm", "rate": 24000},
                   },
               },
           },
       }
       await send_event(ws, event)
       print("Session configured.")
   
   
   async def handle_function_call(ws, call_id: str, name: str, arguments: str) -> None:
       if not name:
           raise Exception("Did not get a function name")
   
       print(f"\n[Function Call] {name}({arguments})")
       tool_fn = TOOL_REGISTRY.get(name)
       if tool_fn is None:
           result = json.dumps({"error": f"Unknown function: {name}"})
       else:
           try:
               args = json.loads(arguments)
               result = tool_fn(**args)
               if asyncio.iscoroutine(result):
                   result = await result
           except Exception as e:
               result = json.dumps({"error": str(e)})
   
       print(f"[Function Result] {result}")
   
       # 함수 호출 출력을 모델에 다시 전달합니다.
       await send_event(ws, {
           "type": "conversation.item.create",
           "item": {
               "type": "function_call_output",
               "call_id": call_id,
               "output": result if isinstance(result, str) else json.dumps(result),
           },
       })
   
       # 모델이 함수 결과를 반영하도록 새 응답을 트리거합니다.
       await send_event(ws, {"type": "response.create"})
   
   
   def play_audio(output_stream: pyaudio.Stream, audio_output_queue: queue.Queue):
       """Runs in a separate thread because pyaudio's write() blocks until the
       sound card consumes the samples. Decoupling playback from the async event
       loop lets us flush the queue on interrupt without waiting for in-flight
       writes to finish."""
       while True:
           data = audio_output_queue.get()
           if data is None:
               break
           output_stream.write(data)
   
   
   async def send_mic_audio(ws, mic) -> None:
       try:
           while True:
               raw_data = mic.read(CHUNK, exception_on_overflow=False)
   
               # 볼륨 미터를 화면에 표시합니다.
               audio_data = np.frombuffer(raw_data, dtype=np.int16).astype(np.float64)
               rms = np.sqrt(np.mean(audio_data**2))
               meter = int(min(rms / 50, 50))
               print(f"Mic Level: {'█' * meter}{' ' * (50 - meter)} |", end="\r")
   
               # 오디오 청크를 Base64로 인코딩하여 전송합니다.
               b64_audio = base64.b64encode(raw_data).decode("utf-8")
               await send_event(ws, {
                   "type": "input_audio_buffer.append",
                   "audio": b64_audio,
               })
   
               await asyncio.sleep(0)
       except asyncio.CancelledError:
           pass
   
   
   async def receive_events(ws, audio_output_queue: queue.Queue) -> None:
       # 여러 delta 이벤트에 걸쳐 들어오는 함수 호출 인수를 누적합니다.
       pending_calls: dict[str, dict] = {}
       async for raw_message in ws:
           if DEBUG_WRITE_LOG:
               with open("data.jsonl", "a", encoding="utf-8") as f:
                   f.write(json.dumps(raw_message) + "\n")
   
           event = json.loads(raw_message)
           event_type = event.get("type", "")
   
           if event_type == "session.created":
               print(raw_message)
   
           elif event_type == "session.updated":
               print(raw_message)
   
           elif event_type == "error":
               print(f"\n[Error] {event}")
   
           elif event_type == "input_audio_buffer.speech_started":
               # AI가 사용자의 말과 겹쳐 말하지 않도록 큐에 쌓인 AI 오디오를 플러시합니다.
               while not audio_output_queue.empty():
                   try:
                       audio_output_queue.get_nowait()
                   except queue.Empty:
                       break
   
           elif event_type == "input_audio_buffer.speech_stopped":
               pass
   
           elif event_type == "input_audio_buffer.committed":
               pass
   
           elif event_type == "response.created":
               pass
   
           elif event_type == "response.output_text.delta":
               pass
   
           elif event_type == "response.output_text.done":
               pass
   
           # 오디오 출력 delta - 재생할 수 있도록 큐에 추가합니다.
           elif event_type == "response.output_audio.delta":
               audio_bytes = base64.b64decode(event.get("delta", ""))
               audio_output_queue.put(audio_bytes)
   
           elif event_type == "response.output_audio_transcript.delta":
               pass
   
           elif event_type == "response.output_audio_transcript.done":
               pass
   
           # 함수 호출 시작 - 대기 중인 호출을 초기화합니다.
           elif event_type == "response.output_item.added":
               item = event.get("item", {})
               if item.get("type") == "function_call" and item.get("status") == "in_progress":
                   item_id = item.get("id", "")
                   pending_calls[item_id] = {
                       "call_id": item.get("call_id", ""),
                       "name": item.get("name", ""),
                       "arguments": "",
                   }
                   print(f"\n[Function Call Started] {item.get('name', '')}")
   
           # 함수 호출 인수 delta - 누적합니다.
           elif event_type == "response.function_call_arguments.delta":
               item_id = event.get("item_id", "")
               if item_id in pending_calls:
                   pending_calls[item_id]["arguments"] += event.get("delta", "")
   
           elif event_type == "response.function_call_arguments.done":
               item_id = event.get("item_id", "")
               call_info = pending_calls.pop(item_id, None)
               if call_info is None:
                   # 폴백: done 이벤트의 데이터를 직접 사용합니다.
                   call_info = {
                       "call_id": event.get("call_id"),
                       "name": event.get("name"),
                       "arguments": event.get("arguments"),
                   }
               try:
                   await handle_function_call(
                       ws,
                       call_info["call_id"],
                       call_info["name"],
                       call_info["arguments"],
                   )
               except Exception as e:
                   print(f"Failed to call function for message {call_info}: error - {e}")
   
           elif event_type == "response.done":
               pass
   
           elif event_type == "rate_limits.updated":
               pass
   
           else:
               print(f"\n[Event: {event_type}]")
   
   
   async def main():
       if not OPENAI_API_KEY:
           print("Error: OPENAI_API_KEY environment variable not set")
           return
   
       p = pyaudio.PyAudio()
   
       input_device_index = int(p.get_default_input_device_info()['index'])
       output_device_index = int(p.get_default_output_device_info()['index'])
   
       # 채널 수는 장치가 지원하는 범위와 일치해야 합니다. 그렇지 않으면 open 시 pyaudio에서 오류가 발생합니다.
       input_info = p.get_device_info_by_index(input_device_index)
       output_info = p.get_device_info_by_index(output_device_index)
       input_channels = min(int(input_info['maxInputChannels']), 1)
       output_channels = min(int(output_info['maxOutputChannels']), 1)
   
       mic = p.open(
           format=FORMAT,
           channels=input_channels,
           rate=RATE,
           input=True,
           output=False,
           frames_per_buffer=CHUNK,
           input_device_index=input_device_index,
           start=False,
       )
       speaker = p.open(
           format=FORMAT,
           channels=output_channels,
           rate=RATE,
           input=False,
           output=True,
           frames_per_buffer=CHUNK,
           output_device_index=output_device_index,
           start=False,
       )
       mic.start_stream()
       speaker.start_stream()
   
       # 사용자가 말을 끊을 때 오디오를 플러시할 수 있도록 오디오를 큐를 거쳐 전달합니다.
       # 스피커에 직접 쓰면 이미 재생 중인 오디오를 취소할 수 없습니다.
       audio_output_queue = queue.Queue()
       threading.Thread(
           target=play_audio, args=(speaker, audio_output_queue), daemon=True
       ).start()
   
       headers = {
           "Authorization": f"Bearer {OPENAI_API_KEY}",
       }
   
       print("Connecting to OpenAI Realtime API...")
   
       async with websockets.connect(
           REALTIME_URL,
           additional_headers=headers,
       ) as ws:
           print("Connected! Configuring session...")
           await configure_session(ws)
   
           print("--- Session Active (Speak into mic) ---")
   
           mic_task = asyncio.create_task(send_mic_audio(ws, mic))
           try:
               await receive_events(ws, audio_output_queue)
           finally:
               mic_task.cancel()
               try:
                   await mic_task
               except asyncio.CancelledError:
                   pass
   
       # 정리
       audio_output_queue.put(None)  # 재생 스레드에 종료 신호를 보냅니다.
       mic.close()
       speaker.close()
       p.terminate()
       print("\nSession ended.")
   
   
   if __name__ == "__main__":
       asyncio.run(main())
   
4. weave.init() 호출에 사용자의 팀 이름과 프로젝트 이름을 입력하세요.
5. OPENAI_API_KEY 환경 변수를 설정하세요.
6. 음성 어시스턴트를 실행하세요:
   python weave_ws_voice_assistant.py