LlamaIndex Agent 类全景解析：AgentWorkflow、FunctionAgent、ReActAgent 与 CodeActAgent 的事件驱动架构

LlamaIndex Agent 类全景解析：AgentWorkflow、FunctionAgent、ReActAgent 与 CodeActAgent 的事件驱动架构

【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index

本文基于 LlamaIndex 核心库的 Agent API 文档页（docs/api_reference/api_reference/agent/index.md），完整讲解llama_index.core.agent.workflow模块对外暴露的 10 个核心类：AgentWorkflow、BaseWorkflowAgent、FunctionAgent、ReActAgent、CodeActAgent、AgentInput、AgentStream、AgentOutput、ToolCall、ToolCallResult。读完后你将能够：用源码级视角理解 Agent 的推理-行动循环、掌握三类内置 Agent 的适用场景与配置差异、并基于工作流事件模型构建单 Agent 与多 Agent 手移（handoff）系统。

一、模块定位与 API 总览

该文档页是 MkDocs 自动生成的 API 索引，指向llama_index.core.agent.workflow模块并列出以下成员：

• Agent 基类：AgentWorkflow（多 Agent 编排）、BaseWorkflowAgent（单 Agent 基类）；
• 内置 Agent：FunctionAgent（函数调用式）、ReActAgent（提示词推理式）、CodeActAgent（代码执行式）；
• 工作流事件：AgentInput、AgentStream、AgentOutput、ToolCall、ToolCallResult。

这些符号的完整导出定义在 workflow/init.py，并从 agent/init.py 提升为llama_index.core.agent的公共 API。各成员对应的源码文件：

类	源码文件
BaseWorkflowAgent	base_agent.py
FunctionAgent	function_agent.py
ReActAgent	react_agent.py
CodeActAgent	codeact_agent.py
AgentWorkflow	multi_agent_workflow.py
事件类（AgentInput等）	workflow_events.py

从源码结构看，整个模块构建在 LlamaIndex 的 Workflow 引擎之上：BaseWorkflowAgent同时继承Workflow、BaseModel（Pydantic）与PromptMixin（base_agent.py#L87-L92），因此 Agent 既是"声明式配置"（字段即参数），也是"事件驱动流程"（@step步骤方法 + 事件流转）。配套测试位于 tests/agent 目录（含workflow/、react/子目录），可作为行为验证参考。

二、BaseWorkflowAgent：所有 Agent 的骨架与配置参数

BaseWorkflowAgent定义了 Agent 的通用字段、推理循环步骤和终止逻辑，是FunctionAgent/ReActAgent/CodeActAgent的共同基类。

2.1 可配置字段一览

以下字段均在 base_agent.py#L94-L142 中以 PydanticField声明：

参数	类型	默认值	说明
name	str	"Agent"	Agent 名称，多 Agent 场景中用于 handoff 定位
description	str	"An agent that can perform a task"	职责描述，会被写入 handoff 提示词供 LLM 判断移交
system_prompt	Optional[str]	None	系统提示词，注入在每轮 LLM 输入最前面
tools	Optional[List[Union[BaseTool, Callable]]]	None	工具列表；普通函数会被自动包装为FunctionTool
tool_retriever	Optional[ObjectRetriever]	None	工具检索器，可替代静态tools，按输入动态取工具
can_handoff_to	Optional[List[str]]	None	允许手移目标 Agent 名称列表（None表示不限制）
llm	LLM	Settings.llm	默认取全局Settings.llm
initial_state	Dict[str, Any]	{}	初始状态，运行中可通过ctx.store.get('state')访问
state_prompt	str / BasePromptTemplate	DEFAULT_STATE_PROMPT	将状态渲染进最后一条消息的模板（须含{state}、{msg}）
output_cls	Optional[Type[BaseModel]]	None	结构化输出 Pydantic 模型；设置后会忽略structured_output_fn
structured_output_fn	Optional[Callable]	None	自定义结构化输出函数，输入List[ChatMessage]
streaming	bool	True	是否把 LLM 流式输出写入事件流
early_stopping_method	"force" / "generate"	"force"	达到最大迭代次数时抛错，或追加一次 LLM 调用生成最终回答

