130 秒速览

1The 30-second version

一句话:LlamaIndex Workflows 是一个事件驱动、async 优先的 Agent 编排库——你写一个个 @step 异步函数,每个函数“接收某种事件、返回另一种事件”,运行时根据函数参数的类型注解自动把事件送到下一个步骤。分支就是 if,循环就是“返回一个更早的事件”,并发就是“返回一个事件列表”。

In one sentence: LlamaIndex Workflows is an event-driven, async-first library for orchestrating agents. You write @step async functions, each of which “takes one kind of event and returns another”, and the runtime routes events to the next step using the type annotation on the function’s parameter. A branch is an if, a loop is “return an earlier event”, and concurrency is “return a list of events”.

为什么今天讲它:7/7 简报的 Agent 技术精选提到 Workflows 1.0“近期讨论升温”,同时 LangGraph 也在推节点级超时与 Saga 补偿。两者代表了 2026 年 Agent 编排的两种主流思路:显式图(LangGraph)与隐式事件路由(Workflows)。弄懂后者,你就能在“我要一张状态机”与“我要几段可组合的 async 代码”之间做出有根据的选择。

Why today: the 7/7 brief’s agent-tech section noted that discussion of Workflows 1.0 was heating up, while LangGraph was shipping node-level timeouts and saga-style compensation. They represent the two mainstream approaches to agent orchestration in 2026: an explicit graph (LangGraph) and implicit event routing (Workflows). Understanding the latter helps you choose, with reasons, between “I need a state machine” and “I need a few composable async functions”.

2核心概念

2Core concepts

1.0 的三个官方新增点(LlamaIndex 博客):类型化状态(Python 与 TS)、资源注入(Python)、可观测性(安装 llama-index-instrumentation 后接 OpenTelemetry、Arize Phoenix 等)。博客强调架构相对 beta 没有大改,主要变化是独立成项目:不装 llama_index 也能用,老的 import 路径通过再导出继续可用。

The three official additions in 1.0 (LlamaIndex blog): typed state (Python and TS), resource injection (Python), and observability (install llama-index-instrumentation to plug into OpenTelemetry, Arize Phoenix and others). The post stresses that the architecture did not change much from the beta; the main change is that it became a standalone project, usable without llama_index, with the old import paths still working through re-exports.

3怎么用:从 Hello World 到并发扇出

3How to use it: from Hello World to concurrent fan-out

安装与最小示例(来自官方仓库 README):

Install and minimal example (from the official repository README):

pip install llama-index-workflows      # Python 3.10+
npm i @llamaindex/workflow-core       # TypeScript

from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent

class HelloWorkflow(Workflow):
    @step
    async def greet(self, ev: StartEvent) -> StopEvent:
        return StopEvent(result=f"Hello, {ev.name}")

# result = await HelloWorkflow(timeout=60).run(name="Ada")

pip install llama-index-workflows      # Python 3.10+
npm i @llamaindex/workflow-core       # TypeScript

from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent

class HelloWorkflow(Workflow):
    @step
    async def greet(self, ev: StartEvent) -> StopEvent:
        return StopEvent(result=f"Hello, {ev.name}")

# result = await HelloWorkflow(timeout=60).run(name="Ada")

注意 ev.name:StartEvent 的字段直接来自 run(name=...) 的关键字参数。timeout=60 是官方入门示例里的写法,用来避免运行无限挂起。

Note ev.name: StartEvent’s fields come straight from run(name=...) keyword arguments. timeout=60 is how the official getting-started example keeps a run from hanging forever.

并发扇出 + 汇合(官方 concurrent execution 文档的写法):一个步骤用 ctx.send_event 发出多个事件,接收步骤用 num_workers 控制并发,下游用 ctx.collect_events 等齐再处理。

Concurrent fan-out + join (as in the official concurrent-execution docs): one step emits several events with ctx.send_event, the receiving step limits concurrency with num_workers, and a downstream step waits with ctx.collect_events until all have arrived.

