LlamaIndex 的 AgentWorkflow、编排者与自定义规划三种多智能体模式怎么选

LlamaIndex 的 AgentWorkflow、编排者与自定义规划三种多智能体模式怎么选

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

当你的任务需要多个 Agent 协作完成（比如先检索资料、再撰写报告、再由另一个 Agent 评审）时，问题往往不是"要不要用多智能体"，而是"控制权该放在哪里"：是让各 Agent 之间自行交接，还是集中在一个编排者手里，或者干脆自己写规划逻辑。LlamaIndex 官方文档给出三种现成路径：AgentWorkflow（内置）、编排者模式（子 Agent 作为工具）、自定义规划器（DIY 提示词 + 解析）。

本文以仓库文档中三个示例共同使用的"研究报告生成"任务（Research → Write → Review）为例，给出每种模式的适用条件、可运行代码和结果验证方式，帮助你按控制粒度需求选型。适用前提：Python 环境、一个支持工具调用的 LLM（示例使用 OpenAI，各 Agent 也可以各用不同 LLM）、用于网页检索的 Tavily API key。

先定判断标准：三种模式各适合什么情况

multi_agent.md 对每种模式给出了明确的适用条件：

• AgentWorkflow：希望开箱即得的多 Agent 行为、几乎不写额外代码，并且接受AgentWorkflow自带的默认 hand-off（交接）启发式。
• 编排者（Orchestrator）：希望有一个"单一决策点"决定每一步、方便注入自定义逻辑，但仍想用声明式的"Agent 即工具"体验，而不是自己写规划器。
• 自定义规划器：追求最大灵活性。你需要强制定制的计划格式（XML / JSON / YAML）、对接外部调度器，或获取前两种模式无法直接提供的额外元数据。

文档同时给出一张对比表：

文档给出的选型顺序是：快速原型先用AgentWorkflow；需要更多控制执行顺序时移到编排者模式；只有当前两者都无法表达你需要的流程时，才上自定义规划器。

三种模式的前置安装相同（示例 notebook 中写作%pip install ...，脚本环境去掉%即可）：

pip install llama-index pip install tavily-python

后续所有代码中的sk-...、sk-proj-...是文档示例里的占位 API key，替换为你自己的 OpenAI API key；tvly-...替换为你自己的 Tavily API key。

主路径：用 AgentWorkflow 跑通"研究—写作—评审"报告

完整的可运行版本见 agent_workflow_multi 示例。如果还没接触过 Agent 基础用法，文档建议先读 agent_workflow_basic 示例。

定义工具与三个子 Agent

四个工具负责网页检索和写入共享状态。record_notes、write_report、review_report通过ctx.store.edit_state()把结果写进工作流的state，这样各 Agent 之间靠状态传递数据，而不只是靠对话：

from tavily import AsyncTavilyClient from llama_index.core.workflow import Context async def search_web(query: str) -> str: """Useful for using the web to answer questions.""" client = AsyncTavilyClient(api_key="tvly-...") # 替换为你的 Tavily API key return str(await client.search(query)) async def record_notes(ctx: Context, notes: str, notes_title: str) -> str: """Useful for recording notes on a given topic. Your input should be notes with a title to save the notes under.""" async with ctx.store.edit_state() as ctx_state: if "research_notes" not in ctx_state["state"]: ctx_state["state"]["research_notes"] = {} ctx_state["state"]["research_notes"][notes_title] = notes return "Notes recorded." async def write_report(ctx: Context, report_content: str) -> str: """Useful for writing a report on a given topic. Your input should be a markdown formatted report.""" async with ctx.store.edit_state() as ctx_state: ctx_state["state"]["report_content"] = report_content return "Report written." async def review_report(ctx: Context, review: str) -> str: """Useful for reviewing a report and providing feedback. Your input should be a review of the report.""" async with ctx.store.edit_state() as ctx_state: ctx_state["state"]["review"] = review return "Report reviewed."

三个 Agent 用FunctionAgent定义。这里的can_handoff_to是 AgentWorkflow 模式的关键配置——它声明了每个 Agent 允许把控制权交给谁，Agent 之间的交接由框架在运行时执行：

from llama_index.core.agent.workflow import FunctionAgent, ReActAgent research_agent = FunctionAgent( name="ResearchAgent", description="Useful for searching the web for information on a given topic and recording notes on the topic.", system_prompt=( "You are the ResearchAgent that can search the web for information on a given topic and record notes on the topic. " "Once notes are recorded and you are satisfied, you should hand off control to the WriteAgent to write a report on the topic. " "You should have at least some notes on a topic before handing off control to the WriteAgent." ), llm=llm, tools=[search_web, record_notes], can_handoff_to=["WriteAgent"], ) write_agent = FunctionAgent( name="WriteAgent", description="Useful for writing a report on a given topic.", system_prompt=( "You are the WriteAgent that can write a report on a given topic. " "Your report should be in a markdown format. The content should be grounded in the research notes. " "Once the report is written, you should get feedback at least once from the ReviewAgent." ), llm=llm, tools=[write_report], can_handoff_to=["ReviewAgent", "ResearchAgent"], ) review_agent = FunctionAgent( name="ReviewAgent", description="Useful for reviewing a report and providing feedback.", system_prompt=( "You are the ReviewAgent that can review the write report and provide feedback. " "Your review should either approve the current report or request changes for the WriteAgent to implement. " "If you have feedback that requires changes, you should hand off control to the WriteAgent to implement the changes after submitting the review." ), llm=llm, tools=[review_report], can_handoff_to=["WriteAgent"], )

llm在示例中为from llama_index.llms.openai import OpenAI创建的OpenAI(model="gpt-4o", api_key="sk-...")实例（key 需替换）。文档同时说明：如果你的 LLM 支持 tool calling 就用FunctionAgent，否则用ReActAgent。

用 AgentWorkflow 连接并运行

接线只需声明 Agent 列表、哪个是入口（root_agent）和初始状态：

from llama_index.core.agent.workflow import AgentWorkflow agent_workflow = AgentWorkflow( agents=[research_agent, write_agent, review_agent], root_agent=research_agent.name, initial_state={ "research_notes": {}, "report_content": "Not written yet.", "review": "Review required.", }, )

AgentWorkflow的执行循环是：把用户消息交给 root Agent → 执行它选中的工具 → 允许它决定 handoff 给下一个 Agent → 循环直到有 Agent 返回最终答案。运行时可以消费事件流观察进度：

from llama_index.core.agent.workflow import ( AgentInput, AgentOutput, ToolCall, ToolCallResult, AgentStream, ) handler = agent_workflow.run( user_msg=( "Write me a report on the history of the internet. " "Briefly describe the history of the internet, including the development of the internet, the development of the web, " "and the development of the internet in the 21st century." ) ) current_agent = None async for event in handler.stream_events(): if ( hasattr(event, "current_agent_name") and event.current_agent_name != current_agent ): current_agent = event.current_agent_name print(f"\n{'='*50}") print(f"🤖 Agent: {current_agent}") print(f"{'='*50}\n") elif isinstance(event, AgentOutput): if event.response.content: print("📤 Output:", event.response.content) if event.tool_calls: print( "🛠️ Planning to use tools:", [call.tool_name for call in event.tool_calls], ) elif isinstance(event, ToolCallResult): print(f"🔧 Tool Result ({event.tool_name}):") print(f" Arguments: {event.tool_kwargs}") print(f" Output: {event.tool_output}") elif isinstance(event, ToolCall): print(f"🔨 Calling Tool: {event.tool_name}") print(f" With arguments: {event.tool_kwargs}")

验证运行结果

事件流中出现handoff工具调用，说明交接按can_handoff_to声明发生了。文档中的示例输出（已删节，仅用于说明流程）：

================================================== 🤖 Agent: ResearchAgent ================================================== 🛠️ Planning to use tools: ['search_web'] 🔨 Calling Tool: search_web With arguments: {'query': 'history of the internet'} 🛠️ Planning to use tools: ['handoff'] 🔨 Calling Tool: handoff With arguments: {'to_agent': 'WriteAgent', 'reason': 'I have gathered and recorded notes on the history of the internet ...'} ... 📤 Output: The report on the history of the internet has been reviewed and approved. ...

最终报告不是从对话里取的，而是存在state里，运行结束后这样取回：

state = await handler.ctx.store.get("state") print(state["report_content"])

需要集中决策点时：编排者模式（子 Agent 作为工具）

当你要"每一步都由一个地方决定"以便注入自定义逻辑时，用编排者示例。它与 AgentWorkflow 的差别在于：子 Agent 之间不互相 handoff，而是把每个子 Agent 的调用包装成工具，交给一个顶层编排者 Agent 调度——工具调用完总是回到编排者手里，控制流因此是集中的。

子 Agent 定义与前一节相同（示例中子 Agent 用gpt-4.1-mini，编排者单独用o3-mini，文档说明每个 Agent 可以用不同 LLM）。核心变化是这层工具包装，它负责调用子 Agent 并把结果写回共享状态：

import re from llama_index.core.workflow import Context async def call_research_agent(ctx: Context, prompt: str) -> str: """Useful for recording research notes based on a specific prompt.""" result = await research_agent.run( user_msg=f"Write some notes about the following: {prompt}" ) async with ctx.store.edit_state() as ctx_state: ctx_state["state"]["research_notes"].append(str(result)) return str(result) async def call_write_agent(ctx: Context) -> str: """Useful for writing a report based on the research notes or revising the report based on feedback.""" async with ctx.store.edit_state() as ctx_state: notes = ctx_state["state"].get("research_notes", None) if not notes: return "No research notes to write from." user_msg = f"Write a markdown report from the following notes. Be sure to output the report in the following format: <report>...</report>:\n\n" # Add the feedback to the user message if it exists feedback = ctx_state["state"].get("review", None) if feedback: user_msg += f"<feedback>{feedback}</feedback>\n\n" # Add the research notes to the user message notes = "\n\n".join(notes) user_msg += f"<research_notes>{notes}</research_notes>\n\n" # Run the write agent result = await write_agent.run(user_msg=user_msg) report = re.search( r"<report>(.*)</report>", str(result), re.DOTALL ).group(1) ctx_state["state"]["report_content"] = str(report) return str(report) async def call_review_agent(ctx: Context) -> str: """Useful for reviewing the report and providing feedback.""" async with ctx.store.edit_state() as ctx_state: report = ctx_state["state"].get("report_content", None) if not report: return "No report content to review." result = await review_agent.run( user_msg=f"Review the following report: {report}" ) ctx_state["state"]["review"] = result return result

然后用三个包装函数作为工具构建编排者：

orchestrator = FunctionAgent( system_prompt=( "You are an expert in the field of report writing. " "You are given a user request and a list of tools that can help with the request. " "You are to orchestrate the tools to research, write, and review a report on the given topic. " "Once the review is positive, you should notify the user that the report is ready to be accessed." ), llm=orchestrator_llm, tools=[ call_research_agent, call_write_agent, call_review_agent, ], initial_state={ "research_notes": [], "report_content": None, "review": None, }, )

注意这里的initial_state中research_notes是列表（与 AgentWorkflow 示例中的字典不同），因为call_research_agent用的是append。

运行时需要为编排者显式创建Context来承载历史和状态，然后消费事件流：

from llama_index.core.workflow import Context ctx = Context(orchestrator) async def run_orchestrator(ctx: Context, user_msg: str): handler = orchestrator.run( user_msg=user_msg, ctx=ctx, ) async for event in handler.stream_events(): if isinstance(event, AgentStream): if event.delta: print(event.delta, end="", flush=True) elif isinstance(event, AgentOutput): if event.tool_calls: print( "🛠️ Planning to use tools:", [call.tool_name for call in event.tool_calls], ) elif isinstance(event, ToolCallResult): print(f"🔧 Tool Result ({event.tool_name}):") print(f" Arguments: {event.tool_kwargs}") print(f" Output: {event.tool_output}") elif isinstance(event, ToolCall): print(f"🔨 Calling Tool: {event.tool_name}") print(f" With arguments: {event.tool_kwargs}") await run_orchestrator( ctx=ctx, user_msg=( "Write me a report on the history of the internet. " "Briefly describe the history of the internet, including the development of the internet, the development of the web, " "and the development of the internet in the 21st century." ), )

验证方式与前一节一致：事件流里应出现call_research_agent→call_write_agent→call_review_agent（以及根据评审反馈再次调用call_write_agent）的工具调用序列；最终报告从状态中取回：

state = await ctx.store.get("state") print(state["report_content"])

文档示例的输出中，编排者先研究、再写作、再评审，评审给出 "Approve with minor revisions" 后再次调用call_write_agent修订，最终在state["report_content"]中得到修订版报告——这段流程由编排者的工具选择决定，而不是子 Agent 之间的 handoff。

需要完全自写规划逻辑时：自定义规划器

当前两种模式都表达不了你需要的流程时，参考custom_multi_agent 示例。思路是：自己写提示词让 LLM 输出结构化计划（XML），再用 Python 代码解析并命令式地执行；子 Agent 可以是FunctionAgent、RAG 流水线或其他服务。这个示例还要求你先了解 Workflow 文档（multi_agent 文档链接的 workflows 章节）。

PlannerWorkflow 分两个@step：plan用提示词让 LLM 生成<plan>块，execute解析后逐步调用子 Agent，然后回到plan判断是否还需要更多步骤。计划格式由PLANNER_PROMPT定义（每个<step>指定要调用的 Agent 和消息）：

PLANNER_PROMPT = """You are a planner chatbot. Given a user request and the current state, break the solution into ordered <step> blocks. Each step must specify the agent to call and the message to send, e.g. <plan> <step agent=\"ResearchAgent\">search for …</step> <step agent=\"WriteAgent\">draft a report …</step> ... </plan> <state> {state} </state> <available_agents> {available_agents} </available_agents> The general flow should be: - Record research notes - Write a report - Review the report - Write the report again if the review is not positive enough If the user request does not require any steps, you can skip the <plan> block and respond directly. """

计划用 pydantic 建模并在plan步骤中解析（无<plan>块时直接作为最终回答返回）：

class PlanStep(BaseModel): agent_name: str agent_input: str class Plan(BaseModel): steps: list[PlanStep]

execute步骤遍历计划逐步调用包装函数（与编排者模式中的call_research_agent等相同），执行完把更新后的状态交回给规划器，询问是否需要继续规划：

@step async def execute(self, ctx: Context, ev: ExecuteEvent) -> InputEvent: chat_history = ev.chat_history plan = ev.plan for step in plan.steps: agent = self.agents[step.agent_name] agent_input = step.agent_input ctx.write_event_to_stream( PlanEvent( step_info=f'<step agent="{step.agent_name}">{step.agent_input}</step>' ), ) if step.agent_name == "ResearchAgent": await call_research_agent(ctx, agent_input) elif step.agent_name == "WriteAgent": # Note: we aren't passing the input from the plan since # we're using the state to drive the write agent await call_write_agent(ctx) elif step.agent_name == "ReviewAgent": await call_review_agent(ctx) state = await ctx.store.get("state") chat_history.append( ChatMessage( role="user", content=f"I've completed the previous steps, here's the updated state:\n\n<state>\n{state}\n</state>\n\nDo you need to continue and plan more steps?, If not, write a final response.", ), ) return InputEvent( chat_history=chat_history, )

PlannerWorkflow中规划器 LLM 在示例里是OpenAI(model="o3-mini", api_key="sk-proj-...")（key 为占位符，需替换）。运行方式：

planner_workflow = PlannerWorkflow(timeout=None) handler = planner_workflow.run( user_msg=( "Write me a report on the history of the internet. " "Briefly describe the history of the internet, including the development of the internet, the development of the web, " "and the development of the internet in the 21st century." ), chat_history=[], state={ "research_notes": [], "report_content": "Not written yet.", "review": "Review required.", }, ) async for event in handler.stream_events(): if isinstance(event, PlanEvent): print("Executing plan step: ", event.step_info) elif isinstance(event, ExecuteEvent): print("Executing plan: ", event.plan) result = await handler print(result.response)

验证点有三处：事件流中的PlanEvent会打印每一步计划（文档示例输出显示依次执行 ResearchAgent、WriteAgent、ReviewAgent、再 WriteAgent 修订四个步骤）；result.response是规划器给出的最终回答（文档示例为 "No further planning steps are needed. The report ... has been completed and reviewed..."）；最终报告和评审仍从状态取回：

state = await handler.ctx.store.get("state") print(state["report_content"]) print(state["review"])

边界与限制

• LLM 不支持 tool calling 时：三个示例 notebook 都注明此时把FunctionAgent换成ReActAgent（from llama_index.core.agent.workflow import FunctionAgent, ReActAgent）。
• AgentWorkflow 的控制权归返：文档明确，任何时刻当前活跃 Agent 都可以选择把控制权交还给用户。
• 自定义规划器只做顺序计划：示例 notebook 说明其提示词假设顺序执行，并行步骤"涉及更复杂的解析和提示词，留作读者练习"；顶层 streaming 也需要你自己实现（对比表中注明 "Top-level is up to you"）。
• 状态结构随模式而异：AgentWorkflow 示例的research_notes初始化为字典并按标题键值写入；编排者与自定义规划器示例初始化为列表并append。照抄代码时保持各自示例的结构，不要混用。

下一步

文档在三种模式之后指向 structured output in single and multi-agent workflows，用于在单/多 Agent 工作流中处理结构化输出。三种模式的完整代码分别见 agent_workflow_multi、agents_as_tools 与 custom_multi_agent 三个 notebook，可直接对照本文代码逐段运行。

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