Agent 工程体系 · 第 51/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

LlamaIndex Agent：Workflow、Tool、Context、RAG 和多 Agent

LlamaIndex 中的 Agent，不只是“让大模型调用函数”。它更准确地表示一种由大语言模型（LLM）、工具（Tool）、记忆或状态（Memory/State）共同组成的半自主系统：系统接收用户目标，决定下一步动作，执行工具，把结果重新放回上下文，再判断是否继续，直到返回最终结果。LlamaIndex 官方文档将 Agent 描述为由 LLM、memory 和 tools 驱动、能够处理外部输入的系统；而 agentic 则是更宽泛的形容词，表示流程中包含 LLM 决策。(docs.llamaindex.ai)

这一区分很重要：

• Workflow 负责控制流程、事件、分支、循环和并发；
• Tool 负责把外部能力暴露给模型；
• Context 负责保存一次运行或多次运行之间的状态；
• RAG 负责把外部知识检索并注入推理过程；
• 多 Agent 负责把复杂任务拆给多个具有不同职责的 Agent。

如果把这些概念混在一起，最终系统通常会表现为：所有逻辑都埋在一个长 Prompt 中，工具没有权限边界，检索失败时仍然生成答案，多 Agent 只是多个模型轮流聊天，却没有明确的状态和终止条件。

---

一、先建立统一模型：Agent 是一个受约束的决策循环

设用户输入为 $x$，当前状态为 $s_t$，可用工具集合为 $T$，当前轮次为 $t$。Agent 在第 $t$ 轮执行以下过程：

$$a_t = \pi_{\theta}(x, s_t, T)$$

其中：

• $a_t$ 是模型选择的动作；
• $\pi_{\theta}$ 是由 Prompt、工具定义和模型共同决定的策略；
• $x$ 是用户目标；
• $s_t$ 是当前上下文，包括历史消息、工具结果、检索文档和业务状态；
• $T$ 是当前 Agent 被允许使用的工具集合。

动作有两类：

$$a_t \in \{\text{FinalAnswer}, \text{ToolCall}(f, args), \text{Handoff}(agent)\}$$

如果动作是工具调用，则执行：

$$o_t = f(args)$$

然后更新状态：

$$s_{t+1} = U(s_t, a_t, o_t)$$

如果模型认为任务完成，输出最终结果；如果选择另一个 Agent，则把控制权和必要状态交给目标 Agent。

这个模型说明了三个常见误解。

第一，Agent 不等于一次 LLM 调用。一次调用只能产生一个决策或答案；Agent 需要在工具结果产生之后重新决策。LlamaIndex 的 FunctionAgent 典型循环正是：读取最新消息和历史记录，发送工具 Schema 与对话历史，让模型直接回答或产生工具调用，执行调用，把结果加入历史，再次调用模型。(docs.llamaindex.ai)

第二，Tool 不是普通函数的别名。普通函数由程序员决定何时调用；Tool 则由模型在给定 Schema、描述和权限范围内选择调用。

第三，Agent 不能天然保证正确性。它只是在工具集合和上下文范围内执行一个概率策略。可靠性必须来自工具约束、状态约束、检索证据、终止条件和评测。

---

二、Workflow：把 Agent 的不确定决策放进确定的程序骨架

2.1 Workflow 的基本语义

LlamaIndex Workflow 是一种事件驱动、基于 Step 的流程模型。一个 Step 接收某种类型的 Event，执行普通 Python 逻辑，然后返回另一个 Event；返回事件的类型决定后续由哪个 Step 接收。分支可以写成普通 if，循环可以通过返回之前的事件类型实现，并发则可以产生一批事件。(developers.llamaindex.ai)

可以把 Workflow 看成一个带类型的状态转移系统：

$$E_i \xrightarrow{\text{step}} E_j$$

其中 $E_i$ 和 $E_j$ 是事件类型，Step 是边上的执行逻辑。

flowchart LR
    S[StartEvent] --> P[plan]
    P -->|需要检索| R[retrieve]
    P -->|无需检索| F[finalize]
    R --> J[judge]
    J -->|证据不足且未超限| P
    J -->|证据充分| F
    J -->|达到预算| F
    F --> Z[StopEvent]

图中的关键点不是“有几个节点”，而是：

1. plan 负责决定下一步；
2. retrieve 只负责检索，不负责最终回答；
3. judge 判断证据是否足够；
4. 循环必须有预算或停止条件；
5. finalize 是唯一的终止路径之一。

这比把“先检索、再判断、必要时重试”全部塞进一个 Agent Prompt 更容易测试，因为每一步的输入和输出都有明确类型。

2.2 一个最小可运行 Workflow

下面的代码展示一个生成—审查—结束流程。它需要：

pip install llama-index-workflows llama-index-llms-openai
export OPENAI_API_KEY="你的密钥"

llama-index-core 也包含 Workflow，可以使用 llama_index.core.workflow 导入；独立安装则可以使用 llama-index-workflows 提供的模块。官方示例说明了这两种安装方式。(developers.llamaindex.ai)

import asyncio

from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent
from llama_index.llms.openai import OpenAI


class DraftEvent(Event):
    text: str


class ReviewEvent(Event):
    text: str
    passed: bool
    feedback: str


class ArticleWorkflow(Workflow):
    llm = OpenAI(model="gpt-4.1")

    @step
    async def draft(self, ev: StartEvent) -> DraftEvent:
        topic = ev.topic

        response = await self.llm.acomplete(
            f"请用中文写一段关于“{topic}”的技术说明，要求准确、简洁。"
        )

        return DraftEvent(text=str(response))

    @step
    async def review(self, ev: DraftEvent) -> ReviewEvent:
        response = await self.llm.acomplete(
            "判断下面内容是否存在明显事实错误。"
            "只输出 PASS 或 FAIL，后面给出一句理由。\n\n"
            f"{ev.text}"
        )

        result = str(response)
        passed = result.upper().startswith("PASS")

        return ReviewEvent(
            text=ev.text,
            passed=passed,
            feedback=result,
        )

    @step
    async def finish(self, ev: ReviewEvent) -> StopEvent:
        if ev.passed:
            result = ev.text
        else:
            result = f"草稿未通过审查：{ev.feedback}\n\n原文：{ev.text}"

        return StopEvent(result=result)


async def main():
    workflow = ArticleWorkflow(timeout=60, verbose=True)
    result = await workflow.run(topic="向量数据库中的余弦相似度")
    print(str(result))


if __name__ == "__main__":
    asyncio.run(main())

执行过程是：

1. run(topic=...) 自动构造起始事件；
2. draft 读取 ev.topic，调用 LLM，返回 DraftEvent；
3. 框架根据返回类型找到 review；
4. review 产生 ReviewEvent；
5. finish 返回 StopEvent，Workflow 结束。

@step 会根据输入和输出类型推断事件连接，并在运行前检查是否存在未消费事件、没有生产者的输入事件或不可达终点。官方文档明确建议优先使用带类型的 Step；只有在事件数量未知、需要动态发送或外部注入事件时，才使用 Context 的动态 API。(developers.llamaindex.ai)

2.3 Workflow 和 Agent 的关系

FunctionAgent 是一个预构建的 Agent Workflow，使用模型供应商提供的 function/tool calling 能力。ReActAgent 使用 ReAct 风格的提示策略，CodeActAgent 则使用代码执行策略。它们都是 Agent，但决策机制不同。(docs.llamaindex.ai)

因此：

• 用 FunctionAgent，你是在使用一个已经实现好的“决策—工具—再决策”循环；
• 继承 Workflow，你是在自己控制事件、状态、并发、循环和失败路径；
• AgentWorkflow 则是在 Workflow 之上进一步提供多 Agent 协作。

一个实用判断是：如果流程的结构比模型的自由度更重要，用 Workflow；如果工具选择本身就是任务的一部分，用 Agent；如果两者都需要，就把 Agent 放进 Workflow 的某个 Step 中。

---

三、Tool：模型可调用能力的边界，而不是“给函数加个装饰器”

3.1 Tool 的组成

一个 Tool 至少包含：

$$\text{Tool} = (\text{name}, \text{description}, \text{input schema}, \text{executor}, \text{policy})$$

其中：

• name：模型调用时使用的名称；
• description：模型判断是否调用它的重要依据；
• input schema：参数名称、类型和约束；
• executor：真正执行操作的代码；
• policy：权限、超时、幂等性、审计和错误处理。

LlamaIndex 支持直接使用 Python 函数，也支持通过 FunctionTool、QueryEngineTool 等类进行进一步定制，还提供针对常见 API 的 Tool Specs。(docs.llamaindex.ai)

3.2 普通函数作为 Tool

from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI


def multiply(a: float, b: float) -> float:
    """计算两个数的乘积。"""
    return a * b


agent = FunctionAgent(
    tools=[multiply],
    llm=OpenAI(model="gpt-4o-mini"),
    system_prompt="你是一个计算助手。需要乘法时调用工具。",
)


async def main():
    response = await agent.run("1234 乘以 4567 是多少？")
    print(str(response))

这里的 Docstring 不是装饰性的注释。它通常会进入工具描述，直接影响模型是否选择该 Tool。一个含糊的描述，例如“处理数据”，无法告诉模型：

• 什么时候应该调用；
• 参数代表什么；
• 是否会修改数据；
• 失败时返回什么；
• 是否支持批量输入。

更好的 Tool 描述应当包含业务语义和边界：

def get_order_status(order_id: str) -> str:
    """
    查询订单当前状态。

    仅用于读取订单状态，不会修改订单。
    order_id 必须是系统中的订单编号，例如 ORD-20260901-001。
    如果订单不存在，返回 NOT_FOUND。
    """
    ...

3.3 Tool 的错误必须变成可推理的结果

如果 Tool 直接抛出未经处理的异常，模型通常只能看到“工具失败”，无法判断下一步该重试、换参数还是向用户澄清。

更稳妥的方式是把可预期错误编码为结构化结果：

from pydantic import BaseModel
from typing import Literal


class OrderResult(BaseModel):
    status: Literal["OK", "NOT_FOUND", "INVALID_ARGUMENT", "TEMPORARY_ERROR"]
    order_id: str
    detail: str


