Async and Concurrency Architecture: Fixing AI's Synchronous Blocking Patterns to Improve System Throughput

There is a very common problem with AI-generated code:

The code runs, but slows down as soon as multiple users access it.

Especially in Python Web projects, even when using FastAPI and functions are written as async def, operations such as database queries, HTTP requests, and large language model (LLM) calls may still block synchronously.

The key issue is:

async def is merely the entry point of an asynchronous function — it does not automatically turn the synchronous code inside it into asynchronous code.

For systems that heavily rely on external I/O — such as web services, knowledge bases, and agent platforms — establishing a unified async and concurrency architecture is an important foundation for improving system throughput.

📌 Technical Profile

Asynchronous Programming

When a program is waiting for external operations — such as databases, networks, or LLMs — to return, it does not hold onto the execution thread idly. Instead, it yields the execution opportunity to other tasks.

Closely related is:

Concurrency

Advancing multiple tasks within the same time period, rather than strictly completing one before starting the next.

Note the distinction:

• Concurrency — multiple tasks progress in an interleaved fashion;
• Parallelism — multiple tasks truly execute simultaneously.

Web backends typically focus on concurrency first.

💡 A Simple Analogy

Imagine the server as a restaurant waiter.

In synchronous mode, after the waiter sends Customer A’s order to the kitchen, they stand there waiting for the food to be ready:

A places order
  ↓
Waiting for kitchen...
  ↓
Food is ready
  ↓
Then serve B

In asynchronous mode:

A places order → Kitchen processes
                   ↓
Waiter → Serves B
       → Serves C
       → Serves D
                   ↓
A's food is ready → Come back to handle A

The number of waiters hasn’t increased — the only difference is:

They no longer just stand around while waiting.

This is the core idea behind asynchronous programming.

1. The Most Common Mistake AI Makes: Asynchronous on the Surface, Blocking in Reality

For example, AI easily generates:

async def handle_request():
    result = requests.get(url)
    return result.text

It appears to use:

async def

But requests.get() is still a synchronous blocking call.

A similar problem is:

time.sleep(5)

appearing inside an asynchronous function.

You should use async-compatible implementations instead, such as:

async with httpx.AsyncClient() as client:
    response = await client.get(url)

And:

await asyncio.sleep(5)

So to judge whether code is truly asynchronous, you can’t just look at whether there is an async def — you also need to check:

Whether any blocking operations have slipped into the entire call chain.

2. Architectural Principles

1. Async Must Run Through the Entire Call Chain

A truly reliable asynchronous architecture should look like:

Layers involving I/O should stay asynchronous as much as possible.

So-called I/O (Input/Output) operations mainly include:

• Database access;
• HTTP requests;
• LLM calls;
• Redis access;
• File read/write;
• Web Search;
• Vector database access.

The biggest cost of these operations is often not CPU computation, but waiting for external results.

Therefore, the principle is simple:

I/O-intensive operations should prefer asynchronous interfaces.

2. Tasks That Can Run Concurrently Should Not Wait Sequentially

Asynchronous does not automatically mean concurrent.

For example:

user = await get_user()
kb = await search_kb()
web = await search_web()

Although all three functions are asynchronous, they still execute sequentially.

If the three tasks are independent, you can do:

user, kb, web = await asyncio.gather(
    get_user(),
    search_kb(),
    search_web(),
)

Suppose the three tasks each take:

User info       0.4s
KB search       1.2s
Web Search      1.5s

Sequential execution takes approximately:

0.4 + 1.2 + 1.5 = 3.1s

Concurrent execution is closer to the slowest task:

≈ 1.5s

But concurrency should not be overused either.

For example:

Create order
 ↓
Deduct inventory
 ↓
Generate payment record

Each step depends on the previous one, so you can’t simply put them into asyncio.gather().

So remember this rule:

Tasks without dependencies are candidates for concurrency; tasks with dependencies should remain sequential.

3. CPU-Intensive Tasks Must Not Block the Event Loop

Asynchrony primarily solves the problem of I/O waiting.

For tasks such as:

• OCR;
• Image processing;
• Video transcoding;
• PDF re-computation;
• Large-scale data computation;
• Local model inference;

These typically fall under:

CPU-bound Task

They are not “waiting for someone else” — they genuinely and continuously occupy the CPU.

If you write:

async def endpoint():
    result = heavy_cpu_task()

Even though the function is async, it may still block:

Event Loop

Therefore, heavy computation should typically be moved to:

Thread Pool

Process Pool

Background Worker

In short:

I/O-intensive
   ↓
async / await

CPU-intensive
   ↓
Thread / Process / Worker

3. Async Resources Should Be Managed Centrally

Resources such as database Engines, SessionFactories, and HTTP Clients should not be repeatedly created inside every business function.

A more reasonable architecture is:

Application Startup
        ↓
ServiceContainer
        ↓
Async Engine / Client
        ↓
Repository
        ↓
Service

Centrally managing these resources reduces redundant connection creation and makes initialization, shutdown, and exception handling much clearer.

3. What Does an Async Architecture Actually Bring?

The most important thing is not making a single database query suddenly faster.

Suppose the database query itself takes 500ms:

Synchronous: 500ms
Asynchronous: likely still 500ms

The difference is whether the server can continue processing other requests while waiting those 500ms.

Synchronous:

Request A
████████████

Request B
            ████████████

Asynchronous:

Request A
████────████

Request B
  ███────████

Request C
    ███────████

Therefore, what an async architecture primarily improves is:

Throughput — the number of requests a system can handle per unit of time.

This is especially important for AI systems, because a single Agent request may involve:

Database
 ↓
LLM
 ↓
Knowledge Base
 ↓
Embedding
 ↓
Reranker
 ↓
Web Search
 ↓
Tool Calls
 ↓
LLM Called Again

A significant amount of time is spent waiting on networks and external services.

This is why AI applications are naturally well-suited for asynchronous architectures.

4. Putting Prompts into Practice: Directly Constraining AI

Rather than manually checking every time, it’s better to write the async rules into the Project Rules:

## Async and Concurrency Rules

This project adopts an async-first architecture for I/O-intensive operations.

1. FastAPI routes that perform I/O operations should use async def.
2. Database access must use SQLAlchemy AsyncSession.
3. Repository database methods must be asynchronous.
4. Never directly call blocking I/O inside async functions.

Avoid using:
- requests
- time.sleep
- synchronous database sessions

Recommended:
- httpx.AsyncClient
- asyncio.sleep
- AsyncSession

5. Independent I/O tasks may use asyncio.gather.
6. Do not parallelize operations that have data dependencies or transactional ordering requirements, and do not block on CPU-intensive operations.
7. CPU-intensive operations must not block the event loop. Use thread pools, process pools, or background workers where appropriate.
8. Maintain the async call chain: API → Service → Repository → Async Driver.
9. Before generating code, check whether all called libraries are synchronous and whether they may block the event loop.

This is far more explicit than simply telling the AI:

“Help me optimize performance.”

5. A Positive Example: The Async Architecture of miniagent

miniagent is an agent platform built on FastAPI, SQLAlchemy, knowledge base retrieval, and an Agent Runtime.

Its database access is not just a few async def at the API layer — it forms a complete asynchronous data access chain.

1. Unified Creation of the Async Engine

In miniagent’s ServiceContainer:

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)

database_url = f"sqlite+aiosqlite:///{db_path}"

self.engine = create_async_engine(
    database_url,
    echo=False,
    future=True,
)

self.session_factory = async_sessionmaker(
    bind=self.engine,
    class_=AsyncSession,
    autoflush=False,
    autocommit=False,
    expire_on_commit=False,
)

What this forms is:

FastAPI
   ↓
SQLAlchemy Async Engine
   ↓
AsyncSession
   ↓
aiosqlite
   ↓
SQLite

Rather than “FastAPI is async, but the underlying database is still accessed synchronously.”

2. Repository Unified Asynchronization

miniagent’s data access layer includes:

async_agent.py
async_chat.py
async_chunk.py
async_document.py
async_embedding.py
async_knowledge_base.py
async_llm.py
...

In other words, asynchrony is adopted as the unified design approach for the Repository (data access layer), rather than a localized optimization for a few interfaces.

3. Session Lifecycle Is Also Asynchronous

miniagent defines a unified AsyncBaseDatabase:

@asynccontextmanager
async def get_session(self):
    async with self.AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except SQLAlchemyError:
            await session.rollback()
            raise
        finally:
            await session.close()

The database Session’s:

Creation
Commit
Rollback
Close

are all uniformly managed within an asynchronous lifecycle.

4. Database Operations Genuinely Use await

For example, miniagent’s chat data access:

async def get_user_session(
    self,
    session_id: int,
    user_id: int,
):
    async with self.get_session() as session:
        stmt = (
            select(ChatSession)
            .where(
                ChatSession.id == session_id,
                ChatSession.user_id == user_id,
            )
        )

        return (
            await session.execute(stmt)
        ).scalar_one_or_none()

What truly matters is not the:

async def

in front of the function, but the database operation itself:

await session.execute(stmt)

This is what gives the event loop the opportunity to continue processing other tasks while waiting for the database.

5. The Complete Async Chain

The main asynchronous chain of miniagent is shown below:

When a request is waiting for the database, execution control can return to the Event Loop, and the server continues advancing other requests.

This is the core of how an async architecture improves system throughput.

The diagram below shows more details of the async chain:

6. A Simple Checklist for AI

After having AI complete backend code, you can ask it to self-check:

Async Architecture Checklist

□ Are I/O operations using async APIs?
□ Is the database using AsyncSession?
□ Are there synchronous blocking libraries called inside async functions?
□ Can independent I/O tasks run concurrently?
□ Are tasks with dependencies incorrectly parallelized?
□ Do CPU-heavy tasks block the Event Loop?
□ Is the following chain maintained:

API
 Service
 Repository
 Async Driver

Complete async chain?

Summary

AI easily produces code like:

Call A
 ↓
Wait
 ↓
Call B
 ↓
Wait
 ↓
Call C

This kind of code is fine for a demo, but once it enters a real concurrent environment, it easily becomes a system bottleneck.

Therefore, a few clear rules should be established upfront:

I/O defaults to async
      ↓
Async runs through the call chain
      ↓
Independent tasks consider concurrency
      ↓
CPU-heavy tasks moved off the Event Loop
      ↓
Async resources managed centrally

Projects like miniagent bake async capability into the infrastructure through AsyncSession, async Repositories, unified Session management, and the FastAPI async call chain — rather than scattering it across a few interfaces.

For AI programming, what we ultimately need to constrain is not just:

“Can this code run?”

We also need to ask:

“Will it block the entire server while it waits?”

This is precisely the value of an async and concurrency architecture.

Open Source Code

• github
• gitee

🪐 Wishing you good luck 🪐