用 LlamaIndex 构建基于 RAG 的 API 文档助手（第二部分：工具、智能体与……）

Towards Dev - Medium 2026-08-13T06:47:18.644982

在第一部分中，我们为一个虚构的电商 API 产品 ShopSphere 构建了一个小型 AI 助手的检索引擎。我们使用 LlamaIndex 加载文档、将其切分成块、创建嵌入向量、存储，最后把所有环节连接到 QueryEngine（查询引擎）上，让它能够搜索文档并利用检索到的信息回答问题。

下面快速回顾一下第一部分的核心概念：

• 组件（Components）是构建模块——包括模型、存储、文档处理和搜索。也就是我们在第一部分构建的东西。
• 工具（Tools）把这些能力转化为代理（Agent）实际可以用于特定任务的东西。
• 代理（Agents）是决策者——它们查看问题，并决定哪个工具更合适。
• 工作流（Workflows）则用于定义一组特定的执行步骤，适用于我们希望流程更可控、更少意外的场景。

我们已经打好了地基。现在，是时候把这些能力变成工具，引入能够决定如何使用这些工具的代理，最后再为那些希望流程可预测的部分定义工作流。

工具——让 AI 不止有一种获取答案的方式

QueryEngine 就像一把锤子。它是一把好锤子，但如果 AI 只有这一件工具，那所有问题看起来都会像钉子。

举个例子，我们的错误码文档中有精确的映射关系，例如 429 → Too Many Requests（请求过多）。我们可以从文档中检索这一信息，然后让大语言模型（LLM）来解释它，但这存在一个小风险：模型可能误解或意外改变精确的值。对于这类信息，我们更希望 AI 调用一个能返回准确答案的函数。这样一来，AI 就不必凭记忆复述或转述这个值，而是直接使用一个可靠的工具去获取它。

def lookup_http_status_code(code: str) -> str:
"""Look up the exact meaning of a ShopSphere API HTTP status code."""
return _STATUS_CODES.get(code.strip(), f"Unknown status code: {code}")

把它包装成 FunctionTool（函数工具），就立刻变成了 AI 智能体可以主动调用的东西：

from llama_index.core.tools import FunctionTool

tool = FunctionTool.from_defaults(lookup_http_status_code)

这里有个大家经常忽略的点：docstring 实际上就是 AI 理解工具的使用说明书。智能体并不会去看你函数内部怎么实现，它只依赖工具的名字、描述、参数这些信息，来决定要不要用、以及怎么用。描述写得含糊，智能体就可能选错工具；描述写得到位，它就能清楚判断什么时候该用这个工具。这就像给厨房抽屉贴标签——标签要清楚到让从没来过你家的人也知道里面装的是什么。

我们还可以把第一部分构建的整个 QueryEngine（查询引擎）也包装成工具。这样一来，智能体就可以把“搜索文档”当作回答问题时可选的工具之一：

from llama_index.core.tools import QueryEngineTool

docs_tool = QueryEngineTool.from_defaults(
    query_engine=query_engine,
    name="shopsphere_docs_search",
    description="Answers questions about the ShopSphere API by searching its documentation.",
)

现在我们就有了一个小工具箱：一个用来搜文档，另外几个用来查具体事实。接下来该雇个人来真正用它们了。

智能体（Agents）——让 AI 自己决定用哪个工具

智能体是我们应用的决策层。它看着用户的问题，盘算手头有哪些工具，然后判断是否需要调用工具——如果需要，具体调用哪个，以及该传什么参数进去。

举个例子，用户问“怎么创建订单？”，智能体可能会选择文档搜索工具。用户要是问“HTTP 429 是什么意思？”，它就会转而使用状态码查询工具。

换句话说，我们不再为每个可能的问题编写 if/else 逻辑，而是给 AI 一组工具，让它自己决定用哪个：

from llama_index.core.agent.workflow import AgentWorkflow

agent = AgentWorkflow.from_tools_or_functions(
    [docs_tool, lookup_http_status_code, lookup_rate_limit],
    llm=llm,
    system_prompt="You are the ShopSphere API Documentation Assistant...",
)

response = await agent.run("How do I create a new order from a cart?")

注意 agent.run(...) 前面的 await。它的意思是告诉 Python：“等待这个 agent 执行完，但在等待大语言模型或工具响应期间，让其他异步任务继续运行。”如果没有 await，代码会在 agent 返回结果之前就继续往下走了。所以说，await 确实表示等待结果，但等待过程中并不会阻塞整个应用。这是 Python 的 async/await 概念，并不是 LlamaIndex 特有的东西。

