LangChain Callback: Hướng Dẫn Toàn Diện Về Giám Sát và Ghi Log Cuộc Gọi LLM

Khi làm việc với các ứng dụng LLM trong production, việc giám sát chi phí, độ trễ và chất lượng phản hồi là yếu tố sống còn. Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến 3 năm với LangChain và cách tích hợp hệ thống callback mạnh mẽ để theo dõi mọi cuộc gọi API.

Callback Trong LangChain Là Gì?

LangChain cung cấp hệ thống callback cho phép bạn hook vào các sự kiện trong chuỗi xử lý: bắt đầu cuộc gọi, nhận token, hoàn thành, hoặc xảy ra lỗi. Điều này giúp ghi log chi tiết, tính toán chi phí thực tế, và phát hiện vấn đề sớm.

Kiến Trúc Callback Handler Tùy Chỉnh

Tôi đã xây dựng một handler toàn diện tích hợp với HolySheep AI để theo dõi chi phí và hiệu suất:

import os
import time
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from collections import defaultdict

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


@dataclass
class TokenUsage:
    """Theo dõi số token và chi phí cho mỗi model"""
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    cost: float = 0.0


@dataclass
class CallMetrics:
    """Metrics chi tiết cho mỗi cuộc gọi LLM"""
    start_time: float = 0.0
    end_time: float = 0.0
    latency_ms: float = 0.0
    success: bool = True
    error_message: Optional[str] = None
    model: str = ""
    usage: TokenUsage = field(default_factory=TokenUsage)


class HolySheepCallbackHandler(BaseCallbackHandler):
    """
    Handler callback tích hợp HolySheep AI với tracking chi phí chi tiết.
    Tích hợp sẵn bảng giá HolySheep 2026 (USD/MToken).
    """
    
    # Bảng giá HolySheep AI - thực tế từ dashboard
    HOLYSHEEP_PRICING = {
        "gpt-4.1": {"input": 8.0, "output": 8.0},
        "gpt-4.1-mini": {"input": 2.0, "output": 8.0},
        "gpt-4o": {"input": 2.50, "output": 10.0},
        "claude-sonnet-4-5": {"input": 3.0, "output": 15.0},
        "claude-3-5-sonnet": {"input": 3.0, "output": 15.0},
        "gemini-2.5-flash": {"input": 0.125, "output": 0.50},
        "deepseek-v3.2": {"input": 0.07, "output": 0.42},
    }
    
    # Pricing từ bảng điều khiển HolySheep
    def __init__(self):
        self.call_history: List[CallMetrics] = []
        self.total_cost = 0.0
        self.total_tokens = 0
        self.total_latency = 0.0
        self.success_count = 0
        self.failure_count = 0
        self._current_call: Optional[CallMetrics] = None
        self._streaming_tokens: List[str] = []
        
    def _calculate_cost(self, model: str, usage: TokenUsage) -> float:
        """Tính chi phí theo bảng giá HolySheep (USD/MTok)"""
        model_lower = model.lower()
        for model_key, prices in self.HOLYSHEEP_PRICING.items():
            if model_key in model_lower:
                input_cost = (usage.prompt_tokens / 1_000_000) * prices["input"]
                output_cost = (usage.completion_tokens / 1_000_000) * prices["output"]
                return input_cost + output_cost
        # Mặc định cho model không có trong bảng giá
        return (usage.total_tokens / 1_000_000) * 10.0
    
    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs
    ) -> None:
        """Bắt đầu cuộc gọi LLM - ghi nhận thời gian bắt đầu"""
        model = kwargs.get("invocation_params", {}).get("model", "unknown")
        self._current_call = CallMetrics(
            start_time=time.time(),
            model=model
        )
        self._streaming_tokens = []
        print(f"[Callback] 🚀 Bắt đầu cuộc gọi LLM: {model}")
    
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        """Mỗi khi có token mới được sinh ra (streaming)"""
        self._streaming_tokens.append(token)
    
    def on_llm_end(self, response: LLMResult, **kwargs) -> None:
        """Hoàn thành cuộc gọi LLM - tính toán chi phí và latency"""
        if self._current_call is None:
            return
            
        self._current_call.end_time = time.time()
        self._current_call.latency_ms = (
            self._current_call.end_time - self._current_call.start_time
        ) * 1000
        
        # Trích xuất usage từ response
        if response.llm_output and "token_usage" in response.llm_output:
            usage = response.llm_output["token_usage"]
            self._current_call.usage = TokenUsage(
                prompt_tokens=usage.get("prompt_tokens", 0),
                completion_tokens=usage.get("completion_tokens", 0),
                total_tokens=usage.get("total_tokens", 0)
            )
            # Tính chi phí
            self._current_call.usage.cost = self._calculate_cost(
                self._current_call.model,
                self._current_call.usage
            )
        
        # Cập nhật stats tổng
        self.total_cost += self._current_call.usage.cost
        self.total_tokens += self._current_call.usage.total_tokens
        self.total_latency += self._current_call.latency_ms
        self.success_count += 1
        
        # Log chi tiết
        print(f"[Callback] ✅ Hoàn thành: {self._current_call.model}")
        print(f"[Callback] ⏱️  Latency: {self._current_call.latency_ms:.2f}ms")
        print(f"[Callback] 💰 Chi phí: ${self._current_call.usage.cost:.6f}")
        print(f"[Callback] 📊 Tokens: {self._current_call.usage.total_tokens:,}")
        
        self.call_history.append(self._current_call)
        self._current_call = None
    
    def on_llm_error(self, error: Exception, **kwargs) -> None:
        """Xử lý khi có lỗi xảy ra"""
        if self._current_call:
            self._current_call.success = False
            self._current_call.error_message = str(error)
            self.failure_count += 1
            print(f"[Callback] ❌ Lỗi: {error}")
            self.call_history.append(self._current_call)
            self._current_call = None
    
    def get_summary(self) -> Dict[str, Any]:
        """Trả về tổng kết tất cả metrics"""
        total_calls = self.success_count + self.failure_count
        avg_latency = self.total_latency / self.success_count if self.success_count > 0 else 0
        
        return {
            "total_calls": total_calls,
            "success_count": self.success_count,
            "failure_count": self.failure_count,
            "success_rate": self.success_count / total_calls if total_calls > 0 else 0,
            "total_cost_usd": round(self.total_cost, 6),
            "total_tokens": self.total_tokens,
            "avg_latency_ms": round(avg_latency, 2),
            "cost_per_1k_tokens": round(
                (self.total_cost / self.total_tokens * 1000) if self.total_tokens > 0 else 0, 6
            )
        }


