Struct ReActAgent 

pub struct ReActAgent {
    pub state: SharedState,
    /* private fields */
}

The classic ReAct reasoning loop: conversation → tool execution fed back → until the model answers directly.

Assembly: three required parameters (Provider / ToolRegistry / system_prompt) + chained optional (with_memory / with_config / with_state / with_event_channel / with_tool_round_executor); for simpler assembly use the macro react_agent!.

Examples

use molo::agent::{Agent, ReActAgent};
use molo::provider::{FakeProvider, FakeReply};
use molo::tool::ToolRegistry;

let mut agent = ReActAgent::new(
    FakeProvider::new([FakeReply::Text("Hello".into())]),
    ToolRegistry::new(),
    "You are a helpful assistant",
);
let answer = agent.run("Are you there").await?;
assert_eq!(answer, "Hello");

Cancellation semantics

ReActAgent observes cancellation and deadlines carried by RunContext. Cancellation is cooperative and only checked at safe points:

• Cancelled mid-conversation: the in-flight request is dropped, and nothing is recorded for this round;
• Cancelled during a tool round: the running tool call is not interrupted (tools are user code; a partial execution risks side effects); the Assistant message already recorded this round and the ToolResult fed back right after remain complete;
• Streaming path: already-dispatched Delta chunks are kept, then the stream concludes with a MessageChunk::Cancelled terminal chunk — no Done;
• After cancellation, already-recorded messages are kept, not rolled back; subsequent runs continue from the retained memory.

Callers that don’t need cancellation can just use run / run_stream — they internally use a token that is never cancelled.

Errors

Tool execution failures are not AgentError — the error text is fed back to the model via ToolResult, the model decides what to do next, and the loop continues. AgentError only represents run-level failures (Memory / Provider failures, exceeding the tool-round limit, cancellation).

Structured output

The ability to require the model’s answer to conform to a given JSON Schema; both entry points share the same validation-and-retry loop (see run_typed and with_structured_output):

• run_typed: typed output — run_typed’s return type directly declares the target type; this run auto-generates a JSON Schema from the type (schemars-derived, the same pipeline as tool parameter schemas), and deserializes after validation;
• with_structured_output: a hand-written Schema (or the serialized result of schemars::schema_for!(T)), paired with Agent::run returning JSON text that you parse yourself.

Both entry points validate the final answer with framework-side jsonschema; on failure the error is fed back to the model for retry (budget AgentConfig::max_structured_retries); compatible endpoints are additionally constrained as best effort via response_format (see ModelOptions::structured).

Fields

state: SharedState

Shared state: the application reads and writes this field directly across runs; it is injected via Tool::call on every tool call, so multiple tools / Agents can share one instance.

Implementations

impl ReActAgent

