Python asyncio 실전: 도구 호출의 전체 시간 예산과 취소 처리

에이전트가 여러 도구를 동시에 호출한다면, 요청 전체의 마감 시간을 먼저 정하고 그 안에 관련 작업을 묶는 편이 좋습니다. Python의 asyncio.timeout_at()과 TaskGroup으로 시작할 수 있습니다. 여기에 “없어도 되는 결과”와 “실패하면 전체를 중단할 결과”를 구분해야 작은 지연이 불필요한 전체 실패로 번지지 않습니다.

이 글은 새 릴리스 소식이 아닌 표준 라이브러리 실전 가이드입니다. 사용 API는 Python 3.11에서 추가됐으며, 예제에는 Python 3.11 이상이 필요합니다. 외부 패키지나 API 키는 사용하지 않습니다.

1. 도구마다 3초를 주기 전에 전체 예산을 정한다

문서 검색, 사용자 설정 조회, 결과 정리를 차례로 수행한다고 가정해 봅시다. 각 단계에 3초씩 부여하면 사용자 한 번의 요청이 기다리는 시간은 길어질 수 있습니다. 먼저 전체 예산을 정하고 준비 과정과 도구 실행을 같은 범위에 넣습니다. 재시도를 추가하더라도 매번 새로운 전체 예산을 주지 않는 것이 핵심입니다.

deadline = loop.time() + budget으로 만든 값을 timeout_at(deadline)에 전달합니다. 여기서 loop.time()은 이벤트 루프의 단조 시계입니다. 날짜·시각을 나타내는 time.time()의 값과 섞지 마세요. 이 숫자는 같은 프로세스의 실행 범위를 관리하는 데 쓰고, 다른 서비스에 전달할 때는 별도의 시간 예산 계약을 설계해야 합니다.

2. 필수 실패와 선택 도구의 시간 초과를 나눈다

TaskGroup은 등록된 작업을 종료 시 기다립니다. 취소 이외의 예외로 한 작업이 실패하면 남은 작업을 취소하고, 일반적인 자식 작업 오류는 ExceptionGroup으로 모아 전달합니다. 반면 기본 gather()는 첫 예외를 전달해도 다른 작업을 자동으로 취소하지 않습니다.

따라서 선택적인 부가 검색까지 모든 예외를 그대로 내보내면 필수 조회도 함께 취소될 수 있습니다. 아래 설계에서는 선택 도구에 배정한 시간이 끝난 경우만 {"status": "timeout"}으로 바꿉니다. 필수 도구의 오류와 예상하지 못한 오류는 숨기지 않습니다. 이는 정답 하나가 아니라, 부분 결과를 허용하는 서비스에 맞춘 정책입니다.

3. 세 가지 종료 경로를 실행해 본다

아래 코드를 deadline_demo.py로 저장하고 python deadline_demo.py로 실행합니다. 본문의 코드 그대로 CPython 3.12.14에서 실행했습니다. 네트워크 대신 asyncio.sleep()을 사용한 모의 실험이며, 실제 외부 API의 취소 동작이나 성능을 검증한 결과는 아닙니다.

import asyncio


async def tool(name, delay, cleaned, fail=False):
    try:
        await asyncio.sleep(delay)  # 외부 호출 대신 기다림만 모의한다.
        if fail:
            raise RuntimeError(f"{name} failed")
        return {"status": "ok", "value": name}
    finally:
        cleaned.add(name)  # 실제 코드에서는 자원 정리를 이 위치에 둔다.


async def optional_tool(cleaned):
    limit = asyncio.timeout(0.08)
    try:
        async with limit:
            return await tool("extra", 0.5, cleaned)
    except TimeoutError:
        if not limit.expired():
            raise  # 도구 자체의 TimeoutError까지 숨기지 않는다.
        return {"status": "timeout"}


async def run_tools(delay, cleaned, fail=False):
    deadline = asyncio.get_running_loop().time() + 0.3
    async with asyncio.timeout_at(deadline):
        await asyncio.sleep(0.05)  # 준비 시간도 전체 예산에 포함한다.
        async with asyncio.TaskGroup() as group:
            required = group.create_task(
                tool("required", delay, cleaned, fail)
            )
            extra = group.create_task(optional_tool(cleaned))
    return {"required": required.result(), "extra": extra.result()}


async def main():
    cases = [("partial", 0.04, False),
             ("deadline", 0.5, False),
             ("failure", 0.02, True)]
    for name, delay, fail in cases:
        cleaned = set()
        try:
            print(name, await run_tools(delay, cleaned, fail))
        except TimeoutError:
            print(name, "overall timeout")
        except ExceptionGroup as errors:
            print(name, "task errors:", len(errors.exceptions))
        print("cleaned:", sorted(cleaned))


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

실행 결과는 다음과 같습니다.

partial {'required': {'status': 'ok', 'value': 'required'}, 'extra': {'status': 'timeout'}}
cleaned: ['extra', 'required']
deadline overall timeout
cleaned: ['extra', 'required']
failure task errors: 1
cleaned: ['extra', 'required']
  • partial: 필수 도구는 성공하고, 선택 도구는 자신의 0.08초 제한을 결과 상태로 반환합니다.
  • deadline: 준비 시간을 포함한 0.3초 마감에 걸려 필수 작업을 취소합니다. 그룹 정리가 끝난 뒤 전체 시간 초과를 출력합니다.
  • failure: 필수 도구의 오류가 먼저 발생해 실행 중인 선택 도구를 취소합니다. 예제에서는 오류 묶음에 하나의 오류가 들어 있습니다.

cleaned는 각 모의 도구의 finally에 도달했음을 표시할 뿐입니다. 실제 연결이나 파일이 닫혔다는 증거는 아닙니다. 실서비스에서는 자원의 비동기 컨텍스트 관리자나 정리 코드를 사용하고, 정리 자체의 실패도 별도로 확인해야 합니다.

바깥 시간 제한이 TaskGroup 전체를 감싸는 순서가 중요합니다. 그래야 그룹을 빠져나오며 작업 완료를 기다리는 구간도 포함됩니다. 시간 제한 컨텍스트가 만드는 TimeoutError는 그 컨텍스트 바깥에서 처리합니다. 예제의 expired() 검사는 도구가 독자적으로 발생시킨 TimeoutError를 선택 도구의 예산 만료로 오인하지 않기 위한 장치입니다.

마지막 except ExceptionGroup은 세 경우를 이어서 보여주기 위한 데모 처리입니다. 운영 코드에서는 내부 예외와 원인을 기록하고, 예상한 유형만 처리하거나 상위로 다시 전달하세요. 이 예제처럼 오류 개수만 남기는 것은 장애 분석에 부족합니다.

4. 시간 제한이 보장하지 않는 것

  • 정확히 0.3초에 강제 종료되는 것은 아닙니다. 취소는 협력적으로 진행됩니다. 정리 작업이 오래 걸리거나 취소에 응하지 않으면 반환이 늦어질 수 있습니다. 예제의 숫자는 동작 설명용이며 지연 시간 보증이 아닙니다.
  • 이벤트 루프를 막으면 타이머도 제때 실행되지 못합니다. 동기 네트워크 호출이나 긴 계산을 비동기 함수 안에 그대로 넣지 마세요. 공식 asyncio 개발 안내도 블로킹 코드가 다른 작업을 지연시킨다고 설명합니다.
  • 스레드로 옮겼다고 취소까지 해결되지는 않습니다. to_thread()를 기다리는 코루틴의 취소와 이미 실행 중인 동기 함수의 종료는 별개입니다. 실행 중인 executor 작업은 Future.cancel()로 취소할 수 없습니다. 해당 클라이언트의 자체 제한 시간이나 협력적 중단 방법을 함께 확인해야 합니다.
  • 외부 효과를 되돌리지는 않습니다. 로컬 대기가 끝났다고 서버에 전달된 작업이 취소됐다고 가정하지 마세요. 생성·전송 같은 작업은 제공자의 상태 조회, 취소 API, 중복 실행 방지 규칙을 따로 설계해야 합니다.

특히 CancelledError를 잡아 성공 값으로 바꾸지 마세요. 공식 문서는 거의 모든 상황에서 다시 발생시켜야 한다고 안내합니다. 이 예외는 BaseException 계열입니다. 정리를 위해 명시적으로 잡았다면 일반적으로 정리 후 raise로 전달하고, 맨몸의 except:로 삼키는 코드도 점검하세요.

5. 도입은 작은 도구 묶음 한 곳부터

같은 응답에 속하는 소수의 비동기 도구를 운영하고, 전체 실패와 부분 성공의 기준이 있다면 이 패턴을 적용할 만합니다. 먼저 읽기 전용 도구 두 개로 시작해 다음을 확인하세요.

  1. 준비·대기·재시도 중 어느 구간까지 전체 예산에 넣을지 정합니다.
  2. 선택 도구의 예상된 시간 초과만 상태로 바꾸고, 인증 오류나 코드 결함은 드러나게 둡니다.
  3. 전체 만료, 필수 실패, 외부 취소를 각각 재현하고, 실행 중인 관련 작업과 자원이 남는지 확인합니다.
  4. 요청 ID와 도구별 결과 상태를 기록하고, 예산 만료와 실제 정리 완료 사이의 시간도 관찰합니다.

반대로 요청 종료 뒤에도 꼭 완료돼야 하는 장기 작업이나 취소를 지원하지 않는 동기 SDK가 중심이라면, 이 코드만으로 전환을 끝내지 마세요. 별도 작업 수명 관리가 먼저입니다. 작업 수가 많다면 동시 실행 제한도 추가해야 합니다. Semaphore로 진행 중인 호출 수를 제한할 수 있지만, 입력 전체에 태스크를 한꺼번에 만드는 방식의 메모리 비용까지 해결해 주지는 않습니다.

적용 기준은 간단합니다. 함께 끝나야 할 작업에는 공통 마감과 TaskGroup을 두고, 선택적 실패는 명시적인 결과로 표현하세요. 이 구분부터 갖추면 “시간은 끝났는데 무엇이 아직 실행 중인가”를 추적하기 쉬워집니다.

자료 확인: 2026년 10월 7일. AI의 도움을 받아 작성하고 Python 공식 문서와 대조했습니다. 예제는 CPython 3.12.14에서 실행한 모의 코드이며, 실제 외부 API는 시험하지 않았습니다.

댓글

댓글 쓰기