Forem: Jun Bae

Coroutine series 3) Coroutines for LLM inference

Jun Bae — Wed, 22 Apr 2026 14:09:43 +0000

Introduction

In this post, I will briefly introduce how to utilize coroutines for LLMs. Using asyncio for LLM inference is straightforward because most AI frameworks support asyncio natively nowadays.

Coroutines for LLM

LLM API SDK

API SDKs—Google GenAI, OpenAI, Claude, and others— provide coroutine functions. You can easily leverage coroutines for AI like this:

# Synchronous inference function of OpenAI SDK
client = OpenAI()
response = client.chat.completions.create(
    model="gpt-5-mini",
    messages=[{"role": "user", "content": prompt}],
)

# Coroutine function of OpenAI SDK
client = AsyncOpenAI()
response = await client.chat.completions.create(
    model="gpt-5-mini",
    messages=[{"role": "user", "content": prompt}],
)

# Synchronous inference function of Google GenAI SDK
client = genai.Client()
response = client.models.generate_content(
    model="gemini-2.5-flash", contents=prompt
)

# Coroutine function of Google GenAI SDK
client = genai.Client()
response = await client.aio.models.generate_content(
    model="gemini-2.5-flash", contents=prompt
)

For the OpenAI SDK, there’s a specific client for async. For Google GenAI, you should add aio after client. Other SDKs, such as Claude and Ollama, also support coroutine functions.

Example comparing async and sync functions

Input

import asyncio
import time
import os
import logging
from dotenv import load_dotenv
from google import genai
from openai import OpenAI, AsyncOpenAI

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%H:%M:%S')
logger = logging.getLogger(__name__)

load_dotenv()

# --- Sync Functions ---

def sync_chat_google(prompt: str):
    logger.info("Starting Sync Google GenAI request...")
    try:
        client = genai.Client()
        response = client.models.generate_content(
            model="gemini-2.5-flash", contents=prompt
        )
        logger.info("Finished Sync Google GenAI request.")
        return f"Google (Sync): {response.text[:50]}..."
    except Exception as e:
        logger.error(f"Sync Google Error: {e}")
        return f"Google (Sync) Error: {e}"

def sync_chat_openai(prompt: str):
    logger.info("Starting Sync OpenAI request...")
    try:
        client = OpenAI()
        response = client.chat.completions.create(
            model="gpt-5-mini",
            messages=[{"role": "user", "content": prompt}],
            reasoning_effort="low"
        )
        logger.info("Finished Sync OpenAI request.")
        return f"OpenAI (Sync): {response.choices[0].message.content[:50]}..."
    except Exception as e:
        logger.error(f"Sync OpenAI Error: {e}")
        return f"OpenAI (Sync) Error: {e}"

def run_sync():
    logger.info("--- Starting Sync Execution ---")
    start_time = time.time()

    prompt = "Explain asyncio in one sentence."
    res_google = sync_chat_google(prompt)
    res_openai = sync_chat_openai(prompt)

    end_time = time.time()
    total_time = end_time - start_time

    print(f"\n[Sync Results]")
    print(res_google)
    print(res_openai)
    print(f"Total Sync Time: {total_time:.2f} seconds\n")
    return total_time

# --- Async Functions ---

async def async_chat_google(prompt: str):
    logger.info("Starting Async Google GenAI request...")
    try:
        client = genai.Client()
        response = await client.aio.models.generate_content(
            model="gemini-2.5-flash", contents=prompt
        )
        logger.info("Finished Async Google GenAI request.")
        return f"Google (Async): {response.text[:50]}..."
    except Exception as e:
        logger.error(f"Async Google Error: {e}")
        return f"Google (Async) Error: {e}"

async def async_chat_openai(prompt: str):
    logger.info("Starting Async OpenAI request...")
    try:
        client = AsyncOpenAI()
        response = await client.chat.completions.create(
            model="gpt-5-mini",
            messages=[{"role": "user", "content": prompt}],
            reasoning_effort="low"
        )
        logger.info("Finished Async OpenAI request.")
        return f"OpenAI (Async): {response.choices[0].message.content[:50]}..."
    except Exception as e:
        logger.error(f"Async OpenAI Error: {e}")
        return f"OpenAI (Async) Error: {e}"

async def run_async():
    logger.info("--- Starting Async Execution ---")
    start_time = time.time()

    prompt = "Explain asyncio in one sentence."
    # Schedule both calls concurrently
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(async_chat_google(prompt))
        task2 = tg.create_task(async_chat_openai(prompt))

    results = [task1.result(), task2.result()]

    end_time = time.time()
    total_time = end_time - start_time

    print(f"\n[Async Results]")
    for res in results:
        print(res)
    print(f"Total Async Time: {total_time:.2f} seconds\n")
    return total_time

async def main():
    logger.info("Starting Comparison...")

    sync_time = run_sync()
    async_time = await run_async()

    print("-" * 30)
    print(f"Sync Time:  {sync_time:.2f}s")
    print(f"Async Time: {async_time:.2f}s")
    if async_time < sync_time:
        print(f"Conclusion: Async was {sync_time / async_time:.2f}x faster!")
    else:
        print("Conclusion: Async was not faster (overhead or network variance).")

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

Output

00:16:17 - INFO - Starting Comparison...
00:16:17 - INFO - --- Starting Sync Execution ---
00:16:17 - INFO - Starting Sync Google GenAI request...
00:16:18 - INFO - AFC is enabled with max remote calls: 10.
00:16:24 - INFO - HTTP Request: POST https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent "HTTP/1.1 200 OK"
00:16:24 - INFO - Finished Sync Google GenAI request.
00:16:24 - INFO - Starting Sync OpenAI request...
00:16:28 - INFO - HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
00:16:28 - INFO - Finished Sync OpenAI request.

[Sync Results]
Google (Sync): Asyncio is Python's library for writing concurrent...
OpenAI (Sync): asyncio is Python's library for writing concurrent...
Total Sync Time: 10.36 seconds

00:16:28 - INFO - --- Starting Async Execution ---
00:16:28 - INFO - Starting Async Google GenAI request...
00:16:28 - INFO - AFC is enabled with max remote calls: 10.
00:16:28 - INFO - Starting Async OpenAI request...
00:16:31 - INFO - HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
00:16:31 - INFO - Finished Async OpenAI request.
00:16:35 - INFO - HTTP Request: POST https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent "HTTP/1.1 200 OK"
00:16:35 - INFO - Finished Async Google GenAI request.

[Async Results]
Google (Async): asyncio is Python's library for writing concurrent...
OpenAI (Async): asyncio is Python's standard-library framework for...
Total Async Time: 6.91 seconds

------------------------------
Sync Time:  10.36s
Async Time: 6.91s
Conclusion: Async was 1.50x faster!

As you can see, the async coroutine functions are 1.5x faster because they call the OpenAI API and the Google API concurrently. On the other hand, the synchronous functions call the Google Genai API first, wait for the response, and then subsequently call the OpenAI API.

Sync vs Async

When you build complex AI inference architectures, such as Agentic AI, you should think carefully about whether to utilize async. It depends on the specific requirements of your architecture.

Sequential Dependencies: When some inferences depend on the previous ones—for example, if a previous step retrieves necessary context (like in RAG), or the result of one inference must be included in the prompt of the subsequent inference—you generally execute them sequentially.
Independent Tasks: When you run independent inferences that do not rely on each other, you can leverage coroutines to run them concurrently.

Example

import asyncio
import time
import os
import logging
from dotenv import load_dotenv
from typing import List

from langchain_google_genai import ChatGoogleGenerativeAI
import ollama

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%H:%M:%S')
logger = logging.getLogger(__name__)

load_dotenv()

# --- 1. Async Retrieval Simulation ---

async def retrieve_docs_db(query: str) -> str:
    """Simulates retrieving documents from a database (IO bound)."""
    logger.info(f"[DB] Searching for: '{query}'...")
    await asyncio.sleep(2)  # Simulate network/DB latency
    result = "Context from DB: Asyncio is single-threaded but concurrent."
    logger.info(f"[DB] Found: {result}")
    return result

async def retrieve_docs_web(query: str) -> str:
    """Simulates retrieving documents from the web (IO bound)."""
    logger.info(f"[Web] Searching for: '{query}'...")
    await asyncio.sleep(2)  # Simulate network latency
    result = "Context from Web: Asyncio uses an event loop to manage tasks."
    logger.info(f"[Web] Found: {result}")
    return result

# --- 2. Sync Ollama Inference ---

def query_ollama(context: List[str], question: str) -> str:

    combined_context = "\n".join(context)
    prompt = f"Context:\n{combined_context}\n\nQuestion: {question}\n\nAnswer:"

    logger.info("Starting Ollama generation...")
    try:
        client = ollama.Client() 
        model = "qwen3:8b" 

        response = client.chat(model=model, messages=[
            {'role': 'user', 'content': prompt},
        ])

        answer = response['message']['content']
        logger.info("Finished Ollama generation.")
        return answer
    except Exception as e:
        logger.error(f"Ollama Error: {e}")
        return f"Ollama Error: {e}"

# --- 3. Sync Google Refinement ---

def refine_with_google(draft_answer: str) -> str:
    logger.info("Starting Google Refinement...")
    try:
        prompt = f"Please refine and polish the following text to be more professional:\n\n{draft_answer}"

        genai_llm = ChatGoogleGenerativeAI(
            model="gemini-3-flash-preview",
            thinking_level="low"
        )
        response = genai_llm.invoke(prompt)
        logger.info("Finished Google Refinement.")
        return response.text
    except Exception as e:
        logger.error(f"Google Refinement Error: {e}")
        return f"Google Error: {e}"

# --- Main Flow ---

async def main():
    start_total = time.time()
    query = "How does asyncio work?"

    print(f"\n=== Starting Concurrent RAG Demo ===")
    print(f"Query: {query}\n")

    # Step 1: Concurrent Retrieval
    logger.info("--- Step 1: Concurrent Retrieval ---")
    start_retrieval = time.time()

    # Launch both retrievals at the same time
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(retrieve_docs_db(query))
        task2 = tg.create_task(retrieve_docs_web(query))
    results = [task1.result(), task2.result()]

    retrieval_time = time.time() - start_retrieval
    logger.info(f"Retrieval complete in {retrieval_time:.2f}s")

    # Step 2: Sync Ollama Inference (Dependent on Step 1)
    logger.info("--- Step 2: Sync Ollama Inference ---")
    start_ollama = time.time()

    draft_answer = query_ollama(results, query)

    ollama_time = time.time() - start_ollama
    print(f"\n[Ollama Draft]:\n{draft_answer}\n")

    # Step 3: Sync Google Refinement (Dependent on Step 2)
    logger.info("--- Step 3: Sync Google Refinement ---")
    start_google = time.time()

    refined_answer = refine_with_google(draft_answer)

    google_time = time.time() - start_google
    print(f"\n[Google Refined]:\n{refined_answer}\n")

    total_time = time.time() - start_total

    print("=" * 40)
    print(f"Retrieval Time: {retrieval_time:.2f}s")
    print(f"Ollama Time:    {ollama_time:.2f}s")
    print(f"Google Time:    {google_time:.2f}s")
    print(f"Total Time:     {total_time:.2f}s")
    print("=" * 40)

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

Output

=== Starting Concurrent RAG Demo ===
Query: How does asyncio work?

20:05:38 - INFO - --- Step 1: Concurrent Retrieval ---
20:05:38 - INFO - [DB] Searching for: 'How does asyncio work?'...
20:05:38 - INFO - [Web] Searching for: 'How does asyncio work?'...
20:05:40 - INFO - [DB] Found: Context from DB: Asyncio is single-threaded but concurrent.
20:05:40 - INFO - [Web] Found: Context from Web: Asyncio uses an event loop to manage tasks.
20:05:40 - INFO - Retrieval complete in 2.01s
20:05:40 - INFO - --- Step 2: Sync Ollama Inference ---
20:05:40 - INFO - Starting Ollama generation...
20:05:57 - INFO - HTTP Request: POST http://127.0.0.1:11434/api/chat "HTTP/1.1 200 OK"
20:05:57 - INFO - Finished Ollama generation.

[Ollama Draft]:
Asyncio is a Python library that enables **single-threaded concurrency** through **asynchronous I/O** and an **event loop**. Here's how it works:

### 1. **Single-Threaded, Concurrent Execution**
   - Asyncio operates in a **single thread**, avoiding the overhead of multi-threading. Instead of using multiple threads, it leverages **non-blocking I/O** to handle multiple tasks concurrently.
   - Concurrency here means tasks can **overlap** in execution, even though they run in the same thread. This is ideal for **I/O-bound tasks** (e.g., network requests, file reads) where waiting for I/O is common.

### 2. **Event Loop as the Core Mechanism**
   - The **event loop** is the heart of asyncio. It manages and schedules tasks, handling I/O operations and switching between coroutines when needed.
   - When a coroutine (an `async def` function) is started, the event loop schedules it. If the coroutine encounters an I/O operation (e.g., a network call), it **yields control** back to the event loop, allowing 
other tasks to run in the meantime.                                                                                                                                                                                  
### 3. **Coroutines and `await`**
   - Coroutines are defined using `async def`. They can be paused and resumed, enabling **cooperative multitasking**.
   - The `await` keyword is used to **delegate control** to another coroutine or I/O operation. While waiting, the event loop can process other tasks, ensuring efficient resource utilization.

### 4. **Non-Blocking I/O**
   - Asyncio avoids blocking the thread by using **asynchronous I/O**. For example, when a coroutine makes a network request, it doesn’t wait for the response to complete. Instead, it **registers a callback** with
 the event loop and continues executing other tasks. Once the I/O operation completes, the event loop resumes the coroutine.                                                                                         
### 5. **Task Scheduling and Collaboration**
   - The event loop manages **tasks** (coroutines) and ensures they run in a coordinated manner. It handles:
     - Scheduling coroutines to run.
     - Handling callbacks for completed I/O operations.
     - Managing exceptions and ensuring orderly execution.

### 6. **Use Cases**
   - Asyncio is best suited for **I/O-bound applications** (e.g., web servers, APIs, data scraping) where waiting for I/O is a bottleneck.
   - It is **not ideal** for CPU-bound tasks, which would benefit more from multiprocessing or threading.

### Summary
Asyncio achieves concurrency by using a **single thread** with an **event loop** to manage **non-blocking I/O** and **coroutines**. This allows multiple tasks to run **simultaneously** without blocking the main th
read, making it efficient for handling many I/O operations in applications like web servers or network clients.                                                                                                      
20:05:57 - INFO - --- Step 3: Sync Google Refinement ---
20:05:57 - INFO - Starting Google Refinement...
20:05:57 - INFO - AFC is enabled with max remote calls: 10.
20:06:06 - INFO - HTTP Request: POST https://generativelanguage.googleapis.com/v1beta/models/gemini-3-flash-preview:generateContent "HTTP/1.1 200 OK"
20:06:06 - INFO - Finished Google Refinement.

[Google Refined]:
Here are three ways to refine the text, depending on the desired tone and context.

### Option 1: Formal & Technical (Best for Documentation or Whitepapers)
This version uses precise terminology and focuses on the architectural benefits of the library.

**Overview of Asyncio in Python**
Asyncio is a specialized Python library designed to facilitate **single-threaded concurrency** through **asynchronous I/O** and an **event loop** architecture. It is structured around the following core principles
:                                                                                                                                                                                                                    
*   **Single-Threaded Concurrency:** Unlike traditional multi-threading, asyncio operates within a single thread. It leverages **non-blocking I/O** to manage multiple execution streams concurrently. This approach 
eliminates the overhead associated with context switching and thread management, making it highly efficient for **I/O-bound tasks**.                                                                                 *   **The Event Loop:** Serving as the central orchestrator, the event loop manages task scheduling and I/O operations. When a coroutine encounters an I/O bottleneck, it **yields control** back to the loop, which 
then executes other pending tasks, ensuring optimal CPU utilization.                                                                                                                                                 *   **Coroutines and Cooperative Multitasking:** Defined via the `async def` syntax, coroutines are the fundamental units of execution. Through the `await` keyword, these functions practice **cooperative multitask
ing**, pausing their execution to allow the event loop to process other operations until the awaited result is available.                                                                                            *   **Non-Blocking Operations:** Asyncio prevents thread blocking by registering callbacks for I/O events. This allows the system to initiate a request—such as a network call—and immediately move to the next task,
 resuming the original coroutine only once the data is ready.                                                                                                                                                        *   **Optimized Use Cases:** Asyncio is the industry standard for high-performance **I/O-bound applications**, including web servers, distributed systems, and real-time data streaming. Conversely, for CPU-intensiv
e computations, multiprocessing remains the preferred parallelization strategy.                                                                                                                                      
---

### Option 2: Professional & Concise (Best for a Presentation or Summary)
This version is streamlined for readability while maintaining a professional tone.

**Understanding Asyncio**
Asyncio enables Python developers to handle high-concurrency workloads within a single thread. By utilizing an event loop and non-blocking I/O, it provides a scalable alternative to traditional threading.

1.  **Concurrency without Threads:** By overlapping task execution rather than running them in parallel, asyncio avoids the memory overhead of multiple threads while efficiently handling thousands of simultaneous 
connections.                                                                                                                                                                                                         2.  **The Event Loop & Coroutines:** The event loop acts as a scheduler. Coroutines (`async def`) cooperatively yield control using `await`, allowing the loop to switch between tasks seamlessly whenever a program 
is waiting for external data.                                                                                                                                                                                        3.  **Efficiency through Non-Blocking I/O:** Instead of halting execution during a network or file operation, asyncio registers the operation and continues with other work. The event loop resumes the paused task o
nly after the I/O operation signals completion.                                                                                                                                                                      4.  **Strategic Application:** Asyncio is ideal for network-heavy applications like APIs and web scrapers. However, for CPU-bound tasks, developers should utilize multiprocessing to bypass the Global Interpreter L
ock (GIL).                                                                                                                                                                                                           
---

### Option 3: Modern & Direct (Best for a Technical Blog or Internal Memo)
This version is punchy and uses active language to explain the concepts.

**Asyncio: Scaling Python through Asynchronous I/O**
Asyncio is Python's built-in solution for **single-threaded concurrency**. It is designed to maximize resource efficiency by ensuring the CPU never sits idle while waiting for network or disk responses.

*   **How it Works:** At the core is the **Event Loop**, which manages **Coroutines** (asynchronous functions).
*   **Cooperative Scheduling:** Using the `await` keyword, a coroutine voluntarily pauses itself during I/O operations. This "cooperation" allows the event loop to rotate through other tasks, creating a highly res
ponsive system.                                                                                                                                                                                                      *   **Non-Blocking Workflow:** By utilizing asynchronous I/O, the application can trigger multiple network requests simultaneously without blocking the main execution thread.
*   **When to use it:** Use Asyncio for I/O-heavy workloads (Web Servers, APIs, Database-heavy apps). Avoid it for heavy mathematical computations, where Multiprocessing is better suited.

