LangGraph Workflows

This content was generated with the assistance of AI and should be verified against the official documentation of each tool and framework before being relied on in production.

This section’s bibliography lists the reference material consulted while preparing these pages.

This page was written against langchain 1.4.3, langchain-core 1.6.5, langgraph 1.2.12 and langchain-ollama 1.1.0 on Python 3.12, as verified on 2026-09-29. The running example is DocsAssistant, which answers questions over this site’s documentation using Ollama’s llama3.1:8b chat model and the nomic-embed-text embedding model over the pgvector store built in Integrating with LangChain. Ollama was not available while the snippets were checked, so every output shown here was produced by running the code against the real packages with a small scripted fake chat model and an in-memory stub in place of the pgvector retriever; no output comes from a live model, and a real model will phrase things differently.

A workflow is a system where LLM calls and tools are orchestrated through predefined code paths, as opposed to an agent, where the model chooses its own path. The framework-neutral theory — what each pattern is for, when to prefer it over an agent, and how the patterns compose — lives in Workflow Patterns. This page is the LangGraph implementation of those patterns, plus self-corrective RAG: for each one, the shape of the graph, a runnable example on DocsAssistant, and the LangGraph feature that makes it work. It assumes the vocabulary from LangGraph Fundamentals (state, nodes, edges, reducers, conditional edges, Send).

Setup: a fake model for every example

Every example imports the same tiny fake chat model from a local fake.py. It replies by applying a function to the text of the last message, which keeps parallel branches deterministic (a queue of canned replies would be consumed in whatever order threads happen to run):

fake.py — a scripted chat model used by every example on this page

from collections.abc import Callable
from typing import Any

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.outputs import ChatGeneration, ChatResult


class RuleChat(BaseChatModel):
    """A fake chat model: a function maps the last message's text to a reply."""

    rule: Callable[[str], str]

    @property
    def _llm_type(self) -> str:
        return "rule-fake"

    def _generate(self, messages: list[BaseMessage], stop=None, run_manager=None, **kwargs: Any) -> ChatResult:
        reply = self.rule(messages[-1].text)
        return ChatResult(generations=[ChatGeneration(message=AIMessage(content=reply))])

To run an example against a real model, replace the RuleChat(…) line with Ollama and, where a snippet parses JSON from the model, use structured output instead of the parser:

Swapping in Ollama and real structured output

from langchain_ollama import ChatOllama

llm = ChatOllama(model="llama3.1:8b", temperature=0)
router = llm.with_structured_output(Route)     # instead of: llm | PydanticOutputParser(pydantic_object=Route)

See ChatOllama and Structured Output. The examples parse the fake model’s JSON text with PydanticOutputParser only because a fake chat model has no native structured-output support.

Prompt chaining

Prompt chaining decomposes a task into a fixed sequence of LLM calls, each consuming the previous call’s output, with optional programmatic gates between steps. In LangGraph this is the plainest graph there is: nodes joined by edges, with a conditional edge wherever a gate can stop the chain early. DocsAssistant’s chain rewrites the question into a search query, retrieves, checks that anything came back, and only then drafts and polishes an answer.

flowchart LR S([START]) --> R[rewrite] R --> V[retrieve] V --> G{context found?} G -->|found| D[draft] D --> P[polish] P --> E([END]) G -->|empty| N[no_answer] N --> E

Figure 1. Prompt chaining with a conditional gate

Prompt chaining with a gate (Graph API)

from typing_extensions import TypedDict

from langgraph.graph import END, START, StateGraph

from fake import RuleChat


def reply(prompt: str) -> str:
    if prompt.startswith("Rewrite"):
        return "weather today" if "weather" in prompt else "HNSW ef_search recall latency pgvector"
    if prompt.startswith("Draft"):
        return "Raise ef_search to trade latency for recall [1]."
    return "Raising hnsw.ef_search trades latency for recall (see pgvector page) [1]."


llm = RuleChat(rule=reply)

CORPUS = {"HNSW ef_search": "pgvector.adoc: SET hnsw.ef_search = 100 raises recall at the cost of latency."}


class State(TypedDict):
    question: str
    query: str
    context: str
    draft: str
    answer: str


def rewrite(state: State):
    return {"query": llm.invoke(f"Rewrite as a search query: {state['question']}").text}


def retrieve(state: State):
    hits = [text for key, text in CORPUS.items() if key.split()[0].lower() in state["query"].lower()]
    return {"context": "\n".join(hits)}


def gate(state: State) -> str:
    return "found" if state["context"] else "empty"


def draft(state: State):
    return {"draft": llm.invoke(f"Draft an answer from:\n{state['context']}\nQ: {state['question']}").text}


def polish(state: State):
    return {"answer": llm.invoke(f"Polish and cite: {state['draft']}").text}


def no_answer(state: State):
    return {"answer": "I could not find this in the documentation."}


g = StateGraph(State)
g.add_node("rewrite", rewrite)
g.add_node("retrieve", retrieve)
g.add_node("draft", draft)
g.add_node("polish", polish)
g.add_node("no_answer", no_answer)
g.add_edge(START, "rewrite")
g.add_edge("rewrite", "retrieve")
g.add_conditional_edges("retrieve", gate, {"found": "draft", "empty": "no_answer"})
g.add_edge("draft", "polish")
g.add_edge("polish", END)
g.add_edge("no_answer", END)
chain = g.compile()

out = chain.invoke({"question": "How do I raise recall for HNSW in pgvector?"})
print(out["query"]); print(out["answer"])
print(chain.invoke({"question": "What is the weather?"})["answer"])

Running it prints the rewritten query, the answer for an in-corpus question, and the gate’s fallback for one the corpus cannot answer:

Output (fake model)

HNSW ef_search recall latency pgvector
Raising hnsw.ef_search trades latency for recall (see pgvector page) [1].
I could not find this in the documentation.

The gate is a conditional edge whose function returns a key of the path map; the empty branch skips both LLM calls, which is the point of a gate — stopping a chain early is cheaper than letting a later step fail. Each node returns only the state keys it writes.

The Functional API expresses the same chain as ordinary Python control flow. @task marks a unit of work whose result is a future, and @entrypoint() marks the workflow function, so the gate becomes an if:

The same chain with the Functional API (continues the module above)

# Functional API
from langgraph.func import entrypoint, task

@task
def rewrite_t(q: str) -> str:
    return llm.invoke(f"Rewrite as a search query: {q}").text

@task
def retrieve_t(query: str) -> str:
    return retrieve({"query": query})["context"]

@entrypoint()
def workflow(q: str) -> str:
    query = rewrite_t(q).result()
    context = retrieve_t(query).result()
    if not context:
        return "I could not find this in the documentation."
    d = llm.invoke(f"Draft an answer from:\n{context}\nQ: {q}").text
    return llm.invoke(f"Polish and cite: {d}").text

print(workflow.invoke("How do I raise recall for HNSW in pgvector?"))

Running the whole module prints the polished answer once more as the final line:

Output of the Functional API run (fake model)

Raising hnsw.ef_search trades latency for recall (see pgvector page) [1].

The Graph API gives you an explicit, drawable structure; the Functional API gives you less ceremony when the flow is mostly straight-line code. The trade-off is covered in LangGraph Fundamentals. See LangGraph — Workflows and agents.

Routing

Routing classifies an input and sends it to a specialised follow-up, so each route can have its own prompt, tools or model. A router must return one of a closed set of labels; that is exactly what structured output (or an enum type) is for. DocsAssistant routes a question to the documentation search, to a bug-report handler or to small talk:

flowchart LR S([START]) --> C{classify} C -->|search_docs| A[search_docs] C -->|report_bug| B[report_bug] C -->|smalltalk| T[smalltalk] A --> E([END]) B --> E T --> E

Figure 2. Routing from START to one of three nodes

Routing with a conditional edge from START

from typing import Literal

from pydantic import BaseModel
from typing_extensions import TypedDict

from langchain_core.output_parsers import PydanticOutputParser
from langgraph.graph import END, START, StateGraph

from fake import RuleChat


class Route(BaseModel):
    step: Literal["search_docs", "report_bug", "smalltalk"]


def reply(prompt: str) -> str:
    if prompt.startswith("Classify"):
        q = prompt.lower()
        step = "report_bug" if "broken" in q else "smalltalk" if "hello" in q else "search_docs"
        return f'{{"step": "{step}"}}'
    return f"[{prompt.split(':')[0]}] handled"


llm = RuleChat(rule=reply)
router = llm | PydanticOutputParser(pydantic_object=Route)


class State(TypedDict):
    question: str
    answer: str


def classify(state: State) -> Literal["search_docs", "report_bug", "smalltalk"]:
    return router.invoke(f"Classify the question: {state['question']}").step


def node(name: str):
    def run(state: State):
        return {"answer": llm.invoke(f"{name}: {state['question']}").text}
    return run


g = StateGraph(State)
for name in ("search_docs", "report_bug", "smalltalk"):
    g.add_node(name, node(name))
    g.add_edge(name, END)
g.add_conditional_edges(START, classify)
app = g.compile()

for q in ("How do I tune HNSW?", "The docs page is broken", "hello there"):
    print(q, "->", app.invoke({"question": q})["answer"])

Output (fake model)

How do I tune HNSW? -> [search_docs] handled
The docs page is broken -> [report_bug] handled
hello there -> [smalltalk] handled

The routing function’s return annotation, Literal[…], is what lets LangGraph draw the possible destinations without a path map; return END from it to stop. Routes can be cheaper models than the one that answers — a llama3.1:8b classifier in front of a hosted model such as claude-sonnet-5 is a common split — and a route can itself be a compiled subgraph or an agent built with create_agent. See LangGraph — Workflows and agents.

Parallelisation

Parallelisation runs independent LLM calls (or retrievals) at the same time and aggregates the results, either to save wall-clock time (sectioning) or to gain confidence by running the same task several times (voting). In LangGraph, a node with several outgoing edges fans out: every target runs in the same superstep. Because several nodes then write to the same state key, that key needs a reducer, here operator.add, which concatenates the lists instead of raising a conflict. DocsAssistant queries a vector, a keyword and a glossary index in parallel and merges their hits:

flowchart LR S([START]) --> V[vector] S --> K[keyword] S --> G[glossary] V --> M[merge] K --> M G --> M M --> E([END])

Figure 3. Fan-out from START and merge

Fan-out from START with a reducer, then merge

import operator
import time
from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    query: str
    hits: Annotated[list[str], operator.add]
    merged: str


def branch(name: str, delay: float):
    def run(state: State):
        time.sleep(delay)
        return {"hits": [f"{name}:{state['query']}"]}
    return run


def merge(state: State):
    return {"merged": " | ".join(sorted(state["hits"]))}


g = StateGraph(State)
g.add_node("vector", branch("vector", 0.3))
g.add_node("keyword", branch("keyword", 0.3))
g.add_node("glossary", branch("glossary", 0.3))
g.add_node("merge", merge)
for n in ("vector", "keyword", "glossary"):
    g.add_edge(START, n)
    g.add_edge(n, "merge")
g.add_edge("merge", END)
app = g.compile()

t = time.perf_counter()
out = app.invoke({"query": "hnsw", "hits": []})
print(out["merged"])
print(f"elapsed {time.perf_counter() - t:.2f}s for three 0.3s branches")

Output

glossary:hnsw | keyword:hnsw | vector:hnsw
elapsed 0.31s for three 0.3s branches

Three branches that each sleep 0.3 seconds finished in about 0.3 seconds, not 0.9: they really ran concurrently (the timing is the evidence). The merge node has one incoming edge from each branch, so it waits until all three have finished. Sort or otherwise canonicalise the merged list, because the order in which branches append is not guaranteed. Voting is the same graph with the same node registered under different names and a merge that counts agreement. When branches differ in length, or the number of branches is not known up front, use Send (next section) rather than static edges. See LangGraph — Workflows and agents.

Orchestrator-worker

An orchestrator-worker workflow differs from plain parallelisation in one respect: the subtasks are not known in advance. An orchestrator LLM plans them from the input, workers handle one each, and a synthesiser combines the results. In LangGraph the dynamic fan-out is the Send API: a conditional edge returns a list of Send(node, state) objects, one per work item, and the graph runs that node once for each, each with its own private input. Worker outputs land in a shared key that has a reducer. DocsAssistant plans sub-questions for a broad question, answers each in a worker and synthesises:

flowchart LR S([START]) --> O[orchestrate] O -. "Send x N" .-> W["worker (one per sub-question)"] W --> Y[synthesize] Y --> E([END])

Figure 4. Orchestrator with Send to workers

Dynamic fan-out with Send

import operator
from typing import Annotated

from pydantic import BaseModel
from typing_extensions import TypedDict

