LlamaIndex 多智能体模式全指南：AgentWorkflow、Orchestrator 与自定义 Planner 实战解析

LlamaIndex 多智能体模式全指南AgentWorkflow、Orchestrator 与自定义 Planner 实战解析【免费下载链接】llama_indexLlamaIndex is the document processing platform for AI项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index导读当一个任务需要多个专家协同完成时LlamaIndex 提供了三种多智能体协作方案内置的AgentWorkflow线性 swarm 模式、Orchestrator 代理模式子代理作为工具以及完全自研的自定义 PlannerDIY prompt 解析。本文基于 multi_agent.md 的核心脉络逐一剖析三种模式的适用场景、最小可用代码骨架并深入仓库源码验证其底层实现机制帮助你根据开发效率与控制灵活性的权衡做出正确选型。多智能体协作流程图何时需要多智能体三种模式的选型全景当单个通用代理无法高效完成复杂任务时多智能体协作成为必然选择。LlamaIndex 为此提供了三种模式它们在便利性与灵活性之间做出了不同的权衡模式代码量灵活性内置流式/事件AgentWorkflow内置⭐ 最少★★是Orchestrator 代理内置⭐⭐★★★是经由 orchestrator自定义 PlannerDIY⭐⭐⭐★★★★★是经由子代理顶层由你掌控核心选择逻辑快速原型优先AgentWorkflow当需要对执行顺序有更多控制时升级到 Orchestrator 代理只有前两种模式无法表达所需流程时才诉诸自定义 Planner。Pattern 1 – AgentWorkflow开箱即用的线性 swarm 模式适用场景与运行机制当你希望以近乎零额外代码获得多智能体行为且接受AgentWorkflow内置的默认交接hand-off启发式策略时选择此模式。AgentWorkflow本身是一个 Workflow事件驱动抽象被预先配置为能够理解 agents、state 与 tool-calling。你只需提供一个或多个 agent 组成的数组并指定哪个 agent 作为root_agent启动它便会自动执行将用户消息交给根rootagent执行该 agent 选择的所有工具允许 agent 在它认为合适时将控制权交接handoff给另一个 agent重复以上步骤直到某个 agent 返回最终答案。注意在任何时刻当前活跃 agent 都可以选择将控制权交还给用户。最小代码骨架报告生成三智能体协作以下代码是 agent_workflow_multi 示例 的浓缩版——三个 agent 协作完成研究 → 撰写 → 评审一份报告…表示为了简洁省略的代码from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent # --- create our specialist agents ------------------------------------------------ research_agent FunctionAgent( nameResearchAgent, descriptionSearch the web and record notes., system_promptYou are a researcher… hand off to WriteAgent when ready., llmllm, tools[search_web, record_notes], can_handoff_to[WriteAgent], ) write_agent FunctionAgent( nameWriteAgent, descriptionWrites a markdown report from the notes., system_promptYou are a writer… ask ReviewAgent for feedback when done., llmllm, tools[write_report], can_handoff_to[ReviewAgent, ResearchAgent], ) review_agent FunctionAgent( nameReviewAgent, descriptionReviews a report and gives feedback., system_promptYou are a reviewer…, # etc. llmllm, tools[review_report], can_handoff_to[WriteAgent], ) # --- wire them together ---------------------------------------------------------- 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., }, ) resp await agent_workflow.run( user_msgWrite me a report on the history of the web … ) print(resp)AgentWorkflow负责全部编排并在运行过程中持续发出流式事件你可以借此向用户实时展示进度。源码级解析交接handoff如何工作从源码结构看AgentWorkflow的核心实现位于 multi_agent_workflow.py其__init__接收agents、root_agent、initial_state、handoff_prompt、handoff_output_prompt、state_prompt、timeout、output_cls等参数并做了多层校验多 agent 必须命名len(agents) 1时任何 agent 若使用默认名AgentDEFAULT_AGENT_NAME或默认描述都会抛出ValueError见 multi_agent_workflow.py#L124-L134。单 agent 场景才允许默认值。root_agent 必须存在只有一个 agent 时自动设为agents[0].name多个 agent 时必须显式提供 root_agent且必须位于 agents 列表中multi_agent_workflow.py#L142-L150。handoff 工具是自动注入的_get_handoff_tool()为每个 agent 动态生成一个FunctionTool.from_defaults(async_fnhandoff, return_directTrue)其描述由handoff_prompt格式化而成包含可交接的 agent 信息can_handoff_to为None的 agent 可以交接给任意其他 agent为空列表则禁止交接multi_agent_workflow.py#L216-L267。状态与内存共享_init_context()在首次运行时把memory默认ChatMemoryBuffer、agents列表、can_handoff_to映射、initial_state的深拷贝、current_agent_name初始为 root_agent、max_iterations默认 20见 base_agent.py#L67写入ctx.storemulti_agent_workflow.py#L269-L312。交接发生时aggregate_tool_results读取next_agent并更新current_agent_name形成下一轮循环multi_agent_workflow.py#L709-L714。最大迭代与早停parse_agent_output中num_iterations max_iterations时early_stopping_methodforce会抛出WorkflowRuntimeError提示通过.run(..., max_iterations...)调高上限或改用generate生成最终响应multi_agent_workflow.py#L527-L547。AgentWorkflow还通过run()支持user_msg、chat_history、memory、max_iterations、early_stopping_method等参数multi_agent_workflow.py#L767-L848。若你只想用单 agent 工具AgentWorkflow.from_tools_or_functions()会根据 LLM 是否为函数调用模型自动选择FunctionAgent或ReActAgentmulti_agent_workflow.py#L850-L900。关键事件流workflow_events.py中定义了AgentInput、AgentSetup、AgentOutput、AgentStream、ToolCall、ToolCallResult、AgentStreamStructuredOutput等事件类型workflow_events.py#L24-L114完整支撑了上述 step 链。Pattern 2 – Orchestrator 代理子代理即工具适用场景当你希望由单一决策点决定每一步的执行便于注入自定义逻辑但又不想自己编写 planner而更偏好声明式的agent 作为工具体验时选择此模式。在此模式下你仍然构建专家 agentResearchAgent、WriteAgent、ReviewAgent但不再让它们相互交接。取而代之的是将每个 agent 的run方法暴露为工具把这些工具交给一个新的顶层 agent——即Orchestrator编排器。完整示例参见 agents_as_tools notebook。最小代码骨架包装 agent.run 为可调用工具import re from llama_index.core.agent.workflow import FunctionAgent from llama_index.core.workflow import Context # assume research_agent / write_agent / review_agent defined as before # except we really only need the search_web tool at a minimum 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, }, ) response await orchestrator.run( user_msgWrite me a report on the history of the web … ) print(response)源码级解析为什么编排器白拿全套能力因为 Orchestrator 本质上仍是一个FunctionAgentfunction_agent.py所以流式输出、工具调用与状态管理全部免费获得——而你依然完整掌控子 agent 的调用方式与整体控制流工具永远把结果返回给 orchestrator。从源码看FunctionAgent的几个关键行为函数调用型 LLM 强约束take_step()中if not self.llm.metadata.is_function_calling_model: raise ValueError(LLM must be a FunctionCallingLLM)即函数调用代理要求底层 LLM 支持函数调用function_agent.py#L101-L110。并行工具调用allow_parallel_tool_calls默认True一次可并行调用多个工具initial_tool_choice可强制首轮调用指定工具function_agent.py#L23-L30。基于 Context 的工具签名上述包装函数通过ctx: Context参数访问ctx.store.edit_state()读写共享状态这正是工具函数注入上下文requires_context/ctx_param_name能力的体现见 multi_agent_workflow.py#L348-L379 中_call_tool对带上下文工具的调用分支。流式事件_get_streaming_response()会把每个增量块封装为AgentStream事件包含delta、tool_calls、current_agent_name写入事件流function_agent.py#L52-L99。注意本模式中状态通过initial_state在 orchestrator 上声明三个包装函数间通过ctx.store共享research_notes、report_content、review字段——这与 Pattern 1 中把状态放在AgentWorkflow(initial_state...)上略有不同体现了状态归属的两种设计。Pattern 3 – 自定义 PlannerDIY 提示词 解析适用场景追求终极灵活性时选择此模式你需要强加一种非常具体的计划格式、对接外部调度器或采集前两种模式无法开箱即用地提供的额外元数据。思路核心你编写一个提示词指示 LLM 输出结构化计划XML / JSON / YAML你自己的 Python 代码解析该计划并命令式地执行它。底层子代理可以是任何东西——FunctionAgent、RAG 流水线或其他服务。最小代码骨架能规划、能执行、能迭代的 Workflow以下是一个最小草图——实现规划 → 执行计划 → 判断是否需要更多步骤的循环。完整示例见 custom_multi_agent notebook。import re import xml.etree.ElementTree as ET from pydantic import BaseModel, Field from typing import Any, Optional from llama_index.core.llms import ChatMessage from llama_index.core.workflow import ( Context, Event, StartEvent, StopEvent, Workflow, step, ) # Assume we created helper functions to call the agents 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 agentResearchAgentsearch for …/step step agentWriteAgentdraft 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. class InputEvent(StartEvent): user_msg: Optional[str] Field(defaultNone) chat_history: list[ChatMessage] state: Optional[dict[str, Any]] Field(defaultNone) class OutputEvent(StopEvent): response: str chat_history: list[ChatMessage] state: dict[str, Any] class StreamEvent(Event): delta: str class PlanEvent(Event): step_info: str # Modelling the plan class PlanStep(BaseModel): agent_name: str agent_input: str class Plan(BaseModel): steps: list[PlanStep] class ExecuteEvent(Event): plan: Plan chat_history: list[ChatMessage] class PlannerWorkflow(Workflow): llm: OpenAI OpenAI( modelo3-mini, api_keysk-proj-..., ) agents: dict[str, FunctionAgent] { ResearchAgent: research_agent, WriteAgent: write_agent, ReviewAgent: review_agent, } step async def plan( self, ctx: Context, ev: InputEvent ) - ExecuteEvent | OutputEvent: # Set initial state if it exists if ev.state: await ctx.store.set(state, ev.state) chat_history ev.chat_history if ev.user_msg: user_msg ChatMessage( roleuser, contentev.user_msg, ) chat_history.append(user_msg) # Inject the system prompt with state and available agents state await ctx.store.get(state) available_agents_str \n.join( [ fagent name{agent.name}{agent.description}/agent for agent in self.agents.values() ] ) system_prompt ChatMessage( rolesystem, contentPLANNER_PROMPT.format( statestr(state), available_agentsavailable_agents_str, ), ) # Stream the response from the llm response await self.llm.astream_chat( messages[system_prompt] chat_history, ) full_response async for chunk in response: full_response chunk.delta or if chunk.delta: ctx.write_event_to_stream( StreamEvent(deltachunk.delta), ) # Parse the response into a plan and decide whether to execute or output xml_match re.search(r(plan.*/plan), full_response, re.DOTALL) if not xml_match: chat_history.append( ChatMessage( roleassistant, contentfull_response, ) ) return OutputEvent( responsefull_response, chat_historychat_history, statestate, ) else: xml_str xml_match.group(1) root ET.fromstring(xml_str) plan Plan(steps[]) for step in root.findall(step): plan.steps.append( PlanStep( agent_namestep.attrib[agent], agent_inputstep.text.strip() if step.text else , ) ) return ExecuteEvent(planplan, chat_historychat_history) 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, )源码级解析事件驱动的循环骨架此模式完全建立在 LlamaIndex 的Workflow 抽象之上——这也是AgentWorkflow本身的底层AgentWorkflow继承自Workflow见 multi_agent_workflow.py#L99。核心机制step装饰器与事件驱动每个step装饰的协程接收一个Event并返回下一个事件Workflow 运行时根据事件类型路由执行。上例中planstep 返回ExecuteEvent | OutputEventexecutestep 返回InputEvent形成规划 → 执行 → 再规划的循环当 LLM 不再输出plan块时planstep 直接返回OutputEvent终止。ctx.store是跨 step 的状态中枢ctx.store.set / get贯穿整个流程ctx.write_event_to_stream()实现自定义流式事件StreamEvent、PlanEvent用于向前端汇报进度。Pydantic 建模计划Plan/PlanStep用 pydanticBaseModel建模解析结果保证类型安全。状态回填再规划execute完成后把最新state作为 user 消息追加到chat_history再次进入planstep 判断是否需要继续——这实现了多轮计划-执行的自适应循环。这种做法意味着编排循环完全由你掌控你可以插入任何自定义逻辑、缓存或人工介入human-in-the-loop检查。如何选择一张决策路线图Pattern代码量灵活性内置流式/事件AgentWorkflow⭐ – 最少★★是Orchestrator agent⭐⭐★★★是经由 orchestratorCustom planner⭐⭐⭐★★★★★是经由子代理。顶层由你决定实战建议快速原型直接使用AgentWorkflow声明式描述 agents 与can_handoff_to关系即可跑通多智能体协作需要控制执行序列迁移到 Orchestrator 代理模式把子 agent 包装为带ctx的工具获得单一决策 声明式子代理的平衡流程无法用前两者表达强约束计划格式、外部调度器、额外元数据、人工审批节点投入自定义 Planner用step 事件循环构建专属编排。进一步探索接下来可学习如何在单个及多智能体工作流中使用结构化输出structured output为多智能体协作产出强类型、可校验的结果。【免费下载链接】llama_indexLlamaIndex is the document processing platform for AI项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考