LlamaAgents & Workflows

The run-llama/llama-agents examples as an ordered learning path — build, serve, and deploy agent workflows with llama-index-workflows.

The run-llama/llama-agents examples teach one thing end to end: how to build, serve, and deploy agent workflows with llama-index-workflows and the llama-agents-* packages. The examples are ordered — each builds on the last — so this guide follows the same six stages, from your first workflow to a multi-replica Kubernetes deployment.

The mental model: steps + events + Context

A workflow is a Workflow subclass whose methods are @step-decorated async functions. A step consumes one event type (its typed parameter) and returns one or more event types. Steps never call each other directly — they communicate only by pushing and pulling events on asyncio queues.

That single idea is why this is not a DAG: the graph is implicit, derived from event types at runtime, rather than declared up front. A step fires whenever an event of its input type is available, so branching, loops, parallel fan-out, and human-in-the-loop pauses all fall out of the same mechanism — no separate DSL.

python

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

class Greet(Workflow):
    @step
    async def start(self, ctx: Context, ev: StartEvent) -> StopEvent:
        return StopEvent(result=f"Hello, {ev.name}!")

result = await Greet().run(name="Ada")   # -> "Hello, Ada!"

Entry / exit: a step taking StartEvent runs first; returning StopEvent ends the run, and its result is what await workflow.run(...) returns.
Branching: annotate the return as a union — -> Approved | Retry | Escalate — and return whichever event should fire next.
Loops: return an event that routes back to an earlier step.
Context & state: every run gets a Context. ctx.store holds serializable, async-safe state (ctx.store.get/set, get_state(), edit_state()); type it with Context[MyState] (a Pydantic model). Pass a Context into run(ctx=...) to resume a prior run.
Streaming: ctx.write_event_to_stream(ev) inside a step; consume with async for ev in handler.stream_events().
Resources (DI): type a step param Annotated[SomeClient, Resource(factory)] to inject shared dependencies (LLM/DB clients), deduplicated across steps.

The package map

Package	Import	Role
`llama-index-workflows`	`workflows`	the core engine (steps, events, Context)
`llama-index-utils-workflow`	`llama_index.utils.workflow`	graph visualization
`llama-agents-server`	`llama_agents.server`	`WorkflowServer` — serve as REST
`llama-agents-client`	`llama_agents.client`	`WorkflowClient` — call a server
`llama-agents-dbos`	`llama_agents.dbos`	DBOS durability runtime
`llamactl`	CLI	dev / deploy

One wrinkle worth knowing: agent.ipynb uses the legacy namespace from llama_index.core.workflow import ..., while the newer examples use the standalone workflows package. Same concepts, different import path.

1 · Start here

feature_walkthrough.ipynb — the guided tour

The single best place to begin: one notebook that walks through workflows, steps, events, Context, branching, loops, cross-run state, human-in-the-loop, resource injection, and observability. If you read only one example, read this.

python

from workflows import Workflow, step, Context
from workflows.events import (
    Event, StartEvent, StopEvent, InputRequiredEvent, HumanResponseEvent,
)

class HITLRequired(InputRequiredEvent): prompt: str
class Response(HumanResponseEvent): answer: str

class Survey(Workflow):
    @step
    async def ask(self, ctx: Context, ev: StartEvent) -> HITLRequired:
        return HITLRequired(prompt="Approve? (y/n)")

    @step
    async def finish(self, ctx: Context, ev: Response) -> StopEvent:
        return StopEvent(result=ev.answer)

agent.ipynb — an agent is a workflow

Builds a function-calling agent from scratch as a Workflow: the agent loop (call the LLM → run the selected tool → feed the result back) is just steps and events, with memory kept in ctx.store. Tools are plain Python functions wrapped in FunctionTool; the loop continues until the LLM stops requesting tools.

python

from llama_index.core.tools import FunctionTool

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

tools = [FunctionTool.from_defaults(fn=multiply)]

# returns FunctionOutputEvent back into the loop until no tool calls remain.

document_processing.ipynb — a realistic pipeline

Iterative extraction over a real document: parse with LlamaParse, have an LLM propose a JSON extraction schema (validated with retries), pause for a human to approve or revise it, then extract with LlamaExtract. Shows typed state, HITL, and resource DI for the parse/extract/LLM clients in one flow. (Needs LlamaCloud + OpenAI keys.)

2 · Serving workflows as an API

server/ — WorkflowServer

Wrap any workflow as a REST service. Standalone, or mounted inside an existing FastAPI app.

python

from llama_agents.server import WorkflowServer

server = WorkflowServer()
server.add_workflow("greeting", Greet())
await server.serve(host="0.0.0.0", port=8000)

Mount it under an existing FastAPI app instead:

python

from fastapi import FastAPI

app = FastAPI()
app.mount("/wf-server", server.app)   # server.app is a mountable ASGI sub-app
# uvicorn.run(app, host="0.0.0.0", port=8000)

Run a workflow over HTTP — POST /workflows/{name}/run with the start event:

bash

curl -X POST localhost:8000/workflows/greeting/run \
  -d '{"start_event": {"name": "Ada"}}'

client/ — WorkflowClient

Drive a running server from Python, including streaming and human-in-the-loop.

python

from llama_agents.client import WorkflowClient

client = WorkflowClient(base_url="http://localhost:8000")
await client.is_healthy()
await client.list_workflows()

handler = await client.run_workflow_nowait("greeting", start_event={"name": "Ada"})
async for event in client.get_workflow_events(handler.handler_id):
    print(event.type, event.value)

done = await client.get_handler(handler.handler_id)   # .result, .status

For human-in-the-loop, the server pauses on an InputRequiredEvent; the client detects it by event.type, collects input locally, and resumes:

python

await client.send_event(handler_id=handler.handler_id, event=response_event)

3 · Durability and scale

durable_workflows.ipynb — save & resume

Workflows are ephemeral by default: when run() returns, state is gone. The notebook escalates through durability strategies, ending at serialize the whole Context and resume later:

python

from workflows.context import JsonSerializer

handler = wf.run(start_event=ev)
result = await handler
snapshot = handler.ctx.to_dict(serializer=JsonSerializer())   # persist this

# later — rehydrate and continue where it left off
ctx = Context.from_dict(wf, snapshot, serializer=JsonSerializer())
await wf.run(ctx=ctx)

dbos/ — production durability with DBOS

DBOS adds crash recovery, resumable runs, and multi-replica coordination on SQLite (zero setup) or Postgres. You hand the WorkflowServer a DBOS-backed store and runtime; interrupted runs resume automatically.

python

from dbos import DBOS
from llama_agents.dbos import DBOSRuntime

DBOS(config={"name": "counter", "run_admin_server": False})
runtime = DBOSRuntime()
server = WorkflowServer(
    workflow_store=runtime.create_workflow_store(),
    runtime=runtime.build_server_runtime(idle_timeout=5.0),
)

The folder demonstrates four things: single-server crash-and-resume, two replicas sharing one Postgres event store (start on A, stream from B), and idle release — long-idle runs are evicted from memory and auto-resumed on the next event.

4 · Deployment

docker/ — containerize a server

The workflows package ships a server launcher, so the container just points it at a module that defines a top-level server object:

dockerfile

FROM python:3.13-alpine3.22
WORKDIR /app
COPY requirements.txt . && RUN pip install -r requirements.txt
COPY app.py .
CMD ["python", "-m", "workflows.server", "app.py"]

bash

docker build -t workflows-example examples/docker
docker run --rm -p 8000:8000 workflows-example

k8s-otel/ — Kubernetes + OpenTelemetry

The full production shape: a FastAPI app with WorkflowServer mounted at /api, DBOS on Postgres for durability, and OpenTelemetry traces exported via OTLP to an in-cluster collector (fanned out to Phoenix and Jaeger). Two replicas share Postgres so human-in-the-loop runs recover across pods. Local dev uses a kind cluster driven by Tilt (tilt up / tilt down); Phoenix UI at localhost:6006. A PVC keeps Postgres across restarts, and OTLP silently drops spans if the collector is down — so verify the collector first.

5 · Observability and evaluation

observability/ — trace your workflows

Workflows are auto-instrumented, so tracing works out of the box. The notebooks cover the built-in event stream, custom + nested spans with step timing, the built-in context logger for structured per-run logs, and exporting to Arize Phoenix and Langfuse.

python

from phoenix.otel import register
from openinference.instrumentation.llama_index import LlamaIndexInstrumentor

register()                      # point at your Phoenix / OTLP endpoint
LlamaIndexInstrumentor().instrument()
# now every workflow run emits spans automatically

eval_driven_prompt_refinement.ipynb — close the loop

A self-contained example (mock evaluator, no API keys) that combines the core patterns into a refinement loop: generate → evaluate → decide, looping with a conditional exit (retry until pass or max iterations), branching three ways (approve / retry-with-feedback / escalate to a human), and carrying prompt history + scores in typed state.

6 · Advanced patterns

streaming_internal_events.ipynb

Beyond streaming your own events, you can stream the framework's internal events — step state changes, inputs/outputs, queue activity — live, which is what makes rich progress UIs and debuggers possible.

python

from workflows.events import StepStateChanged

handler = wf.run(start_event=ev)
async for ev in handler.stream_events():
    if isinstance(ev, StepStateChanged):
        print(ev)          # step lifecycle transitions as they happen
result = await handler

state_management_with_vector_databases.ipynb

Persist workflow state across sessions by snapshotting Context into a vector DB (Qdrant + sentence-transformers embeddings) — a first step toward workflows that recall past runs, i.e. "self-learning" pipelines.

document_agents/ — a finance triage agent

A single workflow that triages incoming emails with attachments: classify each attachment (invoice vs. expense) with LlamaClassify, extract typed fields with LlamaExtract against Pydantic Invoice/Expense schemas, then act — acknowledge the sender and check expenses against a budget. A compact, realistic multi-service agent.

visualization/ — see the graph

Because the graph is implicit, it helps to render it. llama-index-utils-workflow exports every possible path two ways:

python

from llama_index.utils.workflow import (
    draw_all_possible_flows, draw_all_possible_flows_mermaid,
)

draw_all_possible_flows_mermaid(RAGWorkflow, filename="rag.mermaid")  # paste to mermaid.live
draw_all_possible_flows(RAGWorkflow, filename="rag.html")            # interactive pyvis

Injected resources render as plum hexagons, edges are labeled with the variable name (db, cache, llm), and a shared resource used by several steps is drawn as a single deduplicated node.

Cheat sheet

Workflow / step: class W(Workflow); @step async def s(self, ctx, ev: In) -> Out | Other. Branch = union return; loop = return an event routing back.
Events: subclass Event; StartEvent (entry), StopEvent(result=...) (exit), InputRequiredEvent/HumanResponseEvent (HITL), StepStateChanged (internal).
Context: ctx.write_event_to_stream(ev); ctx.store.get/set, get_state(), edit_state(); Context[StateModel]; resume via run(ctx=...); JsonSerializer.
Handler: h = wf.run(...); await h; h.ctx; async for e in h.stream_events(); await h.cancel_run().
Server: WorkflowServer(workflow_store=?, runtime=?), .add_workflow(name, wf), await .serve(host, port), .app to mount. CLI python -m workflows.server app.py. REST POST /workflows/{name}/run body {"start_event": {...}}.
Client: WorkflowClient(base_url), list_workflows(), run_workflow_nowait(...), get_workflow_events(id), get_handler(id), send_event(id, event=).
DBOS: DBOS(config=...), DBOSRuntime(), .create_workflow_store(), .build_server_runtime(idle_timeout=?).
Visualize: draw_all_possible_flows, draw_all_possible_flows_mermaid.

Where to go next

The examples index: github.com/run-llama/llama-agents/tree/main/examples
Package README: llama-index-workflows (the engine) and the repo root README.
Related deep dives on this site: LangGraph for the graph-first take on the same problems, and Agent Autonomy for the control loop behind long-running agents.

Code above is illustrative of the documented API; run the notebooks for the authoritative, cell-for-cell source.