TL;DR:

  • LlamaIndex Workflows is an event-driven framework for building multi-step AI pipelines where each step receives typed events, processes them, and emits new events — giving you explicit state and control flow without a central orchestrator
  • Workflows handle async natively, support human-in-the-loop interrupts, and can serialize state for long-running tasks that survive restarts
  • Best suited for sequential-to-branching pipelines where you want strong typing and observability; for fully autonomous agent loops, LangGraph or raw agent frameworks may fit better

When LlamaIndex first became popular, people used it primarily for RAG — chunking, indexing, querying. Over time it grew into a broader agent framework, but the primitives felt like they were designed for retrieval-first workflows. Building anything with complex branching or multi-step coordination required wiring together callbacks and query pipelines in ways that got messy quickly.

Workflows is LlamaIndex’s architectural response to that problem. Released in 2024 and significantly matured through 2025, it’s a genuinely different way to think about agent pipeline structure — one that’s worth understanding even if you end up not using LlamaIndex for retrieval.

The Core Abstraction: Events and Steps

A LlamaIndex Workflow is a collection of steps connected by events. Each step is a Python function decorated with @step, and it declares what event types it handles and what events it emits. The runtime dispatches events to the appropriate steps automatically.

from llama_index.core.workflow import (
    Workflow,
    StartEvent,
    StopEvent,
    Event,
    step,
    Context,
)

class QueryEvent(Event):
    query: str

class ResultEvent(Event):
    result: str
    confidence: float

class ResearchWorkflow(Workflow):

    @step
    async def handle_start(self, ctx: Context, ev: StartEvent) -> QueryEvent:
        # Extract the user query and kick off the pipeline
        return QueryEvent(query=ev.get("query"))

    @step
    async def research(self, ctx: Context, ev: QueryEvent) -> ResultEvent:
        # Do retrieval or tool calls here
        result = await self.run_retrieval(ev.query)
        return ResultEvent(result=result, confidence=0.85)

    @step
    async def finalize(self, ctx: Context, ev: ResultEvent) -> StopEvent:
        return StopEvent(result=ev.result)

The explicit event types give you something that most agent frameworks don’t: self-documenting data flow. You can read a Workflow class and understand exactly what information moves between steps without tracing through callback chains or reading orchestrator logic.

State Management with Context

The Context object passed to every step is the shared state container for a workflow run. You can read and write arbitrary values to it, and those values persist across steps within the same run.

@step
async def gather_context(self, ctx: Context, ev: StartEvent) -> QueryEvent:
    # Store metadata that later steps might need
    await ctx.set("user_id", ev.get("user_id"))
    await ctx.set("session_start", datetime.now().isoformat())
    return QueryEvent(query=ev.get("query"))

@step
async def log_result(self, ctx: Context, ev: ResultEvent) -> StopEvent:
    user_id = await ctx.get("user_id")
    # user_id is available here even though this step doesn't receive it directly
    await self.log_to_db(user_id, ev.result)
    return StopEvent(result=ev.result)

This is a cleaner model than passing state through event payloads or relying on closures. The Context is also what enables workflow serialization — you can checkpoint mid-run and resume later, which matters for long-running pipelines.

Branching and Fan-Out

Workflows support branching through conditional event emission. A step can return different event types based on logic, and downstream steps are only triggered when their corresponding event arrives.

class HighConfidenceEvent(Event):
    result: str

class LowConfidenceEvent(Event):
    result: str
    needs_human: bool = True

class TriageStep(Workflow):

    @step
    async def triage(
        self, ctx: Context, ev: ResultEvent
    ) -> HighConfidenceEvent | LowConfidenceEvent:
        if ev.confidence > 0.8:
            return HighConfidenceEvent(result=ev.result)
        else:
            return LowConfidenceEvent(result=ev.result, needs_human=True)

    @step
    async def handle_high_confidence(
        self, ctx: Context, ev: HighConfidenceEvent
    ) -> StopEvent:
        return StopEvent(result=ev.result)

    @step
    async def request_human_review(
        self, ctx: Context, ev: LowConfidenceEvent
    ) -> StopEvent:
        # In practice, this might write to a queue and pause the workflow
        return StopEvent(result=f"[NEEDS REVIEW] {ev.result}")

Union return types (HighConfidenceEvent | LowConfidenceEvent) communicate the branching to both the runtime and to anyone reading the code. The type checker can verify that all event types in a union are handled somewhere in the workflow.

Fan-out — emitting multiple events from a single step — is also supported. Combined with ctx.collect_events(), this enables map-reduce patterns where a step waits for all parallel sub-tasks to complete before proceeding.

class SubtaskEvent(Event):
    task_id: int
    result: str

