LangGraph Patterns

Three of the five coordination patterns from Agent Orchestration Patterns — implemented as real, compiled, tested LangGraph graphs, not diagrams.

Free for evaluation, personal, and internal use. Production use requires a commercial license — see LICENSE.md.

Which three, and why these three

• Pattern 1 — Supervisor / Worker (app/supervisor_worker.py)
• Pattern 4 — Human-in-the-Loop Approval Gate (app/human_in_the_loop.py)
• Pattern 5 — Deterministic-First Escalation (app/deterministic_escalation.py)

Patterns 2 (Sequential Handoff) and 3 (Parallel Fan-Out/Fan-In) aren't included here — Sequential Handoff is close to trivial in LangGraph (a straight chain of add_edge calls with little to demonstrate beyond what the diagram already shows), and RAG Reference Pipeline is already a real, working sequential handoff if you want that pattern as code. Fan-Out/Fan-In is a reasonable candidate for a future addition, not omitted for a principled reason — just not built yet.

The three included here were picked because each demonstrates something a diagram can't: real conditional routing (pattern 1), LangGraph's actual pause/resume mechanism, not a polling convention pretending to be one (pattern 4), and a genuine escalation ladder as a graph with a real conditional edge (pattern 5).

Pattern 5 specifically

deterministic_escalation.py is the third implementation of the same negation-aware escalation ladder that powers Sentinel and RAG Reference Pipeline's groundedness.py. Same algorithm, same regression test (test_negated_citation_blocks_not_allows), this time expressed as an actual LangGraph with a real conditional edge between the deterministic check and the judge node — not a Python if/else chain that happens to produce the same output.

The shared AgentBrain abstraction

All three patterns need a "decide between options given context" step somewhere — routing (pattern 1), risk classification (pattern 4), the judge call (pattern 5). Rather than write three separate stub decision-makers, app/brain.py defines one interface (AgentBrain.decide), an offline deterministic stub (StubBrain), and a shell for a real model call (RealBrain) — same pattern as Sentinel's judges and RAG Reference Pipeline's Embedder.

StubBrain is not a production decision-maker. It's a crude token-overlap matcher with a 5-character-prefix stem (enough to catch "crash"/"crashes"/"crashing" without real NLP). Two real bugs surfaced while building this pack, both now covered by regression tests:

• Exact-token matching alone missed "crashing"/"error" against "crashes"/"errors" in the option text — every technical-sounding query silently misrouted to whichever worker was listed first. Fixed with the prefix-stem approach; see tests/test_brain.py::test_crude_stemming_matches_plural_and_ing_forms.
• A query sharing zero vocabulary with any option also fell back to the first-listed option by default — arbitrary, not a real decision. Fixed by adding an explicit default parameter to the interface; see tests/test_brain.py::test_no_signal_returns_default_not_first_option.

Both bugs are the same class of thing as RAG Reference Pipeline's BM25 edge case: found by actually running the code against realistic input, not by code that merely looked correct.

Running it

pip install -r requirements.txt
pytest tests/ -v   # 19 tests, no API key needed

# Pattern 1
from app.supervisor_worker import build_graph
graph = build_graph()
result = graph.invoke({"task": "I need a refund", "routed_to": None, "result": None})

# Pattern 4 — requires a thread_id for the checkpointer
from app.human_in_the_loop import build_graph
from langgraph.types import Command

graph = build_graph()
config = {"configurable": {"thread_id": "example-1"}}
result = graph.invoke(
    {"proposed_action": "issue a refund of $5000", "risk_level": None, "approved": None, "result": None},
    config=config,
)
# result["__interrupt__"] is present — the graph is paused here.
resumed = graph.invoke(Command(resume={"approved": True}), config=config)

# Pattern 5
from app.deterministic_escalation import build_graph
graph = build_graph()
result = graph.invoke({
    "response": "Per Section 9.2, employees get unlimited PTO.",
    "context": "The handbook confirms no Section 9.2 exists.",
    "citation_found": None, "negated": None, "decided_by": None, "verdict": None,
})
# result["verdict"] == "block", result["decided_by"] == "heuristic"

What's a placeholder vs. what's real

• All three graphs, StubBrain, and pattern 4's interrupt/resume mechanism are real and tested — 19 tests, all exercising actual LangGraph execution, not mocked assertions.
• StubBrain is a deterministic token-overlap matcher, not a real decision-maker. Good enough to exercise routing/classification logic in tests; should never make a decision that matters in production.
• RealBrain is an unimplemented shell showing the AgentBrain interface a real model call satisfies — a reference point, not a stub pretending to work.
• Pattern 4's MemorySaver checkpointer is in-memory and does not survive a process restart. Production use needs a persistent checkpointer (Postgres-backed, for instance) — same "don't ship the demo's infrastructure choice to production" caveat as RAG Reference Pipeline's InMemoryVectorStore.

Using this in your own project

1. Clone and run pytest tests/ -v — confirm everything passes.
2. Swap StubBrain for a real implementation of AgentBrain — fill in RealBrain.decide with an actual chat model call, or write your own class satisfying the same interface.
3. For pattern 4 in production: replace MemorySaver() with a persistent checkpointer. LangGraph ships Postgres and SQLite checkpointer implementations — build_graph(checkpointer=...) already accepts any of them.
4. If you're combining patterns (per Agent Orchestration Patterns' closing note — most real systems nest two or three together), the AgentBrain instance can be shared across them: one real model client, injected into whichever pattern's build_graph() needs a decision made.

Project layout

app/
  brain.py                     shared AgentBrain protocol + StubBrain + RealBrain shell
  supervisor_worker.py         pattern 1
  human_in_the_loop.py         pattern 4
  deterministic_escalation.py  pattern 5
tests/                         19 tests, one file per module

License

Free to use for evaluation, personal projects, and internal experimentation. A commercial license is required to run this in production. Full terms in LICENSE.md; pricing at mleg.tech.