23 Async at Scale

Async has been in this book for a while, one piece at a time. Chapter 8 timed a checkpointer in its "async" durability mode. Section 22.7 put async def on the streaming endpoint and spent one paragraph on the event loop, because the endpoint could not be written without it. Section 25.9 will compare threads and asyncio side by side in a table. Here is what async is actually for:

Async does not make anything faster. It lets the process do something else while it waits, and it pays only where there is waiting to overlap.

23.1 When to use

Async pays off in two situations.

A node that waits on several things. The doc-bot’s research asks the model about the question from three angles and looks one thing up on the web. Four waits, one after another, is the sum of four waits. If the node can have all four in flight, its time is the longest of them.
A process that serves several callers. A server holds one compiled graph and answers whoever arrives. Run after run, it is one caller at a time and the machine is idle while each model answers. With async the process can hold twenty runs, each parked at its own await, on one thread and without a lock.

The price is that async is infectious: one async def node closes the synchronous door for every caller of that graph, now and later.

23.2 Just enough asyncio

asyncio is the Python standard-library module that runs coroutines and owns the event loop. Minimal terms asyncio needs for this chapter:

Term	What it means here
coroutine	What calling an `async def` returns. The `async def` declares a coroutine function; calling it does not run it, it hands back a coroutine, an object that can be run.
`await`	Where a coroutine runs, and where it may stop. `await something` means “give me the result of `something`, and if it is not ready yet, let somebody else have the processor until it is.”
event loop	The somebody else: the scheduler every coroutine shares. (1) It runs the current coroutine until it reaches an `await` that is not ready, (2) then runs whichever other coroutine is ready.
`gather`	Hands the event loop several coroutines at once and gives back their results in order. It is how a single node waits on three things in the time of the slowest.

Async does not make any one piece of work faster. A model call that takes 1.7 seconds takes 1.7 seconds under await llm.ainvoke too; a database query that takes 3 milliseconds still takes 3. What changes is what the process does during those 1.7 seconds: instead of sitting idle it can send a second request or serve a second caller. So the only time async can save is time that would otherwise have been spent waiting.

Python’s threads do the same job a different way. They are real operating-system threads, but the GIL¹ lets only one of them run Python code at a time, so threads do not speed up computation either. A thread does release the GIL while it waits on I/O, a socket read, a time.sleep, a file, so threads can overlap waiting, and that is what a thread pool is for. asyncio overlaps waiting on one thread instead: the event loop hands the processor to whichever coroutine is ready, and a coroutine gives it up only at an await. That costs no thread per wait and needs no locks, since nothing is interrupted between two awaits. The price is one rule, which the rest of the chapter keeps coming back to: a blocking call anywhere on that thread stops everything on it.

Either way, Python overlaps the waiting, never the work. Three model calls sent at once still queue for the same GPU at the other end. Running computation in parallel takes more processes or more hardware, and every ratio in this chapter is bounded by that.

23.3 The graph, and the three layers

We’ll build a three nodes graph, search_docs → research → generate: the first and last are Chapter 22’s, its approval gate and web search are left out, and the middle one is new. research asks the model about the question from three angles — what the thing is, how to configure it, what people get wrong — and looks one thing up on the web, and under async it waits on all four at once. Two of the three nodes are async def; the first is a plain def, on purpose, because it has nothing to wait on. The boxes inside research are function calls, not graph nodes: the graph has three nodes, and research is one of them. The badge still marks whatever calls the model, node or function.

%%{init: {'securityLevel': 'loose', 'flowchart': {'diagramPadding': 24}, 'themeCSS': 'foreignObject{overflow:visible}span.bot{display:flex;justify-content:flex-start;align-items:flex-start;width:100%;height:0;overflow:visible;line-height:0}span.bot img{display:block!important;width:18px!important;height:18px!important;padding:2px;background:#fff;border-radius:50%;box-shadow:0 0 0 1px #999}span.bot.r img{margin:0;transform:translate(-24px,-24px)}span.bot.d{justify-content:center}span.bot.d img{margin-top:-50%;transform:translateY(-24px)}'}}%%
flowchart LR
    START(["START"]) --> search_docs["search_docs<br/><i>def</i>"]
    search_docs --> a1 & a2 & a3 & w
    subgraph research ["research &nbsp;<i>async def</i> &nbsp;"]
        a1[["<span class='bot r'><img src='data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIyNCIgaGVpZ2h0PSIyNCIgdmlld0JveD0iMCAwIDI0IDI0IiBmaWxsPSJub25lIiBzdHJva2U9IiMzMzMiIHN0cm9rZS13aWR0aD0iMiIgc3Ryb2tlLWxpbmVjYXA9InJvdW5kIiBzdHJva2UtbGluZWpvaW49InJvdW5kIj48cGF0aCBkPSJNMTIgOFY0SDgiLz48cmVjdCB3aWR0aD0iMTYiIGhlaWdodD0iMTIiIHg9IjQiIHk9IjgiIHJ4PSIyIi8+PHBhdGggZD0iTTIgMTRoMiIvPjxwYXRoIGQ9Ik0yMCAxNGgyIi8+PHBhdGggZD0iTTE1IDEzdjIiLz48cGF0aCBkPSJNOSAxM3YyIi8+PC9zdmc+'/></span>ask: what it is<br/><i>await</i>"]]
        a2[["<span class='bot r'><img src='data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIyNCIgaGVpZ2h0PSIyNCIgdmlld0JveD0iMCAwIDI0IDI0IiBmaWxsPSJub25lIiBzdHJva2U9IiMzMzMiIHN0cm9rZS13aWR0aD0iMiIgc3Ryb2tlLWxpbmVjYXA9InJvdW5kIiBzdHJva2UtbGluZWpvaW49InJvdW5kIj48cGF0aCBkPSJNMTIgOFY0SDgiLz48cmVjdCB3aWR0aD0iMTYiIGhlaWdodD0iMTIiIHg9IjQiIHk9IjgiIHJ4PSIyIi8+PHBhdGggZD0iTTIgMTRoMiIvPjxwYXRoIGQ9Ik0yMCAxNGgyIi8+PHBhdGggZD0iTTE1IDEzdjIiLz48cGF0aCBkPSJNOSAxM3YyIi8+PC9zdmc+'/></span>ask: how to configure it<br/><i>await</i>"]]
        a3[["<span class='bot r'><img src='data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIyNCIgaGVpZ2h0PSIyNCIgdmlld0JveD0iMCAwIDI0IDI0IiBmaWxsPSJub25lIiBzdHJva2U9IiMzMzMiIHN0cm9rZS13aWR0aD0iMiIgc3Ryb2tlLWxpbmVjYXA9InJvdW5kIiBzdHJva2UtbGluZWpvaW49InJvdW5kIj48cGF0aCBkPSJNMTIgOFY0SDgiLz48cmVjdCB3aWR0aD0iMTYiIGhlaWdodD0iMTIiIHg9IjQiIHk9IjgiIHJ4PSIyIi8+PHBhdGggZD0iTTIgMTRoMiIvPjxwYXRoIGQ9Ik0yMCAxNGgyIi8+PHBhdGggZD0iTTE1IDEzdjIiLz48cGF0aCBkPSJNOSAxM3YyIi8+PC9zdmc+'/></span>ask: the most common mistake<br/><i>await</i>"]]
        w[["web lookup<br/><i>await</i>"]]
        a1 & a2 & a3 & w --> g[["gather"]]
    end
    classDef fn fill:#fff,stroke:#999,color:#000
    class a1,a2,a3,w,g fn
    g --> generate["<span class='bot r'><img src='data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIyNCIgaGVpZ2h0PSIyNCIgdmlld0JveD0iMCAwIDI0IDI0IiBmaWxsPSJub25lIiBzdHJva2U9IiMzMzMiIHN0cm9rZS13aWR0aD0iMiIgc3Ryb2tlLWxpbmVjYXA9InJvdW5kIiBzdHJva2UtbGluZWpvaW49InJvdW5kIj48cGF0aCBkPSJNMTIgOFY0SDgiLz48cmVjdCB3aWR0aD0iMTYiIGhlaWdodD0iMTIiIHg9IjQiIHk9IjgiIHJ4PSIyIi8+PHBhdGggZD0iTTIgMTRoMiIvPjxwYXRoIGQ9Ik0yMCAxNGgyIi8+PHBhdGggZD0iTTE1IDEzdjIiLz48cGF0aCBkPSJNOSAxM3YyIi8+PC9zdmc+'/></span>generate<br/><i>async def</i>"]
    generate --> END(["END"])

