Merge remote-tracking branch 'pydantic/main' into qian/dbos-agent

Author: qianl15

docs/cli.md

@@ -108,3 +108,25 @@ async def main():
 ```
 
 _(You'll need to add `asyncio.run(main())` to run `main`)_
+
+### Message History
+
+Both `Agent.to_cli()` and `Agent.to_cli_sync()` support a `message_history` parameter, allowing you to continue an existing conversation or provide conversation context:
+
+```python {title="agent_with_history.py" test="skip"}
+from pydantic_ai import Agent
+from pydantic_ai.messages import ModelMessage, ModelRequest, ModelResponse, UserPromptPart, TextPart
+
+agent = Agent('openai:gpt-4.1')
+
+# Create some conversation history
+message_history: list[ModelMessage] = [
+    ModelRequest([UserPromptPart(content='What is 2+2?')]),
+    ModelResponse([TextPart(content='2+2 equals 4.')])
+]
+
+# Start CLI with existing conversation context
+agent.to_cli_sync(message_history=message_history)
+```
+
+The CLI will start with the provided conversation history, allowing the agent to refer back to previous exchanges and maintain context throughout the session.

docs/message-history.md

@@ -18,9 +18,9 @@ and [`StreamedRunResult`][pydantic_ai.result.StreamedRunResult] (returned by [`A
 
     E.g. you've awaited one of the following coroutines:
 
-    * [`StreamedRunResult.stream()`][pydantic_ai.result.StreamedRunResult.stream]
+    * [`StreamedRunResult.stream_output()`][pydantic_ai.result.StreamedRunResult.stream_output]
     * [`StreamedRunResult.stream_text()`][pydantic_ai.result.StreamedRunResult.stream_text]
-    * [`StreamedRunResult.stream_structured()`][pydantic_ai.result.StreamedRunResult.stream_structured]
+    * [`StreamedRunResult.stream_responses()`][pydantic_ai.result.StreamedRunResult.stream_responses]
     * [`StreamedRunResult.get_output()`][pydantic_ai.result.StreamedRunResult.get_output]
 
     **Note:** The final result message will NOT be added to result messages if you use [`.stream_text(delta=True)`][pydantic_ai.result.StreamedRunResult.stream_text] since in this case the result content is never built as one string.

docs/output.md

