Impossible to stream text responses ahead of final result when requiring structured output

[Blue9]

Initial Checks

• I confirm that I'm using the latest version of Pydantic AI
• I confirm that I searched for my issue in https://github.com/pydantic/pydantic-ai/issues before opening this issue

Description

This is a follow-up from a Slack conversation: https://pydanticlogfire.slack.com/archives/C083V7PMHHA/p1751066192272409

I want my agent to stream text, make tool calls, stream more text if needed, and make a final tool call for its structured output (I'm using Gemini which does not support tool calls and native structured output at the same time). To do so, I have set output_type to [str, FinalOutput] where FinalOutput is my structured output type.

This streams text as expected, but the final_result tool call is not always made. My understanding is that since the end node can either be FinalOutput or text, it will sometimes output text and there is no guarantee it outputs a FinalOutput. In my repro script, I have two example queries. One of them will call the final output tool call, and one of them will not (although this is dependent on the model's output being reproducible which isn't guaranteed).

I think what I would ideally want to do is to set output_type to FinalOutput so that it is always called but still enable streaming text output. That is currently prevented by this line in _agent_graph.py.

Is there a way to have pydantic-ai consistently stream text output AND make a final result tool call? One workaround is to just prompt the LLM to make the final result tool call, but that is not guaranteed.

Example Code

"""
To run: python <name>.py.

Requires GEMINI_API_KEY environment variable (available at https://aistudio.google.com/apikey).
Alternatively you can change the model but this will change the reproducibility of the script.
"""

from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.messages import (
    FinalResultEvent,
    FunctionToolCallEvent,
    FunctionToolResultEvent,
    PartDeltaEvent,
    PartStartEvent,
    TextPart,
    TextPartDelta,
)
from pydantic_ai.models import KnownModelName


class FinalOutput(BaseModel):
    lines: list[str]


QUERY = "Write a short poem about Python."  # Does not call final tool call.
# QUERY = "Write a poem about Python."  # Does call final tool call.

MODEL: KnownModelName = "google-gla:gemini-2.5-flash"
AGENT = Agent(
    model=MODEL,
    model_settings={"temperature": 0},
    output_type=[str, FinalOutput],
)


async def main():
    # Based on https://ai.pydantic.dev/agents/#streaming
    async with AGENT.iter(user_prompt=QUERY) as run:
        async for node in run:
            if Agent.is_user_prompt_node(node):
                print("\nUser prompt:", node.user_prompt)
            elif Agent.is_model_request_node(node):
                async with node.stream(run.ctx) as request_stream:
                    async for event in request_stream:
                        if isinstance(event, PartStartEvent) and isinstance(event.part, TextPart):
                            print("\nText start:", event.part.content)
                        elif isinstance(event, PartDeltaEvent) and isinstance(event.delta, TextPartDelta):
                            print("\nText delta:", event.delta.content_delta)
                        elif isinstance(event, FinalResultEvent):
                            print("\nFinal output tool call:", event.tool_name)
            elif Agent.is_call_tools_node(node):
                async with node.stream(run.ctx) as handle_stream:
                    async for event in handle_stream:
                        if isinstance(event, FunctionToolCallEvent):
                            print("\nTool call event:", event)
                        elif isinstance(event, FunctionToolResultEvent):
                            print("\nTool result event:", event)
            elif Agent.is_end_node(node):
                print("\nFinal output:", node.data.output)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Python, Pydantic AI & LLM client version

Python 3.11.13
pydantic-ai 0.4.0

[DouweM]

@Blue9 Thanks for the report.

I want my agent to stream text, make tool calls, stream more text if needed, and make a final tool call for its structured output (I'm using Gemini which does not support tool calls and native structured output at the same time).

The problem is that when using the tool output mode, the only way we can force a model to actually call the final_result tool to generate structured output is by forcing it to call tools instead of outputting text. Per the Gemini docs:

ANY: The model is constrained to always predict a function call and guarantees function schema adherence. [...] Use this mode when you require a function call response to every prompt (if applicable).

The tool_choice setting on other models APIs behaves the same, where "always use a tool" means "never use text". So the allow_text_output=False property effectively means force_tool_call=True.

Unfortunately using a model that supports native structured output (output_type=NativeOutput(...)) alongside tool calls doesn't fix this, as that would still stop the model from outputting regular text responses. Per the OpenAI docs:

Structured Outputs is a feature that ensures the model will always generate responses that adhere to your supplied JSON Schema, so you don't need to worry about the model omitting a required key, or hallucinating an invalid enum value.

Again, "always give me structured output" means "never use text".

The only workaround I'm aware of to get the experience you're looking for would be to add a write_to_user tool for the LLM to use. Then you can stream the tool argument deltas and manually construct the string to show to the user.

[Blue9]

Thanks for the response and this makes sense. I think your suggestion to add a separate tool is the cleanest solution for now.

Interestingly, if I set output_type=FinalOutput and hardcode allow_text_output to True in _agent_graph.py here, I get streaming text responses as well as the final result tool call. I do see that this skips the tool config in google.py though, which, from my understanding and based on the docs you linked, does not enforce the tool call (although in practice I have not seen it omit the call).

I've only tested this with Gemini but it has been reliable. What I don't understand is why this behaves differently than output_type=[str, FinalOutput] for the same input (e.g., the queries in the example code above). Do you know why this is the case?

[DouweM]

Interestingly, if I set output_type=FinalOutput and hardcode allow_text_output to True in _agent_graph.py here, I get streaming text responses as well as the final result tool call. I do see that this skips the tool config in google.py though, which, from my understanding and based on the docs you linked, does not enforce the tool call (although in practice I have not seen it omit the call).

@Blue9 Right, if we wanted to support streaming text + structured output officially (without treating text as final output), we'd need to handle the case where the LLMs just responds with text and no tool calls, possibly by sending back a "You haven't produced final output yet, use the final_result tool" message. This would only work in tool output mode, as with the model's native structured output mode there'd be no way to return non-structured text anyway.

What I don't understand is why this behaves differently than output_type=[str, FinalOutput] for the same input (e.g., the queries in the example code above). Do you know why this is the case?

@Blue9 What difference exactly are you seeing?

[Blue9]

In the above example, the query "Write a short poem about Python." outputs text but does not result in a tool call when running as-is (this is expected).

However, if I change output_type=[str, FinalOutput] to output_type=FinalOutput, and I modify _agent_graph.py by hardcoding allow_text_output = True, then it outputs both text and makes a final result tool call. I would have expected these to behave the same, but maybe there is a difference in the way these requests are made.

That said you are right in that there is no clean, guaranteed way to enable text output as well as a structured final result, so I think we can close this issue. I'm curious if you are able to observe the same behavior above though and if you have any ideas on why they behave differently.

[DouweM]

@Blue9 Sorry for the late response, I lost this GitHub notification somehow.

The reason for that difference is this validation, which doesn't use allow_text_output but rather the output_schema that was created directly from the output_type (and thus will fail if str is not included as an option):

pydantic-ai/pydantic_ai_slim/pydantic_ai/_agent_graph.py

Lines 541 to 547 in 090ec23

	if isinstance(output_schema, _output.TextOutputSchema):
	result_data = await output_schema.process(text, run_context)
	else:
	m = _messages.RetryPromptPart(
	content='Plain text responses are not permitted, please include your response in a tool call',
	)
	raise ToolRetryError(m)

That means the conversation with the model ends up looking like this: