How to Fix 'async event loop error when scaling' in LangChain (Python)

If you see RuntimeError: This event loop is already running or asyncio.run() cannot be called from a running event loop while scaling a LangChain app, it usually means you mixed sync and async code in the wrong place. This shows up when you move from a single request to concurrent workers, FastAPI endpoints, notebooks, Celery tasks, or background jobs.

In LangChain, the failure is usually not “LangChain is broken.” It’s almost always your app calling an async chain from a context that already has an event loop, or creating nested loops during concurrency.

The Most Common Cause

The #1 cause is wrapping async LangChain calls with asyncio.run() inside code that is already running inside an event loop.

This happens a lot when people take working notebook code and drop it into FastAPI, Streamlit, or an async worker.

# WRONG
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("Summarize: {text}")
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm

async def handle_request(text: str):
    # RuntimeError: asyncio.run() cannot be called from a running event loop
    result = asyncio.run(chain.ainvoke({"text": text}))
    return result

# RIGHT
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("Summarize: {text}")
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm

async def handle_request(text: str):
    result = await chain.ainvoke({"text": text})
    return result

If you are in sync code and need to call async LangChain code, move the boundary up instead of nesting loops:

# Sync entrypoint calling async code once at the top level
import asyncio

def main():
    return asyncio.run(handle_request("hello"))

if __name__ == "__main__":
    print(main())

Other Possible Causes

1) Calling .invoke() on an async-only path inside an async server

Some LangChain components work fine synchronously until you introduce tools, retrievers, or callbacks that expect async behavior. Then .invoke() can block badly or trigger loop-related issues indirectly.

# Problematic in async apps
result = chain.invoke({"text": "hello"})

Use the async API in async contexts:

result = await chain.ainvoke({"text": "hello"})

2) Using Jupyter or IPython with asyncio.run()

Notebook environments already run an event loop. That’s why this error often appears during local testing before production.

# WRONG in Jupyter
import asyncio
response = asyncio.run(chain.ainvoke({"text": "hello"}))

Use direct await in notebooks:

response = await chain.ainvoke({"text": "hello"})

3) Mixing thread pools with shared async clients

If you fan out work with ThreadPoolExecutor and reuse the same async LangChain client or callback manager across threads, you can hit loop ownership issues.

from concurrent.futures import ThreadPoolExecutor

def worker(text):
    # Bad if this calls async internals from thread context incorrectly
    return chain.invoke({"text": text})

with ThreadPoolExecutor(max_workers=20) as pool:
    results = list(pool.map(worker, texts))

Prefer one concurrency model. For async-heavy workloads, use asyncio.gather():

results = await asyncio.gather(*(chain.ainvoke({"text": t}) for t in texts))

4) Async callbacks or retrievers created outside the running loop

Some objects capture loop state at construction time. If you instantiate them globally and then reuse them across requests, they may hold stale state under load.

# Risky global setup in some server setups
retriever = vectorstore.as_retriever()
chain = retriever | llm

Create request-scoped objects when needed:

async def build_chain():
    retriever = vectorstore.as_retriever()
    return retriever | llm

How to Debug It

1. •

   Read the exact exception text

   • •RuntimeError: This event loop is already running
   • •RuntimeError: asyncio.run() cannot be called from a running event loop
   • •Task got Future attached to a different loop

   These point to different failure modes. The first two are nested-loop problems; the last one is usually cross-loop object reuse.

2. •

   Search for every asyncio.run()

   • •Remove it from anything except your top-level process entrypoint.
   • •If it appears inside FastAPI route handlers, Celery tasks, callbacks, or notebook cells, that’s your bug.
3. •

   Check whether your call stack is already async

   • •FastAPI route with async def
   • •Any function using await
   • •Any notebook cell using IPython

   In those cases, use await chain.ainvoke(...), not asyncio.run(...).

4. •

   Test with one minimal path Strip everything down to one chain call:

result = await chain.ainvoke({"text": "ping"})
print(type(result), result)

If this works alone but fails in your app, the problem is around it: worker model, callback setup, retriever lifecycle, or thread usage.

Prevention

• •

  Keep one concurrency model per service.

  • •If the request path is async, stay async all the way down.
  • •Don’t mix thread pools, sync invokes, and nested event loops unless you have a hard reason.
• •

  Use LangChain’s matching API:

  • •.invoke() for sync code at the edge of your app.
  • •.ainvoke() for async request handlers and workers.
  • •Same rule for batch methods: .batch() vs .abatch().
• •

  Don’t create long-lived global objects that depend on request-scoped runtime state.

  • •Build chains cleanly per process startup.
  • •Build request-specific retrievers/tools inside the request flow if they touch async resources.

If your error says This event loop is already running, start by removing every nested asyncio.run() call. In LangChain Python apps, that fixes most scaling failures fast.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.