@@ -573,7 +573,7 @@ agent = Agent(
 async def main():
     user_input = 'My name is Ben, I was born on January 28th 1990, I like the chain the dog and the pyramid.'
     async with agent.run_stream(user_input) as result:
-        async for profile in result.stream():
+        async for profile in result.stream_output():
             print(profile)
             #> {'name': 'Ben'}
             #> {'name': 'Ben'}
@@ -609,9 +609,9 @@ agent = Agent('openai:gpt-4o', output_type=UserProfile)
 async def main():
     user_input = 'My name is Ben, I was born on January 28th 1990, I like the chain the dog and the pyramid.'
     async with agent.run_stream(user_input) as result:
-        async for message, last in result.stream_structured(debounce_by=0.01):  # (1)!
+        async for message, last in result.stream_responses(debounce_by=0.01):  # (1)!
             try:
-                profile = await result.validate_structured_output(  # (2)!
+                profile = await result.validate_response_output(  # (2)!
                     message,
                     allow_partial=not last,
                 )
@@ -627,8 +627,8 @@ async def main():
             #> {'name': 'Ben', 'dob': date(1990, 1, 28), 'bio': 'Likes the chain the dog and the pyramid'}
 ```
 
-1. [`stream_structured`][pydantic_ai.result.StreamedRunResult.stream_structured] streams the data as [`ModelResponse`][pydantic_ai.messages.ModelResponse] objects, thus iteration can't fail with a `ValidationError`.
-2. [`validate_structured_output`][pydantic_ai.result.StreamedRunResult.validate_structured_output] validates the data, `allow_partial=True` enables pydantic's [`experimental_allow_partial` flag on `TypeAdapter`][pydantic.type_adapter.TypeAdapter.validate_json].
+1. [`stream_responses`][pydantic_ai.result.StreamedRunResult.stream_responses] streams the data as [`ModelResponse`][pydantic_ai.messages.ModelResponse] objects, thus iteration can't fail with a `ValidationError`.
+2. [`validate_response_output`][pydantic_ai.result.StreamedRunResult.validate_response_output] validates the data, `allow_partial=True` enables pydantic's [`experimental_allow_partial` flag on `TypeAdapter`][pydantic.type_adapter.TypeAdapter.validate_json].
 
 _(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_
 

examples/pydantic_ai_examples/chat_app.py

@@ -127,7 +127,7 @@ async def stream_messages():
         messages = await database.get_messages()
         # run the agent with the user prompt and the chat history
         async with agent.run_stream(prompt, message_history=messages) as result:
-            async for text in result.stream(debounce_by=0.01):
+            async for text in result.stream_output(debounce_by=0.01):
                 # text here is a `str` and the frontend wants
                 # JSON encoded ModelResponse, so we create one
                 m = ModelResponse(parts=[TextPart(text)], timestamp=result.timestamp())

examples/pydantic_ai_examples/stream_markdown.py

@@ -42,7 +42,7 @@ async def main():
             console.log(f'Using model: {model}')
             with Live('', console=console, vertical_overflow='visible') as live:
                 async with agent.run_stream(prompt, model=model) as result:
-                    async for message in result.stream():
+                    async for message in result.stream_output():
                         live.update(Markdown(message))
             console.log(result.usage())
         else:

examples/pydantic_ai_examples/stream_whales.py

@@ -51,7 +51,7 @@ async def main():
         ) as result:
             console.print('Response:', style='green')
 
-            async for whales in result.stream(debounce_by=0.01):
+            async for whales in result.stream_output(debounce_by=0.01):
                 table = Table(
                     title='Species of Whale',
                     caption='Streaming Structured responses from GPT-4',

pydantic_ai_slim/pydantic_ai/agent/__init__.py

@@ -678,22 +678,28 @@ def _run_span_end_attributes(
         self, state: _agent_graph.GraphAgentState, usage: _usage.RunUsage, settings: InstrumentationSettings
     ):
         if settings.version == 1:
-            attr_name = 'all_messages_events'
-            value = [
-                InstrumentedModel.event_to_dict(e) for e in settings.messages_to_otel_events(state.message_history)
-            ]
+            attrs = {
+                'all_messages_events': json.dumps(
+                    [
+                        InstrumentedModel.event_to_dict(e)
+                        for e in settings.messages_to_otel_events(state.message_history)
+                    ]
+                )
+            }
         else:
-            attr_name = 'pydantic_ai.all_messages'
-            value = settings.messages_to_otel_messages(state.message_history)
+            attrs = {
+                'pydantic_ai.all_messages': json.dumps(settings.messages_to_otel_messages(state.message_history)),
+                **settings.system_instructions_attributes(self._instructions),
+            }
 
         return {
             **usage.opentelemetry_attributes(),
-            attr_name: json.dumps(value),
+            **attrs,
             'logfire.json_schema': json.dumps(
                 {
                     'type': 'object',
                     'properties': {
-                        attr_name: {'type': 'array'},
+                        **{attr: {'type': 'array'} for attr in attrs.keys()},
                         'final_result': {'type': 'object'},
                     },
                 }

pydantic_ai_slim/pydantic_ai/durable_exec/temporal/__init__.py

@@ -1,15 +1,24 @@
 from __future__ import annotations
 
 import warnings
-from collections.abc import Sequence
+from collections.abc import AsyncIterator, Sequence
+from contextlib import AbstractAsyncContextManager
 from dataclasses import replace
 from typing import Any, Callable
 
 from pydantic.errors import PydanticUserError
-from temporalio.client import ClientConfig, Plugin as ClientPlugin
+from temporalio.client import ClientConfig, Plugin as ClientPlugin, WorkflowHistory
 from temporalio.contrib.pydantic import PydanticPayloadConverter, pydantic_data_converter
-from temporalio.converter import DefaultPayloadConverter
-from temporalio.worker import Plugin as WorkerPlugin, WorkerConfig
+from temporalio.converter import DataConverter, DefaultPayloadConverter
+from temporalio.service import ConnectConfig, ServiceClient
+from temporalio.worker import (
+    Plugin as WorkerPlugin,
+    Replayer,
+    ReplayerConfig,
+    Worker,
+    WorkerConfig,
+    WorkflowReplayResult,
+)
 from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner
 
 from ...exceptions import UserError
@@ -31,17 +40,15 @@
 class PydanticAIPlugin(ClientPlugin, WorkerPlugin):
     """Temporal client and worker plugin for Pydantic AI."""
 
-    def configure_client(self, config: ClientConfig) -> ClientConfig:
-        if (data_converter := config.get('data_converter')) and data_converter.payload_converter_class not in (
-            DefaultPayloadConverter,
-            PydanticPayloadConverter,
-        ):
-            warnings.warn(  # pragma: no cover
-                'A non-default Temporal data converter was used which has been replaced with the Pydantic data converter.'
-            )
+    def init_client_plugin(self, next: ClientPlugin) -> None:
+        self.next_client_plugin = next
 
-        config['data_converter'] = pydantic_data_converter
-        return super().configure_client(config)
+    def init_worker_plugin(self, next: WorkerPlugin) -> None:
+        self.next_worker_plugin = next
+
+    def configure_client(self, config: ClientConfig) -> ClientConfig:
+        config['data_converter'] = self._get_new_data_converter(config.get('data_converter'))
+        return self.next_client_plugin.configure_client(config)
 
     def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
         runner = config.get('workflow_runner')  # pyright: ignore[reportUnknownMemberType]
@@ -67,7 +74,35 @@ def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
             PydanticUserError,
         ]
 
-        return super().configure_worker(config)
+        return self.next_worker_plugin.configure_worker(config)
+
+    async def connect_service_client(self, config: ConnectConfig) -> ServiceClient:
+        return await self.next_client_plugin.connect_service_client(config)
+
+    async def run_worker(self, worker: Worker) -> None:
+        await self.next_worker_plugin.run_worker(worker)
+
+    def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:  # pragma: no cover
+        config['data_converter'] = self._get_new_data_converter(config.get('data_converter'))  # pyright: ignore[reportUnknownMemberType]
+        return self.next_worker_plugin.configure_replayer(config)
+
+    def run_replayer(
+        self,
+        replayer: Replayer,
+        histories: AsyncIterator[WorkflowHistory],
+    ) -> AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]:  # pragma: no cover
+        return self.next_worker_plugin.run_replayer(replayer, histories)
+
+    def _get_new_data_converter(self, converter: DataConverter | None) -> DataConverter:
+        if converter and converter.payload_converter_class not in (
+            DefaultPayloadConverter,
+            PydanticPayloadConverter,
+        ):
+            warnings.warn(  # pragma: no cover
+                'A non-default Temporal data converter was used which has been replaced with the Pydantic data converter.'
+            )
+
+        return pydantic_data_converter
 
 
 class AgentPlugin(WorkerPlugin):
@@ -76,8 +111,24 @@ class AgentPlugin(WorkerPlugin):
     def __init__(self, agent: TemporalAgent[Any, Any]):
         self.agent = agent
 
+    def init_worker_plugin(self, next: WorkerPlugin) -> None:
+        self.next_worker_plugin = next
+
     def configure_worker(self, config: WorkerConfig) -> WorkerConfig:
         activities: Sequence[Callable[..., Any]] = config.get('activities', [])  # pyright: ignore[reportUnknownMemberType]
         # Activities are checked for name conflicts by Temporal.
         config['activities'] = [*activities, *self.agent.temporal_activities]
-        return super().configure_worker(config)
+        return self.next_worker_plugin.configure_worker(config)
+
+    async def run_worker(self, worker: Worker) -> None:
+        await self.next_worker_plugin.run_worker(worker)
+
+    def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig:  # pragma: no cover
+        return self.next_worker_plugin.configure_replayer(config)
+
+    def run_replayer(
+        self,
+        replayer: Replayer,
+        histories: AsyncIterator[WorkflowHistory],
+    ) -> AbstractAsyncContextManager[AsyncIterator[WorkflowReplayResult]]:  # pragma: no cover
+        return self.next_worker_plugin.run_replayer(replayer, histories)

pydantic_ai_slim/pydantic_ai/durable_exec/temporal/_logfire.py

@@ -25,10 +25,13 @@ def __init__(self, setup_logfire: Callable[[], Logfire] = _default_setup_logfire
         self.setup_logfire = setup_logfire
         self.metrics = metrics
 
+    def init_client_plugin(self, next: ClientPlugin) -> None:
+        self.next_client_plugin = next
+
     def configure_client(self, config: ClientConfig) -> ClientConfig:
         interceptors = config.get('interceptors', [])
         config['interceptors'] = [*interceptors, TracingInterceptor(get_tracer('temporalio'))]
-        return super().configure_client(config)
+        return self.next_client_plugin.configure_client(config)
 
     async def connect_service_client(self, config: ConnectConfig) -> ServiceClient:
         logfire = self.setup_logfire()
@@ -45,4 +48,4 @@ async def connect_service_client(self, config: ConnectConfig) -> ServiceClient:
                     telemetry=TelemetryConfig(metrics=OpenTelemetryConfig(url=metrics_url, headers=headers))
                 )
 
-        return await super().connect_service_client(config)
+        return await self.next_client_plugin.connect_service_client(config)

pydantic_ai_slim/pydantic_ai/models/instrumented.py

@@ -236,27 +236,36 @@ def handle_messages(self, input_messages: list[ModelMessage], response: ModelRes
             if response.provider_details and 'finish_reason' in response.provider_details:
                 output_message['finish_reason'] = response.provider_details['finish_reason']
             instructions = InstrumentedModel._get_instructions(input_messages)  # pyright: ignore [reportPrivateUsage]
+            system_instructions_attributes = self.system_instructions_attributes(instructions)
             attributes = {
                 'gen_ai.input.messages': json.dumps(self.messages_to_otel_messages(input_messages)),
                 'gen_ai.output.messages': json.dumps([output_message]),
+                **system_instructions_attributes,
                 'logfire.json_schema': json.dumps(
                     {
                         'type': 'object',
                         'properties': {
                             'gen_ai.input.messages': {'type': 'array'},
                             'gen_ai.output.messages': {'type': 'array'},
-                            **({'gen_ai.system_instructions': {'type': 'array'}} if instructions else {}),
+                            **(
+                                {'gen_ai.system_instructions': {'type': 'array'}}
+                                if system_instructions_attributes
+                                else {}
+                            ),
                             'model_request_parameters': {'type': 'object'},
                         },
                     }
                 ),
             }
-            if instructions is not None:
-                attributes['gen_ai.system_instructions'] = json.dumps(
-                    [_otel_messages.TextPart(type='text', content=instructions)]
-                )
             span.set_attributes(attributes)
 
+    def system_instructions_attributes(self, instructions: str | None) -> dict[str, str]:
+        if instructions and self.include_content:
+            return {
+                'gen_ai.system_instructions': json.dumps([_otel_messages.TextPart(type='text', content=instructions)]),
+            }
+        return {}
+
     def _emit_events(self, span: Span, events: list[Event]) -> None:
         if self.event_mode == 'logs':
             for event in events: