Python 에이전트에 71줄짜리 블랙박스를 추가하여 DuckDB로 200달러의 손실을 추적한 방법

사건은 지루한 고객 지원 자동화 작업에서 시작되었습니다.

사용자 요청을 받아 프라이빗 문서 인덱스(private document index)를 검색하고, 답변을 요약하여 검토자에게 전달하는 작업입니다. 영웅적인 일은 아닙니다. 데모가 끝나고 실제 워크플로우가 시작될 때 구축하게 되는 전형적인 Python 에이전트(Python agent)의 모습이죠.

그러던 중 한 번의 실행이 재시도 루프(retry loop)에 빠져 멈춰버렸습니다.

제가 발견하기 전까지 200달러를 다 써버린 것은 아닙니다. 실제 테스트 실행 비용은 더 저렴했습니다. 문제는 투영(projection)이었습니다. 동일한 잘못된 루프, 동일한 문서 검색, 동일한 모델 호출(model calls)이 밤샘 배치(overnight batch) 작업 내부에 그대로 남겨진 것이 문제였습니다. 피할 수 있었던 단 한 번의 실패로 인해 예상 비용은 200달러에 육박했습니다.

에이전트가 생성한 답변은 졸린 눈으로 검토하기에 충분할 만큼 그럴싸해 보였습니다. 하지만 그 이면의 트레이스(trace)는 전혀 그럴싸하지 않았습니다. 에이전트는 잘못된 입력값으로 올바른 도구(tool)를 호출했고, 오래된 컨텍스트(stale context)를 대상으로 재시도했으며, 과거의 결과를 요약했고, 매 턴(turn)마다 비용을 계속 지불하고 있었습니다.

그때 저는 에이전트를 단순한 채팅 기능처럼 다루는 것을 멈췄습니다.

대신 블랙박스(black box)가 필요한 시스템처럼 다루기 시작했습니다.

대시보드(dashboard)도 아니고, 전체 관측성 스택(observability stack)도 아니며, 또 다른 호스팅 서비스도 아닙니다.

그저 다음과 같은 질문에 답할 수 있는 로컬 파일 하나면 충분했습니다:

에이전트가 무엇을 시도했는가?
어떤 도구를 호출했는가?
도구가 어떤 입력을 받았는가?
도구가 실패했는가?
시간이 얼마나 걸렸는가?
실행이 비용 또는 턴 제한을 초과했는가?
모든 것이 끝난 후 실행 기록을 쿼리(query)할 수 있는가?

우리는 순수 Python으로 이 블랙박스를 구축한 다음, DuckDB를 사용하여 이를 작은 크래시 데이터베이스(crash database)처럼 조사할 것입니다.

전과 후

수정 전의 디버깅(debugging)은 다음과 같았습니다:

최종 답변이 틀렸습니다.
모델이 환각(hallucination)을 일으킨 것 같습니다.
아마 검색 도구가 잘못된 데이터를 반환했을 수도 있습니다.
...

이것은 디버깅이 아닙니다. 구문 강조(syntax highlighting)가 적용된 추측일 뿐입니다.

수정 후의 디버깅은 다음과 같습니다:

Turn 1에서 잘못된 쿼리로 search_docs를 호출했습니다.
도구가 147.82ms 후에 타임아웃(timeout)되었습니다.
재시도 시 오래된 컨텍스트를 사용했습니다.
...

버그는 동일합니다. 하지만 하루가 완전히 달라졌습니다.

문제의 형태

일반적인 Python 스크립트는 보통 한 곳에서 실패합니다.

하지만 에이전트는 체인(chain) 전체에 걸쳐 실패합니다.

사용자 요청 (User Request) -> 모델 결정 (Model Decision) -> 도구 호출 (Tool Call) -> 도구 결과 (Tool Result) -> 다음 턴 (Next Turn) -> 최종 답변 (Final Answer)

최종 답변만 기록한다면, 그것은 일기장에 불과합니다.

체인 (chain) 전체를 기록한다면, 그것은 증거가 됩니다.

가장 단순하면서도 유용한 형식은 JSONL입니다. 한 줄에 하나의 이벤트(event)를 기록합니다.

{"type":"tool_start","tool":"search_docs","input":{"query":"rate limits"}}
{"type":"tool_end","tool":"search_docs","duration_ms":83.4,"ok":true}
{"type":"turn_end","turn":2,"total_cost_usd":0.0041}

JSONL은 딱 적당할 정도로 지루한 형식입니다. 깔끔하게 추가(append)할 수 있고, 하나의 커다란 JSON 문서보다 충돌(crash) 상황에서 더 잘 견디며, 일반적인 도구들로 검색할 수 있습니다.

실제로 작동하는 작은 기록기 (Recorder)

여기에 기록기가 있습니다.

이 기록기는 네 가지 일을 수행합니다:

모든 실행(run)에 고유한 ID 부여
추가 전용 (append-only) JSONL 이벤트 기록
도구 실행 시간 (duration) 측정
디스크에 기록하기 전 명백한 비밀 정보(secrets)를 정화 (sanitize)

from __future__ import annotations

import json
...

sanitize() 함수가 완벽하지는 않습니다. 이것은 금고(vault)가 아니라 안전벨트(seatbelt)입니다.

그럼에도 불구하고, 이 패턴에서 가장 당혹스러운 상황, 즉 API 키를 조용히 저장해버리는 유용한 디버그 추적(debug trace)을 만드는 일을 방지해 줍니다.

먼저 도구 하나를 감싸기 (Wrap)

도구 하나로 시작하세요. 첫날부터 모든 것에 계측 (instrument)을 적용하지 마세요.

import random
import time

...

이제 호출을 기록합니다:

box = AgentBlackBox("traces/run.jsonl")

query = "python agent trace format"
...

traces/run.jsonl을 열어보면 키(key)가 가려져(redacted) 있습니다.

{"tool":"search_docs","input":{"query":"python agent trace format","api_key":"[redacted]"}}

그 작은 디테일이 중요합니다. 디버깅 (Debugging)이 두 번째 사고를 유발해서는 안 됩니다.

저렴한 실행 가드 (Run Guard) 추가하기

대부분의 폭주하는 에이전트 (Agent) 사례는 무해해 보였던 루프 (Loop)에서 시작됩니다.

따라서 블랙박스 (Black box)는 단순히 무슨 일이 일어났는지만 기록해서는 안 됩니다. 실행을 거부했을 때가 언제인지도 기록해야 합니다.

class RunStopped(RuntimeError):
    pass

...

이것은 정확한 과금이 아닙니다. 실제 토큰 수 (Token counts)를 알 수 있는 경우에는 제공업체 (Provider)의 응답을 사용하세요.

여기서의 목표는 로컬 트립와이어 (Local tripwire, 현지 지뢰)를 만드는 것입니다. 실행이 중단될 때 명확한 이유를 남기도록 해야 합니다.

작은 에이전트 루프 (Agent Loop)

이 가짜 루프는 가동 부품을 작게 유지해 줍니다.

가짜 모델 섹션을 실제 모델 호출로 교체하세요.

def estimate_cost(input_tokens: int, output_tokens: int) -> float:
    return input_tokens * 0.0000005 + output_tokens * 0.0000015

...

일반적인 질문으로 한 번 실행해 보세요.

print(run_agent("How should I debug Python agent tools?"))

그 다음, 잘못된 질문으로 실행해 보세요.

print(run_agent("timeout during document search"))

두 번째 실행은 실패해야 하지만, 이제는 흔적을 남기며 실패하게 됩니다.

테스트를 위해 예산 중단을 강제하려면, 임시로 max_usd = 0.0001로 설정하세요. 그러면 다음 가드 체크 (Guard check) 시 루프가 조용히 계속 진행되도록 두는 대신 guard_stop 이벤트를 기록할 것입니다.

DuckDB로 충돌 쿼리하기

이 부분은 JSONL이 단순한 로깅 (Logging)이 아니라 디버깅 도구 (Debugging tool)처럼 느껴지게 만드는 핵심입니다.

DuckDB를 설치하세요:

pip install duckdb

그 다음 트레이스 (Trace)를 쿼리하세요:

import duckdb

def query_trace(path: str = "traces/run.jsonl") -> None:
...

이제 실행합니다:

query_trace()

결과물은 다음과 같은 모습일 것입니다:

Event counts
+-------------+--------+
| type        | events |
...

이제 충돌이 발생한 행은 미스터리가 아니라 쿼리 결과로 나타납니다:

일반적인 출력 로그(print logs)가 짜증스럽게 만드는 다음과 같은 질문들에 답할 수 있습니다:

어떤 도구가 가장 자주 실패했는가?
어떤 도구가 가장 느렸는가?
어느 턴(turn)에서 예산을 초과했는가?
동일한 입력이 반복적으로 실패했는가?
가드레일(guard)이 실행을 중단시켰는가, 아니면 도구가 먼저 충돌(crash)했는가?

이것이 바로 업그레이드입니다.

"로그를 가지고 있다"가 아니라,

"실행 과정을 심문(interrogate)할 수 있다"는 것입니다.

실제 프로젝트에서 내가 기록할 것들

데모용으로는 위의 트레이스(trace)만으로도 충분합니다.

실제 프로젝트라면, 다음과 같은 필드들을 추가할 것입니다:

model (모델)
provider (제공자)
prompt_hash (프롬프트 해시)
tool_schema_version (도구 스키마 버전)
input_tokens (입력 토큰)
output_tokens (출력 토큰)
finish_reason (종료 사유)
retry_count (재시도 횟수)
user_id_hash (사용자 ID 해시)
environment (환경)

기본적으로는 다음 항목들을 기록하지 않을 것입니다:

raw access tokens (가공되지 않은 액세스 토큰)
private documents (개인 문서)
full customer prompts (고객의 전체 프롬프트)
full tool responses with sensitive data (민감한 데이터가 포함된 도구의 전체 응답)
cookies or request headers (쿠키 또는 요청 헤더)

지루하지만 단순한 보안 규칙은 이렇습니다:

동작을 디버깅(debug)할 수 있을 만큼만 기록하세요. 누군가에게 해를 끼칠 수 있을 만큼 기록하지 마세요.

한 문장으로 요약한 패턴

모든 에이전트 실행은 보관하기 안전하고, 쿼리(query)하기 쉬우며, 프로세스가 충돌한 후에도 유용한, 로컬의 추가 전용(append-only) 이벤트 스트림을 생성해야 합니다.

이 문장은 새로운 프롬프트 기술(prompt trick)보다 덜 흥미로울 수 있습니다.

하지만 당신의 주말을 지켜줄 가능성은 더 높습니다.

전체 파일

여기에 전체 예제를 한곳에 모아두었습니다.

from __future__ import annotations

import json
...

이 전체 파일에는 유심히 살펴볼 가치가 있는 한 줄이 있습니다:

box.record("run_start", question=question, max_turns=max_turns, max_usd=max_usd)

이 한 줄이 프로그램의 태도를 바꿉니다.

실행(run)은 더 이상 모델과의 사적인 대화가 아닙니다. 그것은 당신이 검사하고, 쿼리하고, 개선할 수 있는 트레이스(trace)가 있는 기록된 실행입니다.

이것이 데모와 신뢰할 수 있는 무언가의 차이입니다.

다음에 무엇을 추가하시겠습니까: 프롬프트 해시, 토큰 수, 스크린샷, 체크포인트?