本文介绍如何使用 LlamaIndex Workflows 从零构建 ReAct 模式的 AI 智能体。ReAct 模式结合推理与行动，通过迭代循环处理复杂任务。文章详细阐述了事件定义、Agent 初始化、用户输入处理、LLM 调用及工具调用等核心步骤，并提供了完整的 Python 代码示例与测试流程，帮助开发者理解底层原理并实现灵活控制。同时补充了内存管理、超时控制及错误处理等最佳实践建议。
LlamaIndex 推出了新特性 Workflows，这是一种事件驱动、用于构建复杂 AI 工作流应用的新方法。在本文中，我们将学习如何基于 Workflows 来构建一个 ReAct 模式的 AI 智能体。尽管 LlamaIndex 框架中已经提供了开箱即用的 ReActAgent 组件，但通过 Workflows 从零构建 ReAct 智能体，可以更深入地了解其内部原理，在未来帮助实现更底层、更灵活的控制能力。
ReAct（Reasoning + Acting）模式的 AI 智能体采用迭代式的推理到行动的工作流程，旨在应对更复杂的人工任务和问题。它通过将推理步骤与实际行动相结合，使得智能体可以逐步理解任务、采取行动，并观察行动获得的新信息以推理后续步骤。过程大致如下：
ReAct 模式具备很好的动态性，使得 AI 能够应对复杂和未知情况，适用于更开放性的问题和探索性的任务，展现出更高的自主决策智能。
一个标准的 ReAct Agent 通常包含以下核心组件：
根据 ReAct 智能体的基本思想，其工作流中最核心的步骤（step）应该包括：
基于 LlamaIndex Workflows 开发 ReAct Agent 的工作流程图如下所示，展示了从用户输入到最终输出的完整闭环。
下面我们将参考官方样例，详细讲解如何使用 Python 代码实现这一逻辑。
参考上面的工作流图，我们需要定义几个需要的 Event 类型来传递状态和数据：
from llama_index.core.llms import ChatMessage
from llama_index.core.tools import ToolSelection, ToolOutput
from llama_index.core.workflow import Event
import os
# 通知事件：用于流程流转的触发信号
class PrepEvent(Event):
pass
# LLM 输入事件：包含输入 LLM 的历史消息列表
class InputEvent(Event):
input: list[ChatMessage]
# 工具调用事件：包含工具调用信息
class ToolCallEvent(Event):
tool_calls: list[ToolSelection]
# 工具输出事件：包含工具输出信息
class FunctionOutputEvent(Event):
output: ToolOutput
工作流初始化，主要是为了给智能体准备必备的'工具'，最重要的就是智能体需要的几大件：LLM 大模型、Memory 记忆、以及可以使用的 Tools 工具。
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)
# 可用的工具集合，支持动态扩展
tools or []
self.tools = tools or []
# 使用的 LLM（大模型），默认使用 OpenAI
self.llm = llm or OpenAI()
# 持久记忆，用于保存对话历史
self.memory = ChatMemoryBuffer.from_defaults(llm=llm)
# 用来把历史对话、已有的推理历史格式化成下一次 LLM 的输入消息历史
.formatter = ReActChatFormatter(context=extra_context )
.output_parser = ReActOutputParser()
.sources = []
这里有两个关键的辅助工具：
这是一次性的步骤，简单的把输入问题/任务放入 memory 即可：
@step
async def new_user_msg(self, ctx: Context, ev: StartEvent) -> PrepEvent:
"""
流程入口：接受用户输入，并放置到 Memory 中；并触发下一步
"""
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()
在这个步骤中，利用上面初始化的 formatter，把对话历史与推理历史格式化，用来输入给 LLM 做推理。注意在一次任务中，这个步骤有可能会被多次循环调用，除非输入问题被 LLM 直接回答（无需借助工具）。
@step
async def prepare_chat_history(
self, ctx: Context, ev: PrepEvent
) -> InputEvent:
"""
将对话与推理历史组装成 LLM 的输入消息列表 (通常是角色 + 内容）。
推理历史包括：
1. LLM 输出的推理结果 (直接回答问题、需要工具调用、观察工具调用结果后可以回答）
2. 工具调用的结果
"""
# 获取历史消息
chat_history = self.memory.get()
print(f'\n------------当前消息历史------------')
for idx, message in enumerate(chat_history, start=1):
print(f'\n{idx}. {message}')
current_reasoning = await ctx.get("current_reasoning", default=[])
print('\n-------------当前推理历史------------')
for idx, reasoning in enumerate(current_reasoning, start=1):
print(f'\n{idx}. {reasoning}')
# 将历史用户消息与推理历史组装成列表
llm_input = self.formatter.format(
self.tools, chat_history, current_reasoning=current_reasoning
)
return InputEvent(input=llm_input)
使用上一步骤准备的输入内容，调用 LLM，并解析结果。以决定下一步动作（返回不同的事件），具体可以参考下面的代码及注释：
@step
async def handle_llm_input(
self, ctx: Context, ev: InputEvent
) -> ToolCallEvent | StopEvent:
"""
调用 LLM;
解析输出结果，获得推理结果；
判断是结束（可以回答问题）, 还是需要调用工具；
"""
chat_history = ev.input
# 调用 LLM
response = await self.llm.achat(chat_history)
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):
tool_name = reasoning_step.action
tool_args = reasoning_step.action_input
return ToolCallEvent(
tool_calls=[
ToolSelection(
tool_id="",
tool_name=tool_name,
tool_kwargs=tool_args,
)
]
)
except Exception as e:
(await ctx.get("current_reasoning", default=[])).append(
ObservationReasoningStep(
observation=f"There was an error in parsing my reasoning: "
)
)
PrepEvent()
这是智能体使用外部工具（Tools）的关键步骤。根据上一步骤 LLM 输出的工具调用需求，调用外部工具（可能有多次调用），并把返回结果放在推理历史中，用于下一次迭代。
@step
async def handle_tool_calls(
self, ctx: Context, ev: ToolCallEvent
) -> PrepEvent:
"""
工具调用，将调用结果作为 LLM 的观察对象;
并将观察内容添加到推理历史
"""
tool_calls = ev.tool_calls
tools_by_name = {tool.metadata.get_name(): tool for tool in self.tools}
# 工具调用
for tool_call in 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()
现在，整个 ReAct Agent 的工作流就完成了，过程还是比较简单清晰的。当然这里也会利用到一些 LlamaIndex 提供的组件，比如用来封装 LLM 推理结果的 xxxReasoningStep 组件等。
我们来测试这个 ReAct Agent 组件，准备两个模拟工具（利用 LlamaIndex 中的 FunctionTool 快速构造基于函数的工具），然后创建一个 Agent：
from llama_index.core.tools import BaseTool, FunctionTool
# 模拟发送邮件
def send_email(subject: str, message: str, email: str) -> None:
"""用于发送电子邮件"""
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:
"""用于查询客户信息"""
result = f"该客户信息为:\n姓名：张三\n积分：50000 分\n邮件：[email protected]"
return result
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
)
现在调用这个 Agent，我们发出一个比较复杂的请求，来看看会发生什么：
async def main():
ret = await agent.run(input="给客户 13688888888 发电子邮件，通知他最新的积分")
print(ret["response"])
if __name__ == "__main__":
import asyncio
asyncio.run(main())
这里的任务很显然需要借助两次工具调用，一次查询客户信息，一次发送邮件，我们从输出中来观察最后一次的迭代信息。
注意到这里的推理历史，完整的反应了 LLM 的'思考'过程：首先需要查询客户信息（调用 tool_customer）；然后观察到客户信息（工具调用结果）；判断需要发送邮件（调用 tool_send_mail）；最后观察返回结果后结束流程。这是一个完整的符合 ReAct 模式（推理 - 行动 - 观察）的工作流，也证明了这里基于 Workflows 构建的 ReAct Agent 的可用性。
在实际生产环境中部署此类智能体时，建议关注以下几点：
timeout 参数，防止智能体陷入死循环或长时间等待工具响应。verbose=True 模式有助于调试，但在生产环境应接入结构化日志系统以便追踪链路。至此我们对 LlamaIndex 所推出的新特性 Workflows 已经有了较为全面的认识。Workflows 是一个与 LangChain 的 LangGraph 相似的另一种智能体开发底层框架，两者都面向复杂的智能体/RAG 应用工作流，但又采取了不同的设计思想。至于哪个更好或许是见仁见智的问题，但对于大量 LLM 应用的开发者来说，的确又多了一个强大的工具。相信随着后续的迭代，Workflows 也会越来越强大，为构建更复杂的 AI 应用提供支持。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online