**Summary:** Asyncio delivers high-performance concurrency by combining a single-threaded event loop with cooperative multitasking, making it an essential tool for modern, scalable Python development.

---

### Key Improvements Made:
*   **Vocabulary:** Changed "heart of" to "central orchestrator," and "ideal" to "industry standard" or "preferred strategy."
*   **Clarity:** Clarified the distinction between concurrency (overlapping tasks) and parallelism (simultaneous tasks).
*   **Precision:** Replaced general terms with more technical equivalents like "context switching," "resource utilization," and "cooperative multitasking."

========================================
Retrieval Time: 2.01s
Ollama Time:    17.12s
Google Time:    9.58s
Total Time:     28.71s
========================================

This is just a simple LLM example using LangChain (Google GenAI) and the Ollama (Qwen3:8b) SDK. As you can see, the RAG component is just a mock simulation for convenience.

This process can retrieve context from two sources and these tasks run concurrently. However, the subsequent LLM inference requires that context data, so it must run sequentially after the retrievals. Similarly, the refinement step must also be sequential, as it relies on refining the previous result.

Note: I used synchronous LLM clients for Langchain and Ollama since this is just a demonstration. However, in production-level development, you should implement these parts as coroutines as well. In a production environment, the server may receive high traffic; using async allows the system to handle multiple incoming requests efficiently while waiting for the LLM to respond.

AI Framework

Most AI frameworks—such as LangGraph, the OpenAI Agents SDK, Google AI SDKs, AutoGen, and CrewAI— build agents or workflows to be asynchronous by default.

But you must ensure that you use coroutine functions within your nodes or agents. Even though the frameworks themselves are asynchronous, performance can crawl if you inadvertently include blocking I/O functions in certain nodes.

Conclusion

Implementing coroutines for LLMs is not that difficult. In fact, it is quite straightforward because most SDKs and frameworks now provide native support for coroutines. The most important aspect, however, is utilizing asyncio patterns correctly. Production-level projects can be complex, making it easy to misuse coroutines or introduce bottlenecks. Therefore, when building AI inference projects, you must carefully consider several factors: which parts should be asynchronous versus synchronous, when to offload tasks using to_thread, what the optimal concurrency limit shoud be.

Coroutine series 2) Useful Asyncio Functions

Jun Bae — Wed, 22 Apr 2026 14:05:19 +0000

This is the second post in the series Coroutine, IO bound and Asyncio for AI.

Introduction

I explained coroutines and asyncio in the previous post: https://dev.to/jun07/coroutine-series-1-what-are-coroutine-asyncio-io-bound-2gh5

In this post, I’m going to show some functions of the asyncio library that are very useful for production-level development.

Coroutine functions

I briefly introduced simple coroutine functions such as async, await, gather, create_task. There are more functions in the asyncio library and you can do so many things with these powerful tools.

`gather` and `TaskGroup`

I demonstrated how to use gather but I didn’t explain much about it. After you make some coroutine functions, you have to make them coordinate well. Often, several coroutine functions should be executed asynchronously. In that case, you can utilize the functions called gather or TaskGroup in asyncio. But are they exactly the same?

Actually, they have a very crucial difference. When one of the tasks in gather fails, the other tasks keep running. On the other hand, TaskGroup cancels the remaining tasks immediately. Look at this example.

import asyncio
import time

async def example_task(name, delay, should_fail=False):
    print(f"Task {name} starting (delay={delay}s)")
    await asyncio.sleep(delay)
    if should_fail:
        print(f"Task {name} failing")
        raise ValueError(f"Task {name} failed")
    print(f"Task {name} finished")
    return f"Result {name}"

async def demonstrate_gather():
    print("\n--- Demonstrating asyncio.gather ---")
    start_time = time.time()

    # Create tasks: B will fail
    tasks = [
        example_task("A", 1),
        example_task("B", 2, should_fail=True),
        example_task("C", 3)
    ]

    # asyncio.gather without return_exceptions=True raises the first exception immediately
    # BUT it does NOT cancel the other tasks. They keep running in the background.
    print("Starting gather with a failing task...")
    try:
        results = await asyncio.gather(*tasks)
    except Exception as e:
        print(f"Gather caught exception: {e}")

    await asyncio.sleep(2)
    print(f"Gather finished (after waiting). Note if other tasks continued printing above.")
    print(f"Gather took: {time.time() - start_time:.2f}s")

async def demonstrate_task_group():
    print("\n--- Demonstrating asyncio.TaskGroup ---")
    start_time = time.time()

    # TaskGroup cancels all other tasks immediately when one fails
    print("Starting TaskGroup with a failing task...")
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(example_task("X", 1))
            tg.create_task(example_task("Y", 2, should_fail=True))
            tg.create_task(example_task("Z", 3))

    except Exception as e:
        print(f"TaskGroup caught exception: {type(e).__name__}: {e}")

    print(f"TaskGroup finished. Other tasks should have been cancelled (no 'finished' print for Z).")
    print(f"TaskGroup took: {time.time() - start_time:.2f}s")

async def main():
    await demonstrate_gather()
    await demonstrate_task_group()

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

Output

--- Demonstrating asyncio.gather ---
Starting gather with a failing task...
Task A starting (delay=1s)
Task B starting (delay=2s)
Task C starting (delay=3s)
Task A finished
Task B failing
Gather caught exception: Task B failed
Task C finished
Gather finished (after waiting). Note if other tasks continued printing above.
Gather took: 4.00s

--- Demonstrating asyncio.TaskGroup ---
Starting TaskGroup with a failing task...
Task X starting (delay=1s)
Task Y starting (delay=2s)
Task Z starting (delay=3s)
Task X finished
Task Y failing
TaskGroup caught exception: ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
TaskGroup finished. Other tasks should have been cancelled (no 'finished' print for Z).
TaskGroup took: 1.99s

As you can see, gather keeps running even if one of the tasks fails while TaskGroup raises an exception and cancels the remaining tasks immediately.

Differences: `TaskGroup` vs. `gather`

Cancellation: In asyncio.gather(), if one task fails, the others continue running (unless explicitly cancelled). TaskGroup cancels siblings immediately upon the first failure, conserving resources.
Error Reporting: gather() returns a list of results or raises the first exception. TaskGroup raises an ExceptionGroup containing all exceptions that occurred, providing a complete picture of the failure state.

Why is `TaskGroup` favored over `gather`?

TaskGroup enforces a "parent-child" relationship where the parent cannot exit until all children are finished. This eliminates "orphaned tasks"—background processes that silently keep running (and consuming resources) after your main function has finished or crashed. That is why modern high-reliability Python architecture favors Structured Concurrency(asyncio.TaskGroup) over unstructured create_task + gather.

By design, TaskGroup operates on a "fail-fast" principle. If one task fails, the group immediately sends a cancellation signal to all other running tasks in that group.

But what if I don’t want to cancel the whole task group even if one fails? Should I just use just gather? In that case, you can still utilize TaskGroup by handling exceptions with try and except blocks inside the individual tasks.

Note: asyncio.TaskGroup is available in Python 3.11+. For older versions, you may need third-party libraries or stick to gather.

Synchronization Primitives: `Queue` and `Event`

What if I want to run some tasks synchronously in certain parts? To do this, asyncio provides synchronization primitives.

asyncio.Queue (The Conveyor Belt): Used for Data Transfer. It allows you to pass data safely between tasks. It handles "backpressure" (slowing down the producer if the consumer is full).
asyncio.Event (The Traffic Light): Used for Signaling. It holds a simple boolean state (True/False). Tasks "wait" for the light to turn green before proceeding.

Asyncio.Queue

As you know, a Queue is a First-In-First-Out(FIFO) data structure. But what is asyncio.Queue then? It is designed to decouple tasks that produce work from tasks that perform work.

What is Queue used for?

Load Balancing: If the producer is faster than the consumer, the queue buffers the excess.
Flow Control: If you set a maxsize, the producer will pause (await put()) until the consumer clears space. This prevents memory explosions.

Before getting into the example code, here are some key methods:

Key Methods

await q.put(item): Add an item. Blocks if the queue is full.
await q.get(): Remove and return an item. Blocks if the queue is empty.
q.task_done(): Tells the queue "I finished processing the item I just got."
await q.join(): Blocks until all items in the queue have been processed (marked via task_done).

Example

Input

import asyncio
import logging
import random
import time

logging.basicConfig(
    level=logging.INFO,
    format='%(relativeCreated)d ms - [%(levelname)s] - %(message)s',
)


async def producer(name: str, queue: asyncio.Queue, items_to_produce: int):
    """Generates work items and puts them into the queue."""
    for i in range(items_to_produce):
        item = f"{name}-item-{i}"

        # Simulate varying production time
        await asyncio.sleep(random.uniform(0.2, 0.8))

        await queue.put(item)
        logging.info(f"[{name}] Added: {item} (Q size: {queue.qsize()})")

async def consumer(name: str, queue: asyncio.Queue):
    """Processes items from the queue continuously."""
    while True:
        # Waits here if queue is empty
        item = await queue.get()

        # Simulate processing work
        logging.info(f"[{name}] Processing {item}...")
        await asyncio.sleep(random.uniform(1.0, 2.0))

        # Signal that this specific item is fully processed
        queue.task_done()
        logging.info(f"[{name}] Finished {item}")

