LangChain and LangGraph — agents, graphs and the infrastructure around them

Hi!

A language model does exactly one thing: it takes text and returns text. Nothing else. It does not remember what you said a minute ago, and it cannot go to your database or to the internet on its own.

Take a simple example: a user asks an assistant what the weather is. The model does not know. For it to answer, this has to happen:

1. Tell the model up front that it has a "get the weather" function, and describe the arguments.
2. Take the reply in which the model asks for that function to be called, and parse it.
3. Call the function from your own code.
4. Send the result back and ask the model to answer again.

That is a loop, not a single request. Add conversation memory so the next question is understood in context; output as it is generated, so the user is not staring at an empty screen for forty seconds; a retry when the provider fails; a record of what the model actually saw, without which a wrong answer cannot be investigated. The amount of code around one simple action adds up.

LangChain and LangGraph are libraries where all of that is already written. The first gives you the parts: models, tools, prompts, the agent loop. The second runs the process: it holds state, branches the steps, saves progress and emits a stream of events.

This article is about using them: what is responsible for what, what an agent is assembled from, when you need a graph instead of an agent, how memory and streaming work, what LangSmith and the ready-made server are for, and what changed in the last major release.

Examples are in Python. Package versions at the time of writing: langchain 1.3.14, langchain-core 1.5.3, langgraph 1.2.10, langchain-classic 1.0.8. Minimum Python is 3.10, and Node 20 for the JavaScript version.

Glossary

The terms below appear throughout, so here are short definitions; each one is unpacked in its own section.

Model — a provider's language model: OpenAI, Anthropic, Google, or a local one through Ollama. In code it is an object with a call method and a streaming call method.

Message — a unit of the conversation. Four types: SystemMessage (instructions), HumanMessage (user input), AIMessage (the model's reply along with any calls it requests), ToolMessage (the result of running a function).

Tool — a function in your code whose description is handed to the model. The model does not run it: it returns a request saying "call this with these arguments", and your code runs it.

Agent — a loop: call the model, run the tools it asked for, return the results, repeat. The loop ends when the model answers without requesting tools.

Graph — a process described explicitly as steps and transitions. You need it when you, not the model, decide the order of actions.

Node — one step of the graph, an ordinary function. Edge — a transition between steps.

State — the data that nodes read and add to as the work proceeds.

Reducer — the rule by which an update from a node is merged into what is already in state.

Super-step — one tick of graph execution: everything scheduled runs, then the updates land in state, then the next tick is scheduled.

Checkpointer — the mechanism that saves state. It is what makes turn two remember turn one, and what lets a crashed run continue from where it stopped.

Thread — the identifier the state is saved under. One user, one conversation, one thread.

Store — long-term memory across all threads: preferences, accumulated facts.

Context — immutable per-run data: who the user is, which tenant, which flags.

Runtime — the object through which a node or tool reaches context, store and the event stream.

Middleware — code that runs around the steps of the agent loop: before the model call, after it, around a tool call.

Interrupt — pausing a run to wait for a human.

Streaming — incremental output: tokens and events arrive as they appear.

What the ecosystem is made of

One brand ships several products with similar names. Let us sort them out.

The layers depend upwards from the bottom:

create_agent from LangChain returns a compiled LangGraph graph — so everything the runtime can do to a graph it can do to an agent: save state, stream it, pause it, nest it inside a bigger graph as an ordinary node. You do not have to choose between an agent and a graph; they are the same thing at different heights of abstraction.

Packages

The library is split across many packages, and you do not need all of them.

Model providers

The library itself is tied to no vendor: a provider plugs in as its own package, and switching between them is a change to one initialisation string rather than a rewrite. Provider packages are versioned independently of the core.

A string like "openai:gpt-5.5" only selects an integration that is already installed — the vendor package has to be in your dependencies.

Beyond models, the other external services plug in the same way: vector stores (langchain-chroma, langchain-postgres for pgvector, langchain-pinecone, langchain-qdrant, langchain-weaviate), web search (langchain-tavily), and anything without a dedicated package lives in langchain-community.

Versioning

There is one thing worth knowing. The main packages — langchain, langchain-core, langgraph — follow semantic versioning: breaking changes only happen at a major, and deprecated features keep working with a warning across the whole line. Hence the practical floor of >=1.0,<2.0. langchain-community, on the other hand, does not follow semantic versioning, and is usually pinned to an exact minor series.

Which layer to take

Check top to bottom and stop at the first match:

The last item applies more often than you would think: an agent loop around a single call adds both tokens and latency.

Installing and the first call

pip install "langchain>=1.0,<2.0" "langchain-core>=1.0,<2.0" "langgraph>=1.0,<2.0" langchain-openai

langchain-core is installed explicitly. It would arrive transitively too, but then its version is governed by something other than your dependency file.

The provider package is mandatory: a string like "openai:gpt-5.5" resolves to a specific integration but does not install it — it is a lookup among what is already there.

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-5.5", temperature=0, timeout=30)

response = await model.ainvoke("Why do parrots have colourful feathers?")
print(response.text)

The string format is provider:model. Examples from the documentation: openai:gpt-5.5, anthropic:claude-sonnet-4-6, google_genai:gemini-2.5-flash-lite. Instead of a string you can construct the provider object directly, which opens up parameters the shared interface does not carry.

response.text is a property, not a method. In the previous version it was a call with parentheses.

Sync and async calls

Every method comes in two forms: invoke and ainvoke, stream and astream. For scripts and notebooks either will do. For a web service the difference matters: a synchronous call blocks the whole event loop, and while it waits for the model every concurrent request waits with it. In a service, only the async forms belong on the request path.

There is also abatch — parallel processing of several independent prompts. It is useful for offline work such as running a dataset, not for a single user's turn.

Model parameters

One caveat about rate_limiter: the built-in InMemoryRateLimiter works within a single process. With five replicas the effective rate is five times the configured one, and a cluster-wide budget needs a shared limiter.

About reasoning_effort: on reasoning models the depth of thinking is set with this parameter. A "think step by step" instruction in the prompt runs alongside the model's own reasoning machinery rather than adding to it. Prompting is a large topic and I have a separate article about it.

Fallbacks

model = init_chat_model("openai:gpt-5.5").with_fallbacks([init_chat_model("openai:gpt-5-nano")])

The switch happens on an exception from the primary model. It is how you survive a provider blip without handing the user an error.

Messages: what comes back

There are four message types: SystemMessage carries instructions, HumanMessage the user input, AIMessage the model's reply along with tool call requests and metadata, ToolMessage the result of a call going back to the model.

The model's reply has two views of the same data:

• content — the provider's raw payload. A string, or its own list of dicts, preserved as is.
• content_blocks — a typed parse over that same data, identical across providers.

Why the second one exists. Providers return semantically identical things in different shapes: one delivers reasoning under the name thinking, another as reasoning. In the parsed view both are a block of type reasoning.

for block in response.content_blocks:
    if block["type"] == "reasoning":
        log_reasoning(block)
    elif block["type"] == "text":
        show_to_user(block)
    elif block["type"] == "tool_call":
        schedule(block)

The block types cover text and reasoning; images, audio, video and files; tool calls and their fragments during streaming; tools executed on the provider's side; and a separate non_standard type for anything that does not fit the taxonomy.

The practical consequence: code that reads content_blocks survives a change of provider. Code that picks apart content by hand is tied to one vendor.

Multimodal input

from langchain.messages import HumanMessage

message = HumanMessage(content=[
    {"type": "text", "text": "What's in this image?"},
    {"type": "image", "url": "https://example.com/photo.png"},
])

Trimming history

Conversations grow, context windows are finite. The simple approach is deterministic trimming:

from langchain.messages import trim_messages

trimmed = trim_messages(messages, max_tokens=8000, strategy="last", token_counter=model)

The smarter approach is summarising the older part of the history, available to agents as ready-made middleware. The difference is that trimming is predictable and loses old material wholesale, while summarisation keeps the meaning but rewrites what the assistant remembers.

Token accounting

from langchain_core.callbacks import UsageMetadataCallbackHandler

callback = UsageMetadataCallbackHandler()
await model.ainvoke("Hello", config={"callbacks": [callback]})
callback.usage_metadata      # {'input_tokens': 8, 'output_tokens': 10, ...}

The same numbers are on every reply in usage_metadata. This is the basis for cost accounting and user quotas.

Prompts

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "You help make sense of build logs. Keep answers short."),
    ("human", "{question}"),
])