class AggregatorStep(Workflow):
    NUM_SUBTASKS = 3

    @step
    async def aggregate(self, ctx: Context, ev: SubtaskEvent) -> StopEvent | None:
        # Collect events until we have all three
        results = ctx.collect_events(ev, [SubtaskEvent] * self.NUM_SUBTASKS)
        if results is None:
            return None  # Still waiting for more
        combined = " | ".join(r.result for r in results)
        return StopEvent(result=combined)

Human-in-the-Loop Interrupts

For workflows that need human approval or input mid-run, LlamaIndex Workflows supports explicit input events — a step can pause and wait for an externally-supplied event rather than one emitted by another step.

class HumanApprovalEvent(Event):
    approved: bool
    feedback: str = ""

class ApprovableWorkflow(Workflow):

    @step
    async def draft_response(self, ctx: Context, ev: QueryEvent) -> HumanApprovalEvent:
        draft = await self.generate_draft(ev.query)
        await ctx.set("draft", draft)
        # This emits an event that external code must handle
        # The workflow pauses here until HumanApprovalEvent is injected
        return HumanApprovalEvent(approved=False)  # placeholder; real approval comes externally

    @step
    async def finalize(self, ctx: Context, ev: HumanApprovalEvent) -> StopEvent:
        if not ev.approved:
            raise WorkflowRejectedError(ev.feedback)
        draft = await ctx.get("draft")
        return StopEvent(result=draft)

In practice, you’d use workflow.run_step() instead of workflow.run() for interrupt-driven flows, allowing your application layer to inject events between steps. This works well for approval workflows, content moderation queues, and any pipeline where a human needs to review intermediate results.

Running a Workflow

import asyncio

async def main():
    workflow = ResearchWorkflow(timeout=60, verbose=True)
    result = await workflow.run(query="What are the main risks of agentic AI systems?")
    print(result)

asyncio.run(main())

Workflows are async-native throughout. For synchronous contexts, you can use asyncio.run() as above. LlamaIndex also provides integrations for running workflows in FastAPI endpoints, which is the typical deployment pattern for production use.

Observability

Workflows emit structured events during execution that integrate with LlamaIndex’s instrumentation layer. If you’re using LangFuse, Arize, or another observability platform through LlamaIndex’s callback system, workflow step executions appear as distinct spans with input/output events logged automatically.

from llama_index.core import Settings
from llama_index.core.callbacks import CallbackManager, LlamaDebugHandler

debug_handler = LlamaDebugHandler(print_trace_on_end=True)
Settings.callback_manager = CallbackManager([debug_handler])

# Your workflow will now emit structured traces
workflow = ResearchWorkflow(timeout=60)
result = await workflow.run(query="...")

This gives you step-level latency, event payloads, and token consumption per step without adding custom instrumentation.

When to Use Workflows vs Alternatives

Use LlamaIndex Workflows when:

  • Your pipeline has a clear, enumerable set of steps with well-defined data moving between them
  • You’re already using LlamaIndex for retrieval and want to keep the stack consistent
  • You need checkpoint/resume for long-running tasks
  • Strong typing between pipeline stages matters for your team

Prefer LangGraph when:

  • You need arbitrary graph topologies with cycles
  • You’re building fully autonomous agents that need to decide their own next action
  • You have complex state machines with many possible transitions

Prefer raw agent loops (OpenAI Agents SDK, Anthropic tool use) when:

  • The pipeline is simple enough that a framework would add overhead
  • You’re building a tightly-scoped single-task agent
  • Framework flexibility matters more than structure

Installation

pip install llama-index-core

# For LLM integrations
pip install llama-index-llms-openai
pip install llama-index-llms-anthropic

Workflows are part of llama-index-core — no additional packages required unless your steps use specific integrations.

Practical Patterns

A few patterns that emerge in production Workflows deployments:

Step timeout handling: Wrap slow steps in asyncio.wait_for() and emit error events to handle timeouts gracefully without terminating the whole workflow.

Retry logic: Steps are plain Python async functions, so tenacity retry decorators work directly. Combine with context storage to avoid re-running expensive steps.

External tool calls: Return an intermediate event from a step, have a “tool execution” step handle the actual HTTP request or subprocess call, then return the result event. This keeps network I/O out of your business logic steps.

Parallel sub-workflows: Run multiple workflow instances concurrently with asyncio.gather() and aggregate results in your application layer. Each workflow run is independent with its own Context.

Workflows occupy a practical middle ground in the Python agent framework ecosystem — more structured than raw tool-calling loops, less complex than LangGraph’s full graph execution model. For teams building multi-step pipelines where data shape matters as much as control flow, it’s a solid choice.