Stream Whales

Stream Whales

고래에 대한 정보를 스트리밍 구조화 응답 검증(streamed structured response validation)으로 보여주는 예제예요. 이 예제는 스트리밍 출력을 실시간으로 검증하면서 데이터를 받아 화면에 그려 주는 흐름을 보여줍니다.

출처: 문서

본문

이 예제에서 확인할 수 있는 것:

이 스크립트는 고래에 대한 구조화 응답을 스트리밍하고, 데이터를 검증한 뒤 rich를 사용해 데이터가 도착하는 대로 동적 테이블로 표시해요.

예제 실행하기 (Running the Example)

의존성을 설치하고 환경 변수를 설정했다면, 다음과 같이 실행합니다:

python -m pydantic_ai_examples.stream_whales

uv run -m pydantic_ai_examples.stream_whales

이렇게 하면 아래 같은 출력이 나와요.

예제 코드 (Example Code)

from typing import Annotated

import logfire
from pydantic import Field
from rich.console import Console
from rich.live import Live
from rich.table import Table
from typing_extensions import NotRequired, TypedDict

from pydantic_ai import Agent

# '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 Whale(TypedDict):
    name: str
    length: Annotated[
        float, Field(description='Average length of an adult whale in meters.')
    ]
    weight: NotRequired[
        Annotated[
            float,
            Field(description='Average weight of an adult whale in kilograms.', ge=50),
        ]
    ]
    ocean: NotRequired[str]
    description: NotRequired[Annotated[str, Field(description='Short Description')]]


agent = Agent('openai:gpt-5.2', output_type=list[Whale])


async def main():
    console = Console()
    with Live('\n' * 36, console=console) as live:
        console.print('Requesting data...', style='cyan')
        async with agent.run_stream(
            'Generate me details of 5 species of Whale.'
        ) as result:
            console.print('Response:', style='green')

            async for whales in result.stream_output(debounce_by=0.01):
                table = Table(
                    title='Species of Whale',
                    caption='Streaming Structured responses from OpenAI',
                    width=120,
                )
                table.add_column('ID', justify='right')
                table.add_column('Name')
                table.add_column('Avg. Length (m)', justify='right')
                table.add_column('Avg. Weight (kg)', justify='right')
                table.add_column('Ocean')
                table.add_column('Description', justify='right')

                for wid, whale in enumerate(whales, start=1):
                    table.add_row(
                        str(wid),
                        whale['name'],
                        f'{whale["length"]:0.0f}',
                        f'{w:0.0f}' if (w := whale.get('weight')) else '...',
                        whale.get('ocean') or '...',
                        whale.get('description') or '...',
                    )
                live.update(table)


if __name__ == '__main__':
    import asyncio

    asyncio.run(main())

더 알아보기 (Learn more)