OpenTelemetry로 LLM 에이전트 추적 — 스팬 계측, 토큰 비용 추적, 프로덕션 디버깅

"이 에이전트가 왜 $3.70을 썼나?" 프로덕션에서 이 질문에 답할 수 없다면 에이전트를 배포할 준비가 안 된 겁니다. OpenTelemetry는 그 답을 구조화합니다.

---

핵심 요약 → 에이전트 옵저버빌리티 ≠ 일반 APM — 비결정적 실행, 토큰 기반 비용, 무결한 실패가 다름 → OpenTelemetry GenAI Semantic Conventions: gen_ai.* 표준 속성 (2026년 초 experimental → stable 진행 중) → 자동 계측: AnthropicInstrumentor().instrument() — 코드 변경 없이 모든 API 호출 자동 추적 → 수동 계측: 에이전트 루프·툴 호출·재시도 스텝을 명시적 스팬으로 감쌈 → 핵심 속성: gen_ai.usage.input_tokens, gen_ai.usage.output_tokens, gen_ai.request.model → 비용 추적: 토큰 수 × 단가를 커스텀 속성으로 추가 → 백엔드: Jaeger (로컬 개발), Grafana Tempo (프로덕션), Datadog/Honeycomb (SaaS) → OTel 오버헤드: 스팬당 <1ms — LLM 레이턴시(100ms~30s) 대비 무시 가능

---

왜 일반 APM으로 부족한가

에이전트 옵저버빌리티가 말해줘야 하는 것: "이 에이전트는 LLM 호출 3회, 툴 실행 2회, 토큰 12,400개($0.037), 두 번째 툴 호출이 타임아웃으로 실패했지만 에이전트가 자가 수정했다." 일반 APM이 말하는 것: "이 엔드포인트가 4.2초 걸렸다."

# 에이전트 옵저버빌리티가 다른 세 가지 이유

1. 비결정적 실행
   같은 입력 → 다른 툴 호출 순서
   → 이슈 재현 시 정확한 실행 경로 캡처 필요

2. 토큰 기반 비용
   CPU/메모리가 아닌 토큰이 비용 단위
   "3,000 토큰으로 해결할 걸 50,000 토큰 쓴 에이전트"
   = 비용 이상이자 로직 버그 신호

3. 무결한 실패 (Silent Failure)
   HTTP 200 반환 + 틀린 답
   → 스팬에 프롬프트/완성 내용·평가 점수 포함해야 감지 가능

---

1. 설치 + 기본 설정

# 핵심 패키지
pip install opentelemetry-api opentelemetry-sdk

# 프로바이더별 자동 계측
pip install opentelemetry-instrumentation-anthropic
pip install opentelemetry-instrumentation-openai
pip install opentelemetry-instrumentation-langchain

# 익스포터 (백엔드 선택)
pip install opentelemetry-exporter-otlp          # Jaeger, Grafana, Datadog 등
pip install opentelemetry-exporter-jaeger         # Jaeger 전용

# OpenLLMetry (LLM 특화 계측 확장)
pip install traceloop-sdk

# otel_setup.py — 앱 시작 시 1회 실행

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
import os

def setup_otel(service_name: str = "llm-agent"):
    """
    OpenTelemetry 초기화 — 앱 진입점에서 1회 호출
    """
    resource = Resource.create({
        "service.name": service_name,
        "service.version": os.getenv("APP_VERSION", "1.0.0"),
        "deployment.environment": os.getenv("ENV", "development"),
    })

    provider = TracerProvider(resource=resource)

    # 익스포터 설정 (환경에 따라 선택)
    exporter = OTLPSpanExporter(
        endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
    )
    provider.add_span_processor(BatchSpanProcessor(exporter))

    trace.set_tracer_provider(provider)
    return trace.get_tracer(service_name)

# ── 자동 계측 등록 ──
# 코드 변경 없이 모든 Anthropic·OpenAI 호출 자동 추적
from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor
from opentelemetry.instrumentation.openai import OpenAIInstrumentor

AnthropicInstrumentor().instrument()
OpenAIInstrumentor().instrument()

---

2. GenAI 표준 속성 — gen_ai.*

OpenAI, Anthropic, LangChain에 대한 자동 계측 패키지가 있으며, 앱 시작 시 한 번 호출하면 이후 모든 API 호출이 자동으로 추적됩니다.

# 자동 계측이 수집하는 gen_ai.* 표준 속성들

GEN_AI_ATTRIBUTES = {
    # 요청
    "gen_ai.system":              "anthropic",           # 프로바이더
    "gen_ai.request.model":       "claude-sonnet-4-6",  # 요청 모델
    "gen_ai.operation.name":      "chat",                # 작업 유형
    "gen_ai.request.max_tokens":  4096,

    # 응답
    "gen_ai.response.model":      "claude-sonnet-4-6",  # 실제 응답 모델
    "gen_ai.response.finish_reasons": ["end_turn"],

    # 토큰 사용량 (비용 계산의 핵심)
    "gen_ai.usage.input_tokens":  2450,
    "gen_ai.usage.output_tokens": 381,

    # Anthropic 전용
    "gen_ai.provider.name":       "anthropic",
    "gen_ai.usage.cache_read_input_tokens":  1200,  # 캐시 히트
    "gen_ai.usage.cache_creation_input_tokens": 800,  # 캐시 생성

    # 스팬 메타
    "gen_ai.agent.name":          "code-review-agent",  # 에이전트 식별
}

# 스팬 이름 패턴: "{operation.name} {gen_ai.system}"
# 예: "chat anthropic", "chat openai"

---

3. 자동 계측 — 코드 변경 없이

# ── 자동 계측 활성화 후 평소처럼 사용 ──

import anthropic

# setup_otel() + AnthropicInstrumentor().instrument() 이후
client = anthropic.Anthropic()

# 이 호출이 자동으로 스팬 생성
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "코드 리뷰해줘"}]
)

# → 자동 생성되는 스팬:
# 이름: "chat anthropic"
# 속성:
#   gen_ai.system = "anthropic"
#   gen_ai.request.model = "claude-sonnet-4-6"
#   gen_ai.usage.input_tokens = 245
#   gen_ai.usage.output_tokens = 89
#   duration = 1.2s
# → 백엔드에서 바로 조회 가능


# 프롬프트·완성 내용 캡처 (기본 OFF — PII 주의)
AnthropicInstrumentor().instrument(
    enrich_token_usage=True,     # 토큰 상세 정보 추가
    capture_content=True,        # 프롬프트·응답 내용 캡처 (개발 환경만)
)

# 프로덕션에서 내용 캡처 활성화 방법 (환경 변수)
# OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true

---

4. 수동 계측 — 에이전트 루프 전체 추적

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
import anthropic
import time

tracer = trace.get_tracer("my-agent")
client = anthropic.Anthropic()

MODEL_PRICES = {
    "claude-sonnet-4-6": {"input": 3.0, "output": 15.0},  # per 1M tokens
    "claude-opus-4-7":   {"input": 5.0, "output": 25.0},
}

def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    prices = MODEL_PRICES.get(model, {"input": 3.0, "output": 15.0})
    return (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000


def run_agent(goal: str, tools: list) -> str:
    """
    에이전트 루프 전체를 루트 스팬으로 감쌈
    내부 LLM 호출·툴 실행이 자식 스팬으로 중첩
    """

    # ── 루트 스팬: 전체 에이전트 실행 ──
    with tracer.start_as_current_span(
        "gen_ai.agent.invoke",
        attributes={
            "gen_ai.agent.name": "code-review-agent",
            "gen_ai.agent.goal": goal[:200],   # 처음 200자만
            "gen_ai.system":     "anthropic",
        }
    ) as agent_span:

        total_input_tokens = 0
        total_output_tokens = 0
        total_cost = 0.0
        step_count = 0

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

        while True:
            step_count += 1

            # ── LLM 호출 스팬 (자식) ──
            with tracer.start_as_current_span(
                f"chat anthropic",
                attributes={
                    "gen_ai.request.model":     "claude-sonnet-4-6",
                    "gen_ai.agent.step_number": step_count,
                    "gen_ai.operation.name":    "chat",
                }
            ) as llm_span:

                try:
                    response = client.messages.create(
                        model="claude-sonnet-4-6",
                        max_tokens=4096,
                        tools=tools,
                        messages=messages
                    )

                    # 토큰 사용량 기록
                    input_tok  = response.usage.input_tokens
                    output_tok = response.usage.output_tokens
                    cost = calculate_cost("claude-sonnet-4-6", input_tok, output_tok)

                    llm_span.set_attributes({
                        "gen_ai.usage.input_tokens":  input_tok,
                        "gen_ai.usage.output_tokens": output_tok,
                        "gen_ai.response.finish_reasons": [response.stop_reason],
                        # 커스텀 비용 속성
                        "llm.cost.usd":          round(cost, 6),
                        "llm.cost.input_usd":    round(input_tok * 3.0 / 1e6, 6),
                        "llm.cost.output_usd":   round(output_tok * 15.0 / 1e6, 6),
                    })

                    total_input_tokens  += input_tok
                    total_output_tokens += output_tok
                    total_cost          += cost

                except Exception as e:
                    llm_span.set_status(Status(StatusCode.ERROR, str(e)))
                    llm_span.record_exception(e)
                    raise

            # 완료 조건 확인
            if response.stop_reason == "end_turn":
                final_text = response.content[0].text
                break

            # ── 툴 실행 스팬 (자식) ──
            for tool_use in [b for b in response.content if b.type == "tool_use"]:
                with tracer.start_as_current_span(
                    f"gen_ai.tool.execute",
                    attributes={
                        "gen_ai.tool.name":       tool_use.name,
                        "gen_ai.tool.call_id":    tool_use.id,
                        "gen_ai.agent.step_number": step_count,
                    }
                ) as tool_span:
                    try:
                        result = execute_tool(tool_use.name, tool_use.input)
                        tool_span.set_attribute("gen_ai.tool.success", True)
                        tool_span.set_attribute("gen_ai.tool.output_length", len(str(result)))
                    except Exception as e:
                        tool_span.set_status(Status(StatusCode.ERROR, str(e)))
                        tool_span.set_attribute("gen_ai.tool.success", False)
                        result = f"오류: {str(e)}"

                messages.append({"role": "assistant", "content": response.content})
                messages.append({
                    "role": "user",
                    "content": [{"type": "tool_result", "tool_use_id": tool_use.id, "content": str(result)}]
                })

        # ── 루트 스팬 최종 집계 ──
        agent_span.set_attributes({
            "gen_ai.agent.total_steps":        step_count,
            "gen_ai.usage.total_input_tokens": total_input_tokens,
            "gen_ai.usage.total_output_tokens": total_output_tokens,
            "llm.cost.total_usd":              round(total_cost, 6),
            "gen_ai.agent.success":            True,
        })

        return final_text

---

5. 비용 대시보드 — Grafana 메트릭 설정

# 토큰·비용을 메트릭으로도 수집 (트레이스 외)

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.prometheus import PrometheusMetricReader

# 메트릭 설정
reader = PrometheusMetricReader()  # Prometheus → Grafana
provider = MeterProvider(metric_readers=[reader])
metrics.set_meter_provider(provider)

meter = metrics.get_meter("llm-agent")

# 메트릭 정의
token_counter = meter.create_counter(
    name="gen_ai.usage.tokens",
    description="LLM 호출당 토큰 사용량",
    unit="token"
)

cost_counter = meter.create_counter(
    name="llm.cost.usd",
    description="LLM 호출 비용",
    unit="USD"
)

llm_duration = meter.create_histogram(
    name="gen_ai.client.operation.duration",
    description="LLM 호출 레이턴시",
    unit="s"
)

# 호출 시 메트릭 기록
def record_llm_metrics(
    model: str,
    input_tokens: int,
    output_tokens: int,
    duration_sec: float,
    agent_name: str
):
    labels = {
        "gen_ai.request.model": model,
        "gen_ai.system": "anthropic",
        "gen_ai.agent.name": agent_name,
    }

    # 토큰 카운터
    token_counter.add(input_tokens,
        {**labels, "gen_ai.token.type": "input"})
    token_counter.add(output_tokens,
        {**labels, "gen_ai.token.type": "output"})

    # 비용 카운터
    cost = calculate_cost(model, input_tokens, output_tokens)
    cost_counter.add(cost, labels)

    # 레이턴시 히스토그램
    llm_duration.record(duration_sec, labels)

# Grafana 대시보드 패널 예시 (PromQL)

# 모델별 시간당 토큰 비용
sum by (gen_ai_request_model) (
  rate(llm_cost_usd_total[1h])
) * 3600

# 에이전트별 일일 지출
sum by (gen_ai_agent_name) (
  increase(llm_cost_usd_total[24h])
)

# 평균 에이전트 스텝 수
avg(gen_ai_agent_total_steps)

# P95 LLM 레이턴시
histogram_quantile(0.95,
  rate(gen_ai_client_operation_duration_bucket[5m])
)

---

6. 프로덕션 트러블슈팅 패턴

# ── 패턴 1: 비용 이상 탐지 ──

def detect_cost_anomaly(span_context):
    """
    단일 에이전트 실행에서 예상 비용 초과 시 알림
    """
    cost = span_context.get("llm.cost.total_usd", 0)
    steps = span_context.get("gen_ai.agent.total_steps", 0)

    # 임계값 초과 시 경고 스팬 이벤트 추가
    if cost > 1.0:  # $1.00 초과
        span_context["span"].add_event(
            "cost_threshold_exceeded",
            attributes={
                "threshold_usd": 1.0,
                "actual_usd": cost,
                "steps": steps,
                "likely_cause": "무한 루프 또는 과도한 컨텍스트" if steps > 20 else "복잡한 태스크"
            }
        )


# ── 패턴 2: 루프 감지 ──

def detect_loop(messages: list) -> bool:
    """
    같은 툴을 3회 이상 연속 호출하면 루프 의심
    """
    if len(messages) < 6:
        return False

    recent_tools = [
        m.get("tool_name") for m in messages[-6:]
        if m.get("role") == "tool"
    ]

    # 같은 툴이 3회 이상 연속
    for i in range(len(recent_tools) - 2):
        if recent_tools[i] == recent_tools[i+1] == recent_tools[i+2]:
            return True
    return False


# ── 패턴 3: 세션·사용자 연결 (컨텍스트 전파) ──

from opentelemetry.baggage import set_baggage, get_baggage
from opentelemetry import context

def run_agent_with_user_context(
    goal: str,
    user_id: str,
    session_id: str
) -> str:
    # 루트 스팬에 사용자 정보 태깅
    with tracer.start_as_current_span(
        "gen_ai.agent.invoke",
        attributes={
            "user.id":       user_id,
            "session.id":    session_id,
            "gen_ai.agent.name": "assistant",
        }
    ) as span:
        return run_agent(goal, tools=[])
        # → Jaeger에서 user.id로 특정 사용자의 전체 에이전트 실행 추적 가능

---

7. 로컬 개발 환경 — Docker Compose로 Jaeger 띄우기

# docker-compose.yml
version: '3.8'
services:
  jaeger:
    image: jaegertracing/all-in-one:latest
    ports:
      - "16686:16686"   # Jaeger UI
      - "4317:4317"     # OTLP gRPC
      - "4318:4318"     # OTLP HTTP
    environment:
      - COLLECTOR_OTLP_ENABLED=true

  # 선택: Prometheus + Grafana
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin

# 실행
docker-compose up -d

# Jaeger UI: http://localhost:16686
# → 서비스 선택 → 트레이스 목록 → 클릭하면 전체 스팬 트리
# → "gen_ai.usage.total_input_tokens" 속성으로 필터링

---

8. LangSmith·Braintrust와 함께 쓰는 법

# LangSmith: LangChain 프레임워크 사용 시
# OTel 스팬이 LangSmith 트레이스와 자동 연결됨

from langchain.callbacks import LangChainTracer
from opentelemetry.instrumentation.langchain import LangChainInstrumentor

LangChainInstrumentor().instrument()
# → LangChain 내 모든 LLM 호출이 OTel 스팬으로 캡처
# → 동시에 LangSmith에도 전송 가능 (이중 계측)


# Braintrust: OTel 스팬 → Braintrust 트레이스 변환
# BraintrustSpanProcessor 추가로 기존 OTel 파이프라인에 Braintrust 연결

from braintrust.otel import BraintrustSpanProcessor
provider.add_span_processor(BraintrustSpanProcessor(
    api_key=os.getenv("BRAINTRUST_API_KEY"),
    project_name="my-agent"
))
# → 프로덕션 실패 → Eval 케이스 1클릭 전환 가능

---

텍스트 요약 — 언제 무엇을 써야 하나

# 백엔드 선택 기준

Jaeger:
  "지금 당장 무료로 시작하고 싶다"
  → Docker Compose 1줄, UI 즉시 사용 가능

Grafana Tempo + Prometheus:
  "메트릭(비용·레이턴시)과 트레이스를 같이 보고 싶다"
  → 프로덕션 표준 스택

Datadog / Honeycomb:
  "관리형 백엔드 원함, 사용량 기반 과금 OK"
  → 팀 규모 크고 인프라 관리 줄이고 싶을 때

LangSmith:
  "LangChain 프레임워크 이미 쓰는 중"
  → 에이전트 트레이스 + 프롬프트 버전 관리 함께

Braintrust:
  "프로덕션 실패를 Eval 데이터셋으로 바꾸고 싶다"
  → 평가 파이프라인 중시하는 팀

---

결론

✅ 지금 당장 추가해야 할 3가지

• AnthropicInstrumentor().instrument() — 코드 변경 없이 모든 호출 자동 추적
• 에이전트 루트 스팬에 llm.cost.total_usd 속성 추가 — 비용 가시성 즉시 확보
• Jaeger Docker Compose 로컬 환경 설정 — 5분 내 트레이스 UI 확인

✅ 에이전트 옵저버빌리티의 핵심 원칙

• 루트 스팬 = 전체 에이전트 실행 (goal → final result)
• 자식 스팬 = LLM 호출 + 툴 실행 + 재시도 각각
• gen_ai.* 표준 속성 → 어떤 백엔드에서도 동일한 쿼리

❌ 주의사항

• capture_content=True → 프로덕션에서 PII 데이터 유출 위험 — 개발 환경만
• GenAI Semantic Conventions = 2026년 초 experimental 상태 — 속성명 변경 가능성
• OTEL_SEMCONV_STABILITY_OPT_IN 환경 변수로 마이그레이션 버퍼 확보 필요