Khởi tạo handler toàn cục
callback_handler = HolySheepCallbackHandler()
print("[Init] HolySheep Callback Handler đã khởi tạo thành công")

Tích Hợp HolySheep AI - Cấu Hình Không Thể Thiếu

Điều quan trọng nhất: cấu hình endpoint chính xác. HolySheep AI cung cấp latency trung bình <50ms và tỷ giá ¥1 = $1 giúp tiết kiệm đến 85% chi phí so với API gốc.

import os
from langchain_openai import ChatOpenAI
from langchain.callbacks.manager import CallbackManager


===== CẤU HÌNH HOLYSHEEP AI =====
⚠️ QUAN TRỌNG: Không bao giờ dùng api.openai.com hoặc api.anthropic.com

os.environ["OPENAI_API_KEY"] = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

Model mapping - chọn model phù hợp với budget
MODEL_CONFIG = {
    "cheap": "deepseek-v3.2",      # $0.42/MTok output - rẻ nhất
    "balanced": "gemini-2.5-flash", # $2.50/MTok - cân bằng
    "quality": "claude-sonnet-4-5", # $15/MTok - chất lượng cao
}

def create_llm(
    model: str = "gpt-4.1",
    temperature: float = 0.7,
    streaming: bool = True
) -> ChatOpenAI:
    """Factory function tạo LLM với HolySheep endpoint"""
    
    return ChatOpenAI(
        # Endpoint HolySheep AI - bắt buộc phải đúng format
        base_url="https://api.holysheep.ai/v1",
        
        # Model từ bảng giá HolySheep
        model=model,
        
        # Temperature kiểm soát creativity
        temperature=temperature,
        
        # Streaming để nhận token nhanh hơn
        streaming=streaming,
        
        # Gắn callback handler để giám sát
        callback_manager=CallbackManager([callback_handler]),
        
        # Timeout hợp lý cho production
        request_timeout=60,
        
        # Retry policy
        max_retries=3,
        max_concurrency=10,
    )


===== DEMO: So sánh chi phí thực tế =====

def demo_cost_comparison():
    """So sánh chi phí giữa các provider thực tế"""
    
    print("=" * 60)
    print("📊 SO SÁNH CHI PHÍ HOLYSHEEP vs OPENAI")
    print("=" * 60)
    
    # Giả định: 1 triệu token input + 500k token output
    test_input_tokens = 1_000_000
    test_output_tokens = 500_000
    
    providers = {
        "OpenAI GPT-4.1": {
            "input_price": 8.0,
            "output_price": 8.0,
            "input_tokens": test_input_tokens,
            "output_tokens": test_output_tokens,
        },
        "Claude Sonnet 4.5 (OpenAI)": {
            "input_price": 15.0,
            "output_price": 15.0,
            "input_tokens": test_input_tokens,
            "output_tokens": test_output_tokens,
        },
        "HolySheep GPT-4.1": {
            "input_price": 8.0,
            "output_price": 8.0,
            "input_tokens": test_input_tokens,
            "output_tokens": test_output_tokens,
        },
        "HolySheep DeepSeek V3.2": {
            "input_price": 0.07,
            "output_price": 0.42,
            "input_tokens": test_input_tokens,
            "output_tokens": test_output_tokens,
        },
        "HolySheep Gemini 2.5 Flash": {
            "input_price": 0.125,
            "output_price": 0.50,
            "input_tokens": test_input_tokens,
            "output_tokens": test_output_tokens,
        },
    }
    
    results = []
    for provider, data in providers.items():
        cost = (
            (data["input_tokens"] / 1_000_000) * data["input_price"] +
            (data["output_tokens"] / 1_000_000) * data["output_price"]
        )
        results.append((provider, cost))
    
    results.sort(key=lambda x: x[1])
    
    for i, (provider, cost) in enumerate(results):
        marker = "🏆 TIẾT KIỆM NHẤT" if i == 0 else ""
        print(f"{i+1}. {provider}: ${cost:.2f} {marker}")
    
    baseline = results[0][1]
    for provider, cost in results[1:]:
        savings = ((cost - baseline) / cost) * 100
        print(f"   → {provider} đắt hơn {savings:.1f}%")


if __name__ == "__main__":
    # Chạy demo
    demo_cost_comparison()
    
    # Tạo LLM instance
    llm = create_llm(model="deepseek-v3.2", temperature=0.5)
    print(f"\n✅ LLM đã sẵn sàng với base_url: {llm.bindings.base_url}")

Ứng Dụng Thực Tế - Theo Dõi Chains và Agents

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.chains import LLMChain, ConversationalRetrievalChain
from langchain.schema import HumanMessage, SystemMessage
import json


def demo_chain_with_callbacks():
    """Demo hoàn chỉnh chain với callback tracking"""
    
    # System prompt cho task cụ thể
    system_template = """Bạn là trợ lý AI chuyên phân tích dữ liệu.
    Trả lời ngắn gọn, chính xác và có cấu trúc."""
    
    prompt = ChatPromptTemplate.from_messages([
        SystemMessage(content=system_template),
        HumanMessage(content="Phân tích xu hướng: {topic}")
    ])
    
    # Tạo chain với callback
    llm = create_llm(model="gemini-2.5-flash", streaming=True)
    chain = LLMChain(
        llm=llm,
        prompt=prompt,
        output_parser=StrOutputParser(),
        verbose=False,  # Tắt verbose vì đã có callback
    )
    
    # Chạy chain với tracking
    topics = [
        "thị trường tiền điện tử Q1 2026",
        "xu hướng AI trong healthcare",
        "công nghệ blockchain cho supply chain",
    ]
    
    print("\n" + "=" * 60)
    print("🚀 CHẠY DEMO CHAIN VỚI CALLBACK TRACKING")
    print("=" * 60 + "\n")
    
    results = []
    for topic in topics:
        print(f"\n📌 Xử lý: {topic}")
        try:
            response = chain.invoke({"topic": topic})
            results.append({
                "topic": topic,
                "success": True,
                "response": response["text"][:100] + "..."
            })
        except Exception as e:
            results.append({
                "topic": topic,
                "success": False,
                "error": str(e)
            })
    
    # In tổng kết
    print("\n" + "=" * 60)
    print("📊 TỔNG KẾT CALLBACK")
    print("=" * 60)
    
    summary = callback_handler.get_summary()
    print(json.dumps(summary, indent=2, ensure_ascii=False))
    
    return results


def create_async_monitoring_chain():
    """Tạo chain với async callback cho production"""
    from langchain.callbacks.tracers.langchain import LangChainTracer
    from langchain.callbacks.tracers.evaluation import EvaluatorCallbackHandler
    
    # Async callback handler cho production
    async def on_llm_complete_async(token_usage: dict):
        """Log async vào database/warehouse"""
        print(f"[Async] Token usage: {token_usage}")
        # Ở đây có thể gửi lên monitoring system như Prometheus, Datadog
        # Hoặc lưu vào ClickHouse, BigQuery để phân tích sau
    
    return on_llm_complete_async


if __name__ == "__main__":
    results = demo_chain_with_callbacks()

Lỗi Thường Gặp và Cách Khắc Phục

1. Lỗi Authentication Error 401 - Sai API Key Hoặc Endpoint

# ❌ SAI: Dùng endpoint gốc của OpenAI
base_url = "https://api.openai.com/v1"  # Lỗi!

✅ ĐÚNG: Dùng endpoint HolySheep AI
base_url = "https://api.holysheep.ai/v1"  # Đúng!

Cách kiểm tra API key còn valid không
import requests

def verify_api_key(api_key: str) -> dict:
    """Verify HolySheep API key"""
    try:
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "deepseek-v3.2",
                "messages": [{"role": "user", "content": "test"}],
                "max_tokens": 10
            },
            timeout=10
        )
        
        if response.status_code == 401:
            return {
                "valid": False,
                "error": "API key không hợp lệ hoặc đã hết hạn. Vui lòng kiểm tra tại https://www.holysheep.ai/dashboard"
            }
        elif response.status_code == 200:
            return {"valid": True, "message": "API key hoạt động tốt"}
        else:
            return {
                "valid": False,
                "error": f"Lỗi {response.status_code}: {response.text}"
            }
    except Exception as e:
        return {"valid": False, "error": str(e)}


Sử dụng
result = verify_api_key("YOUR_HOLYSHEEP_API_KEY")
print(result)

2. Lỗi Rate Limit 429 - Vượt Quá Giới Hạn Request

import time
import threading
from functools import wraps
from collections import deque


class RateLimitHandler:
    """
    Handler xử lý rate limit với exponential backoff.
    HolySheep AI có giới hạn riêng, kiểm tra tại dashboard.
    """
    
    def __init__(self, max_requests_per_minute: int = 60):
        self.max_rpm = max_requests_per_minute
        self.request_times = deque()
        self.lock = threading.Lock()
        self.retry_count = 0
        self.max_retries = 5
        
    def wait_if_needed(self):
        """Chờ nếu vượt rate limit"""
        with self.lock:
            now = time.time()
            # Loại bỏ requests cũ hơn 1 phút
            while self.request_times and self.request_times[0] < now - 60:
                self.request_times.popleft()
            
            if len(self.request_times) >= self.max_rpm:
                # Tính thời gian chờ
                wait_time = 60 - (now - self.request_times[0])
                print(f"[RateLimit] Chờ {wait_time:.2f}s trước request tiếp theo...")
                time.sleep(wait_time)
            
            self.request_times.append(time.time())
    
    def execute_with_retry(self, func, *args, **kwargs):
        """Execute với retry logic"""
        for attempt in range(self.max_retries):
            try:
                self.wait_if_needed()
                return func(*args, **kwargs)
            except Exception as e:
                if "429" in str(e) or "rate limit" in str(e).lower():
                    wait_time = (2 ** attempt) * 1.5  # Exponential backoff
                    print(f"[Retry] Attempt {attempt + 1} thất bại, chờ {wait_time}s...")
                    time.sleep(wait_time)
                    continue
                raise
        raise Exception(f"Failed after {self.max_retries} retries")


Khởi tạo rate limiter cho HolySheep
rate_limiter = RateLimitHandler(max_requests_per_minute=60)


def safe_llm_call(llm, prompt: str) -> str:
    """Gọi LLM an toàn với rate limit handling"""
    def _call():
        return llm.invoke(prompt)
    return rate_limiter.execute_with_retry(_call)

3. Lỗi Model Not Found - Sai Tên Model

# Mapping model names chính xác cho HolySheep
MODEL_NAME_MAPPING = {
    # GPT Models
    "gpt-4": "gpt-4.1",
    "gpt-4-turbo": "gpt-4.1-mini",
    "gpt-4o": "gpt-4o",
    
    # Claude Models - CHÚ Ý: format khác với Anthropic
    "claude-3-opus-20240229": "claude-sonnet-4-5",
    "claude-3-sonnet-20240229": "claude-sonnet-4-5",
    "claude-3-5-sonnet-20241022": "claude-sonnet-4-5",
    
    # Gemini Models
    "gemini-pro": "gemini-2.5-flash",
    "gemini-1.5-pro": "gemini-2.5-flash",
    "gemini-2.0-flash": "gemini-2.5-flash",
    
    # DeepSeek - giá rẻ nhất
    "deepseek-chat": "deepseek-v3.2",
    "deepseek-coder": "deepseek-v3.2",
}


def normalize_model_name(model: str) -> str:
    """
    Normalize tên model để tương thích với HolySheep API.
    HolySheep hỗ trợ: GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2
    """
    model_lower = model.lower().strip()
    
    # Kiểm tra exact match trước
    if model_lower in MODEL_NAME_MAPPING:
        return MODEL_NAME_MAPPING[model_lower]
    
    # Kiểm tra partial match
    for key, value in MODEL_NAME_MAPPING.items():
        if key in model_lower or model_lower in key:
            print(f"[Warning] Model '{model}' → '{value}' (mapped)")
            return value
    
    # Trả về model gốc nếu không match
    print(f"[Info] Sử dụng model: {model}")
    return model


def validate_model_available(model: str) -> bool:
    """Kiểm tra model có trong danh sách hỗ trợ của HolySheep không"""
    supported = [
        "gpt-4.1", "gpt-4.1-mini", "gpt-4o",
        "claude-sonnet-4-5", "claude-3.5-sonnet",
        "gemini-2.5-flash", "gemini-2.0-flash",
        "deepseek-v3.2", "deepseek-chat"
    ]
    return any(m in model.lower() for m in supported)


Test
test_models = [
    "gpt-4",
    "claude-3-5-sonnet-20241022",
    "gemini-1.5-pro",
    "deepseek-chat"
]

print("Testing model name normalization:")
for model in test_models:
    normalized = normalize_model_name(model)
    print(f"  {model} → {normalized}")

Bảng So Sánh Chi Phí Thực Tế

Model	Input ($/MTok)	Output ($/MTok)	Phù hợp cho
DeepSeek V3.2	$0.07	$0.42	Massive usage, testing
Gemini 2.5 Flash	$0.125	$0.50	Daily tasks, high volume
GPT-4.1 Mini	$2.00	$8.00	Fast responses
GPT-4.1	$8.00	$8.00	Complex reasoning
Claude Sonnet 4.5	$3.00	$15.00	Long context, analysis

Kết Luận

Qua 3 năm thực chiến với LangChain và nhiều provider LLM, tôi nhận thấy HolySheep AI là lựa chọn tối ưu cho dự án production nhờ:

• Tiết kiệm 85%+: Với tỷ giá ¥1=$1 và DeepSeek V3.2 chỉ $0.42/MTok output
• Latency <50ms: Gần như instant cho các ứng dụng real-time
• Thanh toán linh hoạt: WeChat, Alipay, Visa - phù hợp với dev Việt Nam
• API tương thích 100%: Chỉ cần đổi base_url là chạy ngay
• Tín dụng miễn phí khi đăng ký: Không rủi ro khi thử nghiệm

Nên Dùng và Không Nên Dùng

✅ Nên dùng HolySheep AI khi:

• Chạy production với volume lớn (tiết kiệm chi phí đáng kể)
• Cần latency thấp cho ứng dụng real-time
• Thích hợp cho dev Việt Nam với thanh toán WeChat/Alipay
• Testing và development với tín dụng miễn phí

❌ Không nên dùng khi:

• Cần model mới nhất chưa được cập nhật trên HolySheep
• Yêu cầu compliance chứng nhận SOC2/HIPAA cụ thể
• Cần SLA cam kết bằng văn bản với mức uptime 99.9%

👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký