Add docs and tests for RunContext.partial_output in output tools

[Danipulok]

Closes #3624

[Danipulok]

When I was working on tests, I discovered that agent.run_stream + result.stream_text(delta=False) (default behavior) does not call TextOutput(my_tool)

It seems this is what was discussed in #3393 (comment), so it does not deserve a new issue?

Here are the tests that demonstrate bugged behavior: link

I think I would like to include the failing streaming test with xfail, but what issue/comment should I link? Or maybe don't commit it at all?

[Danipulok]

About tests: I created a new class designed to test partial_output and separated tests into tests for sync and streaming mode.
The tests were not deleted, just moved and formatted to look alike

Here are the changes:

test_output_validator_partial_sync

before: test_agent.py::test_output_validator_partial_sync

pydantic-ai/tests/test_agent.py

Lines 363 to 377 in ccefddc

	def test_output_validator_partial_sync():
	"""Test that output validators receive correct value for `partial_output` in sync mode."""
	call_log: list[tuple[str, bool]] = []
	agent = Agent[None, str](TestModel(custom_output_text='test output'))
	@agent.output_validator
	def validate_output(ctx: RunContext[None], output: str) -> str:
	call_log.append((output, ctx.partial_output))
	return output
	result = agent.run_sync('Hello')
	assert result.output == 'test output'
	assert call_log == snapshot([('test output', False)])

after: test_agent.py::TestPartialOutput::test_output_validator_text
https://github.com/Danipulok/pydantic-ai/blob/d08191ee0e84ea2ae54fd4c28e8d5076986c281e/tests/test_agent.py#L370-L387

test_output_validator_partial_stream_text
before: test_agent.py::test_output_validator_partial_stream_text

pydantic-ai/tests/test_agent.py

Lines 380 to 409 in ccefddc

	async def test_output_validator_partial_stream_text():
	"""Test that output validators receive correct value for `partial_output` when using stream_text()."""
	call_log: list[tuple[str, bool]] = []
	async def stream_text(messages: list[ModelMessage], info: AgentInfo) -> AsyncIterator[str]:
	for chunk in ['Hello', ' ', 'world', '!']:
	yield chunk
	agent = Agent(FunctionModel(stream_function=stream_text))
	@agent.output_validator
	def validate_output(ctx: RunContext[None], output: str) -> str:
	call_log.append((output, ctx.partial_output))
	return output
	async with agent.run_stream('Hello') as result:
	text_parts = []
	async for chunk in result.stream_text(debounce_by=None):
	text_parts.append(chunk)
	assert text_parts[-1] == 'Hello world!'
	assert call_log == snapshot(
	[
	('Hello', True),
	('Hello ', True),
	('Hello world', True),
	('Hello world!', True),
	('Hello world!', False),
	]
	)

after: test_streaming.py::TestPartialOutput::test_output_validator_text
https://github.com/Danipulok/pydantic-ai/blob/d08191ee0e84ea2ae54fd4c28e8d5076986c281e/tests/test_streaming.py#L762-L789

test_output_validator_partial_stream_output

before: test_agent.py::test_output_validator_partial_stream_output

pydantic-ai/tests/test_agent.py

Lines 412 to 439 in ccefddc

	async def test_output_validator_partial_stream_output():
	"""Test that output validators receive correct value for `partial_output` when using stream_output()."""
	call_log: list[tuple[Foo, bool]] = []
	async def stream_model(messages: list[ModelMessage], info: AgentInfo) -> AsyncIterator[DeltaToolCalls]:
	assert info.output_tools is not None
	yield {0: DeltaToolCall(name=info.output_tools[0].name, json_args='{"a": 42')}
	yield {0: DeltaToolCall(json_args=', "b": "f')}
	yield {0: DeltaToolCall(json_args='oo"}')}
	agent = Agent(FunctionModel(stream_function=stream_model), output_type=Foo)
	@agent.output_validator
	def validate_output(ctx: RunContext[None], output: Foo) -> Foo:
	call_log.append((output, ctx.partial_output))
	return output
	async with agent.run_stream('Hello') as result:
	outputs = [output async for output in result.stream_output(debounce_by=None)]
	assert outputs[-1] == Foo(a=42, b='foo')
	assert call_log == snapshot(
	[
	(Foo(a=42, b='f'), True),
	(Foo(a=42, b='foo'), True),
	(Foo(a=42, b='foo'), False),
	]
	)

after: test_streaming.py::TestPartialOutput::test_output_validator_structured
https://github.com/Danipulok/pydantic-ai/blob/d08191ee0e84ea2ae54fd4c28e8d5076986c281e/tests/test_streaming.py#L791-L818

[DouweM]

Let's rename this Handling partial output and also drop in output validators from the other section as it's already under "Output validators"

[DouweM]

I'd prefer for this to be top-level text rather than the entire docs section to be an indented warning block

[DouweM]

I think we should say this in the "Handling partial output in output validators" section as well. Maybe the 2 sections can be merged as output validators and functions are very similar?

[DouweM]

As I mentioned on the other PR, it's just specific to sync mode or run_sync: run (async), iter and run_stream_events have the same behavior. It's really just "run_stream vs the rest"

[DouweM]

I'd rather this be a paragraph than a list btw

[DouweM]

We don't need this example as it's the "base case" already covered above

[DouweM]

This is actually not a great example because both of these fields are required, meaning the partial data wouldn't validate anyway until the output is complete, so there would not be any partial output in this case.

[DouweM]

We're not using run_stream here 😄

This example needs some work to actually show the problem + use case

[DouweM]

No "fake" subheadings please, I'd rather have paragraphs

[DouweM]

This is duplicative with For output functions with **side effects** (e.g., sending notifications, logging, database updates), you should check the [RunContext.partial_output][pydantic_ai.tools.RunContext.partial_output] flag to avoid executing side effects on partial data. above. I'd rather have one authoritative paragraph

[DouweM]

Let's add that test_output_function_text test with TextOutput(func), create a new issue for that bug, and then skip the test