此外，构造时还可透传 Workflow 引擎参数timeout、verbose、service_manager、resource_manager、num_concurrent_runs（base_agent.py#L70-L76 中的WORKFLOW_KWARGS过滤逻辑）。

两个值得注意的校验细节：

1. 工具自动包装：validate_tools校验器会把非BaseTool的普通函数经FunctionTool.from_defaults()转换（base_agent.py#L206-L224）；
2. 保留工具名：名为handoff的工具会被拒绝，因为它是多 Agent 系统内部生成的移管工具名（base_agent.py#L226-L230）。

2.2 推理循环：由 @step 步骤串联的事件状态机

基类以 5 个@step方法实现了完整的"推理—工具调用—汇总"闭环（base_agent.py#L383-L690）：

AgentWorkflowStartEvent │ ▼ init_run ──► AgentInput （初始化 memory / state / max_iterations， │ 把 user_msg 或 chat_history 写入记忆） ▼ setup_agent ──► AgentSetup （前置 system_prompt、注入 state_prompt 格式化状态） │ ▼ run_agent_step ──► AgentOutput （调用子类实现的 take_step 获取 LLM 一步输出） │ ▼ parse_agent_output ──┬─ 无工具调用 ──► finalize() ──► StopEvent(result=AgentOutput) ├─ 有工具调用 ──► 为每个调用发送 ToolCall 事件 └─ 超过 max_iterations ──► force 抛错 / generate 收尾 │（有工具调用时） ▼ call_tool ──► ToolCallResult （执行工具，异常被捕获为 is_error 的 ToolOutput） │ ▼ aggregate_tool_results ──► 收集全部 ToolCallResult ──► 回发 AgentInput（进入下一轮）

关键机制说明：

• 记忆与状态隔离：_init_context（base_agent.py#L284-L312）把memory（默认ChatMemoryBuffer）、state（initial_state深拷贝）、max_iterations（默认DEFAULT_MAX_ITERATIONS = 20）、num_iterations等存入 workflow 的ctx.store，因此同一次.run()的多轮迭代共享同一份记忆。
• 迭代保护：parse_agent_output每轮自增计数；达到上限时，early_stopping_method="force"抛出WorkflowRuntimeError并提示用.run(.., max_iterations=...)调整；"generate"则追加DEFAULT_EARLY_STOPPING_PROMPT再做一次 LLM 调用收尾（base_agent.py#L482-L542）。
• 工具异常不中断：_call_tool捕获工具执行异常，转为ToolOutput(is_error=True)回传给 LLM 自行修正，只有 Workflow 引擎的内部"等待事件"异常才向上抛出（base_agent.py#L347-L381）。
• return_direct 短路：若某工具声明return_direct且执行成功，其输出直接作为最终回答走finalize，跳过后续 LLM 轮次（base_agent.py#L687-L720）。

2.3 抽象接口：自定义 Agent 需要实现什么

子类必须实现三个抽象方法（base_agent.py#L245-L264）：

async def take_step(self, ctx, llm_input, tools, memory) -> AgentOutput """执行一步推理，产出 AgentOutput（含 response 与 tool_calls）""" async def handle_tool_call_results(self, ctx, results, memory) -> None """把工具结果写回上下文/记忆""" async def finalize(self, ctx, output, memory) -> AgentOutput """收尾：把中间消息并入长期记忆等"""

这正是三种内置 Agent 差异化的唯一落点——循环骨架、事件模型、结构化输出、流式推送全部复用基类。

2.4 run() 的调用签名

BaseWorkflowAgent.run()提供两套签名（base_agent.py#L733-L814）：

# 推荐：标准 workflow 签名 agent.run(user_msg="你好", max_iterations=10, early_stopping_method="generate") agent.run(ctx=ctx, start_event=event, user_msg="你好") # 旧的位置参数签名（已标记 @deprecated，未来版本移除） agent.run(user_msg, chat_history, memory, ctx)

参数说明：user_msg（str或ChatMessage）与chat_history至少提供其一，否则init_run会抛出ValueError("Must provide either user_msg or chat_history")（base_agent.py#L428-L429）；memory可传入自定义BaseMemory实例；返回WorkflowHandler，其最终结果即StopEvent携带的AgentOutput事件。

最小可运行示例（单 Agent + 工具）：

from llama_index.core.agent.workflow import FunctionAgent from llama_index.core.tools import FunctionTool def get_weather(city: str) -> str: """查询指定城市的天气。""" return f"{city}：晴，25°C" agent = FunctionAgent( name="weather_agent", description="回答天气问题的 Agent", system_prompt="你是一个友好的天气助手。", tools=[FunctionTool.from_defaults(get_weather)], # 直接传函数也可，会自动包装 early_stopping_method="generate", # 达到上限时生成最终回答而非报错 ) result = agent.run("北京今天天气怎么样？", max_iterations=10) print(result.response.content)

三、事件模型详解：AgentInput、AgentStream、AgentOutput、ToolCall、ToolCallResult

事件是 Agent 对外暴露一切中间状态的唯一通道，全部定义在 workflow_events.py。

3.1 AgentInput —— 每轮进入 LLM 前的输入

class AgentInput(Event): input: list[ChatMessage] # 本轮完整 LLM 输入（记忆 + scratchpad） current_agent_name: str # 当前执行 Agent 名称

AgentInput在两个时机发出：init_run完成记忆装配后（base_agent.py#L435），以及aggregate_tool_results汇总工具结果后开启新一轮推理时（base_agent.py#L725）。监听它即可获得"Agent 下一步到底喂给 LLM 什么"的完整视图。

3.2 AgentStream —— 流式增量输出

class AgentStream(Event): delta: str # 本次增量文本 response: str # 累计响应 current_agent_name: str tool_calls: list[ToolSelection] = [] # 流中解析出的工具调用（FunctionAgent 特有） raw: Optional[Any] # 原始 LLM 响应（序列化时被排除） thinking_delta: Optional[str] # 思维链增量（如支持 reasoning 的模型）

基类的_get_llm_response在streaming=True时逐块调用astream_chat，把每个增量写为AgentStream事件（base_agent.py#L314-L345）。前端实时渲染 Agent 打字效果的依据就是它。

3.3 AgentOutput —— 单步（或最终）输出

class AgentOutput(Event): response: ChatMessage structured_response: Optional[Dict[str, Any]] current_agent_name: str tool_calls: list[ToolSelection] = [] # 本步解析出的工具选择 retry_messages: list[ChatMessage] = [] # 重试时附加的纠正消息 raw: Optional[Any] def get_pydantic_model(self, model) -> Optional[BaseModel] def __str__(self) -> str # 直接返回 response.content

两个实用点（workflow_events.py#L71-L95）：

• get_pydantic_model(model)把structured_response字典校验为任意 Pydantic 模型，校验失败只发出PydanticConversionWarning而返回None，不会中断流程；
• run()的最终返回值就是最后一个AgentOutput，str(result)即最终文本回答，result.structured_response为结构化结果。

结构化输出有两条生成路径，均在parse_agent_output的终止分支执行（base_agent.py#L569-L605）：structured_output_fn（自定义函数，支持协程）优先于output_cls（由generate_structured_response借助 LLM 抽取）；两者同时设置时output_cls生效、structured_output_fn被置空（base_agent.py#L173-L174）。生成成功会额外发出AgentStreamStructuredOutput事件（同样支持get_pydantic_model）。

3.4 ToolCall 与 ToolCallResult —— 工具执行的两端

class ToolCall(Event): tool_name: str tool_kwargs: dict tool_id: str class ToolCallResult(Event): tool_name: str tool_kwargs: dict tool_id: str tool_output: ToolOutput # 含 content / is_error / raw_output / exception return_direct: bool # 该工具是否声明直接返回

流程为：LLM 决定调用工具后，parse_agent_output对每个调用ctx.send_event(ToolCall(...))；call_tool步骤查找工具并执行（工具不存在时不报错，而是回一条is_error=True的ToolOutput，提示 LLM 换用可用工具，见 base_agent.py#L636-L645）；aggregate_tool_results用ctx.collect_events等齐本批全部ToolCallResult后再统一回写记忆（base_agent.py#L661-L685）——这保证了"并行工具调用"在结果语义上是原子的。

3.5 AgentWorkflowStartEvent —— 启动事件

run()内部构造的启动事件会自动把chat_history中的 dict 逐条校验转换为ChatMessage（workflow_events.py#L116-L136），因此历史消息可以直接以字典列表形式传入。

四、FunctionAgent：原生函数调用 Agent

function_agent.py 中的FunctionAgent是默认推荐实现，依赖 LLM 的 OpenAI 风格 function calling 能力。

4.1 特有配置

参数	默认值	说明
initial_tool_choice	None	首轮强制指定某个工具名（通过tool_choice传给 LLM），仅对最后一条 user 消息生效
allow_parallel_tool_calls	True	允许 LLM 一次性并行调用多个工具
scratchpad_key	"scratchpad"	工作流 store 中存放"本轮中间消息"的键名

4.2 实现要点

• 前置检查：take_step会验证self.llm.metadata.is_function_calling_model，非函数调用模型直接抛ValueError（function_agent.py#L109-L110）——若 LLM 不支持函数调用，应改用ReActAgent；
• scratchpad 设计：每轮 LLM 响应与工具结果（role="tool"消息）不写入长期记忆，而是累积在ctx.store的 scratchpad 中（function_agent.py#L112-L134）；只有finalize时才整体memory.aput_messages(scratchpad)并清空（function_agent.py#L180-L195）。这样长期记忆保持干净的"对话级"粒度，而中间推理细节只在当次运行内可见；
• 流式解析工具调用：流式路径下通过llm.get_tool_calls_from_response(..., error_on_no_tool_call=False)从每个增量中解析工具调用并随AgentStream事件透出（function_agent.py#L76-L99）；
• return_direct 短路：工具声明return_direct时，handle_tool_call_results会把工具输出再补一条 assistant 消息进 scratchpad 并break，配合基类逻辑直接终止（function_agent.py#L165-L176）。

FunctionAgent典型用法——接入知识库检索工具做 RAG Agent：

from llama_index.core.agent.workflow import FunctionAgent from llama_index.core.tools import QueryEngineTool from llama_index.core import VectorStoreIndex, SimpleDirectoryReader docs = SimpleDirectoryReader("path/to/docs").load_data() index = VectorStoreIndex.from_documents(docs) qtool = QueryEngineTool.from_defaults( query_engine=index.as_query_engine(), description="检索公司文档回答问题", ) agent = FunctionAgent( name="rag_agent", description="基于公司文档回答问题的 Agent", tools=[qtool], ) result = agent.run("我们的退款政策是什么？")

五、ReActAgent：提示词驱动的推理-行动 Agent

react_agent.py 的ReActAgent面向不具备原生函数调用能力的 LLM：它通过在提示词中约定 Thought/Action/Observation 格式，让 LLM 以纯文本表达工具调用意图，再由解析器提取。源码注释也明确指出该路径服务于 "non function calling llms"（react_agent.py#L103）。

5.1 特有配置

参数	类型	说明
output_parser	ReActOutputParser	解析 LLM 输出中的 Action 文本为ToolSelection
formatter	ReActChatFormatter	把聊天历史与推理步骤组装为带 ReAct 头部模板的 LLM 输入
reasoning_key	str，默认"current_reasoning"	store 中存放当前推理状态的键名

两个实现细节值得注意：

• system_prompt 自动注入：validate_formatter校验器会在formatter.system_header含{context}占位符时，把用户传入的system_prompt拼入 formatter 上下文，并切换到带 context 的CONTEXT_REACT_CHAT_SYSTEM_HEADER头部模板（react_agent.py#L50-L67）；
• 提示词可替换：_get_prompts暴露react_header提示模板，配合PromptMixin可用agent.set_prompts({"react_header": ...})在运行前改写 ReAct 头部（react_agent.py#L69-L82）。

5.2 FunctionAgent 与 ReActAgent 选型

维度	FunctionAgent	ReActAgent
LLM 要求	必须is_function_calling_model	任意 Chat LLM
工具调用表达	结构化tool_calls字段	提示词约定 + 文本解析
工具参数可靠性	由 LLM API 结构化保证	依赖解析器对文本的容错
额外组件	无	ReActOutputParser、ReActChatFormatter
适用场景	主流商用/开源函数调用模型	不支持 function calling 的模型或需要显式推理轨迹的场景

两者均继承BaseWorkflowAgent，对外.run()用法、事件类型、output_cls结构化输出完全一致，切换只需替换类名。

六、CodeActAgent：以"写代码执行"替代函数调用

codeact_agent.py 的CodeActAgent让 LLM 输出 Python 代码并在受控沙箱中执行，工具以"可导入函数"形式出现在代码里，适合需要循环、数据处理、多步组合工具的任务。

6.1 构造参数与约束

CodeActAgent( code_execute_fn, # 必填：async def code_execute_fn(code: str) -> Dict[str, Any] name="code_act_agent", description="A workflow agent that can execute code.", system_prompt=None, # 会追加在默认 CodeAct 提示词之后 tools=None, # 其他工具；会被注入代码上下文（不能 requires_context） can_handoff_to=None, llm=None, # 需要函数调用能力（execute 也是 tool） code_act_system_prompt=DEFAULT_CODE_ACT_PROMPT, streaming=True, )

实现要点（codeact_agent.py#L63-L140）：

• 强制注入 execute 工具：构造函数把code_execute_fn包装为名为execute的FunctionTool（EXECUTE_TOOL_NAME = "execute"）追加到工具列表，LLM 通过该工具提交代码块；
• 输出协议：默认提示词DEFAULT_CODE_ACT_PROMPT要求代码写在<execute>...</execute>标签内，其余文本直接面向用户；并约定"上一轮代码的顶层变量可被后续代码引用"（codeact_agent.py#L25-L58）；
• 工具过滤：_get_tool_fns会跳过handoff/execute两个保留名，并拒绝requires_context的工具（代码沙箱内无法注入 workflow 上下文）。

6.2 完整示例

import asyncio from llama_index.core.agent.workflow import CodeActAgent async def code_execute_fn(code: str) -> dict: """本地受限执行器：生产环境建议替换为远程沙箱（隔离、限资源）。""" g: dict = {"__builtins__": __builtins__} exec(code, g) # 仅演示用途 return {"ok": True} agent = CodeActAgent(code_execute_fn=code_execute_fn) result = agent.run("计算 1 到 100 中所有质数之和，并列出前 10 个质数") print(result.response.content)

注意code_execute_fn的执行安全完全由你负责：本地exec只适合演示，生产部署应接入带资源限制的网络隔离沙箱。

七、AgentWorkflow：多 Agent 编排与 handoff 机制

multi_agent_workflow.py 中的AgentWorkflow是"管理多个 Agent 并支持对话移交"的顶层 Workflow（docstring 原文："A workflow for managing multiple agents with handoffs"）。

7.1 构造参数与校验规则

AgentWorkflow( agents: List[BaseWorkflowAgent], # 至少一个 Agent initial_state: Optional[Dict] = None, # 全局状态（不支持按 Agent 单独设置） root_agent: Optional[str] = None, # 多 Agent 时必填 handoff_prompt: Optional[str] = None, # 须包含 {agent_info} 占位符 handoff_output_prompt: Optional[str] = None, # 须包含 {to_agent} 与 {reason} state_prompt: Optional[str] = None, # 须包含 {state} 与 {msg} timeout: Optional[float] = None, output_cls: Optional[Type[BaseModel]] = None, structured_output_fn: Optional[Callable] = None, early_stopping_method: Literal["force", "generate"] = "force", )

构造函数内置一组严格校验（multi_agent_workflow.py#L120-L182），多 Agent 场景下尤其容易踩坑：

1. 多个 Agent 时每个都必须有非默认name与description（默认值即报错，因为这两个字段会被写入 handoff 提示词供 LLM 路由）；
2. 不允许任何 Agent 设置initial_state（状态统一由AgentWorkflow(initial_state=...)提供）；
3. 单 Agent 时自动以其为 root；多 Agent 时root_agent必填且必须存在于agents中；
4. 三个提示模板的占位符缺失会直接抛ValueError。

7.2 handoff 的内部实现

多 Agent 时，AgentWorkflow会为当前 Agent 动态生成一个名为handoff的FunctionTool（multi_agent_workflow.py#L216-L246）：

• 工具描述由handoff_prompt.format(agent_info=str({name: description}))生成——这就是为什么 description 写得好坏直接影响路由质量；
• 该工具return_direct=True：LLM 一旦调用handoff(to_agent, reason)，基类的 return_direct 短路逻辑让控制权立即移交给目标 Agent，但不会停止整个工作流（base_agent.py中对tool_name == "handoff"做了专门豁免，base_agent.py#L717-L720）；
• handoff函数本身在 store 中校验目标合法性：Agent 不存在、或超出can_handoff_to白名单时返回错误文本让 LLM 重新选择，而不是硬失败（multi_agent_workflow.py#L73-L92）。

AgentWorkflow同样实现了完整的迭代/终止/结构化输出逻辑（复用DEFAULT_MAX_ITERATIONS与early_stopping_method），其事件流与单 Agent 一致：AgentStream/ToolCall/ToolCallResult/AgentOutput上的current_agent_name字段用于区分是哪个 Agent 产生的。

7.3 多 Agent 示例：路由 + 专业域 Agent

from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent from llama_index.core.llms import ChatMessage root = FunctionAgent( name="router", description="判断用户问题类型：天气问题移交给 weather_agent，其他问题自行回答。", can_handoff_to=["weather_agent"], # 只允许移交给天气 Agent ) weather = FunctionAgent( name="weather_agent", description="专门查询并播报天气，工具为 get_weather。", tools=[get_weather_tool], # FunctionTool.from_defaults(get_weather) ) workflow = AgentWorkflow( agents=[root, weather], root_agent="router", initial_state={"locale": "zh-CN"}, output_cls=None, # 如需结构化结果可传 Pydantic 模型 ) result = workflow.run("北京明天适合跑步吗？") print(result.response.content, result.current_agent_name)

八、事件消费与生产集成建议

把上述机制落到工程上，有三条实践要点（均有源码依据）：

1. 实时 UI 只订阅两类事件：AgentStream（打字机效果，含thinking_delta）与ToolCallResult（展示工具执行进度），最终答案等run()返回的AgentOutput。raw字段在序列化时被排除（exclude=True），适合调试但不必透传前端；
2. 结构化输出优先用output_cls：传入 Pydantic 模型即可，框架在终止分支自动调用generate_structured_response并流式发出AgentStreamStructuredOutput；消费端用event.get_pydantic_model(YourModel)做防御性校验（转换失败仅告警不崩，workflow_events.py#L81-L92）；
3. 预算控制：默认max_iterations=20，长任务通过.run(max_iterations=...)覆盖；配合early_stopping_method="generate"可把"撞上限报错"转为"强制收尾回答"，适合面向终端用户的场景。

九、小结

llama_index.core.agent.workflow模块的设计可以用一句话概括：用统一的 Workflow 事件骨架（AgentInput → AgentSetup → AgentOutput → ToolCall/ToolCallResult → 循环或终止）承载三种工具调用范式——FunctionAgent走 LLM 原生函数调用，ReActAgent走提示词解析，CodeActAgent走代码执行；BaseWorkflowAgent提供记忆、状态、迭代保护与结构化输出等横切能力，AgentWorkflow则在多 Agent 维度上叠加了基于handoff工具的动态路由。所有中间状态都以类型化事件（AgentStream、ToolCall、ToolCallResult、AgentOutput）外泄，使 Agent 系统天然可观测、可流式、可嵌入更大的 Workflow 编排。相关实现可分别追溯至 base_agent.py、function_agent.py、react_agent.py、codeact_agent.py、multi_agent_workflow.py 与 workflow_events.py，行为验证可参考 tests/agent 下的测试套件。

【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index