Three different things get called “concurrency” around this graph, and keeping them apart is most of what the rest of the chapter is:

Layer	What it is	What it bounds
Python’s `asyncio`	coroutines interleaving on one event loop: `async def`, `await`, `gather`, `Semaphore`, `to_thread`	how many awaits are outstanding on the thread
The graph’s door	`invoke`/`stream` against `ainvoke`/`astream`, and which of the two a node can answer	whether the event loop is involved at all, and which thread a node gets
LangGraph’s own scheduling	a superstep starts every ready node as a task: parallel branches, `Send` fan-out, and the `max_concurrency` key	how many node tasks run at once

Which model answers decides most of what this chapter measures. A laptop takes about as long to answer four requests at once as to answer them one after another; a hosted model has a fleet of GPUs to put all four on, and answers them in about the time of one. Every second recorded here was taken against a hosted model:

code/ch21_async.py

llm = ChatOpenAI(                                  # after tools/use_llm.py openai
    base_url="https://api.openai.com/v1",
    api_key=os.environ["OPENAI_API_KEY"],
    model="gpt-4.1-mini",
    temperature=0,
    reasoning_effort=None,
    timeout=240,
)

The companion ships pointed at the local server, like every other program in the book. python tools/use_llm.py openai (Appendix A) rewrites the base_url, api_key and model lines above, and nothing else; Section 23.13 is the same measurements taken back on the local one.

One question drives the single-run sections and four drive the many-caller ones; research asks about each from three angles.

code/ch21_async.py

ANGLES = ["what it is", "how to configure it", "the most common mistake"]

QUESTION = "How do I configure Kafka batch consumers?"
QUESTIONS = [
    "How do I configure Kafka batch consumers?",
    "How should I size a Redis connection pool?",
    "Why is my MySQL query slow?",
    "What does Kafka's max-poll-records do?",
]

research is built from two coroutines. The first asks the model about the question from one angle. Note what it does and does not do: it formats a prompt, it awaits the model, it returns a string. The await on llm.ainvoke is the only place it can stop, and it stops there for as long as the model takes.

code/ch21_async.py

async def ask(angle: str, question: str, docs: list[str]) -> str:
    """One model call, one angle. The unit `research` is built from."""
    prompt = (f"In two sentences, {angle} for: {question}\n"
              f"Use only these notes: {' '.join(docs) or '(none)'}")
    reply = await llm.ainvoke(prompt)
    return f"[{angle}] {reply.content.strip()}"

The second stands in for a tool that waits on something other than the model. It sleeps on the event loop for a second and a half, the way an HTTP client sleeps between sending a request and reading the first byte:

code/ch21_async.py

async def web_lookup() -> str:
    """A stand-in for a tool that waits on the network: a web search, a database.

    It sleeps on the event loop, which is exactly what an `httpx` call does between
    sending the request and the first byte of the reply.
    """
    await asyncio.sleep(1.5)
    return "[web] Kafka batch listeners deliver a List<ConsumerRecord> per poll."

The node itself hands all four coroutines to gather:

code/ch21_async.py

async def research_node(state: State) -> dict:
    """Three model calls and the web lookup, all in flight at once.

    `gather` hands the four coroutines to the event loop; each yields at its own
    `await`, and the event loop waits on four sockets instead of one. The two slower
    ways to write this node are in section 3.
    """
1    *notes, web = await asyncio.gather(
        *(ask(a, state["question"], state["docs"]) for a in ANGLES),
        web_lookup(),
    )
    return {"notes": [*notes, web]}

1: gather starts all four and gives the results back in the order the coroutines were passed, not the order they finished. Four coroutines in, four results out, so *notes, web takes the three ask results into notes and the web line into web. The node’s time is the slowest of the four waits instead of their sum.

The program builds this one graph and swaps in whichever research the section is measuring, so build takes the node as an argument:

code/ch21_async.py

1def build(research=research_node, checkpointer=None, wrap=lambda node: node, timeout=None):
    """The graph, with `research` swappable: every section swaps in the version it measures."""
    b = StateGraph(State)
    b.add_node("search_docs", wrap(search_docs_node))
    b.add_node("research", wrap(research), timeout=timeout)
    b.add_node("generate", wrap(generate_node))
    b.add_edge(START, "search_docs")
    b.add_edge("search_docs", "research")
    b.add_edge("research", "generate")
    b.add_edge("generate", END)
    return b.compile(checkpointer=checkpointer)

1: wrap and timeout are for later sections. wrap has the shape of a decorator: it takes a node function and gives one back. It is applied by calling it, wrap(node), instead of with @wrap above the def, because each section wraps the same three nodes differently.

flowchart TD
    START(["START"]) --> search_docs["search_docs"]
    search_docs --> research["research"]
    research --> generate["generate"]
    generate --> END(["END"])

The program is one async def main() that awaits the nine sections in turn:

code/ch21_async.py

async def main() -> None:
    await two_doors()                    # 1. invoke and ainvoke on the same graph
    await the_threads()                  # 2. which thread each node runs on
    await the_cost_of_waiting()          # 3. one node, several waits
    await across_runs()                  # 4. one process, many callers
    await a_blocking_library()           # 5. a library that blocks
    await a_ceiling()                    # 6. too many calls in flight
    await a_run_that_must_stop()         # 7. a run that must stop
    await state_that_outlives_the_run()  # 8. state that outlives the run
    await a_coroutine_never_awaited()    # 9. a coroutine nobody awaited

if __name__ == "__main__":
1    asyncio.run(main())

1: This line is where the event loop comes from. asyncio.run makes a new event loop, runs main() on it until it returns, then closes it. graph.ainvoke is itself an async def, so calling it only hands back a coroutine, and that coroutine runs on whichever event loop is already going.

