async await LLM call

LLM APIs are I/O-bound operations — your program spends most of its time waiting for the network response from the API server, not doing computation. In synchronous code, each LLM call blocks the entire thread until the response returns. With async/await, the event loop can handle other tasks while waiting for the response, dramatically improving throughput.
LLM API是I/O密集型操作——你的程序大部分时间都在等待API服务器的网络响应，而不是在进行计算。在同步代码中，每次LLM调用都会阻塞整个线程直到响应返回。使用async/await后，事件循环可以在等待响应时处理其他任务，大幅提升吞吐量。

Core Concepts

async / await

async def declares a coroutine function. When called, it returns a coroutine object, not the result immediately. await suspends the coroutine until the awaited operation completes, allowing the event loop to run other tasks.
async def声明一个协程函数。调用时返回一个协程对象，而不是立即返回结果。await挂起协程直到被等待的操作完成，让事件循环可以运行其他任务。

import asyncio

# async def 声明这是一个协程函数
# async def declares this as a coroutine function
async def fetch_llm_response(prompt: str):
    """
    模拟LLM API调用 - 等待2秒后返回结果
    Simulate an LLM API call - wait 2 seconds then return result
    """
    # await 挂起当前协程，让事件循环处理其他任务
    # await suspends the current coroutine, 
    # allowing event loop to handle other tasks
    await asyncio.sleep\(2\)  # 模拟网络延迟 / Simulate network latency
    # sleep\(2\) <-- '\' is to let wordpass correctly save it, nothing else. 
    # 这个 '\(' 奇怪的写法是为了网站 wordpress 能正确存储页面，和python无关
    return f"Response to: {prompt}"



async def main():
    # 调用async函数返回协程对象，需要await才能获取结果
    # Calling an async function returns a coroutine object; need await to get result
    result = await fetch_llm_response("Hello")
    print(result)
    


# asyncio.run() 创建事件循环并运行main协程
# asyncio.run() creates an event loop and runs the main coroutine
asyncio.run(main())

Event Loop

The event loop is the core of asyncio. It manages and schedules all coroutines, switching between them when they hit await points. Think of it as a traffic controller — when one task is waiting for I/O, it switches to another task that’s ready to run.
事件循环是asyncio的核心。它管理和调度所有协程，在它们遇到await点时进行切换。可以把它想象成一个交通调度员——当一个任务在等待I/O时，它切换到另一个可以运行的任务

import asyncio
import time

async def task(name: str, delay: float):
    print(f"[{time.strftime('%H:%M:%S')}] Task {name}: starting")
    await asyncio.sleep(delay)  # 模拟I/O等待 / Simulate I/O wait
    print(f"[{time.strftime('%H:%M:%S')}] Task {name}: done")
    return f"Result from {name}"

async def main():
    # 并发执行三个任务 - 总耗时约2秒而非6秒
    # Run three tasks concurrently - total ~2s not 6s
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 1.5)
    )
    print(f"All results: {results}")

asyncio.run(main())

# 输出示例 / Example output:
# [14:30:01] Task A: starting
# [14:30:01] Task B: starting
# [14:30:01] Task C: starting
# [14:30:02] Task B: done
# [14:30:02.5] Task C: done
# [14:30:03] Task A: done
# All results: ['Result from A', 'Result from B', 'Result from C']

OpenAI Async Calls

Basic Async Call

基础异步调用

import asyncio
import os
from openai import AsyncOpenAI  # 异步客户端 / Async client

# ============================================================
# 1. 初始化异步客户端
# 1. Initialize async client
# ============================================================
# AsyncOpenAI 提供与OpenAI相同的接口，但所有方法都是异步的
# AsyncOpenAI provides the same interface as OpenAI, but all methods are async
# 推荐从环境变量读取API Key / Recommended to read API Key from env var
client = AsyncOpenAI(
    api_key=os.environ.get("OPENAI_API_KEY"),
    # timeout=60.0,  # 可配置超时 / Optional timeout
    # max_retries=2,  # 可配置重试 / Optional retries
)

