企业官网建设流程全解析

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 代理模式（子代理作为工具），以及完全自研的自定义 Planner（DIY prompt + 解析）。本文基于 multi_agent.md 的核心脉络，逐一剖析三种模式的适用场景、最小可用代码骨架，并深入仓库源码验证其底层实现机制，帮助你根据"开发效率"与"控制灵活性"的权衡做出正确选型。

多智能体协作流程图

何时需要多智能体：三种模式的选型全景

当单个通用代理无法高效完成复杂任务时，多智能体协作成为必然选择。LlamaIndex 为此提供了三种模式，它们在"便利性"与"灵活性"之间做出了不同的权衡：

核心选择逻辑：快速原型优先AgentWorkflow；当需要对执行顺序有更多控制时，升级到 Orchestrator 代理；只有前两种模式无法表达所需流程时，才诉诸自定义 Planner。

Pattern 1 – AgentWorkflow：开箱即用的线性 swarm 模式

适用场景与运行机制

当你希望以近乎零额外代码获得多智能体行为，且接受AgentWorkflow内置的默认交接（hand-off）启发式策略时，选择此模式。

AgentWorkflow本身是一个 Workflow（事件驱动抽象），被预先配置为能够理解 agents、state 与 tool-calling。你只需提供一个或多个 agent 组成的数组，并指定哪个 agent 作为root_agent启动，它便会自动执行：

1. 将用户消息交给根（root）agent；
2. 执行该 agent 选择的所有工具；
3. 允许 agent 在它认为合适时将控制权交接（handoff）给另一个 agent；
4. 重复以上步骤，直到某个 agent 返回最终答案。

注意：在任何时刻，当前活跃 agent 都可以选择将控制权交还给用户。

最小代码骨架：报告生成三智能体协作

以下代码是 agent_workflow_multi 示例 的浓缩版——三个 agent 协作完成"研究 → 撰写 → 评审"一份报告（…表示为了简洁省略的代码）：

from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent # --- create our specialist agents ------------------------------------------------ research_agent = FunctionAgent( name="ResearchAgent", description="Search the web and record notes.", system_prompt="You are a researcher… hand off to WriteAgent when ready.", llm=llm, tools=[search_web, record_notes], can_handoff_to=["WriteAgent"], ) write_agent = FunctionAgent( name="WriteAgent", description="Writes a markdown report from the notes.", system_prompt="You are a writer… ask ReviewAgent for feedback when done.", llm=llm, tools=[write_report], can_handoff_to=["ReviewAgent", "ResearchAgent"], ) review_agent = FunctionAgent( name="ReviewAgent", description="Reviews a report and gives feedback.", system_prompt="You are a reviewer…", # etc. llm=llm, tools=[review_report], can_handoff_to=["WriteAgent"], ) # --- wire them together ---------------------------------------------------------- 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.", }, ) resp = await agent_workflow.run( user_msg="Write 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 若使用默认名"Agent"（DEFAULT_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_fn=handoff, return_direct=True)，其描述由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.store（multi_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_method="force"会抛出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或ReActAgent（multi_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 作为工具"体验时，选择此模式。

在此模式下，你仍然构建专家 agent（ResearchAgent、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_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, }, ) response = await orchestrator.run( user_msg="Write me a report on the history of the web …" ) print(response)

源码级解析：为什么编排器"白拿"全套能力

因为 Orchestrator 本质上仍是一个FunctionAgent（function_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 – 自定义 Planner：DIY 提示词 + 解析

适用场景

追求终极灵活性时选择此模式：你需要强加一种非常具体的计划格式、对接外部调度器，或采集前两种模式无法开箱即用地提供的额外元数据。

思路核心：你编写一个提示词，指示 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 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. """ class InputEvent(StartEvent): user_msg: Optional[str] = Field(default=None) chat_history: list[ChatMessage] state: Optional[dict[str, Any]] = Field(default=None) 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( model="o3-mini", api_key="sk-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( role="user", content=ev.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( [ f'<agent name="{agent.name}">{agent.description}</agent>' for agent in self.agents.values() ] ) system_prompt = ChatMessage( role="system", content=PLANNER_PROMPT.format( state=str(state), available_agents=available_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(delta=chunk.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( role="assistant", content=full_response, ) ) return OutputEvent( response=full_response, chat_history=chat_history, state=state, ) 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_name=step.attrib["agent"], agent_input=step.text.strip() if step.text else "", ) ) return ExecuteEvent(plan=plan, chat_history=chat_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_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, )

源码级解析：事件驱动的循环骨架

此模式完全建立在 LlamaIndex 的Workflow 抽象之上——这也是AgentWorkflow本身的底层（AgentWorkflow继承自Workflow，见 multi_agent_workflow.py#L99）。核心机制：

• @step装饰器与事件驱动：每个@step装饰的协程接收一个Event并返回下一个事件，Workflow 运行时根据事件类型路由执行。上例中planstep 返回ExecuteEvent | OutputEvent，executestep 返回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）检查。

如何选择：一张决策路线图

实战建议：

• 快速原型：直接使用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