턴과 중단

턴과 중단

Realtime 프로바이더는 보통 음성 활동 감지(VAD)로 사용자가 말을 시작하고 멈추는 시점과 모델이 언제 응답해야 하는지를 결정해요. Pydantic AI는 이식 가능한 동작을 위한 공유 구성, 지원하는 프로바이더를 위한 명시적 중단, 푸시-투-토크 애플리케이션을 위한 수동 턴 제어를 노출해요.

출처: 문서

본문

자동 턴 감지

자동 감지는 기본으로 활성화돼요. TurnDetection로 공통 동작을 구성해요. sensitivity는 가장 가까운 프로바이더 제어로 매핑되고, prefix_padding_ms와 silence_duration_ms는 지원되는 곳에서 통과해요.

from pydantic_ai.realtime.openai import OpenAIRealtimeModel, OpenAIRealtimeModelSettings

settings = OpenAIRealtimeModelSettings(
    turn_detection={'sensitivity': 'high', 'silence_duration_ms': 400}
)
model = OpenAIRealtimeModel('gpt-realtime', settings=settings)

공유 제어로 충분하지 않을 때만 프로바이더별 설정을 사용하세요. openai_turn_detection, xai_turn_detection, google_vad는 turn_detection을 완전히 덮어써요. 그것들의 허용 값, 기본, 한계는 OpenAI, Azure OpenAI, Google Gemini, xAI 페이지에 문서화돼 있어요.

텍스트 턴

문자열을 보내면 완전한 사용자 턴을 만들고 모델에게 응답을 요청해요:

from pydantic_ai import BinaryImage
from pydantic_ai.realtime import RealtimeSession


async def send_turns(session: RealtimeSession, image: BinaryImage) -> None:
    await session.send('Greet the visitor.')

    # Add context for a later voice or text turn without asking for a reply.
    await session.send('The visitor is called Ada.', respond=False)

    # Show an image and ask for a reply in one operation.
    await session.send(image, respond=True)

이미지는 기본적으로 컨텍스트 전용이에요. 이미지에 대한 응답을 요청하려면 수동 턴 제어를 지원하는 모델이 필요해요.

send('...') 뒤에 create_response()를 호출하지 마세요. 텍스트 턴이 이미 응답을 요청하므로, 그 쌍은 두 번 요청해 모델이 같은 것을 두 번 말하게 할 수 있어요.

응답이 이미 진행 중일 때 OpenAI 프로토콜 프로바이더는 텍스트 턴을 큐에 넣고 다음에 답해요. Gemini 2.5도 마찬가지예요. 반면 Gemini 3.1은 진행 중인 응답을 중단하고, RealtimeResponseInterruptedEvent를 방출하며, 부분 응답을 중단된 것으로 기록하고, 새 텍스트 턴에 답해요.

마이크 음소거

서버 VAD에서 음소거는 오디오 스트림을 보존해야 해요. 오디오 프레임을 단순히 멈추면 프로바이더의 현재 음성 세그먼트가 무한정 열려 있을 수 있어요. 그래서 음소거 중에도 정상 주기로 0으로 채워진 PCM16 프레임을 계속 보내세요. 수동 턴 제어를 쓰면 보내기를 멈추고 clear_audio()를 호출해 부분 입력을 버리세요.

서버 VAD는 임의의 신호 에너지가 아니라 음성을 인식해요. 순수 톤은 음성 세그먼트를 시작하지 않을 수 있어요. 그래서 턴 감지 테스트에는 톤보다 녹음된 음성을 사용하세요.

Barge-in

서버 측 턴 감지에서 프로바이더는 새 사용자 음성을 감지하면 모델을 중단해요. 남는 것은 문제의 로컬 절반이에요. 재생을 위해 이미 큐에 넣은, 사용자가 절대 듣지 못할 오디오와, 그렇지 않으면 듣지도 못한 단어를 기록할 프로바이더 측 전사예요.