chain = prompt | model

ChatPromptTemplate assembles a list of messages with variable substitution. Roles stay separate, and that matters: a provider treats a system instruction and user input differently, with different weight and different levels of trust. Glue them into one string and the user's text ends up where the model expects your instructions — that is both a quality regression and an open door for prompt injection.

The vertical bar is composition: the left object's result is passed to the right one. The resulting chain itself supports ainvoke, astream and abatch, and can be part of something bigger.

Structured output

Often what you need from the model is not text but data: the category of a request, extracted fields, a routing decision.

from pydantic import BaseModel, Field

class BuildFailure(BaseModel):
    module: str
    reason: str
    is_flaky: bool = Field(description="Does this look like a flaky test")

structured = model.with_structured_output(BuildFailure)
result = await structured.ainvoke(log_text)   # -> BuildFailure(...)

Pydantic models, TypedDict and a bare JSON Schema are all accepted. How it is implemented underneath — the provider's native mode, function calling or JSON mode — is chosen from the model's capabilities.

Asking for JSON in the prompt and parsing prose was removed in the first major version as unreliable, and reintroducing it by hand is not worth it: a parsed schema fails immediately, while parsing prose breaks quietly and shows up further down the code.

Tools

A tool is a function in your code whose description is sent to the model along with the prompt. The model does not execute it: it returns a call request with arguments, your code executes it, and the result goes back to the model.

from langchain.tools import tool

@tool
def search_listings(city: str, max_price: int = 1_000_000) -> str:
    """Search active property listings in a city.

    Args:
        city: City name, e.g. "Odesa"
        max_price: Upper bound in USD
    """
    return render(query(city, max_price))

Three elements here are load-bearing:

Type hints form the schema the model sees. Without them an argument has no type and its value has to be guessed.

The docstring is part of the prompt. It is the only thing explaining to the model when this tool applies. How it is worded directly decides whether the tool gets called at the right moment.

A snake_case name. Some providers reject other shapes.

When hints are not enough — you need a constrained set of values, per-field descriptions or validation — declare an explicit argument schema:

from pydantic import BaseModel, Field
from typing import Literal

class WeatherInput(BaseModel):
    location: str = Field(description="City name or coordinates")
    units: Literal["celsius", "fahrenheit"] = "celsius"

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius") -> str:
    """Get current weather."""
    ...

ToolRuntime — reaching the environment

A tool often needs data that is not in its arguments: who the user is, what state the conversation is in, where to write long-term memory. For that you add a ToolRuntime parameter to the signature — the framework injects it, and it does not appear in the schema the model sees.

from langchain.tools import tool, ToolRuntime

@tool
def get_account_info(runtime: ToolRuntime[UserContext]) -> str:
    """Get the current user's account information."""
    user_id = runtime.context.user_id
    return describe(lookup(user_id))

Through it you get: runtime.state — the agent's current state including message history; runtime.context — the immutable per-run context; runtime.store — long-term memory; runtime.stream_writer — the writer for the event stream; runtime.tool_call_id — the id of this call; runtime.execution_info — thread and run identifiers; plus the run configuration.

A point that matters for security: user identity, tenant id and secrets come from runtime.context, never from arguments. Arguments are filled in by the model, and the model can be influenced by the text of a request — a user id in a tool schema means someone else's can be substituted simply by asking for it in the chat.

What a tool returns

from langchain.messages import ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.types import Command

@tool
def set_language(language: str, runtime: ToolRuntime) -> Command:
    """Set the language to answer in from now on."""
    return Command(update={
        "preferred_language": language,
        "messages": [ToolMessage(content=f"Language switched to {language}.",
                                 tool_call_id=runtime.tool_call_id)],
    })

When a tool returns a Command, the ToolMessage is yours to include in the update. Without it the provider sees a tool call with no result and rejects it on the next turn.

return_direct ends the loop and hands the tool's output back as the final answer. It fits when the output is the answer, and does not when the model should comment on it or combine it with something else.

Progress from a tool

A long-running tool can report how it is going, and that reaches the event stream:

@tool
def index_documents(folder: str, runtime: ToolRuntime) -> str:
    """Index a folder of documents."""
    writer = runtime.stream_writer
    writer({"type": "progress", "done": 0, "total": 100})
    ...

Tool errors

By default an exception from a tool aborts the whole run. For some cases that is not what you want: a search that found nothing is information the model can work with.

from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage

@wrap_tool_call
def handle_tool_errors(request, handler):
    try:
        return handler(request)
    except Exception as exc:
        return ToolMessage(content=f"The tool failed: {exc}",
                           tool_call_id=request.tool_call["id"])