from langchain_core.output_parsers import PydanticOutputParser
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

from fake import RuleChat


class Plan(BaseModel):
    subquestions: list[str]


def reply(prompt: str) -> str:
    if prompt.startswith("Plan"):
        return '{"subquestions": ["What is HNSW?", "How is pgvector configured?", "How is recall measured?"]}'
    if prompt.startswith("Answer"):
        return "answer to " + prompt.split(": ", 1)[1]
    return "SYNTHESIS"


llm = RuleChat(rule=reply)
planner = llm | PydanticOutputParser(pydantic_object=Plan)


class State(TypedDict):
    question: str
    subquestions: list[str]
    notes: Annotated[list[str], operator.add]
    answer: str


class WorkerState(TypedDict):
    subquestion: str


def orchestrate(state: State):
    return {"subquestions": planner.invoke(f"Plan sub-questions for: {state['question']}").subquestions}


def assign(state: State):
    return [Send("worker", {"subquestion": s}) for s in state["subquestions"]]


def worker(state: WorkerState):
    return {"notes": [llm.invoke(f"Answer: {state['subquestion']}").text]}


def synthesize(state: State):
    return {"answer": f"{llm.invoke('Synthesize: ' + ' / '.join(state['notes'])).text} ({len(state['notes'])} notes)"}


g = StateGraph(State)
g.add_node("orchestrate", orchestrate)
g.add_node("worker", worker)
g.add_node("synthesize", synthesize)
g.add_edge(START, "orchestrate")
g.add_conditional_edges("orchestrate", assign, ["worker"])
g.add_edge("worker", "synthesize")
g.add_edge("synthesize", END)
app = g.compile()

out = app.invoke({"question": "How do I set up HNSW search with pgvector?", "notes": []})
print(out["subquestions"])
print(sorted(out["notes"]))
print(out["answer"])

Output (fake model)

['What is HNSW?', 'How is pgvector configured?', 'How is recall measured?']
['answer to How is pgvector configured?', 'answer to How is recall measured?', 'answer to What is HNSW?']
SYNTHESIS (3 notes)

The workers' answers are sorted in the output above for the same reason as before: they finish in no fixed order. The worker takes a different state type (WorkerState) from the graph, which is what Send is for — the payload need not match the parent schema. If a worker must report failure or partial results, put those in the reducer key too, so the synthesiser can see which subtask failed. Each worker can be a small agent or a compiled subgraph; for worker agents that talk to each other or to a supervisor, see Multi-Agent & Deep Agents. See LangGraph — Workflows and agents and LangGraph — Use the Graph API for the Send API.

Evaluator-optimizer and reflection

In an evaluator-optimizer workflow one call generates a response and another evaluates it against explicit criteria; the evaluator’s feedback is fed back into the generator until the response passes or a budget runs out. Reflection is the same loop when the critic is the same model (or the same prompt family) reviewing its own draft; using a separate, stronger or differently prompted judge makes the evaluation less biased toward the draft. DocsAssistant’s criterion is whether the answer names the setting and cites its page:

flowchart LR S([START]) --> G[generate] G --> V[evaluate] V --> D{grounded, or 3 attempts used?} D -->|yes| E([END]) D -->|no: feedback| G

Figure 5. Generate-evaluate loop with an attempt budget

A generate-evaluate loop with an attempt budget

from typing import Literal

from pydantic import BaseModel
from typing_extensions import TypedDict

from langchain_core.output_parsers import PydanticOutputParser
from langgraph.graph import END, START, StateGraph

from fake import RuleChat

MAX_ATTEMPTS = 3
calls = {"draft": 0}


class Verdict(BaseModel):
    grounded: bool
    feedback: str


def reply(prompt: str) -> str:
    if prompt.startswith("Write"):
        calls["draft"] += 1
        if "Feedback:" in prompt:
            return "Set hnsw.ef_search = 100 to raise recall [pgvector.adoc]."
        return "Just make the index bigger."
    if "[pgvector.adoc]" in prompt:
        return '{"grounded": true, "feedback": ""}'
    return '{"grounded": false, "feedback": "Cite the page and name the setting."}'


llm = RuleChat(rule=reply)
critic = llm | PydanticOutputParser(pydantic_object=Verdict)


class State(TypedDict):
    question: str
    draft: str
    feedback: str
    grounded: bool
    attempts: int


