Realtime(음성-대-음성)

Realtime(음성-대-음성)

Pydantic AI의 realtime 지원은 에이전트가 라이브 말로 하는 대화를 하게 해줘요. 사용자의 오디오를 음성-대-음성 모델로 스트리밍하고, 모델의 말로 하는 응답을 하나의 영구 연결로 다시 스트리밍해요. 그래서 지연이 낮고 중단이 자연스러워요.

realtime 세션은 Pydantic AI의 나머지와 같은 에이전트 툴, 의존성, 인스트럭션, 메시지 이력, capabilities, 사용량 한도, 옵저버빌리티를 사용해요. 그리고 그것이 핵심이에요. 호출 중간에 에이전트가 주문을 조회하거나, 가용성을 확인하거나, 로그인한 사용자의 데이터에 텍스트 에이전트가 쓰는 것과 같은 툴과 의존성으로 행동할 수 있어요. 호출 자체는 요약이나 구조화된 후속 처리를 위해 Agent.run()에 넘길 수 있는 평범한 메시지 이력이 되고, 같은 코드가 네 프로바이더에서 실행되며, 사용량 한도와 Logfire 추적이 내장돼요. 당신의 애플리케이션은 오디오 전송(백엔드를 통해 브리지되거나, OpenAI와 Azure에서는 WebRTC로 브라우저 직접)을 소유하고, Pydantic AI는 프로바이더 무관 에이전트 루프를 실행해요.

출처: 문서

본문

빠른 시작

OpenAI realtime 의존성으로 Pydantic AI를 설치하고 OPENAI_API_KEY를 설정하세요:

pip install "pydantic-ai-slim[openai-realtime]"

uv add "pydantic-ai-slim[openai-realtime]"

완전한 음성 에이전트는 에이전트 하나, 세션 하나, 그리고 세 개의 작은 루프(마이크 입력, 스피커 출력, 전사 로그)예요. 모델이 사용자를 듣고, 백엔드에서 당신의 툴을 호출하며, 소리 내어 답해요:

import asyncio
import contextlib
from collections.abc import AsyncIterator

from pydantic_ai import Agent

agent = Agent(instructions='You take reservations for The Terrace. Keep replies short.')


@agent.tool_plain
async def check_availability(day: str, party_size: int) -> str:
    """Check whether a table is free."""
    return f'One table for {party_size} is free at 7 pm {day}.'


async def microphone_chunks() -> AsyncIterator[bytes]:
    yield b'...'  # capture signed 16-bit mono PCM chunks from your microphone


async def play_audio(chunks: AsyncIterator[bytes]) -> None:
    async for chunk in chunks:
        ...  # write the PCM chunk to your speaker


async def main():
    async with agent.realtime('openai:gpt-realtime').session() as session:
        microphone = asyncio.create_task(session.send_audio(microphone_chunks()))
        speaker = asyncio.create_task(play_audio(session.stream_audio()))

        async for part in session.stream_transcripts():
            print(f'{part.speaker}: {part.transcript}')
            #> user: Hi! Do you have a table for two tomorrow night?
            #> assistant: We do: 7 pm, table for two. Want me to book it?
            if part.speaker == 'assistant':
                break  # keep listening in a real call; we stop after one exchange
        # 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 the speaker's audio stream --
    # but the microphone reads an external source, so stop it explicitly.
    microphone.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await microphone
    await speaker


if __name__ == '__main__':
    asyncio.run(main())

(이 예제는 완전해서 — 오디오 스택에 의존하는 두 오디오 플레이스홀더를 채우면 — "그대로" 실행할 수 있어요)

모델이 기대하는 샘플 비율로 캡처하고 재생하세요. 그것들은 모델의 프로필에서 보고되고 입력·출력이 다를 수 있어요(아래 프로바이더 지원 참고). 음성 어시스턴트 예제는 listentome로 플레이스홀더를 채워 실행 가능한 마이크·스피커 루프를 만들어요. 텍스트-음성 예제는 텍스트 프롬프트를 보내고 말로 한 응답을 WAV 파일로 저장함으로써 오디오 입력을 완전히 건너뛰어요.

세션이 어떻게 작동하는가

백엔드가 프로바이더 연결을 열고 RealtimeSession을 실행해요. send()나 send_audio()로 콘텐츠를 스트리밍하고, 세션을 반복해 이벤트 스트림(콘텐츠, 툴, 턴, 오류, 재연결 이벤트)을 얻거나, 퀵스타트처럼 전용 stream_audio()와 stream_transcripts() 뷰를 소비하세요.

device ↔ media bridge ↔ RealtimeSession ↔ provider
                         ├── typed tools
                         └── message history
                         (your backend)

미디어 브리지 는 사용자 디바이스와 백엔드 사이에서 오디오를 옮기는 무엇이든이에요. 브라우저 WebSocket 또는 텔레포니 브리지요. 로컬 마이크를 넘어 이렇게 배포하며, 각 설정은 프론트엔드 연결을 참고하세요. OpenAI와 Azure에서는 브라우저가 대신 WebRTC로 프로바이더와 직접 미디어를 교환하고, 백엔드는 미디어 브리지 대신 제어 평면 사이드밴드로 이 같은 루프를 실행해요.

작업별로 배우기

• 오디오, 이미지, 전사가 PCM 와이어 계약, 재생, 캡션, 입력 전사, 이미지 입력을 다뤄요.
• 이벤트가 세션 이벤트 어휘, 표준 실행과 공유되는 이벤트, 턴 경계를 다뤄요.
• 턴과 중단이 자동 턴 감지, barge-in, 출력 자르기, 푸시-투-토크를 다뤄요.
• 툴이 함수 툴, 프로바이더 네이티브 툴, 동시성, 승인, 호출 중 위임을 다뤄요.
• 기능과 훅이 기능과 그것의 훅이 세션에 어떻게 매핑되는지 다뤄요.
• 이력과 핸드오프가 보존된 전사, 오디오와 이미지, 세션 시딩, 표준 텍스트 에이전트로 계속하기를 다뤄요.
• 프론트엔드 연결이 사용자 디바이스와 백엔드 사이의 전송 옵션을 다뤄요.
• 연결 라이프사이클이 세션 라이프사이클, 재연결, 세션 한도, 오류를 다뤄요.
• 사용량과 옵저버빌리티가 사용량 한도, 비용 회계, Logfire, gateway 추적 전파를 다뤄요.
• 문제 해결이 증상별로 흔한 문제를 색인해요.
• API 참조가 세션과 코덱 타입을 나열하고 다른 프로바이더를 구현하는 방법을 설명해요.

프로바이더 지원

모든 프로바이더는 같은 RealtimeModel 인터페이스를 구현해요. 프로바이더 페이지가 설치, 모델 이름, 설정, 기능 지원, 특이점의 표준 출처예요:

이식 가능한 분기를 위해 RealtimeModel.profile 또는 RealtimeSession.profile을 검사하세요. RealtimeModelProfile은 캡처·재생할 오디오 샘플 비율과 위 표의 기능별 플래그(그 이상 포함)를 보고해요. 프로필은 표준 Model과 같은 방식으로 해석돼요(모델의 프로필 검사 참고) — 기본값, 프로바이더의 모델 이름 지식, 그 위의 당신의 profile= 인자. 모델 이름이 모델을 식별하지 않고 추론된 사실이 잘못됐을 때(가장 흔히 Azure 배포가 모델과 다른 이름), profile=을 전달하세요:

from pydantic_ai.realtime.azure import AzureRealtimeModel

# The deployment serves a reasoning model, but nothing in its name says so.
model = AzureRealtimeModel('voice-prod', profile={'supports_thinking': True})

부분 dict는 해석된 프로필 위에 병합돼요. 통째로 바꾸려면 callable (resolved) -> RealtimeModelProfile을 전달하세요.

공유 설정

Realtime 세션은 표준 실행에서 모델 실행 설정이 하는 역할을 하는 자체 설정 타입을 가져요. RealtimeModelSettings는 tool_choice부터 turn_detection까지 realtime 프로바이더 전반에 공유되는 설정을 정의해요. realtime 모델 생성자에 settings=로 기본값을 설정하거나, 한 세션에 realtime(model_settings=...)을 전달하세요. 세션별 값이 모델 기본값을 덮어써요:

from pydantic_ai import Agent
from pydantic_ai.realtime import RealtimeModelSettings

agent = Agent(instructions='You are a helpful voice assistant.')
realtime = agent.realtime(
    'openai:gpt-realtime', model_settings=RealtimeModelSettings(output_modality='audio')
)

음성과 세부 제어는 프로바이더별이에요. openai_voice, google_voice, xai_voice 등은 해당 프로바이더 설정 클래스에 있고, 기본과 한계는 프로바이더 페이지에 있어요.

에이전트의 일반 model_settings와 기능 get_model_settings() 기여는 realtime 세션을 구성하지 않아요. 요청-응답 모델처럼 지원되지 않는 공유 설정은 무시되는데, 한 가지 의도적인 예외가 있어요:

말 전용 모델에서 텍스트를 요청하면 빠르게 실패한다

supports_text_output=False를 보고하는 프로파일을 가진 모델(Gemini Live와 xAI)에서 output_modality='text'는 연결 전에 UserError를 발생시켜요. 조용히 음성으로 답하는 것이 시작하지 않는 것보다 더 나쁘기 때문이에요.

표준 에이전트 실행과의 관계

Agent.realtime()은 run()과 iter()의 장수, 양방향 형제예요. 매개변수는 그것들을 거울처럼 반영해요:

agent.realtime(
    model,                # 'openai:gpt-realtime', or a RealtimeModel instance
    deps=...,             # dependencies, as in run()/iter()
    model_settings=...,   # RealtimeModelSettings
    instructions=...,     # combined with the agent's instructions
    toolsets=...,         # additional toolsets for the session
    capabilities=...,     # additional capabilities for the session
    usage=..., usage_limits=...,
    message_history=...,  # prior conversation to seed the session with
)

표준 실행과 같은 의존성, 인스트럭션, 툴셋, capabilities, 사용량 한도, message_history를 받아요. 입력은 단일 user_prompt 대신 라이브 세션을 통해 도착해요:

전체 매핑은 기능과 훅을, 구조화된 출력이나 더 깊은 추론은 텍스트 에이전트로 핸드오프를 참고하세요.

음성 구축의 다른 방법

같은 realtime 루프는 WebRTC 또는 WebSocket 릴레이로 에이전트 코드를 바꾸지 않고 브라우저나 전화에 배포돼요. realtime 에이전트 루프가 제품에 맞지 않는다면, 두 대안이 그것 밖에 있어요:

• 배치 STT → 텍스트 에이전트 → TTS. 특정 텍스트 모델, 구조화된 출력, 또는 독립적으로 선택된 음성 구성요소를 원할 때, 표준 에이전트를 자체 음성-텍스트·텍스트-음성 서비스와 구성하세요.
• 브라우저에서 프로바이더로 직접. 일시적 토큰을 쓰는 프로바이더 네이티브, UI 전용 경험. 프로바이더의 자체 SDK가 세션을 소유하므로 서버 측 에이전트 루프, 툴, 공유 이력이 없어요. WebRTC 사이드밴드는 브라우저가 미디어를 소유하지만 백엔드가 여전히 에이전트를 실행해요. Pydantic AI는 여전히 별도 백엔드 워크플로우를 구동할 수 있어요.

한계

더 알아보기 (Learn more)

• Pydantic AI Realtime 시리즈
• Pydantic AI 문서 홈