VIII. Agent Loop + Tool Dispatcher 다음 단계인 예외 처리(Exception), 재시도(Retry), Logging 스터디

 

    #-------- 1. KeyError — 딕셔너리에 없는 키를 찾았을 때

    # def get_units(record):

    #     return record["units"]

 

    # get_units({"sku":"51001"})

    # 결과: Exception has occurred: KeyError 'units'

 

    #-------- 2. try / except / else / finally

    def to_int(value):

        try:

            number = int(value)

        except ValueError:

            print(f" {value!r}을 숫자로 변환할 수 없습니다.")

            return None

        except TypeError:

            print(f" {value!r}은 완전히 잘못된 유형입니다.")

            return None

        else:

            print(f" 정상적으로 변환되었습니다.")

            return number

        finally:

            print(f" {value!r}에 대한 처리가 완료되었습니다.")

 

    for v in ["55", "eight", None]:

        print(f"- 입력 {v!r}:")

        print(" 결과:", to_int(v))

 

    #-------- 3. 사용자 정의 Exception

    class ToolError(Exception):

        pass

 

    def divide(a, b):

        if b==0:

            raise ToolError("분모 0으로 나눗셈을 할 수 없습니다.")

        return a/b

 

    for pair in [(100, 25), (50, 0)]:

        try:

            print(f"{pair[0]}/{pair[1]} = {divide(*pair)}")

        except ToolError as e:

            print(f"툴사용 거부: {e}")

 

    #-------- 4. Retry — 실패하면 다시 시도하기

    import time

 

    attempt_log = []

 

    def flaky_call(attempt):

        attempt_log.append(attempt)

        if attempt < 3:

            raise ConnectionError("일시적 사용불가")

        return "success on attempt 3"

 

    def with_try(func, tries=4, base_delay=1.5):

        for attempt in range(1, tries+1):

            try:

                return func(attempt)

            except ConnectionError as e:

                if attempt == tries:

                    raise

                delay = base_delay*(2**(attempt -1))

                print(f"attempt {attempt} failed ({e}), waiting {delay}")

                time.sleep(delay)

 

    print(with_try(flaky_call))

    print("attempt made: ", attempt_log)

 

    #-------- 5. Logging

    import logging

 

    logging.basicConfig(

        level=logging.INFO,

        format="%(asctime)s %(levelname)-8s %(message)s",

        datefmt="%H%M%S",

    )

 

    log = logging.getLogger("agent")

 

    log.debug("이 로그는 INFO 레벨에서 숨겨져 있습니다")

    log.info("실행 시작, max_steps = 5")

    log.warning("도구 'search' 실행에 4.2초 소요")

    log.error("도구 'read_file' 실패: 파일을 찾을 수 없음")

 

    #-------- 6. 가장 실전적인 코드

    import logging

    import time

 

    logging.basicConfig(level=logging.INFO, format="%(levelname)-7s %(message)s")

    log = logging.getLogger("agent")

 

    RETRYABLE = (ConnectionError, TimeoutError)

 

    def safe_call(func, *args, tries=3, base_delay=0.5, **kwargs):

        for attempt in range(1, tries+1):

            try:

                value = func(*args, **kwargs)

                return {"ok":True, "value":value, "error":None, "attempts":attempt}

            except RETRYABLE as e:

                log.warning(f"attempt {attempt}/{tries} failed: {e}")

                if attempt == tries:

                    return {"ok":False, "value":None, "error":str(e), "attempts":attempt}

            except Exception as e:

                log.error(f"permanent failure in {func.__name__}: {type(e).__name__}: {e}")

                return {"ok":False, "value":None, "error":str(e), "attempts":attempt}

 

    calls = {"n":0}

 

    def unstable_api(query):

        calls["n"] += 1

        if calls["n"] < 3:

            raise ConnectionError("속도가 제한됩니다.")

        return f"results for {query}"

 

    def broken_tool(x):

        return x["missing"]

 

    print(safe_call(unstable_api, "재고 상황"))

    print(safe_call(broken_tool, {"현재": 1}))

 

실제 AI Agent를 만들 때 매우 중요합니다. 왜냐하면 현실의 Tool은 항상 성공하지 않기 때문입니다.

