TLDR
- A graph is a control contract, not a picture of cleverness. It defines which state exists, which routes may run and where authority can interrupt action.
- State should behave like a typed case file: versioned, minimal, provenance-aware and explicit about what merges when concurrent work returns.
- Model proposals belong inside deterministic boundaries. Conditional edges, tool contracts, policy checks and human authorisation decide what may happen next.
- Checkpoints preserve workflow continuity; user memory, enterprise knowledge, world state and effect evidence are different stores with different retention rules.
- A controlled graph proves its route. Release evidence joins state transitions, tool receipts, policy decisions, readback and recovery rather than celebrating a plausible final answer.
Reader and route
This book is for engineers, architects, model-risk practitioners and technical leaders designing long-running AI workflows. Read Parts I and II for the mechanism, Part III for consequential authority, and Parts IV and V for retrieval, coordination and operation. The Merehaven lab and glossary can be used as a field manual.
Evidence boundary
LangGraph APIs, provider interfaces, cloud features and code fragments are version-pinned learning specimens. Revalidate them against the selected package versions and runtime before use. Merehaven Bank is wholly fictional; every customer, case, metric, incident and route is synthetic.
The paused payment
A payment-support assistant gathers a customer’s request, checks account context and proposes a route. Halfway through, the customer changes the destination account. A linear script has two bad choices: restart and lose the approval trail, or continue with stale context. A graph can resume, but only if its state distinguishes the old proposal from the new world state.
Now add a timeout. The transfer tool returns no receipt. Retrying may duplicate an effect; assuming success may mislead the customer. The difficult part is not model reasoning. It is preserving identity, state, authority and evidence while the route changes underneath the conversation.
The graph contract
Every transition should answer five questions: what state was read, what new facts were proposed, which policy authorised the transition, which effect receipt returned and how an unknown outcome will be reconciled. A graph that cannot answer them is an orchestration sketch, not an operating control.
Part I: State before action
The first design decision is not the model. It is the state boundary: what the workflow knows, what it may infer and which facts remain independently authoritative.
Why a graph changes the problem
Why the chain paradigm finally broke
Then something peculiar started happening. Product owners at banks began asking harder questions. “Can the assistant more than answer the customer’s question about their mortgage, but also check whether they are in financial difficulty, and if so, route them to the Consumer Duty workflow?” “Can it more than summarise a loan contract, but compare it against ten peer contracts and flag non-standard clauses?” “Can it call the transaction system, check the actual overdraft balance, and then explain it in plain English?” Every one of these requests had the same shape. They all required the system to decide, at runtime, what to do next, based on what it had just discovered.
That is the chain. Beautiful, predictable, reproducible, and profoundly limited. Now look at what an agent does when the same question arrives, but the customer turns out to be in arrears:
That is an agent. The difference is more than the diamond shapes. The difference is that the diamond shapes are real decisions, made by a language model at runtime, based on the actual content of the state. The chain’s author has to predict every path. The agent discovers its own path.
Worked example: the state as the shared case file
Start with a concrete case with the smallest possible banking example. Suppose the Merehaven worked scenario has a prototype relationship manager copilot. A question arrives: “What is Meridian Logistics’s current exposure across all products?” Trace what happens with a chain and with an agent.
In the chain version, the system has four steps. Step one parses the question. Step two calls the core banking API for exposures. Step three formats the numbers. Step four returns. Total tokens used: about 800. Total time: 2.3 seconds. Perfect, until the RM asks a follow-up: “And show me the covenant status on each facility.” Now the chain is stuck. It has no step for covenants. A human would know to call a different API. The chain does not.
In the agent version, the same question arrives and flows into a state object. The state is, roughly, a Python dictionary that travels through the graph and collects evidence as it goes. It looks something like this, expressed as a small Python type:
from typing import TypedDict, List, Optional
from typing_extensions import Annotated
class RMCopilotState(TypedDict):
# What the RM actually asked
query: str
# What the agent has decided to do next
next_action: Optional[str]
# Structured results that have been gathered so far
exposures: Optional[List[dict]]
covenants: Optional[List[dict]]
peer_comparison: Optional[dict]
# The final answer that will be returned
draft_answer: Optional[str]
# Any flags that need human attention
escalations: List[str]The state is the shared case file. Every node in the graph reads from it and writes to it. A router node at the top inspects the query and decides which tools to invoke. If the query mentions “exposure”, it routes to the exposure tool. If it mentions “covenant” or “breach”, it routes to the covenant tool. If it mentions both, it runs both in parallel. If the RM’s follow-up arrives with the same state already populated, the agent does not re-fetch the exposures. It sees them already in the case file and moves on to covenants.
This is the quiet revolution. The state lets the agent accumulate evidence. The graph lets the agent choose which evidence to gather next. The combination is what produces behaviour that feels like reasoning.
Look carefully at the loop from H back to C. A chain cannot do that. A chain is acyclic by definition. The moment your workflow needs to revisit an earlier step, the chain breaks down. LangGraph, as its name suggests, embraces cycles. Cycles are where agency lives.
The failure mode: when “agentic” goes wrong
The lesson, which LangGraph does not eliminate but rather makes more tractable, is that agent capability is exactly as valuable as the guardrails that constrain it. A chain fails by being unhelpful. An agent can fail by being too helpful in the wrong direction. Three patterns in particular recur in post-mortems.
The first is unbounded action space. If an agent can call any tool, it will sometimes call the wrong tool for the wrong reason. The discipline is to scope the tool set tightly at each node, not globally.
The second is state contamination. If the same graph instance serves many users and state leaks between them, the failure mode is data breach. Session isolation, enforced at the state object level, is required by the design in a banking context. In the Merehaven worked scenario, this means every graph run has a uniquely namespaced state bucket in Firestore, tied to the authenticated user’s subject identifier.
The third is loop divergence. If an agent can loop back to an earlier node, it can, in principle, loop forever. LangGraph has a native recursion limit (default 25 steps) to prevent this, but the deeper discipline is to design your graphs so that each node either makes measurable progress on the state or terminates.
Each of these failure modes has a fix. Each fix is easier to implement in a graph than in a chain. That is the quiet virtue of the graph abstraction: it makes the failure modes explicit and therefore addressable.
A thought experiment
Imagine you are architecting a new mortgage pre-application agent for Merehaven Bank. The customer lands on the mortgage pages, starts chatting with the copilot, and within five minutes the copilot needs to have: collected enough information to produce an Agreement in Principle, verified that the customer is not in financial difficulty (Consumer Duty), flagged any vulnerability indicators, checked sanctions and PEP status, and either approved provisionally or politely deferred.
Now ask yourself: what is the smallest possible state object that could carry all of this? A minimal answer looks like this.
from typing import TypedDict, List, Optional, Literal
from datetime import datetime
class MortgageIntakeState(TypedDict):
# Conversation tracking
session_id: str
subject_identifier: str # authenticated customer ID
conversation: List[dict] # history of turns
current_turn: int
# Structured data captured so far
property_value: Optional[float]
requested_loan: Optional[float]
household_income: Optional[float]
household_dependants: Optional[int]
existing_commitments: Optional[List[dict]]
source_of_deposit: Optional[str]
# Checks and assessments
affordability_assessment: Optional[dict]
consumer_duty_flags: List[str]
vulnerability_indicators: List[str]
sanctions_check_status: Literal["pending", "clear", "hit"]
pep_check_status: Literal["pending", "clear", "hit"]
# Decision
decision: Optional[Literal["provisional_approve", "defer", "decline"]]
decision_reasons: List[str]
# Audit
model_versions: dict
timestamps: List[datetime]
next_best_action: Optional[str]The design shows how the state carries more than the customer’s answers but also the agent’s in-progress assessments, the audit trail (model versions, timestamps), and the regulatory flags. This is what a state object looks like when it has to survive scrutiny. The graph itself is simpler than you might expect: an intake node, a verification node, a parallel block for sanctions / PEP / affordability, a Consumer Duty node, a vulnerability node, and a decision node. The simplicity of the graph is bought by the richness of the state.
Deeper mechanism: the ontology of agency
The word “agent” is used so loosely that the underlying system boundary can disappear. The word is used so loosely in industry that it has become almost meaningless. A “fraud agent” in one bank is a rule engine. In another it is a multi-step LLM workflow. In a third it is a team of specialised LLM-backed roles coordinating under a supervisor. All three get marketed as “agents” and all three behave very differently.
Stuart Russell and Peter Norvig, in Artificial Intelligence: A Modern Approach, defined an agent as anything that perceives its environment through sensors and acts upon that environment through actuators. By that definition, a thermostat is an agent. A Roomba is an agent. Google Search is an agent. This is too broad to be useful in the context of LangGraph.
The useful definition, specific to the LLM era, is narrower. An LLM-based agent is a system in which a language model, given a specification of a task and a set of tools, can autonomously decide the sequence of reasoning steps and tool invocations needed to accomplish the task. The key words are autonomously and decide. If a human encodes the sequence (even cleverly, with many branches), it is not an agent. If the language model chooses the sequence at runtime based on intermediate observations, it is.
This narrower definition has three consequences. Agency belongs to the architecture, not to a model name. Agency varies by degree, from choosing between fixed routes to constructing long tool sequences. Every additional choice adds latency, cost and failure surface, so grant only the freedom the task can justify.
A useful distinction separates workflows, where code fixes the route and a model performs bounded sub-tasks, from agents, where a model may choose the next operation. Regulated systems should prefer the narrowest freedom that the task actually needs.
This spectrum, from left to right, is also roughly a cost and complexity gradient. Left-hand systems are cheap and simple. Right-hand systems are expensive and capable. Most banks run a portfolio. You will find single-LLM-call services in copywriting tools, prompt chains in FAQ responders, workflows in document summarisation, tool-using workflows in the majority of RM copilots, true agents in the most ambitious credit memo drafters, and multi-agent systems in the rare strategic initiatives that have had enough capital and patience to mature.
The consumer duty dimension
In a LangGraph-based mortgage agent, this translates into concrete design patterns. Every draft response is passed through a Consumer Duty node that checks for clarity (consumer understanding), flags vulnerability indicators, ensures no pressure selling language, and verifies that the product being offered is genuinely appropriate for the customer’s needs and circumstances. This node is not a rubber stamp. It is a separately-prompted call to the language model, given the draft response and the full context, asked to critique from a Consumer Duty perspective. If it raises concerns, the graph routes to a revision node or to human review.
class ConsumerDutyAssessmentState(TypedDict):
draft_response: str
customer_profile: dict
product_being_discussed: Optional[str]
# Four Consumer Duty outcomes
products_and_services_ok: Optional[bool]
price_and_value_ok: Optional[bool]
consumer_understanding_ok: Optional[bool]
consumer_support_ok: Optional[bool]
# Specific flags
pressure_selling_detected: bool
vulnerability_indicators: List[str]
accessibility_concerns: List[str]
clarity_issues: List[str]
# Decision
passes_consumer_duty: Optional[bool]
revision_required: bool
human_review_required: boolThat is a state object carrying regulatory intent. It makes the Consumer Duty explicit rather than implicit, and that explicitness is, in practice, the single biggest architectural concession that banking teams need to make to get their agent systems past the conduct risk team.
Build a reproducible workshop
Why environment discipline matters more in banking
Environment discipline is part of governance. A reproducible build lets a team connect observed behaviour to an exact dependency set, investigate a defect and restore a known route.
The third is change management. When your agent system is behind a change advisory board, you need to be able to describe what has changed between releases. “We upgraded LangGraph from 0.3.15 to 0.3.16” is an auditable change. “We ran pip install without pinning versions” is not.
The fourth is regulatory portability. UK banks operate across multiple jurisdictions. A mortgage agent deployed in the UK may share code with a wealth management assistant deployed in Singapore. The same codebase needs to behave identically in both. Environment discipline is what makes this practical.
With those reasons in mind, Build.
Worked example: install everything, step by step
Step 1: confirm your Python version.
Open a terminal and type:
python3 --versionUse a Python version supported by the selected LangGraph release, then pin the interpreter and resolved dependencies in the build record. The command below shows one local installation path; verify the current package requirements before running it.
brew install python@3.11On Windows, use the official installer from python.org, and importantly, tick the box that says “Add Python to PATH.” On Ubuntu:
sudo apt update && sudo apt install python3.11 python3.11-venvStep 2: pick a home for your LangGraph work.
mkdir -p ~/LangGraphProjects
cd ~/LangGraphProjectsThis directory will hold all your agent code for the rest of this edition. Inside it, each chapter’s code will go in its own subdirectory. This keeps the topology flat and inspectable.
Step 3: create a virtual environment.
A virtual environment is a directory that contains its own Python interpreter and its own set of installed packages, isolated from the system Python. Create one with:
python3 -m venv langgraph_envThis creates a langgraph_env directory containing
bin/, lib/, and a few configuration files.
That directory now holds everything Python-specific for this
project.
Step 4: activate the virtual environment.
On macOS and Linux:
source langgraph_env/bin/activateOn Windows PowerShell:
.\langgraph_env\Scripts\Activate.ps1(You may need to set the execution policy first with
Set-ExecutionPolicy RemoteSigned run as an
administrator.)
Your shell prompt should now show (langgraph_env) at the
start. This means that when you type python or
pip in this shell, you get the virtual environment’s
versions, not the system’s. Activation is per-shell; if you open a new
terminal, you have to activate again.
Step 5: install LangGraph and core packages.
pip install langgraph langchain langchain-openai langchain-google-vertexai langchain-community python-dotenv pydanticYou will also want development tools:
pip install jupyter ipykernel pytest pytest-asyncio black ruff mypyThese add: - jupyter and ipykernel: for interactive notebooks, useful when exploring agent behaviour - pytest and pytest-asyncio: for unit testing (we will use these from Chapter 5 onwards) - black: an opinionated code formatter - ruff: a fast linter that catches common mistakes - mypy: a static type checker that is invaluable when working with TypedDict state schemas
Step 6: pin your versions.
The single most important step for reproducibility:
pip freeze > requirements.txtThis writes the exact versions of every installed package (direct and transitive) to a file. Check this file into your source control. From now on, anyone else who wants to reproduce your environment can run:
pip install -r requirements.txtand get the identical setup. This is the primitive of banking-grade reproducibility. Without pinned versions, your “it worked yesterday” can become “it does not work today” because some transitive dependency released a minor version overnight.
Step 7: configure your environment variables.
Create a file called .env in your project directory:
touch .envOpen it in your editor and add (for development, using an OpenAI key):
OPENAI_API_KEY=sk-your-key-herefor an operating deployment, Merehaven Bank engineers would instead
configure Application Default Credentials for Vertex AI and reference
Secret Manager-held secrets at runtime. But for the exercises in this
edition, an OpenAI development key is simplest. Critically, add
.env to your .gitignore file immediately:
echo ".env" >> .gitignoreChecking API keys into source control is the single most common security incident in AI engineering. Do not be the person who has to file an incident report because a customer committed their key to a public repository.
Step 8: verify the installation.
python -c "import langgraph; print(langgraph.__version__)"You should see a version string like 0.3.x or higher. If
you see an import error, back up and check that your virtual environment
is activated (your prompt should show (langgraph_env)). If
activation is fine but the import fails, check your
requirements.txt against the versions that actually
installed.
That is your workshop. A dedicated Python, a dedicated virtual environment, pinned versions, secrets kept out of source control, development tools for formatting and linting. It is not glamorous. It is the foundation everything else sits on.
Deeper: what a container looks like for an operating deployment
The development setup we just built is fine for your laptop. for an operating deployment, we containerise. A container is a self-contained package that bundles your code, your Python interpreter, your dependencies, and just enough operating system to run them. The advantage is that the container runs identically on your laptop, on a CI server, on a staging GKE cluster, and on production. Here is a minimal release-tested Dockerfile for a LangGraph service:
# Use a specific, pinned base image
FROM python:3.11.9-slim-bookworm
# Create a non-root user (banking security standard)
RUN useradd -m -u 1000 appuser
# Set up the application directory
WORKDIR /app
RUN chown appuser:appuser /app
# Copy dependency specification first for layer caching
COPY --chown=appuser:appuser requirements.txt .
# Install dependencies as root, then drop privileges
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY --chown=appuser:appuser . .
# Switch to non-root user
USER appuser
# Expose the service port
EXPOSE 8080
# Run the service
CMD ["python", "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]Three build properties matter. The base image is pinned to a specific
version of Python on a specific Debian release; not
python:3.11 but python:3.11.9-slim-bookworm.
This matters because “3.11” will drift as Python releases patch
versions, and “slim” bases change frequently. Pinning to the specific
patch version gives you bit-level reproducibility. The second is that
the scenario runs as a non-root user, which is standard banking security
practice (container escapes are easier from root processes). The third
is that we install dependencies before copying the application code,
which lets Docker cache the dependency layer separately from the
fast-changing application layer.
The first graph: hello, LangGraph
Now that the workshop is ready, Build the smallest possible graph.
Create a file called lesson1.py in your project
directory:
# lesson1.py
# The simplest possible LangGraph: one node that greets.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# Define the state structure
class HelloWorldState(TypedDict):
greeting: str
# Define the node function
def greet_node(state: HelloWorldState) -> HelloWorldState:
"""Prepend 'Hello World, ' to whatever greeting is in the state."""
state["greeting"] = "Hello World, " + state["greeting"]
return state
# Initialise the graph and add the node
graph = StateGraph(HelloWorldState)
graph.add_node("greet", greet_node)
# Define the flow of execution using edges
graph.add_edge(START, "greet")
graph.add_edge("greet", END)
# Compile and run the graph
runnable = graph.compile()
result = runnable.invoke({"greeting": "from LangGraph!"})
# Output the result
print(result)
# Expected: {'greeting': 'Hello World, from LangGraph!'}Run it:
python lesson1.pyYou should see
{'greeting': 'Hello World, from LangGraph!'} on your
screen.
Walk through every line, because every line demonstrates something that will matter later.
The TypedDict definition of
HelloWorldState declares the shape of data that flows
through the graph. Python will not enforce this at runtime, but your
type checker will, and LangGraph will use the schema for its own
internal validation. When you start writing state objects with twenty
fields, this discipline becomes priceless. Every field you declare here
becomes a field the graph will track.
The node function greet_node takes a
state and returns a state. In this case, it mutates the state in place
and returns the same object, which is one valid pattern. Another valid
pattern is to return a dictionary containing only the fields you want to
update; LangGraph will merge this with the existing state. We will see
both in later chapters.
The graph construction starts with
StateGraph(HelloWorldState). The state schema is passed to
the graph constructor, which uses it both for validation and for
documentation. Every node you add will be expected to take a
HelloWorldState and return a HelloWorldState
(or a subset of its fields).
add_node("greet", greet_node) registers the function
under a name. The name matters: it is how you reference the node in
edges, in visualisations, and importantly, in audit traces. Pick names
that would make sense to a compliance reviewer. “greet” is fine for this
toy; “cross_reference_covenants” is better than “n17” for a production
node.
add_edge(START, "greet") and
add_edge("greet", END) define the flow. START
and END are special sentinel nodes built into LangGraph.
START is where execution begins; END is where
it terminates. Every graph must have at least one edge from
START to a real node, and at least one path from real nodes
to END.
compile() validates the graph structure and returns an
executable object. If your graph has errors (a node with no incoming
edge, a dangling reference to a node that does not exist), compile will
raise an exception with a useful error message. This is a point of
safety: you cannot deploy a structurally broken graph to production.
invoke() runs the graph synchronously, from
START to END, passing the initial state
through every node. It returns the final state. There are other
execution modes (stream, astream, abatch) which we will meet in later
chapters, but invoke is the one you will use most when developing and
testing.
Extending the toy: two nodes, one cycle-free graph
Extend the example to show composition. Add a second node:
# lesson1b.py
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class HelloWorldState(TypedDict):
greeting: str
def greet_node(state: HelloWorldState) -> HelloWorldState:
state["greeting"] = "Hello World, " + state["greeting"]
return state
def exclaim_node(state: HelloWorldState) -> HelloWorldState:
"""Append an exclamation mark to the greeting."""
state["greeting"] = state["greeting"] + "!"
return state
graph = StateGraph(HelloWorldState)
graph.add_node("greet", greet_node)
graph.add_node("exclaim", exclaim_node)
graph.add_edge(START, "greet")
graph.add_edge("greet", "exclaim")
graph.add_edge("exclaim", END)
runnable = graph.compile()
result = runnable.invoke({"greeting": "from LangGraph!"})
print(result)
# Expected: {'greeting': 'Hello World, from LangGraph!!'}The design shows how each node sees the result of the previous one.
The greet node produces "Hello World, from LangGraph!". The
exclaim node takes that and produces
"Hello World, from LangGraph!!". This composition
of transformations is the essence of the graph abstraction.
Each node is a pure transformation. The state carries state. The graph
coordinates.
This is also the first moment where the graph abstraction starts to pay for itself. In a chain-based world, you would have had to decide at the outset whether “greet then exclaim” was the right sequence. In the graph world, you can add, remove, or reorder nodes by changing edges. If next week you decide that the exclaim node should come before the greet node, you change two lines. If you decide that the exclaim node is only needed on certain inputs, you add a conditional edge. The architecture is mutable in a way that chains are not.
Visualising graphs: your future self will thank you
LangGraph has built-in support for rendering graphs to images, which
is invaluable for debugging, documentation, and the audit trail. Add
this utility module as display_graph.py:
# display_graph.py
import os
import platform
import random
import subprocess
def display_graph(runnable, output_folder: str = "output") -> str:
"""Render a compiled LangGraph to a PNG and open it."""
from langchain_core.runnables.graph import MermaidDrawMethod
os.makedirs(output_folder, exist_ok=True)
# Render the graph through the Mermaid API
png_bytes = runnable.get_graph(xray=1).draw_mermaid_png(
draw_method=MermaidDrawMethod.API
)
# Write the bytes to a file with a semi-random name
file_path = os.path.join(
output_folder, f"graph_{random.randint(1, 999999)}.png"
)
with open(file_path, "wb") as f:
f.write(png_bytes)
# Open the PNG with the OS's default viewer
if platform.system() == "Darwin":
subprocess.run(["open", file_path])
elif platform.system() == "Windows":
os.startfile(file_path)
else: # Linux
subprocess.run(["xdg-open", file_path])
return file_pathNow from any of your lesson files:
from display_graph import display_graph
display_graph(runnable)and you get a visual rendering of the graph in your default image viewer. for an operating deployment graphs, this visualisation becomes part of the architectural documentation. The compliance team will ask for diagrams. You want to be able to generate them automatically from the source of truth (the code), not maintain them separately (where they will drift out of date).
The failure mode: the dependency hell that eats your saturday
The failure mode in this section is not a bug in LangGraph. It is a bug in the Python packaging ecosystem, and it will bite you. Here is the shape of it.
One Saturday morning, you install a new version of LangGraph to try a new feature. Your virtual environment happily resolves dependencies, upgrades a few packages, and reports success. You run your existing code. It fails. The error is something obscure about Pydantic v1 versus v2 incompatibility, or about a method that moved from one module to another.
What has happened: a new version of LangGraph had updated constraints on one of its dependencies (typically Pydantic or LangChain core), and pip’s resolver upgraded that dependency to a version that is incompatible with code you wrote earlier. The fix, once you know what to do, is to pin your versions carefully and upgrade in deliberate steps, not in an eager blanket.
A second worked example: parameterising the greet node
Extend the Hello World example with explicit configuration. Suppose we want the greet prefix to be configurable rather than hard-coded. Here is a version with a config:
# lesson1c.py
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class HelloWorldState(TypedDict):
greeting: str
prefix: str # now in the state
def greet_node(state: HelloWorldState) -> HelloWorldState:
prefix = state.get("prefix", "Hello World, ")
state["greeting"] = prefix + state["greeting"]
return state
def exclaim_node(state: HelloWorldState) -> HelloWorldState:
state["greeting"] = state["greeting"] + "!"
return state
graph = StateGraph(HelloWorldState)
graph.add_node("greet", greet_node)
graph.add_node("exclaim", exclaim_node)
graph.add_edge(START, "greet")
graph.add_edge("greet", "exclaim")
graph.add_edge("exclaim", END)
runnable = graph.compile()
# Two invocations with different prefixes
result_en = runnable.invoke(
{"greeting": "from LangGraph!", "prefix": "Hello World, "}
)
result_fr = runnable.invoke(
{"greeting": "de LangGraph!", "prefix": "Bonjour le monde, "}
)
print(result_en) # {'greeting': 'Hello World, from LangGraph!!', ...}
print(result_fr) # {'greeting': 'Bonjour le monde, de LangGraph!!', ...}Here, we made the prefix a field on the state. This is the idiomatic way to parameterise a graph at invocation time: pass the parameters in the initial state. It keeps the graph pure, makes testing trivial (just invoke with different states), and means that the parameters are captured in the audit trail alongside everything else.
There is a separate mechanism, configurable fields, for cross-cutting concerns like the model to use, the API endpoint, the temperature setting. We will meet these in later chapters. For now, put parameters on the state.
A second thought experiment: what the state really is
The discussion has used “state” repeatedly, but the next example makes it concrete. When LangGraph passes state through a graph, what is actually happening under the hood? Inspect the mechanism.
When you call graph.compile(), LangGraph builds an
internal representation of your nodes and edges as a
pregel-style graph, named after Google’s classic paper
on graph computation. At runtime, when you call invoke,
LangGraph initialises a state object from your input, then for each
step, it identifies which nodes should execute based on the current
state and the graph’s structure. It runs those nodes (potentially in
parallel), collects their outputs, and merges them into the state using
the schema’s merging rules.
The merging rules are important. By default, when a node returns a
dictionary with a field, that field overwrites the state’s previous
value. But for certain field types, you want to accumulate rather than
overwrite. Consider a conversation history: each turn should append to
the list, not replace it. LangGraph supports this through
reducers, which are functions specified on fields via
Annotated types:
from typing import TypedDict, List, Annotated
from operator import add
class ConversationState(TypedDict):
messages: Annotated[List[str], add] # append, not overwrite
final_answer: str # overwriteAny node that returns {"messages": ["new turn"]} will
cause LangGraph to concatenate that list with the existing state’s
messages. Any node that returns {"final_answer": "foo"}
will overwrite the previous value. This subtlety will matter enormously
when we build memory-augmented agents in Chapter 6.
The reducer pattern is worth internalising because it is how you distinguish between values that replace and values that accumulate. In banking terms, a customer’s current account balance replaces (you do not “add” today’s balance to yesterday’s; you replace with the new value). A customer’s transaction history accumulates (you append each new transaction, you do not replace). Your state schema should declare this explicitly.
Another walk: a three-node example you can hold in your head
Consider one more complete worked example, this time with three nodes, because it captures a pattern you will see repeatedly: gather, transform, respond.
# lesson1d.py
# Three-node example: gather customer context, transform, respond.
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
class CustomerQueryState(TypedDict):
customer_id: str
query: str
# Filled in by nodes:
account_balance: Optional[float]
available_overdraft: Optional[float]
response: Optional[str]
def gather_node(state: CustomerQueryState) -> CustomerQueryState:
"""Simulate fetching account details from the core banking system."""
# in an operating environment this would be a real API call
state["account_balance"] = 2345.67
state["available_overdraft"] = 500.00
return state
def transform_node(state: CustomerQueryState) -> CustomerQueryState:
"""Compute the effective available funds."""
balance = state.get("account_balance", 0.0)
overdraft = state.get("available_overdraft", 0.0)
state["available_overdraft"] = balance + overdraft # total spendable
return state
def respond_node(state: CustomerQueryState) -> CustomerQueryState:
"""Generate a human-readable response."""
total = state.get("available_overdraft", 0.0)
state["response"] = (
f"Your available funds, including agreed overdraft, "
f"are £{total:,.2f}."
)
return state
graph = StateGraph(CustomerQueryState)
graph.add_node("gather", gather_node)
graph.add_node("transform", transform_node)
graph.add_node("respond", respond_node)
graph.add_edge(START, "gather")
graph.add_edge("gather", "transform")
graph.add_edge("transform", "respond")
graph.add_edge("respond", END)
runnable = graph.compile()
result = runnable.invoke({
"customer_id": "C-00123456",
"query": "How much can I spend today?"
})
print(result["response"])This is a toy but it shows the gather, transform, respond pattern that underlies most banking agents. The gather node fetches evidence. The transform node applies business logic. The respond node produces the customer-facing output. In a operating system, each of these would be much richer: the gather node might call three APIs in parallel; the transform node might apply regulatory business logic and Consumer Duty checks; the respond node might call an LLM to produce natural language. But the shape is the same.
Deeper mechanism: configuring structured logging from day one
One thing I wish every engineer at a UK bank would internalise on day
one is structured logging. Unstructured log lines
(print("something happened")) are useless for an operating
deployment debugging. Structured logs, where every log line is a JSON
object with consistent field names, let you run queries, build
dashboards, and trigger alerts.
Here is a minimal structured logging setup for a LangGraph project:
# logging_config.py
import logging
import sys
import json
from datetime import datetime, timezone
from typing import Any
class JSONFormatter(logging.Formatter):
"""Emit logs as JSON objects for Cloud Logging / CloudWatch ingestion."""
def format(self, record: logging.LogRecord) -> str:
log_obj: dict[str, Any] = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"severity": record.levelname,
"message": record.getMessage(),
"logger": record.name,
"module": record.module,
"function": record.funcName,
"line": record.lineno,
}
# Attach any extra fields passed via extra=
if hasattr(record, "state_snapshot"):
log_obj["state_snapshot"] = record.state_snapshot
if hasattr(record, "graph_node"):
log_obj["graph_node"] = record.graph_node
if hasattr(record, "session_id"):
log_obj["session_id"] = record.session_id
if record.exc_info:
log_obj["exception"] = self.formatException(record.exc_info)
return json.dumps(log_obj)
def configure_logging(level: int = logging.INFO) -> None:
root = logging.getLogger()
root.setLevel(level)
# Remove any existing handlers
for h in list(root.handlers):
root.removeHandler(h)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
root.addHandler(handler)And at the entry point of your application:
# main.py
from logging_config import configure_logging
import logging
configure_logging()
logger = logging.getLogger(__name__)
logger.info("service starting", extra={"version": "1.0.3"})Now every log line is a JSON object. Cloud Logging on GCP (and
CloudWatch on AWS) will ingest these natively and let you query by
field. You can search for all logs from a specific
session_id, all errors at a specific
graph_node, all events in a time window. For compliance and
debugging in a regulated environment, this is priceless.
The extra= parameter in the log call. Any keyword
arguments you pass via extra become fields in the JSON
output. In a LangGraph application, you will want to log the current
graph node name, the session ID, a snapshot of the state (or at least a
digest), and any tool calls or LLM responses. Structured logging makes
all of this queryable.
Type the case file
Why Python’s typing matters now more than ever
In a banking context, typing does more than catch bugs. It serves
three distinct governance purposes. First, it is a form of
documentation that cannot rot. A docstring can become
stale as code evolves; a type annotation cannot, because if it becomes
wrong, mypy or the test suite will flag it. Second, it is a form of
schema contract. When a node declares that it takes a
CreditMemoState and returns a CreditMemoState,
this is a contract that both parties (the graph and the node) can rely
upon. Third, it is a form of audit artefact. During
model risk validation, a reviewer can read the type signatures and
understand, in minutes, what each component does without having to trace
execution.
Good Python, in a bank, is typed Python. LangGraph makes this easy by using typing heavily in its own API.
TypedDict: the primary state mechanism
TypedDict, from PEP 589, is Python’s way of saying “I
want a dictionary, but with declared keys and types.” It is Python’s
preferred mechanism for LangGraph state. Here is the canonical form:
from typing import TypedDict, List, Optional
from datetime import datetime
class CreditMemoState(TypedDict):
# Always-present fields
customer_id: str
request_type: str
submitted_at: datetime
# Filled as the graph progresses (hence Optional)
financial_statements: Optional[dict]
sector_analysis: Optional[str]
covenant_check: Optional[dict]
consumer_duty_flags: Optional[List[str]]
# Always present, accumulates through the graph
narrative_paragraphs: List[str]
audit_entries: List[dict]Why TypedDict rather than a regular class? Three reasons.
First, LangGraph expects dict-shaped state because
it uses dictionary merging to apply node outputs to state. A TypedDict
is a dict at runtime; you can subscript it with
state["customer_id"]. A regular class would require
attribute access, and LangGraph’s internals would need a different code
path.
Second, TypedDict is lightweight. It is essentially a type hint on top of a regular dictionary; there is no runtime overhead beyond what you would pay for a dict anyway. Pydantic, by contrast, adds validation overhead on every assignment (which is often worth the cost, but not free).
Third, TypedDict plays well with JSON serialisation. LangGraph needs to serialise state for checkpointing, for cross-process passing, and for audit. A TypedDict maps cleanly to JSON. A Pydantic model does too, but TypedDict is one fewer conversion step.
Some notes on TypedDict that catch engineers out:
- Optional does not mean “may be absent”; it means “may be
None.”
Optional[str]is equivalent toUnion[str, None]. To say “this key may or may not be in the dict,” you useNotRequired[str](fromtyping_extensionsor Python 3.11+’styping). In the Merehaven worked scenario, the scenario uses NotRequired for fields that might not appear in the initial state. - TypedDict inheritance works but is shallow. If you inherit from a TypedDict and add fields, the child has all the parent’s fields plus the new ones. But there is no runtime class hierarchy to query; at runtime these are all just dicts.
- Total vs partial. By default a TypedDict is
total=True, meaning every declared field is required. You can saytotal=Falseto make all fields optional. In practice, you will usually useNotRequiredon specific fields rather than making the whole class partial.
Pydantic: the stricter alternative
Pydantic is a separate library that provides validated data classes. You declare your schema, Pydantic enforces it at runtime. When a state object violates the schema (missing required field, wrong type), Pydantic raises a clear error immediately rather than letting the error propagate silently.
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
class CreditMemoModel(BaseModel):
customer_id: str = Field(..., regex=r"^C-\d{8}$")
request_type: str = Field(..., pattern=r"^(extension|new|review)$")
submitted_at: datetime
financial_statements: Optional[dict] = None
sector_analysis: Optional[str] = None
narrative_paragraphs: List[str] = Field(default_factory=list)
audit_entries: List[dict] = Field(default_factory=list)
@validator("customer_id")
def customer_id_must_be_active(cls, v):
# Could call out to a database here
return vLangGraph supports Pydantic models as state schemas. The tradeoff is: Pydantic gives you runtime validation (great for catching bugs early and producing clear error messages at the boundary), at a small performance cost (maybe 5-10% overhead on state mutations) and some complexity overhead (field_validators, pre-validators, custom serialisation).
A useful boundary rule is: use Pydantic for the outer-most state schema (the one that crosses service boundaries), use TypedDict for internal, tightly-scoped state schemas used within a single graph or tool. This gives you validation where it matters (at trust boundaries) and avoids overhead where it doesn’t.
Annotated and reducers: the state merging primitive
We saw this in Chapter 2, but it deserves deeper treatment. When a
LangGraph node returns a dict, LangGraph needs to know how to merge that
dict into the existing state. For most fields, the default behaviour is
“overwrite.” For some fields, you want to “append” or “sum” or “union.”
This is what the Annotated type and reducers are for.
from typing import TypedDict, List, Annotated
from operator import add
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
class ChatbotState(TypedDict):
# Conversation history: append, never overwrite.
messages: Annotated[List[BaseMessage], add_messages]
# Working notes: append strings.
notes: Annotated[List[str], add]
# Retry count: replace (standard behaviour).
retry_count: int
# Cumulative cost: sum.
total_cost: Annotated[float, lambda a, b: a + b]
# Final answer: replace.
answer: stradd_messages is a special LangGraph reducer that
intelligently merges lists of BaseMessage objects. It handles a few
tricky cases: if you return a message with an ID that already exists in
the state, it replaces rather than appends (so you can update a message
in place); it preserves ordering; and it handles both single messages
and lists.
For custom reducers, any two-argument function that can combine the
current value and the new value will work. operator.add
works for lists and numbers. Lambda functions like
lambda a, b: {**a, **b} work for dicts (shallow merge). You
can write richer reducers for domain-specific needs (such as “merge two
IFRS 9 provision schedules by date, keeping the latest”).
The reducer is the seam between nodes. Every piece of state has either a default merger (replace) or a custom one. When you design a graph, thinking about the reducer for each field is part of the design. Get it wrong and your state will grow unboundedly, or lose information, or produce mysterious bugs.
Async and await: the concurrency model
LangGraph nodes can be either synchronous or asynchronous. The choice matters because it determines how the graph schedules work. A graph with all-sync nodes runs one at a time. A graph with async nodes can do I/O concurrently, which is critical for performance.
A synchronous node function looks like this:
def fetch_exposures_sync(state: MyState) -> MyState:
import requests
resp = requests.get(f"https://api.bank/exposures/{state['customer_id']}")
state["exposures"] = resp.json()
return stateAn asynchronous version looks like this:
import httpx
async def fetch_exposures_async(state: MyState) -> MyState:
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://api.bank/exposures/{state['customer_id']}"
)
state["exposures"] = resp.json()
return stateThe async version allows the graph to process other nodes while this one is waiting on network I/O. In a graph that needs to call three APIs in parallel, the async version completes in the time of the slowest call; the sync version takes the sum of the calls.
In LangGraph, you register async nodes the same way as sync ones:
graph.add_node("fetch_exposures", fetch_exposures_async)LangGraph’s executor will detect the node’s async-ness and schedule
it appropriately. You run an async graph with
runnable.ainvoke(...) or
await runnable.ainvoke(...). Mixing sync and async nodes in
the same graph is allowed; the executor handles this transparently.
Useful project rules:
- I/O-bound nodes (API calls, database queries, LLM
invocations) are async. The exception is if the underlying
client does not support async, in which case run the sync client inside
asyncio.to_threadto avoid blocking the event loop. - CPU-bound nodes (embedding computation, local inference, heavy data processing) are sync, and ideally run in a worker pool or a dedicated service. Async Python does not help with CPU-bound work because of the GIL.
- Graph invocation is async in an operating
environment. Even if all nodes are sync, using
ainvokekeeps your service’s event loop free to handle other requests.
The failure mode: reducers that lose data
The bug: after the graph ran, only the last node’s audit entry was in the state. All the earlier entries had been overwritten. The engineer spent two days assuming it was a threading issue.
The fix: audit_entries: Annotated[List[dict], add]. One
line. Now every node’s contribution is appended rather than
overwriting.
The lesson: whenever a field should accumulate, declare the reducer explicitly. If you are not sure, err on the side of declaring. The cost of a reducer is negligible; the cost of silent data loss can be weeks of investigation.
A thought experiment: designing the state for an autonomous mortgage agent
Consider the state design for something real. Imagine you are designing the state schema For the Merehaven worked scenario, an autonomous mortgage agent. The goal: from initial enquiry to agreement-in-principle within a single session, with Consumer Duty, vulnerability, affordability, sanctions, and PEP checks all performed.
What fields should be in the state? Build it up.
from typing import TypedDict, List, Optional, Annotated, Literal
from datetime import datetime
from operator import add
from pydantic import BaseModel
# --- Component models ---
class ApplicantDetails(BaseModel):
customer_id: Optional[str] = None
full_name: Optional[str] = None
date_of_birth: Optional[str] = None
current_address: Optional[str] = None
employment_status: Optional[str] = None
annual_income: Optional[float] = None
# Vulnerability indicators
declared_disability: Optional[bool] = None
declared_health_issues: Optional[bool] = None
class PropertyDetails(BaseModel):
property_address: Optional[str] = None
property_value: Optional[float] = None
property_type: Optional[str] = None
deposit_amount: Optional[float] = None
deposit_source: Optional[str] = None
class AffordabilityAssessment(BaseModel):
monthly_household_income: Optional[float] = None
monthly_household_commitments: Optional[float] = None
proposed_monthly_repayment: Optional[float] = None
stress_test_rate: Optional[float] = None
stressed_repayment: Optional[float] = None
passes_affordability: Optional[bool] = None
affordability_rationale: Optional[str] = None
# --- Main state ---
class MortgageAgentState(TypedDict):
# Session
session_id: str
subject_identifier: str
channel: Literal["web", "mobile", "branch", "phone"]
session_started_at: datetime
# Conversation
messages: Annotated[List[dict], add] # Each message is {role, content, ts}
current_turn: int
# Application data
applicant: ApplicantDetails
property: PropertyDetails
affordability: AffordabilityAssessment
# Compliance checks
consumer_duty_flags: Annotated[List[str], add]
vulnerability_indicators: Annotated[List[str], add]
sanctions_status: Literal["pending", "clear", "hit"]
pep_status: Literal["pending", "clear", "hit"]
kyc_status: Literal["pending", "clear", "incomplete", "failed"]
# Decision
decision: Optional[Literal["provisional_approve", "defer", "decline", "escalate"]]
decision_rationale: Optional[str]
# Audit
audit_trail: Annotated[List[dict], add]
model_versions: dict # replaced (overwritten with current snapshot)The design choices:
- The state nests Pydantic models for structured sub-domains (applicant, property, affordability). These are validated at the boundary.
- Scalar fields use TypedDict directly, with
Literaltypes for enumerated values. - Accumulating fields (messages, flags, indicators,
audit_trail) use
Annotated[..., add]so nodes can append without overwriting. - Singleton fields that are gradually filled in (decision,
rationale) are
Optionaland overwritten when the node produces the final value. model_versionsis a dict that is replaced each time, capturing a snapshot of which versions were in use.
A closer look at generics: list[str] versus list[str]
One detail that confuses many engineers moving to modern Python: the
difference between List[str] (from typing) and
list[str] (built-in generics). Both are valid; they mean
the same thing to the type checker. The built-in form is available from
Python 3.9 onwards. The typing form is available from Python 3.5
onwards.
For LangGraph work on Python 3.11+, both are fine. The built-in form
is slightly preferred because it avoids an import and reads more
naturally. The typing form is still necessary for
Annotated, Union, Optional, and
other constructs that have no built-in equivalent until Python
3.10+.
A useful rule: use list[str], dict[str, X],
tuple[X, Y] for simple containers; keep
from typing import ... for Annotated, TypedDict, Protocol,
Literal, and friends. This is the style the LangGraph source itself
tends toward, and it makes migration between projects smoother.
worked typing: a commercial banking state schema
from typing import Annotated, List, Optional, Literal
from typing_extensions import TypedDict
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage
from pydantic import BaseModel, Field
from datetime import datetime
from decimal import Decimal
import operator
class CustomerSnapshot(BaseModel):
"""Validated customer identity block. Crosses a trust boundary (ingress)."""
customer_id: str = Field(pattern=r"^CUS-\d{10}$")
segment: Literal["retail", "commercial", "corporate", "private"]
risk_band: Literal["low", "medium", "high"]
sanctions_cleared_at: datetime
consumer_duty_vulnerable: bool = False
class CreditExposure(BaseModel):
"""A single exposure record. Decimal for money, not float."""
product: Literal["mortgage", "card", "loan", "overdraft", "revolving_credit"]
balance: Decimal
limit: Optional[Decimal] = None
originated_at: datetime
ifrs9_stage: Literal[1, 2, 3]
class CreditMemoState(TypedDict, total=False):
# Identity, validated at boundary
customer: CustomerSnapshot
# Accumulating fields with explicit reducers
exposures: Annotated[List[CreditExposure], operator.add]
audit_log: Annotated[List[str], operator.add]
messages: Annotated[List[BaseMessage], add_messages]
# Single-writer fields (replace semantics are fine)
draft_memo: Optional[str]
risk_rating: Optional[Literal["A", "B", "C", "D", "E"]]
requires_human_review: bool
reviewer_notes: Optional[str]
# Telemetry, set once by the observability node
trace_id: str
started_at: datetime
model_versions: dict[str, str]Look at this schema carefully. Every type choice is deliberate.
CustomerSnapshot is a Pydantic model because it enters the
graph from an external API and must be validated. If a malformed
customer object arrived, the validator would reject it immediately with
a clear error message, exactly what you want at a trust boundary.
CreditExposure uses Decimal not
float for monetary balances. Floats cannot represent 0.1
exactly. In a banking system, rounding errors compound across millions
of calculations and eventually cause regulatory reporting discrepancies.
Use Decimal for money. Always.
exposures and audit_log use
operator.add as reducers because multiple nodes append to
them. messages uses add_messages because it
has special semantics (ID-based replacement for tool call updates).
draft_memo, risk_rating, and
reviewer_notes are single-writer fields; the default
replace semantics are correct. trace_id,
started_at, and model_versions are telemetry
fields written once by the observability node.
The total schema is about thirty lines and fully captures the contract of a credit memo graph. A new engineer joining the team can read this schema and know, without reading any node code, what the graph operates on and how fields combine.
Context managers: the unsung heroes of resource safety
Every production LangGraph application touches resources that must be
cleaned up: database connections, HTTP sessions, spans in a tracing
system, locks, file handles. Python’s context manager protocol, the
with statement, is the cleanest way to handle these.
LangGraph nodes almost always use context managers internally.
from contextlib import asynccontextmanager
from opentelemetry import trace
import httpx
tracer = trace.get_tracer(__name__)
@asynccontextmanager
async def traced_call(node_name: str):
"""Create a span and ensure it closes even on exception."""
with tracer.start_as_current_span(node_name) as span:
try:
yield span
except Exception as e:
span.record_exception(e)
span.set_status(trace.StatusCode.ERROR, str(e))
raise
async def fetch_exposures(state: CreditMemoState) -> dict:
async with traced_call("fetch_exposures") as span:
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(
f"https://api.internal.merehaven.test/exposures/{state['customer'].customer_id}"
)
resp.raise_for_status()
span.set_attribute("exposures.count", len(resp.json()))
return {"exposures": resp.json()}Three nested context managers, the tracer span, the HTTP client, and (implicitly) the resp object, are all guaranteed to clean up properly even if the node raises. This is how production nodes avoid leaking sockets, orphaned spans, and half-open database connections under load. It’s the kind of detail that makes the difference between a graph that runs fine in a lab and one that survives ninety-ninth percentile spikes in an operating environment.
Generators, iterators, and streaming
Production LLM applications need to stream. A chat UI that displays tokens as they arrive feels an order of magnitude more responsive than one that waits for the full response. LangGraph supports streaming through Python’s async iterator protocol, which builds on generators.
from typing import AsyncIterator
async def stream_tokens(state: ChatState) -> AsyncIterator[str]:
async for chunk in llm.astream(state["messages"]):
if chunk.content:
yield chunk.content
# In the FastAPI handler:
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
async def event_stream():
async for token in graph.astream({"messages": request.messages}):
yield f"data: {json.dumps(token)}\n\n"
return StreamingResponse(event_stream(), media_type="text/event-stream")The async for and yield combination is the
Python idiom for async generation. It’s simple to write and produces a
resource-efficient stream that releases backpressure naturally. For a
conversational AI in a banking app, Merehaven Bank RM copilot, for
example, this is required by the design. Relationship managers do not
want to wait three seconds for a full response; they want the first
tokens immediately.
Dataclasses versus Pydantic versus TypedDict: a final summary
Python now has three common ways to define structured records. The distinctions matter for LangGraph:
@dataclass: lightweight, has__init__,__repr__,__eq__generated. No validation. Good for internal objects that don’t cross boundaries and don’t need dict-style access.TypedDict: typed dict-style access. No validation, no methods. The natural fit for LangGraph state because state merging is dict-based.pydantic.BaseModel: full runtime validation, serialisation to/from JSON, custom validators. The right choice for ingress/egress boundaries.
For state, use TypedDict. For request/response models at API
boundaries, use Pydantic. For small internal value objects (for example,
a ScoringWeights struct used inside a node), a dataclass is
fine and has the smallest overhead.
Putting it all together: a typed async RAG node
This final example synthesises everything in the chapter: TypedDict state, Pydantic validation at the boundary, Annotated with add_messages, async/await, context managers, and Protocol-based substitution.
from typing import Annotated, List, Protocol, runtime_checkable
from typing_extensions import TypedDict
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage
from pydantic import BaseModel, Field
from contextlib import asynccontextmanager
import operator
class RAGQuery(BaseModel):
"""Validated at graph ingress."""
question: str = Field(min_length=1, max_length=2000)
customer_segment: str
locale: str = "en-GB"
@runtime_checkable
class VectorRetriever(Protocol):
async def aretrieve(self, query: str, k: int = 5) -> list[dict]: ...
class RAGState(TypedDict, total=False):
query: RAGQuery
retrieved: Annotated[List[dict], operator.add]
citations: Annotated[List[str], operator.add]
messages: Annotated[List[BaseMessage], add_messages]
answer: str
async def retrieve_node(state: RAGState, retriever: VectorRetriever) -> dict:
async with traced_call("retrieve"):
docs = await retriever.aretrieve(
state["query"].question,
k=5,
)
return {
"retrieved": docs,
"citations": [d["source"] for d in docs],
}
async def generate_node(state: RAGState) -> dict:
async with traced_call("generate"):
context = "\n\n".join(d["text"] for d in state["retrieved"])
prompt = f"Context:\n{context}\n\nQuestion: {state['query'].question}"
response = await llm.ainvoke(prompt)
return {
"messages": [AIMessage(content=response.content)],
"answer": response.content,
}If you can read this block and understand every choice, you are ready for Chapter 4, and for an operating deployment.
Protocols and structural typing: the escape hatch
Occasionally you need typing that goes beyond TypedDict and Pydantic.
LangGraph nodes sometimes need to accept “anything that behaves like a
retriever” or “anything that has an ainvoke method.”
Python’s answer is typing.Protocol, which provides
structural typing (duck typing with static checking).
from typing import Protocol, runtime_checkable
@runtime_checkable
class Retriever(Protocol):
async def aretrieve(self, query: str, k: int = 5) -> list[str]: ...
async def rag_node(state: RAGState, retriever: Retriever) -> dict:
docs = await retriever.aretrieve(state["query"])
return {"docs": docs}A deeper walkthrough: evolving a state schema
Version 1: the naïve approach
from typing import TypedDict
class CreditMemoStateV1(TypedDict):
customer_id: str
memo_text: str
approved: boolThis works for a trivial prototype but fails every real requirement. It has no audit trail (you cannot tell which LLM generated the memo, or when). It has no intermediate data (no exposures, no financials, no ratings). It has no error handling (if the credit search fails, where does the failure go?). A demo built on this state is the kind of demo that wins an internal hackathon and dies the first time a production incident requires root-cause analysis.
Version 2: adding messages and tools
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class CreditMemoStateV2(TypedDict):
customer_id: str
messages: Annotated[List[BaseMessage], add_messages]
exposures: List[dict]
financials: dict
proposed_rating: str
memo_text: str
approved: boolNow the scenario has a conversation history and intermediate data. This is enough to build a working agent. But it is still missing the audit fields that a bank will demand before production. Who ran this? When? Against which model version? Which policy version? What was the retrieval corpus at the time the memo was drafted? The moment model risk validation gets their hands on this, version 2 will come back with a list of required additions.
Version 3: release-tested with audit
from typing import TypedDict, Annotated, List, Optional
from datetime import datetime
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from pydantic import BaseModel
class Exposure(BaseModel):
product: str
balance: float
arrears_months: int
opened_on: datetime
class Financials(BaseModel):
turnover: float
ebitda: float
net_debt: float
interest_cover: float
class AuditTrail(BaseModel):
run_id: str
started_at: datetime
model_name: str
model_version: str
policy_pack_version: str
user_id: str
tool_calls: List[dict] = []
class CreditMemoStateV3(TypedDict):
# Inputs
customer_id: str
requested_facility: float
# Agent working memory
messages: Annotated[List[BaseMessage], add_messages]
exposures: List[Exposure]
financials: Optional[Financials]
# Agent outputs
proposed_rating: Optional[str]
proposed_rating_justification: Optional[str]
memo_text: Optional[str]
# Governance
audit: AuditTrail
requires_human_review: bool
human_decision: Optional[str]
# Error surface
errors: List[str]This is the shape of real production state in a banking agent. The
properties it now has. Separation of concerns: inputs,
working memory, outputs, governance, errors are grouped logically.
Typed sub-schemas: Exposure, Financials, and AuditTrail
are Pydantic models, so they get full runtime validation.
Explicit optionality: every field that is populated by
the agent is Optional, clarifying what is expected at input
versus what is produced. Audit first-class: AuditTrail
is a required field, not an afterthought; every node that changes
meaningful state appends a tool call entry.
The cost of this design is about thirty lines of code more than version 1. The payoff is that when an auditor or regulator asks “how did this memo get produced?” you can answer that question exhaustively.
Deeper mechanism: reducer semantics and concurrency
LangGraph’s reducers are what make parallel execution safe. Most state schemas can get away with the default “last write wins” semantics, but the moment you have any parallel branch (say, one node fetches exposures while another fetches market data in parallel) you need reducers that can merge those branches without losing data.
The built-in add_messages reducer is the canonical
example. It takes a list of messages from the incoming state and appends
the new messages, deduplicating by id. But you can write
your own.
Consider a credit memo that aggregates findings from three parallel investigators: a financial analyst, a compliance reviewer, and a relationship manager. Each produces a dict of findings keyed by topic. You want to merge these into a single findings dict in state.
from typing import TypedDict, Annotated, Dict, List
from operator import add
def merge_findings(left: Dict[str, List[str]], right: Dict[str, List[str]]) -> Dict[str, List[str]]:
"""Merge two finding dicts: concatenate lists for shared keys, union the rest."""
result = {**left}
for key, value in right.items():
if key in result:
result[key] = result[key] + value
else:
result[key] = value
return result
class InvestigationState(TypedDict):
customer_id: str
findings: Annotated[Dict[str, List[str]], merge_findings]
notes: Annotated[List[str], add]Now if three parallel nodes each emit
{"findings": {"financials": ["concern A"]}} and
{"findings": {"compliance": ["concern B"]}} and
{"findings": {"financials": ["concern C"]}}, the merged
result will be
{"financials": ["concern A", "concern C"], "compliance": ["concern B"]}.
Without the reducer, the last writer would clobber the others and you
would silently lose information.
Three production rules for reducers.
Rule 1: reducers must be associative and
commutative. LangGraph does not guarantee the order in which
parallel branches complete. Your reducer must produce the same result
regardless of order. add for lists is commutative; list
concatenation with dedup is commutative; “last write wins” is not
commutative and is exactly why it is dangerous in parallel contexts.
Rule 2: reducers must be pure. Do not mutate the input arguments. Return a new object. Mutation of a shared state reference across parallel threads is a recipe for Heisenbugs that disappear when you add logging.
Rule 3: reducers should be fast. They run on every state update. A reducer that does an expensive database call or a remote lookup will become a bottleneck. Keep the work in the node; keep the reducer to in-memory merging.
The failure mode: state bloat and its consequences
State bloat is the silent killer of long-running agents. Every time a node appends to state and nothing trims it, state grows. After twenty turns of a conversation, you might have 200KB of messages. After fifty, 500KB. At 1MB, serialisation to your checkpointer starts to take seconds. At 10MB, your graph grinds to a halt and you start getting timeouts from downstream services.
Three defensive patterns.
Pattern 1: message summarisation. When the message history exceeds a threshold (say, twenty messages or 4000 tokens), summarise the older portion into a single system message and retain only the last N raw messages. This is the pattern most production chat agents use by default.
from langchain_core.messages import SystemMessage, RemoveMessage
def summarise_if_needed(state: AgentState):
messages = state["messages"]
if len(messages) < 20:
return {} # nothing to do
old = messages[:-8]
summary = llm.invoke([
SystemMessage("Summarise this conversation concisely."),
*old,
])
# Use RemoveMessage to drop the old messages from state
remove_ops = [RemoveMessage(id=m.id) for m in old]
summary_msg = SystemMessage(content=f"Summary of earlier conversation: {summary.content}")
return {"messages": remove_ops + [summary_msg]}Pattern 2: selective field pruning. Mark some state fields as ephemeral (working memory that need not be persisted) and drop them before checkpointing. LangGraph supports this via explicit state key reset in nodes that finish a phase.
Pattern 3: external storage for large blobs. If you need to include a 2MB document in the reasoning flow, do not put it in state. Store it in object storage (GCS or S3) and put a reference in state (a URI plus a hash). Nodes that need the content fetch it on demand. This keeps state small and the checkpoint fast.
In BFSI contexts, bloat is especially pernicious because your checkpointer is typically a regulated database like Cloud SQL or DynamoDB, with its own row-size limits and its own performance profile. A 5MB row is not only slow; it may trip alerts, get flagged by the DBA team, and cause your agent to be escalated as an infrastructure abuser.
A thought experiment: typing a multi-step mortgage application
Sketch the state schema on paper before reading further. What fields do you need? What types? Which are Optional? Which need reducers? Which need audit fields?
Here is a candidate design.
from typing import TypedDict, Annotated, List, Optional, Literal
from datetime import date
from pydantic import BaseModel, Field
class Applicant(BaseModel):
customer_id: str
date_of_birth: date
employment_status: Literal["employed", "self_employed", "retired", "other"]
gross_annual_income: float
other_income: float = 0
class Property(BaseModel):
postcode: str
property_type: Literal["freehold", "leasehold"]
estimated_value: float
years_remaining_on_lease: Optional[int] = None
class AffordabilityAssessment(BaseModel):
monthly_income_net: float
committed_expenditure: float
stress_tested_rate: float
max_supportable_ltv: float
passed: bool
failure_reasons: List[str] = []
class MortgageApplicationState(TypedDict):
# Inputs
applicants: List[Applicant]
subject_property: Property
requested_loan: float
requested_term_years: int
# Agent working state
messages: Annotated[List, add_messages]
affordability: Optional[AffordabilityAssessment]
credit_data: Optional[dict]
outstanding_queries: List[str]
# Decision outputs
proposed_rate: Optional[float]
proposed_term: Optional[int]
decision: Optional[Literal["offer", "refer_to_human", "decline"]]
decision_reasons: List[str]
# Governance
audit: AuditTrail
fca_consumer_duty_checks: dict
mcob_compliance_checks: dictThe omissions matter. It does not store the full conversation transcript in place of structured fields. It does not conflate inputs with outputs. It does not let the agent’s free-text reasoning become the source of truth. Those are common anti-patterns that feel natural early in a project and become painful to refactor later.
Operating boundary: typing, linting, and CI gates
On the engineering side, how do you make sure your team actually uses these types rigorously? Four things, in escalating severity.
First, mypy strict mode as a CI gate. Any PR that
introduces a type error fails the build. Start with
--strict on new modules and gradually extend to older ones.
In the Merehaven worked scenario, the Python code for AI services runs
in strict mode, with Any explicitly disallowed except in
explicitly labelled interop boundaries.
Second, Pydantic for all external boundaries. Anything that crosses a network boundary (API input, Pub/Sub message, database row) uses a Pydantic model with runtime validation. TypedDict is fine for internal graph state but not for crossing trust boundaries.
Third, schema evolution reviews. Any change to a state schema that is persisted (such as a checkpointer schema) must be reviewed and migrated intentionally. Adding a required field is a breaking change; adding an optional field with a default is backwards compatible. Removing a field requires a deprecation cycle.
Fourth, runtime assertions at critical points. Even with mypy and Pydantic, runtime assertions at the start of each node (“I expect this state shape”) catch the cases where bad data has crept in from a path not covered by static typing. These should be cheap to evaluate and loud when they fail.
Give the graph a grammar
The four primitives
A LangGraph has exactly four primitive elements. Everything else, subgraphs, tool calls, ReAct loops, hierarchical teams, plan-and-execute architectures, is built from these four things. Learn them cold.
The first primitive is state. State is the ledger of the computation. It is a Python object (a TypedDict or a Pydantic model) that carries all the information the graph cares about. At every step, state flows from one node to the next, being updated according to each field’s reducer.
The second primitive is the node. A node is a function (usually async) that takes state, does some work (calls an LLM, queries a database, runs a tool), and returns a partial update to the state. Nodes are pure with respect to their declared state updates: whatever they return is merged into state via reducers, and nothing else. A node that sneakily writes to a global variable is a node that will bite you in an operating environment.
The third primitive is the edge. An edge is a fixed arrow: “when this node finishes, go to that node.” Edges are deterministic. The simplest graph, input → process → output, is three nodes connected by two edges.
The fourth primitive is the conditional edge. A conditional edge is a branch: “after this node, look at state, and pick one of several next nodes.” Conditional edges are what make a LangGraph more than a linear chain. They let you route based on classification, retry on failure, escalate to a human, or loop back for another iteration.
These four primitives, state, node, edge, conditional edge, are sufficient to express every agent architecture in this edition. They are also sufficient to confuse engineers who’ve never seen them framed this way. The rest of the chapter walks through each in depth.
Nodes: the functional unit
A node in LangGraph is, The mechanism is direct: a callable. Here is the simplest node possible:
def add_one(state: dict) -> dict:
return {"count": state.get("count", 0) + 1}That’s it. A function that takes state, returns a partial update. No registration ceremony, no base class to inherit from, no decorator required. LangGraph embraces Python functions as first-class agents.
in an operating environment, nodes become more interesting:
from langchain_core.messages import AIMessage
from langchain_google_vertexai import ChatVertexAI
llm = ChatVertexAI(model_name="gemini-1.5-pro", temperature=0.2)
async def draft_memo_node(state: CreditMemoState) -> dict:
"""Draft a credit memo from customer and exposure data."""
prompt = build_memo_prompt(state["customer"], state["exposures"])
response = await llm.ainvoke(prompt)
return {
"draft_memo": response.content,
"messages": [AIMessage(content=response.content)],
"audit_log": [f"memo drafted by {llm.model_name} at {now_iso()}"],
}This node reads customer data and exposures from state, asks Gemini
to draft a memo, and returns three updates: the draft text, an AI
message (which will be appended to messages via
add_messages), and an audit log entry (which will be
appended via operator.add). It never mutates the input
state; it returns a partial update and lets LangGraph merge it.
Nodes can be synchronous or asynchronous. For banking work, they should almost always be asynchronous because almost every node waits on I/O: an LLM call, a database query, a REST API. Async nodes release the event loop during waits and let the service process other requests concurrently.
Nodes can also be objects with an __call__ method, which
is useful when a node needs configuration. For example, a “score_credit”
node might need a scoring model to be injected:
class CreditScorer:
def __init__(self, model, threshold: float = 0.6):
self.model = model
self.threshold = threshold
async def __call__(self, state: CreditMemoState) -> dict:
features = extract_features(state["customer"], state["exposures"])
score = await self.model.apredict(features)
return {
"credit_score": score,
"risk_rating": self.bucket(score),
"audit_log": [f"score {score:.3f} bucketed to {self.bucket(score)}"],
}
def bucket(self, score: float) -> str:
if score > 0.85: return "A"
if score > 0.70: return "B"
if score > 0.55: return "C"
if score > 0.40: return "D"
return "E"This pattern, a class with __call__ that is registered
as a node, is how you inject dependencies cleanly. It’s also how you
make nodes testable: you construct a CreditScorer with a mock model and
call it with synthetic state.
Edges: the simple connectors
Edges are the simplest primitive. An edge says: after node X, go to node Y. Nothing more. Here’s the syntax:
from langgraph.graph import StateGraph, END
builder = StateGraph(CreditMemoState)
builder.add_node("fetch", fetch_node)
builder.add_node("draft", draft_node)
builder.add_node("score", score_node)
builder.set_entry_point("fetch")
builder.add_edge("fetch", "draft")
builder.add_edge("draft", "score")
builder.add_edge("score", END)
graph = builder.compile()This graph is fully linear: fetch → draft → score → END. It is, in effect, a chain. And that’s fine, sometimes a chain is what you need, and LangGraph makes it trivial to express. The power of LangGraph isn’t that you can’t build chains with it; it’s that you can start with a chain and add conditional branches later without rewriting.
The END sentinel is special. It’s imported from
langgraph.graph and marks the graph’s terminal node. When
execution reaches END, the graph returns the final state to
the caller. You can have multiple edges leading to END, a
graph can have many exit points, but you cannot exit without hitting END
somewhere. Some graphs have a single END; others (for example, a
customer service triage) have several, one per resolved path.
Conditional edges: the branching primitive
Conditional edges are where LangGraph earns its name. A conditional edge is a function that, given state, returns the name of the next node. Here’s the classic shape:
def route_after_score(state: CreditMemoState) -> str:
if state["risk_rating"] in ("D", "E"):
return "human_review"
if state["requires_second_opinion"]:
return "committee_review"
return "format_output"
builder.add_conditional_edges(
"score",
route_after_score,
{
"human_review": "human_review",
"committee_review": "committee_review",
"format_output": "format_output",
}
)Three properties matter. First, the router is a pure function of
state: no side effects, no I/O. It reads state, returns a string.
Second, the third argument to add_conditional_edges is a
mapping from router-returned strings to node names. This dual-layer
addressing, the router can return any string; the map binds it to a
concrete node, makes refactors safer. If you rename a node, you change
the map, not every router. Third, conditional edges support multiple
branches natively; there’s no nested if/else pattern to
manage.
Conditional edges can do more than pick one next node. They can also return a list of next nodes, which causes fan-out: execution proceeds in parallel along multiple branches, and state merges when they rejoin. This is how you implement map-reduce patterns inside a graph.
def fan_out_to_scorers(state: CreditState) -> list[str]:
"""Run multiple scoring models in parallel, then merge."""
return ["score_lgd", "score_pd", "score_ead"]Each branch runs independently. When all three complete, LangGraph merges their updates into state using the reducers defined on each field, and the next edge fires. This parallelism is sometimes called a “barrier join” in concurrency literature; LangGraph implements it for you.
Building the graph: the stategraph API
The StateGraph class is the builder you use to assemble
a graph from nodes and edges. The pattern is always the same:
- Instantiate
StateGraph(StateType), passing the state schema so LangGraph knows the reducers. - Add nodes with
.add_node(name, function). - Set the entry point with
.set_entry_point(name). - Add edges with
.add_edge(from_name, to_name). - Add conditional edges with
.add_conditional_edges(from_name, router_fn, mapping). - Compile with
.compile(). Optionally pass acheckpointerto enable persistence.
Here is a full, production-shaped example for a commercial banking credit memo graph:
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
def build_credit_memo_graph(llm, scorer, retriever, saver):
builder = StateGraph(CreditMemoState)
# Ingestion and enrichment
builder.add_node("validate", validate_node)
builder.add_node("fetch_exposures", fetch_exposures_node)
builder.add_node("retrieve_policies", make_retrieve_node(retriever))
# Analysis
builder.add_node("compute_ratios", compute_ratios_node)
builder.add_node("score", CreditScorer(scorer))
builder.add_node("draft_memo", make_draft_node(llm))
# Review paths
builder.add_node("committee_review", committee_review_node)
builder.add_node("human_review", human_review_node)
builder.add_node("finalise", finalise_node)
# Flow
builder.set_entry_point("validate")
builder.add_edge("validate", "fetch_exposures")
builder.add_edge("fetch_exposures", "retrieve_policies")
builder.add_edge("retrieve_policies", "compute_ratios")
builder.add_edge("compute_ratios", "score")
builder.add_edge("score", "draft_memo")
builder.add_conditional_edges(
"draft_memo",
route_after_draft,
{
"human_review": "human_review",
"committee_review": "committee_review",
"finalise": "finalise",
}
)
builder.add_edge("human_review", "finalise")
builder.add_edge("committee_review", "finalise")
builder.add_edge("finalise", END)
return builder.compile(checkpointer=saver)That is not a demo graph. That is roughly the shape of a real credit
memo pipeline at a UK high-street bank. The nodes are named for what
they do; the edges follow a clear narrative (ingest, enrich, analyse,
route to review, finalise); the router makes a single branching decision
based on risk and complexity. A new engineer can read the
builder section and understand the control flow in under a
minute.
Visualising the graph
Here is the graph we just built, rendered as Mermaid:
Figure 4.1: A production credit memo graph. Note the single conditional edge (route_after_draft) doing all the routing work. A graph with one well-placed decision is usually better than one with five scattered ones.
Execution semantics: how LangGraph actually runs
When you call graph.invoke(initial_state), what happens?
Understanding this is the difference between debugging in an hour and
debugging in a week.
LangGraph executes the graph as a series of supersteps. A superstep is a single discrete advance of the graph. In each superstep:
- LangGraph looks at the current state and determines which nodes are active (ready to run).
- All active nodes run concurrently (in parallel, if async).
- When all active nodes return, their updates are merged into state via the reducers defined on each field.
- LangGraph examines the edges leaving the nodes that just ran and determines the next set of active nodes.
- If the next set is empty or reaches END, execution halts.
This design has several consequences that trip up newcomers. First,
nodes in the same superstep see the same input state; they do not see
each other’s updates until the next superstep. This is by design, it
makes parallel execution safe. Second, if two nodes both write to the
same field in the same superstep, their updates are merged according to
the field’s reducer. If the reducer is the default (replace), only one
of the updates survives (which one is deterministic but depends on
internal order); if the reducer is operator.add, both
updates are concatenated. This is why reducers are so important for
parallel execution. Third, a node cannot peek ahead at what the next
node will do. A node only has access to current state plus its own
logic.
Figure 4.2: Supersteps are barriers. All updates in a superstep are merged before the next one begins.
Thinking about execution: the pregel heritage
LangGraph’s execution model is inspired by Pregel, Google’s bulk-synchronous parallel graph processing framework. In Pregel, compute happens in supersteps, and messages passed during one superstep become visible in the next. LangGraph inherits this cleanliness: the superstep discipline makes parallelism safe and debugging tractable. When you inspect a trace of a LangGraph run, you see a sequence of well-defined superstep transitions, not a tangled thread-of-execution chart.
Failure modes in graph design
Three patterns reliably break graphs in an operating environment, and all three are about graph topology rather than node correctness.
The first is unreachable nodes. You add a node and
forget to add an edge into it. The node is in the graph; it will never
run. This can silently break functionality for months. Mitigation: a
test that verifies every node in the graph is reachable from the entry
point. LangGraph exposes this via graph introspection; write a test that
asserts
graph.nodes.keys() == reachable(graph, entry_point).
The second is infinite loops. A conditional edge
routes back to a node that, given the updated state, routes back again.
Without a cycle-breaker, the graph runs forever (until it hits a
recursion depth limit and crashes). Mitigation: (a) include a loop
counter in state and test it in the router; (b) set
recursion_limit on the graph to a sane upper bound; (c) for
ReAct agents specifically, use the max_iterations
convention and have the router force a terminal path after the
limit.
The third is unbounded fan-out. A fan-out router returns a list of nodes whose length is driven by data, for example, “one branch per account.” If an edge case causes the list to be enormous (ten thousand accounts), the graph may spawn more concurrent work than the event loop can handle. Mitigation: cap fan-out width in the router, or chunk the work into batches of reasonable size.
[!warning] failure specimen Early in the credit memo rollout, we had a graph that fanned out one branch per exposure. For a normal retail customer, this meant three or four branches. For a large commercial customer with over four hundred exposures, the graph spawned four hundred parallel branches, saturated the event loop, and timed out the Cloud Run instance. The fix was a fan-out cap of twenty with pagination over the remainder. No production incidents since.
Regulated banking walkthrough: the rm copilot graph
Consider another full graph, the Relationship Manager copilot, to reinforce the patterns.
def build_rm_copilot_graph(llm, crm_tool, docs_retriever, saver):
builder = StateGraph(RMState)
builder.add_node("classify_intent", make_classify_node(llm))
builder.add_node("fetch_customer", make_crm_node(crm_tool))
builder.add_node("retrieve_policies", make_retrieve_node(docs_retriever))
builder.add_node("answer_simple", make_answer_node(llm, style="concise"))
builder.add_node("answer_complex", make_answer_node(llm, style="detailed"))
builder.add_node("escalate", escalate_node)
builder.set_entry_point("classify_intent")
builder.add_conditional_edges(
"classify_intent",
route_by_intent,
{
"simple_question": "answer_simple",
"customer_specific": "fetch_customer",
"policy_question": "retrieve_policies",
"escalate": "escalate",
}
)
builder.add_edge("fetch_customer", "answer_complex")
builder.add_edge("retrieve_policies", "answer_complex")
builder.add_edge("answer_simple", END)
builder.add_edge("answer_complex", END)
builder.add_edge("escalate", END)
return builder.compile(checkpointer=saver)The shape. Entry point is a classifier node (runs an LLM with a
simple taxonomy prompt). The router branches on the classification:
simple chat gets a fast path; customer-specific queries fetch CRM data
first; policy questions run a retrieval step first; anything flagged as
escalation goes to a human. All paths converge to END (except the fan-in
through answer_complex).
Figure 4.3: The RM copilot graph. One classifier, one router, three worker paths, plus an escalation. Most useful graphs are not much more complex.
Testing graphs: unit, integration, and graph-shape
Graphs deserve three levels of testing. Unit tests verify individual node behaviour with mocked dependencies. Integration tests run the compiled graph end-to-end with real or fake LLMs against a golden set of inputs. Graph-shape tests verify structural properties: every node is reachable; every router returns a valid target; no cycles exist without a bound.
Here is a graph-shape test in pytest style:
def test_credit_memo_graph_shape():
graph = build_credit_memo_graph(mock_llm, mock_scorer, mock_retriever, None)
schema = graph.get_graph()
assert "validate" in schema.nodes
assert "finalise" in schema.nodes
assert any(e.target == "human_review" for e in schema.edges)
assert any(e.target == END for e in schema.edges)
def test_router_exhaustiveness():
"""Every value in the router mapping must be an actual node."""
graph = build_credit_memo_graph(mock_llm, mock_scorer, mock_retriever, None)
schema = graph.get_graph()
node_names = set(schema.nodes.keys()) | {END}
for edge in schema.edges:
assert edge.target in node_namesThese tests run in milliseconds because they don’t invoke any nodes. They catch the most common class of graph bug: a misspelled node name in a conditional edge map.
🗜 subgraphs: composition for the large
As a graph grows beyond a dozen nodes, it becomes hard to reason about. LangGraph supports subgraphs, a compiled graph used as a single node in a parent graph. Subgraphs are how you manage complexity.
retrieval_graph = build_retrieval_subgraph(retriever, reranker)
scoring_graph = build_scoring_subgraph(scorer_models)
parent_builder = StateGraph(CreditMemoState)
parent_builder.add_node("retrieve", retrieval_graph) # subgraph as node
parent_builder.add_node("score", scoring_graph) # subgraph as nodeState flows into the subgraph, through its internal nodes, back out to the parent. The subgraph is a black box from the parent’s perspective, which means you can develop it, test it, and reason about it independently.
Subgraphs are also the unit of reuse across projects. In the Merehaven worked scenario, the scenario has a library of shared subgraphs: a “sanctions check” subgraph used by every money-movement graph; a “consumer duty gate” subgraph used by every customer-facing graph; a “PII redaction” subgraph used by every logging path. A new agent In the Merehaven worked scenario, snaps these in as nodes, and inherits the compliance posture for free.
Node state, edge state, and the visibility rule
One subtle property of LangGraph: nodes have full access to state, edges (routers) have full access to state, but nodes cannot “see” edges directly. This asymmetry matters because it keeps nodes portable. A node written for one graph can be dropped into another graph with a different topology, because the node doesn’t know anything about the graph around it. It only knows about state.
Treat attempts to bypass the visibility rule as a design smell. Promote the required information into typed state so the dependency is explicit and testable.
Thought experiment: redesigning the scoring graph
[!tip] Exercise You have a scoring graph with three scoring nodes (PD, LGD, EAD) running in parallel, then a combining node that produces the final rating. The compliance team asks: “Can we guarantee that if PD fails, we don’t produce a rating at all, we escalate?” How would you change the graph?
The answer: use a conditional edge after the fan-in. After the
combining node, branch based on whether any input was missing. If PD (or
any critical input) is None, route to
escalate_missing_input rather than
produce_rating. The beauty of LangGraph is this kind of
compliance invariant is expressible declaratively in the graph, not
hidden in imperative code.
def route_after_combine(state: ScoringState) -> str:
if state["pd"] is None or state["lgd"] is None or state["ead"] is None:
return "escalate_missing"
if state["combined_score"] is None:
return "escalate_missing"
return "produce_rating"Simple, auditable, and easy to evolve: tomorrow the compliance team may add “escalate if PD > 0.8,” and it’s a one-line change.
Part II: Bound the agent
An agent is a route that may choose its next operation. That freedom is useful only when tools, state and stopping rules make the choice inspectable.
Assemble a bounded tool-using agent
What we mean by “agent” in this chapter
The word “agent” is overloaded. In academic literature, it can mean almost anything autonomous. In the LangGraph world, we’ll use a precise definition: an agent is a graph whose behaviour includes an LLM making choices about what to do next, typically by calling tools. Agents observe, think, act, and observe again. The thinking is done by the LLM; the acting is done by tools; the observing is done by reading the results back into state.
Our first agent is going to be small. It will accept a natural-language question about a banking customer, decide whether it needs to call a tool (such as a customer-lookup or a policy-search), call it if needed, interpret the result, and reply. This is the simplest useful agent pattern in banking, and once you understand it, everything else in the remaining patterns are variations.
The materials: what we are assembling
A minimum viable agent needs four components in addition to the LangGraph primitives from Chapter 4:
Here is the entire code, and then we’ll take it apart piece by piece.
from typing import Annotated, List, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END, add_messages
from langgraph.prebuilt import ToolNode
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage
from langchain_core.tools import tool
from langchain_google_vertexai import ChatVertexAI
# 1. Define tools
@tool
def get_customer(customer_id: str) -> dict:
"""Return basic details for a customer by ID."""
# In real code, this calls the CRM.
return {"id": customer_id, "name": "Alice Smith", "segment": "retail"}
@tool
def search_policies(query: str, k: int = 3) -> list[dict]:
"""Search the internal policy library."""
return [{"title": "Chargeback policy", "snippet": "..."}]
tools = [get_customer, search_policies]
# 2. Define state
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
# 3. Build LLM with tools bound
llm = ChatVertexAI(model_name="gemini-1.5-pro", temperature=0.1)
llm_with_tools = llm.bind_tools(tools)
# 4. Define nodes
SYSTEM_PROMPT = """You are an assistant for Merehaven Bank peers.
Use the tools to look up customer information and policies.
Always cite policy titles when you reference policies."""
def agent_node(state: AgentState) -> dict:
system = SystemMessage(content=SYSTEM_PROMPT)
response = llm_with_tools.invoke([system] + state["messages"])
return {"messages": [response]}
tool_node = ToolNode(tools)
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "tools"
return "__end__"
# 5. Build graph
builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.set_entry_point("agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")
agent = builder.compile()That is a complete, working, production-shaped agent in fifty lines. It has a system prompt, two tools, a state schema, a graph with one conditional edge, and a loop between the agent node and the tools node. Everything else in this section is making sure you understand each line.
Piece 1: tools and the
@tool decorator
The @tool decorator is how LangChain turns a Python
function into something the LLM can call. It does three things. It
captures the function’s signature, its argument names,
types, and defaults. It captures the docstring as the
description the LLM sees. And it wraps the function in a class that
knows how to be invoked from a model-generated tool-call object.
Two patterns I enforce on every team. First, the docstring must start with a verb phrase describing what the tool does (“Return…” or “Compute…” or “Send…”). Second, every argument must have a type hint, because LangChain uses those hints to generate the JSON Schema the LLM sees. A tool with untyped arguments is a tool the LLM cannot call reliably.
@tool
def compute_ltv(
loan_amount: float,
property_value: float,
) -> float:
"""Return the loan-to-value ratio as a percentage.
loan_amount and property_value are in GBP.
"""
return 100.0 * loan_amount / property_valueGemini and Claude both parse this as: a function named
compute_ltv that takes two floats (loan_amount and
property_value) and returns a float. When the user asks “what’s the LTV
for a £300k loan on a £400k property?”, the model generates a tool-call
with loan_amount=300000 and
property_value=400000, invokes the tool, receives 75.0, and
reports that back to the user. All of that happens via the graph
loop.
Piece 2: state with
add_messages
The agent’s state is deliberately minimal: just a single field,
messages, annotated with add_messages as the
reducer. Why? Because messages are the medium of the agent’s thinking.
Every LLM call appends new messages. Every tool invocation appends a
ToolMessage with the result. The agent’s memory of what has
happened is the messages list.
add_messages is a clever reducer. It doesn’t just
concatenate lists; it dedupes by message ID and allows updates to
existing messages (which is important for streaming and for tool-call
updates). Always use add_messages for message fields in
LangGraph state. The default reducer (replace) would destroy the
conversation history.
In richer agents, you will add more state fields: an
answer field that the final node populates, an
escalate boolean, a tool_call_count integer
for limiting loops, a trace_id for observability. But start
with just messages and add fields as you find concrete needs.
Piece 3: the agent node
The agent node is simple: it builds a list of messages starting with
a system prompt, adds the conversation so far, and invokes the LLM. The
LLM returns either a regular AIMessage (if it’s answering directly) or
an AIMessage with tool_calls populated (if it wants to call
a tool).
def agent_node(state: AgentState) -> dict:
system = SystemMessage(content=SYSTEM_PROMPT)
response = llm_with_tools.invoke([system] + state["messages"])
return {"messages": [response]}Three subtleties to note. First, the system prompt is prepended at
every invocation, not stored in state. This keeps the system prompt as a
constant owned by the code, not subject to manipulation by the
conversation. Second, the node returns a partial update, a single
message, rather than replacing the whole messages field.
add_messages handles the merge. Third, the LLM is
llm_with_tools, not raw llm. Binding the tools
makes them available via the model’s function-calling interface; the LLM
literally sees the tool schemas in its prompt template.
Piece 4: the tools node
ToolNode is a prebuilt node from
langgraph.prebuilt. It takes the list of tools at
construction time and knows how to execute any tool the LLM has
requested via a tool-call. When the tools node runs, it looks at the
last message in state, finds the tool calls, executes them (in parallel
if there are multiple), and returns ToolMessage objects with the
results.
You could write this yourself, it’s maybe twenty lines, but
ToolNode handles the error cases (invalid arguments,
exceptions during tool execution, missing tool names) more thoroughly
than most first-draft implementations. Use it.
If you need custom behaviour (for example, logging every tool call,
enforcing authorization before a tool runs, injecting rate limits),
subclass ToolNode or wrap it in a higher-level node.
Piece 5: the conditional edge
The single conditional edge is the brain of the loop. After the agent node runs, we check the last message: does it have tool calls? If yes, go to the tools node; if no, end.
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "tools"
return "__end__"That __end__ sentinel is how you return END from a
conditional edge. After the tools node runs, the fixed edge
builder.add_edge("tools", "agent") sends control back to
the agent for the next iteration. The loop continues, agent → tools →
agent → tools → …, until the agent produces a tool-call-free response
and the router returns __end__.
This is the canonical “ReAct loop” from Yao et al. (2023), albeit simpler than the paper’s textual-reasoning version because we’re using native function calling. Chapter 8 will cover the deeper ReAct pattern with explicit reasoning traces.
The first worked agent: an rm copilot q&a
Consider a concrete case. Imagine a Merehaven Bank RM asks: “What is the current balance for customer CUS-0012345678 and does their profile require Consumer Duty vulnerable-customer handling?”
Here is, roughly, what happens inside the graph:
Superstep 1 (agent): The LLM reads the message. It
recognises it needs to look up the customer. It emits an AIMessage with
a tool_call for
get_customer(customer_id="CUS-0012345678").
Superstep 2 (tools): ToolNode executes
get_customer, receives a dict with the customer’s name,
balance, and vulnerability flag. It emits a ToolMessage with that dict
as the content.
Superstep 3 (agent): The LLM reads the tool result. It now has enough to answer. It emits an AIMessage with the answer, citing the vulnerability policy by name (because the system prompt told it to).
Superstep 4 (END): The router sees no tool_calls in
the last message and returns __end__. The graph halts. The
final state contains four messages: human → AI (with tool_call) →
tool_result → AI (answer).
This is thinking. Not thinking in the philosophical sense, but thinking in the operational sense: the agent decided, acted, observed, and decided again. It did what a human RM does in miniature, albeit with vastly less context and judgment. But the structure is the same.
Banking-grade considerations: streaming, timeouts, retries
The fifty-line example runs fine in a notebook. Running it in an operating environment requires a few more concerns.
Streaming: Relationship managers do not want to wait
three seconds for a full response. Use graph.astream() to
stream messages back. Pair it with FastAPI’s
StreamingResponse to produce a Server-Sent-Events stream to
the browser. The user sees tokens appear as the LLM generates them.
@app.post("/agent/stream")
async def agent_stream(req: AgentRequest):
async def event_stream():
async for chunk in agent.astream(
{"messages": [HumanMessage(content=req.message)]}
):
yield f"data: {json.dumps(serialise(chunk))}\n\n"
return StreamingResponse(event_stream(), media_type="text/event-stream")Timeouts: Every LLM invocation and every tool
invocation should have a timeout. Vertex AI and Bedrock clients accept a
request_timeout parameter. Tools should use
httpx or aiohttp with explicit timeouts on
every HTTP call. A single hanging tool can hold a graph invocation open
for minutes, which in a Cloud Run service means your instance is tied up
and other requests queue.
Retries: Use tenacity or the client-library built-in retry for transient failures, 429 rate limits, 503 service unavailable, connection resets. Retry at the call site, not at the graph level. The graph should see either a successful call or a final failure; it shouldn’t be involved in the retry loop.
Bounded loops: Always set a max-iterations guard.
The simplest is
graph.compile().with_config({"recursion_limit": 25}) which
caps total supersteps. A more domain-aware version: track a
tool_call_count field in state and force termination in the
router if it exceeds a threshold.
Security: bind exactly what you mean
A common early mistake: binding too many tools to the LLM. Every tool
you bind is a tool the LLM may decide to call. In banking, that has
security implications. A tool named transfer_funds must not
be bound to an LLM that handles unauthenticated or low-authorization
traffic. It’s not enough to implement authorization inside the tool; the
LLM should not even see it.
A useful rule: each distinct authorization level gets its own LLM
binding. the scenario has llm_with_read_tools,
llm_with_advisory_tools,
llm_with_action_tools. A customer-facing chatbot uses read;
an RM copilot uses advisory; a back-office workflow with human-approval
gates uses action. The same Python function is never wired as a tool in
more than one binding class.
The failure modes: three you will see in week one
Failure mode A: the tool that always fails. The LLM
calls get_customer, which raises because CRM is down.
ToolNode catches the exception and returns a ToolMessage with an error
string. The LLM sees the error, tries again, and the tool fails again,
and again, and again, until recursion limit stops the loop. The agent
appears unresponsive.
Mitigation: design tools to surface “not available right now, try later” explicitly in their error content. Tune the system prompt to instruct the model: “If a tool returns an error, apologise and do not retry.” Cap the per-tool retry count in state.
Failure mode B: the agent that never stops looking things up. The user asks a simple chat question; the LLM calls a tool anyway because the system prompt implies it should. Every response ping-pongs through a tool call. This is a prompt-engineering problem. Fix by making the system prompt explicit: “Only call tools when you need specific customer data or policy content. For conversational turns, reply directly.”
Failure mode C: prompt injection. User text or retrieved material may contain instructions that compete with application policy. Treat retrieved content as untrusted data, expose only least-privilege tools, validate typed actions outside the model and require independent authorisation for irreversible effects.
📈 observability for agents
Every agent run should emit:
- A trace showing every superstep, every LLM call, every tool call, with latencies and token counts.
- Logs with the trace ID, user ID, and a structured record of any tool invocations.
- Metrics for success rate, end-to-end latency (p50/p95/p99), tool-call-per-run distribution, and cost per run.
On GCP, we integrate OpenTelemetry with Cloud Trace. Each node emits a span; tool calls are child spans. On AWS, the same OpenTelemetry integration exports to X-Ray. The SDK abstractions (LangChain’s callbacks, LangSmith) plug into this for free, giving you both a structured trace and a human-readable trace viewer.
For banking, the trace is also the audit trail. When a regulator asks “why did the agent decline this?”, the answer is: here are the tool results the agent saw, here is the LLM’s response, here is the routing decision. All of that is in the trace. This is why observability is not an afterthought; it is a regulatory requirement.
A richer example: a RAG-aware rm copilot
Extend the agent with a retrieval tool. This adds an evidence route to the worked copilot.
@tool
async def retrieve_policy(query: str, k: int = 5) -> list[dict]:
"""Search the internal policy library and return top snippets."""
results = await matching_engine.aretrieve(query, k=k)
return [{"title": r.title, "snippet": r.snippet, "source_url": r.url} for r in results]
@tool
async def get_customer(customer_id: str) -> dict:
"""Return basic details for a customer by ID."""
return await crm_client.aget(customer_id)
@tool
async def get_exposures(customer_id: str) -> list[dict]:
"""Return list of credit exposures for a customer."""
return await exposure_service.aget(customer_id)
tools = [retrieve_policy, get_customer, get_exposures]With these three tools bound, the agent can answer:
- “What is our chargeback policy for digital-only customers?” (retrieve_policy)
- “What products does CUS-0012345678 hold?” (get_customer, get_exposures)
- “Does the vulnerable-customer policy require a cooling-off period for credit-card applications?” (retrieve_policy)
The same graph topology handles all of them. The LLM decides which tool (or tools) to call based on the question. If a question requires both customer data and policy context, the LLM may fan out two tool calls in parallel in a single superstep; ToolNode handles the parallelism automatically.
Thought experiment: what happens if you remove the system prompt?
[!tip] Exercise You remove the system prompt from the agent node. What happens?
The LLM still receives the user’s message and the tool schemas. It will still make reasonable decisions in many cases, modern LLMs are good at inferring “this is an assistant context.” But consistency drops. The model may give inconsistent tone (formal vs casual turn to turn). It may refuse to use tools because it doesn’t know what role it’s playing. It may cite policies inaccurately because it doesn’t know it should cite at all.
The system prompt is not decoration; it is the agent’s identity and standing orders. Removing it is like asking an employee to do a job without telling them what the job is.
Separate persistence from memory
The taxonomy of agent memory
Cognitive psychology distinguishes several kinds of memory. LangGraph gives us tools to implement each.
Short-term memory is the conversation buffer: the
messages in the current thread. In our Chapter 5 agent, this is the
messages field in state. It lasts for the duration of the
graph invocation; when the invocation ends, it’s gone.
Long-term memory persists across invocations. “Alice prefers morning calls. Her last mortgage query was on 2026-03-14. She is flagged Consumer Duty vulnerable because of a recent bereavement.” These facts are not in the current message thread; they’re in a separate store that the agent can query.
Episodic memory records specific events. “On 2026-02-12 at 14:23, Alice asked about overpaying her mortgage. The conversation resolved with an offer of a callback, which the RM scheduled for 2026-02-14.” Episodic memory is structured log of discrete interactions.
Semantic memory is general knowledge: “Merehaven Bank policy allows mortgage overpayments of up to 10% of the outstanding balance per year without early-repayment charges.” Semantic memory often lives in the knowledge base and is accessed via retrieval (Chapter 11).
Procedural memory is “how to do things”: the system prompts, tool descriptions, and system rules that shape the agent’s behaviour. These live in code and configuration, not in a runtime store.
LangGraph addresses short-term memory through the checkpointer (thread-level persistence) and long-term/episodic memory through the store abstraction (cross-thread persistence). We’ll cover both.
Short-term memory: the checkpointer
A checkpointer is an object that persists graph state
after each superstep. LangGraph ships with several implementations:
MemorySaver (in-memory, for development),
SqliteSaver (local file, for experiments),
PostgresSaver (production). You attach one at compile
time.
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string("postgresql://user:pass@host/db") as saver:
agent = builder.compile(checkpointer=saver)Once compiled with a checkpointer, every graph invocation can be associated with a thread_id. Invocations sharing the same thread_id share state; the checkpointer reads prior state at the start of each invocation and writes updated state at the end. The thread_id is the unit of conversation.
# First turn
await agent.ainvoke(
{"messages": [HumanMessage(content="Hi, I'm Alice")]},
config={"configurable": {"thread_id": "alice-session-001"}}
)
# Second turn, same thread
await agent.ainvoke(
{"messages": [HumanMessage(content="What did we talk about last time?")]},
config={"configurable": {"thread_id": "alice-session-001"}}
)The second invocation loads the state left by the first, appends the
new human message (via add_messages), runs the graph, and
saves the updated state. The agent sees the full message history and can
respond accordingly.
Thread IDs are opaque strings; use what makes sense for your domain.
For a logged-in web app,
thread_id = f"{user_id}:{conversation_id}" is natural. For
a voice channel, thread_id = f"call:{call_sid}". For a
multi-session RM engagement, it might be
thread_id = f"customer:{customer_id}" (all conversations
with a customer share a thread).
What gets checkpointed
Every superstep produces a checkpoint. A checkpoint contains:
- The full state after the superstep.
- Metadata, which node ran, what edges fired, the superstep number.
- A parent_id pointing at the previous checkpoint, forming a linked list (actually a tree, if branches occur).
The checkpointer writes all of this to its backing store. On the next invocation with the same thread_id, the checkpointer reads the most recent checkpoint and restores state.
For PostgresSaver, the default schema is a few tables: checkpoints, writes, and pending writes. These scale to millions of threads if indexed correctly. The migration script ships with LangGraph.
Thread-level persistence in an operating environment: lessons learned
Lesson 3: thread hygiene matters. Do not reuse thread_ids across semantically different conversations. If a customer starts a new topic, start a new thread (or explicitly clear prior state). Otherwise, old context leaks into new conversations and causes surreal model behaviour. One rule of thumb: new subject → new thread.
Viewing thread history: time-travel and debugging
LangGraph exposes thread history via
graph.get_state_history(config). You can iterate over
checkpoints, inspect the state at any point, and even rewind.
config = {"configurable": {"thread_id": "alice-session-001"}}
history = [state async for state in agent.aget_state_history(config)]
for checkpoint in history:
print(checkpoint.metadata.get("step"), checkpoint.values.get("messages", [])[-1])This is astonishingly useful in an operating environment debugging. A customer reports “the assistant said something weird ten minutes ago.” You fetch the thread by ID, inspect each checkpoint, and can see exactly what state looked like at each step. For regulatory audit, this is a ready-made trail.
You can also time-travel: update the state at a specific checkpoint and re-run from that point, producing a divergent branch. This is the foundation for “what if I had answered differently?” patterns, which Chapter 9 (human-in-the-loop) uses heavily.
# Rewind to checkpoint X, modify state, re-run
await agent.aupdate_state(
config={"configurable": {"thread_id": "alice-session-001", "checkpoint_id": "X"}},
values={"draft_memo": "Revised version..."}
)
await agent.ainvoke(None, config=...) # Continue from the updated checkpointLong-term memory: the store
Checkpointer persistence is thread-scoped. Long-term memory is cross-thread: it follows the user, not the conversation. LangGraph provides the store abstraction for this.
from langgraph.store.postgres import PostgresStore
store = PostgresStore(conn_string="postgresql://...")
# Put a memory
store.put(
namespace=("memories", "alice-0012345678"),
key="preferred_contact_time",
value={"value": "morning", "reason": "stated during 2026-03-14 call"}
)
# Retrieve all memories in a namespace
memories = store.search(namespace=("memories", "alice-0012345678"))The namespace is a tuple that organises memories;
(memories, user_id) is a common pattern. key
is a string identifier; value is a JSON-serialisable
payload. You can also attach vector embeddings for semantic search over
memories.
In graph nodes, the store is accessed via a config parameter:
def agent_node(state: AgentState, config, *, store) -> dict:
user_id = config["configurable"]["user_id"]
memories = store.search(("memories", user_id), limit=10)
context = format_memories(memories)
system = SystemMessage(content=f"{BASE_PROMPT}\n\nRelevant memories:\n{context}")
response = llm_with_tools.invoke([system] + state["messages"])
return {"messages": [response]}The node injects long-term memories into the system prompt at every invocation. Combined with short-term memory via the checkpointer, the agent now has both “what we said in this conversation” and “what I know about this person from across conversations.”
Writing memories: when and how
An underappreciated question: when does an agent write to long-term memory? Three common patterns.
Explicit memory node: a dedicated node in the graph that, after the agent node, extracts facts from the conversation and writes them. “The customer said their preferred contact time is mornings, store that as preferred_contact_time.” This is reliable because the write is deterministic.
Tool-based memory: a remember tool that
the LLM can call. This gives the LLM agency over what to remember. More
flexible but harder to control; the LLM may write spurious facts or omit
important ones.
Background memory: a separate, asynchronous process consumes conversation transcripts and extracts facts in batch. This decouples memory extraction from live response, reducing latency. In the Merehaven worked scenario, the scenario uses a combination: explicit memory nodes for known-important facts (preferences, flags), plus a nightly batch process that runs on fresh transcripts to extract additional semantic memories.
An important banking consideration: never write unverified
claims to long-term memory. If a customer says “I’m a doctor,”
do not write occupation=doctor unless you have
corroboration. The agent is not a verification service; claims should be
stored as claims, not facts. At most, write
customer_claimed(occupation, "doctor", at=2026-03-14). The
distinction between fact and claim is the difference between a useful
memory store and a liability.
Semantic memory search via embeddings
Long-term memory scales naturally via embeddings. Every memory is embedded; retrieval is by vector similarity. This lets the agent ask questions like “what did this customer tell us about their financial goals?” and get relevant memories ranked by semantic match.
store = PostgresStore(
conn_string="...",
index={
"dims": 768,
"embed": vertex_ai_embedding_model,
}
)
# Store a memory with automatic embedding
store.put(
namespace=("memories", customer_id),
key="goal_2026",
value={"text": "Planning to buy a home in the next two years, saving for a deposit"}
)
# Search semantically
results = store.search(
namespace=("memories", customer_id),
query="financial goals",
limit=5
)This pattern is directly applicable to BFSI. For a wealth RM, pulling the “what are this client’s known goals?” memories into the system prompt before every conversation step means the agent speaks with context no human could hold in their head across two hundred clients.
A richer graph with memory
Let’s update the Chapter 5 agent with both short-term (checkpointer) and long-term (store) memory.
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
def retrieve_memories_node(state, config, *, store):
user_id = config["configurable"]["user_id"]
memories = store.search(("memories", user_id), query=last_user_text(state), limit=5)
context = "\n".join(f"- {m.value['text']}" for m in memories)
# inject into state as a system message? Or keep separate?
return {"memory_context": context}
def agent_node(state, config, *, store):
system = SystemMessage(
content=f"{BASE_PROMPT}\n\nLong-term memories about this user:\n{state.get('memory_context', '(none)')}"
)
response = llm_with_tools.invoke([system] + state["messages"])
return {"messages": [response]}
def write_memory_node(state, config, *, store):
# Extract new facts via a small structured-output LLM call
user_id = config["configurable"]["user_id"]
facts = extract_facts(state["messages"][-4:])
for fact in facts:
store.put(
namespace=("memories", user_id),
key=fact.key,
value={"text": fact.text, "source_turn": len(state["messages"])}
)
return {}
builder = StateGraph(AgentState)
builder.add_node("retrieve_memories", retrieve_memories_node)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_node("write_memory", write_memory_node)
builder.set_entry_point("retrieve_memories")
builder.add_edge("retrieve_memories", "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")
# After agent produces a final response, write memories before END
builder.add_edge("agent", "write_memory") # conditional: only when no tool calls
builder.add_edge("write_memory", END)
agent = builder.compile(checkpointer=saver, store=store)This graph runs two additional nodes around the core loop:
retrieve_memories at the start, write_memory
at the end. Both interact with the store. The checkpointer handles the
messages field automatically.
Figure 6.1: An agent with both short-term and long-term memory. The retrieve_memories node injects relevant long-term memories at the start; the write_memory node extracts new facts at the end.
Regulated banking memory patterns: a specific design
For the Merehaven worked scenario, a wealth-management RM copilot, the scenario maintains five kinds of memory:
- Identity memories: name, preferred name, contact details, preferred contact time. Small, stable, written rarely.
- Goal memories: stated financial goals with dates and amounts. Written when the customer shares a goal; updated if they revise.
- Constraint memories: “Do not recommend crypto products,” “Strongly risk-averse,” etc. Written at onboarding or when stated.
- Event memories: significant life events that affect advice, bereavement, redundancy, relocation. Written (with consent) when disclosed.
- Conversation-summary memories: a distilled summary of each significant conversation, written at end of session. Keeps context without requiring the full transcript.
We namespace by customer ID:
("memories", customer_id, kind). Each namespace has its own
retention policy. Identity memories live indefinitely (with deletion
support for GDPR); conversation-summary memories expire after two years;
goal and constraint memories are reviewed annually during the
suitability review.
Memory failure modes
Failure A: memory drift. Over time, the store accumulates contradictory memories (“prefers morning” from 2025 and “prefers evening” from 2026). Mitigations: timestamp every memory and, at retrieval, prefer the most recent; implement a consolidation job that merges or supersedes older memories; design the extraction prompt to recognise updates.
Failure B: memory pollution. An LLM hallucinates a fact and writes it to memory. Later retrievals return the hallucination. Mitigations: never let the LLM write directly; route writes through a validation layer; keep an audit trail of every write; offer a “forget” path for users.
Failure C: memory overreach. The agent retrieves a memory that’s embarrassingly relevant (“you told us about your divorce”) and surfaces it in an off-topic conversation. Mitigations: tag memories with sensitivity levels; only retrieve sensitive memories when the current turn explicitly warrants it; err on the side of not bringing things up.
Route by evidence and limit
The three kinds of routing
Think of routing in three tiers of sophistication.
Static routing is the fixed edge:
add_edge(A, B). A to B, always. Use this for sequencing
that never varies.
Conditional routing is the
add_conditional_edges pattern. A routing function examines
state and returns the name of the next node (or a list, for fan-out).
The router is a pure function of state.
Command-based routing is a newer LangGraph pattern
where a node itself returns a Command object specifying the
next node. This is the cleanest way to express “the agent decided what
to do next”, the decision lives in the node (close to the reasoning)
rather than in a separate router function.
All three compose. A production graph uses static routing for deterministic sequences, conditional routing for structural decisions (classification, validation), and command-based routing for LLM-driven handoffs.
Conditional routing: the disciplined pattern
The conditional edge pattern shines when the decision is based on structural state, not LLM judgement. For example: “if the credit score is below 0.5, escalate to human; otherwise finalise.”
def route_after_score(state: CreditState) -> str:
if state["credit_score"] < 0.5:
return "escalate"
if state["requires_committee"]:
return "committee_review"
return "finalise"
builder.add_conditional_edges(
"score",
route_after_score,
{
"escalate": "human_review",
"committee_review": "committee_review",
"finalise": "finalise",
}
)The router is deterministic, pure, and testable. I can write unit
tests for route_after_score that cover every branch in
seconds, without touching an LLM. in an operating environment, when a
regulator asks “why was this case escalated?”, the answer is “because
credit_score < 0.5 in the router.” Clear and
auditable.
A useful rule: any routing decision that can be expressed as structural logic should be expressed that way, not delegated to an LLM. LLM-based routing is a hammer; sometimes you need a scalpel.
LLM-based routing: classification as a node
When routing does require LLM judgement, “is this question a general enquiry, a complaint, or a compliance issue?”, express it as a classification node whose output feeds a structural router.
class Intent(BaseModel):
kind: Literal["general", "complaint", "compliance"]
confidence: float
def classify_intent_node(state, *, llm):
prompt = f"Classify the user's intent: {last_user(state).content}"
intent: Intent = llm.with_structured_output(Intent).invoke(prompt)
return {"intent": intent}
def route_by_intent(state) -> str:
if state["intent"].confidence < 0.6:
return "clarify"
return state["intent"].kind
builder.add_node("classify_intent", classify_intent_node)
builder.add_conditional_edges("classify_intent", route_by_intent, {...})Two advantages over embedding the classification in the router directly. First, the classification result is now in state, where observability and audit can see it. Second, the router is still pure: it reads state and returns a name. This separation of concerns, LLM produces structured output; router dispatches on it, is the cleanest pattern for LLM-assisted routing.
Command-based routing: the modern idiom
LangGraph 0.2+ supports Command, a construct that lets a
node return both state updates and a routing directive in one
object.
from langgraph.types import Command
def agent_node(state, config):
response = llm_with_tools.invoke([...])
if response.tool_calls:
return Command(
update={"messages": [response]},
goto="tools"
)
return Command(
update={"messages": [response]},
goto="__end__"
)This collapses the agent node and the should_continue
router into a single place. The decision “should this edition calls
tools or finish?” is made where the reasoning happens, not in a separate
function looking at artifacts of the reasoning. For multi-agent
architectures (Chapter 12), Command is essential: one agent finishing
decides which other agent should pick up, and that decision is naturally
expressed in the finishing agent’s node.
Command also supports cross-subgraph navigation via
graph=Command.PARENT, allowing a subgraph to hand control
back to its parent explicitly. This is the foundation for hierarchical
agent teams.
Dynamic routing: fan-out and fan-in
Sometimes a router doesn’t pick one next node; it picks several. Returning a list of names causes fan-out: all named nodes run in parallel in the next superstep.
def fan_out_scoring(state) -> list[str]:
"""Run PD, LGD, and EAD scoring models in parallel."""
return ["score_pd", "score_lgd", "score_ead"]All three scoring nodes run concurrently. When they complete, their
state updates merge, and this is where reducers matter. If each writes a
distinct field (pd, lgd, ead), there’s no conflict. If they all write to
scores as a list, you need
Annotated[List, operator.add] so that the three updates
concatenate instead of overwriting each other.
Fan-in happens implicitly: once all parallel branches complete their current superstep, LangGraph moves to the next superstep, which sees all updates merged. A combining node then reads the multi-value state and produces a single decision.
Loop routing: the bounded cycle
Many agent patterns involve controlled loops: a ReAct agent loops until done; a reflection pattern loops until a critic approves. Loops are expressed by edges that go “back” in the graph, with a router enforcing termination.
def should_continue_reflection(state) -> str:
if state["iteration"] >= 3:
return "finalise"
if state["critic_approved"]:
return "finalise"
return "reviser"
builder.add_edge("generator", "critic")
builder.add_conditional_edges("critic", should_continue_reflection, {
"reviser": "reviser",
"finalise": "finalise",
})
builder.add_edge("reviser", "critic")Generator → Critic → Reviser → Critic → … until the critic approves or the iteration cap is reached. Every loop in a LangGraph should have both a content-driven exit (the critic’s decision) and a count-driven exit (the iteration cap). The content exit handles the happy path; the count exit prevents pathological looping.
Routing in a regulated banking context
Consider a worked routing pattern from the Merehaven Bank RM copilot. The RM asks a question. The graph classifies the question, fetches appropriate context, routes to a specialised answerer, and writes an audit log. The routing diagram:
Figure 7.1: A BFSI routing pattern. One classification, three specialised paths, one compliance gate, one audit. Each diamond is a routing decision.
The classifier does one job: decide the kind of request. The compliance gate does one job: verify the proposed answer is compliant (no restricted advice, no unauthorised customer data disclosure). Three specialist answerers each own their domain. The graph has three routing decisions; each is small and testable.
Make reasoning observable without worshipping it
Why ReAct still matters
In October 2022, Shunyu Yao and collaborators at Princeton published a paper with a wordy title: ReAct: Synergizing Reasoning and Acting in Language Models. The pattern it described, interleave reasoning traces (“Thought: the customer asked about overpayments; I should look up the policy”) with actions (“Action: search_policies(‘overpayment’)”) and observations (“Observation: …”), was simple enough that many reviewers thought it was obvious. It was not obvious to the engineers actually shipping agents, and the paper became one of the most cited in the agent literature.
Why? Because ReAct did three things at once. It made reasoning explicit (not a black box inside the model). It made actions grounded in reasoning (the model had to justify each tool call). And it produced traces that humans could read and debug. In a banking context, all three of those properties are not nice-to-haves; they are compliance requirements.
The structure of ReAct
A ReAct turn has three parts: Thought, Action, Observation. The agent produces a thought describing its reasoning, then an action (tool call with arguments), then reads back an observation (tool result). The loop continues until the agent’s thought concludes no more action is needed and produces an Answer.
In explicit form:
Thought: I need to find the customer's recent exposures before I can draft a memo.
Action: get_exposures(customer_id="CUS-0012345678")
Observation: [{product: "mortgage", balance: 250000, ...}, ...]
Thought: I have the exposures. Now I need the latest credit policies to reference.
Action: search_policies(query="commercial credit memo template")
Observation: [{title: "...", snippet: "..."}]
Thought: I have enough context. Drafting the memo now.
Answer: <memo text>This is the literal format in Yao’s paper. In modern implementations, we often separate the Thought into its own message, emit the Action as a tool call, and get the Observation back as a tool message. The structure is preserved; only the format differs.
Building a ReAct agent in LangGraph
Here’s a ReAct-style agent that emits explicit thoughts:
from langchain_core.messages import AIMessage, ToolMessage, SystemMessage
REACT_SYSTEM = """You are a credit memo drafting assistant.
For each step, emit a "Thought:" section explaining your reasoning, then either
call a tool or produce a final "Answer:" block. Always justify tool calls."""
class ReActState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
iteration: int
def agent_node(state: ReActState):
response = llm_with_tools.invoke([
SystemMessage(content=REACT_SYSTEM),
*state["messages"]
])
return {
"messages": [response],
"iteration": state.get("iteration", 0) + 1
}
def should_continue(state: ReActState) -> str:
last = state["messages"][-1]
if state["iteration"] >= 8:
return "__end__" # safety bound
if hasattr(last, "tool_calls") and last.tool_calls:
return "tools"
return "__end__"The iteration counter is the bound. Without it, a confused model can loop indefinitely, racking up cost. Eight iterations is a reasonable default for most BFSI tasks; complex multi-step drafting may warrant twelve.
ReAct with a dedicated thinking phase
A richer variant separates “thinking” from “acting” as distinct nodes. The thinker node produces reasoning; the actor node converts the thought into a tool call.
def thinker_node(state):
prompt = "Given the conversation so far, what should we do next? Think step by step."
thought = thinking_llm.invoke([*state["messages"], HumanMessage(content=prompt)])
return {"messages": [AIMessage(content=f"Thought: {thought.content}")]}
def actor_node(state):
prompt = "Based on the thought above, produce a tool call or a final answer."
response = acting_llm_with_tools.invoke([*state["messages"], HumanMessage(content=prompt)])
return {"messages": [response]}
builder = StateGraph(ReActState)
builder.add_node("thinker", thinker_node)
builder.add_node("actor", actor_node)
builder.add_node("tools", tool_node)
builder.set_entry_point("thinker")
builder.add_edge("thinker", "actor")
builder.add_conditional_edges("actor", should_continue)
builder.add_edge("tools", "thinker")Why separate? Because the thinker can be a different model. For routine decisions, use a cheaper/faster model (a pinned model or Claude Haiku) for thinking; for actions, use the strong model. Or the opposite: use the strong model for deliberation and a smaller model for the well-constrained tool-call generation.
Figure 8.1: ReAct with separated thinking and acting phases, allowing different models per phase.
Self-reflection: ReAct with a critic
The pure ReAct loop trusts the agent to know when it’s done. Real agents sometimes finish wrong. Reflection adds a critic node that evaluates the proposed answer and either approves it or sends the agent back for another iteration.
class Review(BaseModel):
approved: bool
reason: str
suggested_fixes: list[str]
def critic_node(state):
proposed = last_answer(state)
review: Review = critic_llm.with_structured_output(Review).invoke(
f"Review this draft memo. Check: (1) all exposures covered; (2) policy citations correct; "
f"(3) risk rating justified. Draft:\n{proposed}"
)
return {"review": review, "messages": [AIMessage(content=f"Critic: {review.reason}")]}
def route_after_critic(state) -> str:
if state["review"].approved:
return "finalise"
if state["iteration"] >= 3:
return "escalate"
return "thinker" # go back and reviseSelf-reflection is capable for structured outputs, credit memos, compliance narratives, customer correspondence, where a critic with a clear rubric can catch errors the generator missed. It is less useful for fast conversational turns, where the latency cost of a critic loop exceeds the quality benefit.
Worked decision rule: reflection for any document that will be saved to a customer file or submitted externally; no reflection for conversational responses shown in-line.
The failure modes
Over-reasoning. The model emits ten thoughts before every action, tripling cost for marginal quality improvement. Fix: tune the system prompt to request concise thoughts (“One sentence of reasoning, then act”) and enforce the iteration cap.
Thought-drift. The model’s thoughts become
disconnected from the actions. It thinks “I should look up policies” and
then calls get_customer. Fix: use function calling with
strict schemas (the thought is advisory; the action is a structured tool
call that the model cannot invent); add a validation step that rejects a
tool call inconsistent with the stated thought.
Loop collapse. The model’s thoughts converge: “I already tried X, let me try X again” on each iteration. Fix: include the iteration count in the system prompt (“This is iteration 5; try something different”); consider a forced-termination path after N iterations that composes whatever partial answer exists.
A regulated banking ReAct in practice: autonomous mortgage triage
At a UK lender, we built a ReAct agent that triages mortgage applications. The agent sees the application, thinks about what checks are needed, calls tools to check affordability, credit bureau, employment, property valuation, identity; and after each observation, thinks about whether enough evidence exists to recommend a decision.
The traces are beautiful. A typical run reads:
Thought: Standard residential application, £320k on £400k, LTV 80%, first-time buyer.
I should check affordability first.
Action: check_affordability(income=65000, dependents=1, ...)
Observation: {dti: 0.28, affordable: true, buffer: 800/month}
Thought: Affordability is fine. Check credit history.
Action: credit_bureau_lookup(customer_id="CUS-...")
Observation: {score: 780, flags: []}
Thought: Strong credit. Employment verification next.
Action: verify_employment(company="Example Ltd", role="Senior Engineer")
Observation: {confirmed: true, years: 5}
Thought: Property valuation.
Action: property_valuation(address="...")
Observation: {value: 412000, confidence: high}
Thought: All checks pass. LTV 80%, DTI 28%, stable employment, good credit,
property valued slightly higher than purchase. Recommending conditional approval
pending solicitor's report.
Answer: Conditional approval. Reasons: [...]Every thought becomes an audit record. Every action is a discrete tool call with arguments logged. Every observation is a tool result. A human underwriter reviewing the decision can read the trace in under a minute and either agree, disagree, or ask clarifying questions that the agent can answer by re-running a specific sub-step.
Part III: Human authority and executable plans
Consequential action needs a clean break between proposal and permission. Human review is a control surface only when it receives evidence, time and a real veto.
Interrupt before consequence
Why human review is not optional in regulated banking
Consider three regulatory regimes that apply to a typical LangGraph agent In the Merehaven worked scenario,.
The common thread across all four regimes is the same. Autonomous systems in financial services need human intervention points, and those points must be designed, not accidental.
The interrupt primitive
See this in code. The simplest HITL pattern in LangGraph uses
interrupt_before or interrupt_after on
specific nodes.
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from typing import TypedDict, List, Optional, Literal
class MemoState(TypedDict):
customer_id: str
draft_memo: Optional[str]
reviewer_decision: Optional[Literal["approve", "reject", "edit"]]
reviewer_edits: Optional[str]
final_memo: Optional[str]
def draft_memo(state: MemoState) -> dict:
# Call LLM to produce draft
draft = llm.invoke([...])
return {"draft_memo": draft.content}
def apply_reviewer_decision(state: MemoState) -> dict:
decision = state["reviewer_decision"]
if decision == "approve":
return {"final_memo": state["draft_memo"]}
elif decision == "edit":
return {"final_memo": state["reviewer_edits"]}
else: # reject
return {"final_memo": None}
builder = StateGraph(MemoState)
builder.add_node("draft", draft_memo)
builder.add_node("apply_decision", apply_reviewer_decision)
builder.set_entry_point("draft")
builder.add_edge("draft", "apply_decision")
builder.add_edge("apply_decision", END)
checkpointer = PostgresSaver.from_conn_string(POSTGRES_URL)
graph = builder.compile(
checkpointer=checkpointer,
interrupt_before=["apply_decision"]
)The control point is the
interrupt_before=["apply_decision"] argument. When
execution reaches the apply_decision node, LangGraph
pauses. The state is persisted to the checkpointer. The API call that
invoked the graph returns with a status of “interrupted” and the current
state.
Some time later (a minute, an hour, or next morning when the reviewer
returns from lunch), the UI updates the state with the reviewer’s
decision and calls graph.invoke(None, config) again. The
graph resumes from apply_decision, runs that node with the
newly-populated state, and proceeds to END.
The important property is that this is durable and idempotent. If the agent container crashes during the wait, the state is safe in Postgres. If the reviewer’s session times out and they log back in, they see exactly the same draft. If the same update arrives twice, it does not corrupt the state. These are the properties a bank requires.
Worked example: the approval interrupt in action
Trace the full lifecycle of one credit memo through a HITL graph.
Step 2. The return value.
print(result)
# {"customer_id": "CUS-0012345678", "draft_memo": "..."}
# But the graph's state has more: it's paused at "apply_decision"The returned dict is the state at the moment of interrupt. The client SDK returns a “paused” status. The UI shows the draft memo to the reviewer with three buttons: Approve, Edit, Reject.
Step 3. The reviewer reads the draft. They take nine minutes. In those nine minutes, they check the customer’s file, confirm the financial metrics, and read the policy justification the agent cited. They decide to approve with one edit (a clarification on the security).
Step 4. The reviewer submits their decision at 14:13. The UI calls:
graph.update_state(
config,
{
"reviewer_decision": "edit",
"reviewer_edits": "<edited memo text>"
}
)
result = graph.invoke(None, config=config)The update_state call writes the reviewer’s decision
into the checkpointed state. The invoke(None, ...) call
resumes the graph from where it paused. The apply_decision
node runs with the updated state, writes final_memo, and
proceeds to END.
Step 5. The final result. The client gets the final
memo back. The audit log now contains: the original input, the draft,
the reviewer’s identity (captured in update_state via
LangGraph’s config metadata), the reviewer’s decision and edits, and the
final memo. Every step is queryable via
graph.get_state_history(config).
What is worth dwelling on here is how much governance this tiny bit of code buys you. The durability across restarts. The reviewer’s decision as a first-class state update. The full history of the run. Without interrupt, you would be building this yourself with Redis queues, custom REST endpoints, ad-hoc persistence, and flaky reconnection logic. LangGraph makes it a one-liner, and it works in an operating environment.
Dynamic interrupts: the interrupt function
Static interrupts (configured at graph compile time) are capable but coarse. You interrupt before a specific node every time. In practice, you often want to interrupt conditionally: only interrupt if the proposed action is high-risk, only interrupt if confidence is below a threshold, only interrupt if the customer is in a vulnerable category.
Dynamic interruption can pause from inside a node when case state crosses a risk boundary. Confirm the exact interrupt and resume semantics against the pinned LangGraph version, then test persistence, expiration and idempotent replay.
from langgraph.types import interrupt, Command
def risk_gate(state: MemoState) -> dict:
draft = state["draft_memo"]
risk_score = compute_risk_score(draft, state)
if risk_score > 0.7:
# Dynamic interrupt: pause here, present the draft for review
human_decision = interrupt({
"draft": draft,
"risk_score": risk_score,
"risk_factors": identify_risk_factors(draft, state),
"question": "This memo contains higher-risk elements. Please review and decide."
})
# Resume with the human's response available here
return {
"reviewer_decision": human_decision["decision"],
"reviewer_edits": human_decision.get("edits"),
}
else:
# Low risk: auto-approve
return {
"reviewer_decision": "approve",
}The interrupt function throws a special exception that
LangGraph catches. The value passed to interrupt becomes
the payload returned to the client. When the client is ready to resume,
it calls graph.invoke(Command(resume=value), config), where
value is the human’s decision. That value becomes the
return value of interrupt inside the node, and execution
continues.
This pattern lets you encode your risk policy in code rather than in graph topology. The risk gate is a normal Python function: it can read any fields of state, call any classifier, apply any heuristic. If risk is low, no human is bothered; the path is fully automated. If risk is high, the human is engaged with full context.
Deeper mechanism: resume patterns and idempotency
The resume pattern has three important subtleties that trip up teams new to HITL.
First, the resume value replaces the interrupt
call’s return value. It does not re-run the node from scratch.
If your node does work before calling interrupt, that work
is done before the pause and is part of the state. When you resume,
execution picks up inside the node, right after
interrupt. This is efficient but it has implications.
Second, state changes made after the interrupt depend on the state at resume time. If you interrupted thirty minutes ago, and in that time some other process updated a piece of referenced data (say, the customer’s credit bureau score refreshed), your resumed node will see the fresh value if it reads it. This can be a feature or a bug. Make it a feature: re-read any time-sensitive data at the point of use.
Third, resume is idempotent in the sense that the same resume payload produces the same outcome. If the client accidentally sends the same resume twice (network retry, user double-clicks), LangGraph will resume once from the checkpoint and discard the duplicate. This is critical for HTTP-based UIs where retries are common.
The worked scenario handles this by having the resume node re-read the customer’s financial state at the moment of resume, not use the snapshot from the time of draft. If the fresh state was materially different from the stale state, the agent would produce a “your situation has changed; here is the updated assessment” draft and loop back for another underwriter review. This felt like extra work but it was the right thing to do: it ensured decisions were always made on current data, not stale snapshots.
The failure mode: the rubber-stamp reviewer
The fix is not “tell them to read more carefully.” The fix is structural. Four mitigations, in order of effectiveness.
Mitigation 1: reduce the review burden. If the reviewer sees 200 drafts per day, they will rubber-stamp. If they see 40, they will actually read. The way to reduce the number is to raise the bar for what the agent auto-produces. Tighten the confidence threshold. Add more risk checks. Accept a lower automation rate in exchange for more meaningful reviews on the ones that need them.
Mitigation 2: make the review tool demanding. Do not show just the draft and Approve/Reject buttons. Show the reviewer’s attention a structured checklist: “Does the memo cite the correct policy version? Is the rating justified? Are the exposures complete?” Each item is a checkbox the reviewer must actively tick. A checkbox they skip turns red. The friction of the UI forces engagement.
Mitigation 3: random audit probes. Inject synthetic cases that are designed to look plausible but contain a deliberate error. Track the reviewer’s detection rate. A reviewer whose probe-detection rate drops below a threshold gets retraining. This is not a gotcha; it is quality assurance, and every reviewer knows it is happening because they are told it is.
The combination of these four mitigations is what turns HITL from theatre into actual risk mitigation. Without them, HITL is a fig leaf; with them, it is meaningful governance.
Escalation patterns: when the human pauses too long
A second failure mode is the abandoned review. A reviewer starts a review, gets interrupted, never comes back. Meanwhile, the customer’s request sits unresolved. After a day, it is late. After three days, it breaches the bank’s service level. After a week, a complaint arrives. The agent did its job; the human broke the loop.
The solution is an escalation ladder built into the graph. The interrupt is time-bounded. If the review is not complete within a defined SLA, a timer fires and the interrupt escalates.
import datetime as dt
from langgraph.types import interrupt, Command
def draft_review_gate(state: MemoState) -> dict:
review_request_time = dt.datetime.utcnow().isoformat()
decision = interrupt({
"draft": state["draft_memo"],
"review_request_time": review_request_time,
"sla_hours": 4,
"next_escalation_at": (dt.datetime.utcnow() + dt.timedelta(hours=4)).isoformat(),
})
return {
"reviewer_decision": decision["decision"],
"reviewer_id": decision["reviewer_id"],
"review_completed_at": decision["completed_at"],
}Externally, a Cloud Scheduler job (on GCP) or an EventBridge rule (on AWS) runs every fifteen minutes. It queries for paused runs that have exceeded their SLA. For each, it triggers an escalation: a notification to the reviewer’s line manager, a reassignment to a back-up reviewer, or an automatic fallback (decline with reasons, offer a callback).
The escalation is itself an update to the graph’s state via
update_state, followed by a resume with the escalation
decision. The graph never silently stalls; every path ends
deterministically within an SLA-bounded window.
Operating boundary: the reviewer queue
In the Merehaven worked scenario, the scenario runs a shared reviewer queue for agent drafts. It is a simple Firestore collection, with documents representing pending reviews. Each document has the thread ID, the graph name, the agent’s proposed action, the risk signals, and the SLA.
The reviewer’s UI pulls from this queue in priority order: SLA-urgent
first, then risk-sorted. A reviewer clicks “Claim next item,” the item
gets locked to them (with an expiring lock), and they see the context.
When they submit their decision, the UI calls the agent service’s
/resume endpoint, which calls
graph.update_state and
graph.invoke(Command(resume=...)), and the graph
continues.
We started with one queue per agent and found that reviewers developed “tunnel vision” for their specific agent’s idiosyncrasies. We now pool reviewers across related agents (credit memo, limit renewal, facility restructure) so they see a variety of contexts and stay alert. The cost is a small amount of context-switching overhead; the benefit is much better detection of unusual drafts.
A second production observation. The SLA you set in code matters less than the SLA your reviewers can actually meet. If you set a 4-hour SLA and your reviewers are a team of six covering London business hours, a draft produced at 5pm will almost certainly miss the SLA. Design for the real staffing; use escalation to absorb the mismatches; do not pretend your reviewers are available 24/7 unless they are.
A thought experiment: consumer duty and a declined credit card
You work on a credit card application agent. The agent decides applications in under ten seconds, which is a competitive necessity for the market. The agent produces a decision (approve with limit, approve with lower limit, decline) and, if declining, a reason. The reason is shown to the customer.
Where should HITL interrupts fit in this agent?
Not on every decline. You would flood your queue. Not on every approve. Same problem. Candidate interrupts, in order of risk: declines where the customer has been declined more than three times in the last year (risk of vulnerable customer being repeatedly harmed); declines where the model’s confidence is low (risk of wrong decline); approves where the proposed limit is significantly above the customer’s income (risk of affordability concerns); declines where the reason would not pass a Consumer Duty test of comprehensibility.
That last one is interesting because it requires the agent to reason about the quality of its own reason. You can build a second LLM call that evaluates the decline reason against a Consumer Duty rubric and, if the reason fails, interrupts for a human to produce a better explanation. This is an LLM-grading-LLM pattern applied to governance.
Sketch the state schema for this agent on paper before reading further. What fields do you need? What decisions can trigger an interrupt? What information does a reviewer need at the moment of interrupt?
Here is a reasonable sketch.
class CreditCardDecisionState(TypedDict):
# Inputs
customer_id: str
requested_limit: float
application_payload: dict
# Working state
messages: Annotated[List, add_messages]
credit_bureau: Optional[dict]
affordability: Optional[dict]
vulnerability_flags: List[str]
# Outputs
decision: Optional[Literal["approve", "approve_lower", "decline"]]
proposed_limit: Optional[float]
decline_reason: Optional[str]
decline_reason_comprehensibility: Optional[float]
# HITL
requires_human: bool
human_reason: Optional[str]
reviewer_decision: Optional[dict]
# Governance
consumer_duty_checks: dict
audit: AuditTrailThe key insight is that the graph needs to be able to answer the
regulator’s question “how did you make sure this decline was
Consumer-Duty compliant?” The answer is visible in
decline_reason_comprehensibility,
consumer_duty_checks, and the reviewer’s decision. All
three are first-class state fields.
Human review anti-patterns
Anti-pattern 1: human as fallback for undertrained model. A team deploys an agent that is only 75% accurate and uses HITL to catch the other 25%. The reviewers buckle under the load. The right approach: fix the model’s accuracy first; use HITL for the residual high-risk cases. HITL is not a substitute for training.
Anti-pattern 2: the decision is already baked in. The UI shows the human only the agent’s proposed decision, with Approve/Reject buttons. The human’s attention gets anchored on the proposed decision. The right approach: show the human the inputs, the data, and the reasoning, and let them form their own view before seeing the agent’s proposal. Or, even better, have the human decide first on a small sample, then show the agent’s proposal for alignment training.
Anti-pattern 3: no feedback loop. The human’s decisions do not flow back into model retraining or evaluation. The agent keeps making the same mistakes; the humans keep catching them. The right approach: every human override is a labelled example. Feed them into the eval set at minimum; into the fine-tuning set if volume allows.
Anti-pattern 4: overconfident HITL SLAs. The team promises four-hour turnaround and staffs for six. Most drafts miss SLA. The right approach: either staff realistically, or set SLA realistically, or segment drafts by urgency and staff for the urgent ones. Do not set an aspirational SLA and hope.
Connecting to earlier chapters: state as the ledger
Go back to Chapter 3’s analogy: state as the ledger. HITL depends
completely on this ledger being trustworthy. When a reviewer updates
state with a decision, that update is a journal entry: timestamped,
attributable, auditable. The checkpointer is the general ledger. The
full history of state updates (available via
graph.get_state_history(config)) is the complete audit
trail.
This is why TypedDict/Pydantic state design matters so much for HITL
agents. A poorly designed state (free-form strings, inconsistent naming,
optional fields everywhere) makes it hard to know what a reviewer’s
decision actually updated. A well-designed state (typed, schematised,
with explicit reviewer_* fields) makes reviewer actions
legible both to humans and to auditors.
Reviewers’ actions are also what feed the evaluation flywheel we will cover in Chapter 13. Every override is a labelled example of “the agent said X, the reviewer changed it to Y, for reason Z.” These are gold-label training and evaluation data. The agent does not get smarter by itself; it gets smarter because humans correct it and those corrections loop back.
Turn plans into revisable contracts
Why ReAct alone is not enough
ReAct, which we built in Chapter 8, is a tight loop: think, act, observe, repeat. It is brilliant for short tasks where each step depends on the previous one. “Look up this customer’s exposures, then draft a memo about them” is two steps; ReAct handles it elegantly.
But ReAct has three structural weaknesses for longer tasks.
First, it is myopic. At each step, the agent thinks only about the immediate next action. It does not maintain a view of the whole task. For a fifteen-step task, this means the agent may take redundant steps, miss necessary ones, or get distracted by intermediate findings that should have been bracketed.
Second, it is expensive. Each step is an LLM call, and each call prompts the model with the full message history including all prior thoughts and observations. Message history grows quadratically in practice, because each call includes all prior ones. For a fifteen-step task, the last call may have 8,000 tokens of context, most of which is irrelevant to the current step.
Third, it is non-reviewable until complete. A human reviewer watching a ReAct agent has no way to see what the agent intends to do; they see only what it has done so far. For regulated tasks (investment proposals, credit strategies), this is unacceptable. Regulators want to review intent, more than execution.
Plan-and-execute fixes all three. The plan is visible upfront and reviewable. Each step’s context is narrower (just the task and the current step). The plan itself can be optimised as a whole, rather than step-by-step myopically.
The basic plan-and-execute graph
Build the simplest possible plan-and-execute graph, then add sophistication.
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from pydantic import BaseModel, Field
class Step(BaseModel):
description: str
tool: Optional[str] = None
expected_output: str
class Plan(BaseModel):
goal: str
steps: List[Step]
class PlanExecuteState(TypedDict):
input: str
plan: Optional[Plan]
past_steps: Annotated[List[dict], lambda a, b: a + b]
response: Optional[str]
planner_prompt = """You are a planner. Given a user task, produce a plan of 3-7 discrete steps.
Each step must have a clear description, optionally a tool to use, and an expected output.
Return the plan as JSON matching the Plan schema."""
def plan_node(state: PlanExecuteState) -> dict:
plan_llm = llm.with_structured_output(Plan)
plan = plan_llm.invoke([
SystemMessage(content=planner_prompt),
HumanMessage(content=state["input"])
])
return {"plan": plan}
def execute_node(state: PlanExecuteState) -> dict:
plan = state["plan"]
past = state["past_steps"]
next_step_idx = len(past)
if next_step_idx >= len(plan.steps):
return {}
step = plan.steps[next_step_idx]
step_prompt = f"Goal: {plan.goal}\nStep {next_step_idx + 1}: {step.description}\nExpected output: {step.expected_output}"
result = agent_executor.invoke({"input": step_prompt})
return {"past_steps": [{"step": step.description, "result": result["output"]}]}
def replan_or_respond(state: PlanExecuteState) -> str:
if len(state["past_steps"]) >= len(state["plan"].steps):
return "respond"
return "execute"
def respond_node(state: PlanExecuteState) -> dict:
synth_prompt = f"Goal: {state['plan'].goal}\nResults so far: {state['past_steps']}\nProduce a final response for the user."
response = llm.invoke(synth_prompt)
return {"response": response.content}
builder = StateGraph(PlanExecuteState)
builder.add_node("plan", plan_node)
builder.add_node("execute", execute_node)
builder.add_node("respond", respond_node)
builder.set_entry_point("plan")
builder.add_edge("plan", "execute")
builder.add_conditional_edges("execute", replan_or_respond, {"execute": "execute", "respond": "respond"})
builder.add_edge("respond", END)
graph = builder.compile()Three properties matter. First, the plan is typed via Pydantic, so the LLM’s output is validated. Any non-conforming plan triggers a retry with a corrective message. Second, the executor is a separate LangGraph subgraph (or a langchain AgentExecutor); it handles the messy details of running each step, including tool calls. Third, the state tracks past steps separately from the plan, so you can always see what has been done versus what remains.
Worked example: the investment proposal agent
Trace a real task through this pattern. The task: “Produce an investment proposal for a new client, the synthetic client, risk profile moderate, investable assets £2.3M, wants capital growth with some income, UK tax resident.”
Step 1. Planner produces the plan. The planner LLM returns:
{
"goal": "Produce an investment proposal for the synthetic client",
"steps": [
{"description": "Retrieve the synthetic client's current portfolio", "tool": "get_portfolio", "expected_output": "List of holdings with values and asset classes"},
{"description": "Pull current house views on asset allocation for moderate risk profile", "tool": "get_house_views", "expected_output": "Recommended allocation across equity, bonds, alternatives"},
{"description": "Identify mismatches between current portfolio and house view", "tool": null, "expected_output": "List of over/underweight positions"},
{"description": "Propose rebalancing trades with tax optimisation", "tool": "propose_rebalance", "expected_output": "List of buy/sell trades with quantities and tax implications"},
{"description": "Draft proposal document with rationale and risk commentary", "tool": null, "expected_output": "A five-page document"}
]
}This plan is visible in state before any step runs. A human wealth manager, in the UI, sees it and can approve, edit, or reject. Say they approve.
Step 2. Executor runs step 1. It calls
get_portfolio for the synthetic client. The tool returns a
list of eighteen holdings across equities, fixed income, and one real
estate fund. The result is stored in past_steps.
Step 3. Executor runs step 2. It calls
get_house_views. The view recommends 60% equity, 30% fixed
income, 10% alternatives.
Step 4. Executor runs step 3. No tool needed; the LLM compares the portfolio to the target. It notes: 47% equity (underweight), 48% fixed income (overweight), 5% alternatives (underweight).
Step 5. Executor runs step 4. It calls
propose_rebalance with the current and target allocations.
The tool returns a list of trades with tax implications.
Step 6. Executor runs step 5. No tool; the LLM drafts the proposal document.
Step 7. Responder synthesises. The response summarises: here is the proposal document, here are the trades with tax implications, here is the risk commentary. The wealth manager reviews and edits before sending to the client.
Seven steps. The plan was visible upfront. Each step was narrowly
focused. The artefacts (portfolio data, house view, rebalance proposal,
document) are all in past_steps and can be inspected.
Adding replanning
The simple version above runs through the plan linearly. But what if
the executor encounters a surprise? Say the
propose_rebalance tool reveals that one of Hastings’s
holdings is in a discontinued share class and cannot be traded normally.
The plan’s step 4 is now wrong; the document in step 5 will be
misleading.
This is where replanning comes in. After each executed step (or after a defined frequency), control returns to the planner, which sees the execution results so far and decides whether to revise the plan.
class ReplannerOutput(BaseModel):
action: Literal["continue", "replan", "respond"]
revised_plan: Optional[Plan] = None
response: Optional[str] = None
replanner_prompt = """You are a replanner. Given the original plan and the results of executed steps,
decide whether to continue with the existing plan, replan with a revised plan, or respond if done."""
def replanner_node(state: PlanExecuteState) -> dict:
replan_llm = llm.with_structured_output(ReplannerOutput)
decision = replan_llm.invoke([
SystemMessage(content=replanner_prompt),
HumanMessage(content=f"Original plan: {state['plan']}\nSteps completed: {state['past_steps']}")
])
if decision.action == "replan":
return {"plan": decision.revised_plan}
elif decision.action == "respond":
return {"response": decision.response}
return {} # continue with existing plan
builder.add_node("replan", replanner_node)
builder.add_edge("execute", "replan")
builder.add_conditional_edges("replan", lambda s: "respond" if s.get("response") else "execute",
{"respond": "respond", "execute": "execute"})Now every step flows through the replanner. If it detects a problem, it revises. If it detects completion, it transitions to respond. If everything is fine, it continues.
Deeper mechanism: plan granularity
A subtle design question: how granular should plan steps be?
Too coarse and you lose the benefits of planning. If the plan is “1. Understand the customer. 2. Produce a proposal,” then each step is really a miniature plan-and-execute, and you have gained nothing.
Too fine and you lose flexibility. If the plan has fifty steps, the planner is essentially writing the execution itself, and the plan becomes brittle to any deviation.
The sweet spot is 3-8 steps for most tasks. Each step is one conceptual unit of work: a tool call with some reasoning, or a synthesis across prior results, or a draft of one section. The planner stays at the level of “what” not “how.” The executor handles the “how” within each step.
The way to get the granularity right is through iteration. Run the agent on realistic tasks. Observe the plans. If steps are too coarse (the executor is doing too much in one step), split them. If they are too fine (the replanner is constantly reorganising), combine them. After five or six iterations, the planner typically settles into a good rhythm.
A practical heuristic: each step should take no more than three LLM calls in the executor. If a step routinely requires more, it is too coarse. If many steps require zero LLM calls (just a tool call), they are probably too fine, and you can fuse them.
Plan-and-execute in a commercial credit restructure
In the Merehaven worked scenario, the scenario uses plan-and-execute for a commercial credit restructure agent. The scenario: an existing corporate borrower is in breach of a covenant. The relationship manager wants options: extend the maturity, restructure the covenant, refinance with additional security, exit. The agent’s job is to help the RM think through the options and produce a restructure proposal paper.
The planner produces something like:
Goal: Produce a restructure options paper for customer X
Steps:
1. Retrieve the customer's current exposures, covenants, and breach details
2. Pull the customer's recent financial performance and projections
3. Retrieve internal restructure policy and playbook for this sector
4. Analyse covenant breach severity and likelihood of cure without restructure
5. Generate 3-4 restructure options with pros/cons each
6. Stress-test each option against a baseline and downside scenario
7. Recommend the preferred option with rationale
8. Draft the options paper (typically 6-8 pages)The RM sees this plan before the agent runs. If they want a different emphasis (say, stress-test against 3 scenarios instead of 2, or skip option 3 because they know the customer will refuse it), they edit the plan. The agent then executes the edited plan.
Replanning kicks in routinely. For example, in step 5, if the agent discovers that the customer has a material guarantee we didn’t know about, the set of viable options changes, and the replanner extends the plan to re-evaluate.
This pattern has been materially better than our original ReAct attempts for this use case. It makes the agent’s intent inspectable by the RM, lets the RM steer without deep-diving into prompt engineering, and produces higher-quality outputs because each step is narrowly focused.
Failure modes of plan-and-execute
Three failure modes are worth understanding.
Failure mode 1: plan that looks good but is wrong. The planner produces a plan that reads plausibly but misses a important step or includes a redundant one. Without close scrutiny, the plan is executed and the output has a gap. Mitigation: have the planner’s output reviewed by a second LLM call that checks for completeness against a task-specific rubric. For high-value tasks, have a human approve the plan.
Failure mode 2: executor drift. The executor, within a step, interprets the step description loosely and does something different than the planner intended. Mitigation: structured step descriptions with explicit expected outputs; validation of the actual output against the expected description; retry or escalation on mismatch.
Failure mode 3: premature commitment. The planner commits to a plan before enough context is gathered, and the first few execution steps reveal that the plan is fundamentally wrong. The replanner can fix this, but at the cost of wasted execution. Mitigation: allow the planner to include “gather-and-plan” as an early step, where it runs a few information-gathering operations before committing to the full plan.
The analogy revisited: plans as contracts
The second analogy for plan-and-execute: plans as contracts. When a law firm engages on a matter, they produce an engagement letter specifying the scope. The engagement letter is the contract. If the scope changes, the letter changes.
A LangGraph plan is an engagement letter. It specifies what the agent will do. The human (RM, wealth manager, underwriter, customer) can read it, agree to it, or modify it. The agent’s execution is bounded by the agreed scope. Scope changes require replanning, which requires re-agreement.
This framing is valuable when you are explaining plan-and-execute to business stakeholders. It is something they recognise: scope, agreement, execution, change orders. The machine is doing something they have done for years in their own work.
A thought experiment: plan-and-execute for a complaint response
Sketch a plan-and-execute decomposition on paper. What are the steps? Which are tool calls? Which are synthesis?
A candidate plan.
Goal: Produce a DISP-compliant response to a customer complaint
Steps:
1. Parse the complaint letter into discrete grievances with supporting detail (no tool)
2. For each grievance: retrieve relevant policy, product terms, and past precedent (tool: search_policy_corpus)
3. For each grievance: assess merit against the evidence (no tool, LLM reasoning)
4. For each meritorious grievance: propose redress (apology, explanation, refund, goodwill payment) aligned to policy (tool: redress_ladder)
5. Consolidate into a single draft response covering all grievances and a clear decision
6. Check the draft against DISP format requirements (tool: disp_format_check)
7. Final draftThe design shows how this plan scales naturally if the complaint has two grievances or seven: step 2, 3, and 4 are per-grievance loops. The planner can produce a plan with 7 steps for a three-grievance complaint or 10 steps for a four-grievance complaint.
This kind of task-specific planning, where the plan scales to the input, is where plan-and-execute shines. ReAct would navigate it, but would be slow and inconsistent. A hand-coded pipeline would be rigid. Plan-and-execute gives you the flexibility of an agent with the structure of a pipeline.
Part IV: Grounding and coordinated expertise
Retrieval and multi-agent patterns add surfaces, not certainty. Their value comes from evidence routing, specialisation boundaries and explicit disagreement.
Treat retrieval as an evidence route
The basic RAG pipeline
The essential steps:
- Ingest: take your authoritative documents (policies, product terms, regulatory guidance), chunk them, embed them, store them in a vector database.
- Retrieve: at query time, embed the user’s question, find the most similar chunks in the vector database.
- Generate: construct a prompt that includes the user’s question and the retrieved chunks; send it to the LLM; return the response.
from langchain_google_vertexai import VertexAIEmbeddings, ChatVertexAI
from langchain_community.vectorstores import MatchingEngine
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
# Ingestion
embeddings = VertexAIEmbeddings(model_name="text-embedding-004")
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=150)
chunks = splitter.split_documents(policy_documents)
vector_store = MatchingEngine.from_documents(
documents=chunks,
embedding=embeddings,
index_id=VERTEX_INDEX_ID,
endpoint_id=VERTEX_ENDPOINT_ID
)
# Retrieval + Generation
llm = ChatVertexAI(model_name="gemini-1.5-pro")
def rag_answer(question: str):
relevant_chunks = vector_store.similarity_search(question, k=5)
context = "\n\n".join(doc.page_content for doc in relevant_chunks)
prompt = f"""Answer the question using only the context provided.
If the context does not contain the answer, say "I don't have that information in my knowledge base."
Context:
{context}
Question: {question}
Answer:"""
return llm.invoke(prompt).contentThis is naïve RAG. It works for simple questions where the answer is likely to be in one chunk. It fails on questions where the answer spans multiple chunks, where retrieval returns wrong chunks, or where the model needs to reason across chunks rather than quote from one.
The chunking problem
Ask any production RAG team what bit them first. The answer, usually, is chunking. It looks trivial: split documents into pieces, embed each piece. In practice it is the most consequential design decision in the pipeline.
Fixed-size chunking (the default: 1000 characters with 150 overlap) is fine for consistently-structured text like a single policy document. It breaks down for documents with mixed structures. A 40-page product terms document with tables, bullet lists, and narrative prose will chunk poorly: some chunks cut mid-sentence, some cut mid-table, some cut mid-bullet. Retrieval returns fragments that are missing context. The generator confabulates.
Hierarchical chunking chunks at multiple granularities: fine (paragraph), medium (section), coarse (chapter). At retrieval time, you can match against fine chunks for specificity and medium chunks for context. This is sometimes called parent-document retrieval: retrieve at fine granularity to get the matching snippet, then expand to the parent section to provide surrounding context to the generator.
In a commercial credit policy corpus In the Merehaven worked scenario, the scenario uses hierarchical chunking. The policy PDF is parsed by unstructured.io into sections and paragraphs. Each paragraph is embedded as a “leaf” chunk with metadata linking to its section and chapter. At query time, we retrieve the top 8 leaf chunks, then expand each to include the surrounding 2 paragraphs. The generator gets about 2500 tokens of contextualised evidence rather than 800 tokens of fragmented excerpts. Answer quality is materially better.
Self-RAG: the model grading its own retrieval
Self-RAG (Asai et al., 2024) adds a feedback loop: the model grades the retrieved chunks for relevance before using them, and grades its own answer for faithfulness to the chunks.
The pattern in LangGraph:
class SelfRAGState(TypedDict):
question: str
documents: List[Document]
answer: Optional[str]
relevance_grades: List[bool]
answer_is_faithful: Optional[bool]
def retrieve_node(state: SelfRAGState) -> dict:
docs = vector_store.similarity_search(state["question"], k=5)
return {"documents": docs}
def grade_relevance_node(state: SelfRAGState) -> dict:
grades = []
for doc in state["documents"]:
grade_prompt = f"Is this document relevant to the question?\nQuestion: {state['question']}\nDocument: {doc.page_content}\nAnswer only 'yes' or 'no'."
resp = llm.invoke(grade_prompt).content.strip().lower()
grades.append("yes" in resp)
return {"relevance_grades": grades}
def filter_docs_node(state: SelfRAGState) -> dict:
filtered = [d for d, g in zip(state["documents"], state["relevance_grades"]) if g]
return {"documents": filtered}
def generate_node(state: SelfRAGState) -> dict:
context = "\n\n".join(d.page_content for d in state["documents"])
answer = llm.invoke(f"Context:\n{context}\n\nQuestion: {state['question']}").content
return {"answer": answer}
def grade_faithfulness_node(state: SelfRAGState) -> dict:
prompt = f"Is this answer fully supported by the context?\nContext:\n{state['documents']}\n\nAnswer: {state['answer']}\n\n'yes' or 'no'?"
resp = llm.invoke(prompt).content.strip().lower()
return {"answer_is_faithful": "yes" in resp}
def route_after_faithfulness(state: SelfRAGState) -> str:
if state["answer_is_faithful"]:
return "end"
return "retrieve" # retry with different strategyThe graph flow: retrieve, grade each chunk for relevance, filter, generate, grade the answer for faithfulness. If unfaithful, loop back and retry (possibly with different retrieval parameters).
The cost is several extra LLM calls per query. The benefit is that plainly irrelevant chunks do not pollute the context, and plainly unfaithful answers get rejected before reaching the user. For high-stakes BFSI contexts (policy interpretations, regulatory guidance), the cost is worth it.
Corrective RAG (crag)
Corrective RAG (Yan et al., 2024) goes further: if retrieval fails (no relevant chunks found), fall back to a different source, typically web search or a second corpus.
class CRAGState(TypedDict):
question: str
documents: List[Document]
retrieval_quality: Optional[Literal["correct", "ambiguous", "incorrect"]]
answer: Optional[str]
def assess_retrieval_node(state: CRAGState) -> dict:
# Assess top chunks against question
prompt = f"Are these documents sufficient to answer the question? Answer 'correct', 'ambiguous', or 'incorrect'.\nQuestion: {state['question']}\nDocuments: {state['documents']}"
resp = llm.invoke(prompt).content.strip().lower()
if "correct" in resp:
return {"retrieval_quality": "correct"}
elif "ambiguous" in resp:
return {"retrieval_quality": "ambiguous"}
return {"retrieval_quality": "incorrect"}
def fallback_node(state: CRAGState) -> dict:
# Fallback: web search or second corpus
web_results = web_search_tool.invoke(state["question"])
docs = [Document(page_content=r["snippet"], metadata={"source": r["url"]}) for r in web_results]
return {"documents": docs}
def refine_query_node(state: CRAGState) -> dict:
# Rewrite for ambiguous retrieval
prompt = f"Rewrite this question to be more searchable: {state['question']}"
new_q = llm.invoke(prompt).content
return {"question": new_q}Adaptive RAG
Adaptive RAG (Jeong et al., 2024) picks the retrieval strategy based on the question type. Some questions need no retrieval (general knowledge, math). Some need single-hop retrieval (simple fact lookup). Some need multi-hop retrieval (reasoning across multiple documents).
Worked example: a customer asking about ISA rules
User question: “Can I transfer my cash ISA from another provider to your bank mid-tax-year, and does it count against my annual allowance?”
Step 1. Adaptive router. Classifies as a single-hop RAG query (one factual topic, one answer).
Step 2. Retrieval. Embed the question, find top 5 chunks. Top chunks include: (i) a page from the bank’s ISA terms on transfers; (ii) a page from HMRC guidance on ISA transfer rules; (iii) a page from an internal product FAQ; (iv) a page from the current ISA allowance rules; (v) an irrelevant page about cash savings accounts.
Step 3. Relevance grading. The bank’s ISA terms and HMRC guidance pages are graded relevant. The FAQ and irrelevant cash savings page are filtered out.
Step 4. Generation. The LLM composes: “Yes, you can transfer your cash ISA mid-tax-year. Transfers between ISAs do not count against your annual ISA allowance. To initiate the transfer, you should complete the transfer form available on the bank’s website. Note that you must use the formal transfer process, not close the existing ISA and deposit the funds yourself, as the latter would count as a new subscription.”
Step 5. Faithfulness grading. The answer is checked against the retrieved chunks. It passes: every claim in the answer is supported by the chunks.
Step 6. Return. The answer is returned with citations: each sentence linked to the chunk that supports it.
That last step, citations, is the piece most naïve RAG implementations skip. In a BFSI context, it is not optional. The customer, the relationship manager, and the complaints team all benefit from being able to trace an answer back to its authoritative source.
Citations: the regulated banking-critical finishing touch
Every answer in a regulated RAG system should have citations. The customer-facing answer might say “Cash ISAs allow mid-year transfers [1]. These transfers do not count against your annual allowance [2].” The citations link to the specific chunks (and their original document sources) that support each claim.
class Citation(BaseModel):
text: str # the claim in the answer
source: str # document title or URL
chunk_id: str
score: float # retrieval similarity
def generate_with_citations(question: str, docs: List[Document]) -> tuple[str, List[Citation]]:
prompt = f"""Answer the question using the context. For each factual claim in your answer,
append a citation marker [N] referring to the document index.
Question: {question}
Context (indexed):
{format_docs_indexed(docs)}
Answer with inline citations:"""
response = llm.invoke(prompt).content
citations = extract_citations(response, docs)
return response, citationsIn the Merehaven worked scenario, every customer-facing RAG answer includes citations surfaced as hoverable links in the UI. The reviewer can click to see the exact source. This is the property that turns a RAG system from “unverified model prose” into “evidence-backed response,” and it is what regulatory reviewers and complaint handlers actually need.
Production RAG: ingestion pipelines and freshness
The pipeline: 1. Source documents land in a GCS bucket (or S3) with metadata tags (document type, version, effective date). 2. A Cloud Run job (or Lambda) watches the bucket; when a new document appears, it triggers ingestion. 3. Ingestion parses the document, chunks it, embeds the chunks, and upserts to the vector store. Old chunks with the same source ID are marked superseded. 4. A freshness dashboard tracks time-since-last-refresh per document type. Alerts fire if any type exceeds its freshness SLA.
RAG failure modes
Five failure modes worth knowing.
Hallucination despite retrieval. The model ignores the retrieved context and answers from its parametric memory. Mitigation: stricter system prompt (“answer only from context”), faithfulness grading, strong citation requirement.
Retrieval miss. The relevant chunk exists in the knowledge base but isn’t retrieved. Mitigation: hybrid search (combine vector + keyword), query expansion, reranking.
Chunk boundary issues. The relevant information is split across two adjacent chunks and neither is fully relevant alone. Mitigation: chunk overlap (already standard), parent-document expansion.
Stale content. The knowledge base is out of date and the model answers confidently from stale chunks. Mitigation: freshness SLAs, version tagging, conservative phrasing for time-sensitive topics.
Context window overflow. Too many retrieved chunks pushed into the generator, exceeding the context or degrading quality. Mitigation: reranking, aggressive relevance filtering, hierarchical summarisation of retrieved content.
Coordinate specialists without losing ownership
Why multi-agent?
Two reasons, one technical, one organisational.
Technically, specialisation helps. A generalist agent with thirty tools and a lengthy system prompt performs worse than specialist agents each with five tools and a tight system prompt. Context confuses LLMs. Long tool lists inflate the prompt and dilute attention. Experimental evidence is consistent: decomposing a complex task into specialist roles tends to improve accuracy, reduce token usage per decision, and simplify debugging.
Organisationally, specialisation matches how work gets done. Banks already have specialists (trade finance, financial crime, tax, legal, credit, treasury) with distinct mandates. Multi-agent architectures let you mirror this structure in code. A credit specialist agent is owned by the credit risk team. A compliance specialist agent is owned by compliance. Each team iterates on its own agent; the supervisor is owned by the platform team. This separation of concerns scales the number of agents you can maintain in an operating environment.
The cost is coordination. Every agent-to-agent handoff is an opportunity for confusion, duplication, or deadlock. The supervisor’s prompt gets complex. The debugging story is harder: when something goes wrong, which agent was responsible? These are real costs, and for simple tasks, a single agent is better.
A rule of thumb. Use single-agent (ReAct or plan-and-execute) up to about 8-10 tools in one domain. Beyond that, or across domains, consider multi-agent.
Supervisor pattern: the canonical example
The canonical multi-agent pattern is one supervisor and N specialists.
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Literal
from langgraph.graph.message import add_messages
class SupervisorState(TypedDict):
messages: Annotated[List, add_messages]
next_agent: Optional[str]
task_complete: bool
AGENT_NAMES = ["credit_specialist", "compliance_specialist", "tax_specialist"]
supervisor_prompt = f"""You are a supervisor coordinating specialist agents. Based on the conversation,
decide which specialist should handle the next turn, or whether the task is complete.
Available agents: {AGENT_NAMES}
Respond with JSON: {{"next": "agent_name"}} or {{"next": "DONE"}}."""
def supervisor_node(state: SupervisorState) -> dict:
response = supervisor_llm.invoke([
{"role": "system", "content": supervisor_prompt},
*state["messages"]
])
decision = parse_json(response.content)
if decision["next"] == "DONE":
return {"task_complete": True, "next_agent": None}
return {"next_agent": decision["next"]}
def credit_specialist_node(state: SupervisorState) -> dict:
response = credit_agent.invoke(state)
return {"messages": [response], "next_agent": "supervisor"}
# Similarly for compliance_specialist and tax_specialist
def route(state: SupervisorState) -> str:
if state["task_complete"]:
return END
return state["next_agent"]
builder = StateGraph(SupervisorState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("credit_specialist", credit_specialist_node)
builder.add_node("compliance_specialist", compliance_specialist_node)
builder.add_node("tax_specialist", tax_specialist_node)
builder.set_entry_point("supervisor")
builder.add_conditional_edges("supervisor", route,
{"credit_specialist": "credit_specialist",
"compliance_specialist": "compliance_specialist",
"tax_specialist": "tax_specialist",
END: END})
builder.add_edge("credit_specialist", "supervisor")
builder.add_edge("compliance_specialist", "supervisor")
builder.add_edge("tax_specialist", "supervisor")Pattern: all paths go through the supervisor. Specialists never hand off to each other directly. This makes control flow easy to reason about: every handoff is centralised. The supervisor has the full message history and can see what each specialist contributed.
Worked example: a trade finance facility request
A corporate relationship manager asks the RM Copilot: “Client X wants to increase their trade finance facility from £5m to £10m. They are in food manufacturing, exporting to Kenya and Ghana. Can you help me think this through?”
Turn 1. Supervisor decides: credit_specialist.
The credit specialist pulls the customer’s financials, current exposures, and covenant status. Produces analysis: “Client has serviced a £5m TF facility for 2 years with no breaches. Financials show 18% YoY turnover growth. Capital adequacy is within policy for the uplift to £10m. Sector-specific guidance does not flag concerns.”
Turn 2. Supervisor decides: compliance_specialist.
Turn 3. Supervisor decides: credit_specialist.
Credit incorporates compliance findings. Proposes a conditional limit of £8m (not the full £10m) pending receipt of the enhanced KYC documentation on buyers, with a review in 6 months. Produces a structured recommendation.
Turn 4. Supervisor decides: DONE.
Four turns total. Each specialist focused on their narrow expertise. The supervisor integrated the findings. The RM gets a coherent answer. If they had asked the credit specialist alone, they would have missed the compliance nuance. If they had asked compliance alone, they would have missed the credit context.
Network pattern: peer-to-peer collaboration
Sometimes specialists need to talk to each other directly without supervisor mediation. Consider a multi-step fraud investigation where the pattern analyst needs to pass findings directly to the case handler, who then passes to the KYC analyst, who may refer back to the pattern analyst.
In the network pattern, each specialist has edges to others. The supervisor role may still exist but as an arbiter for deadlocks, not a routine gatekeeper.
def pattern_analyst_node(state):
response = pattern_analyst.invoke(state)
next_agent = determine_next(response) # the agent itself decides
return {"messages": [response], "next_agent": next_agent}
builder.add_conditional_edges("pattern_analyst", lambda s: s["next_agent"],
{"case_handler": "case_handler", "kyc_analyst": "kyc_analyst", "END": END})Network patterns scale less well than supervisor patterns, because every added specialist increases the number of potential edges. The decision graph in each specialist’s head (“who should I pass to next?”) gets complex. Use network patterns only when domain flow genuinely requires peer-to-peer (like the fraud case above), and keep the total number of specialists to 3-4.
Hierarchical pattern: teams of teams
For complex enterprise workflows, a single supervisor with many specialists becomes unwieldy. A hierarchical pattern layers supervisors: a top-level supervisor delegates to sub-team supervisors, each of whom coordinates their own specialists.
Example In the Merehaven worked scenario,: a relationship manager copilot has a top-level supervisor that routes between three sub-teams. - Credit team: credit specialist, exposures specialist, collateral specialist. - Compliance team: financial crime specialist, sanctions specialist, KYC specialist. - Product team: lending specialist, trade finance specialist, FX specialist.
Each sub-team has its own supervisor. The top-level supervisor decides which team to engage. The team supervisor decides which specialist within their team handles the turn.
In LangGraph, this is implemented via subgraphs: each sub-team is a compiled graph with its own state; the top-level graph treats each sub-team as a single node.
# Sub-team graph: credit team
credit_team_builder = StateGraph(CreditTeamState)
credit_team_builder.add_node("credit_supervisor", credit_supervisor_node)
credit_team_builder.add_node("credit_specialist", credit_specialist_node)
credit_team_builder.add_node("exposures_specialist", exposures_specialist_node)
credit_team_builder.add_node("collateral_specialist", collateral_specialist_node)
# ... edges and entry point
credit_team_graph = credit_team_builder.compile()
# Top-level graph uses sub-team graph as a node
top_builder = StateGraph(TopState)
top_builder.add_node("top_supervisor", top_supervisor_node)
top_builder.add_node("credit_team", credit_team_graph)
top_builder.add_node("compliance_team", compliance_team_graph)
top_builder.add_node("product_team", product_team_graph)
# ... conditional edges from top_supervisor to each teamThe hierarchical pattern scales to 20+ specialists by keeping each level’s decision space small (3-4 options per supervisor). No supervisor is overwhelmed with a long list. Debugging is easier because you can localise failures to a specific team.
Specialised agent architectures
Some agent patterns deserve their own section because they recur across BFSI use cases.
The verifier pattern. One agent produces; another verifies. Used when output quality matters more than speed. Credit memo drafter + policy-compliance verifier is the canonical pair In the Merehaven worked scenario,. The verifier has its own policy corpus retrieval and runs checks that are too expensive to bake into the drafter.
The researcher-writer pattern. One agent gathers information broadly; another composes a narrative from the gathered material. Used for research-heavy outputs like market commentary or sector reports. The researcher focuses on recall; the writer focuses on synthesis.
The conductor-orchestra pattern. A conductor agent sequences many specialist calls, often in parallel, and assembles the results. Used for batch operations like scoring a portfolio or running a quarterly review across many customers.
The adversarial pair pattern. Two agents: one proposes, one critiques; they iterate until convergence. Used for negotiation simulations, dispute resolution drafts, or any task where pressure-testing improves the output. More expensive than single-agent but produces notably better outputs on open-ended tasks.
Multi-agent failure modes
Three failure modes every production multi-agent system encounters.
Failure 1: infinite ping-pong. Specialist A hands to specialist B, who hands back to A, who hands to B, indefinitely. Mitigation: iteration counter in supervisor state; force a decision after N turns. Detect thought-loops by comparing recent supervisor decisions.
Failure 2: supervisor dilution. The supervisor’s prompt grows with each added specialist, and its decisions become unreliable. Mitigation: migrate to hierarchical pattern; keep each supervisor’s option set to 3-4. Or use a learned router (small classifier fine-tuned on routing examples) instead of an LLM supervisor, which scales better to 10+ routes.
Failure 3: inconsistent context. Specialists see different slices of state and produce contradictory findings. Mitigation: agree on a canonical state schema up front; have the supervisor include a “state summary” in each specialist’s prompt; use structured outputs so findings are machine-mergeable.
Part V: Release, recovery and operating evidence
The graph earns wider authority only through operating evidence. Test the route, constrain the release, preserve receipts and design recovery before scale.
Operate the graph after the demo
The four-phase lifecycle
Every LangGraph agent that reaches production moves through four phases, and each has its own risks and checks.
Phase 1: Prototype. A developer builds a working agent on their laptop. The scope is narrow, the data is synthetic or sampled, and nothing is at stake. Output: a demonstrable proof-of-concept that the task is feasible.
Phase 2: Pilot. A small group of real users (typically 10-50) uses the agent for real tasks. Outputs are closely monitored. Feedback is collected systematically. Failures are captured and categorised. Output: a confident claim about whether the agent can handle worked tasks and where its weaknesses are.
Phase 3: Rollout. The agent is expanded to larger populations, often incrementally (10%, 25%, 50%, 100%). Metrics are watched for regression. Feedback mechanisms scale. Output: broad adoption with predictable quality.
Phase 4: Operation. The agent runs steadily. Monitoring detects drift, performance changes, regulatory events. Updates are deployed carefully. Eventually, the agent is retired or succeeded. Output: durable value with acceptable risk.
Most teams nail phase 1. Many fail at phase 2 because they don’t instrument it properly. The most common failure is between 2 and 3: a pilot that looked good gets rolled out and breaks because the broader population has different patterns.
Testing LangGraph agents
Testing an LLM agent looks strange to engineers trained on conventional software. The outputs are non-deterministic. The inputs are natural language, so the input space is essentially infinite. Correctness is often subjective. Traditional unit tests don’t quite work.
But structured testing is possible and essential. Three layers.
Layer 1: unit tests for nodes. Each graph node is a Python function. Test it like any function: given input state, expect output state to have certain properties. For nodes that call the LLM, either mock the LLM (for structural tests) or use a small deterministic model (for regression).
def test_credit_memo_draft_node_includes_required_sections():
state = {
"customer_id": "CUS-TEST-001",
"exposures": [...],
"financials": {...},
}
result = draft_memo_node(state)
draft = result["draft_memo"]
assert "Executive Summary" in draft
assert "Financial Analysis" in draft
assert "Risk Rating" in draft
assert "Recommendation" in draftLayer 2: integration tests for graph paths. Invoke the whole graph with realistic inputs and assert on final state. These are slower (real LLM calls) but catch issues that unit tests miss: incorrect routing, state shape mismatches between nodes, missing fields.
def test_happy_path_credit_memo_completion():
config = {"configurable": {"thread_id": "test-happy-001"}}
result = graph.invoke(
{"customer_id": "CUS-INTEGRATION-001", "requested_facility": 5_000_000},
config=config
)
assert result["final_memo"] is not None
assert result["audit"]["run_id"] == config["configurable"]["thread_id"]
assert len(result["audit"]["tool_calls"]) > 0Layer 3: evaluation suites. A curated set of (input, expected-quality) pairs, scored by LLM-as-judge or human. Run on every change. Track scores over time.
class EvalCase(BaseModel):
input: dict
expected_properties: List[str]
expected_decision: Optional[str]
def run_evals(graph, cases: List[EvalCase]) -> dict:
scores = []
for case in cases:
output = graph.invoke(case.input)
score = score_output(output, case.expected_properties)
scores.append(score)
return {"mean": sum(scores) / len(scores), "pass_rate": sum(1 for s in scores if s > 0.8) / len(scores)}Evaluation suites are the backbone of confident LLM deployment. Every prompt change, every model version upgrade, every policy update runs the eval suite before any release. If scores drop, release is blocked.
LLM-as-judge for evaluation
How do you score an LLM’s output when “correctness” is subjective? A common answer is LLM-as-judge: use another LLM to grade outputs against a rubric.
judge_prompt = """You are evaluating a credit memo. Score it 1-5 per criterion.
Memo: {memo}
Criteria:
1. Completeness: exposures, financials, risk rating, recommendation?
2. Faithfulness: all claims supported by data?
3. Policy compliance: citations by version?
4. Clarity: would a new RM understand?
5. Tone: professional?
Return JSON: {"completeness":5,"faithfulness":4,"policy_compliance":5,"clarity":4,"tone":5,"reasoning":"..."}"""
def judge_memo(memo: str) -> dict:
return judge_llm.with_structured_output(JudgeScores).invoke(judge_prompt.format(memo=memo))Three subtleties.
First, judges are biased. They tend to rate longer outputs higher, favour verbose over concise, and miss subtle factual errors. Calibrate your judge against human ratings for a sample, and adjust prompts or use a different judge if calibration is poor.
Second, judges need rubrics, not free-form scoring. A rubric with named criteria produces more consistent scores than “rate this memo from 1 to 10.”
Judge diversity is a testable design choice. A different model family may reduce some common-mode errors, but disagreement alone is not quality. Calibrate each judge against blinded human adjudication and preserve failure exemplars.
In the Merehaven worked scenario, the scenario uses a trio: the generator is a pinned model; the judge for policy compliance is a pinned model (via Anthropic’s API); the judge for tone is a smaller model (a pinned model) because tone is a relatively easy signal. The diversification reduces common-mode blindness.
Deployment patterns
Three deployment patterns in an operating environment.
Pattern 1: Synchronous request/response. The user asks; the agent answers in decision-time. Used for chat interfaces, short queries, and anything interactive. Latency targets are 2-8 seconds. Implementation: agent behind a load balancer, stateless (state lives in checkpointer), auto-scaling on CPU/latency.
Pattern 2: Asynchronous job. The user submits a request; the system returns a job ID; the user polls or receives a webhook when done. Used for longer tasks (memo drafts, research reports, batch operations) that may take 30 seconds to several minutes. Implementation: Cloud Tasks or SQS queue; workers pull jobs, run the agent, store results; the UI polls.
Pattern 3: Event-triggered. An event fires (new application received, policy change detected, scheduled trigger), and the agent runs without direct user interaction. Used for monitoring, bulk processing, or routine operations. Implementation: Eventarc (GCP) or EventBridge (AWS) routes events to the agent service.
Observability: the three pillars
Production LLM agents need three kinds of observability.
Metrics. Counters and gauges: requests per second, p50/p95/p99 latency, error rate, tool call count, token usage. Dashboarded in Cloud Monitoring or CloudWatch. Alerts on anomalies.
Traces. Every agent run produces a distributed trace showing every node, every LLM call, every tool call, every retrieval. Used for debugging individual runs. Implemented via OpenTelemetry, propagating baggage across all spans.
Logs. Structured logs with consistent fields
(thread_id, user_id, customer_id,
agent_name, node_name, message).
Used for audit, debugging, and compliance. Retained per the bank’s
retention policy (typically 7 years for audit-tagged events).
LangSmith (from LangChain) provides LLM-specific observability: prompt/response pairs, chain steps, token usage per step. the scenario uses it alongside Cloud Trace because it surfaces LLM-specific details (prompt versions, model versions, temperature) that generic APM misses.
Canary rollouts and feature flags
Rolling out an update to a production agent requires care. Two mechanisms.
Canary rollouts. Deploy the new version to a small percentage of traffic first (5%, then 25%, then 50%, then 100%). Compare metrics between canary and stable. If canary is worse, roll back automatically.
import random
def route_to_version(user_id: str, canary_pct: float = 0.05) -> str:
hash_val = hash(user_id) % 100
return "canary" if hash_val < canary_pct * 100 else "stable"Feature flags. Control which users see which version at feature granularity. Flags are stored in a config service (Cloud Runtime Config, AWS AppConfig, or LaunchDarkly) and checked at runtime.
In the Merehaven worked scenario, the specimen combines both. Every significant update (prompt change, graph topology change, model version) is rolled out via canary. Features where we want user-level control (opt-in betas, pilots) are behind feature flags.
Monitoring: what to alert on
A pager goes off. Which signals matter?
Signal 1: error rate. Increase in HTTP 5xx or exception rate. Usually an infrastructure or code issue.
Signal 2: latency. P99 blowing up means something is slow. Often a downstream API issue or a blocking call.
Signal 3: token usage. Sudden increase means either usage spike (check traffic) or prompt bloat (check recent deployments). Token cost alerts are common because costs can balloon quickly.
Signal 4: evaluation score drift. If the online eval suite shows scores dropping, the agent’s quality is degrading. Usually caused by upstream changes (new customer patterns, model version update, corpus drift).
Signal 5: human override rate. If reviewers are overriding agent outputs more than usual, the agent is making mistakes. Track override rate per use case; alert if it exceeds the rolling baseline.
Signal 6: retrieval hit rate. If RAG retrieval is failing to find relevant chunks more often, the corpus may be stale or the embeddings may have drifted.
Signal 7: toxicity or safety flags. If the model’s safety filters are firing more often on outputs, something may have changed in inputs or in the model.
The evaluation flywheel
A key property of a mature production system is the evaluation flywheel.
- Production runs produce traces and outputs.
- A sample of outputs is labelled for quality (LLM-judge at volume, human-rated for gold set).
- Labels feed the evaluation suite.
- The suite detects regressions on every change.
- New failure modes discovered in an operating environment are added as new eval cases.
- The suite grows over time, covering the real distribution.
Without the flywheel, the eval suite becomes a museum piece: a set of tests written once, slowly drifting from reality. With the flywheel, every production failure becomes a future regression guard.
A thought experiment: your agent is being audited
What do you include?
At minimum: the model card (purpose, inputs, outputs, limitations). The training data lineage (where did the prompts come from, what RAG corpus). The evaluation results (how is the model tested, what are the scores, what is the gold set). The monitoring setup (what do you watch, what alerts, what SLOs). The governance record (who approved this, when, against what policy). The change history (every significant change since launch). The incident log (anything that went wrong, what caused it, what was changed).
Every one of these is a specific data source you need to have ready. The eval suite’s history (from the flywheel) is your evaluation evidence. The CI/CD pipeline’s log is your change history. The incident tooling’s records are your incident log. The observability stack is your monitoring evidence.
If you build these as you go, the audit is a week of pulling together existing artefacts. If you wait to build them at audit time, it is months of scrambling and the audit almost certainly fails.
The Merehaven controlled-graph lab
The synthetic Merehaven payment-support route receives a request, compiles permitted context, proposes a next step and pauses before any consequential action. The graph never treats a message history as world state. Account status, identity and payment outcome come from authoritative systems at decision time.
Typed state and receipts
from typing import Literal, NotRequired, TypedDict
class PaymentState(TypedDict):
case_id: str
customer_id: str
request_text: str
context_version: str
proposed_action: NotRequired[dict]
policy_result: NotRequired[Literal["allow", "review", "deny"]]
effect_receipt: NotRequired[dict]
outcome: NotRequired[Literal["verified", "failed", "unknown"]]
def may_execute(state: PaymentState) -> bool:
return (
state.get("policy_result") == "allow"
and state.get("outcome") is None
and state.get("effect_receipt") is None
)The predicate is deliberately boring. Boring logic is desirable at the authority boundary. The model may propose a structured action, but policy, idempotency and effect reconciliation stay deterministic.
Thought experiment: the flawless explanation
Suppose the assistant produces a perfectly clear explanation for a transfer that never occurred. Is the system successful? A language metric may say yes; the customer’s account says no. Now suppose the transfer occurred twice while the assistant apologised elegantly. The answer is worse than unhelpful because it hides an unresolved effect.
The thought experiment separates narrative quality from operational truth. Process philosophy treats entities as unfolding events rather than static substances. The engineering analogy is limited but useful: a graph state is not a permanent self. It is a versioned account of becoming, and each transition must preserve what changed, why and under whose authority.
Route contract
| Transition | Proposal | Independent control | Required evidence | Failure route |
|---|---|---|---|---|
| Request to context | Customer intent | Identity and access policy | Context lineage | Ask, deny or narrow |
| Context to proposal | Model output | Schema and policy pre-check | Prompt and model versions | Repair or review |
| Proposal to approval | Suggested action | Human or deterministic authorisation | Decision receipt | Deny or expire |
| Approval to effect | Typed action | Idempotency and entitlement | Effect receipt | Reconcile unknown |
| Effect to outcome | Claimed result | Authoritative readback | Outcome receipt | Contain and investigate |
First-hour graph incident runbook
- Freeze route identity: graph, prompt, model, tool, policy and schema versions.
- Withdraw consequential authority while leaving safe inspection paths available.
- Preserve the last checkpoint, attempted action contract and all returned receipts.
- Reconcile every unknown outcome before replay; do not convert uncertainty into failure or success by assumption.
- Reproduce the fault with synthetic state and a non-effecting tool double.
- Restore the last accepted route, then test the failing transition and adjacent veto dimensions.
- Record residual risk, owner, recovery evidence and the condition that would reopen the incident.
Release ladder
| Stage | Allowed route | Evidence gate | Withdrawal trigger |
|---|---|---|---|
| Inspect | Read-only traces | State and schema tests | Any lineage gap |
| Shadow | Propose without effect | Route replay and slice coverage | Policy disagreement |
| Assisted | Human-authorised effect | Decision and effect receipts | Unknown outcome |
| Bounded | Policy-limited delegation | Canary, reconciliation and harm measures | Drift or control failure |
Appendix: field glossary
Workflow checkpoint: a resumable snapshot of graph execution. User memory: retained information about a person, subject to purpose and deletion controls. Enterprise knowledge: governed reusable information. World state: authoritative time-qualified facts. Decision receipt: evidence of policy or human authorisation. Effect receipt: evidence returned by an action system. Unknown outcome: an attempted effect whose success cannot yet be proved. Bounded autonomy: delegated choice within explicit authority, evidence and recovery limits. Readback: independent verification after action.