async def main():
    # maxsize=3 to easily demonstrate backpressure (producers waiting for consumers)
    q = asyncio.Queue(maxsize=5)

    logging.info("--- Starting Pipeline ---")

    # 1. Start Consumers as daemon tasks
    # They run forever until cancelled.
    consumers = asyncio.create_task(consumer(f"Consumer-0", q))

    # 2. Run Producers to completion
    # We use TaskGroup to wait for all producers to finish.
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer("Producer-A", q, 5))
        tg.create_task(producer("Producer-B", q, 5))
        # Total 10 items will be produced

    logging.info("--- All producers finished. Waiting for processing... ---")

    # 3. Wait for the queue to be fully processed
    # execution blocks here until q.task_done() is called for every item
    await q.join()

    logging.info("--- All work completed. helper tasks cancelled ---")

    # 4. Cancel the consumer tasks since they are permanent loops
    consumers.cancel()

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

Output

48 ms - [INFO] - --- Starting Pipeline ---
608 ms - [INFO] - [Producer-A] Added: Producer-A-item-0 (Q size: 1)
609 ms - [INFO] - [Consumer-0] Processing Producer-A-item-0...
788 ms - [INFO] - [Producer-B] Added: Producer-B-item-0 (Q size: 1)
1079 ms - [INFO] - [Producer-A] Added: Producer-A-item-1 (Q size: 2)
1325 ms - [INFO] - [Producer-A] Added: Producer-A-item-2 (Q size: 3)
1325 ms - [INFO] - [Producer-B] Added: Producer-B-item-1 (Q size: 4)
1587 ms - [INFO] - [Producer-B] Added: Producer-B-item-2 (Q size: 5)
1850 ms - [INFO] - [Consumer-0] Finished Producer-A-item-0
1851 ms - [INFO] - [Consumer-0] Processing Producer-B-item-0...
1851 ms - [INFO] - [Producer-A] Added: Producer-A-item-3 (Q size: 5)
3511 ms - [INFO] - [Consumer-0] Finished Producer-B-item-0
3511 ms - [INFO] - [Consumer-0] Processing Producer-A-item-1...
3512 ms - [INFO] - [Producer-B] Added: Producer-B-item-3 (Q size: 5)
4822 ms - [INFO] - [Consumer-0] Finished Producer-A-item-1
4822 ms - [INFO] - [Consumer-0] Processing Producer-A-item-2...
4822 ms - [INFO] - [Producer-A] Added: Producer-A-item-4 (Q size: 5)
6038 ms - [INFO] - [Consumer-0] Finished Producer-A-item-2
6038 ms - [INFO] - [Consumer-0] Processing Producer-B-item-1...
6038 ms - [INFO] - [Producer-B] Added: Producer-B-item-4 (Q size: 5)
6038 ms - [INFO] - --- All producers finished. Waiting for processing... ---
7538 ms - [INFO] - [Consumer-0] Finished Producer-B-item-1
7538 ms - [INFO] - [Consumer-0] Processing Producer-B-item-2...
9468 ms - [INFO] - [Consumer-0] Finished Producer-B-item-2
9469 ms - [INFO] - [Consumer-0] Processing Producer-A-item-3...
10698 ms - [INFO] - [Consumer-0] Finished Producer-A-item-3
10699 ms - [INFO] - [Consumer-0] Processing Producer-B-item-3...
11836 ms - [INFO] - [Consumer-0] Finished Producer-B-item-3
11836 ms - [INFO] - [Consumer-0] Processing Producer-A-item-4...
12919 ms - [INFO] - [Consumer-0] Finished Producer-A-item-4
12919 ms - [INFO] - [Consumer-0] Processing Producer-B-item-4...
13927 ms - [INFO] - [Consumer-0] Finished Producer-B-item-4
13927 ms - [INFO] - --- All work completed. helper tasks cancelled ---

In the example above, the producers generate items and put them into the queue, and the consumer processes them. Given that the maximum queue size is 5, some blocking occurs during the process. Overflowing requests wait for the queue to free up space. Thus, you can utilize this pattern as a Message Queue.

Asyncio.Event

An Event is much simpler. It manages an internal flag that can be set to True or False. It is a broadcast mechanism: one event can wake up many waiting tasks. That’s why I called it a ‘Traffic light’.

What is Event used for?

Startup Sequences: "Don't start accepting HTTP requests until the Database connection is ready."
Graceful Shutdown: "Tell all background workers to stop loop processing."

And here’re some key methods:

Key Methods

await event.wait(): Blocks the task until the flag becomes True. If it's already True, it proceeds immediately.
event.set(): Sets the flag to True. All tasks waiting on wait() are immediately woken up.
event.clear(): Resets the flag to False. Subsequent wait() calls will block again.

Example

Input

import asyncio
import random
import logging
import time

logging.basicConfig(
    level=logging.INFO,
    format='%(relativeCreated)d ms - [%(levelname)s] - %(message)s',
)

async def database_connection_manager(event: asyncio.Event):
    """
    Simulates a database connection manager that handles connection states.
    It manages the event flag to signal when the DB is ready or down.
    """
    logging.info("Manager: Database connection initializing...")
    await asyncio.sleep(2)  # Simulate startup time

    # 1. Event Set: Signal that the resource is ready
    logging.info("Manager: Database CONNECTED. Setting event (Green Light).")
    event.set()

    # Let workers process for a while
    await asyncio.sleep(3)

    # 2. Event Clear: Signal that the resource is unavailable (e.g., maintenance or crash)
    logging.info("Manager: Connection LOST! Clearing event (Red Light). blocking workers...")
    event.clear()

    # Simulate downtime loop
    await asyncio.sleep(3)

    logging.info("Manager: Reconnecting...")
    await asyncio.sleep(2)

    # 3. Event Set again: Recovery
    logging.info("Manager: Database RECONNECTED. Setting event again.")
    event.set()

async def query_worker(worker_id: int, event: asyncio.Event):
    """
    Simulates a worker that needs the database to be ready to process queries.
    """
    logging.info(f"Worker {worker_id}: Ready to start processing.")

    for i in range(5):
        # 4. Event Wait: Block until the event is set
        # If event.is_set() is True, it returns immediately.
        # If False, it waits until some other coroutine calls event.set().
        logging.info(f"Worker {worker_id}: Waiting for DB connection to process request {i+1}...")

        await event.wait()

        # If we reach here, the event is set!
        logging.info(f"Worker {worker_id}: Processing request {i+1} (DB is Online)")

        # Simulate processing time
        process_time = random.uniform(0.5, 1.5)
        await asyncio.sleep(process_time)

async def main():
    # Create the Event object
    # Internal flag is initially False
    db_ready_event = asyncio.Event()

    async with asyncio.TaskGroup() as tg:
        # Create tasks
        tg.create_task(database_connection_manager(db_ready_event))

        workers = [
            tg.create_task(query_worker(i, db_ready_event))
            for i in range(1, 4)
        ]

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

Output

48 ms - [INFO] - Manager: Database connection initializing...
49 ms - [INFO] - Worker 1: Ready to start processing.
49 ms - [INFO] - Worker 1: Waiting for DB connection to process request 1...
49 ms - [INFO] - Worker 2: Ready to start processing.
49 ms - [INFO] - Worker 2: Waiting for DB connection to process request 1...
49 ms - [INFO] - Worker 3: Ready to start processing.
49 ms - [INFO] - Worker 3: Waiting for DB connection to process request 1...
2057 ms - [INFO] - Manager: Database CONNECTED. Setting event (Green Light).
2057 ms - [INFO] - Worker 1: Processing request 1 (DB is Online)
2057 ms - [INFO] - Worker 2: Processing request 1 (DB is Online)
2058 ms - [INFO] - Worker 3: Processing request 1 (DB is Online)
2702 ms - [INFO] - Worker 1: Waiting for DB connection to process request 2...
2702 ms - [INFO] - Worker 1: Processing request 2 (DB is Online)
2959 ms - [INFO] - Worker 3: Waiting for DB connection to process request 2...
2959 ms - [INFO] - Worker 3: Processing request 2 (DB is Online)
3501 ms - [INFO] - Worker 2: Waiting for DB connection to process request 2...
3501 ms - [INFO] - Worker 2: Processing request 2 (DB is Online)
3945 ms - [INFO] - Worker 1: Waiting for DB connection to process request 3...
3945 ms - [INFO] - Worker 1: Processing request 3 (DB is Online)
4315 ms - [INFO] - Worker 3: Waiting for DB connection to process request 3...
4315 ms - [INFO] - Worker 3: Processing request 3 (DB is Online)
4632 ms - [INFO] - Worker 1: Waiting for DB connection to process request 4...
4632 ms - [INFO] - Worker 1: Processing request 4 (DB is Online)
4785 ms - [INFO] - Worker 2: Waiting for DB connection to process request 3...
4785 ms - [INFO] - Worker 2: Processing request 3 (DB is Online)
4836 ms - [INFO] - Worker 3: Waiting for DB connection to process request 4...
4836 ms - [INFO] - Worker 3: Processing request 4 (DB is Online)
5056 ms - [INFO] - Manager: Connection LOST! Clearing event (Red Light). blocking workers...
5834 ms - [INFO] - Worker 1: Waiting for DB connection to process request 5...
5890 ms - [INFO] - Worker 2: Waiting for DB connection to process request 4...
6070 ms - [INFO] - Worker 3: Waiting for DB connection to process request 5...
8056 ms - [INFO] - Manager: Reconnecting...
10063 ms - [INFO] - Manager: Database RECONNECTED. Setting event again.
10063 ms - [INFO] - Worker 1: Processing request 5 (DB is Online)
10064 ms - [INFO] - Worker 2: Processing request 4 (DB is Online)
10064 ms - [INFO] - Worker 3: Processing request 5 (DB is Online)
10848 ms - [INFO] - Worker 2: Waiting for DB connection to process request 5...
10848 ms - [INFO] - Worker 2: Processing request 5 (DB is Online)

Asyncio.to_thread

Sometimes, you might encounter a problem where the functions you want to use do not have asynchronous capabilities. These days, many libraries and frameworks support async operations, but some legacy libraries still do not. In this situation, you can use to_thread.

asyncio.to_thread is a bridge. It allows you to take Blocking I/O (like reading a huge file or using a synchronous library like requests) or Light CPU work and run it without "stopping the world" for your other async tasks.

The Mechanics of `asyncio.to_thread`

When you call to_thread, the following sequence occurs:

The Wrapper: asyncio wraps your synchronous function in a coroutine object.
The Context: It captures the current contextvars (so your database sessions or tokens stay available).
The Executor: It pushes the function into a separate ThreadPoolExecutor.
The Yield: The calling coroutine awaits a Future object. This allows the Event Loop to go do other work while the thread is busy.

import asyncio
import time


def brew_coffee(n: int) -> None:
    print(f"Start brewing coffee #{n}...")

    time.sleep(5)

    print(f"Coffee #{n} is ready!")


async def main() -> None:
    start = time.time()

    tasks = [asyncio.to_thread(brew_coffee, i + 1) for i in range(3)]

    await asyncio.gather(*tasks)

    end = time.time()
    print("Coffee ready!")
    print(f"Total time: {end - start:.2f} seconds")


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

This code is a similar example to the one in the previous post. When we ran that code with asyncio.create_task, it executed synchronously because of the blocking I/O function: time.sleep.

But in this example, if we run this code, it returns

Start brewing coffee #1...
Start brewing coffee #2...
Start brewing coffee #3...
Coffee #2 is ready!
Coffee #1 is ready!
Coffee #3 is ready!
Coffee ready!
Total time: 5.00 seconds

As you can see, the tasks ran asynchronously. You can run legacy blocking I/O functions asynchronously by leveraging to_thread.

Is this the same as Multi-threading? What is the difference?

Feature	with ThreadPoolExecutor()	asyncio.to_thread()
Context Switching	OS-level (Preemptive)	OS-level (Preemptive execution)
Context Variables	Lost (e.g., Trace IDs, Auth tokens)	Preserved (Propagates `contextvars`)
Error Handling	Needs manual `future.exception()` check	Integrated with `try/except` and `TaskGroups`
Lifecycle	Manual (must use `with` or `.shutdown()`)	Automatic (managed by the event loop)
Configuration	Explicit (`max_workers=...`)	Uses the loop's default executor

Actually, to_thread uses multi-threading, but wraps it in a coroutine interface. It doesn’t incur data loss because it automatically propagates context variables, unlike a raw ThreadPool.

Someone might ask “Then, I don’t need to use async coroutine functions, right? Because I can run both sync and async functions asynchronously with to_thread.”

Limitation of `to_thread`

That is absolutely not true. to_thread utilizes threads for I/O bound work, whereas native async tasks are single-threaded. to_thread is a hybrid feature.
It utilizes a global thread pool for processing I/O bound work with legacy blocking I/O functions. However, it can’t process more tasks than the number of threads in the global pool. In contrast, native coroutines can process a much larger number of I/O bound tasks with just one thread.

Note: Default numberof threads in the pool

os.cpu_count()+4) \mathbf{N}_{threads}=\mathrm{min}(32,\, \mathrm{os.cpu\_count}() + 4)

Conclusion

I have introduced some useful functions of asyncio. In the next post, I will demonstrate how to apply these functions to LLM inferences in the next post.

Coroutine series 1) What are Coroutine, Asyncio, I/O bound?

Jun Bae — Wed, 22 Apr 2026 13:40:59 +0000

Introduction

In many cases, we have to run several jobs concurrently. Most developers are likely familiar with multi-threading or multi-processing, both of which Python supports through ThreadPoolExecutor and ProcessPoolExecutor.

However, there is another powerful tool: the ‘coroutine’. While multi-threading and multi-processing are Hardware/OS-based methods, coroutines provide concurrency at the software level.

Python introduced coroutines via asyncio library of Python 3.5 (2015). For a long time, it was just a niche tool primarily for backend developers. However, since the LLM boom of the 2020s, coroutines have become an essential feature for AI engineers.

About this series

I think it would be difficult to cover everything about coroutines in a single post, which is why I created this coroutine series. I will explore the features of coroutine and their specific applications in AI step by step.

What is a Coroutine?

While multi-processing is hardware-based parallellism and multi-threading is OS-level concurrency, coroutines are software-based concurrency.

Method	Multi-Processing	Multi-Threading	Coroutines (Asyncio)
Type	Hardware Parallelism	Kernel Concurrency	Software Concurrency
The Scheduler	The OS Kernel	The OS Kernel	The Python Event Loop
Switching Style	Preemptive (Forceful)	Preemptive (Forceful)	Cooperative (Polite)
Resource Cost	High (Separate Memory)	Medium (Stack + Kernel Objects)	Low (Tiny Object on Heap)
Awareness	The CPU knows.	The OS knows.	Only Python knows.

Multi-processing requires multiple CPU logical cores, while multi-threading ostensibly utilizes multiple threads within a single process. However, as you may know, technically Python multi-threading is not truly capable of executing tasks simultaneously due to the Global Interpreter Lock (GIL). Consequently, it is better suited for I/O-bound concurrency rather than true parallellism.

For CPU-bound tasks: Threads must hold the GIL to execute bytecode. They contend for the lock, causing overhead. A multi-threaded CPU-bound program in Python is often slower than a single-threaded one due to lock contention and context switch overheads.
For I/O-bound tasks: The GIL is released when a thread performs a blocking I/O operation (e.g., socket.recv, time.sleep, file I/O). This allows other threads to run Python code while the first thread waits in the OS kernel. Consequently, standard threading is effective for concurrent I/O in Python.

But what if we don’t even need to allocate multiple threads in an execution? In most I/O-bound scenarios, having several threads is actually unnecessary—especially when you are making API calls and simply waiting for a response. This is precisely why Python introduced support for coroutines.

Why has it become essential?

As mentioned above, when coroutines were first introduced in Python, they were a niche tool used by only a small group of developers. Today, however, we rely heavily on LLMs, most of which are accessed through API calls. Unless you are running a model locally on your own PC, you must request an inference from a server—even in an on-premises environment. (What developer would write code on H100 while it is already serving an LLM model?)

When working with LLM APIs, over 90% of your execution time is spent waiting for a response. Instead of sitting idle while waiting, your process could be handling the next task, not just be laid back.

If you have ten concurrent requests, you don’t need to wait for the first user's prompt to finish. While waiting for the first response, you can send the second, third, and eventually all ten prompts almost simultaneously (provided your LLM server has enough throughput).

Furthermore, in the Agentic AI structure, many inferences are independent of one another. Then sending prompts to LLM servers concurrently can significantly reduce your total latency.

In this context, your Python process is not the Calculator; it is the Traffic Controller.

How can I leverage coroutines?

Python has a built-in library called asyncio that allows you to implement this.

Imagine you are making several cups of coffee and have multiple coffee machines. But you always use only one machine and do nothing while wating for one cup of coffee because you are too lazy to manage tasks concurrently. In code, it looks like this:

Input

import time

def brew_coffee(n: int) -> None:
    print(f"Start brewing coffee #{n}...")
    # This blocks the entire thread. Nothing else can happen.
    time.sleep(5)
    print(f"Coffee #{n} is ready!")

def main() -> None:
    start = time.time()

    # Sequential execution
    for i in range(3):
        brew_coffee(i + 1)

    end = time.time()
    print(f"Total time: {end - start:.2f} seconds")

if __name__ == "__main__":
    main()

Output

Start brewing coffee #1...
Coffee #1 is ready!
Start brewing coffee #2...
Coffee #2 is ready!
Start brewing coffee #3...
Coffee #3 is ready!
Total time: 15.00 seconds

But as your cafe grows, you have to become more efficient now that you can’t keep customers waiting by using only one machine at a time. This is where you—the barista—act as a coroutine. You start one machine, and while it's brewing, you immediately start the second and third.

Input

import asyncio
import time
from typing import List, Coroutine, Any

## 1. Define a coroutine using 'async def'
async def brew_coffee(n: int) -> None:
    print(f"Start brewing coffee #{n}...")

    # 2. 'await' hands control back to the Event Loop
    # This is non-blocking sleep.
    await asyncio.sleep(5)

    print(f"Coffee #{n} is ready!")

async def main() -> None:
    start = time.time()

    # 3. Create a list of coroutine objects (they haven't run yet!)
    tasks: List[Coroutine[Any, Any, None]] = [brew_coffee(i + 1) for i in range(3)]

    # 4. Schedule them concurrently and wait for all to finish
    await asyncio.gather(*tasks)

    end = time.time()
    print("Coffee ready!")
    print(f"Total time: {end - start:.2f} seconds")

if __name__ == "__main__":
    # 5. The entry point for the async world
    asyncio.run(main())

Output

Start brewing coffee #1...
Start brewing coffee #2...
Start brewing coffee #3...
Coffee #1 is ready!
Coffee #2 is ready!
Coffee #3 is ready!
Coffee ready!
Total time: 5.00 seconds

Now you can make three cups of coffee in just five seconds. You didn’t hire more workers, you didn’t move faster, and you didn’t buy extra equipment. You just took advantage of the idle time you previously wasted.

