Production LangChain + LangGraph: Code Structure & Runtime

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.

THE RIGHT PLAYING

contracts
  → state / context / schemas / settings

components
  → model / tools / prompts / middleware / clients

graph
  → nodes / routes / subgraphs / compile

runtime
  → api / invoke / stream / resume / persistence / tests

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.

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.

SUGGESTED TREE

app/
  agent/
    contracts/
      state.py
      context.py
      schemas.py
      settings.py
    components/
      models.py
      prompts.py
      tools.py
      middleware.py
      clients.py
    graph/
      nodes.py
      routes.py
      subgraphs.py
      build.py
    runtime/
      service.py
      stream.py
      persistence.py
  api/
    chat.py
    resume.py
  tests/
    test_tools.py
    test_nodes.py
    test_graph.py

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 ”.

HTTP request
→ build input state
→ build config {"configurable": {"thread_id": ...}}
→ build runtime context
→ invoke / astream
→ inspect final state or __interrupt__
→ return HTTP response

If 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.

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.

from 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 | None

Context 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.

from dataclasses import dataclass

@dataclass
class Context:
    user_id: str
    tenant_id: str
    can_send_email: bool

Structured 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.

from pydantic import BaseModel

class AnswerSchema(BaseModel):
    answer: str
    citations: list[str]
    needs_human_review: bool

Settings 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.

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.

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.

from 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.

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.

from 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.

@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-a

It 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.

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.

def retrieve_node(state: AgentState):
    docs = retriever.search(state["user_query"])
    return {"retrieved_docs": docs}

def retrieve_node(state: AgentState):
    state["retrieved_docs"] = retriever.search(...)
    return state

When 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.

from 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 END

This 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.

from 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.

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=..)

from langgraph.types import interrupt

def review_node(state: AgentState):
    approved = interrupt({
        "question": "Approve answer?",
        "draft": state["structured_answer"],
    })
    return {"approved": bool(approved)}

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'.

config = {"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.

config = {"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.

async 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.

state = 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.

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.

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.

graph = 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.

1. trim_messages utility
2. before model middleware that cuts history
3. summarize old messages and store summary
4. custom filtering strategy

The 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.

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.

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.”

TEST PYRAMID

tool tests
  ↓
node tests
  ↓
route / middleware tests
  ↓
compiled graph integration tests
  ↓
interrupt / resume / replay tests

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.

MINIMAL SKELETON

contracts/state.py
contracts/context.py
contracts/schemas.py

components/models.py
components/tools.py
components/middleware.py

graph/nodes.py
graph/routes.py
graph/build.py

runtime/service.py
runtime/persistence.py
api/chat.py

tests/test_graph.py

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.

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.