def get_order_status(order_id: str) -> OrderResult:
    if not order_id.startswith("ORD-"):
        return OrderResult(
            status="INVALID_ARGUMENT",
            order_id=order_id,
            detail="订单号必须以 ORD- 开头",
        )

    # 实际系统中这里应调用数据库或订单服务
    if order_id == "ORD-404":
        return OrderResult(
            status="NOT_FOUND",
            order_id=order_id,
            detail="未找到订单",
        )

    return OrderResult(
        status="OK",
        order_id=order_id,
        detail="已发货",
    )

这样 Agent 能够区分：

• 参数错误：修正参数；
• 资源不存在：不要无限重试；
• 临时错误：在预算允许时重试；
• 成功：继续回答。

Tool 的返回值是 Agent 的观测值，而不是普通函数的内部实现细节。 这个返回值是否结构化，决定了后续决策是否稳定。

3.4 Query Engine 也是 Tool

RAG 系统中的 Query Engine 可以被封装为 Tool。此时 Agent 不再直接看到向量库、Embedding 或节点，而是看到一个带名称和描述的知识查询能力。

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.tools import QueryEngineTool, ToolMetadata


documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine(
    similarity_top_k=5,
)

knowledge_tool = QueryEngineTool(
    query_engine=query_engine,
    metadata=ToolMetadata(
        name="company_knowledge",
        description=(
            "查询公司内部技术文档。"
            "适用于产品架构、部署流程、API 约定和故障排查。"
            "如果文档中没有答案，不要自行编造。"
        ),
    ),
)

agent = FunctionAgent(
    tools=[knowledge_tool],
    llm=OpenAI(model="gpt-4o-mini"),
    system_prompt=(
        "回答公司内部问题时优先使用 company_knowledge。"
        "无法从检索结果确认时，明确说明证据不足。"
    ),
)

这段代码的前置条件是 ./data 中存在可读取的文档，并且当前环境已经配置 LLM。它展示的是“RAG 作为 Tool”的组合方式，而不是让 Agent 直接管理所有检索细节。

---

四、Context：运行状态、共享状态和记忆不是同一个概念

4.1 三种容易混淆的状态

在 Agent 系统中，至少要区分三种东西：

对话历史

表示模型已经看到的消息序列：

user -> assistant -> tool -> tool_result -> assistant

它用于恢复语言层面的上下文，但不一定适合保存结构化业务状态。

Workflow Context

Context 是一次 Workflow 运行及其跨运行状态的容器，可以保存状态、传递事件、发送动态事件，也可以序列化后恢复。官方文档说明，AgentWorkflow 默认在不同 run() 之间无状态；若要跨运行保留状态，需要复用同一个 Context。(developers.llamaindex.ai)

业务状态

例如：

{
    "tenant_id": "acme",
    "user_id": "u-100",
    "retrieved_doc_ids": ["doc-1", "doc-7"],
    "query_attempts": 2,
    "approval_required": True
}

业务状态不应仅通过聊天消息表达，否则模型可能修改、遗漏或误解它。需要改变业务状态的操作，应由受控代码完成。

4.2 复用 Context 实现跨轮次记忆

from llama_index.core.workflow import Context

workflow = AgentWorkflow.from_tools_or_functions(
    [set_name],
    llm=llm,
    system_prompt="你可以记录和查询用户姓名。",
    initial_state={"name": "unset"},
)

ctx = Context(workflow)

await workflow.run(
    user_msg="我叫 Laurie。",
    ctx=ctx,
)

response = await workflow.run(
    user_msg="我叫什么？",
    ctx=ctx,
)

print(str(response))

同一个 ctx 被传给两次 run()，因此第二次运行可以读取第一次运行产生的状态。若每次都新建 Context，这段记忆就不会自然延续。

工具可以通过把 Context 放在第一个参数位置访问状态：

from llama_index.core.workflow import Context


async def set_name(ctx: Context, name: str) -> str:
    async with ctx.store.edit_state() as ctx_state:
        ctx_state["state"]["name"] = name

    return f"姓名已设置为 {name}"

这里应注意并发问题。若多个并行 Tool 同时修改同一状态字段：

state["score"] += 1

就可能出现丢失更新。应使用状态存储提供的编辑机制、按任务划分命名空间，或让并行任务返回结果，最后由一个聚合 Step 统一写入。

4.3 Context 的序列化边界

官方文档说明，Context 可以通过 JsonSerializer 或 JsonPickleSerializer 序列化和恢复。JSON 适合字符串、数字、列表和字典等普通数据；Pickle 可以处理更多 Python 对象，但不应反序列化不可信来源。(developers.llamaindex.ai)

from llama_index.core.workflow import (
    Context,
    JsonSerializer,
)

ctx_dict = ctx.to_dict(serializer=JsonSerializer())

# 保存 ctx_dict 到数据库或文件之后
restored_ctx = Context.from_dict(
    workflow,
    ctx_dict,
    serializer=JsonSerializer(),
)

不要把以下对象直接塞入可持久化状态：

• 数据库连接；
• HTTP 客户端；
• 向量数据库连接；
• 未定义序列化协议的模型对象；
• 大量原始文档内容；
• 密钥和访问令牌。

这些对象应通过依赖注入、资源对象或服务容器重新创建。Context 应保存“恢复业务流程所需的状态”，而不是保存整个运行时。

---

五、RAG：从“检索后生成”到 Agentic RAG

5.1 RAG 的基本链路

RAG（Retrieval-Augmented Generation，检索增强生成）把外部知识作为生成模型的输入：

$$D_q = \operatorname{Retrieve}(q, D)$$

$$y = \operatorname{Generate}(q, D_q)$$

其中：

• $q$ 是用户查询；
• $D$ 是知识库；
• $D_q$ 是检索到的文档片段；
• $y$ 是最终回答。

传统 RAG 通常由固定流程组成：

用户问题
  -> 向量检索
  -> 取 Top-K 节点
  -> 拼接上下文
  -> LLM 生成答案

它的问题不是“向量检索一定不好”，而是流程固定。例如：

• 简单问题不需要检索；
• 一个复杂问题可能需要拆成多个子问题；
• 向量召回结果可能需要关键词检索补充；
• 召回结果可能相关但排序不佳；
• 第一次检索没有足够证据时，需要改写查询；
• 外部知识源暂时失败时，需要回退到已有证据或人工处理。

Agentic RAG 将“检索是否需要、检索什么、是否继续检索”也变成决策问题：

$$\text{RetrieveDecision} = \pi(q, s_t, E_t)$$

其中 $E_t$ 是当前证据集合。

5.2 检索决策

Agent 不应对所有问题无条件调用 RAG Tool。可以定义一个最小决策函数：

$$d(q)= \begin{cases} \text{direct}, & q \text{ 是通用推理或闲聊问题}\\ \text{retrieve}, & q \text{ 依赖私有、时效或专门知识}\\ \text{clarify}, & q \text{ 缺少必要约束} \end{cases}$$

例如：

• “2 的 10 次方是多少？”通常可以直接计算；
• “公司 2026 年 8 月的发布流程是什么？”需要内部文档；
• “帮我查订单”缺少订单号，应该澄清。

这不是让模型凭感觉决定，而是可以通过工具描述、路由器或 Workflow Step 约束。

5.3 查询分解

对于合取问题：

“比较 A 和 B 的部署方式、成本约束以及故障恢复流程。”

单次检索可能只召回“部署方式”，遗漏成本和恢复。可以将问题分解为：

$$q \rightarrow \{q_1, q_2, \ldots, q_n\}$$

例如：

q1 = A 和 B 的部署方式分别是什么？
q2 = A 和 B 的成本约束分别是什么？
q3 = A 和 B 的故障恢复流程有什么差异？

然后并行检索：

flowchart TD
    Q[原始问题] --> P[查询规划]
    P --> Q1[子查询 1：部署]
    P --> Q2[子查询 2：成本]
    P --> Q3[子查询 3：恢复]
    Q1 --> R1[检索]
    Q2 --> R2[检索]
    Q3 --> R3[检索]
    R1 --> M[结果合并]
    R2 --> M
    R3 --> M
    M --> V[证据校验]
    V --> A[生成答案]

并行并不意味着可以无条件提高性能。并发检索会增加：

• 向量库 QPS；
• LLM 查询改写成本；
• 结果合并复杂度；
• 单个子查询失败的处理路径。

如果某个子查询失败，聚合器必须知道它是“无结果”还是“执行失败”：

class RetrievalResult(BaseModel):
    subquery: str
    status: Literal["OK", "EMPTY", "ERROR"]
    nodes: list[str]
    error: str | None = None

将错误和空结果混为一谈，会导致 Agent 错误地认为“知识库没有答案”。

5.4 重排

初始召回通常追求召回率，返回较大的候选集 $C$。重排器再根据查询与文档的精细相关性排序：

$$C = \operatorname{Recall}(q, D, k_0)$$

$$R = \operatorname{Rerank}(q, C, k_1)$$

其中通常 $k_0 > k_1$。

例如先召回 20 个节点，再重排取前 5 个。重排的目标不是增加知识，而是降低上下文中的噪声。

一个重要边界是：重排分数不是答案正确率。分数高只说明候选节点在模型看来更相关，不代表节点内容真实、完整或没有冲突。因此还需要：

1. 检查节点是否真正回答了子问题；
2. 检查多个节点之间是否存在冲突；
3. 保留来源信息；
4. 对无法支持的结论拒答或降级。

5.5 迭代检索与停止条件

Agentic RAG 最容易失控的地方是循环：

检索 -> 判断不足 -> 改写 -> 检索 -> 判断不足 -> ...

必须显式定义停止条件。可以使用以下判定：

$$\operatorname{Stop}(t, E_t)= \begin{cases} 1, & \operatorname{Coverage}(E_t,q)\ge \tau\\ 1, & t \ge T_{\max}\\ 1, & \Delta(E_t,E_{t-1}) < \epsilon\\ 0, & \text{otherwise} \end{cases}$$

变量含义：

• $t$：当前检索轮次；
• $T_{\max}$：最大轮次；
• $\operatorname{Coverage}$：证据覆盖问题约束的程度；
• $\tau$：最低覆盖阈值；
• $\Delta$：新一轮证据相对于上一轮的增量；
• $\epsilon$：收益过低阈值。

