LangGraph vs LlamaIndex: One Agent, Two Architectures

Multi-step LLM workflows need control flow beyond a single chat() call. LangGraph models agents as graph traversal over shared state; LlamaIndex Workflows models them as event-driven @step functions whose type signatures define wiring.

This post builds the same customer-support email router in both frameworks, then adds a three-reply cap with human escalation. Full code: code/.

1 Task definition

Inbound email router:

Classify into billing, support, or sales.
Retrieve top-2 policy snippets from that department’s knowledge base (RAG).
Draft a reply grounded only in retrieved context.

Data: nine hand-written policy snippets (three per department) — refund window, error-code fix, Team plan price. Stand in for help-center or internal wiki text; small enough for a blog post.

Embeddings: text-embedding-3-small → in-memory vector store per department.

Model: gpt-4o-mini (architecture comparison, not model comparison).

Classification asymmetry: LangGraph uses with_structured_output; LlamaIndex uses free-form text + parse fallback — idiomatic per framework, noted below.

2 Agent abstraction

An LLM is stateless: messages in → one assistant message out (answer or tool call).

An agent wraps it with transcript, extra state, tools, and a max-turn stop condition. Only llm.chat() is stochastic.

Code

@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 itself

An 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.

Code

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 None

Frameworks differ in how transcript, routing, and stop conditions are modeled — not in the loop shape above.

3 LangGraph: shared state graph

Nodes are functions; edges (including conditional) define order. One State object flows through the graph.

3.1 State schema

Code

class State(TypedDict):
    email: str
    department: str
    context: str
    draft: str


# Structured output schema so classification is typed, not string-parsed
class RouteDecision(BaseModel):
    department: Literal["billing", "support", "sales"] = Field(
        description="The department best suited to answer this email."
    )

Node signature: State in → partial dict out. LangGraph merges partial updates.

Code

def classify(state: State) -> dict:
    decision = llm.with_structured_output(RouteDecision).invoke(
        "Classify this customer email into one department.\n\n"
        f"EMAIL:\n{state['email']}"
    )
    return {"department": decision.department}


def make_retrieve_node(dept: str):
    """Factory: one retrieval node per department."""

    def retrieve(state: State) -> dict:
        docs = STORES[dept].similarity_search(state["email"], k=2)
        return {"context": "\n\n".join(d.page_content for d in docs)}

    return retrieve

Three retrieval nodes (retrieve_billing, etc.) via factory — branch visible in graph topology for logging/tracing/draw_mermaid().

3.2 Conditional routing

Code

builder.add_conditional_edges(
    "classify",
    route,                      # returns "billing" | "support" | "sales"
    {dept: f"retrieve_{dept}" for dept in DEPARTMENT_DOCS},
)

Branch table declared at graph construction; every destination must be wired.

%%{init: {'theme':'base', 'themeVariables': {'primaryColor':'#ffffff','primaryBorderColor':'#4A3AA7','primaryTextColor':'#1a1a1a','lineColor':'#4A3AA7','edgeLabelBackground':'#ffffff'}}}%%
flowchart LR
    START([START]) --> classify
    classify -->|billing| rb[retrieve_billing]
    classify -->|support| rs[retrieve_support]
    classify -->|sales| rsl[retrieve_sales]
    rb --> draft
    rs --> draft
    rsl --> draft
    draft --> END([END])

4 LlamaIndex Workflows: inferred wiring

No explicit graph object. Steps declare consumed/emitted Event types; framework infers edges from signatures.

4.1 Event types

Code

class RoutedEvent(Event):
    department: str


class ContextReadyEvent(Event):
    department: str
    context: str

@step methods: input/output types define wiring. No add_edge calls.

Code

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)

Single retrieve step branches on ev.department inside the function — not visible in static workflow structure.

4.2 Classification note

LlamaIndex classify uses free-form text + fallback; LangGraph uses structured output. LlamaIndex can use structured output; this script does not.

4.3 Context store

Values not carried on events (e.g. original email) go in ctx.store — async key-value scoped to one workflow run.

Code

@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()})

StopEvent ends the run.

%%{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])

5 v1 comparison

Single-email routers are comparable in length and complexity.

Branching: LangGraph — named nodes + branch dict (inspectable topology). LlamaIndex — if inside one step.
Incidental state: LangGraph — any State key. LlamaIndex — explicit ctx.store.
Validation: LlamaIndex events validated at construction. LangGraph TypedDict documented but not runtime-validated (typos create silent new keys).

6 Reply cap requirement

New rule: max three agent replies per customer thread; fourth inbound email escalates to human inbox.

Demands:

Thread-scoped state across multiple invocations.
Reply counter persistence.
New escalation branch skipping draft.

6.1 LangGraph v2

Code

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 escalation

Annotated[..., add_messages] — reducer appends message updates instead of replacing the list.

New gate node before classify; route_gate reads reply_count:

Code

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 returns {} — anchor node for conditional edge only. classify skips re-routing if department already set.

Code

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: checkpointer + thread_id:

Code

graph = builder.compile(checkpointer=MemorySaver())

thread = {"configurable": {"thread_id": "customer-4482"}}
result = graph.invoke({"messages": [HumanMessage(content=email)]}, thread)

Each invoke sends only the new message; checkpointer restores prior state. MemorySaver is in-process; swap for DB-backed checkpointer at compile().

%%{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])

6.2 LlamaIndex v2

No graph insertion point. Entry step = whichever consumes StartEvent → logic folds into renamed intake:

Code

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)

Branch via return type union RoutedEvent | EscalateEvent — no edge table.

Persistence: reuse same Context per customer; manual ctx.store read/write. No built-in checkpointer; survives only while Context reference is held.

Code

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)

%%{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])

7 v2 comparison

Similar code length; logic lands in different places:

Insertion vs absorption: LangGraph adds gate node; LlamaIndex expands entry intake step.
Persistence: LangGraph checkpointer + thread_id (declarative). LlamaIndex manual Context reuse (imperative).
Memoization: Both require hand-written “classify once” guards; neither auto-memoizes per thread.
Branching: LangGraph branch = runtime dict (drawable). LlamaIndex branch = type union (statically checkable).

8 Selection guide

	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”

LangGraph — multi-turn agents, thread-scoped state, human-in-the-loop, inspectable/replayable graphs.

LlamaIndex Workflows — async event handlers with minimal framework vocabulary; you own persistence.

Framework choice determines whether a future requirement is a one-line edit or a refactor — e.g. reply cap lands as new node vs expanded entry step.

9 References

LangGraph documentation
LlamaIndex Workflows module guide
OpenAI gpt-4o-mini and text-embedding-3-small (used in post code under code/)