Add Agent.run_stream_sync method and sync convenience methods on StreamedRunResult

[ajac-zero]

Fixes #3005.

This PR tries implementing a possible solution to #3005 by adding a new run_stream_sync method that uses run_stream under the hood, based on the run_sync implementation that uses run internally.

Currently, it uses an 'eager' approach by loading the async iterable upfront and then providing normal iterable to access the stream. This has an impact on latency as it is not true streaming, but it avoids some opaque race conditions that can appear in a 'lazy' implementation.

I'd like to get some feedback on whether this approach is worthwhile or if it would be better to try and get a working 'lazy' implementation. My thinking is that, if latency is a concern, it would probably be best to point users towards the async version regardless, leaving the sync methods as conveniences for certain cases.

from pydantic_ai import Agent

agent = Agent('openrouter:openai/gpt-4o')

with agent.run_stream_sync('What are the capitals of France and Germany?') as response:
    for text in response.stream_text():
        print(text)

# The capital of France is Paris, and the capital
# The capital of France is Paris, and the capital of Germany is Berlin
# The capital of France is Paris, and the capital of Germany is Berlin.

[DouweM]

I'd like to get some feedback on whether this approach is worthwhile or if it would be better to try and get a working 'lazy' implementation. My thinking is that, if latency is a concern, it would probably be best to point users towards the async version regardless, leaving the sync methods as conveniences for certain cases.

@ajac-zero Thanks for working on this! I do think that streaming means people want to see the value as they come in, even when running sync code -- it'll just block until the next chunk is received, or buffer chunks if sync processing takes longer than the next chunk comes in. Otherwise, there's not much point in using this new method over just run_sync without streaming.

[ajac-zero]

@DouweM I took a shot at true streaming implementation, and so far so good. What do you think?

[DouweM]

Have you seen pydantic_ai.direct.StreamedResponseSync? I wonder if there's some overlap we could factor out.

[DouweM]

I don't love copying the private values here, even if this is a subclass...

Would it help to add a new class, or to add _sync methods to StreamedRunResult directly?

[ajac-zero]

Hi @DouweM. I removed the new class in favor of _sync methods in StreamedRunResult. The public API changes a bit more but the code is much cleaner.

[DouweM]

Since it's sync, do we need to yield or could we just return?

[DouweM]

Let's add this note that's in the agent.run_sync docstring to all these methods:

This is a convenience method that wraps [`self.X`][pydantic_ai.result.StreamedRunResult.X] with `loop.run_until_complete(...)`.
        You therefore can't use this method inside async code or if there's an active event loop.

[DouweM]

Also to run_stream_sync

[DouweM]

We need all the overloads we have for run_stream itself.

[DouweM]

Also let's mention this in agents.md where we also introduce run_sync and run_stream. It doesn't need a new list item, but can be at the end of the run_stream explanation.