오디오, 이미지, 전사(transcript)

realtime 세션은 라이브 오디오, 텍스트, 지원되는 이미지를 받아들이고, 재생과 캡션을 위한 별도 뷰를 노출해요. 미디어와 전사에는 하이레벨 세션 뷰를 사용하고, 툴, 턴 경계, 재연결, 오류에는 메인 이벤트 스트림을 사용하세요.

출처: 문서

본문

오디오 와이어 계약

당신은 원시 오디오 샘플을 보내고 받아요. 라이브 경로에는 컨테이너나 코덱이 없어요. send_audio()는 원시 부호 있는 16비트 리틀엔디안 모노 PCM을 받아요. 단일 청크이거나, 청크의 async 이터러블(마이크 스트림, WebSocket 수신 루프)이고, 이터러블이 끝날 때까지 전달해요. 그래서 전체 캡처 루프가 하나의 태스크일 수 있어요. stream_audio()는 같은 형식을 반환해요. session.audio_input_sample_rate에서 캡처하고 session.audio_output_sample_rate에서 재생하세요. 입력과 출력 비율은 다를 수 있어요.

인터랙티브 주기와 청크당 오버헤드의 균형을 맞추려면 100ms 입력 청크로 시작하고, 당신의 전송에 맞게 조정하세요. 프로바이더 페이지들이 모델별 비율과 제약을 나열해요. OpenAI, Azure OpenAI, Google Gemini, xAI.

바운드된 버퍼, 재생 회계, 깔끔한 종료가 있는 완전한 마이크·스피커 루프를 위해서는 realtime 음성 어시스턴트 예제를 사용하세요.

오디오와 전사 소비

메인 이터레이터와 함께 미디어 뷰를 실행하세요:

import asyncio
from collections.abc import AsyncIterator

from pydantic_ai import Agent
from pydantic_ai.messages import SpeechPart
from pydantic_ai.realtime import RealtimeTurnCompleteEvent

agent = Agent(instructions='You are a helpful voice assistant.')


async def play_audio(chunks: AsyncIterator[bytes]) -> None:
    async for chunk in chunks:
        ...  # Write the PCM16 chunk to your speaker or audio output stream.


async def show_transcripts(parts: AsyncIterator[SpeechPart]) -> None:
    async for part in parts:
        print(part.speaker, part.transcript)
        #> assistant Hello from the realtime assistant.


async def main():
    async with agent.realtime('openai:gpt-realtime').session() as session:
        audio_task = asyncio.create_task(play_audio(session.stream_audio()))
        transcript_task = asyncio.create_task(show_transcripts(session.stream_transcripts()))
        async for event in session:
            if isinstance(event, RealtimeTurnCompleteEvent):
                break
        # Let the speaker consume every generated chunk before closing the session.
        await session.wait_for_playback()

    # Leaving the `async with` block closes the session, which ends every live view.
    await asyncio.gather(audio_task, transcript_task)

각 뷰는 독립적으로 바운드돼요. 느린 소비자는 툴, 턴 추적, 다른 소비자를 멈추는 대신 가장 오래된 항목을 버려요. 구독은 stream_audio()나 stream_transcripts()가 호출될 때 시작돼요. 그래서 asyncio.create_task로 태스크에 넘겨진 뷰는 이벤트 루프에서 첫 턴을 기다리는 동안 버퍼 바운드까지만 아무것도 놓치지 않아요. 위처럼 태스크가 생성되는 곳에서 메서드를 호출하고 이터레이터를 전달하세요. 태스크 본문 안의 async for chunk in session.stream_audio()는 태스크가 처음 실행될 때만 구독해서, 그 전에 방출된 오디오는 절대 안 보여요. 소비되지 않는 뷰는 바운드까지 버퍼링하고, 가득 차면 가장 오래된 항목을 버려요. 수집될 때까지요. close()는 대기 중인 항목을 버리고 모든 라이브 이터레이터를 끝내요. closed가 그 상태를 보고해요.

아무도 세션을 반복하지 않으면, 세션은 가장 최근 512개 파트 델타 이벤트(오디오, 전사, 텍스트)와 가장 최근 512개 구조적 이벤트를 늦은 async for를 위해 유지해요. 더 오래된 것은 버려져요. 파트의 시작을 버리면 그 파트의 나머지도 함께 버려져서, 늦은 이터레이터는 파트에 붙일 수 없는 델타를 절대 받지 않아요. 소비자를 위해 대기 중인 실패는 절대 버려지지 않아요.

응답 생성이 끝난 후, 세션을 닫거나 마이크를 열기 전에 wait_for_playback()을 await 하세요. 단일 stream_audio() 소비자가 지금까지 방출된 모든 오디오를 회계했을 때 반환해요. 재생됐거나(재생된 것이 하나의 청크 지연으로 회계되는, played_audio_bytes와 같은), 아예 재생되지 않았거나(barge-in이나 뷰의 버퍼 오버플로로 버려지거나, 뷰가 구독하기 전에 방출)요. 정확히 하나의 오디오 뷰를 요구하고, 그 뷰나 세션이 닫히면 그것도 반환해요.

라이브 캡션

라이브 캡션을 위해 stream_transcripts()에 delta=True를 전달하세요. 각 TranscriptUpdate는 스피커, 새 델타, 지금까지의 전체 전사, 턴을 식별하는 인덱스를 포함해요. 음성 인식이 이전 단어를 수정할 수 있으므로, 캡션을 맹목적으로 덧붙이지 말고 인덱스로 교체하세요:

from pydantic_ai.realtime import RealtimeSession

bubbles: dict[int, tuple[str, str]] = {}


async def show_captions(session: RealtimeSession) -> None:
    async for update in session.stream_transcripts(delta=True):
        bubbles[update.index] = (update.speaker, update.transcript)

입력 전사

공유 input_transcription_model 설정은 사용자 음성이 텍스트로 이력에 도달하는지 제어해요:

값	동작
`'auto'`(기본)	프로바이더의 권장 전사 경로를 사용
모델 ID	지원하는 프로바이더에서 전용 전사 모델을 고정
`None`	입력 전사 비활성화

OpenAI, Azure OpenAI, xAI는 전용 전사 모델을 사용해요. Gemini는 네이티브 전사를 사용하며, google_input_transcription으로 구성해요. 공유 설정의 고정된 모델 ID는 무시되고(네이티브 전사 유지), None만 그것을 끄요. 프로바이더별 기본과 배포 제약은 프로바이더 페이지에 있어요.

사용자 전사는 별도의 전사 패스이지 realtime 모델이 오디오에서 직접 들은 것의 판독이 아니에요. 덜 정확하거나 그냥 다를 수 있으므로, 모델이 왜 그렇게 응답했는지의 근거 진실이 아니라 캡션과 이력 기록으로 취급하세요.

전사 비활성화는 말로 한 턴이 이력, 재생, 텍스트 에이전트 핸드오프에 기여하는 것을 바꿔요. 의존하기 전에 이력과 핸드오프를 참고하세요. WebRTC 사이드밴드는 보존할 오디오 바이트를 받지 않아서, 입력 전사 없이는 사용자 턴에 말로 한 텍스트가 없어요.

이미지

오디오와 텍스트 외에도, 세션은 표준 실행의 멀티모달 입력과 같은 이미지 콘텐츠를 받아요. send()로 컨텍스트로서 이미지를 보내세요. 이미지는 그 자체로 응답을 트리거하지 않아요. 모델이 다음 음성, 텍스트, 또는 수동 생성 턴에서 사용해요. 이미지에 대한 응답을 요청하려면 respond=True를 전달하세요. respond 동작은 텍스트 턴을 참고하세요.

from pydantic_ai import BinaryContent


async def send_image(session):
    jpeg_bytes = b'...'
    await session.send(BinaryContent(data=jpeg_bytes, media_type='image/jpeg'))

이미지를 연속적으로 스트리밍하면 라이브 비디오에 근접해요. 카메라 예제는 마이크 오디오와 함께 초당 한 프레임을 보내요. 그런 연속 스트림에서는 세션의 이미지 보존 컨트롤로 로컬 이력을 바운드하세요. 그것은 프로바이더가 받는 프레임을 바꾸지 않아요. 이미지 보존을 참고하세요. Gemini 특화 라이브 비디오 설정은 Gemini 프로바이더 페이지에 있어요.

엣지 케이스

오디오와 전사 이터레이터는 소비자가 뒤처질 때 오래된 버퍼링 항목을 의도적으로 버려요. Logfire 속성이 그 드롭을 보고해요.
이 뷰들만 소비할 때 세션 실패는 다른 전파 경로를 가져요. 오류 참고.
프로바이더 음성/중단 신호는 달라요. 프로바이더 이름으로 분기하지 말고 프로필 플래그와 턴 가이드를 사용하세요.

오디오, 이미지, 전사(transcript)