Orchestrating LLM workflows with LangChain & LangGraph

01Why a graph, not a script

A single prompt works for a single turn. Real tasks need several steps — classify, retrieve, call a tool, validate, maybe loop back and try again. Wiring that as ad-hoc if/else around chained prompts becomes brittle fast: there is no shared state, no clean way to retry one step, and no visibility into why a run went the way it did.

LangGraph models the workflow as a directed graph of nodes (units of work) and edges (what happens next), all operating over a shared, typed state. That makes the flow explicit, inspectable, resumable, and testable — the difference between a demo and something you can operate.

02Key terms in plain English

The vocabulary of a LangGraph workflow, before any code.

03Minimal worked example — the whole thing, broken down

The smallest useful graph: generate, self-check, and loop back a bounded number of times.

• 1. Define state. Carry the question, the draft answer, whether it passed, and an attempt counter.
• 2. Write nodes. One node drafts an answer; one node checks if the draft is good enough.
• 3. Route. A conditional edge decides: finish, retry, or give up after N attempts.
• 4. Wire & compile. Connect the nodes and let the graph run the loop for you.

# The smallest self-correcting graph — generate -> check -> (loop or stop)
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class State(TypedDict):              # 1. shared state
    question: str
    draft: str
    ok: bool
    attempts: int

def generate(state):                # 2a. node: produce a draft
    state["draft"] = llm_answer(state["question"])
    state["attempts"] += 1
    return state

def check(state):                   # 2b. node: self-grade the draft
    state["ok"] = is_good(state["draft"])
    return state

def route(state):                   # 3. conditional edge = control logic
    if state["ok"]:            return END          # good enough -> stop
    if state["attempts"] < 3: return "generate"   # retry (bounded!)
    return END                                       # give up gracefully

g = StateGraph(State)                # 4. wire it together
g.add_node("generate", generate); g.add_node("check", check)
g.add_edge(START, "generate"); g.add_edge("generate", "check")
g.add_conditional_edges("check", route)
app = g.compile()
app.invoke({"question": "Is flood damage covered?", "attempts": 0})

04Example flow — a self-correcting document assistant

A graph that answers a question over documents, checks whether it actually has enough grounding, and loops to refine the query if not — escalating to a human when confidence stays low.

05The core building blocks

06Defining the graph — annotated snippet

Conceptual LangGraph-style pseudocode showing state, nodes and the conditional loop.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List

class State(TypedDict):                 # shared, typed state flows through every node
    question: str
    context: List[str]
    draft: str
    attempts: int
    grounded: bool

def retrieve(state: State) -> State:
    state["context"] = search(state["question"])     # tool/retriever call
    return state

def generate(state: State) -> State:
    state["draft"] = llm_answer(state["question"], state["context"])
    return state

def grade(state: State) -> State:
    state["grounded"] = is_supported(state["draft"], state["context"])  # self-check
    state["attempts"] += 1
    return state

def route(state: State) -> str:           # conditional edge = the control logic
    if state["grounded"]:        return "answer"
    if state["attempts"] < 3:     return "retrieve"   # loop back and refine
    return "human"                                  # give up gracefully → escalate

g = StateGraph(State)
g.add_node("retrieve", retrieve); g.add_node("generate", generate)
g.add_node("grade", grade);       g.add_node("human", human_review)
g.add_edge(START, "retrieve"); g.add_edge("retrieve", "generate"); g.add_edge("generate", "grade")
g.add_conditional_edges("grade", route, {"answer": END, "retrieve": "retrieve", "human": "human"})
app = g.compile(checkpointer=memory)     # checkpointing → resumable + human-in-the-loop

07Design trade-offs

08Observability & LLMOps

09Limitations & next steps

• Honest scope: a PoC-level design walkthrough to show how I structure controllable LLM workflows.
• Builds on RAG: the retrieve/generate nodes reuse the RAG pipeline patterns.
• Toward autonomy: swap fixed routing for an LLM-driven planner when the task space is open-ended — see agentic systems.
• Hardening: add timeouts, tool sandboxing, and per-tenant rate limits before production.