UI Event Streams

import asyncio
from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel
 
agent = Agent(
    GoogleModel('gemini-3-flash-preview'),
    instructions='与えられたお題で短い物語を日本語で書いてください。',
)
 
async def main():
    async with agent.run_stream('森の中を歩く猫の話') as result:
        async for chunk in result.stream_text(delta=True):
            print(chunk, end='', flush=True)
        print()  # 末尾改行
 
asyncio.run(main())

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel
 
class Profile(BaseModel):
    name: str = Field(description='氏名')
    role: str = Field(description='役職')
    bio_ja: str = Field(description='100 字以内のバイオ')
 
agent = Agent(
    GoogleModel('gemini-3-flash-preview'),
    output_type=Profile,
    instructions='与えられた人物名で架空のプロフィールを作ってください。',
)
 
async def main():
    async with agent.run_stream('山田太郎') as result:
        async for partial in result.stream_output():
            print(f'name={partial.name!r}, role={partial.role!r}, bio={partial.bio_ja!r}')
 
asyncio.run(main())
# name='', role='', bio=''
# name='山田太郎', role='', bio=''
# name='山田太郎', role='エンジニア', bio=''
# name='山田太郎', role='エンジニア', bio='Python と Web 開発が...'

import asyncio
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.google import GoogleModel
 
agent = Agent(GoogleModel('gemini-3-flash-preview'))
 
@agent.tool
async def search(ctx: RunContext, query: str) -> str:
    """検索ツール (擬似)。"""
    return f'{query} の検索結果: ...'
 
async def main():
    async with agent.iter('東京の天気を検索して') as run:
        async for node in run:
            print(f'  → node: {type(node).__name__}')
            if Agent.is_model_request_node(node):
                # モデルの応答を chunk で観測
                async with node.stream(run.ctx) as stream:
                    async for event in stream:
                        # FunctionToolCallEvent / PartDeltaEvent / FinalResultEvent etc.
                        print(f'    event: {type(event).__name__}')
 
asyncio.run(main())

async with agent.run_stream(prompt) as result:
    async for chunk in result.stream_text(delta=True):
        if user_pressed_cancel:
            await result.cancel()
            break
 
if result.state == 'interrupted':
    # 部分結果は使わない、または明示的にクリーンアップ
    ...