에이전트 핸드오프

에이전트 핸드오프 (Agent Handoff)

Realtime 음성-음성 모델은 훌륭한 대화 상대지만 구조화된 출력을 만들지는 못해요. 이 예시는 견고한 패턴을 보여줘요: realtime 모델이 라이브 대화를 실행하게 한 다음, 그 메시지 히스토리를 output_type을 가진 일반 Agent.run()에 넘겨서 타입이 있는 결과를 추출해요.

realtime 세션은 텍스트 에이전트가 만드는 것과 동일한 ModelMessage 히스토리를 기록하기 때문에, 핸드오프는 session.all_messages()를 그대로 넘기는 것뿐이에요 — realtime 실행과 비-realtime 실행은 메시지 히스토리를 통해 상호운용되는 동료 관계예요.

이 예시가 보여주는 것:

realtime 세션
텍스트 에이전트 핸드오프를 통한 구조화된 출력
realtime 및 비-realtime 실행 간에 공유되는 메시지 히스토리

이 예시는 짧은 지원 통화를 모델링해요: 발신자가 realtime 음성 에이전트에게 문제를 설명하면, 축적된 대화가 이를 타입 있는 SupportTicket으로 정제하는 텍스트 에이전트에게 전달돼요. 발신자 쪽은 텍스트 턴으로 구동되어 마이크 없이 예시가 실행돼요 — 실제 앱은 대신 send_audio()로 마이크 오디오를 스트리밍할 거예요 (보이스 어시스턴트 예시 참고).

핸드오프는 스크립트된 각 발신자 턴이 RealtimeTurnCompleteEvent를 받은 뒤에만 실행돼요. realtime 연결이 일찍 끝나면, 부분 통화에서 티켓을 만들지 않고 예시가 오류를 일으켜요.

예시 실행하기

realtime gpt-realtime 모델과 텍스트 트라이지 에이전트 모두 OpenAI에서 실행되므로 OPENAI_API_KEY 환경 변수로 OpenAI API 키를 설정해야 해요.

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

터미널

python -m pydantic_ai_examples.realtime_handoff

터미널

uv run -m pydantic_ai_examples.realtime_handoff

예제 코드

realtime_handoff.py

from __future__ import annotations

import asyncio
from typing import Literal

import logfire
from pydantic import BaseModel

from pydantic_ai import Agent, PartEndEvent, SpeechPart
from pydantic_ai.realtime import RealtimeTurnCompleteEvent

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


class SupportTicket(BaseModel):
    """The structured ticket distilled from the spoken support call."""

    summary: str
    category: Literal['hardware', 'software', 'billing', 'other']
    priority: Literal['low', 'medium', 'high']
    follow_up_questions: list[str]


# The realtime model runs the live conversation.
voice_agent = Agent(
    instructions='You are a friendly, concise phone support agent. Ask one question at a time.'
)

# A normal text agent turns the finished conversation into a typed result -- something a realtime
# model can't do itself.
triage_agent = Agent(
    'openai:gpt-5.2',
    output_type=SupportTicket,
    instructions='Summarize the support call as a structured ticket.',
)

CALLER_TURNS = [
    "Hi, my laptop won't charge anymore -- the light doesn't come on when I plug it in.",
    'I already tried a different outlet and it still does nothing. I need it for a presentation tomorrow.',
]


async def main() -> None:
    async with voice_agent.realtime('openai:gpt-realtime').session() as session:
        remaining_turns = iter(CALLER_TURNS)
        first_turn = next(remaining_turns)
        print(f'caller: {first_turn}')
        await session.send(first_turn)

        async for event in session:
            match event:
                case PartEndEvent(
                    part=SpeechPart(speaker='assistant', transcript=transcript)
                ) if transcript:
                    print(f'agent: {transcript}')
                case RealtimeTurnCompleteEvent():
                    next_turn = next(remaining_turns, None)
                    if next_turn is None:
                        break
                    print(f'caller: {next_turn}')
                    await session.send(next_turn)
                case _:
                    pass
        else:
            raise RuntimeError(
                'The realtime session ended before the support call completed'
            )

        handoff_history = session.all_messages()

    ticket = await triage_agent.run(
        'Create the support ticket for this call.', message_history=handoff_history
    )
    print(f'\nStructured ticket:\n{ticket.output.model_dump_json(indent=2)}')


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

출처: 문서

에이전트 핸드오프