보이스 어시스턴트

보이스 어시스턴트 (Voice Assistant)

realtime 음성-음성 모델 위에 구축된 보이스 어시스턴트 예시예요. 마이크를 OpenAI의 gpt-realtime 모델로 스트리밍하고, 모델의 음성 답변을 스피커로 재생해요. 말을 걸어보세요 — 그리고 모델이 말하는 동안 끼어들어 보세요: 모델이 멈추고 듣기 시작해요 (barge-in).

이 예시가 보여주는 것:

realtime 세션
tools
barge-in (모델이 문장 중간에 말하는 걸 끊기)

에이전트는 대화 중간에 모델이 호출할 수 있는 get_weather 도구 하나를 노출하고, 터미널은 양쪽 대화와 도구 호출의 실시간 트랜스크립트를 보여줘요.

오디오 I/O는 listentome에서 실행돼요. 이 라이브러리의 마이크는 send_audio()가 직접 소비하는 비동기 이터레이터이고, 스피커의 write()는 stream_audio()의 각 청크를 장치가 재생할 때까지 일시 중지돼요. 두 오디오 방향 모두 제한 없이 늘어나지 않도록 유지돼요: 마이크 스트림과 세션의 오디오 버퍼는 각각 소비자가 뒤처지면 가장 오래된 블록을 버려서, 멈추는 머신은 통화를 끝내는 대신 글리치만 발생해요.

Barge-in은 예시 코드가 전혀 들지 않아요. 재생이 단일 장치 페이스 stream_audio() 루프이기 때문에 세션이 재생 위치를 스스로 추적할 수 있고, handle_barge_in=True가 로컬 쪽 절반을 처리해요 — 사용자가 절대 듣지 못할 버퍼링된 오디오를 버리고, 공급자의 트랜스크립트를 실제로 들린 부분으로 잘라내며, 이전 답변이 끝까지 들린 일반 턴에서는 방해하지 않아요. 유일하게 닿지 못하는 것은 이미 스피커 안에 있는 블록이라, 최대 한 청크의 오래된 오디오가 재생을 마쳐요. 세션이 따라갈 수 없는 재생 루프나 직접 소유하고 싶은 트리거는 barge-in 가이드의 수동 경로를 쓰면 돼요.

예시 실행하기

이 예시의 의존성에는 마이크 및 스피커 접근을 위한 listentome이 포함돼요. 또한 PortAudio 시스템 라이브러리도 필요해요: macOS는 brew install portaudio, Debian/Ubuntu는 apt install libportaudio2.

realtime 모델은 gpt-realtime에서 실행되므로 OPENAI_API_KEY 환경 변수로 OpenAI API 키를 설정해야 해요.

의존성 설치와 환경 변수 설정이 끝나면 실행하세요:

터미널

python -m pydantic_ai_examples.realtime_voice

터미널

uv run -m pydantic_ai_examples.realtime_voice

예제 코드

realtime_voice.py

from __future__ import annotations

import anyio
import listentome
import logfire

from pydantic_ai import (
    Agent,
    FunctionToolCallEvent,
    FunctionToolResultEvent,
    PartEndEvent,
    SpeechPart,
)
from pydantic_ai.realtime import RealtimeSession

# '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()

agent = Agent(
    instructions='You are a friendly voice assistant. Keep your replies short and conversational.'
)


@agent.tool_plain
def get_weather(city: str) -> str:
    """Look up the current weather in a city."""
    return f'It is currently 21 degrees and sunny in {city}.'


async def conversation(session: RealtimeSession) -> None:
    """Wire the microphone and speaker to the session and run the conversation."""
    mic = listentome.InputStream(
        samplerate=session.audio_input_sample_rate,
        channels=1,
        dtype='int16',
        blocksize=session.audio_input_sample_rate // 10,  # 100 ms per block
    )
    speaker = listentome.OutputStream(
        samplerate=session.audio_output_sample_rate, channels=1, dtype='int16'
    )

    async with mic, speaker, anyio.create_task_group() as tg:
        tg.start_soon(session.send_audio, mic)

        async def play_audio() -> None:
            async for chunk in session.stream_audio():
                await speaker.write(chunk)

        tg.start_soon(play_audio)

        print('Listening -- start talking (Ctrl-C to quit).')
        async for event in session:
            match event:
                case PartEndEvent(part=SpeechPart() as part) if part.transcript:
                    print(f'{part.speaker}: {part.transcript}')
                case FunctionToolCallEvent(part=call):
                    print(f'[calling {call.tool_name}]')
                case FunctionToolResultEvent(part=result):
                    print(f'[{result.tool_name} returned: {result.content}]')
                case _:
                    pass
        tg.cancel_scope.cancel()


async def main():
    realtime = agent.realtime('openai:gpt-realtime')
    async with realtime.session(handle_barge_in=True) as session:
        await conversation(session)


if __name__ == '__main__':
    try:
        anyio.run(main)
    except KeyboardInterrupt:
        pass

출처: 문서

보이스 어시스턴트