LangChain Async Support: Building High-Throughput Applications

Why your LangChain app stalls under real traffic

Most LangChain tutorials teach you to call .invoke() and move on. That works fine on your laptop with one request at a time. It falls apart the moment you put your chain behind a FastAPI endpoint and ten users hit it in the same second. Each synchronous call to an LLM provider blocks a thread while it waits on a network round trip that can take anywhere from a few hundred milliseconds to tens of seconds. Multiply that by concurrent users, tool calls, retriever lookups, and chained sub-calls, and you get a server that looks fine in a demo and falls over in production.

The fix is not "add more servers." The fix is understanding that an LLM call is fundamentally an I/O-bound operation — your CPU is idle for almost the entire duration, waiting on a socket. That is exactly the problem async I/O was built to solve, and LangChain has first-class support for it across nearly every component: models, chains, retrievers, tools, and agents. Most developers never touch it because the sync API is the one shown in the quickstart. This article is about closing that gap — what ainvoke, abatch, astream, and asyncio.gather actually buy you, where the gains are real versus cosmetic, and how to wire it into a FastAPI service without shooting yourself in the foot with blocking calls hidden three layers deep in a custom tool.

We'll build up from a single async call to a batched, concurrency-controlled pipeline, then talk about the failure modes that only show up once you're running this in front of real traffic: connection pool exhaustion, sync code accidentally blocking your event loop, and how to reason about concurrency limits so you don't get rate-limited into oblivion.

Sync vs async: what's actually different under the hood

Every runnable in LangChain — a ChatModel, a Chain, a Retriever, a Tool — implements the Runnable interface, which ships with both sync and async entry points: invoke/ainvoke, batch/abatch, stream/astream. The sync versions run on the current thread and block until the underlying HTTP call returns. The async versions are coroutines — they yield control back to the event loop while waiting on the network, which means your process can be doing useful work (serving other requests, kicking off other LLM calls) during that wait instead of sitting idle.

Here's the same call written both ways:

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Synchronous — blocks the thread until the response arrives
def sync_call(question: str) -> str:
    response = model.invoke(question)
    return response.content

# Asynchronous — yields control while waiting on the network
async def async_call(question: str) -> str:
    response = await model.ainvoke(question)
    return response.content

On its own, async_call isn't faster than sync_call for a single request — you're still waiting on the same network round trip. The win shows up when you have *many* of these to run at once. With sync code, running ten calls means running them one after another, or spinning up ten threads and dealing with the GIL and thread pool overhead. With async code, you can launch ten coroutines and let the event loop interleave them, and the wall-clock time for all ten approaches the time for *one*, not ten times that.

A common misconception: async does not make an individual LLM call faster. It does not reduce token generation time or network latency. What it changes is how many of those calls you can have in flight at once without proportionally increasing wall-clock time or server resource usage. If your workload is "one user, one question, wait for the answer," async buys you very little. If your workload is "process 500 documents" or "serve 200 concurrent chat users," async is the difference between a service that scales and one that doesn't.

ainvoke: the building block

ainvoke is the async equivalent of invoke and exists on every Runnable, including chains built with LangChain Expression Language (LCEL). Because LCEL chains compose Runnables with the pipe operator, calling .ainvoke() on the composed chain automatically calls .ainvoke() on each link, provided every link actually implements an async path (more on the components that don't, later).

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise technical writer."),
    ("human", "Explain {topic} in two sentences."),
])
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
parser = StrOutputParser()

chain = prompt | model | parser

async def explain(topic: str) -> str:
    return await chain.ainvoke({"topic": topic})

Nothing about the chain definition changes between sync and async usage — that's the point of the Runnable interface. You write the chain once, and the caller decides whether to run it with invoke or ainvoke. This matters for library code: if you're building a reusable chain that other developers will import, you get both call styles for free without writing two versions of your pipeline.

The place this gets interesting is when your chain includes a custom step — a RunnableLambda wrapping your own function, or a custom retriever hitting a database. If that custom function is a regular def doing blocking I/O (a requests.get, a synchronous database driver call), wrapping it in ainvoke doesn't make it async. It just runs the blocking call inside the event loop, which blocks *everything* else scheduled on that loop — arguably worse than not using async at all, because now you've hidden the blocking behavior behind an await that looks async but isn't.

