LangChain Callback 메커니즘으로 AI 앱 모니터링하기: 완전 초보자 가이드

들어가며: 왜 콜백 모니터링이 중요한가요?

AI 앱을 만들 때 가장 큰 고민 중 하나는 "내가 보낸 요청이 제대로 처리되고 있을까?"입니다. LangChain의 Callback 메커니즘은 이 문제를 깔끔하게 해결해줍니다. 저는 HolySheep AI로 여러 AI 앱을 개발하면서 이 콜백 시스템의 강력함을 직접 체감했습니다.

기본 개념을 쉽게 설명하면: Callback은 AI 모델이 응답을 생성하는 과정을 '감시하는 감시자'입니다. 요청이 시작되었을 때, 토큰이 생성될 때, 에러가 발생했을 때마다 알려주는 일종의 알림 시스템이라고 생각하시면 됩니다.

1단계: 필요한 도구 설치하기

먼저 파이썬 환경에서 필요한 패키지를 설치합니다. 터미널(명령 프롬프트)을 열고 다음 명령어를 입력하세요:

# 기본 LangChain 패키지 설치
pip install langchain langchain-openai langchain-core

콜백 모니터링에 필요한 패키지
pip install langchain-callbacks

실시간 스트리밍 확인용
pip install sseclient-py

설치 중红色的 오류 메시지가 보이면, 먼저 python --version으로 파이썬 버전을 확인하세요. 파이썬 3.8 이상이어야 합니다.

2단계: HolySheep AI API 키 설정하기

HolySheep AI에서 API 키를 발급받아야 합니다. 지금 가입하면 무료 크레딧을 받을 수 있어서 테스트하기 매우 좋습니다. 가입 후 대시보드에서 API 키를 복사하세요.

환경변수로 API 키를 설정하는 방법을 알려드립니다:

# Windows의 경우
set HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY

Mac/Linux의 경우
export HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY

또는 .env 파일을 만들어서 관리
.env 파일 내용:
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY

3단계: 기본 Callback 핸들러 만들기

이제 실제 코드를 작성해봅시다. 가장 기본이 되는 Custom Callback Handler를 만드는 예제입니다:

import os
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_openai import ChatOpenAI
from datetime import datetime

class MyCallbackHandler(BaseCallbackHandler):
    """나만의 모니터링 콜백 핸들러"""
    
    def __init__(self):
        self.start_time = None
        self.token_count = 0
        self.request_count = 0
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        """AI 모델이 응답 생성을 시작할 때"""
        self.start_time = datetime.now()
        self.request_count += 1
        print(f"📤 요청 #{self.request_count} 시작!")
        print(f"   질문: {prompts[0][:50]}...")
        print(f"   시간: {self.start_time.strftime('%H:%M:%S')}")
    
    def on_llm_new_token(self, token, **kwargs):
        """토큰이 하나씩 생성될 때 (스트리밍)"""
        self.token_count += 1
        # 매 10번째 토큰마다 표시 (너무 많으면 화면이 복잡해져서)
        if self.token_count % 10 == 0:
            print(f"   📝 토큰 생성 중... ({self.token_count}개)")
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        """AI 모델이 응답을 완료할 때"""
        end_time = datetime.now()
        duration = (end_time - self.start_time).total_seconds()
        
        # 생성된 토큰 수 계산
        total_tokens = 0
        if response.llm_output:
            total_tokens = response.llm_output.get('token_usage', {}).get('total_tokens', 0)
        
        print(f"✅ 응답 완료!")
        print(f"   소요 시간: {duration:.2f}초")
        print(f"   총 토큰 수: {total_tokens}개")
        print(f"   종료 시간: {end_time.strftime('%H:%M:%S')}")
    
    def on_llm_error(self, error, **kwargs):
        """에러가 발생했을 때"""
        print(f"❌ 오류 발생: {str(error)[:100]}")

HolySheep AI를 사용한 ChatOpenAI 설정
api_key = os.getenv("HOLYSHEEP_API_KEY")

llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url="https://api.holysheep.ai/v1",  # 반드시 HolySheep 주소 사용
    streaming=True,
    callbacks=[MyCallbackHandler()]
)

테스트 실행
response = llm.invoke("안녕하세요! 간단한 자기소개를 해주세요.")
print(f"\n최종 응답:\n{response.content}")

이 코드를 실행하면 요청 시작, 토큰 생성 과정, 응답 완료까지 모든 단계를 실시간으로 확인할 수 있습니다.

4단계: 파일로 기록하는 모니터링 시스템

콘솔 출력만으로는 나중에 분석하기 어렵습니다. 파일로 기록하는 시스템을 만들어보겠습니다:

import json
import os
from datetime import datetime
from typing import Any, Dict, List
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_openai import ChatOpenAI

class FileLoggingCallback(BaseCallbackHandler):
    """모든 요청을 JSON 파일로 기록하는 핸들러"""
    
    def __init__(self, log_dir: str = "logs"):
        self.log_dir = log_dir
        self.log_file = os.path.join(log_dir, f"llm_logs_{datetime.now().strftime('%Y%m%d')}.jsonl")
        os.makedirs(log_dir, exist_ok=True)
        
        # 현재 세션의 로그를 메모리에 저장
        self.session_logs: List[Dict[str, Any]] = []
    
    def _save_log(self, log_entry: Dict[str, Any]):
        """로그를 파일에 저장"""
        with open(self.log_file, 'a', encoding='utf-8') as f:
            f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        self.current_log = {
            "event": "start",
            "timestamp": datetime.now().isoformat(),
            "model": serialized.get("name", "unknown"),
            "prompt_preview": prompts[0][:100] if prompts else "",
            "status": "running"
        }
    
    def on_llm_new_token(self, token, **kwargs):
        if hasattr(self, 'current_log'):
            if 'tokens' not in self.current_log:
                self.current_log['tokens'] = []
            self.current_log['tokens'].append(token)
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        if hasattr(self, 'current_log'):
            self.current_log["event"] = "complete"
            self.current_log["status"] = "success"
            
            if response.llm_output:
                self.current_log["token_usage"] = response.llm_output.get('token_usage', {})
                
            # 응답 시간 계산
            if "timestamp" in self.current_log:
                start = datetime.fromisoformat(self.current_log["timestamp"])
                self.current_log["duration_seconds"] = (datetime.now() - start).total_seconds()
            
            # 첫 번째 생성 결과 저장
            if response.generations and response.generations[0]:
                first_gen = response.generations[0][0]
                self.current_log["response_preview"] = first_gen.text[:200]
            
            self._save_log(self.current_log)
            print(f"📁 로그 저장 완료: {self.log_file}")
    
    def on_llm_error(self, error, **kwargs):
        if hasattr(self, 'current_log'):
            self.current_log["event"] = "error"
            self.current_log["status"] = "failed"
            self.current_log["error_message"] = str(error)
            self._save_log(self.current_log)

HolySheep AI 설정
api_key = os.getenv("HOLYSHEEP_API_KEY")

llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url="https://api.holysheep.ai/v1",
    callbacks=[FileLoggingCallback()]
)

여러 요청 테스트
test_questions = [
    "파이썬에서 리스트와 튜플의 차이는?",
    "HTTP GET과 POST 메서드의 차이점은?",
    "Git에서 브랜치를 만드는 방법을 알려줘"
]

for i, question in enumerate(test_questions, 1):
    print(f"\n--- 요청 {i}/3 ---")
    response = llm.invoke(question)
    print(f"응답 미리보기: {response.content[:100]}...")

이 코드를 실행하면 logs/llm_logs_20240101.jsonl 같은 파일이 생성되고, 모든 요청의 상세 정보가 저장됩니다.

5단계: 비용 추적 모니터링하기

저는 HolySheep AI를 사용할 때 매달 비용이 얼마나 나오는지 신경 쓰입니다. HolySheep AI의 가격을 정리하면:

• GPT-4.1: $8.00 / 1M 토큰 (약 0.0008센트/토큰)
• Claude Sonnet 4: $15.00 / 1M 토큰
• Gemini 2.5 Flash: $2.50 / 1M 토큰 (가장 경제적)
• DeepSeek V3: $0.42 / 1M 토큰 (초저렴)

이제 모델별 비용을 추적하는 콜백을 만들어보겠습니다:

import os
from datetime import datetime
from langchain_core.callbacks import BaseCallbackHandler
from langchain_openai import ChatOpenAI

HolySheep AI 가격表 (2024년 기준)
MODEL_PRICES = {
    "gpt-4.1": {"input": 2.50, "output": 10.00},  # $/1M tokens
    "gpt-4-turbo": {"input": 10.00, "output": 30.00},
    "claude-sonnet-4": {"input": 3.00, "output": 15.00},
    "gemini-2.5-flash": {"input": 0.40, "output": 2.50},  # HolySheep 특별가
    "deepseek-v3": {"input": 0.07, "output": 0.42},  # 초저렴
}

class CostTrackingCallback(BaseCallbackHandler):
    """비용 추적 전용 콜백 핸들러"""
    
    def __init__(self):
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        self.total_cost = 0.0
        self.request_details = []
    
    def get_model_price(self, model: str) -> dict:
        """모델 가격 정보 가져오기"""
        return MODEL_PRICES.get(model, {"input": 0, "output": 0})
    
    def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """토큰 사용량으로 비용 계산"""
        prices = self.get_model_price(model)
        input_cost = (input_tokens / 1_000_000) * prices["input"]
        output_cost = (output_tokens / 1_000_000) * prices["output"]
        return input_cost + output_cost
    
    def on_llm_end(self, response, **kwargs):
        if response.llm_output and response.generations:
            model = response.generations[0][0].generation_info.get('model_name', 'unknown')
            token_usage = response.llm_output.get('token_usage', {})
            
            input_tokens = token_usage.get('prompt_tokens', 0)
            output_tokens = token_usage.get('completion_tokens', 0)
            cost = self.calculate_cost(model, input_tokens, output_tokens)
            
            self.total_input_tokens += input_tokens
            self.total_output_tokens += output_tokens
            self.total_cost += cost
            
            detail = {
                "model": model,
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
                "cost_usd": cost,
                "timestamp": datetime.now().isoformat()
            }
            self.request_details.append(detail)
            
            print(f"💰 비용 발생: ${cost:.6f}")
            print(f"   입력: {input_tokens}토큰, 출력: {output_tokens}토큰")
    
    def print_summary(self):
        """전체 비용 요약 출력"""
        print("\n" + "="*50)
        print("📊 HolySheep AI 비용 요약")
        print("="*50)
        print(f"총 입력 토큰: {self.total_input_tokens:,}")
        print(f"총 출력 토큰: {self.total_output_tokens:,}")
        print(f"총 비용: ${self.total_cost:.6f}")
        print(f"총 요청 수: {len(self.request_details)}건")
        print("="*50)

HolySheep AI 설정
api_key = os.getenv("HOLYSHEEP_API_KEY")

여러 모델 테스트
models_to_test = [
    ("gpt-4.1", "안녕하세요!"),
    ("gemini-2.5-flash", "안녕하세요!"),
    ("deepseek-v3", "안녕하세요!")
]

cost_tracker = CostTrackingCallback()

for model_name, prompt in models_to_test:
    print(f"\n🔄 {model_name} 테스트 중...")
    
    llm = ChatOpenAI(
        model=model_name,
        api_key=api_key,
        base_url="https://api.holysheep.ai/v1",
        callbacks=[cost_tracker]
    )
    
    try:
        response = llm.invoke(prompt)
        print(f"   응답: {response.content[:50]}...")
    except Exception as e:
        print(f"   ❌ 에러: {e}")

cost_tracker.print_summary()

이 코드를 실행하면 각 모델의 토큰 사용량과 비용을 한눈에 비교할 수 있습니다. HolySheep AI에서는 DeepSeek V3가 가장 저렴해서 반복적인 태스크에 적합하고, GPT-4.1은 복잡한 reasoning 작업에 적합합니다.

6단계: 체인에서 Callback 사용하기

LangChain의 체인(Chain)에서도 콜백을 사용할 수 있습니다. 여러 단계를 거치는 복잡한 작업에서 각 단계별 모니터링이 가능합니다:

import os
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain

class ChainStepCallback(BaseCallbackHandler):
    """체인 내 각 단계별 모니터링"""
    
    def on_chain_start(self, serialized, inputs, **kwargs):
        chain_name = serialized.get('name', 'unknown')
        print(f"\n🔗 체인 시작: {chain_name}")
        print(f"   입력 키: {list(inputs.keys())}")
    
    def on_chain_end(self, outputs, **kwargs):
        print(f"✅ 체인 완료")
        print(f"   출력 키: {list(outputs.keys())}")
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        tool_name = serialized.get('name', 'unknown')
        print(f"🔧 도구 실행: {tool_name}")

HolySheep AI 설정
api_key = os.getenv("HOLYSHEEP_API_KEY")

llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url="https://api.holysheep.ai/v1"
)

단계 1: 질문 분석 프롬프트
analysis_prompt = PromptTemplate.from_template(
    "다음 질문의 핵심 주제를 3단어로 요약해줘: {question}"
)

단계 2: 답변 생성 프롬프트
answer_prompt = PromptTemplate.from_template(
    "주제 '{topic}'에 대해 2문장으로 설명해줘."
)

체인 생성
analysis_chain = LLMChain(llm=llm, prompt=analysis_prompt)
answer_chain = LLMChain(llm=llm, prompt=answer_prompt)

Combined Chain (간단한 시퀀스)
from langchain.chains import SimpleSequentialChain

combined_chain = SimpleSequentialChain(
    chains=[analysis_chain, answer_chain],
    verbose=True,
    callbacks=[ChainStepCallback()]
)

실행
result = combined_chain.run("인공지능이 미래에 어떻게 발전할까요?")
print(f"\n최종 결과:\n{result}")

자주 발생하는 오류 해결

오류 1: API 키 인증 실패

# ❌ 잘못된 예 - base_url이 HolySheep이 아님
llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url="https://api.openai.com/v1"  # ❌ 이것은 안 됨!
)

✅ 올바른 예
llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url="https://api.holysheep.ai/v1"  # ✅ HolySheep 주소 사용
)

환경변수에서 API 키를 못 찾는 경우
import os
api_key = os.environ.get("HOLYSHEEP_API_KEY")

if not api_key:
    print("⚠️ API 키가 설정되지 않았습니다!")
    print("터미널에서 다음 명령어를 실행하세요:")
    print("  Windows: set HOLYSHEEP_API_KEY=your_key_here")
    print("  Mac/Linux: export HOLYSHEEP_API_KEY=your_key_here")

해결 방법: HolySheep AI 대시보드에서 새로운 API 키를 발급받고, 반드시 https://api.holysheep.ai/v1을 base_url로 사용해야 합니다.

오류 2: 스트리밍与非스트리밍 충돌

# ❌ 콜백에 streaming=True가 없으면 오류 발생 가능
llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url="https://api.holysheep.ai/v1",
    # streaming 설정 없음
    callbacks=[MyCallbackHandler()]  # ❌ streaming 콜백에는 streaming=True 필요
)

✅ 올바른 예 - 스트리밍 콜백 사용 시
llm = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url="https://api.holysheep.ai/v1",
    streaming=True,  # ✅ 스트리밍 모드 활성화
    callbacks=[MyCallbackHandler()]  # 이제 on_llm_new_token이 동작함
)

또는 스트리밍 없이 사용
llm_no_stream = ChatOpenAI(
    model="gpt-4.1",
    api_key=api_key,
    base_url="https://api.holysheep.ai/v1",
    streaming=False,  # 스트리밍 비활성화
    callbacks=[MyCallbackHandler()]
)

해결 방법: on_llm_new_token 콜백을 사용하려면 반드시 streaming=True를 설정하세요. 스트리밍이 필요 없으면 streaming=False로 설정하세요.

오류 3: 토큰 사용량 정보 누락

# ❌ LLMResult에서 토큰 정보를 못 가져오는 경우
def on_llm_end(self, response, **kwargs):
    # response.llm_output이 None일 수 있음
    if response.llm_output:
        tokens = response.llm_output.get('token_usage', {})
    else:
        print("⚠️ 토큰 정보 없음")

✅ 안전한 접근 방법
def on_llm_end(self, response, **kwargs):
    try:
        if response.llm_output:
            token_usage = response.llm_output.get('token_usage', {})
            input_tokens = token_usage.get('prompt_tokens', 0)
            output_tokens = token_usage.get('completion_tokens', 0)
            total_tokens = token_usage.get('total_tokens', input_tokens + output_tokens)
        else:
            # 대체 방법: generations에서 추측
            total_tokens = 0
            for gen_list in response.generations:
                for gen in gen_list:
                    if hasattr(gen, 'generation_info'):
                        # 대략적인 토큰 수 (문자 길이 / 4)
                        total_tokens += len(gen.text) // 4
            print(f"⚠️ API 토큰 정보 없음, 추정: {total_tokens}")
    except Exception as e:
        print(f"토큰 계산 오류: {e}")

해결 방법: 일부 모델은 토큰 정보를 반환하지 않을 수 있습니다. 항상 response.llm_output이 None인지 확인하고, 없으면 대안 계산 방법을 사용하세요.

오류 4: 비동기 콜백 혼용 문제

# ❌ 비동기 콜백을 동기 컨텍스트에서 사용
import asyncio

async def async_handler():
    return "async result"

❌ 이렇게 혼용하면 오류 발생
handler = async_handler()
result = handler  # 이것은 코루틴 객체일 뿐

✅ 올바른 방법 1: async/await 사용
async def run_async_example():
    from langchain_core.callbacks import AsyncCallbackHandler
    
    class AsyncLoggingHandler(AsyncCallbackHandler):
        async def on_llm_end(self, response, **kwargs):
            print("비동기 로그 기록")
            await asyncio.sleep(0.1)  # 비동기 작업
    
    llm = ChatOpenAI(
        model="gpt-4.1",
        api_key=api_key,
        base_url="https://api.holysheep.ai/v1",
        callbacks=[AsyncLoggingHandler()]
    )
    
    await llm.ainvoke("테스트")

✅ 올바른 방법 2: 동기 코드에서는 async-chain을 사용하지 않기
위에서 만든 MyCallbackHandler처럼 BaseCallbackHandler를 상속받는 동기 핸들러 사용
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[MyCallbackHandler()])
result = chain.invoke({"input": "테스트"})  # ✅ 동기 호출

해결 방법: 동기 코드(.invoke(), .run())에서는 동기 콜백 핸들러를, 비동기 코드(.ainvoke())에서는 AsyncCallbackHandler를 사용하세요.

실전 활용 팁

저는 HolySheep AI로 프로덕션 앱을 운영하면서 이런 패턴을 주로 사용합니다:

• 분 단위 로깅: 매 요청마다 파일로 기록하고, 하루가 끝나면 비용 합계를 계산해서 HolySheep AI 대시보드와 비교합니다.
• 에러 알림: on_llm_error에서 Slack이나 이메일로 알림을 보내서 밤에도 앱 상태를 모니터링합니다.
• 성능 벤치마크: HolySheep AI에서 여러 모델의 응답 속도를 비교해서 최적의 모델을 선택합니다.
• 토큰 최적화: 비용 추적 결과를 바탕으로 프롬프트를 최적화해서 불필요한 토큰 사용을 줄입니다.

정리하며

LangChain의 Callback 메커니즘은 AI 앱 개발에서 빼놓을 수 없는 중요한 기능입니다. 요청 추적, 비용 관리, 에러 모니터링 등 다양한 용도로 활용할 수 있습니다. HolySheep AI를 사용하면 글로벌 모델들을 단일 API로 편하게 관리할 수 있어서 콜백 구현도 훨씬 수월해집니다.

특히 HolySheep AI의 DeepSeek V3 모델은 토큰당 $0.00000042로 엄청 저렴해서 대량 요청 작업에 최적화되어 있고, Gemini 2.5 Flash는 가격과 성능의 균형이 좋아서日常적인 작업에 적합합니다. 자신의 용도에 맞는 모델을 골라서 콜백 모니터링과 함께 효율적으로 운영해보세요.

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