class ConcurrentFlow(Workflow):
    @step
    async def start(self, ctx: Context, ev: StartEvent) -> StepTwoEvent | None:
        ctx.send_event(StepTwoEvent(query="Query 1"))
        ctx.send_event(StepTwoEvent(query="Query 2"))
        ctx.send_event(StepTwoEvent(query="Query 3"))

    @step(num_workers=4)
    async def step_two(self, ctx: Context, ev: StepTwoEvent) -> StepThreeEvent:
        return StepThreeEvent(result=await search(ev.query))

    @step
    async def step_three(self, ctx: Context, ev: StepThreeEvent) -> StopEvent | None:
        result = ctx.collect_events(ev, [StepThreeEvent] * 3)   # wait for all 3
        if result is None:
            return None
        return StopEvent(result=result)

class ConcurrentFlow(Workflow):
    @step
    async def start(self, ctx: Context, ev: StartEvent) -> StepTwoEvent | None:
        ctx.send_event(StepTwoEvent(query="Query 1"))
        ctx.send_event(StepTwoEvent(query="Query 2"))
        ctx.send_event(StepTwoEvent(query="Query 3"))

    @step(num_workers=4)
    async def step_two(self, ctx: Context, ev: StepTwoEvent) -> StepThreeEvent:
        return StepThreeEvent(result=await search(ev.query))

    @step
    async def step_three(self, ctx: Context, ev: StepThreeEvent) -> StopEvent | None:
        result = ctx.collect_events(ev, [StepThreeEvent] * 3)   # wait for all 3
        if result is None:
            return None
        return StopEvent(result=result)

更简洁的写法:步骤直接返回 list[Event] 即为一个批次的扇出;下游参数写成 list[XxxEvent] 时,会在整批到齐后只触发一次。官方文档还写明了几条细节:批次元素按完成顺序到达(要固定顺序得自己排序);某个 worker 返回 None 会被剔除,汇合照常触发;用 Collect(Take(n)) 包住列表参数可以“先到 n 个就放行”,但不会取消其余任务。num_workers 只限制该步骤的并发副本数,不限制同时有多少次运行。

A shorter form: a step that returns list[Event] fans out one batch; a downstream parameter typed list[XxxEvent] fires once, after the whole batch has arrived. The official docs add a few details: batch elements arrive in completion order (sort them yourself if order matters); a worker that returns None is dropped and the join still fires; wrapping the list parameter in Collect(Take(n)) releases the join after the first n results but does not cancel the rest. num_workers caps concurrent copies of that step, not how many runs execute at once.

类型化状态(官方 state 文档):多个步骤可能同时改状态时,用 edit_state() 做原子更新,并且把慢的 LLM/网络调用移出加锁块。

Typed state (official state docs): when several steps may change state at once, use edit_state() for atomic updates, and keep slow LLM or network calls out of the locked block.

class CounterState(BaseModel):
    count: int = Field(default=0)

class MyWorkflow(Workflow):
    @step
    async def start(self, ctx: Context[CounterState], ev: StartEvent) -> StopEvent:
        async with ctx.store.edit_state() as state:   # locked read-modify-write
            state.count += 1
        return StopEvent(result="Done!")

class CounterState(BaseModel):
    count: int = Field(default=0)

class MyWorkflow(Workflow):
    @step
    async def start(self, ctx: Context[CounterState], ev: StartEvent) -> StopEvent:
        async with ctx.store.edit_state() as state:   # locked read-modify-write
            state.count += 1
        return StopEvent(result="Done!")

4人在回路与持久化

4Human in the loop and durability

人在回路(HITL):官方推荐“两步式”——一个步骤返回 InputRequiredEvent,另一个步骤接收 HumanResponseEvent;调用方在事件流里看到“需要输入”时,把回复送回同一次运行。

Human in the loop (HITL): the docs recommend a two-step pattern. One step returns InputRequiredEvent and another accepts HumanResponseEvent; when the caller sees “input required” in the event stream, it sends the reply back into the same run.

class NumberWorkflow(Workflow):
    @step
    async def ask(self, ev: StartEvent) -> InputRequiredEvent:
        return InputRequiredEvent(prefix="Enter a number: ")

    @step
    async def answer(self, ev: HumanResponseEvent) -> StopEvent:
        return StopEvent(result=ev.response)

handler = workflow.run()
async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        await handler.send_event(HumanResponseEvent(response=input(event.prefix)))
final_result = await handler