There are also shipped middleware for this — ToolErrorMiddleware turns an exception into a message, ToolRetryMiddleware retries with backoff. Inside a hand-built graph the same job is done by the handle_tool_errors flag on ToolNode.

The cases are worth separating, because they are treated differently: a transient network failure or a 429 — retry; nothing found or a bad argument — return a message and let the model try differently; missing data only a human has — an interrupt; a genuine defect — let the exception out.

Separately: the text of an internal exception should not go into a ToolMessage on a user-facing path. It goes to the model, and from there it can reach the user along with a stack trace and connection details.

Provider-side tools

Some tools are executed by the provider itself — web search, a code interpreter. They appear in the reply as server_tool_call and server_tool_result blocks, are billed by the provider and never run your code. It follows that your timeouts, retries and logging do not apply to them.

There is also the opposite case — headless tools: the schema is declared, the implementation is not. The run pauses with a description of the action, your application performs it wherever it belongs (in the browser, in another service) and resumes with the result.

Agent

from langchain.agents import create_agent

agent = create_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[search_listings, send_email],
    system_prompt="You are a real-estate assistant. Be short and to the point.",
)

result = agent.invoke({"messages": [{"role": "user", "content": "What's in Odesa under 200k?"}]})
print(result["messages"][-1].content)

What has just been assembled:

The loop ends when the model answers without calling tools. If it requested several tools in one turn, they run in parallel and each returns its own ToolMessage.

The loop itself is small and fixed. The places where you would want to intervene are moved into middleware — that is, an agent is configured through hooks around the loop rather than through constructor arguments.

create_agent parameters

Examples set the model with a string; production code usually takes it from configuration together with the timeout, retries and the fallback chain.

The result is a dict

result["messages"][-1].content    # the answer text
result["structured_response"]     # the parsed schema, if response_format was set

What comes back is the final state, not a message, so something like result.content raises AttributeError.

Response format — a schema inside the loop

from pydantic import BaseModel

class Weather(BaseModel):
    temperature: float
    condition: str

agent = create_agent("openai:gpt-5.5", tools=[weather_tool], response_format=Weather)

result = agent.invoke({"messages": [{"role": "user", "content": "Weather in Odesa?"}]})
result["structured_response"]   # Weather(temperature=27.0, condition='clear')

The schema is applied inside the loop, not as a second model call on top of a finished answer. There are two strategies: ProviderStrategy uses the provider's native structured output (more reliable, fewer tokens, not available everywhere), ToolStrategy emulates it through a tool call (works on any model with tool support and lets you union several alternative schemas). Pass the schema without naming a strategy and LangChain reads the model profile and picks one.

ToolStrategy takes handle_errors — what to do when validation fails: by default the request is retried with the error text handed to the model. You can set your own message, restrict retries to particular exception types, pass a handler function, or turn retries off entirely.

Profile — what a model can do

model.profile
# {'max_input_tokens': 400000, 'tool_calling': True, 'reasoning_output': True, 'multimodal': True, ...}

A description of a specific model's capabilities. Use it to check tool support before binding, or to size context against the window, instead of branching on model names. The same profile is what lets response_format pick its strategy.

Context and state are different things

from dataclasses import dataclass

@dataclass
class Context:
    user_id: str

agent.invoke(
    {"messages": [{"role": "user", "content": "What's my balance?"}]},
    config={"configurable": {"thread_id": "t-1"}},
    context=Context(user_id="user-123"),
)

User identity goes into context, not state: state is checkpointed and replayed, so resuming an old conversation would drop yesterday's user into a fresh run.

Memory between turns

from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(model=..., tools=tools, checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "conversation-42"}}

agent.invoke({"messages": [{"role": "user", "content": "My name is Alice"}]}, config=config)
agent.invoke({"messages": [{"role": "user", "content": "What's my name?"}]}, config=config)

Both parts are needed: a checkpointer and a thread_id. A checkpointer without a thread_id saves nothing and raises nothing — the symptom is an assistant that does not remember the previous turn.

InMemorySaver lives until the process restarts and is meant for development; production has Postgres.

Recursion limit

The loop is bounded by a number of super-steps, 25 by default. Exhausting it raises.

result = agent.invoke(payload, config={"recursion_limit": 40})

This is protection against an endless loop. If what you want is a product limit — no more than five searches per conversation, say — that is what ModelCallLimitMiddleware and ToolCallLimitMiddleware are for: they finish cleanly instead of throwing.

Middleware — six points of control

Middleware is code that runs around the steps of the agent loop. Everything is configured through it: swapping the model on the fly, limits, retries, redacting personal data, pausing for a human, assembling a dynamic prompt.

As layers around the loop it looks like this:

There are two shapes. Node-style hooks (before_*, after_*) receive state and return an update to it — they observe and add. Wrap-style hooks (wrap_*) sit in the call path itself, receiving the request and a continuation function: they can retry, override parameters, or return their own result instead of the real one.

Firing order

For a list of three middleware the order is:

before_* hooks run first to last, after_* last to first, and wrap_* nest inside each other with the first middleware in the list outermost.

Two practical consequences follow. A retry that should wrap everything else goes first. A check that must see the final answer also goes near the start of the list — on the way back out it runs last.

How it is written

A single hook is a decorator:

from langchain.agents.middleware import before_model, wrap_model_call
from langchain.messages import AIMessage

@before_model(can_jump_to=["end"])
def stop_long_conversation(state, runtime):
    if len(state["messages"]) >= 50:
        return {"messages": [AIMessage("This conversation has grown too long.")], "jump_to": "end"}
    return None

@wrap_model_call
def upgrade_for_experts(request, handler):
    if request.runtime.context.tier == "expert":
        return handler(request.override(model=strong_model, tools=advanced_tools))
    return handler(request)

ModelRequest carries the messages, the model, the tool set, the system message, state and runtime. It is not mutated: request.override(...) returns a modified copy, and that is what you pass on.

Node-style hooks can skip the rest of the loop and jump straight to the end, to the tools or to the model. Such a jump is declared up front with a list of possible targets — the can_jump_to parameter, as above; in the class form the same thing is declared with the @hook_config decorator on the method.

Wrap-style hooks must return a value: a generator with yield inside raises at runtime.

Several hooks at once, your own state or your own tools mean a class:

from langchain.agents.middleware import AgentMiddleware, AgentState
from typing_extensions import NotRequired