pub fn builder(provider: impl Provider + 'static) -> ReActAgentBuilder

Starts a ReActAgentBuilder with a provider.

pub fn new( provider: impl Provider + 'static, tools: ToolRegistry, system_prompt: impl Into<String>, ) -> Self

Simple construction: three required parameters — provider (Provider) + tools (ToolRegistry) + system_prompt (empty string = no system prompt); run returns the model’s answer text (consistent with Agent::run).

Memory defaults to a bounded window (WindowMemory, 128k token budget, long conversations auto-trim the oldest rounds); when you need something custom (full history / persistence, etc.), chain with_memory to replace it; other optional behaviors (such as the max tool rounds) go through with_config.

Structured output: typed output goes straight to run_typed (this run generates the Schema from the target type, no constructor configuration needed); for hand-written Schemas use with_structured_output.

For simpler assembly see the macro react_agent!.

pub fn kernel(tools: ToolRegistry, system_prompt: impl Into<String>) -> Self

Constructs a provider-free ReAct kernel for use with an outer HarnessRuntime.

The runtime owns provider execution; this kernel only assembles model requests, runs immediate tools, requests governed effects, and consumes observations.

pub fn with_memory(self, memory: impl Memory + 'static) -> Self

Replace the default Memory (default WindowMemory, 128k token budget, long conversations auto-trim the oldest rounds).

The Agent owns the Memory: Memory::record needs &mut self, which a shared (Arc) form can’t write through; sharing the same conversation history across Agents is a Workflow-orchestration concern, out of this method’s scope.

Examples

Replace it with InMemoryMemory, which keeps all history verbatim:

use molo::agent::{Agent, ReActAgent};
use molo::memory::InMemoryMemory;
use molo::provider::{FakeProvider, FakeReply};
use molo::tool::ToolRegistry;

let mut agent = ReActAgent::new(
    FakeProvider::new([FakeReply::Text("Hello".into())]),
    ToolRegistry::new(),
    "",
)
.with_memory(InMemoryMemory::default());

assert_eq!(agent.run("Are you there").await.unwrap(), "Hello");

pub fn with_config(self, config: AgentConfig) -> Self

Replace the default config (default AgentConfig::default): the max_tool_rounds round limit and the options model parameters (temperature / max_tokens / extra parameters).

See AgentConfig for how to write a config.

pub fn with_structured_output(self, schema: Value) -> Self

Enable structured output: the final answer must be JSON conforming to this JSON Schema document.

schema is usually the serialized RootSchema produced by schemars::schema_for!(T) (the same pipeline as tool parameter schemas); it can also be hand-written. Semantics:

• Compatible endpoints constrain the model as best effort via response_format once they receive it;
• Framework-side fallback validation: when the final answer doesn’t conform, the validation error is fed back to the model for retry (independent budget AgentConfig::max_structured_retries; exceeding it fails);
• Once validation passes, the JSON text is returned as-is as the answer.

Typed output goes through run_typed: the schema is auto-generated from the target type and deserialized, ignoring the hand-written schema set here — the two paths don’t combine (the hand-written schema serves Agent::run’s text form; run_typed is always “the type is the schema”).

Examples

use molo::{react_agent, Agent, FakeProvider, FakeReply};
use serde_json::json;

let mut agent = react_agent!(
    FakeProvider::new([FakeReply::Text(r#"{"city":"Beijing"}"#.into())]),
    "You are a structured-output assistant",
)
.with_structured_output(json!({
    "type": "object",
    "properties": { "city": { "type": "string" } },
    "required": ["city"],
}));

let answer = agent.run("How's the weather in Beijing").await?;
assert_eq!(answer, r#"{"city":"Beijing"}"#);

pub fn with_state(self, state: SharedState) -> Self

Attach shared state (empty by default). The application reads and writes via state across runs, and tools access the same instance through the state parameter at call time.

pub fn with_event_channel(self, channel: impl EventChannel + 'static) -> Self

Attach an observation channel (not attached by default, zero cost): the loop pushes process events into the pipeline, and the host side subscribes.

The pipeline is long-lived: attached once, events from many runs flow into the same pipeline (the Agent is the event source); each run is one segment of the pipeline, delimited by ReActEvent::RunStarted / ReActEvent::RunEnded; both the streaming and non-streaming paths publish. See ReActEvent for the event set.

pub fn with_tool_round_executor( self, executor: impl ToolRoundExecutor + 'static, ) -> Self

Replace the default tool-round executor (default: SerialToolRoundExecutor — the round’s calls run one by one, in request order) with a custom round policy: order / concurrency / approval gates.

The executor decides how the round’s tool calls are executed; the per-call mechanics (tool span, ToolStarted / ToolCompleted events, registry dispatch) stay framework-owned via ToolRoundCtx::run. Each outcome is recorded (and, on the streaming path, dispatched) by the agent as the executor’s stream yields it. The round stays atomic by contract — the executor owns that responsibility.

Examples

A round policy that runs the round’s calls concurrently; outcomes flow back as each completes:

use molo::agent::{
    Agent, ReActAgent, ToolCallOutcome, ToolRoundCtx, ToolRoundExecutor,
};
use molo::message::ToolCall;
use molo::provider::{FakeProvider, FakeReply};
use molo::tool::ToolRegistry;
use futures::stream::{BoxStream, FuturesUnordered};
use futures::StreamExt;

#[derive(Default)]
struct ParallelToolRoundExecutor;

#[molo::async_trait]
impl ToolRoundExecutor for ParallelToolRoundExecutor {
    async fn execute_round<'a>(
        &'a mut self,
        ctx: ToolRoundCtx<'a>,
        calls: Vec<ToolCall>,
    ) -> BoxStream<'a, ToolCallOutcome> {
        // The run futures borrow ctx, so they materialize inside the
        // generator (which owns ctx), not across the return.
        Box::pin(async_stream::stream! {
            let mut tasks = FuturesUnordered::new();
            for call in calls {
                tasks.push(ctx.run(call));
            }
            while let Some(outcome) = tasks.next().await {
                yield outcome;
            }
        })
    }
}

let mut agent = ReActAgent::new(
    FakeProvider::new([FakeReply::Text("Hello".into())]),
    ToolRegistry::new(),
    "",
)
.with_tool_round_executor(ParallelToolRoundExecutor);

assert_eq!(agent.run("Are you there").await?, "Hello");

impl ReActAgent

pub async fn run_typed<U>(&mut self, input: &str) -> Result<U, AgentError>

where U: DeserializeOwned + JsonSchema + Send + Sync,

Typed run: same semantics as Agent::run (records input, drives the loop), but the final answer is deserialized into the type parameter U after validation — this run auto-generates a JSON Schema from U (schemars-derived, the same pipeline as tool parameter schemas), with no constructor configuration needed.

Relation to Agent::run: run returns the model’s answer text (free text, or the JSON text of a hand-written Schema from with_structured_output); run_typed generates the Schema from the type and returns the type. The return type is declared in the let annotation (see # Examples); no turbofish needed.

Errors

• AgentError::StructuredRetriesExhausted: validation failures are fed back to the model for retry, but the AgentConfig::max_structured_retries budget is exhausted without success;
• AgentError::StructuredParse: validation passed but deserialization failed (when a derived schema customized with #[schemars(...)] disagrees with the serde representation);
• otherwise the same as Agent::run (Memory / Provider / round limit / cancellation).

Examples

use molo::{FakeProvider, FakeReply, ReActAgent};
use schemars::JsonSchema;
use serde::Deserialize;

#[derive(Deserialize, JsonSchema)]
struct Weather {
    city: String,
}

let mut agent = ReActAgent::new(
    FakeProvider::new([FakeReply::Text(r#"{"city":"Beijing"}"#.into())]),
    molo::tool::ToolRegistry::new(),
    "You are a weather assistant",
);
let weather: Weather = agent.run_typed("How's the weather in Beijing").await?;
assert_eq!(weather.city, "Beijing");

Trait Implementations

impl Agent for ReActAgent

fn run_request_with_context<'life0, 'async_trait>( &'life0 mut self, request: RunRequest, context: RunContext, ) -> Pin<Box<dyn Future<Output = Result<RunOutput, AgentError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

One structured run with caller-provided execution context.

fn run_stream_request_with_context<'a, 'async_trait>( &'a mut self, request: RunRequest, context: RunContext, ) -> Pin<Box<dyn Future<Output = Result<BoxStream<'a, Result<MessageChunk, AgentError>>, AgentError>> + Send + 'async_trait>>

where Self: 'async_trait, 'a: 'async_trait,

Streaming structured run with caller-provided execution context.

fn run_request<'life0, 'async_trait>( &'life0 mut self, request: RunRequest, ) -> Pin<Box<dyn Future<Output = Result<RunOutput, AgentError>> + Send + 'async_trait>>

where Self: Send + 'async_trait, 'life0: 'async_trait,

One structured run with a generated RunContext.

fn run<'life0, 'life1, 'async_trait>( &'life0 mut self, input: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<String, AgentError>> + Send + 'async_trait>>

where Self: Send + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

One run: record the user input, drive the reasoning loop, and return the model’s final answer as text.

fn run_stream_request<'a, 'async_trait>( &'a mut self, request: RunRequest, ) -> Pin<Box<dyn Future<Output = Result<BoxStream<'a, Result<MessageChunk, AgentError>>, AgentError>> + Send + 'async_trait>>

where Self: Send + 'async_trait, 'a: 'async_trait,

Streaming structured run with a generated RunContext.

fn run_stream<'a, 'async_trait>( &'a mut self, input: &'a str, ) -> Pin<Box<dyn Future<Output = Result<BoxStream<'a, Result<MessageChunk, AgentError>>, AgentError>> + Send + 'async_trait>>

where Self: Send + 'async_trait, 'a: 'async_trait,

Streaming run: same semantics as run, with the reply returned as a stream of message chunks (see MessageChunk), ending with MessageChunk::Done; errors are produced as Err items and terminate the stream (no Done afterwards).

impl AgentKernel for ReActAgent

fn start<'life0, 'life1, 'async_trait>( &'life0 mut self, request: RunRequest, context: &'life1 RunContext, ) -> Pin<Box<dyn Future<Output = Result<AgentAction, AgentError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Starts a run and returns the first action.

fn observe<'life0, 'life1, 'async_trait>( &'life0 mut self, observation: Observation, context: &'life1 RunContext, ) -> Pin<Box<dyn Future<Output = Result<AgentAction, AgentError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Consumes an observation and returns the next action.

impl Debug for ReActAgent

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl TypedAgent for ReActAgent

fn run_typed_request_with_context<'life0, 'async_trait, U>( &'life0 mut self, request: RunRequest, context: RunContext, ) -> Pin<Box<dyn Future<Output = Result<TypedRunOutput<U>, AgentError>> + Send + 'async_trait>>

where U: DeserializeOwned + JsonSchema + Send + Sync + 'async_trait, Self: 'async_trait, 'life0: 'async_trait,

Typed structured run with caller-provided execution context.

fn run_typed_request<'life0, 'async_trait, U>( &'life0 mut self, request: RunRequest, ) -> Pin<Box<dyn Future<Output = Result<TypedRunOutput<U>, AgentError>> + Send + 'async_trait>>

where U: DeserializeOwned + JsonSchema + Send + Sync + 'async_trait, Self: Send + 'async_trait, 'life0: 'async_trait,

Typed structured run with a generated RunContext.

fn run_typed<'life0, 'life1, 'async_trait, U>( &'life0 mut self, input: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<U, AgentError>> + Send + 'async_trait>>

where U: DeserializeOwned + JsonSchema + Send + Sync + 'async_trait, Self: Send + 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Typed run: the final answer is deserialized into U once validation passes.