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라는 값이 숫자가 아닙니다.
그래서:
ValueErrorTypeError
애초에 타입이 맞지 않음.
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.00으로 나누기
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
ERRORLogger 생성
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 = 1if calls["n"] < 3:True
↓
ConnectionError↓
safe_call이 잡음
↓
WARNING↓
재시도
두 번째 호출
n = 2또 실패.
세 번째 호출
n = 3조건:
calls["n"] < 3False
따라서:
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를 만드는 데 매우 중요한 설계 원칙입니다.