재생이 세션의 단일 stream_audio() 이터레이터를 소진할 때(다음 청크를 당기기 전에 각 청크를 디바이스에 쓰는) 세션은 그 절반을 스스로 처리할 수 있어요. 세션을 열 때 handle_barge_in=True를 전달하세요:

import asyncio
from collections.abc import AsyncIterator

from pydantic_ai import Agent

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


async def play_audio(chunks: AsyncIterator[bytes]) -> None:
    async for chunk in chunks:
        ...  # write the chunk to your speaker, waiting until the device consumed it


async def main():
    realtime = agent.realtime('openai:gpt-realtime')
    async with realtime.session(handle_barge_in=True) as session:
        playback = asyncio.create_task(play_audio(session.stream_audio()))
        ...  # stream the microphone and handle events; barge-in is handled for you
    await playback  # the audio view ends once the session has closed

사용자가 모델 위에서 말하면 세션은 사용자가 절대 듣지 못할 버퍼링된 오디오를 버리고, 프로바이더의 전사를 실제로 재생된 것까지 자르며, 응답을 취소해요. 이전 응답이 완전히 들렸을 때는 아무것도 하지 않아요. speech-start 신호가 일반 사용자 턴에서도 발화하니까요. 첫 오디오 청크에 도달하지 못한 응답도 멈춰져서, 모델의 thinking 시간에 말하는 것이 목소리에 말하는 것과 같이 작동해요. 프로바이더 차이는 흡수돼요. 출력 자르기가 없는 모델(xAI)에서는 중단점 없이 응답이 취소되고, 프로바이더가 발화 시작을 보고하지 않고 스스로 중단할 때(Gemini)는 로컬 플러시만 수행돼요. 이벤트는 당신의 이터레이터에 여전히 도달하고, 이미 처리돼 있어요. UI 상태나 세션이 닿을 수 없는 유일한 버퍼인 당신의 오디오 레이어 자체의 진행 중 블록을 플러시하려면 그것에 반응하세요. 중단점은 디바이스가 도달한 마지막 청크 경계라서, 실제로 들은 것보다 많아야 한 청크 덜 귀속돼요. 그 단일 이터레이터가 없으면(stream_audio() 소비자 없음 또는 여러 개) 귀속할 재생 위치가 없고, 플래그는 아래 수동 경로에 양보해요.

대안으로 barge-in을 스스로 처리할 수도 있어요. 신호: emits_input_speech_events를 선언하는 프로필을 가진 프로바이더(OpenAI, Azure OpenAI, xAI)는 사용자 음성이 시작될 때 RealtimeInputSpeechStartEvent를 방출해요. Gemini는 모델 출력을 중단할 때 RealtimeResponseInterruptedEvent를 방출해요. 프로바이더가 절대 보내지 않는 이벤트를 기다리지 말고 플래그를 읽으세요.

재생이 단일 디바이스 페이스 이터레이터를 유지하는 동안 트리거를 제어하는 것은 한 줄이면 돼요. 세션이 여전히 played_audio_bytes로 재생 위치를 추적하고(청크는 소비자가 다음 것을 위해 돌아올 때 재생된 것으로 셈), 그것을 interrupt(played_bytes=...)에 전달하면 handle_barge_in=True와 같은 플러시-귀속-자르기-취소 처리를 받아요:

import asyncio
from collections.abc import AsyncIterator

from pydantic_ai.realtime import RealtimeInputSpeechStartEvent, RealtimeSession


async def conversation(session: RealtimeSession) -> None:
    async def play_audio(chunks: AsyncIterator[bytes]) -> None:
        async for chunk in chunks:
            ...  # write the chunk to your speaker, waiting until the device consumed it

    playback = asyncio.create_task(play_audio(session.stream_audio()))
    async for event in session:
        if isinstance(event, RealtimeInputSpeechStartEvent):
            await session.interrupt(played_bytes=session.played_audio_bytes)
    playback.cancel()