전체 흐름은 이렇게 보시면 됩니다.

Agent
  │
  ├─ Tool 실행
  │     │
  │     ├─ 성공 → 결과 반환
  │     │
  │     └─ 실패
  │           │
  │           ├─ 일시적 오류 → Retry
  │           │
  │           └─ 영구적 오류 → 실패 처리
  │
  └─ Logging으로 모든 상황 기록

 

하나씩 보겠습니다.

---

1. KeyError — 딕셔너리에 없는 키를 찾았을 때

주석 처리된 첫 번째 코드입니다.

 

def get_units(record):
    return record["units"]

get_units({"sku":"51001"})

 

함수에 전달된 데이터는:

 

{
    "sku": "51001"
}

 

입니다.

그런데 코드에서는:

 

record["units"]

 

를 찾습니다.

하지만 딕셔너리 안에는:

sku

 

만 있고:

units

 

가 없습니다.

그래서:

KeyError: 'units'

 

가 발생합니다.

---

왜 KeyError일까?

 

record = {
    "sku": "51001"
}

 

여기서:

 

record["sku"]

 

→ 정상

하지만:

 

record["units"]

 

→ 없음

그래서 Python이:

"units라는 Key를 찾을 수 없습니다."

라고 오류를 발생시키는 것입니다.

---

Agent에서는 이런 상황이 자주 발생합니다

예를 들어 Tool 결과가:

 

{
    "sku": "51001"
}

 

인데 Agent가:

 

result["units"]

 

라고 하면 프로그램이 멈출 수 있습니다.

조금 더 안전하게 하려면:

 

record.get("units")

 

를 사용할 수도 있습니다.

그러면 키가 없을 때:

 

None

 

을 반환합니다.

---

2. try / except / else / finally

이번 코드에서 아주 중요한 부분입니다.

 

def to_int(value):
    try:
        number = int(value)

    except ValueError:
        print(f" {value!r}을 숫자로 변환할 수 없습니다.")
        return None

    except TypeError:
        print(f" {value!r}은 완전히 잘못된 유형입니다.")
        return None

    else:
        print(f" 정상적으로 변환되었습니다.")
        return number

    finally:
        print(f" {value!r}에 대한 처리가 완료되었습니다.")

 

구조부터 이해하면 쉽습니다.

try
 │
 ├─ 성공 → else
 │
 └─ 실패 → except
              │
              └─ 마지막에는 finally

 

중요한 것은:

finally는 성공하든 실패하든 항상 실행

 

입니다.

---

① "55"

 

to_int("55")

 

먼저:

 

number = int("55")

 

성공합니다.

number = 55

 

그러면 except는 건너뜁니다.

그리고:

 

else:
    print("정상적으로 변환되었습니다.")
    return number

 

실행됩니다.

하지만 return을 했더라도:

 

finally

 

는 실행됩니다.

즉 흐름:

try 성공
   ↓
else 실행
   ↓
return 준비
   ↓
finally 실행
   ↓
55 반환

 

출력은 대략:

정상적으로 변환되었습니다.
'55'에 대한 처리가 완료되었습니다.
결과: 55

 

---

② "eight"

 

to_int("eight")

 

실행:

 

int("eight")

 

숫자로 변환할 수 없습니다.

그래서:

ValueError

 

가 발생합니다.

이 부분이 실행됩니다.

 

except ValueError:
    print(f" {value!r}을 숫자로 변환할 수 없습니다.")
    return None

 

그 후에도:

 

finally

 

는 실행됩니다.

흐름:

try
 ↓
ValueError
 ↓
except ValueError
 ↓
finally
 ↓
None 반환

 

---

③ None

 

to_int(None)

 

실행:

 

int(None)

 

이건 단순히 숫자로 바꿀 수 없는 문자열 문제가 아닙니다.

아예 데이터 타입이 맞지 않습니다.

그래서:

TypeError

 

가 발생합니다.

따라서:

 

except TypeError:

 

가 실행됩니다.

---

ValueError와 TypeError 차이

이 부분을 꼭 기억하세요.

ValueError

타입은 맞는데 값이 이상함.

 

int("eight")

 

문자열(String)은 받을 수 있지만:

