텍스트를 오디오로
텍스트를 오디오로 (Text to Audio)
가장 작은 realtime 세션 예시예요: Python에서 일반 텍스트를 보내고 모델이 답변을 말하는 걸 들으면 돼요. OpenAI realtime 세션에 텍스트를 보내면 모델이 즉시 응답하도록 요청돼서, 마이크, 음성 활동 감지, 수동 턴테이킹을 관리할 필요가 없어요 — 그냥 send()와 세션의 이벤트 반복만 있으면 돼요.
이 예시가 보여주는 것:
- realtime 세션
- 텍스트 입력 / 오디오 출력 경로 (오디오 하드웨어 불필요)
SpeechPartDelta오디오 및 트랜스크립트 델타 스트리밍
스크립트는 음성 답변을 다시 스트리밍하고, 도착하는 대로 트랜스크립트를 출력하며, 오디오를 나중에 재생할 수 있는 .wav 파일로 저장해요. 기존 텍스트 챗봇을 말하는 챗봇으로 바꾸거나, 음성 사서함 인사말 같은 음성 스니펫을 만드는 데 유용한 시작점이에요.
예시 실행하기
realtime 모델은 gpt-realtime에서 실행되므로 OPENAI_API_KEY 환경 변수로 OpenAI API 키를 설정해야 해요.
의존성 설치와 환경 변수 설정이 끝나면 실행하세요:
터미널
python -m pydantic_ai_examples.realtime_text_to_audio "Tell me a fun fact about octopuses."
터미널
uv run -m pydantic_ai_examples.realtime_text_to_audio "Tell me a fun fact about octopuses."
스트리밍된 PCM 오디오는 realtime-response.wav에 저장되어 나중에 들을 수 있어요. 오디오 없이 턴이 완료되면 스크립트는 오류를 일으키고 빈 WAV 파일을 만들지 않아요.
예제 코드
realtime_text_to_audio.py
from __future__ import annotations
import asyncio
import sys
import wave
import logfire
from pydantic_ai import Agent, PartDeltaEvent, SpeechPartDelta
from pydantic_ai.realtime import RealtimeTurnCompleteEvent
from pydantic_ai.realtime.openai import OpenAIRealtimeModelSettings
# 'if-token-present' means nothing will be sent (and the example will work) if you don't have logfire configured
logfire.configure(send_to_logfire='if-token-present')
logfire.instrument_pydantic_ai()
# OpenAI's realtime models speak in 24 kHz mono PCM16 audio.
SAMPLE_RATE = 24000
DEFAULT_PROMPT = 'Tell me a fun fact about octopuses.'
OUTPUT_PATH = 'realtime-response.wav'
agent = Agent(
instructions='You are a friendly voice assistant. Keep your replies short and conversational.'
)
def save_wav(path: str, audio: bytes) -> None:
"""Wrap the streamed raw PCM16 audio in a WAV container so it can be played back."""
with wave.open(path, 'wb') as wav_file:
wav_file.setnchannels(1) # mono
wav_file.setsampwidth(2) # 16-bit samples
wav_file.setframerate(SAMPLE_RATE)
wav_file.writeframes(audio)
async def main(prompt: str, output_path: str) -> None:
audio = bytearray()
async with agent.realtime(
'openai:gpt-realtime',
model_settings=OpenAIRealtimeModelSettings(openai_voice='marin'),
).session() as session:
await session.send(prompt)
print(f'you: {prompt}')
print('assistant: ', end='', flush=True)
async for event in session:
match event:
case PartDeltaEvent(delta=SpeechPartDelta() as delta):
if delta.audio_chunk:
audio.extend(delta.audio_chunk)
if delta.transcript_delta:
print(delta.transcript_delta, end='', flush=True)
case RealtimeTurnCompleteEvent():
break
case _:
pass
print()
if not audio:
raise RuntimeError('The realtime response completed without any audio')
save_wav(output_path, bytes(audio))
print(f'\nSaved {len(audio)} bytes of audio to {output_path}')
if __name__ == '__main__':
prompt = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_PROMPT
asyncio.run(main(prompt, OUTPUT_PATH))
출처: 문서