ractogateway.pipelines.agent

AgentPipeline — autonomous ReAct agent with pluggable tools.

Exports

AgentPipeline

Sync run() + async arun() ReAct loop.

AsyncAgentPipeline

Async-only wrapper suitable for FastAPI / async servers.

AgentResult, AgentStep, AgentUsage, StopReason

Result and step models.

AgentRateLimitExceededError

Raised when the rate limiter blocks a run.

Tool factories (for advanced use)

make_finish_tool, make_rag_tool, make_rag_tool_async, make_sql_tool, make_http_tool, make_memory_tools

Pre-built tool factories used internally; exposed for custom wiring.

ToolExecutor

Low-level sync / async tool runner.

Install

No extra dependencies required for the core agent. HTTP tool requires httpx:

pip install ractogateway[pipelines-agent-http]

class ractogateway.pipelines.agent.AgentPipeline(kit, *, tools=None, rag_pipeline=None, sql_pipeline=None, enable_http=False, agent_memory=None, max_steps=10, max_consecutive_errors=3, tool_retries=0, max_step_extension=0, max_parallel_tools=4, system_prompt=None, extra_rules='', safe_mode=False, tracer=None, metrics=None, rate_limiter=None, user_id='default')

Bases: object

Autonomous ReAct agent with sync run() and async arun().

Parameters:

• kit (Any) – Any RactoGateway developer kit (must support chat() and achat() methods).

• tools (list[Callable[..., Any]] | None) – List of plain callables or @tool-decorated functions to register. Each is registered by its __name__ (or __tool_name__ if set by the @tool decorator).

• rag_pipeline (Any) – Optional RactoRAG instance — auto-registers a rag_search tool.

• sql_pipeline (Any) – Optional SQLAnalystPipeline — auto-registers a sql_query tool.

• enable_http (bool) – When True, registers an http_get tool that fetches URLs via httpx (requires pip install ractogateway[pipelines-agent-http]).

• agent_memory (Any) – Any dict-like or object with get/set methods. Auto-registers memory_read and memory_write tools.

• max_steps (int) – Hard cap on reasoning steps before the loop stops with StopReason.MAX_STEPS.

• max_consecutive_errors (int) – Number of consecutive tool errors that trigger the circuit breaker (StopReason.CIRCUIT_BREAK). Default 3.

• tool_retries (int) – How many times to retry a failing tool before reporting the error to the LLM. Default 0 (no retry).

• max_step_extension (int) – Maximum additional steps the agent may request via request_more_steps. 0 disables the feature.

• max_parallel_tools (int) – Maximum number of tools to run simultaneously when the LLM requests a parallel batch. 1 forces sequential execution. Default 4.

• system_prompt (str | None) – Fully override the auto-generated system prompt (advanced).

• extra_rules (str) – Append an extra numbered rule to the default system prompt.

• safe_mode (bool) – Catch all exceptions and surface them in result.error instead of re-raising.

• tracer (Any) – Optional ractogateway.telemetry.RactoTracer.

• metrics (Any) – Optional ractogateway.telemetry.GatewayMetricsMiddleware.

• rate_limiter (Any) – Duck-typed rate limiter with check_and_consume(user_id, tokens) and get_remaining(user_id) methods.

• user_id (str) – Default user identifier passed to the rate limiter.

run(goal, *, max_steps=<object object>, user_id=<object object>, session_id=None, response_format=None)

Run the agent synchronously until it finishes or hits max_steps.

Parameters:

• goal (str) – The task or question the agent should solve.

• max_steps (Any) – Per-call override for the constructor max_steps.

• user_id (Any) – Per-call override for the rate-limiter user identifier.

• session_id (str | None) – Optional session tag passed to tracer spans (informational).

• response_format (type | None) – A Pydantic BaseModel subclass. When provided, the agent’s final answer is parsed and validated into an instance stored in AgentResult.parsed_output.

Return type:

AgentResult

Returns:

AgentResult – Contains the final answer, all steps, stop reason, token usage, and (when response_format is set) the parsed structured output.

async arun(goal, *, max_steps=<object object>, user_id=<object object>, session_id=None, response_format=None)

Async variant of run().

