"프롬프트 입력 후 기도하기(Prompt and Pray)"에서 프로덕션 준비가 된 AI 에이전트로

_Build Log_에 다시 오신 것을 환영합니다. 저는 Nick Creighton입니다. 이번 에피소드에서 우리는 AutoGPT가 헤드라인을 장식한 이후로 개발자들을 괴롭혀 온 llm.call()-while-loop 방식의 문제를 과감히 도려냈습니다. 만약 여러분이 에이전트가 제자리걸음만 하며 크레딧을 낭비하다가 처참하게 충돌하는 모습을 본 적이 있다면, "AI를 만들었지만 프로덕션(Production) 환경에서 하루도 버티지 못한다"는 느낌이 무엇인지 잘 알 것입니다. 이 동반 블로그 포스트에서는 팟캐스트의 에너지 넘치는 대화를 구체적이고 단계적인 플레이북(Playbook)으로 번역하여 전달하고자 합니다. 여러분은 각 실패 모드(Failure mode)의 이유(Why), 회복 탄력성이 있는 프레임워크를 구축하기 위해 필요한 무엇(What), 그리고 코드 스니펫(Code snippets), 도구 추천, 모니터링 팁을 포함한 **방법(How)**을 배우게 될 것입니다. 이 글을 다 읽을 때쯤이면, 여러분은 그 순진한 루프(Loop)를 신뢰할 수 있는 관측성(Observability)을 갖추고, 안정적이며, 비용 효율적으로 실행되는 프로덕션 등급의 에이전트로 교체할 수 있을 것입니다.

왜 고전적인 "프롬프트 입력 후 기도하기(Prompt-and-Pray)" 루프가 깨지는가

문제가 있는 패턴의 구조부터 살펴보겠습니다:

• 단일 프롬프트 → LLM 호출 → 파싱(Parse) → 반복. 이 루프는 "완료"와 "정체"의 차이를 전혀 알지 못합니다.
• 지속적인 상태(Persistent state) 없음. 매 반복마다 컨텍스트(Context)를 버리기 때문에, 에이전트는 특정 도구를 이미 시도했다거나 특정 API가 에러를 반환했다는 사실을 기억하지 못합니다.
• 에러 핸들링(Error handling) 없음. 도구가 잘못된 형식의 JSON을 반환하거나 예상치 못한 HTTP 상태 코드를 반환하면, 루프는 이를 성공적인 답변으로 취급하고 계속해서 나선형으로 악화됩니다.
• 관측성(Observability) 없음. 로그도, 메트릭(Metrics)도 없으며, 에이전트가 왜 멈췄는지 또는 왜 예산을 초과했는지 알 수 있는 방법이 없습니다.

그 결과는 무엇일까요? 무한 루프, 환각(Hallucinated)된 API 응답, 그리고 마치 무작위적인 LLM 환각처럼 보이는 갑작스러운 "크레딧 부족" 에러입니다. 프로덕션 환경에서 이는 다운타임, 화난 사용자, 그리고 신용카드 고지서를 보고 우는 회계사로 이어집니다.

신뢰할 수 있는 에이전트 프레임워크 구축의 핵심 기둥

프로덕션 환경에 적합한(production-ready) 에이전트를 구축하는 것은 화려한 프롬프팅(prompting)의 문제라기보다 엔지니어링 규율(engineering discipline)의 문제입니다. 어떤 LLM(Large Language Model)이라도 감독 없이 실행되도록 하기 전에 반드시 강화해야 할 다섯 가지 기둥은 다음과 같습니다.

1. 지속적인 상태 관리 (Persistent State Management)
   에이전트를 상태 머신(state machine)으로 생각하십시오. 각 단계는 내구성이 있는 저장소(Redis, DynamoDB, 또는 간단한 SQLite 파일)로부터 읽고 써야 하며, 이를 통해 다음과 같은 작업을 수행할 수 있어야 합니다:

• 동일한 작업을 재실행하지 않고도 충돌(crash) 발생 후 재개할 수 있습니다.
• 최근 도구 호출(tool calls)의 해시(hash)를 추적하여 루프(loop)를 감지할 수 있습니다.
• 사후에 결정 경로(decision path)를 감사(audit)할 수 있습니다.

예시 (Python + Redis):

import redis, json, uuid

r = redis.Redis(host='localhost', port=6379, db=0)

def get_session(session_id):
    raw = r.get(session_id)
    return json.loads(raw) if raw else {"history": [], "variables": {}}

def save_session(session_id, data):
    r.set(session_id, json.dumps(data))

2. 견고한 오류 복구 (Robust Error Recovery)
   모든 외부 호출(LLM, HTTP API, 데이터베이스)에는 다음 중 하나를 수행할 수 있는 try / except 블록이 필요합니다:

• 지수 백오프(exponential back-off)를 적용하여 재시도(Retry)하거나,
• 구조화된 에러(structured error)를 LLM에 다시 전달하여 실패에 대해