import httpx
from langchain_core.runnables import RunnableLambda

# WRONG: looks async-compatible, but requests.get blocks the event loop
def fetch_context_sync(query: str) -> str:
    import requests
    resp = requests.get(f"https://internal-api/search?q={query}")
    return resp.json()["summary"]

# RIGHT: an actual async function using an async HTTP client
async def fetch_context_async(query: str) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://internal-api/search?q={query}")
        return resp.json()["summary"]

retriever_step = RunnableLambda(fetch_context_async)

RunnableLambda will pick up an async function and route it through ainvoke correctly. Pass it a sync function and call .ainvoke() on the chain, and LangChain will run that sync function in a thread pool executor behind the scenes to avoid blocking the loop directly — which works, but reintroduces thread pool overhead and defeats some of the purpose. If you're writing custom steps for a chain that will run under load, write them as async def using async-native libraries (httpx, asyncpg, motor, async ORMs) from the start.

abatch: processing many inputs without writing your own loop

If you have a list of inputs to process — a batch of documents to summarize, a set of support tickets to classify — the naive approach is a for loop calling ainvoke on each one with asyncio.gather. LangChain gives you this pattern built-in via abatch, which also applies concurrency limits so you don't accidentally fire 500 requests at your LLM provider simultaneously and get rate-limited.

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "Classify the support ticket as billing, technical, or account."),
    ("human", "{ticket}"),
])
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | model | StrOutputParser()

tickets = [
    "My card was charged twice this month.",
    "The app crashes when I upload a PDF.",
    "I can't reset my password.",
    "Why is my invoice showing the wrong plan?",
]

async def classify_all(tickets: list[str]) -> list[str]:
    inputs = [{"ticket": t} for t in tickets]
    results = await chain.abatch(inputs, config={"max_concurrency": 5})
    return results

results = asyncio.run(classify_all(tickets))
for ticket, label in zip(tickets, results):
    print(f"{label.strip():<12} -> {ticket}")