def generate(state: State):
    prompt = f"Write an answer to: {state['question']}"
    if state.get("feedback"):
        prompt += f"\nFeedback: {state['feedback']}"
    return {"draft": llm.invoke(prompt).text, "attempts": state.get("attempts", 0) + 1}


def evaluate(state: State):
    v = critic.invoke(f"Grade this answer: {state['draft']}")
    return {"grounded": v.grounded, "feedback": v.feedback}


def decide(state: State) -> Literal["generate", "__end__"]:
    if state["grounded"] or state["attempts"] >= MAX_ATTEMPTS:
        return END
    return "generate"


g = StateGraph(State)
g.add_node("generate", generate)
g.add_node("evaluate", evaluate)
g.add_edge(START, "generate")
g.add_edge("generate", "evaluate")
g.add_conditional_edges("evaluate", decide)
app = g.compile()

out = app.invoke({"question": "How do I raise HNSW recall in pgvector?"})
print(out["attempts"], out["grounded"], out["draft"])

Output (fake model)

2 True Set hnsw.ef_search = 100 to raise recall [pgvector.adoc].

The first draft is rejected with feedback, the second cites its page and passes, so the loop ends after two attempts. Two details keep the loop safe. The attempt counter lives in state and is part of the exit condition, so a critic that never approves cannot spin forever. And the evaluator returns a structured verdict (a boolean plus feedback), so the routing function reads a field instead of parsing prose. LangGraph also enforces a global ceiling of its own, covered under Guarding loops. To let a person be the evaluator, replace the evaluate node with an interrupt (see Human-in-the-Loop & Time Travel). See LangGraph — Workflows and agents.

Self-corrective RAG

Self-corrective RAG puts a grading step between retrieval and generation: if the retrieved chunks are not relevant to the question, the workflow rewrites the query and retrieves again instead of generating from bad context, and it gives up honestly after a bounded number of rewrites. It is a routing decision (generate, rewrite or give up) inside a loop (rewrite back to retrieve). The vector-store side of the pattern — retrievers, relevance scores, the agent-style variant — is in Agentic RAG: grade, rewrite, retry and is not repeated here; this section shows the workflow shape and its termination rules with a stub index.

flowchart LR S([START]) --> R[retrieve] R --> G{grade documents} G -->|relevant| A[generate] G -->|not relevant, rewrites left| W[rewrite] W --> R G -->|not relevant, budget spent| X[give_up] A --> E([END]) X --> E

Figure 6. Self-corrective RAG with a rewrite budget

Grade, rewrite and retry with a rewrite budget

from typing import Literal

from pydantic import BaseModel
from typing_extensions import TypedDict

from langchain_core.output_parsers import PydanticOutputParser
from langgraph.graph import END, START, StateGraph

from fake import RuleChat

MAX_REWRITES = 2

INDEX = {  # stand-in for the pgvector retriever: query keyword -> chunk
    "hnsw": "pgvector.adoc: SET hnsw.ef_search = 100 raises recall at the cost of latency.",
}


class Grade(BaseModel):
    relevant: bool


def reply(prompt: str) -> str:
    if prompt.startswith("Grade"):
        return '{"relevant": %s}' % ("true" if "ef_search" in prompt else "false")
    if prompt.startswith("Rewrite"):
        return "weather forecast" if "weather" in prompt else "hnsw ef_search"
    return "Set hnsw.ef_search = 100 [pgvector.adoc]."


llm = RuleChat(rule=reply)
grader = llm | PydanticOutputParser(pydantic_object=Grade)


class State(TypedDict):
    question: str
    query: str
    documents: list[str]
    rewrites: int
    answer: str


def retrieve(state: State):
    q = state.get("query") or state["question"]
    return {"query": q, "documents": [d for k, d in INDEX.items() if k in q.lower()]}


def grade(state: State) -> Literal["generate", "rewrite", "give_up"]:
    if any(grader.invoke(f"Grade relevance to {state['question']}: {d}").relevant for d in state["documents"]):
        return "generate"
    return "rewrite" if state.get("rewrites", 0) < MAX_REWRITES else "give_up"


def rewrite(state: State):
    return {"query": llm.invoke(f"Rewrite for retrieval: {state['question']}").text,
            "rewrites": state.get("rewrites", 0) + 1}


