Python에서의 비동기 프로그래밍

Python에서의 비동기 프로그래밍 (Async Programming in Python)

Python 비동기 프로그래밍이 처음이라면 이 페이지가 딱이에요. LlamaIndex의 많은 연산이 비동기 실행을 지원하는데, 그 핵심 개념을 이해해 두면 처리량과 성능을 크게 높일 수 있습니다.

출처: 문서

본문

Python 비동기 프로그래밍이 처음이라면 이 페이지가 도움이 될 것입니다.

특히 LlamaIndex에서는 많은 연산과 함수가 비동기 실행을 지원합니다. 이는 메인 스레드를 차단하지 않고 여러 연산을 동시에 실행할 수 있게 해주며, 많은 경우 전체 처리량과 성능을 개선하는 데 도움이 됩니다.

이해해야 할 핵심 개념은 다음과 같습니다.

1. asyncio의 기초

• 이벤트 루프 (Event Loop): 이벤트 루프는 비동기 연산의 스케줄링과 실행을 처리합니다. 태스크(코루틴)를 계속 검사하고 실행합니다. 모든 비동기 연산은 이 루프에 의해 실행되며, 스레드당 이벤트 루프는 하나만 존재할 수 있습니다.

• asyncio.run(): 이 함수는 비동기 프로그램을 실행하기 위한 진입점입니다. 이벤트 루프를 만들고 관리하며 완료 후 정리합니다. 스레드당 한 번만 호출되도록 설계되었다는 점을 기억하세요. FastAPI 같은 일부 프레임워크는 이벤트 루프를 대신 실행해 주지만, 다른 프레임워크에서는 직접 실행해야 합니다.

• 비동기 + Python 노트북: Python 노트북은 이벤트 루프가 이미 실행 중인 특수한 경우입니다. 즉, asyncio.run()을 직접 호출할 필요 없이 비동기 함수를 직접 호출하고 await할 수 있습니다.

2. 비동기 함수와 await

• 비동기 함수 정의: async def 구문으로 비동기 함수(코루틴)를 정의합니다. 비동기 함수를 호출하면 즉시 실행되는 대신, 스케줄링되어 실행되어야 하는 코루틴 객체를 반환합니다.

• await 사용: 비동기 함수 안에서 await는 대기 중인 태스크가 완료될 때까지 해당 함수의 실행을 일시 중지하는 데 사용됩니다. await some_fn()을 작성하면 함수가 이벤트 루프로 제어권을 넘겨 다른 태스크가 스케줄링되고 실행될 수 있게 합니다. 한 번에 하나의 비동기 함수만 실행되며, 이들은 await로 양보(yield)하며 협력합니다.

3. 동시성(Concurrency) 설명

• 협력적 동시성 (Cooperative Concurrency): 여러 비동기 태스크를 스케줄링할 수 있지만, 한 번에 하나의 태스크만 실행됩니다. 이는 여러 태스크가 동시에 실행되는 진정한 병렬성(parallelism)과 다릅니다. 태스크가 await를 만나면 실행을 일시 중지하여 다른 태스크가 실행될 수 있게 합니다. 이 때문에 비동기 프로그램은 LLM 및 기타 서비스에 대한 API 호출처럼 대기가 흔한 I/O 바운드 태스크에 탁월합니다.

• 진정한 병렬성은 아님: Asyncio는 동시성을 가능하게 하지만 태스크를 병렬로 실행하지는 않습니다. 병렬 실행이 필요한 CPU 바운드 작업에는 threading이나 multiprocessing을 고려하세요. LlamaIndex는 대부분의 경우 multiprocessing을 피하며, 안전하고 효율적으로 구현하기 복잡할 수 있으므로 사용자에게 맡깁니다.

4. 블로킹(동기) 코드 처리

• asyncio.to_thread(): 때로 비동기 프로그램을 멈추지 않고 동기(블로킹) 코드를 실행해야 할 수 있습니다. asyncio.to_thread()는 블로킹 코드를 별도 스레드로 오프로드하여 이벤트 루프가 다른 태스크를 계속 처리할 수 있게 합니다. 오버헤드가 있고 디버깅이 어려워질 수 있으므로 주의해서 사용하세요.

• 대안: Executors: loop.run_in_executor()를 사용해 블로킹 함수를 처리하는 방법도 볼 수 있습니다.

5. 실용적인 예제

다음은 asyncio로 비동기 함수를 작성하고 실행하는 방법을 보여주는 예제입니다.

import asyncio




async def fetch_data(delay):
    print(f"Started fetching data with {delay}s delay")


    # Simulates I/O-bound work, such as network operation
    await asyncio.sleep(delay)


    print("Finished fetching data")
    return f"Data after {delay}s"




async def main():
    print("Starting main")


    # Schedule two tasks concurrently
    task1 = asyncio.create_task(fetch_data(2))
    task2 = asyncio.create_task(fetch_data(3))


    # Wait until both tasks complete
    result1, result2 = await asyncio.gather(task1, task2)


    print(result1)
    print(result2)
    print("Main complete")




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

더 알아보기 (Learn more)

• 설치 (Installation)
• 스타터 예제