The sections below need one small time profiler, to put a number on what waiting costs and on what async wins back:

code/ch21_async.py

class timed:
    """`with timed("label"):` times the block and prints the label and the seconds.

    `compare_to=` names an earlier `timed` block, and the line then ends with the
    ratio: how many times faster this block was than that one.
    """

    def __init__(self, label: str, compare_to: "timed | None" = None):
        self.label, self.compare_to = label, compare_to

    def __enter__(self):
        self.t0 = time.perf_counter()
        return self

    def __exit__(self, *exc):
        self.seconds = time.perf_counter() - self.t0
        ratio = f"   ({self.compare_to.seconds / self.seconds:.1f}x faster than the first)" if self.compare_to else ""
        print(f"   {self.label:<40} {self.seconds:6.1f}s{ratio}")

ch21_max_concurrency.py, the second program this chapter runs, imports it from here rather than writing it again.

23.4 `invoke` is closed, `ainvoke` is open

A compiled graph has a sync door and an async door: invoke and stream on one side, which Chapter 18 used, and ainvoke and astream on the other, which Section 22.7 had to use. (batch and abatch are the same pair for a list of inputs.) Both open onto the same graph, but they are not symmetric. Section 1 tries each door:

code/ch21_async.py

async def two_doors() -> None:
    print("\n===== 1. two doors: invoke and ainvoke on the same graph =====")
    result = await graph.ainvoke({"question": QUESTION})
    print(f"   ainvoke : notes={len(result['notes'])}  answer starts: {result['answer'][:60]!r}")
    try:
        graph.invoke({"question": QUESTION})     # the sync door, on a graph with async nodes
    except TypeError as exc:
        lines = str(exc).splitlines()            # two lines: what is wrong, and what to do
        print("   invoke  : TypeError: " + "\n             ".join(lines))
    print("   a sync node answers both doors; an async node is reachable through ainvoke only")

code/ch21_async.py

===== 1. two doors: invoke and ainvoke on the same graph =====
   ainvoke : notes=4  answer starts: 'To configure Kafka batch consumers, set the `spring.kafka.li'
   invoke  : TypeError: No synchronous function provided to "research".
             Either initialize with a synchronous function or invoke via the async API (ainvoke, astream, etc.)
   a sync node answers both doors; an async node is reachable through ainvoke only

Where does that error come from?

add_node wraps each node in an object with two attributes, func for the sync path and afunc for the async one:

a plain def gets both; afunc runs it on a worker thread;
an async def gets only afunc, and func is None.

graph.invoke then calls each node’s .invoke() in turn, and the first one whose func is None raises, with the name the node was registered under; graph.ainvoke calls each node’s .ainvoke(), and all three have an afunc:

flowchart LR
    invoke(["graph.invoke()"]) --> sd_f["search_docs.invoke()<br/>func = search_docs_node"]
    sd_f --> rs_f["research.invoke()<br/>func = None"]
    rs_f -.-> err["TypeError: No synchronous<br/>function provided to &quot;research&quot;"]
    ainvoke(["graph.ainvoke()"]) --> sd_a["search_docs.ainvoke()<br/>afunc = search_docs_node, on a thread"]
    sd_a --> rs_a["research.ainvoke()<br/>afunc = research_node"]
    rs_a --> gn_a["generate.ainvoke()<br/>afunc = generate_node"]
    gn_a --> done(["END"])
    classDef bad fill:#f8d7da,stroke:#b02a37,color:#000;
    class err bad;

So search_docs ran before the error, and generate was never reached; the error is raised at run time, by the first async node the sync walk meets. The error is correct: a coroutine cannot be run without an event loop, and invoke does not have one.

Deep Dive: why the two slots are not symmetric

This is what add_node leaves in the two slots:

Node	`func`	`afunc`
`def search_docs_node`	the function itself	`partial(run_in_executor, None, search_docs_node)`²
`async def research_node`	`None`	the function itself

A plain def can always be made awaitable: hand it to a thread and the event loop gets a future to wait on, the function itself unchanged. So afunc is filled for every def.

A coroutine cannot be made callable the same way, because it runs only while an event loop drives it, and invoke is called where there is no event loop. LangGraph could start one per call with asyncio.run, but that raises RuntimeError: asyncio.run() cannot be called from a running event loop in the case where it would be needed most: a caller that is already on an event loop, such as a FastAPI handler. So it fills the slot it can fill and leaves the other empty.

That is the price named at the top of the chapter, and it is paid once. The front end that calls invoke today calls ainvoke tomorrow, and so does the test suite.

23.5 Where each node runs

The sync door is closed to an async node, but the async door is open to a sync node, and the question is which thread it runs on.

The thread matters because the event loop is one thread: a node that blocks on that thread stops every other coroutine in the process. An async def node is trusted not to block: it runs on the event loop’s thread, and it is expected to put every slow thing behind an await; breaking that promise is a common mistake. A plain def node makes no such promise, so the runtime has to put it somewhere else.

We will make an instrument to see the thread each node is on: wrap every node of the graph in a function that prints threading.current_thread().name and then runs the node as before, and drive the graph through ainvoke.

code/ch21_async.py

def reporting_thread(node):
    """Wrap a node so that it prints the thread it runs on, then runs as before."""
    name = node.__name__.removesuffix("_node")
1    if inspect.iscoroutinefunction(node):
        async def wrapped(state: State) -> dict:
            print(f"   {name:<12} async def  ran on thread: {threading.current_thread().name}")
            return await node(state)
    else:
        def wrapped(state: State) -> dict:
            print(f"   {name:<12} def        ran on thread: {threading.current_thread().name}")
            return node(state)
    return wrapped

1: LangGraph decides sync or async by this same test on the object you register, so the wrapper has to be async def exactly when the node is. A plain def wrapper that merely returns a coroutine is treated as sync and hands the runtime a coroutine where it expected a dict.

Section 2 is the chapter’s graph with the wrapper on each node:

code/ch21_async.py

async def the_threads() -> None:
    print("\n===== 2. where nodes run: the same graph, each node reporting its thread =====")
    await build(wrap=reporting_thread).ainvoke({"question": QUESTION})
    print("   a plain def is pushed to a worker thread so it cannot stall the event loop; an async def runs on the event loop itself")

code/ch21_async.py

===== 2. where nodes run: the same graph, each node reporting its thread =====
1   search_docs  def        ran on thread: asyncio_2
   research     async def  ran on thread: MainThread
   generate     async def  ran on thread: MainThread
   a plain def is pushed to a worker thread so it cannot stall the event loop; an async def runs on the event loop itself

1: MainThread is the event loop’s own thread. A name like asyncio_2 is a worker thread from a pool.

So the contract for mixing them is this. An async def node runs on the event loop’s own thread and must wait that thread’s way: every slow thing inside it behind an await. A plain def node is pushed onto a worker thread, where it may block as much as it likes without stalling anyone. Convert a node to async def only when it has awaits to make. That is why, in the chapter’s graph, search_docs is a plain def: it is a dictionary lookup, with nothing to wait on.

