LlamaIndex Workflows：用步骤与事件编排智能体

这篇适合已经会用 Python 写函数、想让多个步骤或多个智能体按顺序协作的读者。读完你能用 Workflow 写出单步和多步的流程，理解事件与状态怎样在步骤之间传递，并搭出一个带状态的多智能体工作流。

Workflow 是什么

LlamaIndex 的 workflow（工作流）把代码组织成按顺序执行的步骤。它的写法是定义若干 Steps（步骤），每个步骤由 Events（事件）触发，执行完又会发出新的事件，去触发下一个步骤^[1]。

它的主要好处有四点：代码按步骤拆分，结构清楚；采用事件驱动，控制流可以灵活安排；步骤之间的通信有类型约束；内置状态管理。它既能支持简单的智能体交互，也能支持复杂的智能体交互^[1]。

准备：安装与运行环境

安装 Workflow 包的命令是 pip install llama-index-utils-workflow^[1]。另有一个可以在 Google Colab 中运行的 notebook，可以对照着代码一步步操作^[1]。

第一步：写一个单步工作流

单步工作流是一个继承自 Workflow 的类，用 @step 装饰其中的函数。StartEvent 和 StopEvent 是两个特殊事件，分别表示工作流的开始和结束^[1]：

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

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


w = MyWorkflow(timeout=10, verbose=False)
result = await w.run()

调用 w.run() 即可运行工作流^[1]。示例创建工作流对象时设置了 timeout=10 和 verbose=False^[1]。

第二步：用事件在步骤之间传数据

要连接多个步骤，就要自定义事件来携带数据。一个继承自 Event 的类可以把第一步的输出传给第二步。示例中 ProcessingEvent 就承担了这个作用^[1]：

from llama_index.core.workflow import Event

class ProcessingEvent(Event):
    intermediate_result: str

class MultiStepWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent) -> ProcessingEvent:
        # Process initial data
        return ProcessingEvent(intermediate_result="Step 1 complete")

    @step
    async def step_two(self, ev: ProcessingEvent) -> StopEvent:
        # Use the intermediate result
        final_result = f"Finished processing: {ev.intermediate_result}"
        return StopEvent(result=final_result)

w = MultiStepWorkflow(timeout=10, verbose=False)
result = await w.run()
result

类型标注在这里很关键，它保证工作流按正确的方式执行。同样的标注机制还能做出分支、循环和汇合^[1]。

第三步：循环

在类型标注中用 | 运算符表示“或”，就能做循环。下面的例子里，step_one 的输入可以是 StartEvent 或 LoopEvent，输出可以是 ProcessingEvent 或 LoopEvent。示例用随机数模拟成功或失败：失败时返回 LoopEvent，流程回到第一步；成功时进入第二步^[1]：

class MultiStepWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent | LoopEvent) -> ProcessingEvent | LoopEvent:
        if random.randint(0, 1) == 0:
            print("Bad thing happened")
            return LoopEvent(loop_output="Back to step one.")
        else:
            print("Good thing happened")
            return ProcessingEvent(intermediate_result="First step complete.")

    @step
    async def step_two(self, ev: ProcessingEvent) -> StopEvent:
        # Use the intermediate result
        final_result = f"Finished processing: {ev.intermediate_result}"
        return StopEvent(result=final_result)

第四步：画出流程图

draw_all_possible_flows 可以把工作流画出来，结果保存为 HTML 文件。示例把 flow.html 作为输出路径^[1]：

from llama_index.utils.workflow import draw_all_possible_flows

w = ... # as defined in the previous section
draw_all_possible_flows(w, "flow.html")

用状态在步骤之间共享数据

有时每个步骤都需要读写同一份数据，这时用状态管理。做法是在步骤函数的一个参数上加上 Context 类型标注，再通过 ctx.store.set 和 ctx.store.get 存取值^[1]：

from llama_index.core.workflow import Context, StartEvent, StopEvent


@step
async def query(self, ctx: Context, ev: StartEvent) -> StopEvent:
    # store query in the context
    await ctx.store.set("query", "What is the capital of France?")

    # do something with context and event
    val = ...

    # retrieve query from the context
    query = await ctx.store.get("query")

    return StopEvent(result=val)

更复杂的细节可以在 LlamaIndex 官方文档的 workflow 部分查到^[1]。

多智能体：用 AgentWorkflow 自动编排

如果不想手写每个步骤，可以用 AgentWorkflow 类创建多智能体工作流。多个智能体可以按各自的专长协作，并把任务交接给彼此^[1]。注意，智能体类要从 llama_index.core.agent.workflow 导入，而不是 llama_index.core.agent^[1]。

构造 AgentWorkflow 时必须指定一个根智能体（root agent）。用户的消息先交给根智能体，它可以直接用自己的工具回答，也可以把任务交给更合适的智能体，或者直接把回复返回给用户^[1]。

下面的例子定义了两个 ReActAgent，一个负责乘法，一个负责加法。工具是普通函数，不必额外包装成 FunctionTool，LlamaIndex 会从函数名和文档字符串中解析出名称和描述^[1]：

from llama_index.core.agent.workflow import AgentWorkflow, ReActAgent
from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI

# Define some tools
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

llm = HuggingFaceInferenceAPI(model_name="Qwen/Qwen2.5-Coder-32B-Instruct")

# we can pass functions directly without FunctionTool -- the fn/docstring are parsed for the name/description
multiply_agent = ReActAgent(
    name="multiply_agent",
    description="Is able to multiply two integers",
    system_prompt="A helpful assistant that can use a tool to multiply numbers.",
    tools=[multiply],
    llm=llm,
)

addition_agent = ReActAgent(
    name="add_agent",
    description="Is able to add two integers",
    system_prompt="A helpful assistant that can use a tool to add numbers.",
    tools=[add],
    llm=llm,
)

# Create the workflow
workflow = AgentWorkflow(
    agents=[multiply_agent, addition_agent],
    root_agent="multiply_agent",
)

# Run the system
response = await workflow.run(user_msg="Can you add 5 and 3?")

给多智能体加上共享状态

AgentWorkflow 可以接收 initial_state，它会在工作流开始前提供一个初始状态字典，所有智能体都能访问。状态保存在工作流上下文的 state 键下，并注入到 state_prompt 中，每条新的用户消息都会带上它^[1]。

下面的例子给两个工具各加了一个计数器，每次调用都会把 num_fn_calls 加一：

from llama_index.core.workflow import Context

# Define some tools
async def add(ctx: Context, a: int, b: int) -> int:
    """Add two numbers."""
    # update our count
    cur_state = await ctx.store.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.store.set("state", cur_state)

    return a + b

async def multiply(ctx: Context, a: int, b: int) -> int:
    """Multiply two numbers."""
    # update our count
    cur_state = await ctx.store.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.store.set("state", cur_state)

    return a * b

workflow = AgentWorkflow(
    agents=[multiply_agent, addition_agent],
    root_agent="multiply_agent",
    initial_state={"num_fn_calls": 0},
    state_prompt="Current state: {state}. User message: {msg}",
)

# run the workflow with context
ctx = Context(workflow)
response = await workflow.run(user_msg="Can you add 5 and 3?", ctx=ctx)

# pull out and inspect the state
state = await ctx.store.get("state")
print(state["num_fn_calls"])

运行结束后，通过同一个 Context 读出状态，就能看到函数被调用了几次^[1]。