structlog를 사용한 예시:
import structlog
log = structlog.get_logger()
def call_tool(name, args, session_id):
log.info("tool_start", tool=name, args=args, session=session_id)
try:
result = tool_registryname
log.info("tool_success", tool=name, result=result, session=session_id)
return result
except Exception as exc:
log.error("tool_failure", tool=name, error=str(exc), session=session_id)
raise
4. 비용 및 비율 제어 (Cost & Rate Controls)
운영 에이전트는 API 예산을 무한정 사용해서는 안 됩니다. 다음을 구현하세요:

• 세션별 토큰 제한 (Token caps per session). 사전에 정의된 예산 초과 시 중단합니다.
• 비율 제한 (Rate limiting). 토큰 버킷 알고리즘(token bucket algorithm)을 사용하여 사용자당 초당 호출 횟수를 제한합니다.
• 비용 알림 (Cost alerts). 24시간 동안 지출액이 $X를 초과할 경우 Slack/webhook 알림을 전송합니다. 간단한 토큰 예산 가드:
  MAX_TOKENS = 25_000
  def check_budget(used):
  if used >= MAX_TOKENS:
  raise RuntimeError("토큰 예산 소진")

5. 선언적 도구 계약 (Declarative Tool Contracts)
   LLM이 실시간으로 JSON 스키마를 추측하게 하는 대신, 모든 도구에 대한 엄격한 계약(contract)을 정의하세요:

contracts/tool_schemas.py

TOOL_SCHEMAS = {
"search_web": {
"type": "object",
"properties": {
"query": {"type": "string"},
"max_results": {"type": "integer", "minimum": 1, "maximum": 10}
},
"required": ["query"]
},
...
}
실제 도구를 호출하기 전에 LLM의 출력을 스키마와 검증하세요. 검증에 실패하면 모델에게 수정 프롬프트(corrective prompt)를 다시 전송합니다.

실행 가능한 청사진: 첫 번째 프로덕션 에이전트 구축

아래는 최소한이지만 프로덕션 환경에서 사용 가능한 파이썬 골격 코드입니다. 좋아하는 LLM SDK로 자유롭게 교체할 수 있습니다. 패턴은 동일하게 유지됩니다.

• 도구 레지스트리(Tool registry) 및 스키마 정의. - 상태 지속성(State persistence) 설정. - 오류 처리 및 토큰 예산(Token budgeting)을 포함한 LLM 호출 래핑. - 명시적인 종료 조건이 포함된 루프 실행.

import json, time, uuid
from typing import Dict, Any
from redis import Redis
from jsonschema import validate, ValidationError

# 1️⃣ 도구 정의 (Tool definitions) -------------------------------------------------

def search_web(query: str, max_results: int = 3) -> list:
    # 플레이스홀더(placeholder) – 실제 검색 클라이언트로 교체하세요
    return [{"title": f"Result {i+1}", "url": f"https://example.com/{i}"} for i in range(max_results)]

TOOL_REGISTRY = {
    "search_web": search_web, # 여기에 더 많은 도구를 추가하세요
}

# 2️⃣ 상태 지속성 (State persistence) ------------------------------------------------

redis_client = Redis()

def load_state(session_id: str) -> Dict[str, Any]:
    raw = redis_client.get(session_id)
    return json.loads(raw) if raw else {"history": [], "tokens_used": 0, "loop_count": 0}

def save_state(session_id: str, state: Dict[str, Any]):
    redis_client.set(session_id, json.dumps(state))

# 3️⃣ LLM 헬퍼 (LLM helper) -------------------------------------------------------

MAX_TOKENS = 20_000

def llm_prompt(prompt: str) -> str:
    # 스텁(Stub) – 여기에 LLM SDK 호출을 주입하세요
    # 반드시 가공되지 않은 문자열 응답(raw string response)을 반환해야 합니다
    raise NotImplementedError

def safe_llm(prompt: str, state: dict) -> str:
    resp = llm_prompt(prompt)
    # 매우 단순한 토큰 추정 방식 – 가능하다면 정확한 카운트로 교체하세요
    tokens = len(resp.split())
    state["tokens_used"] += tokens
    check_budget(state["tokens_used"])
    return resp

def check_budget(used: int):
    if used > MAX_TOKENS:
        raise RuntimeError("예산 초과 (Budget exceeded)")

# 4️⃣ 핵심 루프 (Core loop) ---------------------------------------------------------

def run_agent(initial_goal: str):
    session_id = str(uuid.uuid4())
    state = load_state(session_id)
    
    while True:
        # 종료 가드(termination guard) – 무한 루프 방지
        if state["loop_count"] >= 20:
            raise RuntimeError("루프 제한 도달 (Loop limit reached)")
        
        state["loop_count"] += 1
        #

컨텍스트를 포함한 프롬프트 구축
system_prompt = f""" 당신은 다음과 같은 임무를 맡은 AI 에이전트입니다: {initial_goal} 당신은 다음 도구(tools)에 접근할 수 있습니다: {', '.join(TOOL_REGISTRY.keys())} 도구를 호출해야 할 때는 도구 스키마(tool schema)와 일치하는 JSON을 출력하세요. 최종 답변에 도달했다면, {{"final": true, "answer": "..."}}로 응답하세요. 동일한 인자(arguments)로 같은 도구 호출을 반복하지 마세요. """
user_prompt = "\n".join(state["history"][-5:]) # 최근 컨텍스트만 포함
full_prompt = system_prompt + "\n" + user_prompt

# 4️⃣ 안전망을 갖춘 LLM 호출
try:
    llm_raw = safe_llm(full_prompt, state)
except Exception as e:
    # 로그를 남기고 탈출합니다 – 재시도하거나 폴백(fallback)할 수도 있습니다
    print(f"LLM failure: {e}")
    break

# 원시 응답(raw response) 기록
state["history"].append(llm_raw)

# --------------------------------------------------------------
# 5️⃣ 응답 파싱 – 도구 호출인지 최종 답변인지 결정
# --------------------------------------------------------------
try:
    parsed = json.loads(llm_raw)
except json.JSONDecodeError:
    # LLM이 JSON이 아닌 형식을 반환한 경우; 수정을 요청합니다
    state["history"].append(
        "Please respond with valid JSON according to the schema."
    )
    continue

# 최종 답변인가?
if parsed.get("final"):
    print("✅ Final answer:", parsed.get("answer"))
    break

# 도구 호출 처리
tool_name = parsed.get("tool")
args = parsed.get("args", {})

if tool_name not in TOOL_REGISTRY:
    state["history"].append(f"Unknown tool '{tool_name}'.")
    continue

# 스키마에 따른 인자(args) 검증
schema = TOOL_SCHEMAS.get(tool_name)
try:
    validate(instance=args, schema=schema)
except ValidationError as ve:
    state["history"].append(
        f"Argument validation failed: {ve.message}. Please correct."
    )
    continue

# 도구 실행
try:
    result = TOOL_REGISTRY[tool_name](**args) # 다음 턴에 결과를 LLM에 다시 전달
    state["history"].append(json.dumps({"tool_result": result}))
except Exception as exc:
    state["history"].append(
        f"Tool execution error: {str(exc)}."
    )

다른 접근 방식을 시도하십시오.") continue # 각 반복 후 상태 유지 save_state(session_id, state) # 정리 (선택 사항) redis_client.delete(session_id) # 실행 예시 ------------------------------------------------------------- if __name__ == "__main__": run_agent("최신 오픈 소스 Rust 웹 프레임워크를 찾아 핵심 기능을 요약해줘.") 이 스켈레톤(skeleton)은 영속적 상태 (Redis), 예산 확인, 스키마 검증 (schema validation), 명시적 루프 종료, 그리고 LLM 출력과 도구 실행(tool execution) 사이의 명확한 핸드오프(hand-off)라는 모든 핵심 요소를 포함하고 있습니다.

### 시간을 투자할 가치가 있는 도구 및 라이브러리
- **LangChain 2.x** – 이제 프롬프트 생성, LLM 호출, 도구 호출을 분리하는 Runnable 추상화(abstractions)를 제공합니다. 더 높은 수준의 DSL이 필요한 경우 사용하세요.
- **OpenTelemetry** – 대부분의 Python 로거(logger)와 통합되어 LLM 호출, DB 쿼리, 외부 API 전반에 걸친 분산 트레이싱 (distributed traces)을 제공합니다.
- **RedisJSON** – 매번 직렬화/역직렬화(serializing/deserializing)를 수행하지 않고도 구조화된 세션 데이터를 저장하고 쿼리할 수 있게 해줍니다.
- **jsonschema** – 가볍고 빠르며 런타임 계약 준수 (runtime contract enforcement)에 완벽합니다.
- **Prometheus + Grafana** – 토큰 사용량, 에러율, 루프 횟수에 대한 실시간 대시보드용으로 사용합니다.

### 실전 모니터링
관측 가능성 (Observability)은 조치를 취하지 않으면 무용지물입니다. 다음과 같은 신호에 대해 알림을 설정하세요:
- **토큰 예산 초과 (Token-budget breach).** 담당 엔지니어를 태그하는 Slack 웹훅 (webhook)을 트리거합니다.
- **루프 제한 도달 (Loop-limit hits).** 프롬프트나 도구 계약 (tool contract)의 논리적 결함을 나타냅니다.
- **도구 실패 급증 (Tool failure spikes).** 상위 API의 중단일 수 있습니다. 캐시된 응답으로 자동 폴백 (auto-fallback) 하세요.
- **지연 시간 이상치 (Latency outliers).** LLM 호출이 X초를 초과하면 속도 제한 (rate limits)에 걸리고 있을 수 있습니다. 토큰에 대한 Prometheus 샘플 규칙...

_이 기사는 저희 팟캐스트에서 계속됩니다..._