eight

 

라는 값이 숫자가 아닙니다.

그래서:

ValueError

 

---

TypeError

애초에 타입이 맞지 않음.

 

int(None)

 

None은 정수 변환 대상으로 적절하지 않습니다.

그래서:

TypeError

 

---

3. 사용자 정의 Exception

 

class ToolError(Exception):
    pass

 

여기서 새로운 오류 종류를 만들었습니다.

Python 기본 오류:

ValueError
TypeError
KeyError
ConnectionError

 

우리가 만든 오류:

ToolError

 

입니다.

---

왜 직접 Error를 만들까?

예를 들어 Tool에서:

API 오류
파일 없음
계산 불가
권한 없음

 

등 다양한 문제가 있을 수 있습니다.

그중 우리가 특별히 처리하고 싶은 오류를 만들 수 있습니다.

---

divide 함수

 

def divide(a, b):
    if b == 0:
        raise ToolError("분모 0으로 나눗셈을 할 수 없습니다.")

    return a / b

 

여기서:

 

raise

 

는:

"일부러 오류를 발생시켜라"

라는 의미입니다.

---

정상 실행

 

divide(100, 25)

 

↓

4.0

 

---

0으로 나누기

 

divide(50, 0)

 

코드:

 

if b == 0:
    raise ToolError(...)

 

↓

ToolError 발생

 

---

예외 잡기

 

try:
    print(...)
except ToolError as e:
    print(f"툴사용 거부: {e}")

 

따라서:

100/25 = 4.0
툴사용 거부: 분모 0으로 나눗셈을 할 수 없습니다.

 

가 됩니다.

---

Agent에서의 의미

예를 들어:

 

raise ToolError("검색 API 사용량 제한 초과")

 

또는:

 

raise ToolError("파일 형식이 지원되지 않습니다")

 

처럼 사용할 수 있습니다.

Agent는 이 오류를 받아서:

Tool 실패
↓
LLM에게 실패 결과 전달
↓
다른 방법 선택

 

할 수 있습니다.

---

4. Retry — 실패하면 다시 시도하기

이 부분은 실제 API를 사용할 때 매우 중요합니다.

 

import time

attempt_log = []

 

재시도 횟수를 기록합니다.

---

일부러 실패하는 함수

 

def flaky_call(attempt):
    attempt_log.append(attempt)

    if attempt < 3:
        raise ConnectionError("일시적 사용불가")

    return "success on attempt 3"

 

의미:

1번째 → 실패
2번째 → 실패
3번째 → 성공

 

입니다.

---

재시도 함수

 

def with_try(func, tries=4, base_delay=1.5):

 

func       → 실행할 함수
tries      → 최대 시도 횟수
base_delay → 기본 대기 시간

 

---

반복문

 

for attempt in range(1, tries+1):

 

tries=4라면:

attempt = 1
attempt = 2
attempt = 3
attempt = 4

 

입니다.

---

함수 실행

 

try:
    return func(attempt)

 

성공하면 즉시 결과를 반환합니다.

예:

1차 실패
2차 실패
3차 성공

 

3차에서:

 

return "success on attempt 3"

 

하고 함수가 종료됩니다.

---

실패하면?

 

except ConnectionError as e:

 

ConnectionError가 발생하면 재시도합니다.

---

마지막 시도인가?

 

if attempt == tries:
    raise

 

예를 들어 최대 4번인데:

1 → 실패
2 → 실패
3 → 실패
4 → 실패

 

했다면 더 이상 재시도하지 않고:

 

raise

 

로 최종 오류를 발생시킵니다.

---

Exponential Backoff

 

delay = base_delay * (2 ** (attempt - 1))

 

이 부분이 중요합니다.

base_delay = 1.5라면:

시도계산대기

1차 실패	1.5 × 2⁰	1.5초
2차 실패	1.5 × 2¹	3초
3차 실패	1.5 × 2²	6초

이 방식을:

Exponential Backoff (지수 백오프)

라고 합니다.

API 서버에 문제가 있을 때 계속 빠르게:

재시도!
재시도!
재시도!
재시도!

 

하는 대신:

1초 기다림
↓
2초 기다림
↓
4초 기다림
↓
8초 기다림

 

처럼 점점 기다리는 방식입니다.

실제 OpenAI API나 네트워크 API에서도 흔히 사용합니다.

---

작은 오타 하나

현재 코드:

 

print(f"attempt [attempt] failed ({e}), waiting {delay}")

 

여기서:

[attempt]

 

는 그냥 문자 그대로 출력됩니다.

아마 의도는:

 

print(f"attempt {attempt} failed ({e}), waiting {delay}")

 

일 것입니다.

그러면:

attempt 1 failed (일시적 사용불가), waiting 1.5
attempt 2 failed (일시적 사용불가), waiting 3.0
success on attempt 3

 

처럼 출력됩니다.

---

5. Logging

 

import logging

 

지금까지 우리는:

 

print()

 

를 많이 사용했습니다.

하지만 실제 프로그램에서는:

언제 실행되었는지
무슨 일이 발생했는지
어떤 Tool이 실패했는지
몇 초가 걸렸는지

 

기록할 필요가 있습니다.

그래서:

 

logging

 

을 사용합니다.

---

기본 설정

 

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)-8s %(message)s",
    datefmt="%H%M%S",
)

 

하나씩 보겠습니다.

---

level=logging.INFO

로그 수준입니다.

일반적인 레벨:

DEBUG
INFO
WARNING
ERROR
CRITICAL

 

순서로 심각도가 높아집니다.

현재 INFO로 설정했으므로:

DEBUG   → 출력 안 함
INFO    → 출력
WARNING → 출력
ERROR   → 출력

 

입니다.

---

format

 

"%(asctime)s %(levelname)-8s %(message)s"

 

출력 형식을 지정합니다.

예:

143022 INFO     실행 시작
143023 WARNING  검색 속도가 느림
143024 ERROR    파일 없음

 

---

%(asctime)s

시간

143022

 

---

%(levelname)-8s

로그 수준을 8칸으로 정렬합니다.

INFO
WARNING
ERROR

 

---

Logger 생성

 

log = logging.getLogger("agent")

 

agent라는 이름의 Logger를 만듭니다.

---

DEBUG

 

log.debug("이 로그는 INFO 레벨에서 숨겨져 있습니다")

 

현재 레벨이:

 

INFO

 

이므로 DEBUG는 출력되지 않습니다.

---

INFO

 

log.info("실행 시작, max_steps = 5")

 

출력됩니다.

---

WARNING

 

log.warning("도구 'search' 실행에 4.2초 소요")

 

문제는 아니지만 주의가 필요한 상황입니다.

---

ERROR

 

log.error("도구 'read_file' 실패: 파일을 찾을 수 없음")

 

실패 상황입니다.

---

6. 가장 실전적인 코드: safe_call

이제 앞에서 배운 것들이 모두 합쳐집니다.

 

def safe_call(func, *args, tries=3, base_delay=0.5, **kwargs):

 

이 함수의 목적:

어떤 Tool/API 함수든 안전하게 실행한다.

구조:

safe_call()
   │
   ├─ 실행
   │
   ├─ 성공 → OK
   │
   ├─ ConnectionError → 재시도
   │
   ├─ TimeoutError → 재시도
   │
   └─ 기타 오류 → 즉시 실패

 

---

재시도 가능한 오류

 

RETRYABLE = (
    ConnectionError,
    TimeoutError
)

 

즉:

ConnectionError
TimeoutError

 

가 발생하면 재시도합니다.

왜냐하면 이런 오류는 보통:

인터넷 일시 장애
서버 과부하
일시적인 API 오류

 

일 가능성이 있기 때문입니다.

---

정상 실행

 

value = func(*args, **kwargs)

 

예:

 

safe_call(
    unstable_api,
    "재고 상황"
)

 

내부적으로:

 

unstable_api("재고 상황")

 

를 실행합니다.

---

성공 결과

 

return {
    "ok":True,
    "value":value,
    "error":None,
    "attempts":attempt
}

 

예:

 

{
    "ok": True,
    "value": "results for 재고 상황",
    "error": None,
    "attempts": 3
}

 

---

일시적인 오류

 

except RETRYABLE as e:

 

이 부분은:

 

except (ConnectionError, TimeoutError) as e:

 

와 같은 의미입니다.

---

로그 기록

 

log.warning(
    f"attempt {attempt}/{tries} failed: {e}"
)

 

예:

WARNING attempt 1/3 failed: 속도가 제한됩니다.

 

---

마지막 시도까지 실패

 

if attempt == tries:
    return {
        "ok":False,
        "value":None,
        "error":str(e),
        "attempts":attempt
    }

 

Agent 프로그램 전체를 죽이지 않고 결과를 반환합니다.

이게 아주 중요합니다.

---

영구적인 오류

 

except Exception as e:

 

예:

KeyError
ValueError
TypeError
AttributeError

 

같은 예상하지 못한 오류입니다.

이런 경우 재시도해도 해결되지 않을 가능성이 높습니다.

예:

 

def broken_tool(x):
    return x["missing"]

 

입력:

 

{"현재": 1}

 

여기에는:

missing

 

키가 없습니다.

따라서:

KeyError

 

가 발생합니다.

이걸 3번 반복한다고 해결될까요?

1번 실행 → KeyError
2번 실행 → KeyError
3번 실행 → KeyError

 

계속 똑같습니다.

그래서 재시도하지 않고 즉시:

 

log.error(...)

 

후 실패를 반환합니다.

---

unstable_api 실행 흐름

 

calls = {"n":0}

 

호출 횟수 저장.

---

첫 번째 호출

n = 1

 

 

if calls["n"] < 3:

 

True

↓

ConnectionError

 

↓

safe_call이 잡음

↓

WARNING

 

↓

재시도

---

두 번째 호출

n = 2

 

또 실패.

---

세 번째 호출

n = 3

 

조건:

 

calls["n"] < 3

 

False

따라서:

 

return f"results for {query}"

 

성공.

최종:

 

{
    "ok": True,
    "value": "results for 재고 상황",
    "error": None,
    "attempts": 3
}

 

---

broken_tool 실행 흐름

 

def broken_tool(x):
    return x["missing"]

 

호출:

 

safe_call(
    broken_tool,
    {"현재": 1}
)

 

실행:

 

x["missing"]

 

하지만:

 

x = {
    "현재": 1
}

 

missing 없음.

↓

KeyError

 

↓

 

except Exception

 

↓

로그:

ERROR permanent failure in broken_tool: KeyError: 'missing'

 

↓

결과:

 

{
    "ok": False,
    "value": None,
    "error": "'missing'",
    "attempts": 1
}

 

---

⭐ 이번 코드에서 가장 중요한 개념

이제 Agent가 상당히 실전적인 구조로 발전하고 있습니다.

지금까지:

LLM
 ↓
Tool 선택
 ↓
Dispatcher
 ↓
Tool 실행

 

이었다면 이제는:

LLM
 │
 ▼
Tool Call
 │
 ▼
Dispatcher
 │
 ▼
safe_call()
 │
 ├── 성공 ───────────→ 결과 반환
 │
 ├── ConnectionError
 │       │
 │       ├─ Retry 1
 │       ├─ Retry 2
 │       └─ Retry 3
 │
 └── Permanent Error
         │
         ▼
      실패 반환

 

그리고 모든 과정은:

logging

 

으로 기록됩니다.

---

제가 보기에는 이번 코드에서 꼭 기억해야 할 것은 6가지입니다

개념의미

try	위험할 수 있는 코드 실행
except	오류 발생 시 처리
else	오류가 없었을 때 실행
finally	성공/실패 관계없이 항상 실행
raise	직접 오류 발생
Retry	일시적 오류일 때 재시도

그리고 Agent 관점에서 가장 중요한 원칙은 이것입니다.

모든 오류를 재시도하면 안 됩니다.

ConnectionError
TimeoutError

 

같은 것은 재시도할 가치가 있습니다.

하지만:

KeyError
잘못된 arguments
없는 Tool 이름
잘못된 JSON

 

등은 재시도해도 같은 오류가 반복될 가능성이 높습니다.

그래서 일시적 오류와 영구적 오류를 구분하는 것이 실제 Agent를 만드는 데 매우 중요한 설계 원칙입니다.