ReActWorkflow¶

Built-in implementation of the ReAct (Reasoning and Acting) pattern.

Quick Example¶

from mamba_agents import Agent
from mamba_agents.workflows import ReActWorkflow, ReActConfig, ReActHooks
from mamba_agents.tools import read_file, run_bash

agent = Agent("gpt-4o", tools=[read_file, run_bash])

config = ReActConfig(
    max_iterations=15,
    expose_reasoning=True,
    auto_compact_in_workflow=True,
)

hooks = ReActHooks(
    on_thought=lambda s, t: print(f"Thought: {t}"),
    on_action=lambda s, tool, args: print(f"Action: {tool}"),
)

workflow = ReActWorkflow(agent=agent, config=config, hooks=hooks)
result = await workflow.run("Analyze the codebase")

# Access scratchpad
for entry in result.state.context.scratchpad:
    print(f"[{entry.entry_type}] {entry.content}")

Classes¶

ReActWorkflow¶

The main workflow class.

ReActConfig¶

Configuration extending WorkflowConfig.

ReActState¶

State tracking for ReAct execution.

ReActHooks¶

Callbacks extending WorkflowHooks.

API Reference¶

ReActWorkflow ¶

ReActWorkflow(
    agent: Agent[DepsT, Any],
    config: ReActConfig | None = None,
    hooks: ReActHooks | None = None,
    prompt_manager: PromptManager | None = None,
)

Bases: Workflow[DepsT, str, ReActState]

ReAct (Reasoning + Acting) workflow implementation.

Implements the Thought -> Action -> Observation loop until the agent calls the final_answer tool to signal task completion.

The workflow: 1. Sends the task to the agent 2. Agent thinks and calls tools 3. Tool results become observations 4. Loop continues until final_answer is called or max_iterations reached

Example

from mamba_agents import Agent from mamba_agents.workflows.react import ReActWorkflow, ReActConfig

agent = Agent("gpt-4o", tools=[read_file, run_bash]) workflow = ReActWorkflow( ... agent=agent, ... config=ReActConfig(max_iterations=10), ... ) result = workflow.run_sync("Find the bug in main.py") print(result.output) # Final answer print(result.state.context.scratchpad) # All steps

Initialize the ReAct workflow.

PARAMETER	DESCRIPTION
`agent`	Agent instance to use for execution. TYPE: `Agent[DepsT, Any]`
`config`	ReAct-specific configuration. TYPE: `ReActConfig \| None` DEFAULT: `None`
`hooks`	Optional hooks for observability. TYPE: `ReActHooks \| None` DEFAULT: `None`
`prompt_manager`	Optional PromptManager for template resolution. TYPE: `PromptManager \| None` DEFAULT: `None`

Source code in src/mamba_agents/workflows/react/workflow.py

def __init__(
    self,
    agent: Agent[DepsT, Any],
    config: ReActConfig | None = None,
    hooks: ReActHooks | None = None,
    prompt_manager: PromptManager | None = None,
) -> None:
    """Initialize the ReAct workflow.

    Args:
        agent: Agent instance to use for execution.
        config: ReAct-specific configuration.
        hooks: Optional hooks for observability.
        prompt_manager: Optional PromptManager for template resolution.
    """
    self._react_config = config or ReActConfig()
    super().__init__(self._react_config, hooks)

    self._agent = agent
    self._react_hooks: ReActHooks = hooks or ReActHooks()
    self._last_state: ReActState | None = None
    self._prompt_manager = prompt_manager

    # Register the final_answer tool
    self._agent.tool_plain(
        create_final_answer_tool(),
        name=self._react_config.final_answer_tool_name,
        retries=self._react_config.tool_retry_count,
    )

name `property` ¶

name: str

Get the workflow name.

agent `property` ¶

agent: Agent[DepsT, Any]

Get the agent instance.

react_config `property` ¶

react_config: ReActConfig

Get the ReAct-specific configuration.

prompt_manager `property` ¶

prompt_manager: PromptManager | None

Get the prompt manager.

Returns None if no prompt manager was provided or created.

get_token_usage ¶

get_token_usage() -> TokenUsage

Get token usage statistics.

Delegates to the agent's usage tracker.

RETURNS	DESCRIPTION
`TokenUsage`	TokenUsage with total prompt/completion/total tokens.

Source code in src/mamba_agents/workflows/react/workflow.py

def get_token_usage(self) -> TokenUsage:
    """Get token usage statistics.

    Delegates to the agent's usage tracker.

    Returns:
        TokenUsage with total prompt/completion/total tokens.
    """
    return self._agent.get_usage()

get_cost ¶

get_cost() -> float

Get estimated cost.

Delegates to the agent's cost estimator.

RETURNS	DESCRIPTION
`float`	Estimated cost in USD.

Source code in src/mamba_agents/workflows/react/workflow.py

def get_cost(self) -> float:
    """Get estimated cost.

    Delegates to the agent's cost estimator.

    Returns:
        Estimated cost in USD.
    """
    return self._agent.get_cost()

get_scratchpad ¶

get_scratchpad() -> list

Get the scratchpad entries from the last run.

RETURNS	DESCRIPTION
`list`	List of ScratchpadEntry objects, or empty list if no run yet.

Source code in src/mamba_agents/workflows/react/workflow.py

def get_scratchpad(self) -> list:
    """Get the scratchpad entries from the last run.

    Returns:
        List of ScratchpadEntry objects, or empty list if no run yet.
    """
    if self._last_state is None:
        return []
    return self._last_state.scratchpad

get_reasoning_trace ¶

get_reasoning_trace() -> str

Get formatted reasoning trace from the last run.

RETURNS	DESCRIPTION
`str`	Formatted scratchpad text, or empty string if no run yet.

Source code in src/mamba_agents/workflows/react/workflow.py

def get_reasoning_trace(self) -> str:
    """Get formatted reasoning trace from the last run.

    Returns:
        Formatted scratchpad text, or empty string if no run yet.
    """
    if self._last_state is None:
        return ""
    return self._last_state.get_scratchpad_text(self._react_config)

ReActConfig ¶

Bases: WorkflowConfig

Configuration for ReAct workflow execution.

Extends WorkflowConfig with ReAct-specific settings for reasoning trace visibility, termination strategy, and context management.

ATTRIBUTE	DESCRIPTION
`expose_reasoning`	Whether Thought steps appear in workflow state. TYPE: `bool`
`reasoning_prefix`	Prefix for reasoning/thought in prompts. TYPE: `str`
`action_prefix`	Prefix for actions in prompts. TYPE: `str`
`observation_prefix`	Prefix for observations in prompts. TYPE: `str`
`termination_strategy`	How to detect workflow completion ("tool" only for MVP). TYPE: `TerminationStrategy`
`final_answer_tool_name`	Name of the tool that signals completion. TYPE: `str`
`auto_compact_in_workflow`	Whether to auto-compact context mid-workflow. TYPE: `bool`
`compact_threshold_ratio`	Ratio of context limit to trigger compaction. TYPE: `float`
`max_consecutive_thoughts`	Max thoughts without action before forcing action. TYPE: `int`
`include_scratchpad`	Whether to include full reasoning history in each iteration. TYPE: `bool`
`tool_retry_count`	Number of retries for failed tool calls. TYPE: `int`
`system_prompt_template`	Optional template config for system prompt. TYPE: `TemplateConfig \| None`
`iteration_prompt_template`	Optional template config for iteration prompts. TYPE: `TemplateConfig \| None`

Example

config = ReActConfig( ... max_iterations=15, ... expose_reasoning=True, ... termination_strategy="tool", ... )

With custom prompt templates¶

from mamba_agents.prompts import TemplateConfig config = ReActConfig( ... system_prompt_template=TemplateConfig(name="workflow/react_system"), ... iteration_prompt_template=TemplateConfig(name="workflow/react_iteration"), ... )

ReActState `dataclass` ¶

ReActState(
    task: str,
    scratchpad: list[ScratchpadEntry] = list(),
    current_thought: str | None = None,
    current_action: str | None = None,
    current_observation: str | None = None,
    is_terminated: bool = False,
    termination_reason: str | None = None,
    final_answer: str | None = None,
    iteration_token_counts: list[int] = list(),
    total_tokens_used: int = 0,
    compaction_count: int = 0,
    consecutive_thought_count: int = 0,
)

Internal state for ReAct workflow execution.

This is stored in WorkflowState.context and tracks the full reasoning trace, termination status, and token usage.

ATTRIBUTE	DESCRIPTION
`task`	The original user task/prompt. TYPE: `str`
`scratchpad`	List of all Thought/Action/Observation entries. TYPE: `list[ScratchpadEntry]`
`current_thought`	The most recent thought generated. TYPE: `str \| None`
`current_action`	The most recent action taken. TYPE: `str \| None`
`current_observation`	The most recent observation received. TYPE: `str \| None`
`is_terminated`	Whether the workflow has terminated. TYPE: `bool`
`termination_reason`	Why the workflow terminated. TYPE: `str \| None`
`final_answer`	The final answer if workflow completed successfully. TYPE: `str \| None`
`iteration_token_counts`	Token count for each iteration. TYPE: `list[int]`
`total_tokens_used`	Total tokens used across all iterations. TYPE: `int`
`compaction_count`	Number of times context was compacted. TYPE: `int`
`consecutive_thought_count`	Consecutive thoughts without action (for forcing action). TYPE: `int`

add_thought ¶

add_thought(thought: str, token_count: int = 0) -> None

Add a thought to the scratchpad.

PARAMETER	DESCRIPTION
`thought`	The reasoning/thought content. TYPE: `str`
`token_count`	Estimated token count. TYPE: `int` DEFAULT: `0`

Source code in src/mamba_agents/workflows/react/state.py

def add_thought(self, thought: str, token_count: int = 0) -> None:
    """Add a thought to the scratchpad.

    Args:
        thought: The reasoning/thought content.
        token_count: Estimated token count.
    """
    self.scratchpad.append(
        ScratchpadEntry(
            entry_type="thought",
            content=thought,
            token_count=token_count,
        )
    )
    self.current_thought = thought
    self.consecutive_thought_count += 1

add_action ¶

add_action(
    action: str,
    token_count: int = 0,
    metadata: dict[str, Any] | None = None,
) -> None

Add an action to the scratchpad.

PARAMETER	DESCRIPTION
`action`	The action description (tool name and args). TYPE: `str`
`token_count`	Estimated token count. TYPE: `int` DEFAULT: `0`
`metadata`	Additional metadata (tool_name, tool_args, etc.). TYPE: `dict[str, Any] \| None` DEFAULT: `None`

Source code in src/mamba_agents/workflows/react/state.py

def add_action(
    self,
    action: str,
    token_count: int = 0,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Add an action to the scratchpad.

    Args:
        action: The action description (tool name and args).
        token_count: Estimated token count.
        metadata: Additional metadata (tool_name, tool_args, etc.).
    """
    self.scratchpad.append(
        ScratchpadEntry(
            entry_type="action",
            content=action,
            token_count=token_count,
            metadata=metadata or {},
        )
    )
    self.current_action = action
    self.consecutive_thought_count = 0  # Reset on action

add_observation ¶

add_observation(
    observation: str,
    token_count: int = 0,
    metadata: dict[str, Any] | None = None,
) -> None

Add an observation to the scratchpad.

PARAMETER	DESCRIPTION
`observation`	The tool result/observation content. TYPE: `str`
`token_count`	Estimated token count. TYPE: `int` DEFAULT: `0`
`metadata`	Additional metadata (tool_name, is_error, etc.). TYPE: `dict[str, Any] \| None` DEFAULT: `None`

Source code in src/mamba_agents/workflows/react/state.py

def add_observation(
    self,
    observation: str,
    token_count: int = 0,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Add an observation to the scratchpad.

    Args:
        observation: The tool result/observation content.
        token_count: Estimated token count.
        metadata: Additional metadata (tool_name, is_error, etc.).
    """
    self.scratchpad.append(
        ScratchpadEntry(
            entry_type="observation",
            content=observation,
            token_count=token_count,
            metadata=metadata or {},
        )
    )
    self.current_observation = observation

get_scratchpad_text ¶

get_scratchpad_text(config: ReActConfig) -> str

Format the scratchpad for prompt inclusion.

PARAMETER	DESCRIPTION
`config`	ReActConfig containing prefix settings. TYPE: `ReActConfig`

RETURNS	DESCRIPTION
`str`	Formatted scratchpad text with prefixes.

Source code in src/mamba_agents/workflows/react/state.py

def get_scratchpad_text(self, config: ReActConfig) -> str:
    """Format the scratchpad for prompt inclusion.

    Args:
        config: ReActConfig containing prefix settings.

    Returns:
        Formatted scratchpad text with prefixes.
    """
    if not self.scratchpad:
        return ""

    lines = []
    for entry in self.scratchpad:
        prefix = {
            "thought": config.reasoning_prefix,
            "action": config.action_prefix,
            "observation": config.observation_prefix,
        }[entry.entry_type]
        lines.append(f"{prefix}{entry.content}")

    return "\n".join(lines)

get_thoughts ¶

get_thoughts() -> list[str]

Get all thoughts from the scratchpad.

Source code in src/mamba_agents/workflows/react/state.py

def get_thoughts(self) -> list[str]:
    """Get all thoughts from the scratchpad."""
    return [e.content for e in self.scratchpad if e.entry_type == "thought"]

get_actions ¶

get_actions() -> list[str]

Get all actions from the scratchpad.

Source code in src/mamba_agents/workflows/react/state.py

def get_actions(self) -> list[str]:
    """Get all actions from the scratchpad."""
    return [e.content for e in self.scratchpad if e.entry_type == "action"]

get_observations ¶

get_observations() -> list[str]

Get all observations from the scratchpad.

Source code in src/mamba_agents/workflows/react/state.py

def get_observations(self) -> list[str]:
    """Get all observations from the scratchpad."""
    return [e.content for e in self.scratchpad if e.entry_type == "observation"]

ReActHooks ¶

ReActHooks(
    on_workflow_start: Callable[..., Awaitable[None] | None]
    | None = None,
    on_workflow_complete: Callable[
        ..., Awaitable[None] | None
    ]
    | None = None,
    on_workflow_error: Callable[..., Awaitable[None] | None]
    | None = None,
    on_step_start: Callable[..., Awaitable[None] | None]
    | None = None,
    on_step_complete: Callable[..., Awaitable[None] | None]
    | None = None,
    on_step_error: Callable[..., Awaitable[None] | None]
    | None = None,
    on_iteration_start: Callable[
        ..., Awaitable[None] | None
    ]
    | None = None,
    on_iteration_complete: Callable[
        ..., Awaitable[None] | None
    ]
    | None = None,
    on_thought: ThoughtHook | None = None,
    on_action: ActionHook | None = None,
    on_observation: ObservationHook | None = None,
    on_compaction: CompactionHook | None = None,
)

Bases: WorkflowHooks['ReActState', str]

Extended hooks for ReAct workflow observability.

Inherits the 8 base workflow hooks and adds 4 ReAct-specific hooks: - on_thought: Called when a reasoning trace is generated. - on_action: Called when a tool is about to be executed. - on_observation: Called when a tool result is received. - on_compaction: Called when context is compacted.

All hooks can be sync or async functions.

Example

def log_thought(state: ReActState, thought: str) -> None: ... print(f"Thought: {thought}")

hooks = ReActHooks(on_thought=log_thought) workflow = ReActWorkflow(agent, hooks=hooks)

Initialize ReAct hooks.

PARAMETER	DESCRIPTION
`on_workflow_start`	Called when workflow begins. TYPE: `Callable[..., Awaitable[None] \| None] \| None` DEFAULT: `None`
`on_workflow_complete`	Called when workflow completes successfully. TYPE: `Callable[..., Awaitable[None] \| None] \| None` DEFAULT: `None`
`on_workflow_error`	Called when workflow fails. TYPE: `Callable[..., Awaitable[None] \| None] \| None` DEFAULT: `None`
`on_step_start`	Called before each step executes. TYPE: `Callable[..., Awaitable[None] \| None] \| None` DEFAULT: `None`
`on_step_complete`	Called after each step completes. TYPE: `Callable[..., Awaitable[None] \| None] \| None` DEFAULT: `None`
`on_step_error`	Called when a step fails. TYPE: `Callable[..., Awaitable[None] \| None] \| None` DEFAULT: `None`
`on_iteration_start`	Called at the start of each iteration. TYPE: `Callable[..., Awaitable[None] \| None] \| None` DEFAULT: `None`
`on_iteration_complete`	Called at the end of each iteration. TYPE: `Callable[..., Awaitable[None] \| None] \| None` DEFAULT: `None`
`on_thought`	Called when a thought/reasoning is generated. TYPE: `ThoughtHook \| None` DEFAULT: `None`
`on_action`	Called when a tool is about to execute. TYPE: `ActionHook \| None` DEFAULT: `None`
`on_observation`	Called when a tool result is received. TYPE: `ObservationHook \| None` DEFAULT: `None`
`on_compaction`	Called when context is compacted. TYPE: `CompactionHook \| None` DEFAULT: `None`

Source code in src/mamba_agents/workflows/react/hooks.py

def __init__(
    self,
    # Inherited from WorkflowHooks
    on_workflow_start: Callable[..., Awaitable[None] | None] | None = None,
    on_workflow_complete: Callable[..., Awaitable[None] | None] | None = None,
    on_workflow_error: Callable[..., Awaitable[None] | None] | None = None,
    on_step_start: Callable[..., Awaitable[None] | None] | None = None,
    on_step_complete: Callable[..., Awaitable[None] | None] | None = None,
    on_step_error: Callable[..., Awaitable[None] | None] | None = None,
    on_iteration_start: Callable[..., Awaitable[None] | None] | None = None,
    on_iteration_complete: Callable[..., Awaitable[None] | None] | None = None,
    # ReAct-specific hooks
    on_thought: ThoughtHook | None = None,
    on_action: ActionHook | None = None,
    on_observation: ObservationHook | None = None,
    on_compaction: CompactionHook | None = None,
) -> None:
    """Initialize ReAct hooks.

    Args:
        on_workflow_start: Called when workflow begins.
        on_workflow_complete: Called when workflow completes successfully.
        on_workflow_error: Called when workflow fails.
        on_step_start: Called before each step executes.
        on_step_complete: Called after each step completes.
        on_step_error: Called when a step fails.
        on_iteration_start: Called at the start of each iteration.
        on_iteration_complete: Called at the end of each iteration.
        on_thought: Called when a thought/reasoning is generated.
        on_action: Called when a tool is about to execute.
        on_observation: Called when a tool result is received.
        on_compaction: Called when context is compacted.
    """
    super().__init__(
        on_workflow_start=on_workflow_start,
        on_workflow_complete=on_workflow_complete,
        on_workflow_error=on_workflow_error,
        on_step_start=on_step_start,
        on_step_complete=on_step_complete,
        on_step_error=on_step_error,
        on_iteration_start=on_iteration_start,
        on_iteration_complete=on_iteration_complete,
    )
    self.on_thought = on_thought
    self.on_action = on_action
    self.on_observation = on_observation
    self.on_compaction = on_compaction

trigger_thought `async` ¶

trigger_thought(state: ReActState, thought: str) -> None

Trigger the on_thought hook.

PARAMETER	DESCRIPTION
`state`	Current ReAct state. TYPE: `ReActState`
`thought`	The generated thought/reasoning. TYPE: `str`

Source code in src/mamba_agents/workflows/react/hooks.py

async def trigger_thought(self, state: ReActState, thought: str) -> None:
    """Trigger the on_thought hook.

    Args:
        state: Current ReAct state.
        thought: The generated thought/reasoning.
    """
    await self._trigger("on_thought", state, thought)

trigger_action `async` ¶

trigger_action(
    state: ReActState,
    tool_name: str,
    tool_args: dict[str, Any],
) -> None

Trigger the on_action hook.

PARAMETER	DESCRIPTION
`state`	Current ReAct state. TYPE: `ReActState`
`tool_name`	Name of the tool being called. TYPE: `str`
`tool_args`	Arguments passed to the tool. TYPE: `dict[str, Any]`

Source code in src/mamba_agents/workflows/react/hooks.py

async def trigger_action(
    self,
    state: ReActState,
    tool_name: str,
    tool_args: dict[str, Any],
) -> None:
    """Trigger the on_action hook.

    Args:
        state: Current ReAct state.
        tool_name: Name of the tool being called.
        tool_args: Arguments passed to the tool.
    """
    await self._trigger("on_action", state, tool_name, tool_args)

trigger_observation `async` ¶

trigger_observation(
    state: ReActState,
    observation: str,
    is_error: bool = False,
) -> None

Trigger the on_observation hook.

PARAMETER	DESCRIPTION
`state`	Current ReAct state. TYPE: `ReActState`
`observation`	The tool result or error message. TYPE: `str`
`is_error`	Whether this observation is an error. TYPE: `bool` DEFAULT: `False`

Source code in src/mamba_agents/workflows/react/hooks.py

async def trigger_observation(
    self,
    state: ReActState,
    observation: str,
    is_error: bool = False,
) -> None:
    """Trigger the on_observation hook.

    Args:
        state: Current ReAct state.
        observation: The tool result or error message.
        is_error: Whether this observation is an error.
    """
    await self._trigger("on_observation", state, observation, is_error)

trigger_compaction `async` ¶

trigger_compaction(result: CompactionResult) -> None

Trigger the on_compaction hook.

PARAMETER	DESCRIPTION
`result`	The compaction result with details. TYPE: `CompactionResult`

Source code in src/mamba_agents/workflows/react/hooks.py

async def trigger_compaction(self, result: CompactionResult) -> None:
    """Trigger the on_compaction hook.

    Args:
        result: The compaction result with details.
    """
    await self._trigger("on_compaction", result)

ReActWorkflow¶

Quick Example¶

Classes¶

ReActWorkflow¶

ReActConfig¶

ReActState¶

ReActHooks¶

API Reference¶

ReActWorkflow ¶

name property ¶

agent property ¶

react_config property ¶

prompt_manager property ¶

get_token_usage ¶

get_cost ¶

get_scratchpad ¶

get_reasoning_trace ¶

ReActConfig ¶

With custom prompt templates¶

ReActState dataclass ¶

add_thought ¶

add_action ¶

add_observation ¶

get_scratchpad_text ¶

get_thoughts ¶

get_actions ¶

get_observations ¶

ReActHooks ¶

trigger_thought async ¶

trigger_action async ¶

trigger_observation async ¶

trigger_compaction async ¶

name `property` ¶

agent `property` ¶

react_config `property` ¶

prompt_manager `property` ¶

ReActState `dataclass` ¶

trigger_thought `async` ¶

trigger_action `async` ¶

trigger_observation `async` ¶

trigger_compaction `async` ¶