Streaming: showing the answer as it is written

run_stream gives you the answer while the model is still writing it, so a chat window can show words straight away instead of after the whole reply.

import asyncio

from pydantic_ai import Agent
from pydantic_ai.models.function import FunctionModel


async def slow_reply(messages, info):
    for word in ["Your ", "refund ", "was ", "sent ", "today."]:
        yield word


agent = Agent(FunctionModel(stream_function=slow_reply))

A streaming model function is an async generator: it yields the text in pieces, the way a provider sends them. It is passed as stream_function=.

async def main():
    async with agent.run_stream("Where is my refund?") as result:
        async for text in result.stream_text(debounce_by=None):
            print(repr(text))


asyncio.run(main())

run_stream is an async context manager. stream_text() yields the answer so far, growing each time. debounce_by=None yields on every piece; the default, 0.1 seconds, groups pieces that arrive close together, which is kinder to a browser redrawing a page.

async def main():
    async with agent.run_stream("Where is my refund?") as result:
        async for piece in result.stream_text(delta=True, debounce_by=None):
            print(piece, end="|")
        print()
        print(result.usage)


asyncio.run(main())

delta=True yields only the new text each time, for code that appends. Once the stream is done, result.usage and the messages are there, as for a normal run.

Typed output streams too

With an output_type, result.stream_output() yields partial objects as fields arrive, and validators run on each. The Pydantic AI docs cover checking ctx.partial_output in a validator, so a rule that needs the complete answer only runs once.