Python async/await 기초

FastAPI·SSE·LangGraph가 모두 의존하는 비동기 모델

FastAPI는 async def, SSE는 async iterator, LangGraph는 astream_events를 사용한다. 세 도구 모두 같은 Python asyncio 모델 위에 있다. 본 글은 동기·비동기 차이의 직관, async/await 문법, 이벤트 루프, asyncio 기본 도구, 자주 만나는 오류 패턴을 정리한다. MINERVA 시리즈를 읽기 전에 이 모델을 먼저 익혀두면 나머지 글들이 훨씬 가볍게 읽힌다.

Engineering
저자

Kwangmin Kim

공개

2026년 05월 06일

1 왜 async/await를 알아야 하는가

MINERVA 시리즈의 거의 모든 백엔드 글이 async를 가정한다.

글 async 사용 위치
04 FastAPI 서빙 lifespan, async def stream
08-1 스트리밍·관측성 SSE event_generator가 yield 기반 비동기 제너레이터
13~16 LangGraph astream_events, astream
02-1 BaseAgent v2 astream_events 메서드

이 글들이 자주 사용하는 패턴(동기 함수와 async 함수의 차이, await이 무엇을 기다리는가, 이벤트 루프가 왜 필요한가)이 이해되어 있지 않으면 시리즈 본문이 어렵게 느껴진다. 본 글은 그 토대를 깐다.

2 동기 vs 비동기 — 한 식당에서 본 차이

정의: 동기와 비동기
  • 동기(synchronous): 한 작업이 끝날 때까지 다음 작업이 시작되지 않는다
  • 비동기(asynchronous): 한 작업이 외부 응답을 기다리는 동안 다른 작업이 진행된다

식당 비유로 정리하면 직관이 명확해진다.

모델 직원 1명이 어떻게 일하는가
동기 손님 A의 주문을 받음 → 주방에 전달 → 음식 나올 때까지 카운터에 서서 대기 → 손님 A에게 서빙 → 그제서야 손님 B 주문받음
비동기 손님 A 주문 → 주방에 전달 → “음식 나오면 알려달라” 말하고 손님 B 주문 받음 → 손님 C 주문 받음 → 주방에서 알림 오면 그때 A에게 서빙

직원 수(스레드 수)는 동일하지만 비동기 모델은 대기 시간을 다른 작업으로 채운다. 한 직원으로 더 많은 손님을 동시 처리할 수 있다.

3 I/O bound vs CPU bound — async가 빛나는 곳

async가 모든 상황에서 빠른 것은 아니다. 작업 종류에 따라 효과가 갈린다.

작업 유형 예시 async 효과
I/O bound LLM API 호출, DB 쿼리, 파일 읽기, 네트워크 요청 큼 — 대기 시간을 다른 작업으로 채움
CPU bound 행렬 곱, 이미지 처리, 압축, 암호화 거의 없음 — 한 코어가 계산 중이면 다른 작업도 못 진행

MINERVA의 핵심 워크로드인 LLM 호출은 전형적인 I/O bound다. Azure OpenAI에 요청을 보내고 응답을 1~10초 기다린다 — 이 대기 시간이 async로 다른 사용자 요청 처리에 활용된다.

CPU bound 작업(예: 임베딩 계산을 로컬에서 직접)은 async보다 멀티프로세싱·GPU가 답이다.

4 async/await 문법 기초

import asyncio


# 1. async def — 비동기 함수 정의
async def fetch_user(user_id: int) -> dict:
    print(f"[fetch] {user_id} 시작")
    await asyncio.sleep(1)              # I/O 대기 시뮬레이션 (네트워크·DB 등)
    print(f"[fetch] {user_id} 완료")
    return {"id": user_id, "name": f"user-{user_id}"}


# 2. await — 비동기 함수의 결과를 기다림
async def main():
    user = await fetch_user(42)         # 1초 대기 후 결과
    print(user)


# 3. asyncio.run — 이벤트 루프 시작
asyncio.run(main())

핵심 규칙

  • async def로 정의된 함수는 호출 즉시 실행되지 않고 코루틴 객체를 반환한다
  • 코루틴은 await 또는 asyncio.run/asyncio.gather 등으로 이벤트 루프에 제출되어야 실제 실행된다
  • await는 다른 async 함수 안에서만 사용 가능 (동기 함수에서 await 쓰면 SyntaxError)

5 await이 실제로 하는 일

async def main():
    print("A")
    await asyncio.sleep(1)
    print("B")

await asyncio.sleep(1)이 일어나는 순간:

  1. 현재 함수 실행을 잠시 중단한다
  2. 이벤트 루프에 “1초 후에 깨워달라”고 등록한다
  3. 이벤트 루프는 그 사이 다른 코루틴을 실행한다
  4. 1초가 지나면 이 함수를 깨워서 print("B")부터 이어 실행한다

await은 “일시정지 + 양보”의 표시다. 단순히 time.sleep(1)이라면 1초간 모든 다른 작업도 멈추지만, await asyncio.sleep(1)은 1초간 다른 작업이 진행될 자리를 양보한다.

6 동시 실행 — asyncio.gather

여러 async 작업을 동시에 시작하려면 gather를 쓴다.

async def fetch_all_users():
    # 1. 순차 실행 — 3초 소요
    u1 = await fetch_user(1)
    u2 = await fetch_user(2)
    u3 = await fetch_user(3)

    # 2. 동시 실행 — 약 1초 소요
    u1, u2, u3 = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )
    return [u1, u2, u3]

gather는 모든 코루틴을 이벤트 루프에 동시 등록하고, 모두 완료되면 결과를 한 번에 받는다. MINERVA에서 여러 사용자 요청이 동시에 들어와도 한 워커가 처리할 수 있는 이유가 이것이다 — 각 요청이 LLM 응답을 기다리는 동안 이벤트 루프가 다른 요청을 진행한다.

7 이벤트 루프 — 짧게

이벤트 루프는 단일 스레드 안에서 코루틴들의 큐를 관리하는 스케줄러다. await을 만나면 현재 코루틴을 큐에서 빼고 I/O 이벤트나 타이머에 등록한 뒤, 다른 준비된 코루틴을 실행한다. 외부 이벤트가 도착하면 해당 코루틴을 다시 큐에 넣어 재개한다.

이벤트 루프의 깊이 있는 작동 원리는 별도 글에서

본 글은 실전 패턴 중심이라 이벤트 루프는 짧게 다룬다. CPU vs I/O 작업의 본질, Task 전환 9단계, 코루틴이 “CPU+I/O가 섞인 얇은 껍질”이라는 통찰은 비동기 프로그래밍 (Asynchronous Programming)에서 깊이 있게 다룬다. 본 글의 실전 패턴(gather·async generator·FastAPI lifespan·pytest-asyncio)을 이해하기 어려우면 그 글을 먼저 읽기를 권장한다.

일반 Python 프로그램에서는 이벤트 루프를 직접 만들 일이 거의 없다. asyncio.run(main())이 알아서 루프를 만들고, 메인 코루틴이 끝나면 루프를 닫는다. FastAPI는 자체 이벤트 루프(uvicorn 통합)를 띄워주므로 라우터 함수를 async def로만 정의하면 된다.

이벤트 루프가 막히면 모든 게 멈춘다

await 없이 무거운 동기 작업(예: 큰 numpy 행렬 연산)을 async 함수 안에서 실행하면 그 시간 동안 이벤트 루프가 막혀 다른 코루틴이 모두 대기한다. 무거운 동기 작업은 asyncio.to_thread() 또는 loop.run_in_executor()로 별도 스레드에 보낸다:

result = await asyncio.to_thread(heavy_numpy_function, data)

8 async iterator와 async generator — SSE의 핵심

스트리밍 응답을 만들 때 async for + yield 패턴을 쓴다.

async def event_generator():
    for token in ["안", "녕", "하", "세", "요"]:
        await asyncio.sleep(0.1)        # LLM이 토큰 생성하는 시간 시뮬레이션
        yield f"data: {token}\n\n"


# 소비 측
async def consume():
    async for chunk in event_generator():
        print(chunk, end="")

yield가 들어간 async 함수는 async generator다. SSE 응답을 만드는 FastAPI 패턴이 정확히 이것이다 — MINERVA 08-1편 SSE 전송 형식이 같은 구조를 사용한다.

# FastAPI 라우터 — async generator를 StreamingResponse에 전달
@router.post("/stream")
async def stream(req: RunRequest):
    async def event_generator():
        for event in agent.stream(query):
            yield f"data: {event.model_dump_json()}\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream")

9 async 컨텍스트 매니저 — lifespan 패턴

리소스를 열고 닫는 패턴도 async 버전이 있다.

class AsyncResource:
    async def __aenter__(self):
        await self.setup()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.teardown()


async def main():
    async with AsyncResource() as resource:
        await resource.use()

FastAPI의 lifespan이 정확히 이 패턴이다.

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 시작 시점
    await warmup_qna()
    await warmup_ds()
    yield                                # ← 이 시점에 앱이 요청을 받기 시작
    # 종료 시점 (선택)
    await cleanup()


app = FastAPI(lifespan=lifespan)

yield 위는 startup, 아래는 shutdown. MINERVA의 lifespan이 이 패턴 위에서 ALBERT weight·Vector 인덱스 warmup을 처리한다 (04편 FastAPI 서빙 참조).

10 FastAPI에서의 async — 두 종류 라우터

FastAPI는 라우터 함수를 두 형태로 받는다.

@router.get("/sync")
def sync_endpoint():
    """동기 함수. 별도 스레드 풀에서 실행."""
    result = blocking_db_query()
    return result


@router.get("/async")
async def async_endpoint():
    """비동기 함수. 이벤트 루프에서 직접 실행."""
    result = await async_db_query()
    return result
라우터 형태 실행 위치 적합한 작업
def (동기) thread pool 동기 라이브러리(예: 일부 ORM, requests)를 쓰는 함수
async def event loop 비동기 라이브러리(httpx, aiohttp, asyncpg)를 쓰는 함수

중요: async def 안에서 requests.get() 같은 동기 호출을 하면 이벤트 루프가 막힌다. 같이 동작은 하지만 동시 처리 능력이 사라진다. httpx나 aiohttp 같은 async 라이브러리를 사용해야 진짜 효과가 난다.

11 자주 발생하는 오류 패턴

WRONG:

async def fetch():
    return {"id": 1}

result = fetch()                        # 코루틴 객체만 반환, 실행 안 됨
print(result)                           # <coroutine object fetch at 0x...>

CORRECT:

async def main():
    result = await fetch()              # await으로 실행
    print(result)

asyncio.run(main())

async def 함수를 그냥 호출하면 코루틴 객체만 만들고 실행하지 않는다. await이나 asyncio.run으로 실행해야 한다.

WRONG:

async def fetch():
    time.sleep(2)                        # 동기 sleep — 이벤트 루프 막힘
    return {"id": 1}

CORRECT:

async def fetch():
    await asyncio.sleep(2)               # 비동기 sleep — 다른 코루틴이 진행 가능
    return {"id": 1}

async 함수 안에서 동기 함수(time.sleep, requests.get, 동기 DB 드라이버)를 호출하면 그 시간 동안 모든 다른 코루틴이 멈춘다. 비동기 대안을 쓰거나 asyncio.to_thread로 격리한다.

WRONG:

def normal_function():
    result = await fetch()               # SyntaxError — 동기 함수에서 await

CORRECT:

def normal_function():
    return asyncio.run(fetch())          # 또는 다른 진입점

await은 async def 안에서만 사용 가능하다. 동기 함수에서 비동기 함수를 호출하려면 asyncio.run이 진입점이 된다 (단, 이미 이벤트 루프가 돌고 있는 환경에서는 asyncio.run이 에러를 내므로 nest_asyncio 같은 우회가 필요할 수 있다).

WRONG:

results = await asyncio.gather(task1(), task2(), task3())
# 한 작업이 예외 발생하면 gather가 즉시 실패하고 다른 결과를 잃음

CORRECT:

results = await asyncio.gather(
    task1(), task2(), task3(),
    return_exceptions=True,              # 예외도 결과로 받음
)
for r in results:
    if isinstance(r, Exception):
        log.warning("task failed: %s", r)
    else:
        process(r)

기본 동작은 첫 예외에서 즉시 실패다. 모든 작업의 결과를 받고 싶다면 return_exceptions=True를 명시한다.

12 정리

항목 핵심
언제 쓰는가 I/O bound (LLM·DB·네트워크) — CPU bound는 효과 없음
async def 호출 시 코루틴 반환, await 또는 이벤트 루프로 실행해야 함
await 일시정지 + 다른 코루틴에 양보
asyncio.gather 여러 코루틴 동시 실행 + 모든 결과 대기
async for + yield async generator (SSE 토큰 스트리밍)
async with async context manager (FastAPI lifespan)
함정 async 함수 안 동기 sleep·requests 호출 → 이벤트 루프 차단

13 응용 분야

분야 async 패턴
FastAPI 라우터 async def로 정의 + 비동기 라이브러리(httpx, asyncpg)
LLM 호출 LangChain·Anthropic·OpenAI SDK 모두 await client.chat.completions.create(...) 지원
SSE 스트리밍 async generator + yield
파일 I/O aiofiles (선택적)
WebSocket async for message in websocket:

14 관련 주제

선행 학습

  • 이 글이 가장 기초 — 별도 선행 없음

바로 이어 읽을 글

MINERVA 시리즈 응용

Subscribe

Enjoy this blog? Get notified of new posts by email: