10 LangGraph and LangChain Features Most Agent Tutorials Still Don't Show You

Ten small features that decide whether your agent survives its first week in production. Every snippet below was run, and every output is copied from the terminal.

Most agent tutorials I read this year still teach the same thing. Import create_react_agent, give it two tools, ask it about the weather, done.

That function is now deprecated. In LangGraph 1.2.20, calling it prints this:

create_react_agent has been moved to `langchain.agents`. Please update your import to `from langchain.agents import create_agent`. Deprecated in LangGraph V1.0 to be removed in V2.0.

The bigger problem is not the import. It is that the tutorial stops at the happy path. Real agents loop forever, call flaky APIs, run on models that get retired, and sometimes try to delete a table. The demo never shows what happens then.

LangChain and LangGraph have answers for most of this now, and they are small. First, two terms you will see all the way through:

• An agent is a model calling tools in a loop until the task is done.
• Middleware is code that wraps that loop. It can look at every model call and every tool call, and change or stop it. LangChain's create_agent ships with built-in middleware, and you can write your own.

Items 1 to 5 use create_agent and a real model (gpt-4o-mini; the whole set cost well under one cent). Items 6 to 10 are LangGraph runtime features and need no model at all. Versions: langchain 1.4.9, langchain-core 1.7.6, langgraph 1.2.20, langchain-openai 1.7.4, Python 3.13.

1. Call limits, so a stuck agent stops by itself

An agent that polls "is it done yet?" will happily poll forever. Each loop is a paid model call.

What people usually write is a for i in range(10) around their own loop, or nothing at all and a prayer.

The better way is two lines of middleware:

from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
from langchain.tools import tool


@tool
def check_deploy_status(service: str) -> str:
    """Check whether a deploy has finished."""
    return f"{service}: still rolling out. Check again."  # never finishes


agent = create_agent(
    "openai:gpt-4o-mini",
    tools=[check_deploy_status],
    system_prompt="Keep checking until the deploy finishes, then report.",
    middleware=[
        ModelCallLimitMiddleware(run_limit=4, exit_behavior="end"),
        ToolCallLimitMiddleware(tool_name="check_deploy_status", run_limit=3),
    ],
)

result = agent.invoke({"messages": [("user", "Is the payments deploy done?")]})
for m in result["messages"]:
    print(f"{m.type:>5}: {str(m.content)[:90] or m.tool_calls[0]['name']}")

When I ran it, the agent did this:

human: Is the payments deploy done?
   ai: check_deploy_status
 tool: payments: still rolling out. Check again.
   ai: check_deploy_status
 tool: payments: still rolling out. Check again.
   ai: check_deploy_status
 tool: payments: still rolling out. Check again.
   ai: check_deploy_status
 tool: Tool call limit exceeded. Do not call 'check_deploy_status' again.
   ai: Model call limits exceeded: run limit (4/4)

The tool limit fired first and told the model to stop calling that tool. The model tried once more anyway (models are optimists), and then the model limit ended the run cleanly.

This helps any agent that pays per call. On-call agents that poll deploys are the classic case.

Experienced engineers like it because the limit lives outside the prompt. "Please do not call this more than three times" is a request. Middleware is a rule.

The trade-off: exit_behavior="end" returns a polite message, not an error, so your code must check for it. If you would rather fail loudly, use exit_behavior="error".

2. Tool retry with backoff, without writing the loop yourself

Internal services drop connections. If the agent gives up on the first error, the user gets "I could not reach the runbook" for a problem that would have fixed itself in 200 ms.

People usually wrap every tool body in try/except and a time.sleep, and copy that block into every tool.

import time

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware
from langchain.tools import tool

attempts = []


@tool
def search_runbook(query: str) -> str:
    """Search the on-call runbook."""
    attempts.append(time.monotonic())
    if len(attempts) < 3:
        raise ConnectionError("runbook service reset the connection")
    return "Runbook: restart the consumer, then check lag on the orders topic."


agent = create_agent(
    "openai:gpt-4o-mini",
    tools=[search_runbook],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            retry_on=(ConnectionError,),
            initial_delay=0.2,
            backoff_factor=2.0,
            jitter=False,
        )
    ],
)

