Streaming

Stream AI responses in real-time

The stream_text function enables real-time streaming of AI responses. This provides a better user experience by showing text as it’s generated rather than waiting for the complete response.

stream_text exposes a unified typed event_stream for text, provider thinking, and lifecycle events. Its text_stream property remains a text-only projection, and on_reasoning_event remains available for callback-based integrations.

Basic Streaming

import asyncio
from ai_query import stream_text
from ai_query.providers import google

async def main():
    result = stream_text(
        model=google("gemini-2.0-flash"),
        prompt="Write a short story about a robot."
    )

    async for chunk in result.text_stream:
        print(chunk, end="", flush=True)

    print()  # New line at the end

asyncio.run(main())

stream_text returns immediately with a TextStreamResult object. The provider request starts when you iterate over event_stream or text_stream, or await a final result property.

Getting Usage Statistics

Token usage is available after streaming completes:

result = stream_text(
    model=google("gemini-2.0-flash"),
    prompt="Explain machine learning."
)

async for chunk in result.text_stream:
    print(chunk, end="", flush=True)

# Await usage after streaming
usage = await result.usage
print(f"\nTokens: {usage.total_tokens}")

Streaming with Tools

Combine streaming with tool calling for powerful agentic workflows:

from ai_query import stream_text
from ai_query.providers import google, tool, Field

@tool(description="Search Wikipedia for information")
async def search_wikipedia(query: str = Field(description="Search query")) -> str:
    # Simulated search
    return f"Wikipedia results for '{query}': ..."

async def main():
    result = stream_text(
        model=google("gemini-2.0-flash"),
        prompt="Search for information about Python programming and summarize it.",
        tools={"search_wikipedia": search_wikipedia}
    )

    async for chunk in result.text_stream:
        print(chunk, end="", flush=True)

asyncio.run(main())

When tools are provided, the streaming handles the full execution loop:

1. The AI decides to call a tool
2. The tool executes
3. The AI receives the result
4. The AI generates the final response (streamed)

Parameters

Parameter	Type	Description
model	LanguageModel	The AI model to use
prompt	str	The user prompt
messages	list[Message]	Full conversation history
system	str	System prompt
tools	dict[str, Tool]	Available tools
stop_when	StopCondition	Stop condition for tool loops
on_step_start	Callable	Step start callback
on_step_finish	Callable	Step finish callback
on_reasoning_event	Callable	Callback for provider thinking output during streaming
reasoning	ReasoningConfig	Normalized reasoning-depth controls like effort and budget
provider_options	dict	Provider-specific options

Typed Events

Consume event_stream when one ordered channel needs to drive a UI, transport, or log:

result = stream_text(model=model, prompt="Explain binary search.")

async for event in result.event_stream:
    if event.type == "text.delta":
        render_text(event.text)
    elif event.type.startswith("reasoning."):
        render_reasoning(event.event)
    elif event.type == "step.finished":
        record_step(event.step)

The stream emits step.started and step.finished around each model step, then one stream.finished event. Both stream views are replayable and share one provider request, including when consumed concurrently.

When a completed tool call is available, the same stream emits tool_call.ready, tool_execution.started, tool_execution.finished, and tool_result. Parallel finish events arrive as tools complete; normalized result events remain in provider order.

OpenAI-compatible, Anthropic, and Bedrock streams also emit tool_call.started and tool_call.delta while function names and JSON arguments are being assembled. Use the step number and provider index for correlation until a tool call ID is available. Providers that only return completed calls, such as Gemini functionCall parts, still get normalized start/delta events immediately before tool_call.ready.

Thinking Events

Reasoning is emitted as reasoning.summary, reasoning.delta, reasoning.signature, or reasoning.state. Summary and delta events are both streamed reasoning output, but neither is included in text_stream. Callback-based integrations can continue using on_reasoning_event:

from ai_query import stream_text
from ai_query.providers import google

def on_reasoning(event):
    print(f"\n[{event.kind}]", event.text or event.data)

result = stream_text(
    model=google("gemini-2.5-pro"),
    prompt="Explain binary search.",
    provider_options={
        "google": {
            "thinking_config": {
                "thinking_budget": 1024,
                "include_thoughts": True,
            }
        }
    },
    on_reasoning_event=on_reasoning,
)

async for chunk in result.text_stream:
    print(chunk, end="")

See Thinking for provider differences and guidance.

TextStreamResult Object

The stream_text function returns a TextStreamResult:

result = stream_text(model=google("gemini-2.0-flash"), prompt="Hello")

# Async iterator for text chunks
async for chunk in result.text_stream:
    print(chunk, end="")

# Await to get final values
usage = await result.usage        # Token usage
text = await result.text          # Complete text (after streaming)
tool_calls = await result.tool_calls  # Any tool calls made

Collecting Full Text

If you need the complete text after streaming:

result = stream_text(
    model=google("gemini-2.0-flash"),
    prompt="Write a poem."
)

# Stream to console
async for chunk in result.text_stream:
    print(chunk, end="")

# Get full text
full_text = await result.text
print(f"\n\nFull response ({len(full_text)} chars)")

Real-time UI Updates

Stream directly to a user interface:

async def stream_to_ui(prompt: str, update_callback):
    """Stream AI response to a UI callback."""
    result = stream_text(
        model=google("gemini-2.0-flash"),
        prompt=prompt
    )

    buffer = ""
    async for chunk in result.text_stream:
        buffer += chunk
        await update_callback(buffer)

    return buffer

Next Steps

Tools

Add function calling to your streams

Agents

Build agents with streaming output