The max_concurrency parameter in the config dict is doing real work here — it caps how many of those abatch calls are in flight at once, regardless of how many inputs you passed in. Without it, abatch will try to fire all requests concurrently, which for a batch of 500 tickets against an API with a 60 requests-per-minute limit is a guaranteed RateLimitError storm. Setting max_concurrency to something sane (5–20 depending on your provider's tier) turns "fire everything at once and hope" into a controlled, self-throttling batch job.

abatch also preserves input order in the output list, which matters if you're zipping results back against the original inputs like in the example above — you don't have to track indices yourself.

For very large batches where you also want to react to results as they land (rather than waiting for the whole batch to finish), abatch_as_completed is the better tool:

async def classify_streaming(tickets: list[str]):
    inputs = [{"ticket": t} for t in tickets]
    async for idx, result in chain.abatch_as_completed(inputs, config={"max_concurrency": 5}):
        print(f"Ticket {idx} classified as: {result.strip()}")

This yields (index, result) tuples as each one finishes, in completion order rather than input order — useful for a live dashboard or a job queue where you want to persist results incrementally instead of holding everything in memory until the last request comes back.

astream and astream_events: getting tokens out the door faster

Throughput isn't only about handling more requests — it's also about perceived latency for a single request. If a user asks a question and your chain takes eight seconds to produce a full answer, ainvoke makes them stare at a blank screen for eight seconds. astream starts yielding chunks as the model produces them, so you can pipe tokens to the client as they're generated.

async def stream_answer(question: str):
    async for chunk in chain.astream({"topic": question}):
        print(chunk, end="", flush=True)

In a web context, this is what lets you build the token-by-token typing effect users expect from chat interfaces. Wired into FastAPI, it looks like this:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/ask")
async def ask(payload: dict):
    async def token_generator():
        async for chunk in chain.astream({"topic": payload["topic"]}):
            yield chunk

    return StreamingResponse(token_generator(), media_type="text/plain")

For chains with multiple steps — retrieval, then generation, then a formatting step — plain astream only shows you output from the final Runnable. If you need visibility into intermediate steps (which documents the retriever pulled, when the model actually started generating versus when a tool call kicked off), astream_events gives you a structured event stream instead of raw text chunks:

async def trace_execution(question: str):
    async for event in chain.astream_events({"topic": question}, version="v2"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            print(event["data"]["chunk"].content, end="")
        elif kind == "on_retriever_end":
            docs = event["data"]["output"]
            print(f"\n[retrieved {len(docs)} documents]")

This is the tool you reach for when debugging why a RAG chain feels slow — you can see exactly how much wall-clock time is spent in retrieval versus generation versus any tool calls in between, instead of guessing from a single aggregate latency number.

Concurrency control: asyncio.gather, semaphores, and provider rate limits

abatch's max_concurrency covers the common case, but once you start composing your own async workflows outside a single chain call — say, fanning out to three different chains per request, or combining retrieval from two vector stores — you need to manage concurrency yourself with asyncio.gather and a semaphore.

import asyncio

async def process_document(doc_id: str, semaphore: asyncio.Semaphore):
    async with semaphore:
        summary = await summarize_chain.ainvoke({"doc_id": doc_id})
        tags = await tagging_chain.ainvoke({"doc_id": doc_id})
        return {"doc_id": doc_id, "summary": summary, "tags": tags}

async def process_all(doc_ids: list[str]):
    semaphore = asyncio.Semaphore(10)  # at most 10 documents in flight
    tasks = [process_document(doc_id, semaphore) for doc_id in doc_ids]
    return await asyncio.gather(*tasks, return_exceptions=True)

Two details worth calling out. First, the semaphore wraps the *whole* per-document workflow, not just one LLM call — so each of the ten "slots" represents one document's full pipeline (summarize, then tag), not ten calls to a single chain. This matters if your per-item work has multiple sequential model calls; without wrapping the whole unit, you can end up with far more concurrent requests than the semaphore count suggests. Second, return_exceptions=True on gather means one failed document doesn't cancel the other nine — you get the exception object back in the results list instead of the gather call raising immediately. You still need to check each result for whether it's an Exception before using it, but at least a single flaky API response doesn't take out your whole batch.

Picking the right concurrency number is mostly about your provider's rate limits, not your own hardware. Check your tier's requests-per-minute and tokens-per-minute limits, and set your semaphore or max_concurrency comfortably under that ceiling — remember that a burst of concurrent requests can trip a rate limiter even if your average usage over a minute is fine. Most teams find a good starting point through trial: start at 5, watch for 429 errors, and increase gradually while monitoring actual throughput, since the "right" number depends on your specific provider tier, average token count per request, and how many other services share that same API key's rate limit budget.

Wiring it into a FastAPI service

The full value of async LangChain shows up when it's serving concurrent HTTP requests, because FastAPI's own request handling is already async — using sync invoke calls inside an async def endpoint blocks FastAPI's event loop for every other request being served by that worker, which defeats the purpose of using an async framework in the first place.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

app = FastAPI()

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant for a course platform."),
    ("human", "{question}"),
])
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
chain = prompt | model | StrOutputParser()

class Question(BaseModel):
    question: str

@app.post("/chat")
async def chat(payload: Question):
    try:
        answer = await chain.ainvoke({"question": payload.question})
    except Exception as exc:
        raise HTTPException(status_code=502, detail=str(exc))
    return {"answer": answer}

This alone lets a single worker process handle many concurrent /chat requests without spinning up a thread per request, because the worker is free to work on other requests while any given ainvoke call is waiting on the model provider. Combined with an ASGI server like Uvicorn running multiple worker processes, you get concurrency at two levels: async I/O within a worker, and multi-process parallelism across workers for CPU-bound work like local tokenization or embedding pre-processing.

One more subtlety: if part of your endpoint logic calls into synchronous code you don't control — a legacy database ORM, a third-party SDK that only ships a sync client — don't call it directly inside an async def handler. Push it into a thread with asyncio.to_thread so it doesn't block the loop:

import asyncio

@app.post("/chat-with-legacy-lookup")
async def chat_with_lookup(payload: Question):
    user_context = await asyncio.to_thread(legacy_sync_lookup, payload.question)
    answer = await chain.ainvoke({"question": payload.question, "context": user_context})
    return {"answer": answer}