result = agent.invoke({"messages": [("user", "Orders consumer is stuck. What does the runbook say?")]})
gaps = [round(b - a, 2) for a, b in zip(attempts, attempts[1:])]
print("tool attempts:", len(attempts), "| gaps between attempts (s):", gaps)
print("answer:", result["messages"][-1].content)

Here is the run. The tool failed twice on purpose:

tool attempts: 3 | gaps between attempts (s): [0.2, 0.41]
answer: The runbook suggests the following steps:

1. Restart the orders consumer.
2. Check the lag on the orders topic.

The gaps double (0.2 s, then 0.41 s) because of backoff_factor=2.0. The model never saw the two failures.

Use it for anything that goes over a network, such as RAG retrieval or a ticketing API.

Experienced engineers prefer this because retry rules sit in one place, and retry_on limits retries to errors that are actually temporary.

The trade-off: only retry tools that are safe to run twice. Retrying "search" is fine. Retrying "charge the card" without an idempotency key (a unique ID that makes a repeat request harmless) is how a customer pays three times. When retries run out, the default on_failure="continue" hands the error text to the model; "error" stops the run instead.

3. Model fallback, for the day your model is retired

Models get retired, and your agent is not told. While writing this post I found that gpt-4.1-nano now returns a 404 on my account. Any agent pinned to it is down.

from langchain.agents import create_agent
from langchain.agents.middleware import ModelFallbackMiddleware

# gpt-4.1-nano was retired on this account: every call to it returns a 404.
agent = create_agent(
    "openai:gpt-4.1-nano",
    tools=[],
    middleware=[ModelFallbackMiddleware("openai:gpt-4o-mini")],
)

result = agent.invoke({"messages": [("user", "In five words: what is a dead-letter queue?")]})
reply = result["messages"][-1]
print("answered by:", reply.response_metadata.get("model_name"))
print("answer:", reply.content)

And the output from a real run:

answered by: gpt-4o-mini-2024-07-18
answer: Message handling for processing failures.

The primary failed. The middleware tried the next model in the list and the user still got an answer.

This is useful for support bots that must answer, and for provider outages and rate limits.

Experienced engineers like it because the fallback list is plain config. You can put a model from a second provider in it.

The trade-off: I read the source. It falls back on any exception, not only on outages. A bug in your request will also be quietly "fixed" by the second model, so log which model answered (as the snippet does). The fallback model may also behave differently with your prompt, so test it on its own.

4. Human approval per tool call, with interrupt_mode="per_call"

Some tools should never run without a person saying yes: restarts, scaling, refunds. The older human-in-the-loop setup paused once per model turn and expected one combined answer for all calls in that turn, matched by position.

interrupt_mode="per_call" (new in langchain 1.4.6) gives each tool call its own pause and its own ID. An interrupt is LangGraph's way of pausing a run and saving it, so it needs a checkpointer (the component that saves the run's state).

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


@tool
def restart_service(name: str) -> str:
    """Restart a production service."""
    return f"{name} restarted"


@tool
def scale_service(name: str, replicas: int) -> str:
    """Change the replica count of a production service."""
    return f"{name} scaled to {replicas}"


agent = create_agent(
    "openai:gpt-4o-mini",
    tools=[restart_service, scale_service],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "restart_service": True,
                "scale_service": {"allowed_decisions": ["approve", "reject"]},
            },
            interrupt_mode="per_call",
        )
    ],
    checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "incident-42"}}

out = agent.invoke(
    {"messages": [("user", "Restart checkout and scale search to 40 replicas. Do both in parallel.")]},
    config,
    version="v2",
)
answers = {}
for intr in out.interrupts:
    v = intr.value
    print("PAUSED:", v["type"], v["name"], v["args"])
    if v["name"] == "scale_service":
        answers[intr.id] = {"type": "reject", "message": "40 is too many. Max is 12."}
    else:
        answers[intr.id] = {"type": "approve"}

out = agent.invoke(Command(resume=answers), config, version="v2")
for m in out.value["messages"]:
    if m.type == "tool":
        print("TOOL RESULT:", m.name, "->", m.content)
for intr in out.interrupts:
    print("PAUSED AGAIN:", intr.value["name"], intr.value["args"])

This is what happened when I ran it:

PAUSED: tool_approval restart_service {'name': 'checkout'}
PAUSED: tool_approval scale_service {'name': 'search', 'replicas': 40}
TOOL RESULT: restart_service -> checkout restarted
TOOL RESULT: scale_service -> User rejected the tool call for `scale_service` with reason: 40 is too many. Max is 12.
PAUSED AGAIN: scale_service {'name': 'search', 'replicas': 12}

The approved restart ran. The rejected scale did not. The model read my reason and came back with 12 replicas, which paused again for approval. That second pause is the part I liked most.

This is useful for any agent that can change production, from on-call bots with write access to refund flows.

Experienced engineers prefer per-call answers because each answer is keyed by its interrupt ID and checked against a schema before it is saved. An answer cannot quietly land on the wrong call.

The trade-off: the source docstring warns that this middleware must be listed before tool retry or error middleware, and that middleware listed before it runs again on resume. Also, people approve most prompts without reading them, so keep approvals for the few tools that matter.

5. Your own wrap_tool_call middleware

Built-ins will not cover your rules. Say your SQL agent must never write.

People usually put "never run DELETE" in the system prompt.

import time

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
from langchain.tools import tool


@tool
def run_sql(query: str) -> str:
    """Run a SQL query against the orders database."""
    return "3 rows"


@wrap_tool_call
def sql_guard(request, handler):
    query = request.tool_call["args"].get("query", "").upper()
    if any(word in query for word in ("DROP", "DELETE", "TRUNCATE", "UPDATE")):
        print(f"BLOCKED: {query}")
        return ToolMessage(
            content="Blocked: this agent has read-only access. Use SELECT only.",
            tool_call_id=request.tool_call["id"],
        )
    start = time.perf_counter()
    result = handler(request)  # run the real tool
    print(f"RAN: {query} in {(time.perf_counter() - start) * 1000:.2f} ms")
    return result


agent = create_agent("openai:gpt-4o-mini", tools=[run_sql], middleware=[sql_guard])
result = agent.invoke({"messages": [(
    "user",
    "Delete all rows in the orders table where status='test'. Use the exact SQL DELETE statement.",
)]})
print("FINAL:", result["messages"][-1].content)

I asked it to delete test orders. It printed:

BLOCKED: DELETE FROM ORDERS WHERE STATUS='TEST';
FINAL: I am unable to execute a DELETE statement as I have read-only access to the database. However, you can use the following SQL statement to delete the rows:

```sql
DELETE FROM orders WHERE status='test';
```

Please execute this statement in your database management interface.

The tool never ran. Then the model helpfully suggested I run the DELETE myself, which is a good reminder that the guard protects the tool, not the user.

wrap_tool_call gets the request and a handler. Call handler(request) to run the tool, or return your own ToolMessage to skip it. The same hook is a good place for audit logs and for redacting secrets.

Experienced engineers like it because a code check runs every time, and a prompt instruction runs when the model feels like it.

The trade-off: my word list is naive. It would also block SELECT last_updated FROM orders, because "UPDATE" is inside the column name. The real fix is a read-only database user. Use the middleware as a second layer, not the only one.

6. Durability modes: exit, async, sync

