上个月，我正在开发一个客服智能体，它需要完成三件事：从数据库查询订单详情、在知识库里搜索排障步骤，以及决定是否要把问题升级转给人工。我一开始用的是标准的链式（chain-based）方法，按线性顺序把各个步骤串联起来。在一切顺利的“理想路径”下，它跑得挺好。但一旦我需要让智能体回头重试——比如客户的第一个排障步骤没管用，我们需要换第二种方案时——整个流程就崩了。有向无环图（DAG）用来做流水线很棒，但对于需要推理、分支和循环的智能体来说，简直是噩梦。
就在那时，我偶然发现了 LlamaIndex Workflows。传统的基于图的编排需要你显式地把节点连在一起，而 Workflows 不同，它使用的是事件驱动架构。各个步骤（Steps）负责发送和接收事件，框架则根据这些事件来决定接下来运行什么。听起来这正是我需要的，所以我花了一个周末用它重构了我的客服智能体。以下是我学到的东西。
核心概念：步骤与事件
Workflows 建立在两个基础模块之上：步骤和事件。步骤就是一个加了 @step 装饰器的函数，它会执行一些操作并发送事件。事件则是类型化的数据类，负责在步骤之间传递信息。神奇之处在于，步骤之间不需要互相认识——它们只需要知道自己接收什么事件、产生什么事件就行了。
这跟基于图的方法相比是一个巨大的转变。在 DAG 中，你需要手动定义边：“节点 A 连接到节点 B”。而在事件驱动的 Workflow 中，步骤 A 发出一个 LookupResultEvent，任何接收 LookupResultEvent 作为输入的步骤都会自动收到它。这让分支和循环变得非常自然，不再是生硬地强加了。
环境配置
首先，安装必要的包。你需要 LlamaIndex 核心库和 workflow 工具包：
pip install llama-index-core>=0.11.16 llama-index-utils-workflow
我还推荐安装 workflow 渲染工具——它能生成 HTML 格式的流程可视化图，对调试非常有帮助：
pip install llama-index-utils-workflow
至于大语言模型（LLM），这里我使用 OpenAI，因为它是默认选项，但你也可以随时换成其他提供商。只需设置你的 API 密钥：
import os
os.environ["OPENAI_API_KEY"] = "sk-your-key-here"
如果你想使用像 Ollama 这样的本地或开源模型，可以通过 Settings 对象进行配置：
from llama_index.core import Settings
from llama_index.llms.ollama import Ollama
Settings.llm = Ollama(model="llama3", request_timeout=120.0)
构建工具调用智能体
让我带你过一遍我实际构建的智能体。我希望这个 Workflow 能够：
- 接收用户查询
- 判断是需要查询订单信息还是搜索知识库
- 执行相应的工具
- 评估结果，然后要么直接回复，要么循环回去执行更多操作
第一步：定义事件
事件是 Workflow 的骨干。每个事件都携带着步骤间需要传递的特定数据。以下是我定义的内容：
from llama_index.core.workflow import Event
class UserQueryEvent(Event):
query: str
class ToolCallEvent(Event):
tool_name: str
tool_args: dict
class ToolResultEvent(Event):
tool_name: str
result: str
class ResponseEvent(Event):
response: str
这里的类型声明非常重要。Workflows 使用这些类型来将事件路由到正确的步骤。我一开始犯了个错，用了普通的字典而不是类型化的事件，结果路由彻底崩了。
第二步：定义工具
我创建了两个简单的工具——一个用于查询订单，一个用于搜索知识库：
from llama_index.core.tools import FunctionTool
def lookup_order(order_id: str) -> str:
"""Look up order details by order ID."""
# 实际场景中，这里会查询数据库
orders = {
"ORD-123": "Order shipped on Jan 15, tracking: 1Z999AA10123456784",
"ORD-456": "Order processing, estimated ship date: Jan 20",
}
return orders.get(order_id, "Order not found")
def search_knowledge_base(query: str) -> str:
"""Search the knowledge base for troubleshooting steps."""
# 实际场景中，这里会查询向量存储
kb = {
"reset password": "Go to Settings > Account > Reset Password",
"wifi connection": "1. Restart router 2. Forget network 3. Reconnect",
}
for key, value in kb.items():
if key in query.lower():
return value
return "No relevant articles found"
order_tool = FunctionTool.from_defaults(fn=lookup_order)
kb_tool = FunctionTool.from_defaults(fn=search_knowledge_base)
第三步：构建 Workflow
现在到了重头戏。Workflow 类将你的步骤组合在一起，并管理共享上下文：
from llama_index.core.workflow import Workflow, Context, StartEvent, StopEvent, step
from llama_index.core.llms import LLM
class SupportAgentWorkflow(Workflow):
def __init__(self, llm: LLM, **kwargs):
super().__init__(**kwargs)
self.llm = llm
@step
async def decide_tool(self, ctx: Context, ev: StartEvent | UserQueryEvent) -> ToolCallEvent | ResponseEvent:
"""Decide which tool to call based on the query."""
query = ev.get("query", "")
# 将查询存入上下文，留待后用
await ctx.set("query", query)
# 询问 LLM 该使用哪个工具
prompt = f"""You are a support agent. Given this user query, decide what to do.
Available tools:
- lookup_order: Look up order details (requires order_id)
- search_knowledge_base: Search for troubleshooting steps (requires query)
User query: {query}
Respond with either:
- TOOL: lookup_order|order_id_value
- TOOL: search_knowledge_base|query_value
- RESPONSE: your direct response if no tool is needed"""
response = await self.llm.acomplete(prompt)
text = str(response).strip()
if text.startswith("TOOL:"):
parts = text.split("|", 1)
tool_name = parts[0].replace("TOOL:", "").strip()
tool_arg_value = parts[1].strip() if len(parts) > 1 else ""
if tool_name == "lookup_order":
return ToolCallEvent(tool_name=tool_name, tool_args={"order_id": tool_arg_value})
elif tool_name == "search_knowledge_base":
return ToolCallEvent(tool_name=tool_name, tool_args={"query": tool_arg_value})
elif text.startswith("RESPONSE:"):
return ResponseEvent(response=text.replace("RESPONSE:", "").strip())
# 兜底回复
return ResponseEvent(response="I'm not sure how to help with that. Could you rephrase?")
@step
async def execute_tool(self, ctx: Context, ev: ToolCallEvent) -> ToolResultEvent:
"""Execute the selected tool."""
if ev.tool_name == "lookup_order":
result = lookup_order(**ev.tool_args)
elif ev.tool_name == "search_knowledge_base":
result = search_knowledge_base(**ev.tool_args)
else:
result = "Unknown tool"
return ToolResultEvent(tool_name=ev.tool_name, result=result)
@step
async def evaluate_result(self, ctx: Context, ev: ToolResultEvent) -> ResponseEvent | UserQueryEvent:
"""Evaluate tool result and decide if we need more action."""
query = await ctx.get("query")
prompt = f"""You are a support agent. Based on the tool result, decide if you can answer the user's query.
User query: {query}
Tool used: {ev.tool_name}
Tool result: {ev.result}
If the result is sufficient, respond with RESPONSE: followed by your answer.
If you need to try another tool or approach, respond with RETRY: followed by a refined query."""
response = await self.llm.acomplete(prompt)
text = str(response).strip()
if text.startswith("RESPONSE:"):
return ResponseEvent(response=text.replace("RESPONSE:", "").strip())
elif text.startswith("RETRY:"):
new_query = text.replace("RETRY:", "").strip()
return UserQueryEvent(query=new_query)
return ResponseEvent(response=ev.result)
@step
async def format_response(self, ctx: Context, ev: ResponseEvent) -> StopEvent:
"""Format and return the final response."""
return StopEvent(result=ev.response)
这里信息量比较大，让我拆解一下关键部分：
StartEvent和StopEvent是内置的事件类型。StartEvent负责启动 Workflow，StopEvent负责结束它。- 类型提示中的
|操作符至关重要。当一个步骤可以返回多种事件类型时，框架会根据这些类型注解来正确路由事件。decide_tool可以返回ToolCallEvent或ResponseEvent，框架会把它们分别送到对应的下一步。 Context是一个共享状态对象。我用它来存储原始查询，以便后续步骤可以访问。这比把状态塞进每一个事件里传参要优雅得多。- 循环发生在
evaluate_result中。如果 LLM 认为工具返回的结果不够好，它就会发出一个UserQueryEvent，这个事件会被路由回decide_tool。这就是在 DAG 中极其痛苦才能实现的循环行为。
第四步：运行
from llama_index.llms.openai import OpenAI
llm = OpenAI(model="gpt-4o-mini")
workflow = SupportAgentWorkflow(llm=llm, verbose=True)
result = await workflow.run(query="Where is my order ORD-123?")
print(result)
# 输出: "Your order ORD-123 was shipped on January 15th.
# Your tracking number is 1Z999AA10123456784."
设置 verbose=True 会在执行时打印出每一步，这对调试来说简直是无价之宝。你可以清楚地看到发出了哪些事件，以及哪些步骤接收了它们。
可视化 Workflow
我最喜欢的功能之一就是 HTML 渲染。定义好 Workflow 后，你可以生成一张可视化的流程图：
from llama_index.utils.workflow import draw_all_possible_flows
draw_all_possible_flows(workflow, filename="support_agent.html")
在浏览器中打开这个 HTML 文件，你会看到一张清晰的图表，展示了 Workflow 中所有可能的路径，包括循环。当我试图向同事解释这个流程时，这个功能帮我省了好几个小时。
让我意外的地方
在开发过程中，有几件事让我踩了坑：
类型提示不是可选的。 我一开始试图省略步骤返回值的联合类型提示，以为框架能自己推断路由。它不能。整个路由机制完全依赖于这些类型注解。如果你不明确指定 decide_tool 返回的是 ToolCallEvent | ResponseEvent，框架根本不知道该把事件发到哪里。
Context 只支持异步。 ctx.set() 和 ctx.get() 方法都是协程。我好几次忘了加 await 关键字，结果得到了让人摸不着头脑的报错。一定要记住：await ctx.set("key", value)。
事件命名对可读性至关重要。 当你的 Workflow 变大时，会有很多事件类型。我一开始用了类似 StepOneEvent 这种泛泛的名字，很快就晕头转向了。换成 ToolCallEvent 和 EvaluationNeededEvent 这种描述性的名字后，一切都清晰多了。
实用建议
从简单开始，然后再加循环。 先跑通一条直线路径（查询 → 工具 → 响应），验证没问题了，再加循环逻辑。在一个连线性都没跑通的 Workflow 里调试循环，那滋味绝对酸爽。
开发时开启
verbose=True。 控制台输出会显示每一次事件发送和步骤执行。在生产环境中为了性能可以关掉它。保持事件精简。 事件应该只携带下一步所需的数据。我一开始把整个对话历史都塞进了事件里，导致 Workflow 又慢又难调试。共享状态应该用
Context对象来存。单独测试步骤。 既然步骤本质上就是异步函数，你完全可以创建相应的事件并直接调用该步骤来进行独立测试。这比每次测试都要跑一遍整个 Workflow 轻松多了。
诚实的局限性
Workflows 并不是万能的。对于简单的线性流水线（摄入 → 文档分块 → 嵌入 → 存储），传统的流水线更简单直接。只有当你涉及分支、循环或条件逻辑时，事件驱动的开销才是划算的。
文档还在不断完善中。我遇到了一些边界情况——比如如何处理多个步骤发出相同事件类型的情况——需要去翻源码。不过社区很活跃，LlamaIndex 的 Discord 频道帮了不少忙。
另外，要注意循环能力意味着你需要防范死循环。我在上下文中加了一个重试计数器，用来限制 evaluate_result 可以循环回去的次数。如果不加这个，一个犯迷糊的 LLM 可能会无限转圈。
总结
LlamaIndex Workflows 精准解决了我面临的问题：构建一个需要推理、行动，并且在搞不定时能回头重试的智能体。对于智能体系统，事件驱动架构感觉非常自然，这是 DAG 永远做不到的。如果你在构建任何超越简单链式调用的东西——尤其是涉及工具调用、多步推理或人机协同（human-in-the-loop）模式的系统——Workflows 绝对值得深入研究。只要记住：留意你的类型提示，限制好你的循环。