대신 디바이스보다 앞서 버퍼링하는 재생 루프는 played_audio_bytes가 너무 앞서 읽게 만들어요. 실제 디바이스 소비를 스스로 세고 그것을 전달하세요. 이 핸들러는 발화 시작을 보고하는 프로바이더를 커버해요. Gemini는 스스로 중단하고 로컬 플러시만 남기므로, 당신을 위해 그 플러시를 수행하는 handle_barge_in=True를 선호하세요.

프로바이더의 발화 시작과 다음 응답 시작 사이에 중단하는 것은, 자신의 턴 감지가 말해지는 응답을 취소하는 모델(OpenAI와 Azure OpenAI 기본, xAI)에서 자르기만 보내요. 프로바이더의 것과 경쟁하는 두 번째 클라이언트 측 취소는 다음 응답에 적용되어 barge-in에 대한 응답을 침묵시킬 수 있어요. 이것은 모든 형태의 interrupt()에 적용돼요. 그 창 밖에서 발생시킨 중단(정지 버튼, 모델을 끊는 툴)은 여전히 취소해요. 다른 무엇도 멈추지 않으니까요.

마지막으로, 재생이 단일 세션 전체 stream_audio() 이터레이터를 소진하지 않을 때(여러 소비자, 디바이스보다 앞서 버퍼링하는 재생 레이어, 세션이 오디오를 건드리지 않는 전송) 자체 회계를 유지하고 played_ms(또는 아무것도)를 전달하세요. 여기서 Speaker는 당신의 재생 레이어 — 버퍼링된 오디오를 보고하고 플러시할 수 있는 무엇이든 — 를 대신해요:

from typing import Protocol

from pydantic_ai.realtime import RealtimeInputSpeechStartEvent, RealtimeSession


class Speaker(Protocol):
    def has_unplayed_audio(self) -> bool: ...
    def flush(self) -> None: ...
    def played_ms(self) -> int: ...


async def handle_events(session: RealtimeSession, speaker: Speaker):
    async for event in session:
        if isinstance(event, RealtimeInputSpeechStartEvent) and speaker.has_unplayed_audio():
            speaker.flush()
            if session.profile.get('supports_output_truncation', False):
                await session.interrupt(played_ms=speaker.played_ms())
            elif session.profile.get('supports_interruption', False):
                await session.interrupt()

played_ms로 위의 세션 측 편의 기능은 모두 당신 몫으로 다시 구현할 수 있어요. 중단하기 전에 재생되지 않은 오디오를 추적하고, 버퍼링된 재생을 스스로 플러시하세요. played_ms가 있는 interrupt()는 절대 플러시하지 않아요.

WebRTC 사이드밴드에서는 그 둘 사이에 세 번째 버퍼가 있어요. 프로바이더가 재생보다 훨씬 앞서 오디오를 생성하고 이미 만든 것을 계속 스트리밍하므로, 모델을 멈추는 것만으로는 목소리를 멈추지 못해요. interrupt()는 그 아웃바운드 버퍼도 버려서, 실제로 청취자에게 턴을 끝내는 것이에요. 브라우저는 여전히 자체 재생 버퍼를 소유하므로 위처럼 barge-in 시 플러시해야 해요.

이력은 SpeechPart.interrupted_at_ms에 알려진 컷오프를 기록하고 응답 상태를 중단된 것으로 표시해요. 이 이력이 텍스트 모델에 보내질 때 Pydantic AI는 저장된 이력을 수정하지 않고 준비된 요청에 읽을 수 있는 중단 노트를 추가해요.

먼저 말하기

재생이 이미 실행되는 상태에서 텍스트 턴을 보내 에이전트가 대화를 열게 하세요. 생성된 후 도착하는 인사말의 완성된 SpeechPart를 기다렸다가, 마이크를 열기 전에 재생 루프가 소진하게 두세요. 고정된 절전은 둘 다 말해주지 않아요.