# ============================================================
# 2. 定义异步调用函数
# 2. Define async call function
# ============================================================
async def get_chat_completion(prompt: str, model: str = "gpt-3.5-turbo") -> str:
    """
    异步获取LLM响应
    Asynchronously get LLM response
    
    Args:
        prompt: 用户提示词 / User prompt
        model: 模型名称 / Model name
    
    Returns:
        LLM响应内容 / LLM response content
    """
    try:
        # await client.chat.completions.create 是非阻塞的
        # await client.chat.completions.create is non-blocking
        # 发起HTTP请求后，事件循环可以处理其他任务
        # After initiating HTTP request, event loop can handle other tasks
        response = await client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.7,
            max_tokens=500,
        )
        
        # 提取响应内容 / Extract response content
        # response.choices[0].message.content 包含完整响应
        # response.choices[0].message.content contains the full response
        return response.choices[0].message.content
        
    except Exception as e:
        # 捕获并处理异常 / Catch and handle exceptions
        print(f"Error calling OpenAI API: {e}")
        return f"Error: {str(e)}"

# ============================================================
# 3. 执行异步调用
# 3. Execute async call
# ============================================================
async def main():
    # 单个调用 / Single call
    result = await get_chat_completion("What is the capital of France?")
    print(f"Response: {result}")

# 启动事件循环 / Start the event loop
if __name__ == "__main__":
    asyncio.run(main())

Concurrent Multiple Calls

并发调用多个LLM

import asyncio
import os
import time
from openai import AsyncOpenAI

client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

async def summarize_text(text: str, idx: int) -> dict:
    """
    异步总结单条文本
    Asynchronously summarize a single text
    """
    start = time.time()
    
    try:
        response = await client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "Summarize the following text in one sentence."},
                {"role": "user", "content": text}
            ],
            max_tokens=100,
        )
        
        elapsed = time.time() - start
        return {
            "index": idx,
            "summary": response.choices[0].message.content,
            "elapsed": elapsed
        }
    except Exception as e:
        return {
            "index": idx,
            "summary": f"Error: {e}",
            "elapsed": time.time() - start
        }

async def main():
    # 模拟10条需要总结的文本 / Simulate 10 texts to summarize
    texts = [
        f"This is sample text number {i}. " * 5 
        for i in range(10)
    ]
    
    print(f"Starting {len(texts)} concurrent summarization requests...")
    start_total = time.time()
    
    # ============================================================
    # asyncio.gather() 并发执行所有协程
    # asyncio.gather() runs all coroutines concurrently
    # ============================================================
    # 所有请求同时发出，总耗时 ≈ 最慢单次请求的耗时
    # All requests are sent simultaneously; total time ≈ slowest single request
    results = await asyncio.gather(*[
        summarize_text(text, i) for i, text in enumerate(texts)
    ])
    
    total_elapsed = time.time() - start_total
    print(f"\nAll {len(results)} requests completed in {total_elapsed:.2f}s")
    
    # 打印结果 / Print results
    for r in results:
        print(f"  [{r['index']}] {r['summary'][:50]}... ({r['elapsed']:.2f}s)")

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

Streaming

流式输出

import asyncio
import os
from openai import AsyncOpenAI

client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

async def stream_response(prompt: str):
    """
    流式获取LLM响应 - 逐词/逐块输出
    Stream LLM response - output token by token / chunk by chunk
    """
    # ============================================================
    # stream=True 启用流式模式
    # stream=True enables streaming mode
    # ============================================================
    # 注意：stream=True时，需要使用 async for 迭代响应流
    # Note: when stream=True, use async for to iterate the response stream
    stream = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "user", "content": prompt}
        ],
        stream=True,  # 启用流式 / Enable streaming
        max_tokens=300,
    )
    
    print("Streaming response:")
    full_content = ""
    
    # ============================================================
    # async for 逐个处理流式chunk
    # async for processes each streaming chunk one by one
    # ============================================================
    # 每个chunk包含一小部分响应内容
    # Each chunk contains a small piece of the response
    async for chunk in stream:
        # delta 是当前chunk的新增内容
        # delta is the new content in this chunk
        delta = chunk.choices[0].delta
        
        if delta.content:
            content = delta.content
            print(content, end="", flush=True)  # 实时输出 / Real-time output
            full_content += content
    
    print("\n\n--- Full response received ---")
    return full_content

async def main():
    await stream_response("Explain quantum computing in simple terms.")

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

Error Handling & Retry

import asyncio
import os
import random
from openai import AsyncOpenAI, APIError, APIConnectionError, RateLimitError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

