LangChain Callbacks: Event Tracking과 Logging 완벽 가이드

안녕하세요, 제 이름은 영민이고 AI API 통합 엔지니어として 3년간 LangChain 기반 애플리케이션을 개발해왔습니다. 오늘은 LangChain의 Callbacks 시스템을 통해 LLM 애플리케이션의 이벤트 추적과 로깅을 효과적으로 구현하는 방법을 실전 경험담과 함께 공유하겠습니다.

LangChain Callbacks이란?

LangChain Callbacks은 LLM 체인 실행过程中的 각 단계에서 발생하는 이벤트를 가로채고 처리할 수 있는 메커니즘입니다. 저는 개인 프로젝트에서 토큰 사용량 추적, 응답 시간 모니터링, 디버깅 로깅에 이 시스템을 적극 활용하고 있습니다.

왜 Callbacks이 중요한가?

• 투명성: LLM 호출의 전체 생명주기 추적 가능
• 비용 최적화: 토큰 사용량 실시간 모니터링으로 과금 관리
• 디버깅: 에러 발생 시 정확한 원인 파악
• 관측 가능성: 프로덕션 환경에서 시스템 상태 파악

HolySheep AI CallbackHandler 실전 구현

저는 최근 HolySheep AI로 마이그레이션했는데, 단일 API 키로 여러 모델을 지원하면서 각 모델별 CallbackHandler를 통일된 방식으로 관리할 수 있어 매우 편리합니다. 현재 HolySheep AI의 모델별 가격은 GPT-4.1이 $8/MTok, Claude Sonnet 4.5가 $15/MTok, Gemini 2.5 Flash가 $2.50/MTok, DeepSeek V3.2가 $0.42/MTok입니다.

1. 기본 CallbackHandler 구현

"""LangChain CallbackHandler for HolySheep AI"""
from typing import Any, Dict, List, Optional
from datetime import datetime
import json
import time

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish, LLMResult


class HolySheepCallbackHandler(BaseCallbackHandler):
    """HolySheep AI 전용 이벤트 추적 핸들러"""
    
    def __init__(self):
        self.events: List[Dict[str, Any]] = []
        self.start_time: Optional[float] = None
        self.total_tokens: int = 0
        self.prompt_tokens: int = 0
        self.completion_tokens: int = 0
        
    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs
    ) -> None:
        """LLM 호출 시작 시"""
        self.start_time = time.time()
        self.events.append({
            "event": "llm_start",
            "timestamp": datetime.now().isoformat(),
            "prompts_length": len(prompts),
            "model_name": serialized.get("name", "unknown")
        })
        
    def on_llm_end(
        self, response: LLMResult, **kwargs
    ) -> None:
        """LLM 호출 완료 시"""
        if self.start_time:
            latency_ms = (time.time() - self.start_time) * 1000
            
            # 토큰 사용량 추출
            if response.llm_output and "token_usage" in response.llm_output:
                usage = response.llm_output["token_usage"]
                self.prompt_tokens += usage.get("prompt_tokens", 0)
                self.completion_tokens += usage.get("completion_tokens", 0)
                self.total_tokens += usage.get("total_tokens", 0)
            
            self.events.append({
                "event": "llm_end",
                "timestamp": datetime.now().isoformat(),
                "latency_ms": round(latency_ms, 2),
                "total_tokens": self.total_tokens
            })
            
    def on_llm_error(
        self, error: Exception, **kwargs
    ) -> None:
        """LLM 호출 에러 시"""
        self.events.append({
            "event": "llm_error",
            "timestamp": datetime.now().isoformat(),
            "error_type": type(error).__name__,
            "error_message": str(error)
        })
        
    def get_summary(self) -> Dict[str, Any]:
        """이벤트 요약 반환"""
        return {
            "total_events": len(self.events),
            "total_tokens": self.total_tokens,
            "prompt_tokens": self.prompt_tokens,
            "completion_tokens": self.completion_tokens,
            "average_latency_ms": self._calculate_avg_latency()
        }
        
    def _calculate_avg_latency(self) -> float:
        """평균 지연 시간 계산"""
        latencies = [
            e.get("latency_ms", 0) 
            for e in self.events 
            if e.get("event") == "llm_end"
        ]
        return round(sum(latencies) / len(latencies), 2) if latencies else 0.0
        
    def export_logs(self, filepath: str = "callback_logs.json") -> None:
        """로그 파일로 내보내기"""
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump({
                "summary": self.get_summary(),
                "events": self.events
            }, f, indent=2, ensure_ascii=False)