class NumberWorkflow(Workflow):
    @step
    async def ask(self, ev: StartEvent) -> InputRequiredEvent:
        return InputRequiredEvent(prefix="Enter a number: ")

    @step
    async def answer(self, ev: HumanResponseEvent) -> StopEvent:
        return StopEvent(result=ev.response)

handler = workflow.run()
async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        await handler.send_event(HumanResponseEvent(response=input(event.prefix)))
final_result = await handler

如果提问和回答发生在不同的请求或进程里(现实中的审批流程大多如此),就要把上下文快照下来:提问后保存 handler.ctx.to_dict()(JSON),再 handler.cancel_run(),避免内存里留一个永远等不到答案的运行;答案到了以后用 Context.from_dict(workflow, d) 恢复、workflow.run(ctx=ctx) 继续,再送入 HumanResponseEvent。

If the question and the answer happen in different requests or processes (as most real approval flows do), snapshot the context: after asking, save handler.ctx.to_dict() (JSON) and call handler.cancel_run() so no in-memory run is left waiting for an answer that will never reach it; when the answer arrives, restore with Context.from_dict(workflow, d), continue with workflow.run(ctx=ctx), and send the HumanResponseEvent.

持久化(durability):官方文档明说工作流“默认是易失的”,run() 返回后状态就没了,而且没有需要打开的内置 checkpointer。做法是在运行中订阅内部事件,每当某步骤结束(StepStateChanged 且 StepState.NOT_RUNNING)就存一次快照。恢复时:已完成的步骤不重跑,挂起的事件会重新派发、汇合缓冲会重建,中断时正在执行的步骤会被回滚重跑——即“至少一次”语义。

Durability: the docs say plainly that workflows are “ephemeral by default”: state is gone once run() returns, and there is no built-in checkpointer to switch on. Instead you subscribe to internal events during the run and save a snapshot whenever a step finishes (StepStateChanged with StepState.NOT_RUNNING). On restore, finished steps do not re-run, pending events are re-dispatched and join buffers rebuilt, and steps that were mid-execution are rewound and run again: at-least-once semantics.

db.save("my-run", json.dumps(handler.ctx.to_dict()))
ctx = Context.from_dict(w, json.loads(db.load("my-run")))
result = await w.run(ctx=ctx)

db.save("my-run", json.dumps(handler.ctx.to_dict()))
ctx = Context.from_dict(w, json.loads(db.load("my-run")))
result = await w.run(ctx=ctx)

不想手写检查点循环?官方提供 DBOS 运行时插件:替换默认的进程内运行时,把步骤状态变化记入数据库,崩溃后自动恢复,工作流代码不用改;WorkflowServer 也能挂同一个插件。

Don’t want to hand-write the checkpoint loop? The docs offer a DBOS runtime plugin that replaces the default in-process runtime, journals step transitions to a database and recovers after a crash without changes to workflow code; WorkflowServer can use the same plugin.

5多智能体、观测与部署

5Multi-agent, observability and deployment

6对比:Workflows vs LangGraph

6Comparison: Workflows vs LangGraph

下表综合 Alice Labs 指南(社区/咨询公司观点)与官方文档;“最适合”一行是评论性判断,不是事实。

The table combines the Alice Labs guide (a consultancy’s view) with the official docs; the “best fit” row is opinion, not fact.

JetBrains PyCharm 博客(2026-06)的框架盘点把 LlamaIndex 定位为“以数据/检索为中心”:索引和长期记忆强,复杂的重动作编排与多智能体支持有限;LangGraph 控制力、HITL 与多智能体更强。Alice Labs 则建议两者可以组合使用:外层循环用 LangGraph,RAG 子任务交给一个 Workflow。

JetBrains’ PyCharm blog roundup (June 2026) positions LlamaIndex as “data/retrieval-centric”: strong indexing and long-term memory, limited for complex action-heavy orchestration and multi-agent setups, while LangGraph offers more control, HITL and multi-agent support. Alice Labs suggests combining them: LangGraph for the outer loop, a Workflow for a RAG sub-task.

7坑与注意事项

7Pitfalls

官方文档明确写到的:

Stated in the official docs:

8资源与来源

8Resources and sources