Step‑by‑Step Guide: Building a ReAct Agent with LlamaIndex Workflows

ReAct Agent Overview

The ReAct paradigm combines iterative reasoning, action, and observation so that an LLM can tackle complex tasks by alternating between thinking and using external tools. The loop proceeds as follows:

Reasoning : the agent analyses the task, the current context, and decides the next action.

Action : it invokes an external tool (e.g., a search API, a calculator, or a custom function).

Observe & Loop : the result of the tool is fed back into the reasoning step, and the cycle repeats until the task is solved.

This dynamic approach enables the agent to handle open‑ended or exploratory queries with higher autonomy.

Designing the ReAct Workflow

The essential steps of a ReAct workflow are:

Feed the LLM with the original question, conversation history, tool definitions, and any previously obtained tool outputs.

If the LLM can answer directly, output the answer and terminate.

If the LLM decides a tool is needed, emit the tool name and arguments.

Execute the requested tool, capture its output, and add it to the reasoning history.

Loop back to step 1, repeating until the LLM produces a final answer.

The following diagram (from the official LlamaIndex sample) visualises this flow:

Implementing the Workflow with LlamaIndex

First, define the custom Event subclasses that represent each stage of the loop.

from llama_index.core.llms import ChatMessage
from llama_index.core.tools import ToolSelection, ToolOutput
from llama_index.core.workflow import Event
import os

# Notification event (no payload)
class PrepEvent(Event):
    pass

# LLM input event – carries the assembled message list
class InputEvent(Event):
    input: list[ChatMessage]

# Tool call event – specifies which tool to invoke and with what arguments
class ToolCallEvent(Event):
    tool_calls: list[ToolSelection]

# Tool output event – wraps the tool's result for the reasoning history
class FunctionOutputEvent(Event):
    output: ToolOutput

Next, create the ReActAgent class that inherits from Workflow. The constructor wires together the LLM, a list of tools, a memory buffer, a formatter that builds the LLM input, and an output parser that interprets the LLM’s ReAct‑style response.

from typing import Any, List
from llama_index.core.agent.react import ReActChatFormatter, ReActOutputParser
from llama_index.core.agent.react.types import ActionReasoningStep, ObservationReasoningStep
from llama_index.core.llms.llm import LLM
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.tools.types import BaseTool
from llama_index.core.workflow import Context, Workflow, StartEvent, StopEvent, step
from llama_index.llms.openai import OpenAI

class ReActAgent(Workflow):
    def __init__(self, *args: Any, llm: LLM | None = None, tools: list[BaseTool] | None = None, extra_context: str | None = None, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.tools = tools or []
        self.llm = llm or OpenAI()
        self.memory = ChatMemoryBuffer.from_defaults(llm=llm)
        self.formatter = ReActChatFormatter(context=extra_context or "")
        self.output_parser = ReActOutputParser()
        self.sources = []

The workflow consists of several @step functions:

@step
async def new_user_msg(self, ctx: Context, ev: StartEvent) -> PrepEvent:
    """Entry point – store the user query in memory and start a fresh reasoning list."""
    self.sources = []
    user_input = ev.input
    user_msg = ChatMessage(role="user", content=user_input)
    self.memory.put(user_msg)
    await ctx.set("current_reasoning", [])
    return PrepEvent()

@step
async def prepare_chat_history(self, ctx: Context, ev: PrepEvent) -> InputEvent:
    """Assemble conversation and reasoning history into the LLM input format."""
    chat_history = self.memory.get()
    current_reasoning = await ctx.get("current_reasoning", default=[])
    llm_input = self.formatter.format(self.tools, chat_history, current_reasoning=current_reasoning)
    return InputEvent(input=llm_input)

@step
async def handle_llm_input(self, ctx: Context, ev: InputEvent) -> ToolCallEvent | StopEvent:
    """Call the LLM, parse its ReAct response, and decide the next event."""
    response = await self.llm.achat(ev.input)
    try:
        reasoning_step = self.output_parser.parse(response.message.content)
        (await ctx.get("current_reasoning", default=[])).append(reasoning_step)
        if reasoning_step.is_done:
            self.memory.put(ChatMessage(role="assistant", content=reasoning_step.response))
            return StopEvent(result={
                "response": reasoning_step.response,
                "sources": [*self.sources],
                "reasoning": await ctx.get("current_reasoning", default=[]),
            })
        elif isinstance(reasoning_step, ActionReasoningStep):
            return ToolCallEvent(tool_calls=[
                ToolSelection(tool_id="", tool_name=reasoning_step.action, tool_kwargs=reasoning_step.action_input)
            ])
    except Exception as e:
        (await ctx.get("current_reasoning", default=[])).append(
            ObservationReasoningStep(observation=f"There was an error in parsing my reasoning: {e}")
        )
    return PrepEvent()

@step
async def handle_tool_calls(self, ctx: Context, ev: ToolCallEvent) -> PrepEvent:
    """Execute the requested tools and feed their outputs back into the reasoning history."""
    tools_by_name = {tool.metadata.get_name(): tool for tool in self.tools}
    for tool_call in ev.tool_calls:
        tool = tools_by_name.get(tool_call.tool_name)
        if not tool:
            (await ctx.get("current_reasoning", default=[])).append(
                ObservationReasoningStep(observation=f"Tool {tool_call.tool_name} does not exist")
            )
            continue
        try:
            tool_output = tool(**tool_call.tool_kwargs)
            self.sources.append(tool_output)
            (await ctx.get("current_reasoning", default=[])).append(
                ObservationReasoningStep(observation=tool_output.content)
            )
        except Exception as e:
            (await ctx.get("current_reasoning", default=[])).append(
                ObservationReasoningStep(observation=f"Error calling tool {tool.metadata.get_name()}: {e}")
            )
    return PrepEvent()

Testing the Agent

Two mock tools are created with FunctionTool.from_defaults: one that simulates sending an email and another that returns fake customer data.

from llama_index.core.tools import BaseTool, FunctionTool

def send_email(subject: str, message: str, email: str) -> None:
    """Simulated email sender"""
    print(f"邮件已发送至 {email}，主题为 {subject}，内容为 {message}")

tool_send_mail = FunctionTool.from_defaults(fn=send_email, name='tool_send_mail', description='用于发送电子邮件')

def query_customer(phone: str) -> str:
    """Simulated customer lookup"""
    return "该客户信息为:
姓名: 张三
积分: 50000分
邮件: [email protected]"

tool_customer = FunctionTool.from_defaults(fn=query_customer, name='tool_customer', description='用于查询客户信息，包括姓名、积分与邮件')

agent = ReActAgent(
    llm=OpenAI(model="gpt-4o-mini"),
    tools=[tool_send_mail, tool_customer],
    timeout=120,
    verbose=True,
)

Running the agent with a request that requires both a customer lookup and an email send demonstrates the full ReAct loop.

async def main():
    ret = await agent.run(input="给客户13688888888发电子邮件，通知他最新的积分")
    print(ret["response"])

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

The resulting reasoning history (shown in the screenshot below) confirms that the agent first called tool_customer, observed the returned data, then called tool_send_mail, and finally produced the answer.

Conclusion

LlamaIndex’s new Workflows feature provides a low‑level, event‑driven framework for building sophisticated ReAct agents, comparable to LangChain’s LangGraph but with its own design choices. The example shows how developers can gain fine‑grained control over reasoning, tool usage, and memory while keeping the implementation concise and readable.