LangGraph saves a checkpoint (a snapshot of the run's state) as the graph moves. The durability argument decides when that write happens. To test it I killed the process inside the third step, the way an out-of-memory kill would, then tried to resume.

import operator
import os
import sqlite3
import subprocess
import sys
from typing import Annotated, TypedDict

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.errors import EmptyInputError
from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    done: Annotated[list[str], operator.add]


def step(name):
    def run(state: State):
        if name == "open_ticket" and os.environ.get("KILL") == "1":
            os._exit(1)  # the pod is OOM-killed: no exception, no cleanup
        return {"done": [name]}
    return run


def build(db):
    b = StateGraph(State)
    for n in ["fetch_logs", "summarise", "open_ticket"]:
        b.add_node(n, step(n))
    b.add_edge(START, "fetch_logs")
    b.add_edge("fetch_logs", "summarise")
    b.add_edge("summarise", "open_ticket")
    b.add_edge("open_ticket", END)
    return b.compile(checkpointer=SqliteSaver(sqlite3.connect(db, check_same_thread=False)))


if len(sys.argv) == 3:  # child process: run the graph, die at open_ticket
    mode, db = sys.argv[1], sys.argv[2]
    build(db).invoke({"done": []}, {"configurable": {"thread_id": "t1"}}, durability=mode)
    sys.exit(0)

for mode in ["exit", "async", "sync"]:
    db = f"/tmp/durability-{mode}.db"
    if os.path.exists(db):
        os.remove(db)
    subprocess.run([sys.executable, __file__, mode, db], env={**os.environ, "KILL": "1"})
    graph = build(db)
    config = {"configurable": {"thread_id": "t1"}}
    print(f"{mode:>5} after kill: saved={graph.get_state(config).values.get('done')}")
    os.environ["KILL"] = "0"
    try:
        ran = [list(u)[0] for u in graph.stream(None, config, stream_mode="updates")]
        print(f"{mode:>5} resume re-ran: {ran}  final={graph.get_state(config).values['done']}")
    except EmptyInputError as exc:
        print(f"{mode:>5} after resume: {type(exc).__name__}: {exc}")
    os.environ["KILL"] = "1"

The output, copied from the terminal:

 exit after kill: saved=None
 exit after resume: EmptyInputError: Received no input for __start__
async after kill: saved=[]
async resume re-ran: ['fetch_logs', 'summarise', 'open_ticket']  final=['fetch_logs', 'summarise', 'open_ticket']
 sync after kill: saved=['fetch_logs', 'summarise']
 sync resume re-ran: ['open_ticket']  final=['fetch_logs', 'summarise', 'open_ticket']

• exit saves only when the run ends. A hard kill means it never ended. Nothing was saved, and the thread looked as if it had never run. The docs warn about exactly this.
• async writes in the background while the next step runs. My steps take microseconds, so the writes had not landed when the process died. All three steps ran again. With slower steps the gap is smaller, but it is not zero.
• sync writes before every next step. Only the step that crashed ran again.

A note on my first attempt: I raised a Python exception instead of killing the process. All three modes kept their progress, because an exception still lets the run end in an orderly way. Test durability with a real kill.

Use sync for steps with side effects (opening tickets, sending emails). Use exit for short, cheap runs you can simply start again.

Experienced engineers prefer to choose this per call instead of accepting the default without knowing it.

The trade-off: sync is the slowest, because every step waits for a database write.

7. Node caching with CachePolicy

RAG pipelines run the same retrieval for the same question all day. Each run is an embedding call plus a vector search.

People usually add functools.lru_cache to the retrieval function, which lives in one process and knows nothing about the graph.

import time
from typing import TypedDict

from langgraph.cache.memory import InMemoryCache
from langgraph.graph import END, START, StateGraph
from langgraph.types import CachePolicy


class State(TypedDict):
    question: str
    docs: list[str]


def retrieve(state: State):
    time.sleep(1.5)  # pretend: embedding call + vector search
    return {"docs": [f"doc about {state['question']}"]}


builder = StateGraph(State)
builder.add_node(
    "retrieve",
    retrieve,
    cache_policy=CachePolicy(ttl=300, key_func=lambda s: s["question"].strip().lower()),
)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", END)
graph = builder.compile(cache=InMemoryCache())

for q in ["What is our refund policy?", "  what is our REFUND policy?", "How do I reset 2FA?"]:
    t = time.perf_counter()
    chunks = list(graph.stream({"question": q}, stream_mode="updates"))
    print(f"{time.perf_counter() - t:.2f}s  {chunks[0]}")

I asked three questions and timed each one:

1.51s  {'retrieve': {'docs': ['doc about What is our refund policy?']}}
0.00s  {'retrieve': {'docs': ['doc about What is our refund policy?']}, '__metadata__': {'cached': True}}
1.51s  {'retrieve': {'docs': ['doc about How do I reset 2FA?']}}

The second question differs only in spaces and case, so key_func mapped it to the same key: 0.00 s instead of 1.51 s, marked 'cached': True.

Use it for retrieval or any other expensive step that is pure (same input, same output).

Experienced engineers like that the cache is per node with its own TTL (time to live, how long an entry stays valid).

The trade-off is visible in the output. The cached result still says "What is our refund policy?" with the first question's exact text. If your key_func throws away something that matters, you serve the wrong answer very quickly. InMemoryCache also disappears on restart; SqliteCache exists for that.

8. Per-node timeouts and an error_handler that returns a Command

A CRM API that hangs will hang your agent with it. LangGraph 1.2 added timeout= per node and error_handler=, a function that runs after a node has failed all its retries. It can return a Command, an object that updates state and says which node to go to next. set_node_defaults sets all of this once for the whole graph.

import asyncio
import time
from typing import TypedDict

from langgraph.errors import NodeError
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, RetryPolicy

calls = {"lookup_customer": 0}


class State(TypedDict):
    ticket: str
    status: str


async def lookup_customer(state: State):
    calls["lookup_customer"] += 1
    await asyncio.sleep(5)  # the CRM API hangs
    return {"status": "found customer"}


async def draft_reply(state: State):
    return {"status": state["status"] + " -> reply drafted"}


async def escalate(state: State):
    return {"status": state["status"] + " -> sent to a human"}


async def on_failure(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"{error.node} failed ({type(error.error).__name__})"},
        goto="escalate",
    )