# ============================================================
# 使用 tenacity 库实现指数退避重试
# Use tenacity library for exponential backoff retry
# ============================================================
# 重试策略：
# - 最多重试3次 / Max 3 retry attempts
# - 等待时间：2秒 × 2^(attempt-1) + 随机抖动
# - Wait time: 2s × 2^(attempt-1) + random jitter
# - 仅对特定异常类型重试 / Only retry on specific exception types
@retry(
    stop=stop_after_attempt(3),  # 最多3次尝试 / Max 3 attempts
    wait=wait_exponential(multiplier=2, min=2, max=30),  # 指数退避 / Exponential backoff
    retry=retry_if_exception_type((APIError, APIConnectionError, RateLimitError)),
)
async def robust_chat_completion(prompt: str) -> str:
    """
    带自动重试的LLM调用
    LLM call with automatic retry
    """
    # 模拟偶尔的网络错误 / Simulate occasional network errors
    if random.random() < 0.3:  # 30%概率触发重试 / 30% chance to trigger retry
        raise APIConnectionError("Simulated network error")
    
    response = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=100,
    )
    return response.choices[0].message.content

async def main():
    try:
        # 即使有30%概率失败，重试机制会保证最终成功
        # Even with 30% failure rate, retry mechanism ensures eventual success
        result = await robust_chat_completion("Hello, world!")
        print(f"Result: {result}")
    except Exception as e:
        print(f"All retries exhausted: {e}")

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

Production Best Practices

生产环境最佳实践

import asyncio
import os
from contextlib import asynccontextmanager
from openai import AsyncOpenAI

# ============================================================
# 1. 客户端单例模式 (Singleton Client)
# ============================================================
# 整个应用共享一个AsyncOpenAI实例，复用连接池
# Share one AsyncOpenAI instance across the app to reuse connection pool
class LLMClient:
    _instance = None
    
    @classmethod
    def get_client(cls) -> AsyncOpenAI:
        if cls._instance is None:
            cls._instance = AsyncOpenAI(
                api_key=os.environ.get("OPENAI_API_KEY"),
                timeout=60.0,
                max_retries=2,
                # httpx客户端配置 / httpx client config
                http_client=None,  # 可自定义 / Can customize
            )
        return cls._instance

# ============================================================
# 2. 上下文管理器自动清理 (Context Manager for Cleanup)
# ============================================================
@asynccontextmanager
async def get_llm_client():
    """
    异步上下文管理器，确保客户端正确关闭
    Async context manager to ensure proper client cleanup
    """
    client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
    try:
        yield client
    finally:
        # 关闭底层httpx连接 / Close underlying httpx connections
        await client.close()

async def use_client():
    async with get_llm_client() as client:
        response = await client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": "Hello"}],
        )
        return response.choices[0].message.content

# ============================================================
# 3. 超时控制 (Timeout Control)
# ============================================================
async def call_with_timeout(prompt: str, timeout_seconds: float = 30.0):
    """
    带超时的LLM调用
    LLM call with timeout
    """
    try:
        # asyncio.timeout 会在超时时抛出 TimeoutError
        # asyncio.timeout raises TimeoutError on timeout
        async with asyncio.timeout(timeout_seconds):
            client = LLMClient.get_client()
            response = await client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
            )
            return response.choices[0].message.content
    except asyncio.TimeoutError:
        return f"Request timed out after {timeout_seconds}s"
    except Exception as e:
        return f"Error: {e}"

async def main():
    # 测试超时 / Test timeout
    result = await call_with_timeout("Tell me a very long story...", timeout_seconds=5.0)
    print(result)

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

Key Takeaways

要点	EN	CN
使用 AsyncOpenAI 替代 OpenAI	Use AsyncOpenAI instead of OpenAI	用 AsyncOpenAI 替换 OpenAI
所有API调用前加 await	Add await before every API call	每个API调用前加 await
asyncio.gather() 实现并发	Use asyncio.gather() for concurrency	用 asyncio.gather() 实现并发
流式输出用 async for	Use async for for streaming	流式输出用 async for
生产环境需处理超时和重试	Handle timeouts and retries in production	生产环境需处理超时和重试
整个应用共享一个客户端实例	Share one client instance across the app	整个应用共享一个客户端实例