import asyncio
from collections.abc import AsyncIterator

from pydantic_ai import Agent
from pydantic_ai.messages import PartEndEvent, SpeechPart

agent = Agent(instructions='You are a welcoming museum guide.')


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 main():
    async with agent.realtime('openai:gpt-realtime').session() as session:
        playback = asyncio.create_task(play_audio(session.stream_audio()))
        await session.send('Greet the visitor.')
        async for event in session:
            if isinstance(event, PartEndEvent) and isinstance(event.part, SpeechPart):
                if event.part.speaker == 'assistant':
                    break
        await session.wait_for_playback()
        ...  # open the microphone and start sending audio
    await playback  # the audio view ends once the session has closed

수동 턴 제어에서 create_response()는 텍스트 턴을 추가하지 않고 인사말을 요청할 수 있어요. 응답이 이미 활성화되어 있으면 요청은 그 응답이 완료될 때까지 보류되고, 사용자가 barge-in하면 버려져요. 그래서 create_response()에서 돌아오는 것이 음성이 시작됐다는 뜻은 아니에요.

서버 VAD는 기본으로 interrupt_response를 활성화해, 감지된 어떤 음성도 진행 중인 인사말을 취소해요. 여기에는 오디오 경로가 열리는 동안의 스피커 에코와 마이크 과도 신호가 포함돼요. 인사말이 재생될 때까지 마이크 음소거에서 설명한 대로 디지털 무음 프레임을 계속 보내며 마이크 캡처를 음소거하세요.

푸시-투-토크

supports_manual_turn_control을 선언하는 프로필을 가진 모델에서 turn_detection=False로 자동 감지를 비활성화하세요. 오디오를 스트리밍하고, commit_audio()를 호출해 사용자 턴을 끝낸 뒤, create_response()를 호출하세요. 명시적 create_response() 호출이 필요한 이유는 턴 감지가 꺼져 있으면 버퍼를 커밋하는 것이 사용자 입력만 확정하고, 당신이 요청하기 전까지 아무것도 응답을 트리거하지 않기 때문이에요. clear_audio()로 커밋되지 않은 입력을 버리세요.

from pydantic_ai import Agent
from pydantic_ai.realtime.openai import OpenAIRealtimeModel, OpenAIRealtimeModelSettings

agent = Agent()
model = OpenAIRealtimeModel(
    'gpt-realtime', settings=OpenAIRealtimeModelSettings(turn_detection=False)
)


async def main():
    async with agent.realtime(model).session() as session:
        await session.send_audio(b'...')
        await session.commit_audio()
        await session.create_response()

Gemini는 Pydantic AI를 통해 수동 턴 동사를 노출하지 않아요. turn_detection=False는 연결 전에 UserError를 발생시켜요.

모델이 지원하는 것 확인

이것들은 프로바이더 연결이 무엇을 할 수 있는지 묘사하는 모델 프로필 플래그예요. 에이전트에 동작을 추가하는 capabilities와 혼동하지 마세요. 프로바이더 이름이 아니라 RealtimeModelProfile로 분기하세요:

프로필 플래그 게이트
supports_manual_turn_control commit_audio(), clear_audio(), create_response()
supports_interruption interrupt()
supports_output_truncation interrupt(played_ms=...)

지원되지 않는 메서드를 호출하면 제어 메시지가 전송되기 전에 UserError가 발생해요. 현재 프로바이더 지원은 각 프로바이더 페이지에 요약돼 있어요.

엣지 케이스

  • 푸시-투-토크 침묵은 보통 commit_audio()나 create_response()를 생략했기 때문이에요.
  • 재생이 음성 감지를 트리거하면 디바이스나 WebRTC 레이어에서 에코 제거를 추가하고 실제 barge-in 시 재생을 신속히 플러시하세요.

더 알아보기 (Learn more)