The worker thread a plain def node runs on is not LangGraph’s. It comes from the event loop’s default thread pool, and that pool has min(32, cpus + 4) threads. It is the same pool you reach yourself with await asyncio.to_thread(fn, ...), which a later section uses. If more sync nodes than that are blocking at the same moment, the extra ones wait in a queue until a thread frees up. Nothing reports the wait; the runs just take longer.

A sync node does not stall the event loop, but it does occupy one of a limited number of threads for as long as it blocks.

23.6 Situation one: a node that waits on several things

Section 23.3 wrote research the fast way: all four waits handed to gather at once. We’ll explore two other ways to write the same node, and explain why they are slower.

The first awaits the four one after another, so its time is the sum:

code/ch21_async.py

async def research_in_series(state: State) -> dict:
    """Every wait finishes before the next one starts: the time is the sum."""
    notes = [await ask(a, state["question"], state["docs"]) for a in ANGLES]
    notes.append(await web_lookup())
    return {"notes": notes}

The second hands the web lookup and the three model calls, still in series, to gather, so the lookup’s wait happens alongside the model’s:

code/ch21_async.py

async def research_web_alongside(state: State) -> dict:
    """The web lookup waits alongside the model; the three model calls still take turns."""
    async def angles_in_series() -> list[str]:
        return [await ask(a, state["question"], state["docs"]) for a in ANGLES]
1    notes, web = await asyncio.gather(angles_in_series(), web_lookup())
    return {"notes": [*notes, web]}

1: Two coroutines in, two results out, in the order they were passed: the list of notes first, the web line second, so the two names on the left pick them up.

Section 3 runs the graph once with each, under the timed of Section 23.3: compare_to= puts the ratio against an earlier block on the line.

code/ch21_async.py

async def the_cost_of_waiting() -> None:
    print("\n===== 3. one node, several waits: research written three ways =====")
    with timed("every wait one after another") as in_series:
        await build(research_in_series).ainvoke({"question": QUESTION})
    with timed("web lookup alongside the model", compare_to=in_series):
        await build(research_web_alongside).ainvoke({"question": QUESTION})
    with timed("everything at once", compare_to=in_series):
        await graph.ainvoke({"question": QUESTION})
    print("   the first saving is the whole web wait, the event loop's; the second is the model server's")

The shape of the first and the third, with the event loop on the left:

sequenceDiagram
    participant L as event loop
    participant M as model server
    participant W as web lookup
    Note over L,W: one after another: ask + ask + ask + web
    L->>M: await ask(...)
    M-->>L: reply
    L->>M: await ask(...)
    M-->>L: reply
    L->>M: await ask(...)
    M-->>L: reply
    L->>W: await web_lookup()
    W-->>L: result
    Note over L,W: at once: gather(ask, ask, ask, web_lookup)
    par
        L->>M: ask(...)
        M-->>L: reply
    and
        L->>M: ask(...)
        M-->>L: reply
    and
        L->>M: ask(...)
        M-->>L: reply
    and
        L->>W: web_lookup()
        W-->>L: result
    end
    Note over L,W: the slowest of the four

code/ch21_async.py

===== 3. one node, several waits: research written three ways =====
   every wait one after another                5.0s
   web lookup alongside the model              3.6s   (1.4x faster than the first)
   everything at once                          2.4s   (2.1x faster than the first)
   the first saving is the whole web wait, the event loop's; the second is the model server's

Two savings, and they come from two different places:

The first is the event loop’s. Moving the web lookup alongside the model saved 1.4 seconds, about the whole of the lookup’s second and a half: it fit inside the time the model was taking anyway. That saving is yours whenever a node waits on two different things.
The second is the server’s, and not the event loop’s to give. Sending the three model calls at once instead of in turn saved 1.2 seconds more. The loop did its part, three requests on the wire at the same moment; how much of the work the server overlaps after that is the server’s business. This one answered all three inside the web lookup’s second and a half, which is why the node’s time stops falling there.

Gotcha: when one of the gathered calls fails

gather also decides what happens when one of the four raises: the node raises at once, and the other three keep running to the end with nobody waiting for their answers. One 429 from the model server and two calls’ worth of tokens are wasted, and the result doesn’t report it. gather(..., return_exceptions=True) makes the failure a value in the list instead of an exception; asyncio.TaskGroup (Python 3.11) cancels the siblings when one fails.

23.7 Situation two: one process, many callers

The second place waiting can be overlapped is between runs, and it is the one Chapter 22‘s streaming endpoint lives in: a process holding several callers’ runs at once. Four questions, first one after another, then all four handed to gather. The gather is the one from Section 23.6; what changes is the unit. There each coroutine was one model call inside one run; here each is a whole run, graph.ainvoke, and the same compiled graph serves all four at once with no lock and no copy:

code/ch21_async.py

async def across_runs() -> None:
    print(f"\n===== 4. one process, many callers: {len(QUESTIONS)} runs =====")
    with timed("one after another") as in_series:
        for q in QUESTIONS:
            await graph.ainvoke({"question": q})
    with timed("at once (asyncio.gather)", compare_to=in_series):
        await asyncio.gather(*(graph.ainvoke({"question": q}) for q in QUESTIONS))
    print("   the gain is bounded by how many requests the model server runs at once, not by this process")

code/ch21_async.py

===== 4. one process, many callers: 4 runs =====
   one after another                          10.1s
   at once (asyncio.gather)                    2.6s   (3.9x faster than the first)
   the gain is bounded by how many requests the model server runs at once, not by this process

Four runs at once is sixteen model calls, twelve of them in flight at the same moment, and the process handled that without a lock or a line of code beyond gather. Four runs took little more than the time of one, which is what it looks like when both sides hold up their end: the event loop kept every call in flight, and the server had somewhere to put them.

That is the shape of “async at scale.” The event loop removes this process as the bottleneck, and the bottleneck moves to the other end of the socket. Whether that is a win depends entirely on what is at the other end, and Section 23.13 is what it looks like when the other end is one GPU.

The two situations above are reasons to use async. In the following sections we’ll check when async is there to save the day.

23.8 Pattern: a library that blocks

Not every client can await. A legacy HTTP library, a database driver with no async build, a vendor SDK: called from an async def node, such a call holds the event loop’s thread for its whole duration. The fix is to wrap it in asyncio.to_thread, which makes a blocking call awaitable.

code/ch21_async.py

def legacy_search(question: str) -> str:
    """A client you may not rewrite: it blocks, the way a sync HTTP library does."""
    time.sleep(2)
    return "[tickets] an old issue about batch listeners"

code/ch21_async.py

async def research_blocking_tool(state: State) -> dict:
    """The legacy call made straight from an async node: it holds the event loop for two seconds."""
    tickets = legacy_search(state["question"])
    web = await web_lookup()
    return {"notes": [tickets, web]}

code/ch21_async.py

async def research_tool_on_a_thread(state: State) -> dict:
    """The same call handed to a worker thread: the event loop keeps running."""
1    tickets, web = await asyncio.gather(asyncio.to_thread(legacy_search, state["question"]),
                                        web_lookup())
    return {"notes": [tickets, web]}

1: asyncio.to_thread(fn, *args) returns an awaitable. It puts the call on the event loop’s default thread pool — the one Section 23.5’s def nodes land in — and hands back something gather can wait on with the rest. The thread blocks; the event loop does not.

