_pretty_print.py

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