当一个任务需要多个专家协作时，LlamaIndex 给了你好几种选择，各有各的取舍：便利换灵活。本页会走一遍最常见的三种模式，说明各自适合什么时候用，并为每种方式给出一个最小代码骨架。

1. AgentWorkflow（内置） —— 声明一组 agent，让 AgentWorkflow 负责它们之间的交接。本节 完整 Notebook
2. 编排者模式（内置） —— 由一个“编排者”agent 决定下一步调用哪个子 agent，这些子 agent 以工具的形式暴露给它。本节 完整 Notebook
3. 自定义规划器（DIY） —— 由你自己编写让 LLM 规划执行顺序的提示词（通常是 XML / JSON），再在代码里以命令式的方式依次调用各个 agent。本节 完整 Notebook

---

模式一 —— AgentWorkflow（即线性“swarm”模式）

何时使用 —— 你希望几乎不写额外代码就获得多 agent 行为，并且能接受 AgentWorkflow 自带的默认交接策略。

AgentWorkflow 本身就是一个已配置好、懂得 agent、状态与工具调用的 Workflow。你提供一个由一个或多个 agent 组成的数组，告诉它该由哪个 agent 起手，它就会：

1. 把用户消息交给根 agent。
2. 执行该 agent 选中的工具。
3. 允许 agent 在它认为合适的时候把控制权“交接”给另一个 agent。
4. 不断重复，直到某个 agent 给出最终回答。

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

下面是多 agent 报告生成示例的精简版本。三个 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 负责全部的编排工作，并在运行过程中不断发出事件，让你能实时向用户汇报进展。

---

模式二 —— 编排者 agent（把子 agent 当工具）

何时使用 —— 你想要有一个统一的地方来决定每一个步骤，以便插入自己的逻辑，同时又不想自己写规划器，仍然偏好声明式的agent 即工具体验。

在这种模式下，你照样构建各领域专家 agent（ResearchAgent、WriteAgent、ReviewAgent），但并不让它们互相交接。相反，你把每个 agent 的 run 方法包装成工具，把这些工具交给一个新的顶层 agent —— 编排者。

完整示例见 agents_as_tools notebook。

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)

由于编排者不过就是另一个 FunctionAgent，你等于免费获得了流式输出、工具调用与状态管理——同时对 agent 的调用方式和整体控制流保有完全的控制权（工具总会把结果返回给编排者）。

---

模式三 —— 自定义规划器（自己写提示词 + 解析）

何时使用 —— 追求极致灵活。你需要强制采用某种特定的规划格式、对接外部调度器，或者收集前两种模式开箱不提供额外元数据。

思路是：你写一段提示词，指示 LLM 输出一个结构化计划（XML / JSON / YAML），然后由你自己的 Python 代码解析这个计划并命令式地执行它。下属的 agent 可以是任何东西 —— FunctionAgent、RAG 流水线或其他服务。

下面是一个极简骨架，演示一个能够制定计划、执行计划并判断是否还需要后续步骤的工作流。完整示例见 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,
        )

这种方式意味着编排循环由你掌握，所以你可以随意插入自己需要的自定义逻辑、缓存或在环人工确认。

---

如何选择模式

模式	代码行数	灵活度	内置流式 / 事件
AgentWorkflow	⭐ —— 最少	★★	有
编排者 agent	⭐⭐	★★★	有（经由编排者）
自定义规划器	⭐⭐⭐	★★★★★	有（经由子 agent）。顶层由你自己决定。

如果你要快速做原型，就从 AgentWorkflow 开始。当你需要对执行顺序有更多控制时，改用编排者 agent。只有当前两种模式无法表达你需要的流程时，才去用自定义规划器。

接下来你会学习如何在单 agent 与多 agent 工作流中使用结构化输出