Four runs, each making the legacy call once, written both ways:

code/ch21_async.py

async def a_blocking_library() -> None:
    print(f"\n===== 5. a library that blocks: {len(QUESTIONS)} runs, each calling it =====")
    with timed("called from the async node"):
        await asyncio.gather(*(build(research_blocking_tool).ainvoke({"question": q})
                               for q in QUESTIONS))
    with timed("handed to asyncio.to_thread"):
        await asyncio.gather(*(build(research_tool_on_a_thread).ainvoke({"question": q})
                               for q in QUESTIONS))

code/ch21_async.py

===== 5. a library that blocks: 4 runs, each calling it =====
   called from the async node                 11.0s
   handed to asyncio.to_thread                 3.5s

Seven and a half seconds between them, on a call that takes two. Written the first way, the four runs’ legacy calls could not overlap each other — each one held the event loop while it ran, so the four took eight seconds between them, and the web lookups and model calls had to fit around them. Written the second way, the four calls sat on four worker threads at once, the event loop was free to run the other runs while they slept, and the whole thing came in under the four sleeps.

23.9 Pattern: too many calls in flight

Once the event loop can send everything at once, the question is whether it should.

A server that accepts four requests queues the fifth; a database has a connection limit; a hosted API returns 429 (HTTP’s “Too Many Requests”: the request was refused, not queued, and the client is expected to wait and retry), most often on a per-minute budget that a cap on calls in flight only approximates. The client-side half of a limit like those is a semaphore: a counter that await can wait on, so that at most N of a thing are in flight and the rest queue in the process instead of at the far end.

Section 6 puts a gate of two around ask, in a research that is otherwise the graph’s own, runs the four questions at once, and prints every call that had to wait for the gate:

code/ch21_async.py

async def a_ceiling(limit: int = 2) -> None:
    print(f"\n===== 6. too many calls in flight: at most {limit}, across {len(QUESTIONS)} runs =====")
    gate = asyncio.Semaphore(limit)

1    async def guarded(angle: str, question: str, docs: list[str]) -> str:
        who = f"question {QUESTIONS.index(question)}, {angle}"
        t0 = time.perf_counter()
2        async with gate:                       # takes the count, or waits here for it
            waited = time.perf_counter() - t0
            if waited > 0.01:
                print(f"   waited {waited:4.1f}s     : {who}")
            else:
                print(f"   went straight in : {who}")
            return await ask(angle, question, docs)

    async def research_capped(state: State) -> dict:
        """research_node with the gate around each model call."""
        *notes, web = await asyncio.gather(*(guarded(a, state["question"], state["docs"]) for a in ANGLES),
                                           web_lookup())
        return {"notes": [*notes, web]}

    g = build(research_capped)
    with timed(f"{len(QUESTIONS) * len(ANGLES)} research calls asked to start"):
        await asyncio.gather(*(g.ainvoke({"question": q}) for q in QUESTIONS))

1: guarded is ask with the gate around it. async with gate decrements the semaphore’s counter, or suspends the coroutine at that await until another one releases; the counter goes back up when the block exits, whether ask returned or raised.
2: async with, not with, because entering can wait. A semaphore whose counter is at zero has to hold the caller until it rises again, and a coroutine can only be held at an await.

code/ch21_async.py

===== 6. too many calls in flight: at most 2, across 4 runs =====
   went straight in : question 0, what it is
   went straight in : question 0, how to configure it
   waited  0.8s     : question 0, the most common mistake
   waited  1.0s     : question 1, what it is
   waited  1.6s     : question 1, how to configure it
   waited  1.8s     : question 1, the most common mistake
   waited  2.5s     : question 2, what it is
   waited  2.5s     : question 2, how to configure it
   waited  3.3s     : question 2, the most common mistake
   waited  3.6s     : question 3, what it is
   waited  4.3s     : question 3, how to configure it
   waited  4.6s     : question 3, the most common mistake
   12 research calls asked to start            6.4s

Four runs asked for twelve model calls between them. Two went straight in, both from question 0; the other ten waited, and the waits grow by about one call’s time at a step, which is the queue draining two at a time. The order the lines come out in is the order the gate let them in. In Chapter 22’s terms the semaphore turns a burst of callers into a queue the server can take: load control on the client side.

Deep Dive: what with gate: would do

asyncio.Semaphore puts its acquire behind __aenter__ and offers no plain __enter__: with gate: is a TypeError, not a slower version of the same thing.

threading.Semaphore does have one, and reaching for it here deadlocks: its acquire blocks the event loop’s thread, so the coroutines that took the count never get a turn to release it, and nothing moves again — not even a wait_for timeout wrapped around the whole thing, because firing that needs the event loop too.

23.10 `max_concurrency` and the superstep

LangGraph’s own scheduling has a cap of its own: graph.ainvoke(input, {"max_concurrency": 4}) runs at most four node tasks of one superstep at a time, and abatch honours the same key across its inputs.

That invites a question. A node fans out to seven tasks — Chapter 10’ Send — and the run is started with a cap of four. Does the fan-out become two supersteps, four tasks and then three?

It does not, and the difference matters more than the arithmetic: tasks in one superstep all read the state as it stood when the step began, so if the cap did split them, the last three would start by reading what the first four had written. The program that settles it needs no model. Seven Sends into one node, each task waiting four tenths of a second, run twice — once without the cap, once with it:

%%{init: {'flowchart': {'diagramPadding': 24, 'subGraphTitleMargin': {'top': 10, 'bottom': 10}}}}%%
flowchart LR
    START(["START"]) --> w0 & w1 & wd & w6
    subgraph step0 ["one superstep &nbsp;"]
        w0["work 0"]
        w1["work 1"]
        wd["…"]
        w6["work 6"]
    end
    w0 & w1 & wd & w6 --> END(["END"])

Two reducers carry the measurement. done is the ordinary one — each task appends the number it was sent. saw is the instrument: each task reports how much of done it could see at its own start, which is the number that answers the question.

code/ch21_max_concurrency.py

FANOUT = 7          # tasks in the one superstep
CAP = 4             # max_concurrency for the capped run
WORK = 0.40         # how long one task waits

T0 = 0.0            # start of the current run; the log's times are relative to it

class State(TypedDict):
    done: Annotated[list[int], operator.add]
    saw: Annotated[list[int], operator.add]      # what each task saw of `done` when it started

The task logs one line as it starts and then waits. The line is the whole instrument: the time says which wave it is in, and len(done) says what it was handed. Counting anything is left to the reader, who can see seven timestamps.

code/ch21_max_concurrency.py

async def work(state: dict) -> dict:
1    n, saw = state["n"], len(state["done"])
    print(f"      {time.perf_counter() - T0:4.1f}s  {threading.current_thread().name:<11}"
          f"  work {n}   saw len(done)={saw}")
    await asyncio.sleep(WORK)
    return {"done": [n], "saw": [saw]}

def build():
    """START -> seven `work` tasks -> END. Chapter 8's Send fan-out and nothing
    else: the seven tasks are the whole superstep."""
    g = StateGraph(State)
    g.add_node("work", work)
    g.add_conditional_edges(
        START,
2        lambda s: [Send("work", {"n": i, "done": s["done"]}) for i in range(FANOUT)],
        ["work"],
    )
    g.add_edge("work", END)
3    return g.compile(checkpointer=InMemorySaver())

1: What this task can see, read before it waits. If the cap split the step, the tasks in the second wave would be handed a done with the first wave’s four entries in it. The thread goes in the line too, free of charge: these are async def nodes, so all seven report the event loop’s own thread (Section 23.5).
2: Seven Sends straight off START. Nothing stands on either side of the fan-out, because neither node would be measuring anything: each Send carries the task’s own n and a copy of done as it stood at the start.
3: A checkpointer, because the count of checkpoints is evidence. One per superstep: if the cap split the fan-out, there would be one more.

Then the run. The cap is a key on the config — beside configurable, not inside it — and the span is timed by the chapter’s own timed, imported from ch21_async.py:

code/ch21_max_concurrency.py

async def one_run(cap: int | None) -> tuple[float, int, int, list[int]]:
    global T0
    graph = build()
    config: dict = {"configurable": {"thread_id": f"cap-{cap}"}}
    if cap is not None:
1        config["max_concurrency"] = cap

    print(f"\n   {f'max_concurrency={cap}' if cap else 'max_concurrency unset'}")
    T0 = time.perf_counter()
2    with timed("   wall clock") as run:
        out = await graph.ainvoke({"done": [], "saw": []}, config)

    history = list(graph.get_state_history(config))
3    fan_step = next(s for s in history if s.metadata and s.metadata["step"] == 0)
    print(f"      tasks in superstep 0 : {len(fan_step.next)}   {fan_step.next}")
    print(f"      checkpoints written  : {len(history)}")
    return run.seconds, len(fan_step.next), len(history), out["saw"]

1: The key under test, and the only difference between the two runs.
2: Section 23.3’ timed, imported from the chapter’s own file: it prints the label and the seconds, and leaves the seconds on run.seconds.
3: Step 0 is the superstep that holds the fan-out; step −1 is the input the run started from. fan_step.next names the tasks that step ran — the claim stated by the runtime itself, rather than inferred from timings.

main runs it twice and asserts what the two runs must agree on, so that a future version of LangGraph that changes any of this fails the file instead of quietly printing something else:

code/ch21_max_concurrency.py

    assert free_tasks == cap_tasks == FANOUT          # one superstep, seven tasks, either way
    assert free_ckpts == cap_ckpts                    # the cap costs no extra superstep
    assert free_saw == cap_saw == [0] * FANOUT        # nobody saw a neighbour's write
    assert cap_wall > free_wall * 1.5                 # it is paid for in wall clock: two waves

code/ch21_max_concurrency.py

===== 7 fan-out tasks, with and without a cap of 4 =====

   max_concurrency unset
       0.0s  MainThread   work 0   saw len(done)=0
       0.0s  MainThread   work 1   saw len(done)=0
       0.0s  MainThread   work 2   saw len(done)=0
       0.0s  MainThread   work 3   saw len(done)=0
       0.0s  MainThread   work 4   saw len(done)=0
       0.0s  MainThread   work 5   saw len(done)=0
       0.0s  MainThread   work 6   saw len(done)=0
      wall clock                               0.4s
      tasks in superstep 0 : 7   ('work', 'work', 'work', 'work', 'work', 'work', 'work')
      checkpoints written  : 3

   max_concurrency=4
       0.0s  MainThread   work 0   saw len(done)=0
       0.0s  MainThread   work 1   saw len(done)=0
       0.0s  MainThread   work 2   saw len(done)=0
       0.0s  MainThread   work 3   saw len(done)=0
1       0.4s  MainThread   work 4   saw len(done)=0
       0.4s  MainThread   work 5   saw len(done)=0
       0.4s  MainThread   work 6   saw len(done)=0
      wall clock                               0.8s
      tasks in superstep 0 : 7   ('work', 'work', 'work', 'work', 'work', 'work', 'work')
      checkpoints written  : 3

1: The second wave, four tenths of a second in — and still reading len(done)=0. These three started once the first four had finished and released their slots, and they were handed the state as it stood before any of them ran.

The timestamps are the two waves. Uncapped, seven tasks start together at 0.0s. Capped at four, four start at 0.0s and the other three at 0.4s — the moment the first four let go of their slots — and the run takes twice as long for it.

Everything else is the same in both columns, and the last line of each task is the one that settles the question: saw len(done)=0, seven times, including the three that started in the second wave. Had the cap split the step, those three would have begun after four writes had been applied and would have been handed a done with four entries in it. They were not. All seven read the state as it stood when the step began, all seven writes landed together at its end — which is why tasks in superstep 0 says seven and not four, and why the run writes the same three checkpoints either way.

Supersteps are decided by the graph’s topology; max_concurrency only decides how many of that step’s tasks run at the same moment. A cap of one would still be one superstep — seven tasks, taken one at a time, all reading the same input and landing in the same checkpoint.

One practical note for the day you reach for the key. It caps the graph’s tasks in one run. It does not reach inside a node: the three ask calls under one gather are Python’s concurrency and LangGraph never sees them, which is why Section 23.9 needed a semaphore. Nor does it cap the other runs in the same process, each of which brings its own, or the thread pool that Section 23.5’s plain def nodes draw from.

23.11 Pattern: a run that must stop

A caller gave up, a request has a deadline, a node is stuck on something that will never answer. Async has two places to put a limit, and the tools differ in what they leave behind.

Where	How you write it	What happens when it fires
around the whole run	`await asyncio.wait_for(graph.ainvoke(...), timeout=1)`	the task it wraps is cancelled, and `TimeoutError` reaches the caller
the same, cancelling nothing	`done, pending = await asyncio.wait([task], timeout=1)`	nothing is raised and nothing is cancelled; what has not finished comes back in `pending`, still running
around one node	`add_node(..., timeout=1)`, or a `TimeoutPolicy`	that node is cancelled and raises `NodeTimeoutError`

asyncio.wait takes tasks, not coroutines — asyncio.create_task(...) first, or it raises TypeError: Passing coroutines is forbidden.

TimeoutPolicy is the longer form of the node timeout: run_timeout is a wall-clock cap on one attempt, idle_timeout a cap on time without progress.

Two stand-ins for research, one per case: one that waits the event loop’s way, and one that takes the same wait on the event loop’s thread.

code/ch21_async.py

SLOW = 5                   # sections 7 and 8: a wait long enough to cut short


async def slow_research(state: State) -> dict:
    """An async research that waits too long: the timeout is meant to cut it off."""
    await asyncio.sleep(SLOW)
    return {"notes": []}


async def blocking_research(state: State) -> dict:
    """The same wait, taken on the event loop's thread instead of behind an await."""
    time.sleep(SLOW)
    return {"notes": []}

code/ch21_async.py

async def a_run_that_must_stop() -> None:
    """Two timeouts: one around the whole run, one around the node."""
    print("\n===== 7. a run that must stop: a timeout around the run, and around a node =====")
    g = build(research=slow_research, checkpointer=InMemorySaver())
    config = {"configurable": {"thread_id": "t-stop"}}
    t0 = time.perf_counter()
    try:
        await asyncio.wait_for(g.ainvoke({"question": QUESTION}, config), timeout=1)
    except TimeoutError:
1        saved = await g.aget_state(config)         # where the cancel actually left the thread
        print(f"   {'wait_for the whole run':<22} -> TimeoutError after "
              f"{time.perf_counter() - t0:.1f}s; state has {sorted(saved.values)}, next={saved.next}")
    await node_timeouts()

1: A checkpointer, so the cancel leaves something to look at. Without one the run would simply be gone; with one the thread can be read back, and .next is the runtime’s own account of where it stopped.

code/ch21_async.py

async def node_timeouts() -> None:
    """What a node timeout can cut off: only a node that gives the event loop a chance."""
    for label, node in (("timeout, node awaits", slow_research),
                        ("timeout, node blocks", blocking_research)):
        g = build(research=node, timeout=TimeoutPolicy(run_timeout=1))
        try:
            await g.ainvoke({"question": QUESTION})
            print(f"   {label:<22} -> no error: the node ran its full {SLOW}s and returned")
        except NodeTimeoutError as exc:
            print(f"   {label:<22} -> {type(exc).__name__}: {str(exc).splitlines()[0]}")

code/ch21_async.py

===== 7. a run that must stop: a timeout around the run, and around a node =====
   wait_for the whole run -> TimeoutError after 1.0s; state has ['docs', 'question'], next=('research',)
   timeout, node awaits   -> NodeTimeoutError: Node 'research' exceeded its run timeout of 1.000s (elapsed: 1.003s).
1   timeout, node blocks   -> no error: the node ran its full 5s and returned

1: The deadline passed with nobody able to act on it. A timeout is a cancellation, and a cancellation is delivered at an await. This node has no await: time.sleep holds the event loop’s thread for the whole five seconds, so the timer that should have fired at one second could not run until the node had already returned.

The run-level timeout lands where the run was waiting: search_docs finished and is on the thread, research did not and is what .next names. The node-level one fires on a node that awaits, a second into a five-second wait. On a node that blocks the event loop it never fires at all — there is nowhere to deliver the cancellation, so the node runs to its end and returns as if nothing had been asked of it.

A timed-out node is a failed node, so Section 21.2’s RetryPolicy applies to it: with max_attempts=3 a node that times out is tried three times, and the NodeTimeoutError from the last attempt is what the caller sees.

23.12 Pattern: state that outlives the run

Everything above kept its state in memory. A server that can be restarted, a run that a caller comes back to, a cancel you intend to resume from: all of them need the checkpointer of Chapter 8, and the async door is picky about which one.

code/ch21_async.py

async def state_that_outlives_the_run() -> None:
    print("\n===== 8. state that outlives the run =====")
    conn = sqlite3.connect(":memory:", check_same_thread=False)
    g = build(checkpointer=SqliteSaver(conn))
    try:
        await g.ainvoke({"question": QUESTION}, {"configurable": {"thread_id": "t1"}})
    except NotImplementedError as exc:
        print(f"   SqliteSaver + ainvoke      -> NotImplementedError: {str(exc).splitlines()[0]}")   # the rest is the install hint
    conn.close()

    g = build(checkpointer=InMemorySaver())      # InMemorySaver serves both doors
    await g.ainvoke({"question": QUESTION}, {"configurable": {"thread_id": "t1"}})
    print("   InMemorySaver + ainvoke    -> fine; it serves both doors, and forgets at exit")

    async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:   # the one that persists
        g = build(checkpointer=saver)
        await g.ainvoke({"question": QUESTION}, {"configurable": {"thread_id": "t1"}})
        saved = await g.aget_state({"configurable": {"thread_id": "t1"}})
    print(f"   AsyncSqliteSaver + ainvoke -> fine; checkpoint holds {len(saved.values['notes'])} notes")

code/ch21_async.py

===== 8. state that outlives the run =====
   SqliteSaver + ainvoke      -> NotImplementedError: The SqliteSaver does not support async methods. Consider using AsyncSqliteSaver instead.
   InMemorySaver + ainvoke    -> fine; it serves both doors, and forgets at exit
   AsyncSqliteSaver + ainvoke -> fine; checkpoint holds 4 notes

SqliteSaver raises, and that first line is the whole of the mistake. Its async methods exist only for that, and ainvoke calls the first of them when it loads the thread’s last checkpoint, before any node runs. The block shows the message’s first line; the rest of it is the install hint for aiosqlite.
InMemorySaver answers both doors, which is why the earlier chapters never noticed, and forgets everything when the process ends.
AsyncSqliteSaver is the one that persists: an async context manager with the same from_conn_string shape as its sync twin, and aget_state beside the methods Chapter 8 used. The production rung of that ladder is AsyncPostgresSaver.

That is also what lets a run outlive the request that started it, which is what this chapter’s Challenge builds.

Gotcha: a coroutine nobody awaited

ask(...) without the await is a coroutine object that was never started. Nothing was sent.

code/ch21_async.py

async def research_forgot_await(state: State) -> dict:
1    notes = [ask(a, state["question"], state["docs"]) for a in ANGLES]   # BUG: no await
    return {"notes": notes}                    # three unstarted coroutines go into state

1: This line is valid Python and raises nothing. ask(...) without the await builds a coroutine instead of calling the model, three times over, and the failure waits for whatever touches them next.

code/ch21_async.py

async def a_coroutine_never_awaited() -> None:
    print("\n===== 9. a coroutine nobody awaited =====")
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always", RuntimeWarning)
        try:
            await build(research_forgot_await).ainvoke({"question": QUESTION})
        except TypeError as exc:
            print(f"   research forgot its awaits -> TypeError, raised in generate: {exc}")
1        gc.collect()                           # the dropped coroutines are collected now ...
        print(f"   ... and only now Python warns: {caught[-1].message}")

1: The warning comes from the garbage collector, not from the line that forgot the await. gc.collect() forces the sweep here so the warning lands on the next line; in real code it arrives wherever the objects happen to die.

code/ch21_async.py

===== 9. a coroutine nobody awaited =====
   research forgot its awaits -> TypeError, raised in generate: sequence item 0: expected str instance, coroutine found
   ... and only now Python warns: coroutine 'ask' was never awaited

23.13 Back on a laptop

Every number above came from a hosted model, gpt-4.1-mini, reached with python tools/use_llm.py openai from Appendix A. That is the setting async was built for: a far end that answers many requests at once, so the only thing in the way was this process.

The rest of this book runs on the local model, and it is worth seeing what these same measurements look like there, because the answer is not “a bit less”. Section 4, four runs at once, against LM Studio on the book’s machine:

code/ch21_async.py

===== 4. one process, many callers: 4 runs =====
   one after another                          31.2s
   at once (asyncio.gather)                   28.5s   (1.1x faster than the first)
   the gain is bounded by how many requests the model server runs at once, not by this process

A tenth, where the hosted model gave nearly four times. The event loop did exactly the same work in both: sixteen model calls, twelve of them in flight at the same moment, no lock and no copy. What changed is that the far end is one GPU, and one GPU generating four answers together is not much faster than generating them in turn. The ratio is not a property of your code. It is a property of what answers it.

Turning the server’s dial does not rescue it either. LM Studio’s Max Concurrent Predictions, in the loader’s advanced settings (lms ps prints it as PARALLEL), is four by default. Set it to one and section 4’s two timings come out the same, which is async against a server that cannot overlap at all. Raise it to twelve, so every call is admitted at once, and on this machine the four runs moved from 28.7 s to 28.6 s. The engine moves it a little: the same model as a GGUF under llama.cpp, with its continuous batching, took the three gathered ask calls from 1.2x to 1.7x and the four runs to about 25 s, and raising its slot count from four to twelve changed nothing either. Flash attention and a quantised KV cache changed nothing at this context length.

The reason is underneath all of them. Generating one token reads the whole model’s weights, and this model’s are about 15 GB on disk at 4 bits. One machine’s memory bandwidth is the budget, and four requests at once spend it four ways. Batching helps when the GPU is waiting on arithmetic; it does not help when the GPU is waiting on memory, which is where a single-user decode sits.

So the two situations of Section 23.1 survive the move to a laptop, but only one of them keeps its payoff:

	hosted	local
a node waiting on several different things	1.4 s back, the lookup’s whole wait	the same: it was never the GPU’s to give
a node’s three model calls at once	all three inside the lookup’s 1.5 s	a fifth, and the server decides
four callers’ runs at once	3.9x	1.1x

The first row is the event loop’s own saving, and it is yours either way. The other two are the server’s, and a laptop has none to spare. Everything else in this chapter — which door a node answers, which thread it runs on, what a blocking call costs, what a timeout can cut — is unchanged, because none of it depends on the model at all.

23.14 The same shape elsewhere

The pattern here is the one every I/O-bound server in every language settled on: a single thread that never blocks, waiting on many sockets at once, and a scheduler that resumes whichever one is ready. Node’s event loop is it; so is nginx’s worker, and so is the readiness poll the operating system offers under both. What LangGraph adds is the contract in Section 23.5: a unit of work declares whether it waits the event loop’s way, and the runtime puts it where that is safe. The failure in Section 23.8 is also the same everywhere — a blocking call on the event loop thread — and in every one of those systems it is the bug that is found in production, under load, by a graph of latency climbing.

23.15 What we learned

Here is a summary of what to reach for when async could help:

When you are here	Reach for
a node waits on several things	`asyncio.gather` over the awaits
one process serves several callers	one `ainvoke` per run, one compiled graph
a client that cannot await	`await asyncio.to_thread(call, ...)`
the far end has a limit	`asyncio.Semaphore` around the call
a run or a node must stop	`asyncio.wait_for`, `add_node(timeout=)`
state must outlive the run	`AsyncSqliteSaver`, `AsyncPostgresSaver`

New pieces in this chapter

Piece	Usage
*API*
`graph.ainvoke()` / `graph.abatch()`	The async door into a compiled graph; the only door once any node is `async def`. `await graph.ainvoke({"question": q})`; `invoke` on such a graph raises `TypeError: No synchronous function provided`. `abatch` is not run here
`llm.ainvoke()`	A model call that yields to the event loop while the model works. `reply = await llm.ainvoke(prompt)`; forgetting the `await` leaves an unstarted coroutine
`asyncio.gather()`	Run several coroutines at once and collect their results in order. `await asyncio.gather(*(ask(a, q, docs) for a in ANGLES), web_lookup())`; the time is the slowest, not the sum
`asyncio.Semaphore`	A counter `await` can wait on: at most N of something in flight. `async with gate:` around `llm.ainvoke`
`asyncio.to_thread()`	Run a blocking call on a worker thread and get an awaitable. `await asyncio.to_thread(legacy_tool, q)`; the fix for a sync client you cannot rewrite
`AsyncSqliteSaver`	The async twin of `SqliteSaver`, for a file on one machine. `async with AsyncSqliteSaver.from_conn_string(path) as saver:`; `SqliteSaver` raises `NotImplementedError` under `ainvoke`. The production rung is `AsyncPostgresSaver`, same shape
`max_concurrency`	A config key, not a `configurable` entry: `graph.ainvoke(x, {"max_concurrency": 4})` caps the node tasks of one superstep and the inputs of `abatch`; it does not reach the awaits inside a node
`asyncio.wait_for()`	A timeout around a run; cancels it at its next `await`. `await asyncio.wait_for(graph.ainvoke(...), timeout=5)`; with a checkpointer the finished steps stay

Code in this chapter

File	What it runs
`code/ch21_async.py`	The doc-bot’s spine with an async `research` node, run through `ainvoke` nine ways: the two doors, where each node’s thread is, the cost of waiting, four runs at once, a library that blocks, a semaphore around the model call, two timeouts, three checkpointers, and a coroutine nobody awaited.
`code/ch21_max_concurrency.py`	Seven `Send` tasks run with and without a cap of four — the peak, the wall clock, the checkpoints, and what each task could see of its neighbours. No model: every call is a sleep, and the claim is asserted rather than printed.

Challenge

🥈 Silver — a deadline for the request, not for the run

Chapter 22’s /ask holds the connection until the graph is done. A caller will not wait thirty seconds for that, but they still want the answer. Those are two different limits, and this challenge is about not confusing them.

Give /ask a deadline of a second or two. If the graph finishes inside it, answer as before. If it does not, free the connection with 202 Accepted and a run id, and let the run carry on, so that GET /run/{id} can hand over the answer when the caller comes back for it.

The standard library has more than one way to put a deadline on a coroutine, and Section 23.11’s table names two. Try both.

office hours Worked solution with tests: code/answers/ch21_deadline_without_cancel.py. A real FastAPI app, driven in-process through httpx’s ASGI transport, so there is no server to start.

The Global Interpreter Lock (GIL): a lock in the standard CPython interpreter that one thread must hold to run Python bytecode, so two threads never execute Python code at the same moment; a thread releases it while it waits on I/O, and C extensions such as numpy release it during their own computation. A free-threaded build without the GIL has existed since Python 3.13 (PEP 703) and is a separate, opt-in build; the book’s venv is the standard one.↩︎
functools.partial fixes the leading arguments now and takes the rest at call time. run_in_executor is run_in_executor(executor_or_config, func, *args), so with the first two fixed, calling the slot as afunc(state) runs run_in_executor(None, search_docs_node, state). None there asks for the event loop’s default thread pool.↩︎

23.1 When to use

23.2 Just enough asyncio

23.3 The graph, and the three layers

23.4 invoke is closed, ainvoke is open

23.5 Where each node runs

23.6 Situation one: a node that waits on several things

23.7 Situation two: one process, many callers

23.8 Pattern: a library that blocks

23.9 Pattern: too many calls in flight

23.10 max_concurrency and the superstep

23.11 Pattern: a run that must stop

23.12 Pattern: state that outlives the run

23.13 Back on a laptop

23.14 The same shape elsewhere

23.15 What we learned

New pieces in this chapter

Code in this chapter

Challenge

🥈 Silver — a deadline for the request, not for the run

23.4 `invoke` is closed, `ainvoke` is open

23.10 `max_concurrency` and the superstep