Pretty print result classes

Makefile

@@ -18,6 +18,14 @@ mypy:
 tests: 
 	uv run pytest 
 
+.PHONY: snapshots-fix
+snapshots-fix: 
+	uv run pytest --inline-snapshot=fix 
+
+.PHONY: snapshots-create 
+snapshots-create: 
+	uv run pytest --inline-snapshot=create 
+
 .PHONY: old_version_tests
 old_version_tests: 
 	UV_PROJECT_ENVIRONMENT=.venv_39 uv run --python 3.9 -m pytest

pyproject.toml

@@ -47,6 +47,7 @@ dev = [
     "mkdocstrings[python]>=0.28.0",
     "coverage>=7.6.12",
     "playwright==1.50.0",
+    "inline-snapshot>=0.20.7",
 ]
 [tool.uv.workspace]
 members = ["agents"]
@@ -116,4 +117,7 @@ filterwarnings = [
 ]
 markers = [
     "allow_call_model_methods: mark test as allowing calls to real model implementations",
-]
+]
+
+[tool.inline-snapshot]
+format-command="ruff format --stdin-filename {filename}"

src/agents/result.py

@@ -17,6 +17,7 @@
 from .logger import logger
 from .stream_events import StreamEvent
 from .tracing import Trace
+from .util._pretty_print import pretty_print_result, pretty_print_run_result_streaming
 
 if TYPE_CHECKING:
     from ._run_impl import QueueCompleteSentinel
@@ -89,6 +90,9 @@ def last_agent(self) -> Agent[Any]:
         """The last agent that was run."""
         return self._last_agent
 
+    def __str__(self) -> str:
+        return pretty_print_result(self)
+
 
 @dataclass
 class RunResultStreaming(RunResultBase):
@@ -216,3 +220,6 @@ def _cleanup_tasks(self):
 
         if self._output_guardrails_task and not self._output_guardrails_task.done():
             self._output_guardrails_task.cancel()
+
+    def __str__(self) -> str:
+        return pretty_print_run_result_streaming(self)

src/agents/util/_pretty_print.py

@@ -0,0 +1,56 @@
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel
+
+if TYPE_CHECKING:
+    from ..result import RunResult, RunResultBase, RunResultStreaming
+
+
+def _indent(text: str, indent_level: int) -> str:
+    indent_string = "  " * indent_level
+    return "\n".join(f"{indent_string}{line}" for line in text.splitlines())
+
+
+def _final_output_str(result: "RunResultBase") -> str:
+    if result.final_output is None:
+        return "None"
+    elif isinstance(result.final_output, str):
+        return result.final_output
+    elif isinstance(result.final_output, BaseModel):
+        return result.final_output.model_dump_json(indent=2)
+    else:
+        return str(result.final_output)
+
+
+def pretty_print_result(result: "RunResult") -> str:
+    output = "RunResult:"
+    output += f'\n- Last agent: Agent(name="{result.last_agent.name}", ...)'
+    output += (
+        f"\n- Final output ({type(result.final_output).__name__}):\n"
+        f"{_indent(_final_output_str(result), 2)}"
+    )
+    output += f"\n- {len(result.new_items)} new item(s)"
+    output += f"\n- {len(result.raw_responses)} raw response(s)"
+    output += f"\n- {len(result.input_guardrail_results)} input guardrail result(s)"
+    output += f"\n- {len(result.output_guardrail_results)} output guardrail result(s)"
+    output += "\n(See `RunResult` for more details)"
+
+    return output
+
+
+def pretty_print_run_result_streaming(result: "RunResultStreaming") -> str:
+    output = "RunResultStreaming:"
+    output += f'\n- Current agent: Agent(name="{result.current_agent.name}", ...)'
+    output += f"\n- Current turn: {result.current_turn}"
+    output += f"\n- Max turns: {result.max_turns}"
+    output += f"\n- Is complete: {result.is_complete}"
+    output += (
+        f"\n- Final output ({type(result.final_output).__name__}):\n"
+        f"{_indent(_final_output_str(result), 2)}"
+    )
+    output += f"\n- {len(result.new_items)} new item(s)"
+    output += f"\n- {len(result.raw_responses)} raw response(s)"
+    output += f"\n- {len(result.input_guardrail_results)} input guardrail result(s)"
+    output += f"\n- {len(result.output_guardrail_results)} output guardrail result(s)"
+    output += "\n(See `RunResultStreaming` for more details)"
+    return output

tests/README.md

@@ -0,0 +1,25 @@
+# Tests
+
+Before running any tests, make sure you have `uv` installed (and ideally run `make sync` after).
+
+## Running tests
+
+```
+make tests
+```
+
+## Snapshots
+
+We use [inline-snapshots](https://15r10nk.github.io/inline-snapshot/latest/) for some tests. If your code adds new snapshot tests or breaks existing ones, you can fix/create them. After fixing/creating snapshots, run `make tests` again to verify the tests pass.
+
+### Fixing snapshots
+
+```
+make snapshots-fix
+```
+
+### Creating snapshots
+
+```
+make snapshots-update
+```

tests/test_pretty_print.py

@@ -0,0 +1,201 @@
+import json
+
+import pytest
+from inline_snapshot import snapshot
+from pydantic import BaseModel
+
+from agents import Agent, Runner
+from agents.agent_output import _WRAPPER_DICT_KEY
+from agents.util._pretty_print import pretty_print_result, pretty_print_run_result_streaming
+from tests.fake_model import FakeModel
+
+from .test_responses import get_final_output_message, get_text_message
+
+
+@pytest.mark.asyncio
+async def test_pretty_result():
+    model = FakeModel()
+    model.set_next_output([get_text_message("Hi there")])
+
+    agent = Agent(name="test_agent", model=model)
+    result = await Runner.run(agent, input="Hello")
+
+    assert pretty_print_result(result) == snapshot("""\
+RunResult:
+- Last agent: Agent(name="test_agent", ...)
+- Final output (str):
+    Hi there
+- 1 new item(s)
+- 1 raw response(s)
+- 0 input guardrail result(s)
+- 0 output guardrail result(s)
+(See `RunResult` for more details)\
+""")
+
+
+@pytest.mark.asyncio
+async def test_pretty_run_result_streaming():
+    model = FakeModel()
+    model.set_next_output([get_text_message("Hi there")])
+
+    agent = Agent(name="test_agent", model=model)
+    result = Runner.run_streamed(agent, input="Hello")
+    async for _ in result.stream_events():
+        pass
+
+    assert pretty_print_run_result_streaming(result) == snapshot("""\
+RunResultStreaming:
+- Current agent: Agent(name="test_agent", ...)
+- Current turn: 1
+- Max turns: 10
+- Is complete: True
+- Final output (str):
+    Hi there
+- 1 new item(s)
+- 1 raw response(s)
+- 0 input guardrail result(s)
+- 0 output guardrail result(s)
+(See `RunResultStreaming` for more details)\
+""")
+
+
+class Foo(BaseModel):
+    bar: str
+
+
+@pytest.mark.asyncio
+async def test_pretty_run_result_structured_output():
+    model = FakeModel()
+    model.set_next_output(
+        [
+            get_text_message("Test"),
+            get_final_output_message(Foo(bar="Hi there").model_dump_json()),
+        ]
+    )
+
+    agent = Agent(name="test_agent", model=model, output_type=Foo)
+    result = await Runner.run(agent, input="Hello")
+
+    assert pretty_print_result(result) == snapshot("""\
+RunResult:
+- Last agent: Agent(name="test_agent", ...)
+- Final output (Foo):
+    {
+      "bar": "Hi there"
+    }
+- 2 new item(s)
+- 1 raw response(s)
+- 0 input guardrail result(s)
+- 0 output guardrail result(s)
+(See `RunResult` for more details)\
+""")
+
+
+@pytest.mark.asyncio
+async def test_pretty_run_result_streaming_structured_output():
+    model = FakeModel()
+    model.set_next_output(
+        [
+            get_text_message("Test"),
+            get_final_output_message(Foo(bar="Hi there").model_dump_json()),
+        ]
+    )
+
+    agent = Agent(name="test_agent", model=model, output_type=Foo)
+    result = Runner.run_streamed(agent, input="Hello")
+
+    async for _ in result.stream_events():
+        pass
+
+    assert pretty_print_run_result_streaming(result) == snapshot("""\
+RunResultStreaming:
+- Current agent: Agent(name="test_agent", ...)
+- Current turn: 1
+- Max turns: 10
+- Is complete: True
+- Final output (Foo):
+    {
+      "bar": "Hi there"
+    }
+- 2 new item(s)
+- 1 raw response(s)
+- 0 input guardrail result(s)
+- 0 output guardrail result(s)
+(See `RunResultStreaming` for more details)\
+""")
+
+
+@pytest.mark.asyncio
+async def test_pretty_run_result_list_structured_output():
+    model = FakeModel()
+    model.set_next_output(
+        [
+            get_text_message("Test"),
+            get_final_output_message(
+                json.dumps(
+                    {
+                        _WRAPPER_DICT_KEY: [
+                            Foo(bar="Hi there").model_dump(),
+                            Foo(bar="Hi there 2").model_dump(),
+                        ]
+                    }
+                )
+            ),
+        ]
+    )
+
+    agent = Agent(name="test_agent", model=model, output_type=list[Foo])
+    result = await Runner.run(agent, input="Hello")
+
+    assert pretty_print_result(result) == snapshot("""\
+RunResult:
+- Last agent: Agent(name="test_agent", ...)
+- Final output (list):
+    [Foo(bar='Hi there'), Foo(bar='Hi there 2')]
+- 2 new item(s)
+- 1 raw response(s)
+- 0 input guardrail result(s)
+- 0 output guardrail result(s)
+(See `RunResult` for more details)\
+""")
+
+
+@pytest.mark.asyncio
+async def test_pretty_run_result_streaming_list_structured_output():
+    model = FakeModel()
+    model.set_next_output(
+        [
+            get_text_message("Test"),
+            get_final_output_message(
+                json.dumps(
+                    {
+                        _WRAPPER_DICT_KEY: [
+                            Foo(bar="Test").model_dump(),
+                            Foo(bar="Test 2").model_dump(),
+                        ]
+                    }
+                )
+            ),
+        ]
+    )
+
+    agent = Agent(name="test_agent", model=model, output_type=list[Foo])
+    result = Runner.run_streamed(agent, input="Hello")
+
+    async for _ in result.stream_events():
+        pass
+
+    assert pretty_print_run_result_streaming(result) == snapshot("""\
+RunResultStreaming:
+- Current agent: Agent(name="test_agent", ...)
+- Current turn: 1
+- Max turns: 10
+- Is complete: True
+- Final output (list):
+    [Foo(bar='Test'), Foo(bar='Test 2')]
+- 2 new item(s)
+- 1 raw response(s)
+- 0 input guardrail result(s)
+- 0 output guardrail result(s)
+(See `RunResultStreaming` for more details)\
+""")