AI 에이전트 블로그
컴퓨터 액세스 시스템

자율 에이전트 시스템의 디버깅과 관찰 가능성

agentsdebuggingobservabilitytutorialarchitecture

조용히 실패하는 자율 에이전트는 에이전트가 없는 것보다 더 나쁩니다. 전통적인 함수가 예외를 던지면 스택 트레이스를 얻습니다. 에이전트가 스무 번의 도구 호출과 세 번의 모델 호출을 거쳐 잘못된 길로 가면 잘못된 답변을 얻습니다 — 명확한 설명 없이.

에이전트 디버깅에는 다른 정신적 모델이 필요합니다. 시스템은 결정론적 경로를 실행하는 것이 아니라 일련의 결정을 내리고 있습니다. 관찰 가능성이란 이러한 결정들을 포착하는 것을 의미합니다 — 입력과 출력뿐만 아니라 그것들을 연결하는 추론까지.

에이전트에 전통적인 디버깅이 실패하는 이유

표준 로깅은 무슨 일이 일어났는지를 포착합니다. 에이전트 관찰 가능성은 왜를 포착해야 합니다 — 모델이 무엇을 결론 내렸는지, 어떤 도구를 왜 선택했는지, 어떤 중간 상태에서 작업하고 있었는지.

실패 모드도 다릅니다:

  • 조용한 환각: 에이전트가 불확실성을 신호하지 않고 자신 있게 잘못된 답변을 생성합니다.
  • 결정 드리프트: 각 단계는 국소적으로 합리적으로 보이지만 시퀀스 전체가 목표에서 벗어납니다.
  • 도구 오용: 에이전트가 올바른 도구를 미묘하게 잘못된 매개변수로 호출합니다.
  • 무한 루프: 에이전트가 실패한 접근 방식을 계속 재시도하며 막힙니다.
  • 컨텍스트 오염: 초기 단계의 잘못된 출력이 이후 모든 추론을 손상시킵니다.

이 중 어느 것도 예외를 생성하지 않습니다. 전체 실행 트레이스를 재구성할 때만 보이는 잘못된 동작을 만들어냅니다.

에이전트 결정을 위한 구조화된 로깅

첫 번째 단계는 모든 에이전트 상호작용을 구조화된 로그로 래핑하는 것입니다. 원시 API 응답을 기록하지 말고 의미적 이벤트를 기록하세요.

import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any
import anthropic


client = anthropic.Anthropic()


@dataclass
class AgentEvent:
    trace_id: str
    step: int
    event_type: str      # "llm_call", "tool_call", "tool_result", "decision", "error"
    model: str | None
    input_tokens: int | None
    output_tokens: int | None
    latency_ms: float | None
    content: dict[str, Any]
    timestamp: float


def log_event(event: AgentEvent):
    print(json.dumps(asdict(event)))  # 로그 싱크로 교체하세요


class TracedAgent:
    def __init__(self, trace_id: str | None = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.step = 0
        self.tools = []


    def add_tool(self, name: str, description: str, input_schema: dict):
        self.tools.append({
            "name": name,
            "description": description,
            "input_schema": input_schema
        })


    def call(self, messages: list[dict], system: str = "") -> str:
        self.step += 1
        start = time.monotonic()


        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            system=system,
            tools=self.tools,
            messages=messages
        )


        latency_ms = (time.monotonic() - start) * 1000


        log_event(AgentEvent(
            trace_id=self.trace_id,
            step=self.step,
            event_type="llm_call",
            model="claude-opus-4-6",
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            latency_ms=latency_ms,
            content={
                "stop_reason": response.stop_reason,
                "text_blocks": [b.text for b in response.content if b.type == "text"],
                "tool_calls": [
                    {"name": b.name, "input": b.input}
                    for b in response.content if b.type == "tool_use"
                ]
            },
            timestamp=time.time()
        ))


        return response

완전한 트레이스 구축

단일 로그 라인으로는 부족합니다 — 각 결정을 결과와 연결하는 전체 실행 트레이스가 필요합니다:

from typing import Callable


def run_traced_agent(
    task: str,
    tools: dict[str, Callable],
    tool_schemas: list[dict],
    system: str,
    max_steps: int = 20,
) -> dict:
    agent = TracedAgent()
    for schema in tool_schemas:
        agent.add_tool(**schema)


    messages = [{"role": "user", "content": task}]
    trace = {"trace_id": agent.trace_id, "task": task, "steps": []}
    step_count = 0


    while step_count < max_steps:
        step_count += 1
        response = agent.call(messages, system=system)


        step_record = {
            "step": step_count,
            "stop_reason": response.stop_reason,
            "model_output": [],
            "tool_results": []
        }


        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    step_record["model_output"].append(block.text)
            trace["steps"].append(step_record)
            trace["final_answer"] = step_record["model_output"][-1] if step_record["model_output"] else ""
            break


        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                step_record["model_output"].append({
                    "tool": block.name,
                    "input": block.input
                })


                tool_fn = tools.get(block.name)
                if not tool_fn:
                    result = f"오류: 알 수 없는 도구 '{block.name}'"
                else:
                    try:
                        result = tool_fn(**block.input)
                    except Exception as e:
                        result = f"도구 오류: {e}"


                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result)
                })
                step_record["tool_results"].append({
                    "tool": block.name,
                    "result_preview": str(result)[:200]
                })


        trace["steps"].append(step_record)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})


    else:
        trace["error"] = f"최대 단계 수 초과 ({max_steps})"


    return trace

루프 감지

무한 루프는 일반적인 실패 모드입니다. 각 LLM 호출의 도구 호출 패턴에 지문을 찍어 감지합니다:

def detect_loop(trace: dict, window: int = 4) -> bool:
    steps = trace["steps"]
    if len(steps) < window:
        return False


    def step_signature(step: dict) -> str:
        tools_called = sorted(
            t["tool"] if isinstance(t, dict) else t
            for t in step.get("model_output", [])
            if isinstance(t, dict) and "tool" in t
        )
        return "|".join(tools_called)


    recent = [step_signature(s) for s in steps[-window:]]
    if len(set(recent)) == 1 and recent[0]:
        return True


    if len(steps) >= 4:
        pattern = [step_signature(s) for s in steps[-4:]]
        if pattern[0] == pattern[2] and pattern[1] == pattern[3]:
            return True


    return False

프로덕션에서 추적할 메트릭

from collections import Counter


def compute_trace_metrics(trace: dict) -> dict:
    steps = trace["steps"]
    errors = [s for s in steps if "error" in s]


    tool_calls_by_name: Counter = Counter()
    for step in steps:
        for output in step.get("model_output", []):
            if isinstance(output, dict) and "tool" in output:
                tool_calls_by_name[output["tool"]] += 1


    return {
        "trace_id": trace["trace_id"],
        "total_steps": len(steps),
        "error_steps": len(errors),
        "tool_call_distribution": dict(tool_calls_by_name),
        "completed": "final_answer" in trace,
        "loop_detected": detect_loop(trace),
    }

알림을 위한 핵심 신호:

  • 루프 비율 > 5% — 에이전트가 막히고 있음
  • 도구별 오류율 > 임계값 — 도구가 고장남
  • 평균 단계 수 증가 추세 — 작업이 어려워지고 있음
  • p99 지연 시간 급등 — 모델 엔드포인트가 느림

OpenTelemetry 통합

이미 OpenTelemetry를 사용하는 팀의 경우 에이전트 트레이스를 스팬으로 내보낼 수 있습니다:

from opentelemetry import trace as otel_trace


tracer = otel_trace.get_tracer("agent")


def run_with_otel(task: str, tools: dict, tool_schemas: list, system: str):
    with tracer.start_as_current_span("agent.run") as root_span:
        root_span.set_attribute("agent.task", task[:200])


        agent = TracedAgent()
        for schema in tool_schemas:
            agent.add_tool(**schema)


        messages = [{"role": "user", "content": task}]


        for step in range(20):
            with tracer.start_as_current_span(f"agent.step.{step}") as step_span:
                response = agent.call(messages, system=system)
                step_span.set_attribute("llm.stop_reason", response.stop_reason)
                step_span.set_attribute("llm.input_tokens", response.usage.input_tokens)


                if response.stop_reason == "end_turn":
                    break

로그에서 PII 삭제

에이전트 로그에는 종종 민감한 데이터가 포함됩니다. 외부 시스템으로 내보내기 전에 삭제합니다:

import re


PII_PATTERNS = [
    (re.compile(r'\b[\w.+-]+@[\w-]+\.[a-z]{2,}\b'), '[이메일]'),
    (re.compile(r'\b01[016789][-\s]?\d{3,4}[-\s]?\d{4}\b'), '[전화번호]'),
    (re.compile(r'\bsk-[a-zA-Z0-9]{20,}\b'), '[API키]'),
]


def redact(text: str) -> str:
    for pattern, replacement in PII_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

가장 중요한 세 가지 메트릭

작업 완료율 — 몇 퍼센트의 실행이 final_answer에 도달하는지 대 max_steps나 오류로 끝나는지. 작업 유형별로 기준선을 설정합니다.

작업당 토큰 비용 — 모든 단계에 걸친 input_tokens + output_tokens의 합계. 완료율 변화 없이 비용이 20% 증가하면 대개 프롬프트 저하를 나타냅니다.

도구 오류율 — error_steps / total_steps. 이 메트릭의 급등은 고장난 도구나 API를 직접 가리킵니다.

에이전트 시스템에서 관찰 가능성은 선택 사항이 아닙니다 — 반복 개선할 수 있는 시스템과 고장나면 재시작만 할 수 있는 시스템의 차이입니다. 구조화된 이벤트와 트레이스 ID로 시작하세요. 루프 감지를 추가하세요. 메트릭을 푸시하세요. 프로덕션에서 첫 번째 장애가 발생하여 추측하는 대신 정확히 무슨 일이 일어났는지 재구성할 수 있을 때 이 투자가 보답합니다.

관련 기사