直觉是：当证据已经足够、预算耗尽或继续检索不再带来新信息时，都必须停止。

一个更接近生产系统的状态可以写成：

state = {
    "original_query": "...",
    "subqueries": [],
    "evidence": [],
    "attempt": 0,
    "max_attempts": 3,
    "seen_doc_ids": set(),
    "coverage": 0.0,
}

每次迭代必须更新 attempt，并记录已经检索过的文档。否则查询改写可能反复召回同一批节点，模型却误以为自己在探索新证据。

---

六、把 Agentic RAG 写成 Workflow

下面给出一个简化的事件设计。它重点展示状态和控制流，检索实现可以替换为 LlamaIndex Query Engine、混合检索器或外部搜索服务。

from pydantic import Field
from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent


class PlanEvent(Event):
    queries: list[str] = Field(default_factory=list)


class EvidenceEvent(Event):
    query: str
    passages: list[str] = Field(default_factory=list)
    attempt: int


class JudgeEvent(Event):
    query: str
    evidence: list[str]
    sufficient: bool
    attempt: int


class AgenticRAG(Workflow):
    def __init__(self, retriever, llm, **kwargs):
        super().__init__(**kwargs)
        self.retriever = retriever
        self.llm = llm

    @step
    async def plan(self, ev: StartEvent | JudgeEvent) -> PlanEvent | StopEvent:
        if isinstance(ev, JudgeEvent):
            if ev.sufficient or ev.attempt >= 3:
                answer = await self.llm.acomplete(
                    "仅依据以下证据回答问题；证据不足时明确说明。\n\n"
                    f"问题：{ev.query}\n"
                    f"证据：\n" + "\n".join(ev.evidence)
                )
                return StopEvent(result=str(answer))

            prompt = (
                "请改写下面的问题，使下一轮检索更具体。"
                "只返回一个检索问题。\n"
                f"{ev.query}"
            )
            rewritten = await self.llm.acomplete(prompt)
            return PlanEvent(queries=[str(rewritten)])

        query = ev.query
        response = await self.llm.acomplete(
            "判断是否需要拆分问题。"
            "如果需要，返回三个互不重复的子问题；否则返回原问题。\n"
            f"{query}"
        )

        # 示例代码为了清晰，实际项目应使用结构化输出解析
        return PlanEvent(queries=[str(response)])

    @step
    async def retrieve(self, ev: PlanEvent) -> EvidenceEvent:
        query = ev.queries[0]
        passages = await self.retriever(query)

        return EvidenceEvent(
            query=query,
            passages=passages,
            attempt=1,
        )

    @step
    async def judge(self, ev: EvidenceEvent) -> JudgeEvent:
        evidence_text = "\n".join(ev.passages)

        verdict = await self.llm.acomplete(
            "判断证据是否足以回答问题。只输出 SUFFICIENT 或 INSUFFICIENT。\n"
            f"问题：{ev.query}\n"
            f"证据：{evidence_text}"
        )

        sufficient = str(verdict).strip().upper().startswith("SUFFICIENT")

        return JudgeEvent(
            query=ev.query,
            evidence=ev.passages,
            sufficient=sufficient,
            attempt=ev.attempt,
        )

这是一个教学骨架，真实实现需要补充两个关键点。

第一，attempt 不能固定为 1。如果 judge 返回 PlanEvent，下一轮 retrieve 必须继承并递增尝试次数。更好的做法是把运行级状态写入 Context，而不是依赖事件之间手工携带所有字段。

第二，模型输出不应依赖字符串前缀。应使用结构化输出，例如：

class RetrievalDecision(BaseModel):
    action: Literal["DIRECT", "RETRIEVE", "CLARIFY"]
    queries: list[str]
    reason: str

然后校验 queries 数量、长度和重复率。否则“请返回三个问题”可能得到自然语言解释、编号列表或空结果，后续代码无法稳定处理。

---

七、多 Agent：多个角色不等于多个自主系统

多 Agent 是指多个具有独立职责、工具集合或上下文边界的 Agent 协同完成任务。一个 Agent 是否应拆成多个，关键不在于“角色名称是否不同”，而在于是否存在：

• 不同的工具权限；
• 不同的上下文需求；
• 不同的成功标准；
• 不同的模型配置；
• 不同的失败和回退策略。

例如：

ResearchAgent：检索资料，只能写入研究笔记
WriteAgent：根据笔记生成报告
ReviewAgent：检查事实和引用，不直接修改原始资料

如果三个 Agent 看到完全相同的工具、相同的状态和相同的 Prompt，那么拆分它们通常只是增加延迟和 token 消耗。

7.1 AgentWorkflow：交接式协作

LlamaIndex 提供的 AgentWorkflow 可以接收多个 Agent，并由当前 Agent 决定是否把控制权交给另一个 Agent。典型流程是：

1. 根 Agent 接收用户消息；
2. 执行当前 Agent 选中的工具；
3. 当前 Agent 通过 handoff 交接给另一个 Agent；
4. 新 Agent 继续处理；
5. 某个 Agent 返回最终答案。(developers.llamaindex.ai)

from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent


research_agent = FunctionAgent(
    name="ResearchAgent",
    description="检索资料并记录研究笔记。",
    system_prompt=(
        "你是研究员。只负责收集证据和记录笔记。"
        "资料充分后交给 WriteAgent。"
    ),
    llm=llm,
    tools=[search_web, record_notes],
    can_handoff_to=["WriteAgent"],
)

write_agent = FunctionAgent(
    name="WriteAgent",
    description="根据研究笔记撰写 Markdown 报告。",
    system_prompt=(
        "你是技术作者。根据研究笔记写报告，"
        "完成后交给 ReviewAgent。"
    ),
    llm=llm,
    tools=[write_report],
    can_handoff_to=["ReviewAgent", "ResearchAgent"],
)

review_agent = FunctionAgent(
    name="ReviewAgent",
    description="审查报告中的事实、结构和证据。",
    system_prompt="你是审稿人。发现证据不足时交回 WriteAgent。",
    llm=llm,
    tools=[review_report],
    can_handoff_to=["WriteAgent"],
)

workflow = AgentWorkflow(
    agents=[research_agent, write_agent, review_agent],
    root_agent=research_agent.name,
    initial_state={
        "research_notes": {},
        "report_content": "",
        "review": "",
    },
)

response = await workflow.run(
    user_msg="写一份关于 Web 历史的技术报告。",
)

print(str(response))

这种模式的优势是代码少，适合快速构建协作式 Agent。它的风险也很明确：交接顺序和下一步选择部分依赖模型判断。如果评审 Agent 不交回作者，或者作者在证据不足时直接结束，程序本身未必能阻止它。

所以必须给交接增加约束：

• can_handoff_to 限制可达 Agent；
• 状态中记录当前阶段；
• 每个阶段设置最大交接次数；
• ReviewAgent 输出结构化审查结果；
• Workflow 层面对非法阶段转换进行拒绝。

7.2 Orchestrator：集中式路由

第二种模式是设置一个 Orchestrator，由它统一决定调用哪个子 Agent。子 Agent 不直接互相交接，而是作为 Tool 暴露给 Orchestrator。官方文档将这种模式描述为：由一个顶层 Agent 决定下一步调用哪个专门 Agent，并把各专门 Agent 暴露为工具。(developers.llamaindex.ai)

async def call_research_agent(ctx: Context, prompt: str) -> str:
    """让研究 Agent 针对指定主题收集资料。"""
    result = await research_agent.run(
        user_msg=f"请针对以下主题收集资料：{prompt}"
    )

    async with ctx.store.edit_state() as state:
        state["state"]["research_notes"].append(str(result))

    return str(result)


async def call_write_agent(ctx: Context) -> str:
    """让写作 Agent 根据研究笔记生成报告。"""
    async with ctx.store.edit_state() as state:
        notes = state["state"].get("research_notes", [])

    if not notes:
        return "没有研究笔记，无法写作。"

    result = await write_agent.run(
        user_msg="请根据以下研究笔记撰写报告：\n" + "\n".join(notes)
    )

    return str(result)


orchestrator = FunctionAgent(
    tools=[call_research_agent, call_write_agent],
    llm=llm,
    system_prompt=(
        "你是总协调器。根据任务选择研究或写作工具。"
        "研究证据不足时不得要求写作 Agent 直接编造内容。"
    ),
)

集中式路由的优点是每次决策都经过同一个地方，比较容易插入：

• 路由规则；
• 权限检查；
• 最大调用次数；
• 审批节点；
• 预算和超时；
• 回退逻辑。

代价是 Orchestrator 可能成为瓶颈。所有子 Agent 的结果都要回到顶层上下文，Prompt 变长，顶层模型还需要理解每个子 Agent 的能力边界。

7.3 自定义 Planner：当流程必须由代码掌控

第三种方式是让模型只生成结构化计划，程序解析计划并按规则执行：

{
  "steps": [
    {"agent": "research", "input": "收集部署资料"},
    {"agent": "research", "input": "收集故障恢复资料"},
    {"agent": "write", "input": "合并两组资料"},
    {"agent": "review", "input": "检查引用和事实"}
  ]
}

程序负责：

1. 校验 Agent 名称；
2. 检查步骤数量；
3. 检查是否允许该顺序；
4. 执行并发或串行任务；
5. 处理失败和重试；
6. 判断是否允许结束。

官方文档把这种模式定位为灵活性最高的方式，适合需要严格计划格式、外部调度器或额外元数据的场景。(developers.llamaindex.ai)

它也最容易出现“计划解析器变成第二个 Workflow 引擎”的问题。因此只有当 AgentWorkflow 和 Orchestrator 无法表达所需流程时，才值得使用。

---

八、多 Agent 路由：分类、能力匹配、动态选择与回退

路由问题可以形式化为：

给定查询 $q$、Agent 集合 $A=\{a_1,\ldots,a_n\}$，选择一个或多个 Agent：

$$a^* = \arg\max_{a_i \in A} \left[ \operatorname{Fit}(q,a_i) -\lambda \operatorname{Cost}(a_i) -\mu \operatorname{Risk}(q,a_i) \right]$$

其中：

• Fit 表示能力匹配度；
• Cost 表示延迟、token 或外部服务费用；
• Risk 表示错误、越权或不可验证回答的风险。