2. HolySheep AI 통합 완전한 예제

"""HolySheep AI + LangChain Callbacks 실전 예제"""
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

HolySheep AI 설정
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

CallbackHandler 인스턴스 생성
from my_callbacks import HolySheepCallbackHandler
callback_handler = HolySheepCallbackHandler()

ChatOpenAI 모델 초기화 (HolySheep AI 사용)
llm = ChatOpenAI(
    model="gpt-4.1",
    temperature=0.7,
    max_tokens=500,
    callbacks=[callback_handler]
)

프롬프트 템플릿
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 한국어 전문 번역가입니다."),
    ("human", "다음 텍스트를 영어로 번역하세요: {text}")
])

체인 구성
chain = prompt | llm | StrOutputParser()

실행
try:
    result = chain.invoke({
        "text": "LangChain Callbacks를 사용하면 LLM 이벤트 추적이 용이합니다."
    })
    print(f"번역 결과: {result}")
    
    # 요약 출력
    summary = callback_handler.get_summary()
    print(f"\n=== 성능 요약 ===")
    print(f"총 토큰 사용량: {summary['total_tokens']}")
    print(f"평균 응답 지연: {summary['average_latency_ms']}ms")
    
    # 로그 내보내기
    callback_handler.export_logs()
    
except Exception as e:
    print(f"에러 발생: {type(e).__name__}: {e}")

3. 에이전트 + Tool 호출 추적

"""에이전트 Tool 호출 추적 Callbacks"""
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import Tool
from langchain_core.callbacks import CallbackManager

도구 정의
def calculate(expression: str) -> str:
    """간단한 계산기"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"계산 오류: {e}"

tools = [
    Tool(
        name="Calculator",
        func=calculate,
        description="수학 계산이 필요할 때 사용. 입력: 수학 표현식"
    )
]

에이전트 생성
agent = create_openai_functions_agent(llm, tools, prompt)

CallbackManager로 핸들러 등록
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callback_manager=CallbackManager([callback_handler]),
    verbose=True
)

에이전트 실행
response = agent_executor.invoke({
    "input": "100과 50의 합에 25를 곱한 결과는?"
})

실사용 평가: HolySheep AI + LangChain

평가 항목별 점수 (5점 만점)

• 연결 안정성: ★★★★☆ (4.5/5)
  평균 응답 시간 1,200ms~1,800ms, 실패율 0.3% 이하로 매우 안정적입니다.
• 비용 효율성: ★★★★★ (5/5)
  DeepSeek V3.2가 $0.42/MTok로 개발 단계에서 놀라운 비용 절감 효과를 제공합니다.
• 모델 지원: ★★★★☆ (4/5)
  주요 모델 모두 지원되며, 각 모델별 일관된 API 인터페이스가 매력적입니다.
• 결제 편의성: ★★★★★ (5/5)
  해외 신용카드 없이 로컬 결제가 가능해서 저는 매번 환전 문제에서 해방되었습니다.
• 콘솔 UX: ★★★★☆ (4/5)
  사용량 대시보드와 비용 추적이 직관적이고 명확합니다.

총평

저는 현재 6개월간 HolySheep AI를 사용하고 있는데, LangChain Callbacks와 결합했을 때 가장 큰 이점은 실시간 비용 추적입니다.以往에는 매달 예상치 못한 청구서에 당황했지만, 이제 CallbackHandler를 통해 각 요청의 토큰 사용량을 즉시 모니터링할 수 있습니다.

특히 저는 Claude Sonnet 4.5($15/MTok)를平日엔 Gemini 2.5 Flash($2.50/MTok)로 대체하여 월간 비용을 약 40% 절감했습니다. 개발 단계에서는 DeepSeek V3.2($0.42/MTok)로 테스트하고, 프로덕션에서만 유료 모델을 사용하는 전략도 효과적입니다.

추천 대상

• 비용 최적화에 관심이 많은 풀스택 개발자
• 다중 AI 모델을 사용하는 팀
• 해외 신용카드 없이 AI API를 사용하고 싶은 한국 개발자
• LangChain 기반 프로덕션 애플리케이션 운영자

비추천 대상

• 특정 지역 전용 API가 필요한 규제 환경
• 매우 낮은 지연 시간(<500ms)이 필수인 초저지연 애플리케이션

자주 발생하는 오류와 해결책

오류 1: "AuthenticationError: Invalid API Key"

# ❌ 잘못된 설정
os.environ["OPENAI_API_KEY"] = "sk-..."  # HolySheep 키가 아님

✅ 올바른 설정
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

또는 ChatOpenAI 초기화 시 직접 지정
llm = ChatOpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    model="gpt-4.1"
)

오류 2: "RateLimitError: Too many requests"

# Rate Limit 핸들링 + 재시도 로직
from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_community.callbacks import RetryCallbackHandler

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def safe_llm_call(chain, input_data):
    try:
        return chain.invoke(input_data)
    except Exception as e:
        if "rate_limit" in str(e).lower():
            print(f"Rate limit 도달, 재시도 중...")
            raise
        raise

배치 처리로 Rate Limit 우회
batch_size = 10
for i in range(0, len(inputs), batch_size):
    batch = inputs[i:i+batch_size]
    for item in batch:
        result = safe_llm_call(chain, item)
    time.sleep(1)  # 배치 간 딜레이

오류 3: "TokenUsageTracking: Missing token_usage in response"

# 토큰 사용량 추적 실패 해결
class RobustTokenTrackingHandler(BaseCallbackHandler):
    """토큰 사용량을 안정적으로 추적하는 핸들러"""
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        # 방법 1: 표준 token_usage 필드 확인
        if (response.llm_output and 
            "token_usage" in response.llm_output):
            return response.llm_output["token_usage"]
            
        # 방법 2: 응답 메타데이터에서 추출
        if response.llm_output and "model_name" in response.llm_output:
            model = response.llm_output["model_name"]
            # 모델별 평균 토큰估算
            prompt_text = response.prompts[0] if response.prompts else ""
            estimated_prompt = len(prompt_text) // 4
            estimated_completion = len(response.generations[0][0].text) // 4
            return {
                "prompt_tokens": estimated_prompt,
                "completion_tokens": estimated_completion,
                "total_tokens": estimated_prompt + estimated_completion
            }
            
        # 방법 3: 토큰 카운팅 API 사용
        # HolySheep AI 대시보드에서 직접 확인
        return {"source": "dashboard"}

오류 4: "CallbackHandler not firing events"

# Callbacks이 발동하지 않는 경우
❌ 잘못: 체인에 callbacks 미지정
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"text": "hello"})  # 콜백 작동 안함

✅ 올바른 방법 1: 체인에 직접 전달
chain = prompt | llm | StrOutputParser()
chain_with_callbacks = chain.with_config({
    "callbacks": [callback_handler]
})
result = chain_with_callbacks.invoke({"text": "hello"})

✅ 올바른 방법 2: LLM 초기화 시 전달
llm = ChatOpenAI(
    callbacks=[callback_handler],
    model="gpt-4.1"
)

✅ 올바른 방법 3: invoke config 사용
result = chain.invoke(
    {"text": "hello"},
    config={"callbacks": [callback_handler]}
)

결론

LangChain Callbacks 시스템은 LLM 애플리케이션의 관측 가능성을 크게 높이는 핵심 기능입니다. HolySheep AI와 결합하면 단일 API 키로 여러 모델을 일관된 방식으로 모니터링하면서 동시에 비용도 최적화할 수 있습니다. 저는 특히 한국 개발자 친화적인 결제 시스템 덕분에 해외 신용카드 걱정 없이 AI 개발에 집중할 수 있게 되었습니다.

궁금한 점이 있으시면 언제든지 댓글로 질문해 주세요. Happy coding!

👉 HolySheep AI 가입하고 무료 크레딧 받기