class UsageState(AgentState):
    model_calls: NotRequired[int]

class UsageMiddleware(AgentMiddleware):
    state_schema = UsageState
    tools = [reset_usage]

    async def aafter_model(self, state, runtime):
        return {"model_calls": state.get("model_calls", 0) + 1}

Node-style hooks have async variants prefixed with a — those are the ones to implement on a request path.

Dynamic prompt

system_prompt is static. Anything that depends on state, context or retrieved documents is assembled in a separate hook:

from langchain.agents.middleware import dynamic_prompt

@dynamic_prompt
def prompt_for_tier(request):
    tier = request.runtime.context.tier
    return f"You are a support agent. The customer is on the {tier} plan."

agent = create_agent(model=..., tools=tools, middleware=[prompt_for_tier], context_schema=Context)

The catalogue of shipped middleware

Most common requirements are already covered.

Context and memory. SummarizationMiddleware summarises history as it approaches a token budget, with settings for the trigger, the tail that is kept and how tokens are counted. ContextEditingMiddleware clears older tool results to reclaim context.

Safety and control. HumanInTheLoopMiddleware pauses for human approval. PIIMiddleware detects and redacts personal data by type, with a choice of strategy and of what it applies to: input, output, tool results. ModelCallLimitMiddleware and ToolCallLimitMiddleware cap the number of calls within a thread or a single run, with a choice of what happens at the limit.

Reliability. ModelRetryMiddleware and ToolRetryMiddleware retry with exponential backoff. ToolErrorMiddleware turns exceptions into messages the model can act on. ModelFallbackMiddleware switches to other models when the primary one fails.

Capability. TodoListMiddleware gives the agent a task list to plan with. LLMToolSelectorMiddleware pre-selects the relevant tools with a cheap model when there are too many. SubAgentMiddleware delegates subtasks to isolated subagents. FilesystemMiddleware and FilesystemFileSearchMiddleware provide a filesystem over a pluggable backend and search over it. ShellToolMiddleware gives a persistent shell session. RubricMiddleware adds self-evaluation against a rubric with iterations. LLMToolEmulator fakes tool execution with a model, for tests.

Provider-specific. Prompt caching and built-in tools on Anthropic, caching on Bedrock, content moderation on OpenAI.

Two warnings about combining them. A fallback mechanism exists both on the model and in middleware — one of them should be doing the work, otherwise you get two layers of switching stacked on each other. And SummarizationMiddleware rewrites history, that is, it changes what the assistant remembers about the conversation: switch it on deliberately.

Middleware or a node

Both run code around a model call. The line goes like this: middleware when the behaviour is cross-cutting and unrelated to the content of the conversation (retries, redaction, limits, tracing, prompt assembly). A node when the step is part of the process: a routing decision, a distinct phase of work, something you would draw on a diagram.

LangGraph: when you set the order of steps

An agent fits while the model chooses the sequence of actions. A different class of task looks like this: always load the user profile first, then classify the request, then either ask a clarifying question or answer, and after the answer write a log entry. Here the order is known in advance, and there is no need to describe it in words in a prompt — it can be described in code.

A graph is the way to describe the process explicitly. Deterministic steps stay code, decisions stay with the model, and together they form a diagram you can draw and discuss.

Here is what a typical chat pipeline looks like as a graph. The labels on the right say what each step is: ordinary code or a model call.

This is where the difference from an agent shows: the order of steps is explicit. The model decides what kind of question it was asked, not what order to do the work in.

Three entities

State is the shared data all the steps work with. Nodes are the steps themselves, ordinary functions, each returning a partial update to state. Edges are the transitions: what runs next.

Execution advances in super-steps. Everything scheduled for the current tick runs — in parallel where the graph allows; then the updates are merged into state; then the next tick is scheduled. The graph is compiled before it runs.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    question: str
    answer: str

def answer(state: State) -> dict:
    return {"answer": f"Answer to {state['question']}"}

graph = (
    StateGraph(State)
    .add_node("answer", answer)
    .add_edge(START, "answer")
    .add_edge("answer", END)
    .compile()
)

START and END are special markers: the first is the entry point where user input arrives, the second is the exit. An edge back into START is invalid; if you need a loop, route it through a named node.

The state schema can be a TypedDict, a Pydantic model or a dataclass. The first is the usual choice as the lightest one.

Reducers — how updates are merged

Every state key has a reducer: the rule by which an update from a node is combined with what is already in state.

from typing import Annotated
import operator
from langgraph.graph.message import add_messages

class State(TypedDict):
    name: str                                     # default: overwrite
    findings: Annotated[list[str], operator.add]  # append
    messages: Annotated[list, add_messages]       # append, deduplicated by id

Here is what happens on a single super-step when two nodes have run in parallel:

The default reducer overwrites the value. So a list without an explicit reducer, written by two nodes, keeps only one of the values — with no error and no warning. Any accumulating field must have a reducer.

For messages there is add_messages: it appends, drops duplicates by id and converts dicts into message objects. There is also a ready-made MessagesState class that includes it.

The second way to bypass reducers is to return the whole state from a node:

def good(state: State) -> dict:
    return {"answer": "..."}          # only what changed

def bad(state: State) -> State:
    state["answer"] = "..."           # mutation goes around the reducers
    return state

You can also give the graph separate input and output schemas, keeping internal working fields out of the public interface.

Nodes

A node takes state, and if needed also a RunnableConfig (which carries the thread_id, tags and configurable values) or a Runtime (context, store, the stream writer). Async nodes are plain async def with the same signatures.

A handy pattern for a service: the node is created by a factory function that closes over its dependencies — the database client, the HTTP session, the configuration — and returns the node function. Dependencies then do not become globals, and the node is easy to test on its own.

Routing

def route(state: State) -> Literal["clarify", "generate"]:
    return "clarify" if state["needs_input"] else "generate"

builder.add_conditional_edges("assess", route, {"clarify": "clarify", "generate": "generate"})

The third argument maps return values to node names. It is optional, but with it the graph renders correctly and the set of reachable nodes becomes explicit.

When you need to update state and choose the next step at once, the node returns a Command:

from langgraph.types import Command
from typing import Literal

def triage(state: State) -> Command[Literal["escalate", "resolve"]]:
    if state["severity"] > 3:
        return Command(update={"assigned": "oncall"}, goto="escalate")
    return Command(update={"assigned": "bot"}, goto="resolve")

The Literal annotation listing the reachable nodes is needed for rendering and for validating the targets.

