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、对接外部调度器或获取前两种模式无法直接提供的额外元数据。文档同时给出一张对比表模式代码量灵活性内置 streaming / eventsAgentWorkflow最少★★Yes编排者 Agent★★★★★Yes经由编排者自定义规划器最多★★★★★子 Agent 有顶层由你自己实现文档给出的选型顺序是快速原型先用AgentWorkflow需要更多控制执行顺序时移到编排者模式只有当前两者都无法表达你需要的流程时才上自定义规划器。三种模式的前置安装相同示例 notebook 中写作%pip install ...脚本环境去掉%即可pip install llama-index pip install tavily-python后续所有代码中的sk-...、sk-proj-...是文档示例里的占位 API key替换为你自己的 OpenAI API keytvly-...替换为你自己的 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_keytvly-...) # 替换为你的 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( nameResearchAgent, descriptionUseful 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. ), llmllm, tools[search_web, record_notes], can_handoff_to[WriteAgent], ) write_agent FunctionAgent( nameWriteAgent, descriptionUseful 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. ), llmllm, tools[write_report], can_handoff_to[ReviewAgent, ResearchAgent], ) review_agent FunctionAgent( nameReviewAgent, descriptionUseful 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. ), llmllm, tools[review_report], can_handoff_to[WriteAgent], )llm在示例中为from llama_index.llms.openai import OpenAI创建的OpenAI(modelgpt-4o, api_keysk-...)实例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_agentresearch_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_msgfWrite 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 fWrite 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 ffeedback{feedback}/feedback\n\n # Add the research notes to the user message notes \n\n.join(notes) user_msg fresearch_notes{notes}/research_notes\n\n # Run the write agent result await write_agent.run(user_msguser_msg) report re.search( rreport(.*)/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_msgfReview 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. ), llmorchestrator_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_msguser_msg, ctxctx, ) async for event in handler.stream_events(): if isinstance(event, AgentStream): if event.delta: print(event.delta, end, flushTrue) 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( ctxctx, 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 分两个stepplan用提示词让 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_infofstep 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 arent passing the input from the plan since # were 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( roleuser, contentfIve completed the previous steps, heres the updated state:\n\nstate\n{state}\n/state\n\nDo you need to continue and plan more steps?, If not, write a final response., ), ) return InputEvent( chat_historychat_history, )PlannerWorkflow中规划器 LLM 在示例里是OpenAI(modelo3-mini, api_keysk-proj-...)key 为占位符需替换。运行方式planner_workflow PlannerWorkflow(timeoutNone) 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换成ReActAgentfrom 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考