Here is one agent: an inbound customer-support email router. Classify the email into a department (billing, support, or sales), retrieve that department’s policy knowledge with RAG, and draft a grounded reply. It’s a small enough task to hold in your head completely, which makes it a good instrument for a different question — not “which framework is more powerful,” but what does each framework’s core abstraction make easy, and what does it make hard?
LangGraph models an agent as a graph traversal over shared state: nodes are plain functions, edges (some conditional) decide what runs next, and a single mutable state object flows through the whole thing. LlamaIndex Workflows models the same kind of agent as event-driven steps: @step-decorated async functions declare what event type they consume and what they emit, and the framework wires the graph implicitly from those type annotations.
This post builds the router twice — once in each framework — then raises the stakes with a second requirement (a 3-reply cap with human escalation) and watches where each architecture bends easily and where it doesn’t. The full code for all four versions lives in code/ alongside this post.
Before comparing two concrete frameworks, it’s worth being precise about what “agent” means, because the LLM itself does surprisingly little. A language model is a stateless function from a message sequence to one new message: hand it the conversation so far, it returns the next assistant message — which is either a final answer or a request to call a tool. It has no memory of having been called a moment ago, and it never executes anything; it only ever emits text.
@dataclass
class Message:
role: str # "system" | "user" | "assistant" | "tool"
content: str
tool_call: dict | None = None # e.g. {"name": "lookup_refund", "args": {...}}
class LLM:
"""Stateless: a sequence of messages in, ONE new message out."""
def chat(self, messages: list[Message]) -> Message:
... # one forward pass; the model never executes anything itselfAn agent is the deterministic machinery wrapped around that stochastic call. Concretely, it’s a class holding four things — a growing message sequence (the conversation transcript, which is the model’s only memory), any extra state the task needs, a registry of tools the model is allowed to request, and a stop condition with a max-turn cap so the loop provably terminates. Its run method is a plain deterministic loop: the only nondeterminism anywhere is inside llm.chat; everything else — append, dispatch, check, repeat — is ordinary code you could unit-test with a mocked model.
class Agent:
"""Deterministic loop around a stochastic LLM call."""
def __init__(self, llm: LLM, tools: dict[str, Callable], max_turns: int = 10):
self.llm = llm
self.tools = tools # name -> callable the agent may run
self.messages: list[Message] = [] # the transcript IS the memory
self.state: dict = {} # task state beyond the transcript
self.max_turns = max_turns
def run(self, user_input: str) -> Message:
self.messages.append(Message(role="user", content=user_input))
for _ in range(self.max_turns):
reply = self.llm.chat(self.messages) # the ONLY stochastic step
self.messages.append(reply)
if self.is_done(reply): # stop condition: no tool
return reply # requested -> final answer
result = self.tools[reply.tool_call["name"]](**reply.tool_call["args"])
self.messages.append(Message(role="tool", content=str(result)))
return self.escalate() # max turns hit: fail safely
def is_done(self, reply: Message) -> bool:
return reply.tool_call is NoneNothing in that sketch is specific to LangGraph or LlamaIndex — every agent framework is selling you some arrangement of those same four ingredients plus that same loop. Where they genuinely differ is in how each ingredient is modeled: is the message sequence one shared mutable list or typed payloads passed between handlers? Is “what happens next” a declared edge in a graph or inferred from type annotations? Is the stop condition a special END node or a sentinel event? That’s precisely the axis the rest of this post walks through, on one concrete agent — and the reply-cap requirement added halfway down is nothing more than max_turns showing up as a business rule.
Both versions solve the identical problem:
billing, support, or sales.Both scripts use the same gpt-4o-mini model and the same three department knowledge bases — the comparison below is about architecture, not model choice. The one deliberate asymmetry is classification itself: LangGraph forces a typed, schema-validated decision via with_structured_output; LlamaIndex asks for free-form text and parses it by hand with a string fallback. That’s not a fairness gap — it’s each framework’s idiomatic way of getting a department out of the model, and it’s called out again where it appears below.
LlamaIndex has no graph object to build. Instead, custom Event subclasses describe the payloads that move between steps:
class RoutedEvent(Event):
department: str
class ContextReadyEvent(Event):
department: str
context: strA @step method declares what it consumes and what it returns in its type signature — that’s the entire wiring mechanism. There’s no separate builder.add_edge(...) call anywhere; the framework infers the graph from which steps consume which event types.
class EmailRouterWorkflow(Workflow):
@step
async def classify(self, ctx: Context, ev: StartEvent) -> RoutedEvent:
email: str = ev.email
await ctx.store.set("email", email) # stash for later steps
resp = await Settings.llm.acomplete(
"Classify this customer email into exactly one department: "
"billing, support, or sales.\n"
"Reply with only the department name, lowercase.\n\n"
f"EMAIL:\n{email}"
)
dept = resp.text.strip().lower()
if dept not in INDEXES:
dept = "support"
return RoutedEvent(department=dept)
@step
async def retrieve(self, ctx: Context, ev: RoutedEvent) -> ContextReadyEvent:
email: str = await ctx.store.get("email")
retriever = INDEXES[ev.department].as_retriever(similarity_top_k=2)
nodes = await retriever.aretrieve(email)
context = "\n\n".join(n.get_content() for n in nodes)
return ContextReadyEvent(department=ev.department, context=context)retrieve is a single step handling all three departments, dispatching on ev.department as ordinary data (INDEXES[ev.department]). Where LangGraph’s author reached for three named nodes plus a branch table, LlamaIndex’s plain-Python dict lookup inside one step does the same job — and the department fan-out is invisible from outside the step; nothing in the workflow’s static structure shows three branches.
Also notice classify here isn’t using structured output the way LangGraph’s version does: it asks the model for a bare lowercase word and parses resp.text by hand, with dept not in INDEXES as a manual fallback to "support". Nothing in LlamaIndex prevents with_structured_output-style validation — this script just didn’t reach for it, so a truly apples-to-apples pair would need one of the two updated to match.
email has to survive from classify to retrieve to draft, but it isn’t part of any event payload after classify emits RoutedEvent. ctx.store is the escape hatch — a plain async key-value store scoped to the running workflow, for exactly this kind of “not really an event, just some state I need later” data:
@step
async def draft(self, ctx: Context, ev: ContextReadyEvent) -> StopEvent:
email: str = await ctx.store.get("email")
resp = await Settings.llm.acomplete(
f"You are a {ev.department} representative. Draft a short, "
"friendly reply to the customer email below. Base your answer "
"ONLY on the company knowledge provided.\n\n"
f"COMPANY KNOWLEDGE:\n{ev.context}\n\n"
f"CUSTOMER EMAIL:\n{email}\n\n"
"REPLY:"
)
return StopEvent(result={"department": ev.department, "draft": resp.text.strip()})%%{init: {'theme':'base', 'themeVariables': {'primaryColor':'#ffffff','primaryBorderColor':'#4A3AA7','primaryTextColor':'#1a1a1a','lineColor':'#4A3AA7','edgeLabelBackground':'#ffffff'}}}%%
flowchart LR
Start([StartEvent: email]) -->|consumes| classify
classify -->|RoutedEvent| retrieve
retrieve -->|ContextReadyEvent| draft
draft --> Stop([StopEvent: result])
At this stateless, single-turn scale the two are close to a wash:
email, needed three steps later) is implicit in LangGraph — it’s just another key in the shared State dict, present everywhere. LlamaIndex requires an explicit ctx.store.set / ctx.store.get pair — more typing, but it’s clear at each call site exactly what a step depends on beyond its declared event.RoutedEvent(department=...) is validated at construction. LangGraph’s State is a TypedDict — structurally documented, but not validated at runtime; a node returning {"deprtment": "billing"} (typo) would silently create a new, unused key rather than raise.Neither architecture is under real pressure yet, because nothing here spans more than one email. The next requirement changes that.
New business rule: the agent may reply to a customer thread at most 3 times. On the 4th inbound email, stop drafting and forward the whole thread to the department inbox for a human.
This is no longer a single-shot function — it’s a thread that spans multiple invocations, needs to remember how many times it has already replied, and needs a genuinely new control-flow branch (escalate) that didn’t exist before.
The state schema grows a message list (with a reducer), a reply counter, and escalation fields:
class State(TypedDict, total=False):
messages: Annotated[list[BaseMessage], add_messages] # full thread
department: str
context: str
reply_count: int
escalated: bool
forward: str # the internal forward composed on escalationAnnotated[list[BaseMessage], add_messages] is a reducer: instead of a node’s return value replacing state["messages"], LangGraph appends to it. That’s the graph declaring, once, how updates to this field should merge — every node can just return {"messages": [new_message]} without knowing or caring about the existing list.
The reply-cap check becomes a new node, gate, wired in before the original entry point:
def gate(state: State) -> dict:
"""Anchor node for the reply-cap branch. No state change."""
return {}
def route_gate(state: State) -> str:
if state.get("reply_count", 0) >= MAX_AGENT_REPLIES:
return "escalate"
return "classify"gate does no work — its only job is to give route_gate’s conditional edge a node to attach to. The original classify node is otherwise untouched, aside from one guard so it doesn’t re-classify a thread that’s already been routed:
def classify(state: State) -> dict:
if state.get("department"): # already routed on a previous turn
return {}
decision = llm.with_structured_output(RouteDecision).invoke(...)
return {"department": decision.department}Cross-invocation persistence — remembering reply_count and department from one graph.invoke() to the next — is handled by compiling with a checkpointer and passing a thread_id:
graph = builder.compile(checkpointer=MemorySaver())
thread = {"configurable": {"thread_id": "customer-4482"}}
result = graph.invoke({"messages": [HumanMessage(content=email)]}, thread)Each call only sends the new inbound message; the checkpointer restores everything else from the last checkpoint under that thread_id, merges the new message in via the add_messages reducer, and persists the result.
%%{init: {'theme':'base', 'themeVariables': {'primaryColor':'#ffffff','primaryBorderColor':'#4A3AA7','primaryTextColor':'#1a1a1a','lineColor':'#4A3AA7','edgeLabelBackground':'#ffffff'}}}%%
flowchart LR
START([START]) --> gate
gate -->|reply_count >= 3| escalate
gate -->|else| classify
classify -->|billing| rb[retrieve_billing]
classify -->|support| rs[retrieve_support]
classify -->|sales| rsl[retrieve_sales]
rb --> draft
rs --> draft
rsl --> draft
draft --> END1([END])
escalate --> END2([END])
LlamaIndex has no separate “graph” to insert a node into. The natural entry point for a workflow is whichever step consumes StartEvent — and there was already exactly one of those (classify). So the reply-cap logic doesn’t get a new node in front; it gets folded into the entry step itself, renamed intake:
class EmailThreadWorkflow(Workflow):
@step
async def intake(
self, ctx: Context, ev: StartEvent
) -> RoutedEvent | EscalateEvent:
"""Record the inbound email, enforce the reply cap, route."""
email: str = ev.email
history: list = await ctx.store.get("history", default=[])
history.append({"role": "customer", "content": email})
await ctx.store.set("history", history)
reply_count: int = await ctx.store.get("reply_count", default=0)
department: str | None = await ctx.store.get("department", default=None)
if reply_count >= MAX_AGENT_REPLIES:
return EscalateEvent(department=department or "support")
if department is None: # classify once per thread; then it sticks
resp = await Settings.llm.acomplete(...)
department = resp.text.strip().lower()
if department not in INDEXES:
department = "support"
await ctx.store.set("department", department)
return RoutedEvent(department=department)The return type RoutedEvent | EscalateEvent — a plain Python union — is what makes this a branching step: LlamaIndex infers from the annotation that intake can lead to either retrieve (which consumes RoutedEvent) or a new escalate step (which consumes EscalateEvent). No separate branch table to populate; the type checker’s union is the branch declaration.
Persistence across wf.run() calls needs no framework-level concept like a checkpointer at all — it’s just Python object reuse. The same Context instance is passed into every call for a given customer:
wf = EmailThreadWorkflow(timeout=120, verbose=False)
ctx = Context(wf) # ONE context = ONE customer thread; reuse across runs
for email in inbound:
result = await wf.run(email=email, ctx=ctx)ctx.store holds history, reply_count, and department explicitly — nothing is restored automatically the way LangGraph’s checkpointer restores State. Every step that needs prior data fetches it with await ctx.store.get(...).
%%{init: {'theme':'base', 'themeVariables': {'primaryColor':'#ffffff','primaryBorderColor':'#4A3AA7','primaryTextColor':'#1a1a1a','lineColor':'#4A3AA7','edgeLabelBackground':'#ffffff'}}}%%
flowchart LR
Start([StartEvent: email]) --> intake
intake -->|"EscalateEvent (reply_count >= 3)"| escalate
intake -->|"RoutedEvent (else)"| retrieve
retrieve -->|ContextReadyEvent| draft
draft --> Stop1([StopEvent: reply])
escalate --> Stop2([StopEvent: escalated])
Both versions ended up almost the same length — this was never a “one framework needs less code” story. The differences are structural, and they trace directly back to what each core abstraction treats as a first-class concept:
Where the new logic went. LangGraph’s graph has an explicit shape you can insert into: gate slots in before classify as a new named node with its own edge, and classify needed only a one-line memoization guard. LlamaIndex’s entry point is defined by “whichever step consumes StartEvent,” and there’s only one natural place for that — so the new gating logic had nowhere to go but into the existing entry step, which is why intake is a visibly heavier function than the classify it replaced. Insertion vs. absorption is a direct consequence of one framework having an explicit graph object to edit and the other not.
What persistence costs. LangGraph’s checkpointer + thread_id is opt-in infrastructure decided once, at compile() time; after that, every node just returns partial updates and trusts the reducer (add_messages) and the checkpointer to do the merging and restoring. LlamaIndex’s Context is an explicit object you must remember to create once and thread through every wf.run() call yourself, and every field it holds is read and written by hand via ctx.store. The first is a declarative promise made to the framework; the second is an imperative promise you keep yourself.
What memoization costs. Interestingly, this part is equally manual in both — classify’s if state.get("department"): return {} and intake’s if department is None: are the same guard, hand-written in both frameworks. Neither one auto-memoizes a step’s work across invocations of the same thread; that logic is on you either way.
Branching mechanics. LangGraph’s new branch is a dict ({"escalate": "escalate", "classify": "classify"}) — a runtime data structure inspectable at graph-build time. LlamaIndex’s new branch is a type annotation (RoutedEvent | EscalateEvent) — a compile-time (or at least static-analysis-time) contract with no runtime object representing “the graph” at all.
| LangGraph | LlamaIndex Workflows | |
|---|---|---|
| Core abstraction | Graph of nodes + edges over shared, typed-dict state | Async steps wired implicitly by the Event types they consume/emit |
| Branching declared as | A dict passed to add_conditional_edges |
A Union return-type annotation |
| Incidental state | Implicit — any key on the shared State |
Explicit — await ctx.store.get/set |
| Cross-invocation persistence | First-class: checkpointer + thread_id, restored automatically |
Manual: reuse the same Context object yourself |
| Inserting logic before an existing entry point | Add a node + edge; original node barely changes | Folds into the existing entry step; that step grows |
| Topology you can inspect/draw | Yes — graph.get_graph().draw_mermaid() |
No separate graph object exists |
| New vocabulary on top of Python | Nodes, edges, conditional edges, reducers, checkpointers | Events, steps — much closer to “just async functions” |
For a single-turn, stateless task like the v1 router, the choice barely matters — both are a thin layer over “call an LLM, do some routing, call an LLM again.” The gap opens once state has to survive across turns and a new branch has to be threaded in without disturbing what’s already there. Reach for LangGraph when the agent is genuinely multi-turn and thread-scoped state is core to the problem (this email router with a reply cap, a human-in-the-loop review queue, anything where you want to pause, inspect, and resume a specific conversation by ID) — the checkpointer earns its keep, and having an actual graph object pays off the moment you want to visualize, replay, or time-travel-debug a run. Reach for LlamaIndex Workflows when the mental model is closer to a set of async functions reacting to typed messages than to named states and transitions — fewer new concepts to learn on top of plain Python and asyncio, at the cost of hand-rolling whatever persistence the task needs via Context.
TypedDict, a reducer) at the cost of a larger vocabulary (nodes, edges, conditional edges, checkpointers).Context store, staying closer to plain asyncio at the cost of manual state threading.