还有一点值得了解：agent 默认不会记住之前的对话。每次调用 .run() 都会从全新状态开始。当你希望进行真正的多轮对话时，这就成了问题。比如用户先问“How do I create an order?”，接着又问“That endpoint needs what scope?”，这时候 agent 需要记得我们说的是哪个 endpoint。在 LlamaIndex 中，可以用一个共享的 Context 对象来保存这种短期对话状态，让它在多次 agent 运行之间保持可用：

from llama_index.core.workflow import Context

ctx = Context(agent)
await agent.run("How do I create a new order?", ctx=ctx)
await agent.run("And what scope does that endpoint need?", ctx=ctx)  # 记得第一轮的内容

当一个 AI 不够用：让专职 agent 互相交接工作

随着工具越来越多，让一个 agent 处理所有事情很快就会变得混乱。这就像让一个客服人员处理所有类型的用户请求——你给他们的职责越多，每件事就越难做好。更好的做法是让 agent 各司其职，在需要时互相交接工作。

我们把助手拆成了两个 agent：

docs_agent —— 前台 agent。

docs_agent 负责回答「如何创建订单？」这类概念性问题；如果用户想查某个具体事实或代码，它会把问题转给专家。reference_agent 就是那位专家，专门处理「HTTP 429 是什么意思？」这类精确问题；一旦遇到需要更宽泛解释或文档检索的问题，它又会把对话交还给 docs_agent。

docs_agent = ReActAgent(
    name="docs_agent",
    description="Answers conceptual questions... hands off to reference_agent for exact codes.",
    tools=[docs_tool],
    llm=llm,
)

reference_agent = ReActAgent(
    name="reference_agent",
    description="Looks up exact status codes and rate limits.",
    tools=[lookup_http_status_code, lookup_rate_limit],
    llm=llm,
)

workflow = AgentWorkflow(agents=[docs_agent, reference_agent], root_agent="docs_agent")

所有消息都会先进入 docs_agent，它作为根 agent 接收请求。它可以自己直接作答，也可以看一眼 reference_agent 的描述，判断「这个问题更适合那个 agent 处理」，然后把问题转交过去。

注意，description 字段对 agent 的作用，和 docstring 对工具的作用是一样的。它相当于一个标签，用来告诉其他 agent：这个 agent 擅长什么、什么时候该调用它。

Workflows——当你不想让 AI 自由发挥时

Agent 适合「让 AI 自己决定下一步」的场景。但有些流程必须每次按固定步骤走。比如：总是先对问题分类，检索到的答案不理想就重试一次，在返回最终答案前附上来源。这种场景下，我们不想让 agent 全权决定整个流程，而是把流程明确写进代码，只在真正需要 LLM 判断力的时候才让它参与。

这就是 Workflow 的用途：由我们自己定义一系列步骤，仅在必要之处引入 LLM。涉及的词汇很少：

• Workflow —— 定义整个流程的类。
• @Step —— 把某个方法标记为流程中的一个步骤。
• Event —— 步骤之间传递的一小包数据。
• StartEvent —— 标记流程从哪儿开始。

StopEvent 用来标记工作流的结束位置。下面是最简单的一个例子：

from llama_index.core.workflow import StartEvent, StopEvent, Workflow, step

class MyWorkflow(Workflow):
    @step
    async def my_step(self, ev: StartEvent) -> StopEvent:
        return StopEvent(result="Hello, world!")

result = await MyWorkflow().run()

现在我们来构建自己的工作流：先对问题分类，再去检索文档；如果拿到的答案不理想，就重试一次；最后把答案和来源一起返回。

class ClassifyEvent(Event):
    question: str

class DocsRetrieveEvent(Event):
    question: str

class ReferenceLookupEvent(Event):
    question: str
    kind: str
    argument: str

class AnswerReadyEvent(Event):
    answer: str
    sources: list[str]

这些事件类本质上就是一个个带标签的盒子，用来把信息从一个步骤传到下一个步骤。

@step
async def classify_question(self, ctx: Context, ev: ClassifyEvent) -> DocsRetrieveEvent | ReferenceLookupEvent:
    # 这里用简单的规则判断，决定这个问题走哪条分支
    ...

