Vol. 22 · Application Engineering
Production LangChain + LangGraph: Code Structure & Runtime
This volume answers a more applied question: how does a real Python project on LangChain + LangGraph spread out into files? Not “which framework is better”, but where exactly “state”, “context”, “tools”, “middleware”, “nodes”, “graph.compile(...)”, “checkpointer”, “store”, HTTP-endpoints, “thread id”, streaming and “resume” live. This is a codebase map, not a concept map.
Production-Repository Map
A good project usually divides code into four layers: contracts, executables, orchestration graph, and serving/runtime. Then LangChain remains a layer of standard building blocks, and LangGraph becomes the place where these blocks are combined into a meaningful lifecycle.
Noda is responsible for the action: call the model, make a tool lookup, check approval. Routing is responsible for controlling flow: whether to go to ‘tools’, ‘review’, ‘END’ or another subgraph. If you mix it in one place, the graph quickly becomes unreadable.
Source-complete notespractice
practice
- node = work
- route = transition
- Runtime in a single function
This is not the only correct structure, but it shows how to share responsibility. The main idea is that ‘contracts’ should not depend on the API layer, ‘tools’ should not know about HTTP, and ‘graph’ should not create database clients directly inside the nodes.
It is very useful to decide in advance what the service does at the input of the query: whether thread creates, how it forms “config”, where it takes “context”, how it causes “graph.invoke” or “graph.astream”, what it returns to the outside, and how it processes “ interrupt ”.
Source-complete notesroutine
textHTTP request
→ build input state
→ build config {"configurable": {"thread_id": ...}}
→ build runtime context
→ invoke / astream
→ inspect final state or __interrupt__
→ return HTTP responseIf the project has support-agent, sales-agent and document-review workflow, do not try to cram everything into one megagraph. Individual graphs are easier to test, compile, version and maintain.
Source-complete noteschip
chip
- different state schema
- Different tools and middleware
- common components can be shuffled below
Contracts and basic schemes
In LangGraph, the state determines what data can live inside the execution. It is better to keep it in a separate file and discuss as a public contract: which keys are required, which transient, which accumulative, which write middleware, which write tools, which node and read.
Source-complete notesstately
sqlfrom langchain.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict, Annotated
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
user_query: str
retrieved_docs: list[str]
structured_answer: dict | None
approved: bool | NoneContext is a dependency injection for a specific start. There live well "user id", feature flags, tenancy, database handles, environment mode. It is not part of the evolving state or part of the default LLM prompt.
Source-complete notesexample
sqlfrom dataclasses import dataclass
@dataclass
class Context:
user_id: str
tenant_id: str
can_send_email: boolStructured output, HTTP payloads, tool input/output contracts, and final API responses are usually best handled through ‘Pydantic’. Then the internal graph state can remain light, and the outer boundaries become validated and stable.
Source-complete notesexample
sqlfrom pydantic import BaseModel
class AnswerSchema(BaseModel):
answer: str
citations: list[str]
needs_human_review: boolSettings are useful to make separately: model name, timeout, max tokens, DSN for checkpointer-a, streaming mode, feature flags. But "settings.py" should not create a graph by itself. Its job is to deliver configuration, not mix wiring with orchestration logic.
Source-complete notesWhat to keep in settings
What to keep in settings
- MODEL_NAME
- DB_DSN
- MAX_TOKENS
- random node side effects
This distinction is better fixed at the level of contracts. State is responsible for what happens inside the current run/thread. The store is responsible for long-lived user or application data through different threads and sessions.
Source-complete notestable
| Essence | For what? |
|---|---|
| State | messages, intermediate results, current approval |
| Store | preferences, profile, cross-thread memory |
Components of LangChain-Layer
The model is best initialized in one place. Then it is easier to change the provider, temperature, timeouts, streaming behavior and runtime-configurable parameters. If the model is smeared on the nodes, you quickly lose control of the call policy.
Source-complete notesexample
sqlfrom langchain.chat_models import init_chat_model
def build_model():
return init_chat_model(
"provider:model",
temperature=0,
timeout=30,
)Even if the prompt is short, keeping it separate is useful: it is easier to review, test, version and change without shoveling the orchestration code. This is especially important if middleware then makes dynamic prompt generation.
Source-complete notesstore
store
- system prompts
- response instructions
- tool-use policy hints
The official approach now is access to the state, context, store, config, stream writer and `tool call id' via `ToolRuntime'. This makes the tools testable and doesn’t make you drag hidden singletons or global request contexts.
Source-complete notesexample
sqlfrom langchain.tools import tool, ToolRuntime
@tool
def fetch_preferences(runtime: ToolRuntime[Context]) -> str:
user_id = runtime.context.user_id
if runtime.store:
memory = runtime.store.get(("users",), user_id)
if memory:
return memory.value["preferences"]
return "No preferences"Through middleware, it is convenient to do dynamic prompt, dynamic model selection, trimming, tool error handling, permission-based tool filtering and audit/logging. These are cross-cutting concerns that should not be smeared on each node with your hands.
Source-complete notesfrequent hooks
text@before_model
@after_model
@wrap_model_call
@wrap_tool_call
@dynamic_prompt
Node-style hooks are suitable for sequential logic, wrap-style – for control around call-aIt is better to depend on a small client wrapper than to directly collect SQL, HTTP-requests and deserialization in place. The tool remains orchestration-aware, but does not turn into 200 lines of integration code.
Source-complete notespattern
pattern
- tool → repository
- tool → raw httpx/sqlalchemy everywhere
Assembling Count by Files
The official examples of LangGraph constantly repeat one idea: the node reads the state and returns the changes. This makes execution predictable, helps reducers work correctly and makes debugging easier through checkpoints/state history.
Source-complete notesfine · badly
pythondef retrieve_node(state: AgentState):
docs = retriever.search(state["user_query"])
return {"retrieved_docs": docs}pythondef retrieve_node(state: AgentState):
state["retrieved_docs"] = retriever.search(...)
return stateWhen the logic of transitions lives apart, it becomes clear why the graph goes to ‘review’, ‘tools’ or ‘END’. This is more important than it seems: routing logic is quickly becoming the main place of business rules, and it should be read separately from node actions themselves.
Source-complete notesexample
sqlfrom langgraph.graph import END
def route_after_model(state: AgentState):
if state.get("approved") is False:
return END
if needs_human_review(state):
return "review"
if has_tool_calls(state["messages"][-1]):
return "tools"
return ENDThis is the place where state schema is associated with node, ribs, checkpointer, store and sometimes static interrupts. A good build graph() contains almost no business logic, it only collects the execution object from the already finished pieces.
Source-complete notesexample
sqlfrom langgraph.graph import StateGraph, START
def build_graph(checkpointer, store):
builder = StateGraph(AgentState)
builder.add_node("retrieve", retrieve_node)
builder.add_node("answer", answer_node)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "answer")
return builder.compile(checkpointer=checkpointer, store=store)Subgraphs are not only useful for multi-agent. It’s also a way to split commands, allocate a separate approval-flow or keep a private internal state at a specific module without clogging the parent graph.
Source-complete notesscenario
scenario
- document review subgraph
- specialized planner subgraph
- private agent memory inside subgraph
If human confirmation is part of the business flow, don’t encode it through the strange external ifs around the graph. It is much cleaner to make explicit approval node, which pauses execution and then continues the graph through Command(resume=..)
Source-complete notesexample
sqlfrom langgraph.types import interrupt
def review_node(state: AgentState):
approved = interrupt({
"question": "Approve answer?",
"draft": state["structured_answer"],
})
return {"approved": bool(approved)}Serving Layer and Runtime Action
HTTP handlers are better made thin. The basic orchestration around calling a graph can live in the service: form a state update, collect a 'thread id', if necessary 'checkpoint id', transmit a runtime context, and select 'invoke' versus 'astream'.
Source-complete notestypical
textconfig = {"configurable": {"thread_id": thread_id}}
context = Context(user_id=user_id, tenant_id=tenant_id, can_send_email=False)
result = graph.invoke(
{"messages": [{"role": "user", "content": text}]},
config=config,
context=context,
)If a project has interrupts, you need a separate endpoint or service method that accepts ‘thread id’ and ‘resume’ payload. It’s not “the same invoke” because it’s important to continue a particular thread rather than start a new run.
Source-complete notesresume path
textconfig = {"configurable": {"thread_id": thread_id}}
result = graph.invoke(
Command(resume=resume_payload),
config=config,
)For production UI, it is usually useful to stream both message chunks and state updates. Then you can show the answer for tokens, and understand which node the graph is now on, and catch " interrupt " in time.
Source-complete notesexample
textasync for mode, chunk in graph.astream(
inputs,
config=config,
stream_mode=["messages", "updates", "custom"],
):
...
In interactive flows, subgraphs are often useful.If the service stores execution history or can replay/debug, the user or internal tooling may need not only “thread id”, but also “checkpoint id”. With them, you can get a specific snapshot or replay execution after a given point.
Source-complete notesdebug / replay
textstate = graph.get_state({
"configurable": {"thread_id": thread_id}
})
history = list(graph.get_state_history({
"configurable": {"thread_id": thread_id}
}))
Checkpoint id becomes useful when the tooling or API can address a particular snapshot.If the task is simple, the service can keep the agent object from create agent. If the orchestration is custom, the service holds compiled 'StateGraph'. In production code, both forms are often found in different modules of the same project.
Source-complete notestable
| Form | When appropriate. |
|---|---|
| `create_agent` | typical agent loop |
| compiled graph | Custom branching / interrupts / subgraphs |
Memory, Persistence and Operator
For local development, “InMemorySaver” is enough. For production, official materials recommend DB-backed persistence. The most useful thing here is to centralize the creation of a checkpointer, rather than scattering it across different build functions.
Source-complete notesmatrix
matrix
- local dev → InMemorySaver
- prod → database-backed checkpointer
- thread id is mandatory for persistence
Checkpointer keeps the graph state on super-steps and threads. A store is needed when information needs to survive conversations and be accessible through different threads. Usually, both resources are initialized side by side, but solve different tasks.
Source-complete notescompile
textgraph = builder.compile(
checkpointer=checkpointer,
store=store,
)In both LangChain agents and LangGraph workflows, short memory lives in state/messages. With the growth of the thread, you either need to trimm the messages or summarize the story. Cost, latency and noise increase in the prompt context.
Source-complete notesoptions
text1. trim_messages utility
2. before model middleware that cuts history
3. summarize old messages and store summary
4. custom filtering strategyThe official materials separately remind: for database-backed persistence, you need to prepare a database scheme. It’s better to frame it as an explicit deploy/startup step, rather than hoping that “it will appear” during the first query.
Source-complete notesrule
rule
- Setup/migrations separately
- Lazy schema creation in the middle of the user request
Interrupt resumes node from the beginning, not from a specific line. Therefore, everything that has been done before "interrupt()" can be done again. If there was a non-idempotent side effect, you will get duplicates, re-entries, or inconsistency.
Source-complete notessafer
safer
- upsert before interrupting
- side effects after interruption
- create row before interrupt
Tests, Skeleton and Anti-Patterns
A good set of tests here is layered: unit tests tools and middleware, unit tests node functions, integration tests compiled graph, separate tests for interrupt/resume and, if necessary, snapshot-state tests. Otherwise, the error in routing will easily hide behind “the answer seems to have turned out.”
If you compress everything to the most useful minimum, there are a few mandatory files. This is already a good starter for a real service, not for a laptop demo.
It is not the library that usually breaks down, but the architecture around it. The most common problems are: giant graph.py, business logic inside route functions without tests, tools with direct access to globals, an HTTP layer that itself builds half the state, and the lack of an explicit resume path.
Source-complete notesred flags
red flags
- whole project in one module
- tool = 200 lines of raw SDK code
- No interrupt/resume test
- No ownership of the state schema
Production 'LangChain' + 'LangGraph' project is more conveniently understood as a regular backend with clear layers: contracts, components, orchestration, runtime. When ‘state’ and ‘context’ are defined explicitly, ‘tools’ use ‘ToolRuntime’ and the graph is assembled in a separate ‘build.py’, things become much less ‘magical’ and much more engineering.
Source-complete notestakeaway
takeaway
- State - Main Contract
- tools/middleware – layer of components
- graph/build — wiring
- service/api — invoke, stream, resume
No dead end
Keep moving through the map.
Continue in sequence, switch to a related guide, or return to the seven-track learning map.