Workflow graphs

A workflow graph describes a process as nodes and edges instead of nested code: each node does one unit of work, and each edge decides what runs next. LangGraph, a framework built around the idea, defines all three plainly. Of nodes: “Functions that encode the logic of your agents. They receive the current state as input, perform some computation or side-effect, and return an updated state.” Of edges: “Functions that determine which Node to execute next based on the current state.” Of state: “A shared data structure that represents the current snapshot of your application”[1]. State is passed from node to node rather than living in local variables.

The graph describes the available steps and transitions. A transition can use a deterministic condition or a model-produced classification; either way, the surrounding application defines how the result selects the next step. Persistence and checkpointing are additional capabilities, not requirements for something to be a workflow graph.

The distinction from agent graphs is the work inside the nodes and who chooses subsequent actions, not simply the presence of edges. A node can run a fixed operation or an agent loop, and a system can combine both. The site’s levels organize these patterns for explanation; they are not universal framework categories. It is also the middle page of the site’s graph engineering thread: knowledge graphs connect information, agent graphs connect work, and a workflow graph is what you draw when the work is settled in advance.

This page provides primary references and illustrative examples. The examples are scripted, not measured model runs; source references do not establish the correctness of every implementation or outcome.

"How often should the DW-300's filter be cleaned?"

def _node_retrieve(state: State, sections: dict[str, Section], model: Model) -> State:
    hits = [s for s, score in bm25_search(sections, state["question"], k=RETRIEVE_K) if score > 0]
    state["source_cites"] = [s.cite for s in hits]
    return state

NODES: dict[str, Callable[[State, dict, Model], State]] = {
    "retrieve": _node_retrieve,
    "no_match": _node_no_match,
    "draft": _node_draft,
    "check": _node_check,
    "revise": _node_revise,
}


def _edge_from_retrieve(state: State) -> str | None:
    return "draft" if state["source_cites"] else "no_match"


def _edge_from_check(state: State) -> str | None:
    if state["verdict"] == "ok" or state["revisions"] >= MAX_REVISIONS:
        return None
    return "revise"


EDGES: dict[str, Callable[[State], str | None]] = {
    "retrieve": _edge_from_retrieve,
    "no_match": lambda state: None,
    "draft": lambda state: "check",
    "check": _edge_from_check,
    "revise": lambda state: "check",
}


def run(
    question: str,
    model: Model,
    embedder: Embedder | None,
    tracer: Tracer,
    *,
    corpus_dir: Path = DEFAULT_CORPUS_DIR,
) -> Answer:
    del embedder  # retrieval here is keyword search, not a vector index
    sections = load_sections(corpus_dir)
    state: State = {"question": question, "source_cites": [], "draft_text": "", "verdict": None, "revisions": 0}

    node_id: str | None = "retrieve"
    while node_id is not None:
        state = NODES[node_id](state, sections, model)
        completion = state.pop("_completion", None)
        if completion is not None:
            tracer.record(
                kind="model", decided_by="code", title=f"Node: {node_id}", detail=completion.text[:200],
                tokens_in=completion.tokens_in, tokens_out=completion.tokens_out, ms=completion.ms,
            )
        else:
            tracer.record(kind="code", decided_by="code", title=f"Node: {node_id}", detail=", ".join(state["source_cites"]) or "none")
        next_id = EDGES[node_id](state)
        tracer.record(
            kind="code", decided_by="code", title=f"Checkpoint after '{node_id}'",
            detail=f"revisions={state['revisions']} verdict={state.get('verdict')!r} -> next: {next_id or 'stop'}",
        )
        node_id = next_id

    citations = cited_sources(state["draft_text"])
    return Answer(text=state["draft_text"], citations=citations, retrieved_sources=state["source_cites"])

python -m examples.workflow_graphs --model stub:scripted

Failure modes

Cost and latency

Illustrative, not measured: no result file exists for this technique yet, so every number below is a worked example. See docs/EVALS.md for how a real run is recorded and scored.

How to Evaluate It

The same 60 questions as every other technique, with a narrow interest in the result. Because the graph runs the same logic as write and check, its score on citation hit rate and revision count should match that page’s result file once one exists: a divergence between the two would point at a bug in the graph wiring, not a difference in what the technique can do. The no_match branch is what unanswerable questions specifically exercise: whether retrieval finding nothing correctly ends the run at a dead end rather than drafting from an empty source list.

No result file exists yet (see docs/EVALS.md). Run python scripts/eval_run.py --example workflow_graphs --model <spec> --dry to project the cost of a real run first.

Run it

Try it

1. Use it

   Find an automation tool with a visual if/else node. Read one branch condition: a fixed rule, or a model deciding?

2. Build it

   Run python -m examples.workflow_graphs --model stub:scripted from the repo root: retrieve, draft, a check that fails on an unretrieved citation, revise, check, stop, checkpointing after every node. Now ask a question of gibberish words: retrieval returns nothing, the graph goes straight to no_match, and two checkpoints stand in for five.

3. Either lane

   Draw a process you run as nodes and edges, then count the edges whose condition you could not write as a line of code. That is where it stops being this level.