LangChain으로 복잡한 LLM 애플리케이션을 개발할 때, 체인의 각 단계에서 어떤 일이 발생하는지 파악하는 것은 디버깅과 성능 최적화의 핵심입니다. 이번 튜토리얼에서는 LangChain의 콜백 시스템과 추적 도구를 활용하여 프로덕션 환경에서 체인 호출을 효과적으로 추적하는 방법을 다룹니다. HolySheep AI를 통해 다양한 모델의 비용과 성능을 한눈에 모니터링하는 실전 기법도 함께 알아보겠습니다.
1. LangChain 콜백 시스템 아키텍처 이해
LangChain의 콜백 시스템은 이벤트 기반 아키텍처로 설계되어 있습니다. 체인 실행 중 발생하는 다양한 이벤트(시작, 종료, 토큰 생성, 오류 등)를 가로채어 후처리할 수 있습니다. 저는 실제 프로젝트에서 이 시스템을 활용하여 토큰 사용량과 응답 지연 시간을 실시간으로 추적하고 있습니다.
1.1 콜백 시스템의 핵심 컴포넌트
LangChain의 콜백 시스템은 크게 세 가지 레벨로 구분됩니다. 첫 번째는 전역 콜백으로, 모든 체인 실행에 적용됩니다. 두 번째는 태그 기반 필터링으로, 특정 태그가 붙은 체인만 추적합니다. 세 번째는 체인별 개별 콜백으로, 특정 체인에만 적용됩니다. 이三层 구조를 이해하면 프로덕션 환경에서 불필요한 오버헤드 없이 필요한 만큼만 추적할 수 있습니다.
1.2 CallbackHandler 인터페이스 구조
커스텀 콜백 핸들러를 만들기 위해서는 CallbackHandler 인터페이스를 구현해야 합니다. 이 인터페이스는 20개 이상의 메서드를 제공하지만, 실제로 구현해야 하는 핵심 메서드는 제한적입니다.
"""
LangChain 커스텀 콜백 핸들러 구현
HolySheep AI API 연동 + 체인 호출 추적
"""
from typing import Any, Dict, List, Optional
from datetime import datetime
import time
import json
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_core.agents import AgentFinish, AgentAction
HolySheep AI 클라이언트
from openai import OpenAI
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
class HolySheepTracingHandler(BaseCallbackHandler):
"""
HolySheep AI 체인 호출 추적을 위한 커스텀 콜백 핸들러
- 토큰 사용량 추적
- 응답 지연 시간 측정
- 모델별 비용 계산
- 체인 실행 흐름 로깅
"""
# HolySheep AI 모델별 비용 ($/MTok)
MODEL_COSTS = {
"gpt-4.1": 8.0,
"gpt-4.1-mini": 1.5,
"claude-sonnet-4-20250514": 15.0,
"claude-haiku-4-20250514": 3.0,
"gemini-2.5-flash": 2.50,
"deepseek-chat-v3.2": 0.42,
}
def __init__(self, trace_name: str = "default"):
self.trace_name = trace_name
self.chains: List[Dict[str, Any]] = []
self.current_chain: Optional[Dict[str, Any]] = None
self.total_tokens = 0
self.total_cost_cents = 0.0
def on_chain_start(
self,
serialized: Dict[str, Any],
inputs: Dict[str, Any],
*,
run_id: str,
parent_run_id: Optional[str] = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> None:
"""체인 실행 시작 시 호출"""
self.current_chain = {
"run_id": run_id,
"parent_run_id": parent_run_id,
"start_time": time.time(),
"start_datetime": datetime.now().isoformat(),
"serialized_name": serialized.get("name", "unknown"),
"tags": tags or [],
"metadata": metadata or {},
"input_tokens": 0,
"output_tokens": 0,
"cost_cents": 0.0,
"llm_calls": [],
"errors": []
}
print(f"[{self.trace_name}] 🔵 Chain Started: {serialized.get('name', 'unknown')}")
print(f" Run ID: {run_id}, Tags: {tags}")
def on_chain_end(
self,
outputs: Dict[str, Any],
*,
run_id: str,
parent_run_id: Optional[str] = None,
**kwargs: Any,
) -> None:
"""체인 실행 종료 시 호출"""
if self.current_chain and self.current_chain["run_id"] == run_id:
self.current_chain["end_time"] = time.time()
self.current_chain["end_datetime"] = datetime.now().isoformat()
self.current_chain["duration_ms"] = (
self.current_chain["end_time"] - self.current_chain["start_time"]
) * 1000
self.current_chain["outputs"] = outputs
# 총합 업데이트
self.total_tokens += (
self.current_chain["input_tokens"] +
self.current_chain["output_tokens"]
)
self.total_cost_cents += self.current_chain["cost_cents"]
print(f"[{self.trace_name}] 🟢 Chain Ended: {self.current_chain['serialized_name']}")
print(f" Duration: {self.current_chain['duration_ms']:.2f}ms")
print(f" Input Tokens: {self.current_chain['input_tokens']}")
print(f" Output Tokens: {self.current_chain['output_tokens']}")
print(f" Cost: ${self.current_chain['cost_cents']:.4f}")
self.chains.append(self.current_chain)
self.current_chain = None
def on_llm_start(
self,
serialized: Dict[str, Any],
prompts: List[str],
*,
run_id: str,
parent_run_id: Optional[str] = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> None:
"""LLM 호출 시작 시 호출"""
if self.current_chain:
llm_call = {
"run_id": run_id,
"start_time": time.time(),
"model": serialized.get("name", "unknown"),
"prompt_length": len(prompts[0]) if prompts else 0,
"num_prompts": len(prompts)
}
self.current_chain["llm_calls"].append(llm_call)
def on_llm_end(
self,
response: LLMResult,
*,
run_id: str,
parent_run_id: Optional[str] = None,
**kwargs: Any,
) -> None:
"""LLM 응답 수신 시 호출"""
if self.current_chain:
# 해당 llm_call 찾기
for llm_call in reversed(self.current_chain["llm_calls"]):
if llm_call["run_id"] == run_id and "end_time" not in llm_call:
llm_call["end_time"] = time.time()
llm_call["duration_ms"] = (
llm_call["end_time"] - llm_call["start_time"]
) * 1000
# 토큰 및 비용 계산
if response.llm_output:
usage = response.llm_output.get("token_usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
llm_call["input_tokens"] = input_tokens
llm_call["output_tokens"] = output_tokens
# 비용 계산
model = llm_call["model"]
cost_per_mtok = self.MODEL_COSTS.get(model, 8.0)
cost = (
(input_tokens / 1_000_000) * cost_per_mtok +
(output_tokens / 1_000_000) * cost_per_mtok
)
llm_call["cost_cents"] = cost * 100
# 총합 업데이트
self.current_chain["input_tokens"] += input_tokens
self.current_chain["output_tokens"] += output_tokens
self.current_chain["cost_cents"] += llm_call["cost_cents"]
break
def on_llm_error(
self,
error: BaseException,
*,
run_id: str,
parent_run_id: Optional[str] = None,
**kwargs: Any,
) -> None:
"""LLM 호출 오류 시 호출"""
if self.current_chain:
self.current_chain["errors"].append({
"run_id": run_id,
"error_type": type(error).__name__,
"error_message": str(error),
"timestamp": datetime.now().isoformat()
})
print(f"[{self.trace_name}] 🔴 LLM Error: {error}")
def get_summary(self) -> Dict[str, Any]:
"""추적 결과 요약 반환"""
return {
"trace_name": self.trace_name,
"total_chains": len(self.chains),
"total_tokens": self.total_tokens,
"total_cost_cents": self.total_cost_cents,
"total_cost_dollars": self.total_cost_cents / 100,
"chains": self.chains
}
def export_to_json(self, filepath: str) -> None:
"""추적 결과를 JSON 파일로 저장"""
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(self.get_summary(), f, indent=2, ensure_ascii=False)
print(f"[{self.trace_name}] 📄 Traces exported to {filepath}")
사용 예제
if __name__ == "__main__":
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
# HolySheep AI를 통한 LLM 초기화
llm = ChatOpenAI(
model="gpt-4.1-mini",
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
# 커스텀 콜백 핸들러 연결
tracing_handler = HolySheepTracingHandler(trace_name="production-chain")
prompt = PromptTemplate.from_template(
"{topic}에 대해 3문장으로 설명해줘."
)
chain = LLMChain(
llm=llm,
prompt=prompt,
callbacks=[tracing_handler] # 체인별 콜백
)
# 체인 실행
result = chain.invoke(
{"topic": "인공지능"},
config={"callbacks": [tracing_handler]}
)
print(f"\n{'='*50}")
print("추적 요약:")
summary = tracing_handler.get_summary()
print(f"총 체인 실행: {summary['total_chains']}")
print(f"총 토큰: {summary['total_tokens']}")
print(f"총 비용: ${summary['total_cost_dollars']:.4f}")
# JSON으로 내보내기
tracing_handler.export_to_json("chain_traces.json")2. LangSmith 통합으로 프로덕션 추적
LangSmith는 LangChain에서 공식 제공하는 디버깅 및 추적 플랫폼입니다. 복잡한 멀티체인을 추적하고 팀과 협력할 때 특히 유용합니다. HolySheep AI와 함께 사용하면 비용 최적화와 성능 모니터링을 통합할 수 있습니다.
2.1 HolySheep AI + LangSmith 연동 설정
"""
HolySheep AI + LangSmith 통합 추적 시스템
다중 모델 비용 추적 + 분산 추적
"""
import os
from typing import Dict, Any, List, Optional
from datetime import datetime
import asyncio
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI
from langsmith import Client
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
HolySheep AI 모델별 비용 설정
MODEL_PRICING = {
# HolySheep AI 모델
"gpt-4.1": {"input": 8.0, "output": 8.0}, # $/MTok
"gpt-4.1-mini": {"input": 1.5, "output": 6.0},
"gpt-4.1-flash": {"input": 0.5, "output": 2.0},
"claude-sonnet-4-20250514": {"input": 15.0, "output": 75.0},
"claude-haiku-4-20250514": {"input": 3.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 10.0},
"deepseek-chat-v3.2": {"input": 0.42, "output": 1.68},
}
class HolySheepCostTracker(BaseCallbackHandler):
"""
HolySheep AI 비용 추적기
- 모델별 토큰 사용량
- 실시간 비용 계산
- LangSmith 메타데이터 연동
"""
def __init__(self, project_name: str = "holysheep-production"):
self.project_name = project_name
self.usage_records: List[Dict[str, Any]] = []
self.current_usage: Dict[str, int] = {"prompt_tokens": 0, "completion_tokens": 0}
self.total_cost_cents = 0.0
def on_llm_end(
self,
response: LLMResult,
*,
run_id: str,
parent_run_id: Optional[str] = None,
**kwargs: Any,
) -> None:
"""LLM 응답에서 토큰 사용량 추출 및 비용 계산"""
if not response.llm_output:
return
usage = response.llm_output.get("token_usage", {})
model = response.llm_output.get("model_name", "unknown")
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
# 비용 계산
pricing = MODEL_PRICING.get(model, {"input": 8.0, "output": 8.0})
input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
output_cost = (completion_tokens / 1_000_000) * pricing["output"]
total_cost = input_cost + output_cost
# 레코드 저장
record = {
"run_id": run_id,
"model": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
"input_cost": input_cost,
"output_cost": output_cost,
"total_cost": total_cost,
"timestamp": datetime.now().isoformat()
}
self.usage_records.append(record)
self.total_cost_cents += total_cost * 100
# 누적 사용량 업데이트
self.current_usage["prompt_tokens"] += prompt_tokens
self.current_usage["completion_tokens"] += completion_tokens
def get_usage_report(self) -> Dict[str, Any]:
"""사용량 보고서 생성"""
return {
"project_name": self.project_name,
"total_requests": len(self.usage_records),
"total_prompt_tokens": self.current_usage["prompt_tokens"],
"total_completion_tokens": self.current_usage["completion_tokens"],
"total_cost_cents": self.total_cost_cents,
"total_cost_dollars": self.total_cost_cents / 100,
"model_breakdown": self._get_model_breakdown(),
"records": self.usage_records
}
def _get_model_breakdown(self) -> Dict[str, Dict[str, Any]]:
"""모델별 사용량 분류"""
breakdown: Dict[str, Dict[str, Any]] = {}
for record in self.usage_records:
model = record["model"]
if model not in breakdown:
breakdown[model] = {
"requests": 0,
"prompt_tokens": 0,
"completion_tokens": 0,
"total_cost_cents": 0.0
}
breakdown[model]["requests"] += 1
breakdown[model]["prompt_tokens"] += record["prompt_tokens"]
breakdown[model]["completion_tokens"] += record["completion_tokens"]
breakdown[model]["total_cost_cents"] += record["total_cost"] * 100
return breakdown
HolySheep AI 클라이언트 팩토리
class HolySheepClientFactory:
"""
HolySheep AI API 기반 LLM 클라이언트 팩토리
단일 API 키로 다중 모델 관리
"""
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@classmethod
def create_client(
cls,
model: str,
temperature: float = 0.7,
max_tokens: Optional[int] = None,
**kwargs
):
"""HolySheep AI를 통한 LLM 클라이언트 생성"""
if model.startswith("gpt-") or model.startswith("o1"):
return ChatOpenAI(
model=model,
api_key=cls.API_KEY,
base_url=cls.BASE_URL,
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
elif model.startswith("claude-"):
return ChatAnthropic(
model=model,
anthropic_api_key=cls.API_KEY,
# Anthropic은 base_url 변경 불가, 직접 호출 시 HolySheep 프록시 사용
timeout=60,
**kwargs
)
elif model.startswith("gemini-"):
return ChatGoogleGenerativeAI(
model=model,
google_api_key=cls.API_KEY, # HolySheep 경유
**kwargs
)
else:
# DeepSeek 등 기타 모델
return ChatOpenAI(
model=model,
api_key=cls.API_KEY,
base_url=cls.BASE_URL,
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
프로덕션 예제: 다중 모델 체인
async def run_multi_model_chain():
"""
HolySheep AI를 통한 다중 모델 파이프라인
- 비용 최적화를 위한 모델 자동 선택
- 통합 추적 시스템
"""
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
# 비용 추적기 초기화
cost_tracker = HolySheepCostTracker(project_name="multi-model-pipeline")
# HolySheep AI를 통한 다양한 모델 초기화
models = {
"fast": HolySheepClientFactory.create_client("gpt-4.1-mini", temperature=0.3),
"balanced": HolySheepClientFactory.create_client("claude-haiku-4-20250514", temperature=0.5),
"powerful": HolySheepClientFactory.create_client("gpt-4.1", temperature=0.7),
"cheap": HolySheepClientFactory.create_client("deepseek-chat-v3.2", temperature=0.5),
}
# 모델 선택 로직
async def select_model(task_complexity: str) -> str:
"""태스크 복잡도에 따른 모델 선택"""
complexity_map = {
"simple": "cheap", # DeepSeek: $0.42/MTok
"moderate": "fast", # GPT-4.1-mini: $1.5/MTok
"complex": "balanced", # Claude Haiku: $3/MTok
"advanced": "powerful" # GPT-4.1: $8/MTok
}
return complexity_map.get(task_complexity, "fast")
# 파이프라인 구성
prompt = ChatPromptTemplate.from_template(
"당신은 {role} 전문가입니다. {question}에 대해 답변해주세요."
)
# 복잡도에 따른 동적 체인 실행
results = []
tasks = [
("simple", "1+1은 무엇인가요?"),
("moderate", "Python에서 리스트와 튜플의 차이는?"),
("complex", "마이크로서비스 아키텍처의 장단점을 설명해주세요."),
("advanced", "트랜스포머 아키텍처의 어텐션 메커니즘을 수학적으로 설명해주세요."),
]
for complexity, question in tasks:
model_key = await select_model(complexity)
model = models[model_key]
chain = prompt | model | StrOutputParser()
result = await chain.ainvoke(
{"role": "기술", "question": question},
config={"callbacks": [cost_tracker]}
)
results.append({
"complexity": complexity,
"model": model_key,
"question": question,
"answer_length": len(result)
})
print(f"[{complexity}] {model_key} 모델 사용: {len(result)}자 응답")
# 비용 보고서 출력
report = cost_tracker.get_usage_report()
print(f"\n{'='*60}")
print("HolySheep AI 비용 보고서")
print(f"{'='*60}")
print(f"총 요청 수: {report['total_requests']}")
print(f"총 토큰: {report['total_prompt_tokens'] + report['total_completion_tokens']:,}")
print(f"총 비용: ${report['total_cost_dollars']:.4f}")
print(f"\n모델별 사용량:")
for model, stats in report["model_breakdown"].items():
print(f" {model}:")
print(f" 요청: {stats['requests']}")
print(f" 토큰: {stats['prompt_tokens'] + stats['completion_tokens']:,}")
print(f" 비용: ${stats['total_cost_cents']/100:.4f}")
if __name__ == "__main__":
asyncio.run(run_multi_model_chain())3. 프로덕션 레벨 추적 시스템 구축
실제 프로덕션 환경에서는 단순한 콘솔 로깅보다 더 강력한 추적 시스템이 필요합니다. 분산 추적, 실시간 대시보드, 알림 시스템까지 통합하는 방법을 알아보겠습니다.
3.1 실시간 모니터링 대시보드
"""
프로덕션 레벨 실시간 모니터링 시스템
HolySheep AI 체인 추적 + Grafana 연동
"""
import asyncio
import threading
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
import json
import logging
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class ChainMetrics:
"""체인 실행 메트릭"""
chain_id: str
chain_name: str
start_time: datetime
end_time: Optional[datetime] = None
duration_ms: float = 0.0
status: str = "running" # running, success, error
input_tokens: int = 0
output_tokens: int = 0
cost_cents: float = 0.0
error_message: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
@property
def total_tokens(self) -> int:
return self.input_tokens + self.output_tokens
class RealTimeMetricsCollector(BaseCallbackHandler):
"""
실시간 메트릭 수집기
스레드 세이프한 메트릭 집계
HolySheep AI 비용 추적 통합
"""
# 모델별 비용 (HolySheep AI 기준)
MODEL_PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"gpt-4.1-mini": {"input": 1.5, "output": 6.0},
"gpt-4.1-flash": {"input": 0.5, "output": 2.0},
"claude-sonnet-4-20250514": {"input": 15.0, "output": 75.0},
"claude-haiku-4-20250514": {"input": 3.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 10.0},
"deepseek-chat-v3.2": {"input": 0.42, "output": 1.68},
}
def __init__(
self,
aggregation_window_seconds: int = 60,
on_alert: Optional[Callable[[str, Dict], None]] = None
):
self._lock = threading.Lock()
self.aggregation_window = timedelta(seconds=aggregation_window_seconds)
self.on_alert = on_alert
# 메트릭 저장소
self._chains: Dict[str, ChainMetrics] = {}
self._window_metrics: List[ChainMetrics] = []
self._last_window_start = datetime.now()
# aggregations
self._total_requests = 0
self._total_tokens = 0
self._total_cost_cents = 0.0
self._error_count = 0
# 슬라이딩 윈도우 Aggregations
self._latencies: List[float] = []
self._model_usage: Dict[str, Dict[str, int]] = defaultdict(
lambda: {"requests": 0, "input_tokens": 0, "output_tokens": 0}
)
def _update_window(self):
"""시간 윈도우 업데이트 및 이전 윈도우 메트릭 초기화"""
now = datetime.now()
if now - self._last_window_start >= self.aggregation_window:
self._window_metrics.clear()
self._last_window_start = now
logger.info("Metrics window reset")
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""토큰 사용량 기반 비용 계산"""
pricing = self.MODEL_PRICING.get(model, {"input": 8.0, "output": 8.0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return (input_cost + output_cost) * 100 # cents
def on_chain_start(self, serialized: Dict, inputs: Dict, **kwargs) -> None:
"""체인 시작 추적"""
run_id = kwargs.get("run_id", "unknown")
chain_name = serialized.get("name", "unknown")
with self._lock:
self._chains[run_id] = ChainMetrics(
chain_id=run_id,
chain_name=chain_name,
start_time=datetime.now(),
metadata={"input_keys": list(inputs.keys()) if inputs else []}
)
self._total_requests += 1
def on_chain_end(self, outputs: Dict, **kwargs) -> None:
"""체인 종료 추적"""
run_id = kwargs.get("run_id", "unknown")
with self._lock:
if run_id in self._chains:
chain = self._chains[run_id]
chain.end_time = datetime.now()
chain.duration_ms = (
chain.end_time - chain.start_time
).total_seconds() * 1000
chain.status = "success"
# 슬라이딩 윈도우 업데이트
self._window_metrics.append(chain)
self._latencies.append(chain.duration_ms)
self._update_window()
# 오버헤드 체크 (>5초)
if chain.duration_ms > 5000:
self._trigger_alert(
"high_latency",
{
"chain": chain.chain_name,
"duration_ms": chain.duration_ms,
"threshold_ms": 5000
}
)
def on_chain_error(self, error: Exception, **kwargs) -> None:
"""체인 오류 추적"""
run_id = kwargs.get("run_id", "unknown")
with self._lock:
if run_id in self._chains:
chain = self._chains[run_id]
chain.end_time = datetime.now()
chain.duration_ms = (
chain.end_time - chain.start_time
).total_seconds() * 1000
chain.status = "error"
chain.error_message = str(error)
self._error_count += 1
self._trigger_alert(
"chain_error",
{
"chain": chain.chain_name,
"error": str(error),
"duration_ms": chain.duration_ms
}
)
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
"""LLM 호출 종료 시 토큰 및 비용 추적"""
run_id = kwargs.get("run_id", "unknown")
if not response.llm_output:
return
usage = response.llm_output.get("token_usage", {})
model = response.llm_output.get("model_name", "unknown")
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
cost_cents = self._calculate_cost(model, input_tokens, output_tokens)
with self._lock:
# 현재 실행 중인 체인에 합산
for chain in self._chains.values():
if chain.status == "running":
chain.input_tokens += input_tokens
chain.output_tokens += output_tokens
chain.cost_cents += cost_cents
break
# 전역 Aggregations 업데이트
self._total_tokens += input_tokens + output_tokens
self._total_cost_cents += cost_cents
# 모델별 사용량 업데이트
self._model_usage[model]["requests"] += 1
self._model_usage[model]["input_tokens"] += input_tokens
self._model_usage[model]["output_tokens"] += output_tokens
def _trigger_alert(self, alert_type: str, data: Dict[str, Any]) -> None:
"""알림 트리거"""
logger.warning(f"Alert: {alert_type} - {data}")
if self.on_alert:
self.on_alert(alert_type, data)
def get_dashboard_metrics(self) -> Dict[str, Any]:
"""대시보드용 메트릭 반환"""
with self._lock:
# P50, P95, P99 지연 시간 계산
sorted_latencies = sorted(self._latencies) if self._latencies else [0]
p50 = sorted_latencies[int(len(sorted_latencies) * 0.5)] if sorted_latencies else 0
p95 = sorted_latencies[int(len(sorted_latencies) * 0.95)] if sorted_latencies else 0
p99 = sorted_latencies[int(len(sorted_latencies) * 0.99)] if sorted_latencies else 0
# 현재 실행 중인 체인
running_chains = [
{
"chain_id": c.chain_id,
"chain_name": c.chain_name,
"running_time_ms": (datetime.now() - c.start_time).total_seconds() * 1000
}
for c in self._chains.values() if c.status == "running"
]
return {
"timestamp": datetime.now().isoformat(),
"summary": {
"total_requests": self._total_requests,
"total_tokens": self._total_tokens,
"total_cost_cents": self._total_cost_cents,
"total_cost_dollars": self._total_cost_cents / 100,
"error_count": self._error_count,
"error_rate": self._error_count / max(self._total_requests, 1)
},
"latency": {
"p50_ms": p50,
"p95_ms": p95,
"p99_ms": p99,
"window_requests": len(self._window_metrics)
},
"model_usage": {
model: {
"requests": stats["requests"],
"tokens": stats["input_tokens"] + stats["output_tokens"],
"cost_cents": (
(stats["input_tokens"] / 1_000_000) * self.MODEL_PRICING.get(model, {"input": 8.0})["input"] +
(stats["output_tokens"] / 1_000_000) * self.MODEL_PRICING.get(model, {"output": 8.0})["output"]
) * 100
}
for model, stats in self._model_usage.items()
},
"running_chains": running_chains
}
def export_prometheus_metrics(self) -> str:
"""Prometheus 형식 메트릭 내보내기"""
metrics = self.get_dashboard_metrics()
lines = [
"# HELP holyheep_total_requests Total number of chain requests",
"# TYPE holyheep_total_requests counter",
f"holyheep_total_requests {metrics['summary']['total_requests']}",
"",
"# HELP holyheep_total_tokens Total tokens processed",
"# TYPE holyheep_total_tokens counter",
f"holyheep_total_tokens {metrics['summary']['total_tokens']}",
"",
"# HELP holyheep_total_cost_cents Total cost in cents",
"# TYPE holyheep_total_cost_cents gauge",
f"holyheep_total_cost_cents {metrics['summary']['total_cost_cents']}",
"",
"# HELP holyheep_latency_p95 P95 latency in milliseconds",
"# TYPE holyheep_latency_p95 gauge",
f"holyheep_latency_p95 {metrics['latency']['p95_ms']}",
"",
"# HELP holyheep_error_rate Error rate",
"# TYPE holyheep_error_rate gauge",
f"holyheep_error_rate {metrics['summary']['error_rate']}",
]
return "\n".join(lines)
모니터링 서비스
class ChainMonitoringService:
"""
체인 모니터링 서비스
HolySheep AI API 호출 추적 + 실시간 대시보드 연동
"""
def __init__(self):
self.collector = RealTimeMetricsCollector(
aggregation_window_seconds=60,
on_alert=self._handle_alert
)
self._running = False
def _handle_alert(self, alert_type: str, data: Dict[str, Any]) -> None:
"""알림 처리"""
# 실제로는 Slack, PagerDuty 등으로 전송
logger.error(f"[ALERT] {alert_type}: {json.dumps(data, ensure_ascii=False)}")
if