day4_practice/failure_agent.py + failure_injection.py

도대체 뭐하는 실습이냐면

외부 API가 죽었을 때 서비스가 500을 내지 않고 백업 데이터로 응답하는지 보는 장애 대응 테스트입니다. 백엔드에서 늘 하던 retry, backoff, fallback 테스트와 같은데, 그 서비스가 LLM 에이전트라는 점만 다릅니다.

한 문단 요약

자산 상태를 조회하는 에이전트가 있습니다. 조회 API를 일부러 고장 내고(FAILURE_MODE), 에이전트가 ① 죽지 않고 ② 백업 도구로 갈아타서 ③ 답을 내는지, 그리고 ④ 그 와중에 로그에 API 키가 새지 않는지를 검사합니다. 4개 다 통과하면 "DoD 충족", 즉 운영에 내보내도 된다는 판정입니다.

1. 평소 백엔드 코드라면 이렇게 짰을 겁니다

일반 백엔드: 코드가 폴백까지 처리 
def get_asset_status(asset_id):
    for attempt in range(3):      # 재시도
        try:
            data = backend_api(asset_id)
            validate(data)         # 스키마 검증
            return data
        except RateLimitError:
            sleep(2 ** attempt)  # 백오프
        except TimeoutError:
            continue
    return backup_cache(asset_id)  # 폴백
이 실습: 폴백은 LLM이 결정 
def get_asset_status(asset_id):
    for attempt in range(3):      # 재시도 (같음)
        try:
            ...                    # 검증 (같음)
        except RateLimitError:
            sleep(2 ** attempt)  # 백오프 (같음)
        ...
    # 폴백을 직접 부르지 않고 "말로" 시킨다
    return "에러: ... get_asset_status_backup
            도구를 사용하세요."

차이는 마지막 줄 하나입니다. 도구는 백업을 직접 부르지 않고 "백업 도구를 쓰라"는 문장을 돌려줍니다. LLM이 그 문장을 도구 결과로 읽고, 다음 턴에 백업 도구를 스스로 호출합니다. 이 연결이 실제로 되는지 확인하는 것이 이 실습의 핵심입니다.

왜 raise가 아니라 return인가? 도구에서 예외가 터지면 에이전트 그래프가 그대로 죽습니다(처리 안 된 예외로 500이 나는 것과 같음). 문자열로 돌려주면 LLM에게는 정상적인 도구 결과라서 대화가 이어집니다.

2. python failure_agent.py를 치면 일어나는 일

위에서 아래로 그대로 실행됩니다. 오른쪽 코드는 실제 파일의 해당 부분입니다.

1

import 시점: 로그 설정과 에이전트 조립

import failure_injection 순간 agent.log 파일 로깅이 켜지고, 장애 스위치 FAILURE_MODE와 더미 API 키가 준비됩니다. 이어서 LLM에 도구 두 개(실시간, 백업)를 붙여 에이전트를 만듭니다.

agent = create_react_agent(llm,
    [get_asset_status, get_asset_status_backup],
    prompt="...get_asset_status를 먼저 사용하고 도구 에러 메시지에
            안내된 지침을 그대로 따르세요.")
2

정상 실행: 장애 없이 한 번

스위치를 "none"으로 두고 "A-1001 자산 상태 알려줘"를 물어봅니다. 실시간 도구가 바로 성공하는지 확인합니다. (대조군)

failure_injection.FAILURE_MODE = "none"
normal = agent.invoke({"messages": [("user", question)]})
3

장애 실행: 스위치를 켜고 같은 질문

환경 변수 FAILURE_MODE를 읽어(없으면 timeout) 장애를 켜고 같은 질문을 다시 합니다. 이번에는 실시간 도구가 실패하므로 백업으로 갈아타야 합니다. (실험군)

mode = os.environ.get("FAILURE_MODE", "timeout")
failure_injection.FAILURE_MODE = mode
broken = agent.invoke(...)
4

trace 출력: 메시지 이력 찍기

장애 실행의 messages를 순서대로 찍어 "어떤 도구를 불렀고 무슨 결과가 왔는지"를 눈으로 봅니다. 실패 지점을 찾는 디버깅 용도입니다.

5

로그 보안 점검

agent.log 파일을 열어 API 키 원문(sk-internal-dummy-1234, AKIA)이 있는지 문자열 검색합니다.

6

DoD 판정: assert 4개

정상 호출 처리     # 2번에서 실시간 도구로 정상 답변했나
실패 호출 처리     # 3번에서 크래시 없이 답변이 나왔나
필수 장애 1개 복구 # 3번에서 백업 도구가 호출됐나
로그 Secret 점검   # 5번에서 키 원문이 없나
→ 4개 다 통과하면 "DoD 충족"

3. 3번(장애 실행) 안에서 벌어지는 일

agent.invoke() 한 번 안에서 누가 누구를 부르는지 시퀀스로 그렸습니다. timeout 모드 기준입니다.

실행 스크립트 __main__ agent (LLM) create_react_agent 실시간 도구 get_asset_status 가짜 백엔드 _call_backend 백업 도구 ..._backup invoke("A-1001 상태?") ① 실시간 도구 호출 시도 1 Timeout (5초) 시도 2 Timeout (5초) 시도 3 Timeout (5초) 도구 함수 안의 for 루프 LLM은 이 15초를 모르고 결과만 기다림 ② "에러: … backup 도구를 사용하세요" ③ 문구를 읽고 LLM이 스스로 백업 도구 호출 ④ "A-1001: 사용중 (백업 스냅샷 기준)" ⑤ 최종 답변

노란 영역은 1단 방어(도구 코드의 재시도)이고, ②→③이 2단 방어(LLM의 경로 전환)입니다. ③은 코드의 if문이 아니라 LLM의 판단이라서, 프롬프트의 "에러 메시지 지침을 따르세요"가 이 연결을 걸어 줍니다.

4. failure_injection.py: 고장 나는 도구

강의 제공 코드. "테스트 대상 + mock"을 한 파일에 담았습니다. 
logging.basicConfig(level=logging.INFO, ...,
    handlers=[logging.FileHandler("agent.log"),
              logging.StreamHandler()])

로그 설정

로그를 화면과 agent.log 파일에 동시에 남깁니다. 5번 보안 점검이 이 파일을 검사합니다.

FAILURE_MODE = os.environ.get("FAILURE_MODE", "none")
API_KEY = os.environ.get("INTERNAL_API_KEY",
                         "sk-internal-dummy-1234")

장애 스위치와 더미 비밀값

FAILURE_MODE는 테스트용 mock 스위치입니다. API_KEY는 "로그에 새면 안 되는 값"의 예시입니다.

def _call_backend(asset_id):
    logger.info("... api_key=%s***", API_KEY[:6])
    if FAILURE_MODE == "timeout":
        time.sleep(5); raise TimeoutError(...)
    if FAILURE_MODE == "rate_limit":
        raise RateLimitError("429 ...")
    if FAILURE_MODE == "bad_schema":
        return {"assetStatus": "사용중인듯"}
    return {"asset_id": ..., "status": "사용중", "owner": "김하늘"}

가짜 백엔드 API (mock)

진짜 자산 관리 서버 대신입니다. 스위치 값에 따라 일부러 느려지거나, 429를 내거나, 필드명이 다른 응답을 줍니다.

첫 줄 로그가 키를 앞 6자만 찍는 것이 마스킹입니다. 그래서 로그에는 sk-int***만 남습니다.

@tool
def get_asset_status(asset_id):
    for attempt in range(1, 4):
        try:
            data = _call_backend(asset_id)
            if "status" not in data ...:
                return "에러: 형식 오류. backup 쓰세요"
            return f"{id}: {status} (사용자: {owner})"
        except RateLimitError:
            if 마지막: return "에러: 한도 초과. backup 쓰세요"
            time.sleep(2 ** attempt)
        except TimeoutError:
            if 마지막: return "에러: 지연. backup 쓰세요"
        except Exception:
            return "에러: 조회 실패. backup 쓰세요"

실시간 조회 도구 (1단 방어)

LLM이 부르는 도구입니다. 장애마다 대응이 다릅니다.

timeout: 그냥 다시 시도합니다.
429: 2초, 4초 기다렸다 시도합니다(지수 백오프).
형식 오류: 다시 불러도 같으니 재시도 없이 바로 포기합니다.

어떤 경우든 마지막에는 raise 대신 다음 행동을 적은 문자열을 return합니다. 이게 LLM에게 넘기는 바통입니다.

@tool
def get_asset_status_backup(asset_id):
    """백업 자산 상태 조회. get_asset_status
    실패 시 사용한다. 1시간 전 스냅샷..."""
    return f"{asset_id}: 사용중 (백업 스냅샷 기준...)"

백업 도구 (폴백)

캐시나 읽기 전용 복제본 역할입니다. 데이터는 1시간 늦지만 항상 성공합니다.

docstring의 "실패 시 사용한다"도 LLM이 읽는 설명이라, 언제 쓸지 판단하는 근거가 됩니다.

5. failure_agent.py: 시험 감독관

실습 코드. 에이전트를 만들고, 돌리고, 채점합니다. 
llm = ChatBedrockConverse(model="...sonnet-4-5...")
agent = create_react_agent(llm,
    [get_asset_status, get_asset_status_backup],
    prompt="... 도구 에러 메시지에 안내된
            지침을 그대로 따르세요.")

에이전트 조립

3일차에 직접 만든 agent ⇄ tools 그래프를 create_react_agent 한 줄로 만든 것입니다. 구조는 같습니다.

프롬프트의 "에러 지침을 따르라"가 2단 방어의 스위치입니다. 이 문장이 없으면 LLM이 에러를 보고 그냥 "조회 실패"라고 답할 수 있습니다.

def get_text(message): ...
def print_trace(result): ...

출력 헬퍼

get_text는 Bedrock 응답에서 텍스트만 뽑습니다. print_trace는 메시지 이력을 "사용자 / 도구요청 / 도구결과 / 최종답변" 한 줄씩으로 찍는 디버깅용 함수입니다(trace.py와 같음).

def tool_names(result):
    return [tc["name"] for m in result["messages"]
            if m.tool_calls for tc in m.tool_calls]

채점용: 어떤 도구를 불렀나

실행 이력에서 LLM이 요청한 도구 이름만 뽑습니다. 결과에 get_asset_status_backup이 있으면 "복구했다"로 판정합니다.

def check_secret_log(path="agent.log"):
    text = open(path).read()
    hits = [p for p in ("sk-internal-dummy-1234", "AKIA")
            if p in text]
    return (not hits), ...

채점용: 로그에 비밀값이 샜나

로그 파일에서 키 원문을 문자열 검색합니다. AKIA는 AWS 액세스 키가 시작하는 접두어입니다.

if __name__ == "__main__":
    # 정상 실행 → 장애 실행 → trace
    # → 로그 점검 → DoD 4개 판정

실행부

2장의 1~6단계가 여기 그대로 들어 있습니다.

6. 주입할 수 있는 장애 3종

나누는 기준은 하나, "다시 부르면 나아질 장애인가?"입니다.

FAILURE_MODEHTTP로 치면도구 안 대응도구가 걸리는 시간
timeout504 Gateway Timeout일시적일 수 있으니 3번 재시도약 15초 (5초 × 3)
rate_limit429 Too Many Requests기다리면 풀리니 2초, 4초 쉬고 재시도약 6초
bad_schema200인데 필드명이 바뀜다시 불러도 같으니 재시도 없이 바로 포기즉시
none200 정상-즉시

세 경우 모두 결국 "backup 도구를 쓰세요" 문자열로 끝나고, 그다음은 LLM 몫입니다.

7. 실제로 돌린 결과 (timeout)

print_trace 출력을 흐름으로
0 사용자: A-1001 상태?→ 1 LLM: get_asset_status 요청→ 2 도구: "에러: … backup 쓰세요"→ 3 LLM: backup 요청→ 4 도구: 사용중 (스냅샷)→ 5 최종 답변 
# agent.log (Bedrock 호출 줄은 생략)
14:56:21 INFO    backend 호출 asset_id=A-1001 api_key=sk-int***   ← 마스킹됨
14:56:26 WARNING timeout (시도 1/3)
14:56:31 WARNING timeout (시도 2/3)
14:56:36 WARNING timeout (시도 3/3)
14:56:39 INFO    backup 경로 사용 asset_id=A-1001                ← LLM이 갈아탐

=== DoD 판정 ===
정상 호출 처리     : 통과
실패 호출 처리     : 통과 (그래프 중단 없이 답변 생성)
필수 장애 1개 복구 : 통과 (장애 timeout, 백업 도구 호출 있음)
로그 Secret 점검   : 통과 (발견 0건)
최종               : DoD 충족

다른 장애로 돌려 보려면 day4_practice에서 PowerShell로:

$env:FAILURE_MODE="rate_limit"; python failure_agent.py   # 2초, 4초 백오프 로그
$env:FAILURE_MODE="bad_schema"; python failure_agent.py   # 재시도 없이 바로 ERROR
Remove-Item Env:FAILURE_MODE                             # 원래대로

8. 이 설계의 약점 (알고 쓰기)