An important detail of the behaviour: Command adds a dynamic edge, it does not replace a static one. If the same node also has an ordinary edge, both destinations run. From the outside this looks like duplicated work.

From a subgraph a command can hand control to the parent graph — that is what Command(goto=..., graph=Command.PARENT) is for.

Send — fan-out

When the number of branches is only known at runtime:

from langgraph.types import Send

class State(TypedDict):
    topics: list[str]
    drafts: Annotated[list[str], operator.add]     # the accumulator is mandatory

def fan_out(state: State):
    return [Send("write_draft", {"topic": topic}) for topic in state["topics"]]

builder.add_conditional_edges(START, fan_out, ["write_draft"])
builder.add_edge("write_draft", "synthesise")

Each Send carries its own private input to one invocation of the worker — this is how "scatter and gather" is built:

The results are collected through a reducer; without one on the collecting field, only the last worker to finish survives.

Subgraphs

A compiled graph is a runnable object, so it can be a node in another graph:

builder.add_node("research", research_graph)

Shared state keys flow through automatically. If the state schemas differ, you need an adapter node that converts on the way in and out.

Per-node controls

from langgraph.types import RetryPolicy, CachePolicy
from langgraph.cache.memory import InMemoryCache

builder.add_node("fetch", fetch, retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0))
builder.add_node("embed", embed, cache_policy=CachePolicy(ttl=300))
graph = builder.compile(cache=InMemoryCache())

From the same release: RunControl.request_drain(), a cooperative stop that asks a run to halt at the next safe point and leaves a checkpoint to continue from. It is the proper way to survive a deploy or a container being stopped mid-run.

The graph's recursion_limit is the same as the agent's and is set in the run configuration. There is also a RemainingSteps state field a node can look at to wind down early instead of hitting the wall.

Checkpointer and store: memory and durability

There are two kinds of memory, solving different problems.

Neither is switched on by default.

Checkpointers

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "conversation-1"}}

await graph.ainvoke({"messages": ["Hello"]}, config)
await graph.ainvoke({"messages": ["And again"]}, config)   # sees the first turn

Two operational details. .setup(), which creates the tables, is a deploy-time operation rather than an application-start one: changing the schema on start races itself across replicas. And keep the thread_id under 255 characters for Postgres — the column is bounded.

Inspecting history and rewinding

snapshot = await graph.aget_state(config)                        # current state and what comes next
history = [s async for s in graph.aget_state_history(config)]    # newest entries first

past = history[-2]
await graph.ainvoke(None, past.config)                           # replay from that point

fork = await graph.aupdate_state(past.config, {"messages": ["edited"]})
await graph.ainvoke(None, fork.config)                           # continue down a new branch

Calling with None instead of input means "resume from the checkpoint". Debugging is built on the same thing: rewind to the step you need, adjust state, and watch what happens next.

One quirk: update_state passes through the reducers. On an appending field it appends where you meant to replace. For replacement there is the Overwrite wrapper:

from langgraph.types import Overwrite

await graph.aupdate_state(config, {"items": Overwrite(["C"])})

Checkpoint retention

Checkpoints are created one per super-step per thread and are not removed on their own. For a long-lived product that means a permanently growing table, so decide the retention policy — how long a thread stays resumable and what deletes the rest — in advance.

Store

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
graph = builder.compile(checkpointer=checkpointer, store=store)

The operations are put, get, search, delete. The namespace is a hierarchical tuple like ("users", user_id, "preferences"), and it is also the isolation boundary. Build it from trusted context, not from text that came from the model: a namespace without a user or tenant id means the data is visible to everyone.

Give the store an embedding function and dimensions through IndexConfig and the search becomes semantic rather than exact. The production implementation is PostgresStore.

Inside a graph, reach the store through the runtime rather than a global:

def recall(state, runtime: Runtime):
    item = runtime.store.get(("users", runtime.context.user_id), "preferences")
    ...

Durability — write modes

How often state is written to disk. Passed at call time:

await graph.ainvoke(payload, config=config, durability="async")

The choice depends on what a lost step costs: if steps send email or take payments, "sync" fits; an analytical pass over a long conversation is fine on "exit".

Replay on resume

A general rule of the runtime: a resume runs the node from its very beginning, and only completed work already in the checkpoint is skipped. Two consequences that affect the code:

• Side effects performed before the stopping point happen a second time.
• Non-deterministic values — timestamps, random ids — come out different on replay. Compute them in a node of their own so the value lands in the checkpoint and is substituted from there.

Human-in-the-loop

Sometimes a run has to stop and wait for a person: confirm a send, fix a draft, answer a clarifying question. That is what interrupt is for.

from langgraph.types import interrupt, Command

def review(state):
    decision = interrupt({"draft": state["draft"], "question": "Publish this?"})
    return {"approved": decision == "yes"}

The call stops the run, saves everything into a checkpoint and surfaces the payload. The run stays suspended — for a second or a week — until it is resumed with Command(resume=...); then interrupt returns the value passed in.

Note the second appearance of review in the diagram: a node with a pause runs from its very beginning on resume. That is a topic of its own, and we come back to it below.

A checkpointer and a thread_id are mandatory here: without them there is nowhere to save the pause.

There is also a static variant — interrupt_before and interrupt_after at compile time, stopping at a node boundary. It is convenient for debugging and stepping through but carries no payload, so product approval flows use the ordinary interrupt.

Approving tool calls

For the most common case — confirming a dangerous call before it runs — there is shipped middleware:

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[write_file, execute_sql, read_data],
    checkpointer=checkpointer,
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={
            "write_file": {"allowed_decisions": ["approve", "edit", "reject"]},
            "execute_sql": {"allowed_decisions": ["approve", "reject"]},
            "read_data": False,
        },
        description_prefix="Awaiting approval",
    )],
)

True switches on the interrupt with default decisions, False lets the call through without asking, and a dict configures which decisions the reviewer may take.

await agent.ainvoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config, version="v2")

When the model requests several guarded tools in one turn, they are presented together, and the list of decisions must be in the same order as the presented actions.

The interrupt can be made conditional with the when parameter: a function looks at the call arguments and decides whether approval is needed. That way only the risky cases are reviewed — writing outside the workspace, say — while the rest pass without questions.

Side effects and the stopping point

The same replay rule as above, with visible consequences:

def send_and_confirm(state):
    send_email(state["draft"])              # sends again on resume
    interrupt({"sent": True})

def confirm_then_send(state):
    decision = interrupt({"draft": state["draft"]})
    if decision == "approve":
        send_email(state["draft"])          # after the pause — exactly once
    return {"sent": decision == "approve"}

