ractogateway.pipelines.agent._models

Models for AgentPipeline.

class ractogateway.pipelines.agent._models.StopReason(*values)

Bases: str, Enum

Why the agent loop terminated.

FINISHED = 'finished'

MAX_STEPS = 'max_steps'

ERROR = 'error'

CIRCUIT_BREAK = 'circuit_break'

class ractogateway.pipelines.agent._models.AgentUsage(**data)

Bases: BaseModel

Token and step accounting across the full agent run.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

total_input_tokens: int

Cumulative input tokens across all LLM calls.

total_output_tokens: int

Cumulative output tokens across all LLM calls.

steps_taken: int

Number of reasoning+action steps executed.

tools_called: int

Number of non-finish tool invocations.

property total_tokens: int

Sum of all input and output tokens.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ractogateway.pipelines.agent._models.AgentStep(**data)

Bases: BaseModel

One reasoning + action step in the ReAct loop.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

step_num: int

1-based position in the execution sequence.

thought: str | None

The LLM’s reasoning text before selecting a tool.

tool_name: str

Name of the tool that was invoked.

tool_input: dict[str, Any]

Arguments passed to the tool as a flat dict.

observation: str

The tool’s return value (truncated to 4000 chars when long).

duration_ms: float

Wall-clock time for tool execution in milliseconds.

is_finish: bool

True when this step invoked the finish() termination tool.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ractogateway.pipelines.agent._models.AgentResult(**data)

Bases: BaseModel

Full output of an AgentPipeline run.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

goal: str

The original user goal / task string.

final_answer: str | None

The agent’s answer when finish() was called (None on max_steps/error).

steps: list[AgentStep]

All reasoning+action steps in execution order.

stop_reason: StopReason

Why the loop terminated.

usage: AgentUsage

Token and step counts.

error: str | None

Exception message when stop_reason is ERROR.

parsed_output: Any

Validated Pydantic model instance when response_format was passed to run(). Not included in JSON serialisation; call result.parsed_output.model_dump() manually.

get_tool_calls()

Return (tool_name, tool_input) for every non-finish step.

Return type:

list[tuple[str, dict[str, Any]]]

get_observations()

Return all tool observations in execution order.

Return type:

list[str]

succeeded()

True when the agent reached a final answer.

Return type:

bool

to_json(path=None, *, indent=2)

Serialise to JSON. Returns string when path is None.

Return type:

str | None

to_markdown(path=None)

Build a step-by-step Markdown trace. Returns string when path is None.

Return type:

str | None

exception ractogateway.pipelines.agent._models.AgentRateLimitExceededError

Bases: RuntimeError

Raised when the rate limiter blocks an agent run.