Uses kit.achat() for LLM calls and runs sync tools in a thread pool so the event loop is never blocked.

Parameters:

• goal (str) – The task or question the agent should solve.

• max_steps (Any) – Per-call override for the constructor max_steps.

• user_id (Any) – Per-call override for the rate-limiter user identifier.

• session_id (str | None) – Optional session tag for tracing.

• response_format (type | None) – A Pydantic BaseModel subclass for structured output parsing.

Return type:

AgentResult

property registered_tools: list[str]

Sorted list of all registered tool names.

property system_prompt: str

The system prompt sent to the LLM each step (read-only).

class ractogateway.pipelines.agent.AsyncAgentPipeline(*args, **kwargs)

Bases: object

Async-only variant of AgentPipeline.

Exposes a single async run() method - suitable for FastAPI endpoints where a sync run() should not be in the public API.

All constructor parameters are identical to AgentPipeline.

async run(goal, **kwargs)

Async-only agent entrypoint. See AgentPipeline.arun().

Return type:

AgentResult

property registered_tools: list[str]

Sorted list of all registered tool names.

property system_prompt: str

The system prompt used by the inner pipeline.

class ractogateway.pipelines.agent.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

class ractogateway.pipelines.agent.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.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.StopReason(*values)

Bases: str, Enum

Why the agent loop terminated.

FINISHED = 'finished'

MAX_STEPS = 'max_steps'

ERROR = 'error'

CIRCUIT_BREAK = 'circuit_break'

exception ractogateway.pipelines.agent.AgentRateLimitExceededError

Bases: RuntimeError

Raised when the rate limiter blocks an agent run.

class ractogateway.pipelines.agent.ToolExecutor(tools, max_retries=0)

Bases: object

Runs registered tools by name with sync and async support.

Parameters:

• tools (dict[str, Callable]) – Mapping of tool name to callable.

• max_retries (int) – How many times to retry a tool that raises an exception before reporting an error observation to the LLM. Default 0 = no retry.

property names: list[str]

Sorted list of registered tool names.

describe_all()

Build the tools section for the agent system prompt.

Return type:

str

execute(tool_name, tool_input)

Execute tool_name synchronously, retrying up to max_retries times on exception.

Return type:

tuple[str, float]

Returns:

tuple[str, float] – (observation, duration_ms)

async aexecute(tool_name, tool_input)

Execute tool_name asynchronously, retrying up to max_retries times on exception.

Async callables are awaited directly; sync callables run in the default thread-pool executor to avoid blocking the event loop.

Return type:

tuple[str, float]

Returns:

tuple[str, float] – (observation, duration_ms)

async aexecute_parallel(calls)

Execute multiple tool calls concurrently via asyncio.gather.

Parameters:

calls (list[tuple[str, dict[str, Any]]]) – List of (tool_name, tool_input) pairs to run in parallel.

Return type:

list[tuple[str, float]]

Returns:

list[tuple[str, float]] – Results in the same order as calls.

ractogateway.pipelines.agent.make_finish_tool()

Return the always-present finish tool.

When the LLM calls finish(answer=...), the agent loop stops and returns the answer as AgentResult.final_answer.

Return type:

tuple[str, Callable]

ractogateway.pipelines.agent.make_http_tool()

Return an http_get tool that fetches URL content via httpx.

Requires httpx: pip install ractogateway[pipelines-agent-http]

Return type:

tuple[str, Callable]

ractogateway.pipelines.agent.make_memory_tools(agent_memory)

Return memory_read and memory_write tools backed by agent_memory.

agent_memory can be any object supporting:

memory.get(key) -> Any
memory.set(key, value) -> None

or a plain dict.

Return type:

list[tuple[str, Callable]]

ractogateway.pipelines.agent.make_rag_tool(rag_pipeline)

Return a rag_search tool backed by a RactoRAG pipeline.

Return type:

tuple[str, Callable]

ractogateway.pipelines.agent.make_rag_tool_async(rag_pipeline)

Return an async rag_search tool backed by an async RactoRAG.

Return type:

tuple[str, Callable]

ractogateway.pipelines.agent.make_sql_tool(sql_pipeline)

Return a sql_query tool backed by a SQLAnalystPipeline.

Return type:

tuple[str, Callable]