Three rules follow: side effects go after the interrupt or are made idempotent; non-deterministic values move into a node of their own; a node containing an interrupt is kept small, because everything in it will repeat.

The same mechanism is used beyond approvals: to edit state, to ask for missing data only a human has, and to run headless tools on the client side.

Streaming

Incremental output is what makes the answer appear as it is generated, and what shows progress during long operations.

Where things come from:

There are two different APIs here.

The first is stream / astream with a stream_mode. It fits most tasks.

Several modes can be requested at once, as a list.

The stream format version

The format has a version, and v1 is the default: there, the shape of the output depends on the call — a single mode yields raw data, several modes yield (mode, data) tuples, and switching subgraphs on adds a namespace to the tuple.

v2 yields everything uniformly — a StreamPart dict with type, ns and data:

async for chunk in graph.astream(payload, stream_mode="messages", version="v2"):
    if chunk["type"] == "messages":
        message, metadata = chunk["data"]
        if metadata["langgraph_node"] in {"generate", "answer_user"}:
            print(message.content, end="")

New code should pass the version explicitly: the default stayed as it was for compatibility.

The metadata carries langgraph_node, tags and run identifiers. Filtering by node name is what keeps tokens from internal calls, such as classification, out of the user's stream. You can approach it from the other side too, tagging such a model nostream:

classifier = model.with_config({"tags": ["nostream"]})

Then its tokens are never emitted at all.

Subgraphs in the stream

async for chunk in graph.astream(payload, stream_mode="messages", subgraphs=True, version="v2"):
    ...

Without subgraphs=True, tokens produced inside a nested graph — including an agent used as a node — never reach the stream. There is no error; the stream is simply incomplete. The ns field on every chunk shows the source: an empty tuple for the root graph, a node name with a task id for a nested one.

Custom events

from langgraph.config import get_stream_writer

def index_documents(state):
    writer = get_stream_writer()
    writer({"type": "progress", "done": 0, "total": 100})

These are read under the custom mode. The same mechanism is how you stream a model that has no LangChain integration at all: call it yourself and write the chunks by hand.

astream_events

The second API is more detailed: every Runnable in the graph reports its start, its chunks and its finish.

async for event in graph.astream_events(state, version="v2"):
    kind = event["event"]          # on_chain_start, on_chat_model_stream, on_chain_end, on_tool_start…
    name = event["name"]
    data = event["data"]
    metadata = event.get("metadata", {})   # langgraph_node lives here too

You need this when it matters which node produced a token: node boundaries become stages for the user ("searching", "checking", "answering"), and tokens from different nodes go to different parts of the interface. The price is volume: the event stream is large, and a consumer needs precise filtering so internal content never reaches the interface.

There is also a newer stream_events(version="v3"), built around content blocks with typed projections for message, value and lifecycle channels. Moving to it means rewriting the consumer, not flipping a flag.

Functional API

A second front end to the same runtime. The process is written in ordinary Python — with if, for and await — while checkpoints, resumption, interrupts and streaming are preserved.

from langgraph.func import entrypoint, task

@task
def write_essay(topic: str) -> str:
    return generate(topic)

@entrypoint(checkpointer=checkpointer)
def workflow(topic: str) -> dict:
    essay = write_essay(topic).result()
    approved = interrupt({"essay": essay, "action": "approve?"})
    return {"essay": essay, "approved": approved}

@entrypoint takes a single positional argument — several values go in a dict. Input and output must be serialisable because they land in the checkpoint. There is no state schema: instead, the previous run's result is injected through the previous parameter. There is also entrypoint.final, for when one value is returned to the caller and a different one is saved for the next run.

@task is a unit of work whose result is written to the checkpoint. Calling one returns a future immediately; several tasks started before the first is resolved run in parallel — that replaces Send here.

The replay rule is the same, phrased for procedural code: on resume the entrypoint body is replayed from the top, and results of completed tasks are substituted from the checkpoint. So every side effect and every non-deterministic value goes inside a @task — ordinary code between tasks runs again.

Choosing between the two APIs. The Functional API is shorter when the flow is mostly linear and there is no shared state worth naming. The Graph API wins when the process has structure, several steps work with the same state, and its shape needs to be discussed: a graph can be rendered, a function with tasks cannot.

Multi-agent

"Multi-agent" usually means one of three shapes:

Graph as a tool. A compiled graph is runnable, so it is wrapped in a tool — and the calling agent delegates the work without knowing anything about the internals.

@tool
def research(topic: str) -> str:
    """Run the deep-research pipeline on a topic and return its report."""
    return research_graph.invoke({"topic": topic})["report"]

Supervisor. A coordinator distributes work among specialists and collects the results. As a graph that is a router node plus one node per specialist, with the route chosen through Command. Such a shape renders and can be inspected.

Swarm. Agents hand control to each other directly, with no central router. More flexible, but every handoff edge is a potential cycle, so a recursion_limit is needed.

Another option is subagents through SubAgentMiddleware: an ordinary agent gains a delegation tool without the rest of the Deep Agents harness.

Subagent or subgraph

The deciding property is context isolation. Delegating to a subagent fits when the subtask would otherwise flood the parent context with detail nobody needs afterwards: reading a long document, a noisy search. A subgraph fits when the step is part of the process and its state matters to what follows.

About cost: every subagent runs its own loop with its own system prompt and its own tool definitions, so running five subagents costs roughly as much as five agents, not one.

Deep Agents

A ready-made assembly of middleware on top of an ordinary agent, aimed at long tasks that do not fit into a single context window.

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[get_weather],
    system_prompt="You are a research assistant.",
)

What is switched on straight away:

• A filesystem with a full set of tools: list, read, write, edit, delete, search by glob and by content. Individual tools can be excluded; the layer itself cannot.
• Summarisation and context offloading, so a long task does not run into the window size.
• Prompt caching for static sections on providers that support it.
• Subagents through a tool that spawns a temporary agent with fresh context.

Task planning became optional in version 0.7 — older material describes it as on by default.

The filesystem is pluggable: memory, local disk, the graph's store, a composition of several sources, or your own implementation. Backends carry declarative access rules by path pattern, which is what defines what the agent may read and write. For running code there are sandboxes with a shell and interpreters that execute JavaScript.

The harness fits tasks that genuinely need planning across many steps, files as working memory, delegation and memory across sessions. For a bounded loop with a few tools it brings capabilities the task does not use but which take up part of the prompt budget.

Retrieval

This is a large topic of its own; I have an article on RAG and function calling and an article on vector databases. Here is the part that concerns the stack.

The pipeline has three stages, each testable on its own:

Indexing runs separately and ahead of time; retrieval and generation run on every user request.

There are two architectures. The classic two-step one: always retrieve, then answer — predictable latency, fewer model calls, a fit for a narrow corpus where retrieval is needed for every question. Agentic RAG: retrieval is a tool and the model decides whether it is needed, when, and with what query — variable latency, more turns, a fit for open-ended questions and several sources. In the first major version the agentic shape is presented as the default.

@tool
def search_docs(query: str) -> str:
    """Search the product documentation. Use for questions about how the product works."""
    return "\n\n".join(d.page_content for d in retriever.invoke(query))

agent = create_agent(model="openai:gpt-5.5", tools=[search_docs])

The docstring here works as the retrieval policy: it tells the model which questions this corpus is right for.

The parts

Loaders return Document objects — text plus metadata. Metadata is attached at load time: source, section, timestamp, tenant. Adding it later means reindexing the corpus.

Splitters:

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""],
)
chunks = splitter.split_documents(docs)

Chunk size is a retrieval-quality parameter. Too small and you lose the context that made the chunk answerable; too large and the embedding is diluted until it matches everything weakly. The usual starting range is 500–1500 characters with 10–20% overlap, then tuned by measurement.

Embeddings:

from langchain.embeddings import init_embeddings

embeddings = init_embeddings("openai:text-embedding-3-small")

The same model and the same dimensions must be used both when indexing and when querying. Mix them and the distances in the store stop meaning anything — with no error to tell you.

Vector stores: in memory for tests, Chroma for local development, pgvector next to an existing Postgres, plus Pinecone, Qdrant and Weaviate as separate packages.

Retrievers:

retriever = store.as_retriever(search_kwargs={"k": 4})

# Diversity instead of near-duplicate chunks
retriever = store.as_retriever(search_type="mmr",
                               search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5})

# Scoped by metadata — the tenant filter belongs here
docs = store.similarity_search(query, k=5, filter={"tenant_id": tenant})

The tenant and permission filter goes in the retriever: asking the prompt not to show other people's data is not an isolation mechanism.

Grounding the answer

Retrieval helps if the prompt makes the model prefer the retrieved text over its own memory. In practice: say explicitly that the answer must come only from the provided context, and that if it is not there the model should say so; number the chunks and require citations, which makes unsupported claims visible; keep the retrieved text in a clearly delimited block.

The last point matters for security as well: retrieved text is untrusted input. A document containing "ignore your previous instructions" is a prompt injection, which is why it is never pasted into the system message.

LangSmith: tracing, datasets and evaluation

Debugging ordinary code leans on a stack trace. A wrong answer from a model leaves no stack trace: you get a plausible paragraph that shows neither what the model saw nor why it decided as it did. So recording runs is part of the workflow.

LangSmith is switched on with environment variables and needs no code changes:

LANGSMITH_TRACING=true
LANGSMITH_API_KEY=<key>
LANGSMITH_PROJECT=<project>

After that every LangChain and LangGraph run is traced: nodes, model calls, tools, tokens, latencies. The older LANGCHAIN_ variable names no longer work.

It is worth attaching tags and metadata to runs — that is how traces are found later:

import langsmith as ls

with ls.tracing_context(enabled=True, project_name="quality-runs"):
    await agent.ainvoke(payload)

await agent.ainvoke(payload, config={"tags": ["production"], "metadata": {"user_id": user_id}})

What else the platform has: dashboards and alerts on quality metrics; rules, webhooks and online evaluation right on the production stream; annotation queues where answers are labelled by hand or user feedback is collected; automatic analysis of recurring problems in traces.

A separate area is prompt work: a playground for experimenting with prompts and model configurations, versioning with commits, a prompt hub with tags and a public part, and access to prompts from code. The point is that a prompt changes more often than code and can be edited by people who do not work with the repository.

An important consideration: traces contain prompts and completions, that is, customer content leaving your perimeter. Sending it to a third-party service is a matter of contracts and data-handling requirements.

The alternative, if that decision has not been made: your own tracing through OpenTelemetry plus one structured record per run. Either way it makes sense to record, for every model call, the model, the node, the latency, the token counts, the status and a correlation id. Full prompts and completions do not go into ordinary logs — that is the same customer data, only without access control.

The extension point for your own recording is callbacks: BaseCallbackHandler with on_llm_start, on_llm_end, on_tool_start and so on. A ready-made handler for token counting already exists.

Evaluating quality

An evaluation has three parts: a dataset (inputs and what a good output looks like), a target (the thing being run) and evaluators (scoring functions). A run of these is called an experiment, and experiments can be compared with previous ones.

from langsmith import Client

client = Client()
dataset = client.create_dataset(dataset_name="Sample dataset")

examples = [
    {"inputs": {"question": "Which country is Mount Kilimanjaro located in?"},
     "outputs": {"answer": "Mount Kilimanjaro is located in Tanzania."}},
]
client.create_examples(dataset_id=dataset.id, examples=examples)

experiment_results = client.evaluate(
    target,
    data="Sample dataset",
    evaluators=[correctness_evaluator],
    experiment_prefix="first-eval",
    max_concurrency=2,
)

Evaluators come in three kinds. Deterministic checks — the schema is valid, a required id is present, the length is within bounds — apply wherever the check is mechanical. LLM-as-judge scores what cannot be checked mechanically: faithfulness, tone, whether a required voice was kept; a judge has position and verbosity biases, and it is itself a prompt that can change. Human labelling is the most expensive and is usually reserved for the sample the automated checks flag.

Running an evaluation makes sense when the prompt changes, when the model changes and before a release — the three moments when quality moves without a single test failing. The dataset must stay the same from run to run.

There is also trajectory evaluation — checking the path the agent took, not only how the answer came out.

Testing

In layers, cheapest first.

Without a model. Graph wiring, routers, reducers and node logic are ordinary functions. A router is a pure function of state, and testing it directly is easier.

With a fake model. The real graph, a fake model with canned responses in order. That is enough to drive routing, streaming and error paths deterministically:

from langchain_core.language_models.fake_chat_models import FakeListChatModel

model = FakeListChatModel(responses=["CLARIFY", "final answer"])

For tools, LLMToolEmulator plays the same role.

With the real model, over a dataset. This is the layer that catches prompt regressions — the case where every test is green and the answers got worse. Such runs are usually kept out of the main build pipeline because they are slow, paid and slightly variable.

What fake models will not show you: whether the prompt really elicits the behaviour, and whether structured output parses against the real provider's grammar.

Running and shipping

Two ways to get a graph to users.

The ready-made server

LangSmith Deployment is a runtime for agents: it takes on the task queue, durable execution, streaming and horizontal scaling. The running part is called Agent Server and is built around three concepts: an assistant — a graph with a specific configuration, a thread — a state context, a run — one execution. There are cron jobs for scheduled runs, a persistence layer with checkpoints and a store, an API for creating runs, reading state and attaching to the stream, and the langgraph-sdk client library.

It can be hosted four ways: fully managed cloud; your own Kubernetes cluster alongside a self-hosted LangSmith; a hybrid where the control plane is with the vendor while servers and data are yours; or a standalone server in Docker or Kubernetes without a control plane.

The project configuration lives in one file:

{
  "dependencies": ["."],
  "graphs": { "agent": "./my_agent/agent.py:graph" },
  "env": "./.env",
  "python_version": "3.12"
}

Two keys are required: dependencies and graphs, a map from graph id to a file and a variable. The optional ones include env, store settings (including semantic search and record lifetimes), checkpointer settings, http parameters such as CORS and disabling routes, webhooks, an auth handler, a base image and a Python version, and pinning the server API version.

The langgraph-cli commands:

pip install -U "langgraph-cli[inmem]"
langgraph new path/to/app --template new-langgraph-project-python
cd path/to/app && pip install -e .
langgraph dev

Once it is up you get the API on port 2024, its documentation, and Studio — a visual interface that connects to the locally running agent. It shows every step: which prompts went to the model, which tool calls happened and with what arguments, what came back, how long it took and how many tokens it used. From there you can also re-run a thread from any step and see how the behaviour changes.

The difference between langgraph dev and langgraph up: the first is fast, needs no Docker and keeps state in memory — it is for iterating; the second brings up an environment close to production, with a real Postgres.

Embedded in your own service

The second way: the graph stays a library call inside your application, the API and the storage are yours, and shipping goes through your existing pipeline. This fits when the agent is part of a larger system where authentication, quotas and the external contract are already implemented elsewhere.

Then four things the Agent Server does for you have to be handled yourself:

• Compile graphs once at application start. Compiling per request adds latency to every turn.
• Do not block the event loop. A synchronous call inside a node stalls every concurrent request on that worker.
• Prepare the checkpointer schema at deploy time, not at start, and share the application's connection pool.
• Stop gracefully. On a deploy the container is killed and a run in flight is cut off; request_drain() leaves a checkpoint to continue from.

Migrating from 0.x

This section is useful even if you are starting from scratch: model training data and most articles on the web describe the previous version, so plausible-looking code often turns out to be a removed API.

Removed outright: Python 3.9 support; binding tools to a model before passing it to an agent (tools are now a create_agent parameter); structured output through asking for JSON and parsing prose; .text() as a method; AgentExecutor, initialize_agent and the zoo of ready-made agent types.

Moved to langchain-classic: the old chains, the whole former retrievers module, the indexing API, hub, CacheBackedEmbeddings and the community re-exports. This is not "deprecated but present" — importing them from the main package fails.

from langchain.chains import LLMChain
from langchain.retrievers import MultiQueryRetriever

# 1.x
from langchain_classic.chains import LLMChain
from langchain_classic.retrievers import MultiQueryRetriever

Deprecated but working: create_react_agent from langgraph.prebuilt; the tracing environment variables with the LANGCHAIN_ prefix; community integrations that now have dedicated packages. Deprecated features keep working with a warning across the whole first major line.

The mapping of old arguments to new ones comes down to one rule: what used to be a constructor argument is now a middleware hook. A callable prompt became @dynamic_prompt, pre- and post-model hooks became before_model and after_model, model selection on the fly became wrap_model_call, a tool error handler became wrap_tool_call, and a ToolNode object in the list became simply a list of tools.

On the LangGraph side almost nothing changed: state, nodes and edges, the execution model, checkpoints, streaming and interrupts all carried over unedited.

The migration order: Python 3.10 first, everything else is blocked on it; then install langchain-classic and repoint the imports — after that the code runs again, and the diff shows what merely moved and what needs reworking; then state schemas, agents one at a time, hooks, prompts, message handling and structured output together with its failure path. The last step is a dataset run: a mechanical migration compiles long before the behaviour is back, and where it shifts is above all prompt handling and structured output.

Within the first major line, upgrades usually amount to raising the floor and refreshing the lock file. Reading the release notes is still worth it for two reasons: middleware appears that replaces code you wrote by hand, and new format versions appear — they break nothing, but they also give nothing until a consumer is deliberately rewritten.

Behaviour that is easy to miss

A list of things that do not announce themselves with an error.

• A list field without a reducer. Two nodes write to it and only one value survives.
• A node returning the whole state instead of a partial update. The mutation bypasses the reducers.
• A Command with goto plus a static edge from the same node. Both destinations run.
• stream_mode="messages" without subgraphs=True. Tokens from a nested agent never reach the stream.
• A checkpointer without a thread_id. Nothing is saved and nothing errors.
• A side effect before an interrupt. The node replays from the start and the action repeats.
• update_state passing through a reducer. It appended where a replacement was intended.
• InMemorySaver in production. State disappears on restart.
• A synchronous model call in an async service. It blocks every concurrent request on that process.
• A user id as a tool argument. Its value is chosen by the model.
• A tool returning a Command without a ToolMessage. The provider sees a call with no result.
• Different embedding models for indexing and for retrieval. Distances stop meaning anything.

In short

What is worth carrying away.

The layers differ by area of responsibility: LangChain — models, tools and the agent loop; LangGraph — state, execution, memory and streaming; Deep Agents — a ready-made harness for long tasks; LangSmith — tracing and evaluation; Agent Server — a ready-made runtime. A LangChain agent is a LangGraph graph, so the runtime's capabilities are available at any level.

Configuring an agent lives in middleware: six hooks around the loop and a catalogue of shipped implementations.

In a graph everything rests on reducers: an accumulating field needs one, and a node returns only what it changed.

Memory is switched on explicitly: a checkpointer and a thread_id, both at once.

And a general property of the stack: a fair share of its failures are silent. Tokens that never appear, state that was not saved, lost results from a fan-out, an answer based on irrelevant chunks. That is why tracing runs here is a basic part of the work rather than an optional extra.