Tweaks, add validate_response_output_sync

Author: DouweM

pydantic_ai_slim/pydantic_ai/_utils.py

@@ -234,6 +234,15 @@ def sync_anext(iterator: Iterator[T]) -> T:
         raise StopAsyncIteration() from e
 
 
+def sync_async_iterator(async_iter: AsyncIterator[T]) -> Iterator[T]:
+    loop = get_event_loop()
+    while True:
+        try:
+            yield loop.run_until_complete(anext(async_iter))
+        except StopAsyncIteration:
+            break
+
+
 def now_utc() -> datetime:
     return datetime.now(tz=timezone.utc)
 

pydantic_ai_slim/pydantic_ai/agent/abstract.py

@@ -12,7 +12,6 @@
 from typing_extensions import Self, TypeIs, TypeVar
 
 from pydantic_graph import End
-from pydantic_graph._utils import get_event_loop
 
 from .. import (
     _agent_graph,
@@ -335,7 +334,7 @@ def run_sync(
         if infer_name and self.name is None:
             self._infer_name(inspect.currentframe())
 
-        return get_event_loop().run_until_complete(
+        return _utils.get_event_loop().run_until_complete(
             self.run(
                 user_prompt,
                 output_type=output_type,
@@ -685,8 +684,7 @@ def main():
             The result of the run.
         """
         if infer_name and self.name is None:
-            if frame := inspect.currentframe():  # pragma: no branch
-                self._infer_name(frame)
+            self._infer_name(inspect.currentframe())
 
         async def _consume_stream():
             async with self.run_stream(
@@ -706,7 +704,7 @@ async def _consume_stream():
             ) as stream_result:
                 yield stream_result
 
-        return get_event_loop().run_until_complete(_consume_stream().__anext__())
+        return _utils.get_event_loop().run_until_complete(anext(_consume_stream()))
 
     @overload
     def run_stream_events(
@@ -1344,6 +1342,6 @@ def to_cli_sync(
         agent.to_cli_sync(prog_name='assistant')
         ```
         """
-        return get_event_loop().run_until_complete(
+        return _utils.get_event_loop().run_until_complete(
             self.to_cli(deps=deps, prog_name=prog_name, message_history=message_history)
         )

pydantic_ai_slim/pydantic_ai/result.py

@@ -438,8 +438,7 @@ def stream_output_sync(self, *, debounce_by: float | None = 0.1) -> Iterator[Out
         Returns:
             An iterable of the response data.
         """
-        async_stream = self.stream_output(debounce_by=debounce_by)
-        yield from _blocking_async_iterator(async_stream)
+        return _utils.sync_async_iterator(self.stream_output(debounce_by=debounce_by))
 
     async def stream_text(self, *, delta: bool = False, debounce_by: float | None = 0.1) -> AsyncIterator[str]:
         """Stream the text result as an async iterable.
@@ -485,8 +484,7 @@ def stream_text_sync(self, *, delta: bool = False, debounce_by: float | None = 0
                 Debouncing is particularly important for long structured responses to reduce the overhead of
                 performing validation as each token is received.
         """
-        async_stream = self.stream_text(delta=delta, debounce_by=debounce_by)
-        yield from _blocking_async_iterator(async_stream)
+        return _utils.sync_async_iterator(self.stream_text(delta=delta, debounce_by=debounce_by))
 
     @deprecated('`StreamedRunResult.stream_structured` is deprecated, use `stream_responses` instead.')
     async def stream_structured(
@@ -539,8 +537,7 @@ def stream_responses_sync(
         Returns:
             An iterable of the structured response message and whether that is the last message.
         """
-        async_stream = self.stream_responses(debounce_by=debounce_by)
-        yield from _blocking_async_iterator(async_stream)
+        return _utils.sync_async_iterator(self.stream_responses(debounce_by=debounce_by))
 
     async def get_output(self) -> OutputDataT:
         """Stream the whole response, validate and return it."""
@@ -614,6 +611,18 @@ async def validate_response_output(
         else:
             raise ValueError('No stream response or run result provided')  # pragma: no cover
 
+    def validate_response_output_sync(
+        self, message: _messages.ModelResponse, *, allow_partial: bool = False
+    ) -> OutputDataT:
+        """Validate a structured result message.
+
+        This is a convenience method that wraps [`validate_response_output()`][pydantic_ai.result.StreamedRunResult.validate_response_output] with `loop.run_until_complete(...)`.
+        You therefore can't use this method inside async code or if there's an active event loop.
+        """
+        return _utils.get_event_loop().run_until_complete(
+            self.validate_response_output(message, allow_partial=allow_partial)
+        )
+
     async def _marked_completed(self, message: _messages.ModelResponse | None = None) -> None:
         self.is_complete = True
         if message is not None:
@@ -638,15 +647,6 @@ class FinalResult(Generic[OutputDataT]):
     __repr__ = _utils.dataclasses_no_defaults_repr
 
 
-def _blocking_async_iterator(async_iter: AsyncIterator[T]) -> Iterator[T]:
-    loop = _utils.get_event_loop()
-    while True:
-        try:
-            yield loop.run_until_complete(async_iter.__anext__())
-        except StopAsyncIteration:
-            break
-
-
 def _get_usage_checking_stream_response(
     stream_response: models.StreamedResponse,
     limits: UsageLimits | None,

tests/test_streaming.py

@@ -2085,3 +2085,24 @@ async def ret_a(x: str) -> str:
             AgentRunResultEvent(result=AgentRunResult(output='{"ret_a":"a-apple"}')),
         ]
     )
+
+
+def test_structured_response_sync_validation():
+    async def text_stream(_messages: list[ModelMessage], agent_info: AgentInfo) -> AsyncIterator[DeltaToolCalls]:
+        assert agent_info.output_tools is not None
+        assert len(agent_info.output_tools) == 1
+        name = agent_info.output_tools[0].name
+        json_data = json.dumps({'response': [1, 2, 3, 4]})
+        yield {0: DeltaToolCall(name=name)}
+        yield {0: DeltaToolCall(json_args=json_data[:15])}
+        yield {0: DeltaToolCall(json_args=json_data[15:])}
+
+    agent = Agent(FunctionModel(stream_function=text_stream), output_type=list[int])
+
+    chunks: list[list[int]] = []
+    result = agent.run_stream_sync('')
+    for structured_response, last in result.stream_responses_sync(debounce_by=None):
+        response_data = result.validate_response_output_sync(structured_response, allow_partial=not last)
+        chunks.append(response_data)
+
+    assert chunks == snapshot([[1], [1, 2, 3, 4], [1, 2, 3, 4], [1, 2, 3, 4]])