So, what exactly are coroutines, asyncio, await, and gather? Let’s dive into them one by one.

Coroutine functions

In Python, a standard block of code starting with def is simply called a function. But if you add async in front of it like async def, it becomes a coroutine function. This tells Python that the function is intended to run asynchronously. ‘Asynchronous’ means that tasks can be handled independently rather than waiting for one to finish before starting the next.

Asynchronous vs Synchronous

The dictionary definitions of these words can sometimes be confusing. Some people might think ‘synchronous jobs? Does it mean the jobs would be run at the same time?’ It helps to think of "synchronous" like "syncing" a device. When you sync two devices, they must be "in step" or connected to share data. Therefore, synchronous programming means tasks are executed in a strict sequence; the previous task must finish before the next one begins. On the other hand, asynchronous means tasks are decoupled, allowing the program to move on to other work without waiting for a specific operation to complete.

Synchronous tasks (Dependent): Washing your clothes and then hanging them out to dry. You cannot hang the laundry until the washer has finished its cycle.
Asynchronous tasks (Independent): Running the dishwasher and microwaving your lunch. You don't need to wait for the dishwasher to finish to start the microwave; both can happen at the same time.

async def and await

As mentioned earlier, the ‘async’ keyword allows you to define a coroutine. Creating one is simple, but running it requires a slightly different approach than a standard function.

import asyncio

async def brew_coffee(n):
    print(f"Start brewing coffee #{n}...")
    await asyncio.sleep(5)
    print(f"Coffee #{n} is ready!")

This is the coffee example, but how can I run this coroutine? Same as normal function, just brew_coffee(1) will this be executed? If you run brew_coffee(1), you will see this output.

<coroutine object brew_coffee at 0x000002957D795BE0>

Unfortunately, it doesn’t start making coffee. Then how can I start it? You have to write await.

await brew_coffee(1)

Start brewing coffee #1...
Coffee #1 is ready!

Now you just made one cup of coffee. await means the process will wait until the job finishes. But wait a minute.. Wait until jobs done? Then what if I start two async jobs like this.

await brew_coffee(1)
await brew_coffee(2)

Then it will return this:

Start brewing coffee #1...
Coffee #1 is ready!
Start brewing coffee #2...
Coffee #2 is ready!

Right.. This was not run concurrently. If it waits on await, how can I start several jobs concurrently?

asyncio.create_task(brew_coffee(1))
asyncio.create_task(brew_coffee(2))

# Or as the first example code
works = [brew_coffee(1), brew_coffee(2)]
await asyncio.gather(*works)

This returns

Start brewing coffee #1...
Start brewing coffee #2...
Coffee #1 is ready!
Coffee #2 is ready!

Now it works!

What is `await` exactly?

When a coroutine encounters an await expression, it follows a three-step process:

Suspend: When await is encountered, the coroutine saves its current instruction pointer and stack variables.
Yield: Control is returned to the Event Loop.
Resume: When the awaited operation completes, the Event Loop restores the coroutine's state and resumes execution at the instruction following the await.

Long story short, when a task hits await, it says, "I’m waiting for something; go ahead and run someone else." And Event Loop will kick off another task. Only after the awaited operation is complete will the loop return to the original task to finish the job.

But a word of caution: async and await are not a silver bullet! You must avoid using "blocking" I/O functions inside a coroutine.

The Danger of Blocking I/O

Consider this version of our coffee example. It looks almost identical to our successful coroutine code, but I have replaced asyncio.sleep(5) with time.sleep(5).

import asyncio
import time
from typing import List, Coroutine, Any


async def brew_coffee(n: int) -> None:
    print(f"Start brewing coffee #{n}...")

    time.sleep(5)

    print(f"Coffee #{n} is ready!")


async def main() -> None:
    start = time.time()

    tasks: List[Coroutine[Any, Any, None]] = [brew_coffee(i + 1) for i in range(3)]

    await asyncio.gather(*tasks)

    end = time.time()
    print("Coffee ready!")
    print(f"Total time: {end - start:.2f} seconds")


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

Output

Start brewing coffee #1...
Coffee #1 is ready!
Start brewing coffee #2...
Coffee #2 is ready!
Start brewing coffee #3...
Coffee #3 is ready!
Coffee ready!
Total time: 15.00 seconds

Yes, this is not a coroutine. Why? Because time.sleep(5) is a blocking I/O operation. If you use blocking functions (like time.sleep, requests.get, or standard file reading) inside a coroutine, you effectively stop the entire "Traffic Controller" from working. For a coroutine to be effective, you must use non-blocking equivalents that know how to await.

Note: You might think this mistake is too obvious to mention. But in production-level environments, you should be very careful about this issue. Real-world projects often consist of complex codebases with dozens or even hundreds of files. You must carefully audit your coroutines to ensure that no blocking I/O is hidden within whole downstream. One small blocking call can bring your entire high-performance system to a crawl.

Conclusion

In this post, we have explored the fundamentals of coroutines and how to implement them at a basic level. I will cover more advanced features of asyncio and demonstrate how to leverage them for AI inference in the next post of this series.

KV Cache and Prompt Caching: How to Leverage them to Cut Time and Costs

Jun Bae — Wed, 22 Apr 2026 13:35:03 +0000

Introduction

A Problem of LLM Inference

In the transformer structure, the model calculates the $K,V\mathbf{K}, \mathbf{V}$ matrices using weight matrices $W\mathbf{W}$ . When an input $x0\mathbf{x}_0$ vector enters the model, it is first multiplied by the $Wq,Wk,Wv\mathbf{W}_q, \mathbf{W}_k, \mathbf{W}_v$ matrices. This yields the $q0,k0,v0\mathbf{q}_0, \mathbf{k}_0, \mathbf{v}_0$ vectors. As you iterate this process and stack the $k,v\mathbf{k}, \mathbf{v}$ vectors, they form the $K,V\mathbf{K}, \mathbf{V}$ matrices.

Assume you have successfully generated an output token after completing the transformer process, and that token is $x1\mathbf{x}_1$ . Here lies the problem: for subsequent inference, the model must calculate not only $k1,v1\mathbf{k}_1, \mathbf{v}_1$ but $k0,v0\mathbf{k}_0, \mathbf{v}_0$ again. Because the attention score is calculated as $q1KTV\mathbf{q}_1\mathbf{K}^{T}\mathbf{V}$ , it requires the $k,v\mathbf{k}, \mathbf{v}$ vectors from all previous inputs. This results in redundant computations every time a new token is generated. As the number of input tokens grows, these recomputations become significantly time-consuming.

What is KV cache exactly?

If you understant the problem, you might have already thought of the solution. Yes, the solution is to sotre the previous $k,v\mathbf{k}, \mathbf{v}$ vectors in what is known as a “cache”. It is a relatively straightforward concept.

A Simple example of KV cache

Let’s continue with the example from the introduction. When generating a token subsequent to $x1\mathbf{x}_1$ , what needs to be computed? First, we calculate $q1=x1Wq\mathbf{q}_1=\mathbf{x}_1\mathbf{W}_q$ . And then, you need the $K,V\mathbf{K}, \mathbf{V}$ matrices. Let’s look at this step-by-step.

We need to form $K\mathbf{K}$ using vectors $k0\mathbf{k}_0$ and $k1\mathbf{k}_1$
This requires computing $k0=x0Wk\mathbf{k}_0=\mathbf{x}_0\mathbf{W}_k$ and $k1=x1Wk\mathbf{k}_1=\mathbf{x}_1\mathbf{W}_k$ . Notice that we have just recomputed $k0\mathbf{k}_0$ .
Similarly, we need $V\mathbf{V}$ , composed of vectors $v0\mathbf{v}_0$ and $v1\mathbf{v}_1$ .
Following the same logic as step 2, we calculate $v0=x0Wv\mathbf{v}_0=\mathbf{x}_0\mathbf{W}_v$ and $v1=x1Wv\mathbf{v}_1=\mathbf{x}_1\mathbf{W}_v$ .
Finally, with the $K,V\mathbf{K}, \mathbf{V}$ matrices ready, we can compute $q1KTV\mathbf{q}_1\mathbf{K}^{T}\mathbf{V}$ .

As you can see, the initial computation for $x0\mathbf{x}_0$ required three vector calculations( $q0,k0,v0\mathbf{q}_0,\mathbf{k}_0,\mathbf{v}_0$ ). However, for the next token, that number jumps to five( $k0,v0,q1,k1,v1\mathbf{k}_0,\mathbf{v}_0,\mathbf{q}_1,\mathbf{k}_1,\mathbf{v}_1$ ). This grows to seven for the third token and nine for the fourth. If your input is 1,000 token long, you would have to perform 2,001 vector computations just for the last token. This is computationally expensive and highly inefficient.

This is where the "cache" comes in. If we store the entire $K,V\mathbf{K}, \mathbf{V}$ matrices after each step, we no longer need to recompute $k0,v0\mathbf{k}_0,\mathbf{v}_0$ . The number of computations remains constant at three per token $qn,kn,vn\mathbf{q}_n,\mathbf{k}_n,\mathbf{v}_n$ , regardless of how many tokens have already been processed.

Note: This explanation assumes a decoder-only architecture. You might wonder why we don't cache the $Q\mathbf{Q}$ matrix in an encoder; however, since most modern LLMs are decoder-only, that isn't a concern here. Additionally, "hitting the cache" requires both the token and its position to be identical due to positional embeddings.

How much fater is it?

Various engineering blogs and papers (e.g., from NVIDIA, Hugging Face, and Databricks) have quantified the benefits of KV caching.

Benchmark: Generating 1,000 Tokens (Llama-2-7B)

With KV Cache: ~20–30 seconds (consistent speed per token).
Without KV Cache: > 2–3 minutes (each token takes progressively longer than the last).

As the number of tokens grows, the gap widens significantly. Given that some SOTA models support now support context lengths of 1M, this type of cache engineering has become indispensable.

How to leverage KV cache when serving models?

In most cases, you don’t need to worry about complex configurations. Since KV cache has become the standard serving method, libraries like vLLM and Huggingface Transformers activate KV cache automatically by default provided you have enough memory.

In Transformers, the use_cache parameter in model.generation_config controls this behavior; its default value is True.

In vLLM, it allocates the remaining memory—defined by gpu_memory_utilization—to the KV cache after the model weights are loaded in the memory.

vLLM is particularly well-known for its sophisticated cache engineering techniques, such as PagedAttention. Because of this, vLLM offers massive throughput with almost no memory fragmentation, making it a unmatched top-tier choice for efficient LLM serving.

The Problem with KV Caching

Memory Consumption

The memory required for the KV cache increases linearly with sequence length and batch size.

KV Cache Memory Formula:

\textbf{Memory}_{KV} = 2\,\times\,\textbf{Batch}\,\times\,\textbf{SeqLen}\,\times\,\textbf{Layers}\,\times\,\textbf{HiddenDim}\,\times\,\textbf{Precision}

2: One for keys and one for values.
Batch: Number of concurrent users being served.
SeqLen: The length of the context/sequence.
Precision: Bytes (e.g., 2 bytes for FP16, 1 byte for FP8).

Let me give you a simple example of how it increases with a typical LLM model.

Example: Llama-3-70B (FP16)

Model Weights: about 140GB
KV Cache (1 user, 1k tokens): ~0.3GB
KV Cache (1 user, 128k tokens): ~40GB
KV Cache (64 users, 16k tokens): ~310GB
KV Cache (64 users, 128k tokens): ~2,560GB

As you can see, the KV cache footprint explodes as you increase context length or batch size. Even with a 1 GB allocation per user, you might need more than one H100 GPU just to serve 100 concurrent users.

Beyond memory capacity, this also impacts latency. While GPUs are exceptionally fast at mathematics, they are limited by memory bandwidth. The GPU must constantly fetch the cached data from VRAM to the chip to generate each new token. This is why you may see high latency even when GPU compute utilization is only at 10%—the memory bandwidth has become the bottleneck.

There are emerging solutions to these problems, such as Grouped-Query Attention(GQA), PagedAttention, Quantization. However I don’t think these are enough. Definitely KV cache is hindering the scalability of LLMs today. If LLMs eventually hit a "dead end," as Yann LeCun has suggested, I believe the inefficiencies of the KV cache may be a contributing factor. It has become so indispensable that developers cannot afford to abandon it. If a new model is too massive to leverage the KV cache with current computing resources, it might be useless in the real world as consumers will not tolerate prohibitively slow inference speeds.

What was once a solution is now a massive problem.. So ironic. We’ll have to see how big tech companies tackle this tricky challenge in the future. Isn’t it exciting?

Prompt (Context) caching

This is another method in LLM engineering. While different from standard KV caching, it significantly improves efficiency and reduces computation time. Furthermore, if you use LLM services via API—such as Gemini or OpenAI—, this method can save you both time and money.

What is Prompt Caching?

Standard KV caching typically occurs within a single turn. Once a turn ends, the cache is cleared and the process restarts from scratch for the next turn.

Prompt caching extends this concept across multiple turns. After one turn is completed, the server saves the $K,V\mathbf{K}, \mathbf{V}$ matrices and retains them for a period of time. When the next turn begins, the model can reuse those exact same $K,V\mathbf{K}, \mathbf{V}$ matrices. Let me give you a simple example.

Example: A Multi-Turn Conversation

First turn

System prompt: Answer user’s question.

User prompt: Hi

Assistant prompt: Hi! How can I help you?

Second turn

User prompt: Who are you?

Assistant prompt:

In the first turn, the input tokens consist of $(Answer),(user’s),(question),(Hi)(\textrm{Answer}), (\textrm{user's}), (\textrm{question}),(\textrm{Hi})$ (I knowingly omitted tokens of ‘\n’, spacing, BOS, EOS or start token of certain prompt for convenience). Of course, the model wil compute the $K,V\mathbf{K}, \mathbf{V}$ matrices for all tokens and begin generating responses. After it finishes generating entire assistant prompt, it will have also stacked the $k,v\mathbf{k}, \mathbf{v}$ vectors for $(Hi!),(How),(can),(I),(help),(you?)(\textrm{Hi!}), (\textrm{How}), (\textrm{can}),(\textrm{I}),(\textrm{help}),(\textrm{you?})$ onto the matrices.

Then, in multi-turn structure, what happens during the second turn? As you might expect, the model reprocesses the $k,v\mathbf{k}, \mathbf{v}$ vectors for the previous context: $(Answer),(user’s),(question),(Hi)(\textrm{Answer}), (\textrm{user's}), (\textrm{question}),(\textrm{Hi})$ and $(Hi!),(How),(can),(I),(help),(you?)(\textrm{Hi!}), (\textrm{How}), (\textrm{can}),(\textrm{I}),(\textrm{help}),(\textrm{you?})$ . it then computes the vectors for the new user prompt, and begin generating an answer.

Of course, the model could leverage a KV cache, but rebuilding that cache from scratch would be highly inefficient. It would consume too much time performing the exact same operations. However, if we cache the $K,V\mathbf{K}, \mathbf{V}$ matrices from the previous turn, we can avoid recomputing the $k,v\mathbf{k}, \mathbf{v}$ vectors for the initial instructions and the first exchange. By retrieving these vectors from memory, the model can begin processing immediately from the new prompt, “Who are you?”. Here is an example image from the OpenAI website.

As you can see in the image, the token positions must remain exactly the same. If even a single word is inserted, the cache will be invalidated from that point forward because the positional encodings will shift. Therefore, to leverage prompt caching effectively, you should keep your token sequences as consistent as possible.

Note: Currently, the terminology for this technique has not been standardized. OpenAI calls it ‘Prompt Caching,’ vLLM uses ‘Prefix Caching,’ and Google refers to it as ‘Context Caching’. They all refer to the same concept. Unfortunately, this kind of naming inconsistency happens occasionally in the AI field. :(

With this method, tokens from previous turns don’t even need to be re-processed by the transformer layers. As long as the cache exists, the model can skip the heavy computation for those segments entirely. You simply load the cache and move forward without needing to re-evaluate the earlier parts of the conversation. It’s an incredibly efficient way to handle long-context interactions

Saving Money with Prompt Caching

When you successfully "hit" the prompt cache during a multi-turn conversation, several LLM API providers—such as OpenAI and Google Gemini—offer significant discounts. Below, I’ve broken down the current discount policies for both platforms.

OpenAI API

They applies a comprehensive half-price policy to input costs.

While the chart above uses specific examples, most recent models—including the GPT-5 family—benefit from this pricing. They still haven’t updated this chart, despite it being over half a year since GPT-5 launched.

Activation: The minimum token length to trigger prompt caching is 1,024 tokens.
TTL (Time to Live): The cache typically persists for 5 to 10 minutes during peak hours, and up to one hour during off-peak times.
Extended Caching: Certain models (such as gpt-5.2 and gpt-4.1) support extended prompt caching, allowing the cache to be retained for up to 24 hours.

Google Gemini API

Gemini offers two distinct caching tiers: Implicit and Explicit.

Implicit Caching: This is automatically enabled for most Gemini models. The minimum threshold is 1,024 tokens for Flash models and 4,096 tokens for Pro models. However, Google does not strictly guarantee cost savings with this tier; it is primarily an optimization for latency.
Explicit Caching: This allows for a cost-saving guarantee but must be configured manually. You can define specific parameters like this:

cache = client.caches.create(
    model=model,
    config=types.CreateCachedContentConfig(
      display_name='sherlock jr movie', # used to identify the cache
      system_instruction=system_prompt,
      contents=content,
      ttl="300s",
  )

The default TTL is 1 hour and no minimum or maximum time. The billing is based on these factors:

Cache token count: The number of input tokens cached, billed at a reduced rate when included in subsequent prompts.

Storage duration: The amount of time cached tokens are stored (TTL), billed based on the TTL duration of cached token count. There are no minimum or maximum bounds on the TTL.

Other factors: Other charges apply, such as for non-cached input tokens and output tokens.

And also they explicitly show the exact costs of cached tokens like this:

This is Gemini-3-Pro example and it varies depending on models.

Important Note: As seen in the Gemini 3 Pro pricing, you are charged a storage fee. Be mindful when setting a long TTL.

Conclusion

In this post, we have explored two pillars of LLM optimization: KV cache and Prompt caching. These techniques are indispensable for reducing latency and lowering operational costs. However, these efficiencies come with a significant trade-off: a massive increase in memory consumption.

This "memory wall" has forced developers and enterprises to secure high-performance GPUs with vast amounts of VRAM, not just for training, but for inference as well. It is no surprise that memory manufacturers like Samsung, SK Hynix, and Micron have seen such a surge in demand; their hardware is the literal foundation upon which these caching methods reside.

Currently, the memory bottleneck is one of the most significant "sticking points" in LLM architecture. Many of the world's leading tech companies are racing to find more efficient ways to handle these tricky difficulties. If the industry can overcome these scaling limits with new, innovative architectures, it will represent the next major breakthrough in the evolution of Artificial Intelligence.

Why LoRA? Understanding the representative PEFT

Jun Bae — Wed, 22 Apr 2026 13:24:04 +0000

Why LoRA?

Low-Rank Adaptation (LoRA) has revolutionized the way we approach Large Language Models (LLMs). As the most prominent Parameter-Efficient Fine-Tuning (PEFT) method, LoRA allows developers to adapt massive models like Llama 3 or GPT-4 to specific tasks without needing a cluster of H100 GPUs.

But how exactly does it work, and why is it so effective? In this post, we’ll dive into the mathematical intuition behind LoRA, the concept of "intrinsic dimension," and why this method is a game-changer for AI engineers.

The Problem: The Cost of Scale

When it comes to traditional statistical models—like simple or generalized linear models, Tree model series (Random Forest, Light GBM, etc)—, we can train our mod

els from scratch using only their blueprints and architectures. Historically, this was the standard approach and was not particularly difficult or cumbersome.

However, the Deep Learning revolution changed the landscape. As model sizes ballooned from millions to billions of parameters, retraining or even fine-tuning a model became logistically impossible for most individuals and even many companies.

The Hypothesis of Intrinsic Dimension

Here is the crucial insight: We don't need to train all the parameters.

Research suggests that over-parameterized models reside on a low intrinsic dimension. Simply put, while a model might have billions of parameters, the "knowledge" required to solve a specific new task can be represented by a much smaller subset of variables.

This premise was validated before LoRA's invention by researchers like Li et al. (2018) and Aghajanyan et al. (2020), who showed that learning happens in a lower-dimensional subspace. LoRA operationalizes this theory.

What is LoRA? The Mechanics

So what exactly is LoRA?

The concept is remarkably straightforward. As mentioned, LoRA is a fine-tuning method that trains the model using only a fraction of the total parameters.

The Mathematical Formulation

Let’s look at the linear algebra. Suppose we have a pre-trained weight matrix $W0∈Rd×k\mathbf{W}_0 \in \mathbb{R}^{d \times k}$ .

In full fine-tuning, we update the weights by learning a cumulative gradient update $ΔW\Delta \mathbf{W}$ : $h=(W0+ΔW)x\mathit{h} = (\mathbf{W}_0 + \Delta \mathbf{W})\mathit{x}$

LoRA constrains this update $ΔW\Delta \mathbf{W}$ by representing it as the product of two smaller matrices, $B\mathbf{B}$ and $A\mathbf{A}$ , with a low rank $r$ :

Where:

$B∈Rd×r\mathbf{B} \in \mathbb{R}^{d \times r}$
$A∈Rr×k\mathbf{A} \in \mathbb{R}^{r \times k}$
$\ll \min(d, k)$ (The rank, usually set between 8 and 64)

All you need to do is to train A and B matrices. Then what are the A and B matrices?

Initialization Strategy

To ensure the training starts exactly where the pre-trained model left off (i.e., $ΔW\Delta \mathbf{W}$ is zero at the beginning), we initialize:

Matrix A: Random Gaussian distribution $N(0,σ2)\mathcal{N}(0, \sigma^2)$
Matrix B: Zeros

Therefore, $BA=0\mathbf{B}\mathbf{A} = 0$ initially, ensuring no destabilizing noise is added to the model at step zero.

The Scaling Factor $α\alpha$

One detail often missed is the scaling factor $α\alpha$ . The update is actually scaled as:

\Delta \mathbf{W} = \frac{\alpha}{r} (\mathbf{B}\mathbf{A})

This allows us to tune the learning rate effectively regardless of the rank $r$ we choose.

Then, the problem is.. is it accurate enough?

The answer is ‘yes’. And even more accurate.

Why is LoRA Effective?

Let’s talk numbers. Assume $d = 10, 000$ and $k = 10, 000$ with a rank $r = 16$ .

Full Fine-Tuning: $\times 10,000 = 100,000,000$ parameters.
LoRA: $\times 16) + (16 \times 10,000) = 320,000$ parameters.

That is a 99.68% reduction in trainable parameters. But is it accurate?

Performance vs. Efficiency

Yes, it is surprisingly accurate.

As shown in the original LoRA paper, on benchmarks like WikiSQL, LoRA performs neck-and-neck with full fine-tuning. In some cases, it even outperforms the baseline because it acts as a form of regularization, preventing the model from overfitting to the small training set.

Then why? It might seem counterintuitive that tuning less than 0.01% of the parameters is sufficient. In reality, it is.

For those with a background in statistics, they can feel a bit familiar with this concept. There is a very famous and conventional method of decreasing dimension: PCA (Principal Component Analysis).

The Intuition: PCA Analogy

Imagine a dataset containing "Biological Species" and "Number of Legs." These two features are highly correlated; knowing the species often tells you the leg count. You don't need two independent dimensions to represent this variance.

There are a few more dimensionality reduction methods like clustering.

Anyway, the important thing is that the concept—reducing the dimension can improve not only training efficiency but performance— is valid.

Similarly, in deep neural networks, weight updates often occur in a subspace with high correlations. We can project these high-dimensional updates into a lower-rank space without losing significant information.

And LoRA takes advantage of this concept and it maximizes this effect. It doesn’t have any specific mathematic formula like PCA. It just creates random subspace matrices and trains them.

According to the authors, the subspaces of LLM parameters are more similar than previously thought. As you see in the image, the first dimension shares very similar subspaces with all of other dimensions. Therefore, they said only one rank worked quite well for fine-tuning GPT-3 model.

Mitigating Catastrophic Forgetting

A unique advantage of LoRA is that it freezes the original weights ( $W0\mathbf{W}_0$ ).

Full Fine-Tuning: Overwrites $W0\mathbf{W}_0$ . If you train heavily on coding tasks, the model might "forget" how to write poetry (Catastrophic Forgetting).
LoRA: Keeps $W0\mathbf{W}_0$ intact. You can train one adapter for Coding and another for Poetry. You can simply swap the adapters at runtime without reloading the base model.

Conclusion

LoRA has become an indispensable tool in the AI toolkit. Today, most individuals—and even many companies—lack the resources to perform full fine-tuning on models with billions of parameters. Even with access to H100 or A100 GPUs, full fine-tuning is often prohibitively slow and, as demonstrated, rarely provides a better return on investment.

Furthermore, in generic models, an individual query effectively activates only a small portion of the parameters. This is a concept that has been further expanded upon in another famous method, Mixture of Experts (MoE), which I will cover in a future post.

You can leverage LoRA very easily because the libraries these days are so convenient such as peft or trl. Just try it. Then you will see that fine-tuning about 10 billions models takes about 3-6 hours with one H100. And the output files are only hundreds Megabytes. (Of course, you have to merge them with original model when you serve it.)

Reference

Chunyuan Li, Heerad Farkhoor, Rosanne Liu, and Jason Yosinski. Measuring the Intrinsic Di mension of Objective Landscapes. arXiv:1804.08838 [cs, stat], April 2018a. URL http: //arxiv.org/abs/1804.08838. arXiv: 1804.08838.

Armen Aghajanyan, Luke Zettlemoyer, and Sonal Gupta. Intrinsic Dimensionality Explains the Effectiveness of Language Model Fine-Tuning. arXiv:2012.13255 [cs], December 2020. URL http://arxiv.org/abs/2012.13255.

Hu, E. J., Shen, Y., Wallis, P., Allen-Zhu, Z., Li, Y., Wang, S., Wang, L., & Chen, W. (2021, June 17). LORA: Low-Rank adaptation of Large Language Models. arXiv.org. https://arxiv.org/abs/2106.09685

Forem: Jun Bae

Coroutine series 3) Coroutines for LLM inference

Introduction

Coroutines for LLM

LLM API SDK

Example comparing async and sync functions

Sync vs Async

AI Framework

Conclusion

Coroutine series 2) Useful Asyncio Functions

Introduction

Coroutine functions

gather and TaskGroup

Differences: TaskGroup vs. gather

Why is TaskGroup favored over gather?

Synchronization Primitives: Queue and Event

Asyncio.Queue

Key Methods

Asyncio.Event

Asyncio.to_thread

The Mechanics of asyncio.to_thread

Limitation of to_thread

Conclusion

Coroutine series 1) What are Coroutine, Asyncio, I/O bound?

Introduction

About this series

What is a Coroutine?

Why has it become essential?

How can I leverage coroutines?

Coroutine functions

Asynchronous vs Synchronous

async def and await

What is await exactly?

The Danger of Blocking I/O

Conclusion

KV Cache and Prompt Caching: How to Leverage them to Cut Time and Costs

Introduction

A Problem of LLM Inference

What is KV cache exactly?

A Simple example of KV cache

How much fater is it?

How to leverage KV cache when serving models?

The Problem with KV Caching

Memory Consumption

Prompt (Context) caching

What is Prompt Caching?

Saving Money with Prompt Caching

Conclusion

Why LoRA? Understanding the representative PEFT

Why LoRA?

The Problem: The Cost of Scale

The Hypothesis of Intrinsic Dimension

What is LoRA? The Mechanics

The Mathematical Formulation

Initialization Strategy

The Scaling Factor α\alphaα

Why is LoRA Effective?

Performance vs. Efficiency

The Intuition: PCA Analogy

Mitigating Catastrophic Forgetting

Conclusion

Reference

`gather` and `TaskGroup`

Differences: `TaskGroup` vs. `gather`

Why is `TaskGroup` favored over `gather`?

Synchronization Primitives: `Queue` and `Event`

The Mechanics of `asyncio.to_thread`

Limitation of `to_thread`

What is `await` exactly?

The Scaling Factor $α\alpha$