8.1 分类路由

分类路由先把请求映射到固定类别：

billing     -> BillingAgent
technical   -> TechnicalAgent
sales       -> SalesAgent
unknown     -> GeneralAgent

它实现简单，容易统计准确率，但无法很好处理跨领域问题。例如：

“这个技术方案的成本和部署风险如何？”

它同时属于 technical 和 billing，强行单选会丢失信息。

8.2 能力匹配

更合理的路由不只看类别，而是比较 Agent 的能力描述：

agents = [
    {
        "name": "TechnicalAgent",
        "capabilities": ["API", "部署", "故障排查"],
        "tools": ["internal_docs", "log_search"],
    },
    {
        "name": "BillingAgent",
        "capabilities": ["计费", "套餐", "账单"],
        "tools": ["billing_api"],
    },
]

对于每个 Agent，计算查询所需能力与其能力集合的覆盖：

$$\operatorname{Coverage}(q,a) = \frac{ |\operatorname{RequiredSkills}(q) \cap \operatorname{Skills}(a)| }{ |\operatorname{RequiredSkills}(q)| }$$

如果覆盖率不足，就不应仅凭语言相似度选择该 Agent。

8.3 动态选择与回退

动态选择允许 Agent 在运行时发现当前 Agent 无法完成任务，然后回退：

stateDiagram-v2
    [*] --> Routed
    Routed --> Specialist
    Specialist --> Answered: 证据充分
    Specialist --> Retry: 参数可修正
    Specialist --> Fallback: 工具不可用
    Specialist --> Handoff: 能力不匹配
    Retry --> Specialist
    Handoff --> OtherSpecialist
    Fallback --> Human
    Answered --> [*]
    Human --> [*]
    OtherSpecialist --> Answered

回退不应只是“再调用一个 Agent”。至少要携带失败原因：

class AgentFailure(BaseModel):
    agent: str
    category: Literal[
        "UNSUPPORTED",
        "INVALID_INPUT",
        "NO_EVIDENCE",
        "TOOL_ERROR",
        "TIMEOUT",
    ]
    detail: str
    retryable: bool

不同失败类型的处理不同：

• INVALID_INPUT：向用户澄清；
• UNSUPPORTED：路由到其他 Agent；
• NO_EVIDENCE：改写查询或返回证据不足；
• TOOL_ERROR：按幂等性和预算重试；
• TIMEOUT：降级、异步化或转人工。

如果所有失败都统一成“请换一个 Agent”，系统可能在多个 Agent 之间循环。

---

九、并发：速度提升来自独立性，而不是简单 gather

在多 Agent 或查询分解中，只有当任务之间满足近似独立条件时才适合并发：

$$\operatorname{Deps}(q_i) \cap \operatorname{Writes}(q_j)=\varnothing$$

也就是说，$q_i$ 不依赖 $q_j$ 的输出，且两者不会并发写入同一状态。

适合并发的任务：

查询“部署方式”
查询“成本约束”
查询“故障恢复”

不适合并发的任务：

先检索资料 -> 再根据资料生成报告

LlamaIndex Workflow 支持通过事件列表表达批量并发，也支持通过 Context 动态发送事件；类型化事件更易验证，而动态 Context API 更灵活但需要开发者自己维护收集、计数和完成条件。(developers.llamaindex.ai)

并发系统必须定义：

• 最大并发数；
• 单任务超时；
• 部分成功是否可接受；
• 取消策略；
• 状态写入顺序；
• 聚合结果的确定性。

例如三个子查询中一个超时，最终答案可以有三种策略：

1. 等待全部完成；
2. 使用两个成功结果并声明缺失；
3. 直接失败，不生成答案。

这不是框架默认能替你决定的业务语义。

---

十、生命周期与错误路径

一个可诊断的 Agent 运行至少包含以下阶段：

创建 Workflow
  -> 校验事件图
  -> 创建或恢复 Context
  -> 接收 StartEvent
  -> Agent 规划
  -> Tool/Agent 执行
  -> 写入观测与状态
  -> 判断继续、交接、重试或结束
  -> 返回 StopEvent

10.1 启动阶段失败

常见原因：

• Step 的输入事件没有生产者；
• 返回事件没有消费者；
• 没有可达的 StopEvent；
• Tool Schema 无法生成；
• LLM 配置缺失。

Workflow 可以在运行前调用：

workflow.validate()

官方文档建议在测试或启动阶段主动验证 Workflow；对于依赖资源的流程，还可以选择验证资源配置。(developers.llamaindex.ai)

10.2 Tool 阶段失败

应记录至少以下字段：

run_id
agent_name
tool_name
arguments_hash
started_at
duration_ms
status
error_category
result_size

不要在日志中直接记录包含密钥、身份证号、完整用户消息或敏感检索内容的原始参数。

10.3 LLM 阶段失败

需要区分：

• 请求超时；
• 限流；
• 上下文长度超限；
• 不支持工具调用；
• 返回格式无法解析；
• 模型返回空内容。

例如某些模型不支持流式输出时，官方 Agent 文档建议通过 streaming=False 关闭流式能力。(docs.llamaindex.ai)

10.4 超时和取消

Workflow 应设置总超时，Tool 还应设置独立超时。总超时不能替代子任务超时，否则一个阻塞的数据库查询可能占满整个运行预算。

重试也必须考虑幂等性：

• 查询类 Tool 通常可以重试；
• 创建订单、发送邮件、扣款等写操作不能仅凭网络错误就重试；
• 非幂等操作应使用幂等键和外部确认状态。

---

十一、评测：分别测路由、检索、工具和最终答案

只评估最终答案会掩盖系统真正的问题。一个多 Agent RAG 系统至少应拆成以下指标：

路由指标

给定标注好的请求，评估：

$$\operatorname{RouteAccuracy} = \frac{\text{正确选择的请求数}}{\text{请求总数}}$$

还应统计：

• 错误路由率；
• 不必要的多 Agent 调用率；
• 回退成功率；
• 路由平均延迟。

检索指标

对于标注证据的查询，统计：

$$\operatorname{Recall@k} = \frac{\text{前 }k\text{ 个结果中包含相关证据的查询数}} {\text{查询总数}}$$

重排后再测 Precision@k、证据覆盖率和重复率。

工具指标

工具层应评估：

• Tool 选择准确率；
• 参数解析正确率；
• 参数校验拒绝率；
• 可重试错误的恢复率；
• 不可重试错误的误重试率。

答案指标

最终答案至少要区分：

• Faithfulness：答案是否能被检索证据支持；
• Answer correctness：答案是否正确；
• Completeness：是否覆盖问题的所有子约束；
• Citation correctness：引用是否真的支持对应结论；
• Abstention quality：证据不足时是否正确拒答。

对于 Agentic RAG，还应记录完整轨迹：

query
  -> route decision
  -> selected agent
  -> tool calls
  -> retrieved nodes
  -> rerank scores
  -> query rewrites
  -> stop decision
  -> final answer

LlamaIndex 提供追踪、调试和评测相关能力；官方文档也将 RAG 评估、响应评估和检索评估作为独立主题。(developers.llamaindex.ai)

---

十二、常见误用与边界

把所有逻辑交给一个 Agent

一个大 Agent 可以快速起步，但随着工具数量增长，模型需要在更多相似工具中选择，Prompt 变长，错误路由增加。解决方案不是立即拆成十个 Agent，而是先划分工具域、收紧描述和权限，再观察轨迹。

把 Context 当作数据库

Context 适合保存运行状态，不适合代替事务数据库、向量数据库或消息队列。需要查询、索引、并发事务和审计的数据，应放在对应的持久化系统中，Context 只保存引用和流程游标。

把检索结果直接当事实

检索到的节点只是候选证据。节点可能过期、重复、被错误切分或来自错误租户。必须在检索器层加入租户过滤、时间过滤、来源标识和权限校验。

用更多 Agent 解决质量问题

如果问题来自错误的文档切分、错误的检索过滤或 Tool 返回值不稳定，增加 Agent 只会扩大错误传播范围。

没有显式停止条件

任何包含“继续搜索直到满意”的 Prompt 都是不完整的。应该至少有最大轮次、最大工具调用数、总 token 预算、总延迟预算以及证据覆盖判定。

---

十三、如何选择 LlamaIndex 的实现层级

可以按照以下顺序增加复杂度：

1. 固定 Query Engine：流程完全确定，不需要模型选择工具；
2. Query Engine 作为 Tool：Agent 决定何时查询知识库；
3. FunctionAgent：需要工具调用循环，但流程仍然简单；
4. 自定义 Workflow：需要分支、循环、并发、审批、重试或显式状态；
5. AgentWorkflow：多个 Agent 以交接方式协作；
6. Orchestrator Agent：需要集中式路由和统一策略；
7. Custom Planner：需要严格计划格式、外部调度或复杂执行编排。

LlamaIndex 官方多 Agent 文档也将 AgentWorkflow、Orchestrator 和自定义 Planner 作为三种主要模式，并指出它们分别在代码量、灵活性和控制能力上进行取舍。(developers.llamaindex.ai)

最终的工程边界可以概括为：

• Agent 决定“下一步做什么”；
• Tool 决定“允许做什么”；
• Context 记录“已经知道什么、当前处于什么阶段”；
• RAG 提供“有哪些外部证据”；
• Workflow 规定“流程如何转移、何时并发、何时失败和何时停止”；
• 多 Agent 解决“不同能力如何隔离、匹配和协作”。

当这些职责清晰分离时，LlamaIndex Agent 才不只是一个会调用函数的聊天机器人，而是一个能够被验证、观测、限流、恢复和评测的工程系统。

---

系列导航与关联阅读

• 系列入口：Agent 工程完整路线：从运行循环、记忆与协议到安全、评测和生产交付
• 上一篇：Semantic Kernel Agent：Plugin、Process、Memory、编排和服务集成
• 下一篇：编程 Agent 工程：仓库上下文、搜索、补丁、命令、验证和提交
• 延伸：Agentic RAG：检索决策、查询分解、重排、迭代和停止
• 延伸：多 Agent 路由：分类、能力匹配、动态选择、回退和评测

官方资料

• LlamaIndex Agent Documentation
• LlamaIndex Repository

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。