graph = (
    StateGraph(State)
    .set_node_defaults(
        timeout=1.0,
        retry_policy=RetryPolicy(max_attempts=2, initial_interval=0.1),
        error_handler=on_failure,
    )
    .add_node("lookup_customer", lookup_customer)
    .add_node("draft_reply", draft_reply)
    .add_node("escalate", escalate)
    .add_edge(START, "lookup_customer")
    .add_edge("lookup_customer", "draft_reply")
    .add_edge("draft_reply", END)
    .add_edge("escalate", END)
    .compile()
)

t = time.perf_counter()
result = asyncio.run(graph.ainvoke({"ticket": "T-1093", "status": ""}))
print(f"took {time.perf_counter() - t:.1f}s, CRM attempts: {calls['lookup_customer']}")
print("status:", result["status"])

The run took a few seconds:

took 3.0s, CRM attempts: 2
status: lookup_customer failed (NodeTimeoutError) -> sent to a human

Two attempts, each cut at one second, then the handler moved the ticket to a human instead of crashing.

My first version failed at compile time:

ValueError: Node timeouts are only supported for async nodes because sync Python execution cannot be safely cancelled in-process. Node '__default_error_handler__' is sync.

The docs do say that timeout defaults apply to error-handler nodes too, and timeouts only work on async code. So when you set a default timeout, your handler must be async def as well. The error message names the node, which helps.

This is useful for support bots that call slow vendor APIs, and for any graph that should degrade instead of crash.

Experienced engineers prefer it because the recovery path is a real node in the graph, visible and testable.

The trade-off: timeouts need async nodes, and a node that blocks with time.sleep will not be cut on time. Defaults are also not inherited by subgraphs.

9. stream(version="v2") and GraphOutput.interrupts

With several stream modes at once, the old stream gave you tuples whose shape depended on the options. And invoke mixed the pause information into your state under a __interrupt__ key.

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.config import get_stream_writer
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt


class State(TypedDict):
    refund: int
    status: str


def check_order(state: State):
    get_stream_writer()({"progress": "order found, paid by card"})
    return {"status": "checked"}


def approve_refund(state: State):
    answer = interrupt({"question": f"Approve refund of {state['refund']} INR?"})
    return {"status": f"refund {answer}"}


builder = StateGraph(State)
builder.add_node("check_order", check_order)
builder.add_node("approve_refund", approve_refund)
builder.add_edge(START, "check_order")
builder.add_edge("check_order", "approve_refund")
builder.add_edge("approve_refund", END)
graph = builder.compile(checkpointer=InMemorySaver())

print("--- stream, version v2: every part has the same shape")
cfg = {"configurable": {"thread_id": "a"}}
for part in graph.stream({"refund": 1200, "status": ""}, cfg,
                         stream_mode=["updates", "custom"], version="v2"):
    print(part["type"], "|", part["data"])

print("--- invoke, default")
old = graph.invoke({"refund": 1200, "status": ""}, {"configurable": {"thread_id": "b"}})
print(old)

print("--- invoke, version v2")
new = graph.invoke({"refund": 1200, "status": ""}, {"configurable": {"thread_id": "c"}}, version="v2")
print("value:", new.value)
print("interrupts:", [i.value for i in new.interrupts])

The printed result shows both shapes:

--- stream, version v2: every part has the same shape
custom | {'progress': 'order found, paid by card'}
updates | {'check_order': {'status': 'checked'}}
updates | {'__interrupt__': (Interrupt(value={'question': 'Approve refund of 1200 INR?'}, id='f2226891b56fbadf838dbeb04bc84ee5', response_schema=None),)}
--- invoke, default
{'refund': 1200, 'status': 'checked', '__interrupt__': [Interrupt(value={'question': 'Approve refund of 1200 INR?'}, id='19c1d978c90c7be7cf56b97568b45f4b', response_schema=None)]}
--- invoke, version v2
value: {'refund': 1200, 'status': 'checked'}
interrupts: [{'question': 'Approve refund of 1200 INR?'}]

With version="v2", every stream part is a dict with type, ns and data, and invoke returns a GraphOutput with .value (your state) and .interrupts (the pauses) kept apart.

Use it in chat UIs that show progress, and in any code that checks "did the run pause for approval?"

Experienced engineers like it because one shape means one parser, and the state no longer carries a key you never put there.

The trade-off: it is opt-in. Without version="v2", invoke still returns the old shape, so a codebase can easily end up with both.

10. Store for long-term memory, not the checkpointer

A user tells the bot their language and timezone on Monday. On Friday, in a new chat, the bot has forgotten.

People usually try to keep one thread alive forever. The checkpointer is short-term memory for one thread. The Store is long-term memory shared across threads, organised by namespace.

import operator
from dataclasses import dataclass
from typing import Annotated, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore


@dataclass
class Context:
    user_id: str


class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    known_prefs: dict


def chat(state: State, runtime: Runtime[Context]):
    ns = ("users", runtime.context.user_id)
    last = state["messages"][-1]
    if last.startswith("remember:"):
        key, value = last.removeprefix("remember:").strip().split("=")
        runtime.store.put(ns, key, {"value": value})
    prefs = {item.key: item.value["value"] for item in runtime.store.search(ns)}
    return {"known_prefs": prefs}


builder = StateGraph(State, context_schema=Context)
builder.add_node("chat", chat)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile(checkpointer=InMemorySaver(), store=InMemoryStore())

ctx = Context(user_id="priya")
monday = {"configurable": {"thread_id": "monday-chat"}}
friday = {"configurable": {"thread_id": "friday-chat"}}

graph.invoke({"messages": ["remember: language=Python"]}, monday, context=ctx)
graph.invoke({"messages": ["remember: timezone=IST"]}, monday, context=ctx)
out = graph.invoke({"messages": ["hi again"]}, friday, context=ctx)

print("Friday thread messages (checkpointer):", out["messages"])
print("Friday thread prefs (store):         ", out["known_prefs"])
print("Monday thread messages:              ", graph.get_state(monday).values["messages"])
other = graph.invoke({"messages": ["hello"]}, {"configurable": {"thread_id": "x"}},
                     context=Context(user_id="arjun"))
print("Another user's prefs:                ", other["known_prefs"])

This is what each thread could see:

Friday thread messages (checkpointer): ['hi again']
Friday thread prefs (store):          {'language': 'Python', 'timezone': 'IST'}
Monday thread messages:               ['remember: language=Python', 'remember: timezone=IST']
Another user's prefs:                 {}

Friday's thread has only its own message, but the preferences from Monday are there. Another user sees nothing, because the namespace includes the user ID.

This is useful for support bots that remember a customer, and assistants that learn preferences.

Experienced engineers keep the two apart because thread history grows without limit, while the Store holds only what you chose to keep.

The trade-off: the Store never forgets on its own. You decide what goes in, how to delete it when a user asks, and where it lives. InMemoryStore is gone on restart; use a database-backed store in production.

One I did not include

langchain 1.4 has built-in MCP support (langchain.mcp, still beta) for loading tools from MCP servers. It needs a running server, so it did not fit the "every snippet runs on its own" rule here. That one can wait for a follow-up post.

Which one would you add?

That is my ten. Most of them are a few lines, and most of them are the difference between a demo and something you can leave running overnight.

Which feature would you add to this list? Tell me in the comments.

---

Versions used: langchain 1.4.9, langchain-core 1.7.6, langgraph 1.2.20, langchain-openai 1.7.4, Python 3.13.15. I write about building and running AI systems at systemdesign.academy.