Streamed Output

import asyncio
from pydantic_ai import Agent
 
agent = Agent('google-gla:gemini-3-flash-preview', instructions='短い日本語で。')
 
async def main():
    async with agent.run_stream('PydanticAI の特徴を 3 つ教えて') as response:
        async for text in response.stream_text(delta=True):
            print(text, end='', flush=True)
        print()  # 末尾改行
 
asyncio.run(main())

import asyncio
from pydantic import BaseModel
from pydantic_ai import Agent
 
class Recipe(BaseModel):
    title: str
    ingredients: list[str]
    steps: list[str]
 
agent_r = Agent(
    'google-gla:gemini-3-flash-preview',
    output_type=Recipe,
    instructions='レシピを構造化して返してください。',
)
 
async def main():
    async with agent_r.run_stream('簡単なオムレツのレシピを教えて') as response:
        async for partial in response.stream():
            # partial は Recipe (の部分的な状態)
            print(f'\\rtitle={partial.title or "(未確定)"} / ingredients={len(partial.ingredients or [])} 件', end='')
        print()
        # ループを抜けたら完成形が確定
        final = await response.get_output()
        print(f'\\nFINAL: {final}')
 
asyncio.run(main())

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic_ai import Agent
 
app = FastAPI()
agent = Agent('google-gla:gemini-3-flash-preview', instructions='短く日本語で。')
 
@app.get('/chat')
async def chat(q: str):
    async def gen():
        async with agent.run_stream(q) as response:
            async for text in response.stream_text(delta=True):
                yield f'data: {text}\\n\\n'
        yield 'data: [DONE]\\n\\n'
    return StreamingResponse(gen(), media_type='text/event-stream')