반응형
클로드 코드 토큰 경제학 시리즈 8편: 운영과 모니터링 - usage 데이터 제대로 활용하기
💡 시리즈 소개: 이 시리즈는 Claude Code 사용자를 위한 토큰 경제학 가이드입니다. 이번 편에서는 API의 usage 데이터를 정확하게 집계하고 모니터링하는 방법을 다룹니다.
정확한 관리의 핵심
토큰 관리는 '추정'이 아니라 '사전 카운트'와 '사후 usage 합산'으로 합니다.
┌────────────────────────────────────────┐
│ 토큰 관리 사이클 │
├────────────────────────────────────────┤
│ │
│ 사전 ─► count_tokens API │
│ (예산 초과 방지) │
│ ↓ │
│ 실행 ─► 요청 수행 │
│ ↓ │
│ 사후 ─► usage 합산 │
│ (실제 소모 확인) │
│ │
└────────────────────────────────────────┘Messages API의 usage 구조
기본 구조
Anthropic Messages API 응답에는 usage 필드가 포함됩니다.
{
"content": [...],
"usage": {
"input_tokens": 1234,
"output_tokens": 567
}
}
기본 집계
total_tokens = usage.input_tokens + usage.output_tokens
⚠️ 많은 경우
total_tokens가 별도 제공되지 않아 직접 합산이 필요합니다.
확장 usage 필드 (중요!)
"input/output만 합치면 끝"이 아닌 경우
Claude Code/에이전트 환경에서는 확장 필드까지 합산해야 합니다.
캐시 토큰
{
"usage": {
"input_tokens": 1234,
"output_tokens": 567,
"cache_creation_input_tokens": 500,
"cache_read_input_tokens": 200
}
}
| 필드 | 설명 |
|---|---|
cache_creation_input_tokens |
캐시 생성(쓰기) 토큰 |
cache_read_input_tokens |
캐시 읽기 토큰 |
Compaction iterations
Compaction이 내부적으로 추가 호출을 만들 수 있어, 토큰이 iterations 배열로 분해될 수 있습니다.
{
"usage": {
"input_tokens": 1000,
"output_tokens": 200,
"iterations": [
{"input_tokens": 500, "output_tokens": 100},
{"input_tokens": 300, "output_tokens": 80}
]
}
}
올바른 집계 공식
단순 집계 (최소)
total = usage.input_tokens + usage.output_tokens
확장 집계 (권장)
def calculate_total_tokens(usage):
total = usage.input_tokens + usage.output_tokens
# 캐시 토큰 추가
total += getattr(usage, 'cache_creation_input_tokens', 0)
total += getattr(usage, 'cache_read_input_tokens', 0)
# iterations 합산 (compaction 사용 시)
if hasattr(usage, 'iterations'):
for iteration in usage.iterations:
total += iteration.input_tokens + iteration.output_tokens
return total
스트리밍에서의 usage 주의점
누적(Cumulative) 보고
스트리밍 모드에서 message_delta 이벤트의 usage는 누적값입니다.
이벤트 1: usage.output_tokens = 10
이벤트 2: usage.output_tokens = 25 (누적)
이벤트 3: usage.output_tokens = 50 (누적)
이벤트 4: usage.output_tokens = 82 (최종 누적)올바른 집계 방법
# ❌ 잘못된 방법: 모두 더하기
# 10 + 25 + 50 + 82 = 167 (중복 집계!)
# ✅ 올바른 방법: 최종 값만 사용
final_output_tokens = 82
스트리밍 집계 규칙
| 용도 | 방법 |
|---|---|
| 실시간 모니터링 | 마지막 usage 누적값 사용 |
| 정산/리포팅 | 최종 message 응답의 usage 사용 |
사전 검증: Token Count API
용도
실제 생성 호출 없이 토큰 수를 미리 계산합니다.
활용 시나리오
- 컨텍스트 초과 방지
- 예산 초과 차단
- 200K 경계 사전 확인
사용 예시
from anthropic import Anthropic
client = Anthropic()
# 요청 전 토큰 수 확인
count_response = client.messages.count_tokens(
model="claude-sonnet-4-20250514",
system="You are a helpful assistant.",
messages=[
{"role": "user", "content": "분석해주세요: " + long_document}
],
tools=[...] # MCP 도구도 포함하여 계산
)
estimated_tokens = count_response.input_tokens
# 200K 경계 확인
if estimated_tokens > 180000: # 90% 임계치
print(f"⚠️ 경고: {estimated_tokens} 토큰 - 200K 근접!")
# 대안 전략 실행
else:
# 정상 진행
response = client.messages.create(...)
주의사항
Token Count API의 결과는 estimate(추정)입니다. 실제 create 시 미세한 차이가 있을 수 있습니다.
운영 대시보드 설계 원칙
1. 확장 필드에 대비
# 확장 가능한 스키마로 설계
usage_record = {
"timestamp": datetime.now(),
"model": model_id,
"input_tokens": usage.input_tokens,
"output_tokens": usage.output_tokens,
# 확장 필드 (있을 경우)
"cache_creation": getattr(usage, 'cache_creation_input_tokens', None),
"cache_read": getattr(usage, 'cache_read_input_tokens', None),
"iterations": getattr(usage, 'iterations', None),
# 계산된 총계
"total_tokens": calculate_total_tokens(usage)
}
2. 200K 경계 모니터링
def check_threshold(total_input_tokens):
thresholds = {
"safe": 150000, # 75%
"warning": 180000, # 90%
"critical": 200000 # 100%
}
if total_input_tokens >= thresholds["critical"]:
return "CRITICAL: 200K 초과 - 프리미엄 요율!"
elif total_input_tokens >= thresholds["warning"]:
return "WARNING: 200K 근접"
elif total_input_tokens >= thresholds["safe"]:
return "CAUTION: 75% 도달"
return "OK"
3. 세션별 누적 추적
class SessionTokenTracker:
def __init__(self):
self.turns = []
self.total_input = 0
self.total_output = 0
def add_turn(self, usage):
self.turns.append({
"turn": len(self.turns) + 1,
"input": usage.input_tokens,
"output": usage.output_tokens,
"cumulative_input": self.total_input + usage.input_tokens
})
self.total_input += usage.input_tokens
self.total_output += usage.output_tokens
# 경계 체크
return check_threshold(self.total_input)
서드파티 통합 주의사항
파싱 실패 사례
Cloudflare AI Gateway 등 서드파티에서 Anthropic API usage 파싱 실패로 토큰/비용 계산이 깨지는 문제가 보고되었습니다.
점검 사항
| 항목 | 확인 내용 |
|---|---|
| 기본 필드 | input_tokens, output_tokens 파싱 확인 |
| 확장 필드 | cache_*, iterations 파싱 확인 |
| 스키마 변화 | 새 필드 추가 시 대응 가능 여부 |
프레임워크 통합 (예: LangChain)
# LangChain Anthropic 통합 예시
from langchain_anthropic import ChatAnthropic
chat = ChatAnthropic(model="claude-sonnet-4-20250514")
response = chat.invoke("Hello!")
# 메타데이터에서 usage 확인
usage = response.response_metadata.get("usage", {})
💡 프레임워크가 기본 필드만 노출하는지, 확장 필드까지 전달하는지 확인하세요.
비용 계산 로직
기본 비용 계산
def calculate_cost(usage, model="sonnet", is_long_context=False):
# 모델별 단가 (per million tokens)
pricing = {
"haiku": {"input": 1, "output": 5},
"sonnet": {"input": 3, "output": 15},
"opus": {"input": 5, "output": 25}
}
# 장문 컨텍스트 프리미엄 (>200K)
long_context_multiplier = {
"sonnet": {"input": 2, "output": 1.5}, # $6/$22.5
"opus": {"input": 2, "output": 1.5} # $10/$37.5
}
base = pricing[model]
if is_long_context and model in long_context_multiplier:
multiplier = long_context_multiplier[model]
input_rate = base["input"] * multiplier["input"]
output_rate = base["output"] * multiplier["output"]
else:
input_rate = base["input"]
output_rate = base["output"]
input_cost = (usage.input_tokens / 1_000_000) * input_rate
output_cost = (usage.output_tokens / 1_000_000) * output_rate
return input_cost + output_cost
캐시 비용 포함
def calculate_cost_with_cache(usage, model="sonnet"):
base_cost = calculate_cost(usage, model)
# 캐시 비용 (입력 단가 기준)
base_input_rate = {"haiku": 1, "sonnet": 3, "opus": 5}[model]
cache_write_cost = (
getattr(usage, 'cache_creation_input_tokens', 0) / 1_000_000
) * base_input_rate * 1.25 # +25%
cache_read_cost = (
getattr(usage, 'cache_read_input_tokens', 0) / 1_000_000
) * base_input_rate * 0.10 # 10%
return base_cost + cache_write_cost + cache_read_cost
모니터링 알림 설정 예시
임계치 기반 알림
alert_rules = [
{
"name": "200K 경계 근접",
"condition": lambda u: u.input_tokens > 180000,
"severity": "warning",
"action": "notify_slack"
},
{
"name": "비정상 출력량",
"condition": lambda u: u.output_tokens > 50000,
"severity": "warning",
"action": "log_and_review"
},
{
"name": "세션 비용 초과",
"condition": lambda cost: cost > 5.0,
"severity": "critical",
"action": "notify_and_pause"
}
]
핵심 정리
집계 공식
| 수준 | 공식 |
|---|---|
| 최소 | input + output |
| 권장 | input + output + cache_creation + cache_read + Σ(iterations) |
| 스트리밍 | 최종 누적값만 사용 |
사전/사후 관리
| 단계 | 도구 | 용도 |
|---|---|---|
| 사전 | count_tokens | 예산/한도 초과 방지 |
| 실행 중 | 스트리밍 usage | 실시간 모니터링 |
| 사후 | 최종 usage | 정확한 비용 계산 |
주의사항
- 확장 필드 누락 시 과소추정 위험
- 스트리밍 중복 집계 주의
- 서드파티 파싱 검증 필요
다음 편 예고
9편: 토큰 전략 종합 정리 - 비용 최적화의 핵심 원칙
시리즈의 마지막 편으로, 지금까지 다룬 내용을 종합하고 실무에 바로 적용할 수 있는 체크리스트를 제공합니다.
참고 자료
Anthropic API 문서
- Messages API - usage 필드 구조
- Count Tokens API - 사전 토큰 계산
- Streaming - 스트리밍 usage 누적 방식
- Compaction - iterations 배열 구조
SDK 및 통합
- anthropic-sdk-python - Python SDK usage 패턴
- LangChain Anthropic - 프레임워크 통합
이슈 사례
- Cloudflare AI Gateway Thread - usage 파싱 실패 사례
📚 클로드 코드 토큰 경제학 시리즈 | 2026년 2월
반응형
'AI' 카테고리의 다른 글
| 생성형 AI 플랫폼 비교 완전 가이드 소개 (0) | 2026.02.12 |
|---|---|
| 클로드 코드 토큰 경제학 시리즈 9편: 토큰 전략 종합 정리 - 비용 최적화의 핵심 원칙 (0) | 2026.02.12 |
| 클로드 코드 토큰 경제학 시리즈 7편: 토큰 최적화 실전 가이드 - Tool Search, 캐시, 컴팩션 (0) | 2026.02.12 |
| 클로드 코드 토큰 경제학 시리즈 6편: 5k/20k/100k 입력 시나리오 - 현실적인 비용 예측하기 (0) | 2026.02.12 |
| 클로드 코드 토큰 경제학 시리즈 5편: 코드 작업의 토큰 경제학 - 파일, diff, 로그의 비밀 (0) | 2026.02.12 |
