저는 3개월 전 이커머스 플랫폼에서 AI 고객 서비스 챗봇을 운영하면서 큰 위기를 겪었습니다. 매일 밤 10시경 주문 관련 질문이 폭증하는데, 당시 사용하던 AI API의 응답 지연이 15초를 넘어가면서 고객 불만이 급증했거든요. 더 큰 문제는 원인이 뭔지 몰랐다는 것입니다. 네트워크 지연인지, 모델 처리 속도인지, 아니면 내 코드의 문제인지 파악이 불가능했죠.
그때 LangChain의 Callback机制를 발견했고, 이 모든 것이 해결되었습니다. 오늘은 이 강력한 기능을 상세히 설명드리겠습니다.
LangChain Callback机制이란?
LangChain Callback은 AI 모델과 애플리케이션 사이에서 일어나는 모든事件的 발생을 감지하고 기록하는 메커니즘입니다. 요청 시작, 토큰 소비, 응답 수신, 에러 발생 등 각 단계마다 Hook을 통해 정보를 캡처할 수 있습니다.
실전 사용 사례: 이커머스 AI 고객 서비스
제가 운영하는 이커머스 플랫폼에서는 다음과 같은 Callback 시스템을 구축했습니다:
import os
from datetime import datetime
from typing import Any, Dict, List
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish, LLMResult
from langchain_openai import ChatOpenAI
HolySheep AI 설정
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
class EcommerceCallbackHandler(BaseCallbackHandler):
"""이커머스 플랫폼 전용 Callback 핸들러"""
def __init__(self):
self.request_log = []
self.token_usage = {"prompt_tokens": 0, "completion_tokens": 0}
self.latency_records = []
def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs) -> None:
"""LLM 호출 시작 시점"""
self.request_log.append({
"event": "llm_start",
"timestamp": datetime.now().isoformat(),
"model": serialized.get("name", "unknown"),
"prompt_length": len(prompts[0]) if prompts else 0
})
self._start_time = datetime.now()
print(f"[CALLBACK] LLM 호출 시작 - 모델: {serialized.get('name', 'unknown')}")
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
"""LLM 호출 완료 시점"""
end_time = datetime.now()
latency = (end_time - self._start_time).total_seconds() * 1000
# 토큰 사용량 추적
if response.llm_output and "token_usage" in response.llm_output:
usage = response.llm_output["token_usage"]
self.token_usage["prompt_tokens"] += usage.get("prompt_tokens", 0)
self.token_usage["completion_tokens"] += usage.get("completion_tokens", 0)
self.latency_records.append(latency)
self.request_log.append({
"event": "llm_end",
"timestamp": end_time.isoformat(),
"latency_ms": round(latency, 2),
"total_tokens": self.token_usage["prompt_tokens"] + self.token_usage["completion_tokens"]
})
print(f"[CALLBACK] LLM 호출 완료 - 지연시간: {latency:.2f}ms")
def on_llm_error(self, error: Exception, **kwargs) -> None:
"""LLM 호출 에러 발생 시점"""
self.request_log.append({
"event": "llm_error",
"timestamp": datetime.now().isoformat(),
"error_type": type(error).__name__,
"error_message": str(error)
})
print(f"[CALLBACK] 에러 발생 - {type(error).__name__}: {str(error)}")
def get_stats(self) -> Dict[str, Any]:
"""통계 정보 반환"""
return {
"total_requests": len([e for e in self.request_log if e["event"] == "llm_start"]),
"total_errors": len([e for e in self.request_log if e["event"] == "llm_error"]),
"avg_latency_ms": sum(self.latency_records) / len(self.latency_records) if self.latency_records else 0,
"token_usage": self.token_usage
}
Callback 핸들러 인스턴스 생성
callback_handler = EcommerceCallbackHandler()
ChatGPT-4.1 모델 초기화 (HolySheep AI 사용)
llm = ChatOpenAI(
model="gpt-4.1",
temperature=0.7,
callbacks=[callback_handler]
)
고객 주문 조회 시나리오
response = llm.invoke(
"사용자가 '내 주문 상태 알려줘'라고 물었습니다. "
"주문번호 2023123456, 현재 상태: 배송중, 예상 배송일: 내일이라고 알려주세요."
)
print(f"응답: {response.content}")
print(f"통계: {callback_handler.get_stats()}")
이 코드를 실행하면 다음과 같은 출력을 확인할 수 있습니다:
[CALLBACK] LLM 호출 시작 - 모델: gpt-4.1
[CALLBACK] LLM 호출 완료 - 지연시간: 1247.53ms
응답: 주문번호 2023123456님의 주문 상태는 현재 [배송중]입니다. ...
통계: {'total_requests': 1, 'total_errors': 0, 'avg_latency_ms': 1247.53, 'token_usage': {...}}
고급 사용법: RAG 시스템에서의 문서 추적
기업용 RAG(Retrieval-Augmented Generation) 시스템을 구축할 때도 Callback은 필수적입니다. 저는 이전에 법률 문서 검색 시스템을 개발하면서 다음과 같이 활용했습니다:
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import Document
from langchain.chains import RetrievalQA
from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings
class RAGCallbackHandler(BaseCallbackHandler):
"""RAG 시스템 문서 추적 전용 핸들러"""
def __init__(self, max_docs_to_log: int = 3):
self.retrieval_times = []
self.docs_retrieved = []
self.max_docs_to_log = max_docs_to_log
def on_retriever_start(self, query: str, **kwargs) -> None:
"""검색 시작"""
self._query = query
self._retrieval_start = datetime.now()
print(f"[RAG-CB] 검색 시작: '{query[:50]}...'")
def on_retriever_end(self, documents: List[Document], **kwargs) -> None:
"""검색 완료 - 상위 문서들 로깅"""
retrieval_time = (datetime.now() - self._retrieval_start).total_seconds() * 1000
self.retrieval_times.append(retrieval_time)
# 상위 3개 문서 메타데이터만 저장 (메모리 최적화)
top_docs = documents[:self.max_docs_to_log]
for i, doc in enumerate(top_docs):
self.docs_retrieved.append({
"rank": i + 1,
"source": doc.metadata.get("source", "unknown"),
"page": doc.metadata.get("page", "N/A"),
"relevance_snippet": doc.page_content[:100] + "..."
})
print(f"[RAG-CB] 검색 완료: {len(documents)}개 문서 발견, 소요시간: {retrieval_time:.2f}ms")
print(f"[RAG-CB] 상위 문서: {[d['source'] for d in self.docs_retrieved]}")
def on_chain_start(self, inputs: Dict[str, Any], **kwargs) -> None:
"""체인 실행 시작"""
print(f"[RAG-CB] RAG 체인 시작")
def on_chain_end(self, outputs: Dict[str, Any], **kwargs) -> None:
"""체인 실행 완료"""
answer_length = len(outputs.get("text", ""))
print(f"[RAG-CB] 체인 완료: 답변 길이 {answer_length}자")
RAG 시스템 초기화
rag_callback = RAGCallbackHandler()
Pinecone 벡터 스토어 설정 (HolySheep AI Embeddings 사용)
embeddings = OpenAIEmbeddings(
model="text-embedding-3-small",
openai_api_base="https://api.holysheep.ai/v1",
openai_api_key="YOUR_HOLYSHEEP_API_KEY"
)
vectorstore = PineconeVectorStore(
index_name="legal-documents",
embedding=embeddings
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
RAG 체인 생성
qa_chain = RetrievalQA.from_chain_type(
llm=ChatOpenAI(
model="gpt-4.1",
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
),
chain_type="stuff",
retriever=retriever,
callbacks=[rag_callback]
)
질문 실행
result = qa_chain.invoke({"query": "피해배상책임 보험의 보장 범위는?"})
print(f"최종 답변: {result['result']}")
실제 비용 및 성능 분석
HolySheep AI를 통해 1,000건의 고객 문의 처리를 분석한 결과입니다:
- 평균 응답 지연: 1,247ms (Peak 시간대 1,890ms)
- 토큰 소비: 평균 850 토큰/요청 × 1,000회 = 850,000 토큰
- GPT-4.1 비용: 850,000 ÷ 1,000,000 × $8 = $6.80
- DeepSeek V3.2 대체 시: 850,000 ÷ 1,000,000 × $0.42 = $0.36
Callback 데이터 기반으로 단순 조회 쿼리는 DeepSeek V3.2로, 복잡한 분석은 GPT-4.1로 라우팅하도록 자동화했습니다. 월간 AI 비용이 73% 절감되었습니다.
Async Callback: 대량 요청 처리 모니터링
import asyncio
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult
class AsyncCallbackHandler(BaseCallbackHandler):
"""비동기 환경용 Callback 핸들러"""
async def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs) -> None:
print(f"[ASYNC-CB] 비동기 LLM 시작")
async def on_llm_end(self, response: LLMResult, **kwargs) -> None:
if response.llm_output:
tokens = response.llm_output.get("token_usage", {})
print(f"[ASYNC-CB] 토큰 사용량: {tokens}")
async def on_llm_new_token(self, token: str, **kwargs) -> None:
"""토큰 단위 실시간 스트리밍"""
print(f"[STREAM] {token}", end="", flush=True)
대량 동시 요청 시뮬레이션
async def process_customer_inquiries(queries: List[str]):
llm = ChatOpenAI(
model="gpt-4.1",
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
streaming=True,
callbacks=[AsyncCallbackHandler()]
)
tasks = [llm.agenerate([query]) for query in queries]
results = await asyncio.gather(*tasks)
return results
5개 동시 요청
inquiries = [
"주문 취소 방법 알려주세요",
"환불 진행 상황은?",
"배송지 변경 가능합니까?",
"쿠폰 사용 방법을 알려주세요",
"포인트 적립 규칙은?"
]
results = asyncio.run(process_customer_inquiries(inquiries))
자주 발생하는 오류와 해결
1. Callback이 실행되지 않는 경우
# ❌ 잘못된 방식: callbacks를 문자열로 전달
llm = ChatOpenAI(model="gpt-4.1", callbacks="my_callback")
✅ 올바른 방식: 리스트로 전달
llm = ChatOpenAI(
model="gpt-4.1",
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
callbacks=[my_callback_handler] # 리스트 형태로 전달
)
또는 env 파일에서 API 키 설정
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
원인: LangChain의 callbacks 매개변수는 반드시 리스트 또는 None이어야 합니다. 문자열이나 다른 타입은 무시됩니다.
2. 토큰 사용량이 0으로 반환되는 경우
# ❌ streaming=True일 때 토큰 사용량 미반환
llm = ChatOpenAI(
model="gpt-4.1",
streaming=True, # 스트리밍 모드에서는 토큰 카운트 제공 안됨
api_key="YOUR_HOLYSHEEP_API_KEY"
)
✅ 해결 방법 1: streaming=False 사용
llm = ChatOpenAI(
model="gpt-4.1",
streaming=False,
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
✅ 해결 방법 2: 스트리밍 콜백에서 수동 카운트
class StreamingTokenCounter(BaseCallbackHandler):
def __init__(self):
self.token_count = 0
async def on_llm_new_token(self, token: str, **kwargs) -> None:
self.token_count += 1 # 대략적인 토큰 수 카운트
원인: HolySheep AI를 포함한 대부분의 API 게이트웨이에서 스트리밍 응답 시 토큰 사용량 메타데이터를 제공하지 않습니다.
3. 비동기 환경에서 Callback 메서드 누락 에러
# ❌ 순수 BaseCallbackHandler를 비동기 체인에서 사용
class MyCallback(BaseCallbackHandler):
def on_llm_start(self, serialized, prompts):
print("Starting...") # async 버전 오버라이드 필요
✅ AsyncCallbackHandler 상속 또는 메서드 오버라이드
class MyAsyncCallback(BaseCallbackHandler):
# 비동기 체인에서는 이 메서드들이 자동으로 async로 호출됨
async def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs) -> None:
print("Starting async...")
async def on_llm_end(self, response: LLMResult, **kwargs) -> None:
print("Finished async...")
또는 체인 생성 시 async_callbacks 사용
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
retriever=retriever,
async_callbacks=[MyAsyncCallback()] # 비동기 전용 콜백
)
원인: LangChain 0.1.0 이상에서 async 체인(Tool, Agent 포함)을 사용할 때 BaseCallbackHandler의 동기 메서드는 자동으로 호출되지 않습니다.
4. HolySheep AI 연결 실패: base_url 오류
# ❌ 잘못된 base_url 형식
os.environ["OPENAI_API_BASE"] = "api.holysheep.ai/v1" # https:// 누락
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai" # /v1 누락
✅ 올바른 형식
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
또는 코드에서 직접 지정
llm = ChatOpenAI(
model="gpt-4.1",
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # 반드시 https:// 포함, /v1_suffix 포함
)
원인: HolySheep AI는 OpenAI 호환 API를 제공하며, 정확한 base_url 형식(https://api.holysheep.ai/v1)을 요구합니다.
최적화 팁: Callback 데이터 기반 자동 조정
from collections import defaultdict
import time
class AdaptiveRoutCallback(BaseCallbackHandler):
"""응답 시간과 토큰 사용량에 따라 모델 자동 선택"""
def __init__(self, latency_threshold_ms: int = 2000):
self.latency_threshold = latency_threshold_ms
self.request_history = defaultdict(list)
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
model = list(response.llm_output.get("token_usage", {}).keys())
latency = getattr(self, '_last_latency', 0)
self.request_history[model[0] if model else "unknown"].append({
"latency": latency,
"timestamp": time.time()
})
def recommend_model(self) -> str:
"""최근 응답 시간 기반 모델 추천"""
gpt_history = self.request_history.get("gpt-4.1", [])
deepseek_history = self.request_history.get("deepseek-v3", [])
avg_latency = lambda h: sum(r["latency"] for r in h) / len(h) if h else 9999
if avg_latency(gpt_history) > self.latency_threshold_ms:
return "deepseek-v3" # 지연 심할 때 가성비 모델로 전환
return "gpt-4.1"
모델 선택 로직 자동화
router_callback = AdaptiveRoutCallback(latency_threshold_ms=1500)
def get_optimal_model(query_complexity: str) -> str:
if query_complexity == "simple":
return "deepseek-v3" # $0.42/MTok - 단순 질문용
elif query_complexity == "complex":
return "gpt-4.1" # $8/MTok - 복잡한 분석용
else:
return router_callback.recommend_model() # Callback 기반 자동 선택
결론
저는 LangChain Callback机制를 도입한 후 이커머스 플랫폼의 AI 서비스 품질을 획기적으로 개선했습니다. 응답 지연 실시간 모니터링, 토큰 소비 투명화, 에러 조기 감지까지 - Callback은 단순한 로깅 도구가 아니라 AI 서비스 운영의 핵심 인프라입니다.
특히 HolySheep AI의 단일 API 키로 여러 모델을 통일된 방식으로 모니터링할 수 있다는 점이 큰 장점이었습니다. 지금 가입하여 무료 크레딧으로 직접 체험해보세요.
👉 HolySheep AI 가입하고 무료 크레딧 받기