This is a common pattern once real projects mix async LangChain code with older synchronous infrastructure — you don't have to rewrite the legacy piece to get most of the benefit, you just have to make sure it doesn't run directly on the event loop thread.

Common pitfalls that quietly kill your throughput gains

A few mistakes show up repeatedly in production LangChain code, and each one silently erases the benefit of switching to async in the first place.

• Calling `.invoke()` inside an async function. This is the single most common mistake. Someone writes async def get_answer(): return chain.invoke(...) instead of await chain.ainvoke(...), and the function is technically a coroutine but behaves exactly like sync code — it blocks the event loop for the full duration of the call.
• Wrapping custom retrievers or tools with blocking libraries. A custom tool that calls a synchronous database driver or requests inside what's meant to be an async chain will block the loop even though the surrounding chain uses ainvoke. Use async-native clients (asyncpg, httpx, motor) for any custom step that will run inside an async chain.
• Ignoring `max_concurrency` on large batches. Firing 200 unthrottled requests at once doesn't make things 200x faster — it usually triggers rate limiting, and you end up with retries and backoff eating any time you saved.
• Not handling partial failures in `gather`. Without return_exceptions=True, one failed call cancels the whole batch, and you lose results that had already completed successfully.
• Mixing sync and async model clients inconsistently. If part of your codebase constructs a chain and calls .invoke() in one place and .ainvoke() in another, you get inconsistent connection pooling behavior and it becomes hard to reason about how many concurrent connections your process is actually holding open to the provider.
• Forgetting that streaming and batching are different axes. astream improves perceived latency for one request; abatch/asyncio.gather improve total throughput for many requests. Conflating the two leads to solutions that optimize the wrong metric — for example, streaming tokens for a batch job that's actually bottlenecked on concurrency limits, not perceived latency.

None of these are exotic — they're the kind of thing that passes code review because the code "looks async" (it has async/await scattered around) without actually behaving that way end to end. The fix is almost always the same: trace one request through the entire call path and check that every I/O-bound step in it is a genuine await on an async-native operation, not a sync call quietly running in the middle of your event loop.

Putting it together: a realistic high-throughput pattern

A pattern that holds up well for real workloads — document processing pipelines, batch classification jobs, RAG endpoints under concurrent user load — combines everything above: ainvoke for individual steps, abatch or gather-with-semaphore for volume, max_concurrency tuned to your provider's actual limits, and astream at the edge wherever a human is waiting on the response.

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
summarize_prompt = ChatPromptTemplate.from_messages([
    ("system", "Summarize the document in one paragraph."),
    ("human", "{document}"),
])
summarize_chain = summarize_prompt | model | StrOutputParser()

async def summarize_corpus(documents: list[str]) -> list[str]:
    inputs = [{"document": doc} for doc in documents]
    return await summarize_chain.abatch(inputs, config={"max_concurrency": 8})

async def main():
    documents = ["... document text 1 ...", "... document text 2 ..."]
    summaries = await summarize_corpus(documents)
    for i, summary in enumerate(summaries):
        print(f"Document {i}: {summary}\n")

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

Scale this pattern up and the same shape holds whether you're processing 20 documents or 20,000 — the only thing that changes is max_concurrency, and possibly whether you reach for abatch_as_completed so a long-running job can checkpoint partial progress instead of waiting for the entire batch before writing anything to disk.

Async support in LangChain isn't a performance trick bolted onto the framework — it's built into the Runnable interface from the ground up, which is why chains, models, retrievers, and tools all expose the same ainvoke/abatch/astream triplet. The work of building a high-throughput application isn't learning new LangChain APIs so much as it's learning to write every custom piece of your pipeline — retrievers, tools, database calls — as genuinely async code, and then trusting the framework to compose it correctly.

If you want to go deeper into this — including async agent executors, concurrent tool calling, and how to profile where an async chain is actually spending its time — that's exactly the kind of production detail we cover hands-on in the LangChain Tutorial 2026 course on teachyou.ai, where we build a full async RAG service from scratch and load-test it under concurrent traffic.