返回值类型 DocsRetrieveEvent | ReferenceLookupEvent 表示这一步可以在两条路径中选择。根据分类结果，它会把流程导向「文档检索」路径，或者「参考查找」路径。

@step
async def docs_retrieve(self, ctx: Context, ev: DocsRetrieveEvent | RetryEvent) -> AnswerReadyEvent | RetryEvent:
    ...
    if len(answer_text.strip()) < 25 and retry_count < 1:
        return RetryEvent(question=f"Explain in more detail: {ev.question}")

    return AnswerReadyEvent(answer=answer_text, sources=sources)

这里就形成了一个循环。如果检索到的答案看起来太短或不完整，这个步骤会构造一个 RetryEvent，把改写后的问题重新扔回工作流，再尝试一次搜索。我们只允许重试一次；之后不管结果有没有改善，工作流都会继续往下走。

那么，工作流是怎么知道要再次执行 docs_retrieve 的呢？

这一段最容易把人绕晕，咱们放慢点讲。第一次看到这种写法，你会觉得像变魔术——代码里从头到尾找不到一行写着“如果返回值是 RetryEvent，就再调一次 docs_retrieve”。实话实说：你也不需要单独去写什么“监听器”。订阅关系就藏在 ev 参数的类型注解里。

你可以把整个流程想象成一个收发室。工作流里的每个步骤都像一个员工，他得到的指令是：“凡是寄给我的信封，我都拆开。”信封上的地址不是人名，而是一个类型，比如 RetryEvent 或 AnswerReadyEvent。

当 LlamaIndex 构建工作流时，它会读取每个步骤的函数签名，然后生成一张小小的分拣表：

• 类型为 DocsRetrieveEvent 或 RetryEvent 的信封 → 交给 docs_retrieve
• 类型为 ClassifyEvent 的信封 → 交给 classify_question
• 类型为 AnswerReadyEvent 的信封 → 交给 finalize

任何一个步骤执行完毕、返回一个事件对象时，这个对象并不会被直接塞给下一个函数，而是先丢进公共的信件堆里。随后，工作流的调度器会看看刚刚到的是什么类型的信封，查一下分拣表，再把它送到所有声明“我接收这类信”的步骤手中。

所以，当 docs_retrieve 返回一个 RetryEvent 时，发生的事情是这样的：信封进了信件堆 → 调度器查表：“谁接收 RetryEvent？”答案是：docs_retrieve 自己也接收——因为它的函数签名里明确写着 ev: DocsRetrieveEvent | RetryEvent。于是 docs_retrieve 被再次调用，这次手里拿的是 RetryEvent，而不是最初的 DocsRetrieveEvent。

整个机关就在这里。这不是递归，也不是你手写的 while 循环——只不过是一个步骤订阅了不止一种信封，而它接受的类型里，恰好包含它自己有时会发出去的那一种。循环就是这么来的。

还有一点很实用：流程中不直接相邻的步骤，有时也需要共享信息——比如我们的重试计数器，docs_retrieve 要在自己的多次重试之间记住它。

这就是 Context 的用途——它就像一本共享笔记本，工作流里的每一步都可以读写：

await ctx.store.set("retry_count", retry_count + 1)
retry_count = await ctx.store.get("retry_count")

整合起来

把这四层堆叠到一起，就能看到完整助手的一次真实对话长什么样：

使用 DocsRetrieveEvent 的工作流执行

使用 ReferenceLookupEvent 的工作流执行

想看完整的可用代码，可以访问这个仓库：github.com/H11199/API-Documentation-Agent

自己动手跑一遍也很简单：

git clone https://github.com/H11199/API-Documentation-Agent.git
cd API-Documentation-Agent
python3 -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\Activate.ps1
pip install -r requirements.txt
cp .env.example .env             # 然后把你的 HF token 填进去，格式为 HF_TOKEN=hf_...
python ingestion_pipeline.py     # 用 data/ 里的示例文档构建索引
python app.py                    # 和助手对话

谢谢阅读！如果这篇文章帮你把之前觉得困惑的地方理清了，欢迎拍个手——这会让我知道该多写些这类内容。如果你还没读第一部分，或者想回顾一下我们是怎么搭建 RAG 和 QueryEngine 基础的，建议先从那篇开始，再回来继续。这样理解本部分的概念会容易得多。

本文原载于 Medium 的 Towards Dev 专栏，读者可以在那里通过标注和回应继续讨论。

查看原文