def generate(state: State):
    return {"answer": llm.invoke(f"Answer from {state['documents']}: {state['question']}").text}


def give_up(state: State):
    return {"answer": "I could not find this in the documentation."}


g = StateGraph(State)
for name, fn in (("retrieve", retrieve), ("rewrite", rewrite), ("generate", generate), ("give_up", give_up)):
    g.add_node(name, fn)
g.add_edge(START, "retrieve")
g.add_conditional_edges("retrieve", grade)
g.add_edge("rewrite", "retrieve")
g.add_edge("generate", END)
g.add_edge("give_up", END)
app = g.compile()

for step in app.stream({"question": "How do I get better recall from vector search?"}, stream_mode="updates"):
    print({k: {kk: vv for kk, vv in v.items() if kk in ("query", "rewrites", "answer")} for k, v in step.items()})
print(app.invoke({"question": "What is the weather?"})["answer"])

Output (fake model)

{'retrieve': {'query': 'How do I get better recall from vector search?'}}
{'rewrite': {'query': 'hnsw ef_search', 'rewrites': 1}}
{'retrieve': {'query': 'hnsw ef_search'}}
{'generate': {'answer': 'Set hnsw.ef_search = 100 [pgvector.adoc].'}}
I could not find this in the documentation.

The first pass retrieves nothing relevant for the vague question, the rewrite node produces a query that contains the setting name, and the second retrieval succeeds. The last line is the unanswerable question: the rewrite does not help, the budget of two rewrites is spent and the workflow gives up with a fixed message instead of hallucinating. Grading with a small local model (llama3.1:8b) and generating with a larger one is a typical cost split. See LangGraph — Build a custom RAG agent, which builds the same grade-and-rewrite loop over a real vector store.

Guarding loops

Two of the patterns above loop, so both need a stop condition that does not depend on the model behaving. Put a counter in state and test it in the routing function, as the two examples do. As a second line of defence, LangGraph limits the number of supersteps a run may take and raises GraphRecursionError when it is exceeded; the limit is the recursion_limit key of the run config (a top-level key, not one under configurable). The default differs between versions (10007 in the installed langgraph 1.2.12), so set it explicitly for any graph that contains a loop:

Catching the recursion limit on a graph that never stops

from typing_extensions import TypedDict
from langgraph.errors import GraphRecursionError
from langgraph.graph import START, StateGraph

class S(TypedDict):
    n: int

g = StateGraph(S)
g.add_node("loop", lambda s: {"n": s.get("n", 0) + 1})
g.add_edge(START, "loop")
g.add_edge("loop", "loop")
app = g.compile()
try:
    app.invoke({"n": 0}, config={"recursion_limit": 5})
except GraphRecursionError as e:
    print(type(e).__name__, str(e)[:90])

Output

GraphRecursionError Recursion limit of 5 reached without hitting a stop condition. You can increase the limit

The message is truncated here. See LangGraph — Graph API overview for recursion_limit.

Choosing and combining patterns

Patterns nest: an orchestrator’s worker can be a self-corrective RAG subgraph, and a routed branch can be an agent. Start with the simplest one that fits and add structure only when a measurement shows it helps.

Table 1. Which pattern for which DocsAssistant problem
Pattern	Reach for it when	LangGraph feature
Prompt chaining	The task splits into fixed, verifiable steps	Edges, a conditional edge as a gate
Routing	Inputs fall into distinct categories with different handling	`add_conditional_edges`, a `Literal` return type
Parallelisation	Independent subtasks, or repeated runs for confidence	Multiple edges from one node, a reducer such as `operator.add`
Orchestrator-worker	The subtasks depend on the input	`Send`, a reducer on the shared key
Evaluator-optimizer / reflection	A clear criterion exists and iteration measurably helps	A conditional edge back to the generator, an attempt counter
Self-corrective RAG	Retrieval quality varies and bad context produces bad answers	A grading route with rewrite and give-up branches

Every graph above compiles without a checkpointer. To keep state across turns or resume a run, compile with one (see Short-Term Memory & Checkpointers). When the model, not the code, should choose the path, use Agents with create_agent.

LangGraph Workflows

Setup: a fake model for every example

Prompt chaining

Routing

Parallelisation

Orchestrator-worker

Evaluator-optimizer and reflection

Self-corrective RAG

Guarding loops

Choosing and combining patterns

References