Orchestrate end-to-end delivery with LangGraph state

Introduction

When you reach the end of an AI engagement, the gap between a working prototype and a production handoff can feel like the hardest mile — discovery artifacts, scoping decisions, deployment configs, and client-ready documentation all need to land in the right order. This lesson teaches you how to tie every delivery phase together using a LangGraph-based orchestrator that routes work through conditional quality gates from discovery to handoff. By the end, you will be able to build and run a stateful delivery workflow that sequences each phase automatically and resumes cleanly after any interruption.

Key Terminology

• StateGraph — A LangGraph class that models the delivery lifecycle as a directed graph of named nodes and edges; instantiated with a typed state schema (e.g., DeliveryState) so every node reads and writes the same shared dictionary.
• DeliveryState — A TypedDict that defines the schema of the shared delivery context, carrying fields like discovery_artifacts, current_phase, and phase_completed across every node in the graph.
• Conditional quality gate — A routing function wired via add_conditional_edges that inspects the current state and returns a string label ("pass" or "fail") to determine which node the workflow transitions to next, enabling automatic rework loops without hard-coded retry logic.
• SqliteSaver — A LangGraph checkpointer that persists the full graph state to a SQLite file after every node completes, allowing a multi-day engagement workflow to survive process restarts and resume from the last successful phase.
• Phase node — An async function (e.g., run_discovery) that follows the contract of reading inputs from state, executing its phase logic, and returning a partial update dictionary that LangGraph merges back into the shared DeliveryState.
• Discovery artifacts — The structured output dictionary produced by the discovery node — containing scored use cases, data readiness profiles, and interview insights — that downstream phases (scoping, prototyping) consume as their primary input.

Concepts

Mapping a Delivery Lifecycle to a Directed Graph

A delivery engagement is not a linear checklist. Phases loop back when quality isn't met, later phases depend on the structured outputs of earlier ones, and the whole workflow may pause for days between steps. A directed graph captures exactly this structure: each phase becomes a node, each transition becomes an edge, and the conditional logic at each gate determines whether work advances or re-enters the same node for rework.

LangGraph's StateGraph gives you this model directly. You declare nodes by name ("discovery", "scope", "prototype", "deploy", "handoff"), wire them with edges, and set an entry point. The graph then enforces the execution order — rather than you manually sequencing function calls that have no awareness of each other's outputs or failure modes. The set_entry_point("discovery") call tells the runtime where every new engagement begins (see Code Walkthrough).

Conditional Gates as Rework Loops

The key insight that separates a resilient orchestrator from a fragile script is how failures are handled. In a simple sequential pipeline, a phase that produces low-quality output either raises an exception or silently passes bad data forward. In a graph-based orchestrator, a quality gate function evaluates the state after each node and routes control flow explicitly — "pass" advances the engagement, "fail" returns to the same node for rework.

add_conditional_edges("discovery", self.check_discovery_gate, {"pass": "scope", "fail": "discovery"}) is not just a routing call — it encodes the business rule that discovery work must meet a quality bar before scoping begins. This decouples the gate logic (what "good enough" means) from the phase logic (what the discovery node actually does), making both independently testable (see Code Walkthrough).

Checkpointing for Durable, Multi-Day Workflows

A delivery engagement rarely completes in a single process run. Compiling the graph with SqliteSaver.from_conn_string(self.db_path) instructs LangGraph to persist the full DeliveryState to disk after every node execution. If the process restarts — whether from a crash, a scheduled pause, or a deliberate stop — invoking the graph with the same thread ID resumes from the last persisted checkpoint rather than from "discovery" again.

This is why the approach matters beyond elegance: without checkpointing, a five-phase workflow that fails midway through deployment requires re-running discovery, scoping, and prototyping from scratch. With it, the orchestrator picks up at the exact node boundary where execution last succeeded.

The Node Contract: Consume, Execute, Emit

Every phase node follows the same three-step contract. First, it reads its required inputs from the shared DeliveryState — run_discovery consumes use_case_descriptions, data_samples, and interview_transcripts. Second, it executes its phase logic — scoring feasibility, profiling data readiness, summarizing insights. Third, it returns a partial dictionary that LangGraph merges back into the live state — updating discovery_artifacts, current_phase, and phase_completed.

This partial-update pattern is what makes phases composable: each node only writes the keys it owns, never the full state object. The downstream node (run_scoping) then reads discovery_artifacts without knowing anything about how it was produced. Adding a new phase means adding a node that follows the same contract — no changes to existing nodes required (see Code Walkthrough).

Code Walkthrough

Now that you understand how the delivery lifecycle maps to a directed graph, let's walk through the two core pieces that make the orchestrator work: the graph construction and the discovery phase node.

LangGraph State Machine Design

The orchestrator defines the delivery lifecycle as a directed graph with conditional transitions:

Code snippetpython
1from langgraph.graph import StateGraph
2from langgraph.checkpoint.sqlite import SqliteSaver
3from typing import TypedDict, Literal
4
5class DeliveryState(TypedDict):
6    use_case_descriptions: list[str]
7    data_samples: list[dict]
8    interview_transcripts: list[str]
9    discovery_artifacts: dict
10    current_phase: str
11    phase_completed: bool
12
13class DeliveryOrchestrator:
14    """Orchestrates end-to-end AI delivery lifecycle."""
15
16    def __init__(self, llm_config: dict, db_path: str):
17        self.llm_config = llm_config
18        self.db_path = db_path
19
20    def build_workflow(self) -> StateGraph:
21        """Construct the delivery lifecycle graph."""
22        graph = StateGraph(DeliveryState)
23        graph.add_node("discovery", self.run_discovery)
24        graph.add_node("scope", self.run_scoping)
25        graph.add_node("prototype", self.run_prototype)
26        graph.add_node("deploy", self.run_deployment)
27        graph.add_node("handoff", self.run_handoff)
28
29        graph.add_conditional_edges(
30            "discovery",
31            self.check_discovery_gate,
32            {"pass": "scope", "fail": "discovery"},
33        )
34        graph.add_conditional_edges(
35            "scope",
36            self.check_scope_gate,
37            {"pass": "prototype", "fail": "scope"},
38        )
39        graph.set_entry_point("discovery")
40        return graph.compile(
41            checkpointer=SqliteSaver.from_conn_string(self.db_path)
42        )

The StateGraph is initialized with a DeliveryState schema so every node reads and writes the same typed dictionary. Each add_conditional_edges call wires a quality gate function: returning "pass" advances the workflow to the next phase, while "fail" loops back for rework. Compiling with SqliteSaver persists the graph's checkpoint after every node, so the workflow survives process restarts and resumes from the last completed phase — critical for multi-day engagements.

Phase Execution

Each node executes its phase logic and writes structured artifacts back into state:

Code snippetpython
1import asyncio
2
3async def run_discovery(self, state: DeliveryState) -> dict:
4    """Execute discovery phase activities."""
5    # Score use cases by feasibility and business value
6    scored_cases = []
7    for desc in state["use_case_descriptions"]:
8        score = {"description": desc, "feasibility": 0.85, "value": 0.90}
9        scored_cases.append(score)
10
11    # Profile data readiness from provided samples
12    readiness = {
13        "sample_count": len(state["data_samples"]),
14        "completeness": 0.92,
15        "schema_consistent": True,
16    }
17
18    # Summarize insights from interview transcripts
19    insights = [
20        f"Insight from transcript {i+1}"
21        for i in range(len(state["interview_transcripts"]))
22    ]
23
24    return {
25        "discovery_artifacts": {
26            "scored_use_cases": scored_cases,
27            "data_readiness": readiness,
28            "interview_insights": insights,
29        },
30        "current_phase": "discovery",
31        "phase_completed": True,
32    }

The node reads from state — the shared delivery context — runs its evaluations, and returns a partial update dictionary that LangGraph merges back into the graph's state. The discovery_artifacts key carries forward everything the scoping phase needs to set boundaries and priorities. This same pattern repeats for each subsequent phase: each node consumes the artifacts its predecessor produced and emits its own.

Confirm that after calling build_workflow() and invoking the compiled graph with an initial state, the current_phase field in the final state reads "discovery" and phase_completed is True before the conditional gate routes to the next node.

Do's and Don'ts

Having walked through the material above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do wire every phase transition through add_conditional_edges with explicit "pass" / "fail" routing — this lets the orchestrator loop a phase back for rework rather than blindly advancing, so a failed discovery gate re-runs discovery until check_discovery_gate returns "pass" instead of silently corrupting downstream scoping artifacts.
2. ✓Do compile the StateGraph with SqliteSaver attached as the checkpointer — without it, a process restart mid-engagement drops all intermediate DeliveryState fields and forces the entire graph to re-execute from the discovery entry point, which is unacceptable for multi-day client engagements.
3. ✓Do return a partial dictionary (only the keys your node writes) from each phase node like run_discovery — LangGraph merges partial updates into the shared DeliveryState; returning the full state object risks overwriting fields set by prior nodes, such as use_case_descriptions or data_samples that the discovery node read but did not produce.

Don'ts

1. ✗Don't bypass check_discovery_gate by hardcoding the edge directly from "discovery" to "scope" — removing the conditional gate means scoping proceeds even when discovery_artifacts is incomplete or its data_readiness signals low completeness, which propagates bad baselines through every subsequent phase with no recovery path.
2. ✗Don't initialize StateGraph without a DeliveryState schema — omitting the TypedDict removes type enforcement on keys like phase_completed and current_phase; nodes can silently write to mismatched keys and the conditional gate functions read None instead of a boolean or phase string, causing unpredictable routing.
3. ✗Don't invoke a later phase node (e.g., run_prototype) directly on raw input instead of through the compiled graph — skipping the compiled graph bypasses both the SqliteSaver checkpoint writes and the conditional gate logic, so the discovery_artifacts and current_phase fields that run_prototype depends on are never populated by the nodes that precede it.