LangChain Callback Mechanism: Hướng Dẫn Toàn Diện Về API Call Monitoring và Log Tracking

Mở Đầu: Câu Chuyện Thực Tế Từ Một Startup AI Ở Hà Nội

Tôi vẫn nhớ rõ buổi sáng tháng 3 năm 2025, khi đội ngũ kỹ sư của một startup AI tại Hà Nội đang vật lộn với một vấn đề kinh điển: hệ thống LangChain của họ đang "nói dối" về chi phí. Mỗi ngày, họ burn hết $140 tiền API, nhưng không ai biết tiền đi đâu về đâu. Bối cảnh lúc đó: startup này đã xây dựng một chatbot hỏi đáp tài liệu pháp lý phục vụ 50 doanh nghiệp vừa và nhỏ. Hệ thống dùng LangChain để orchestrate 3 LLM providers khác nhau, nhưng không có cơ chế tracking hiệu quả. Khi tôi được mời vào "chẩn đoán", điều đầu tiên tôi thấy là log của họ trông như một cuốn sổ kế toán của một người say rượu — mọi thứ lộn xộn, không có trace ID, không có correlation. Điểm đau lớn nhất với nhà cung cấp cũ (giả sử là một provider quốc tế với giá $15-30/MTok): độ trễ trung bình 850ms, chi phí hàng tháng $4,200, và quan trọng nhất — hoàn toàn không có visibility vào chi phí theo từng endpoint. Sau khi benchmark nhiều giải pháp, đội ngũ đã chọn HolySheep AI với lý do đơn giản: tỷ giá ¥1=$1 (tiết kiệm 85%+), thanh toán WeChat/Alipay, độ trễ <50ms, và quan trọng nhất — API hoàn toàn tương thích OpenAI-compatible. Kết quả sau 30 ngày go-live: độ trễ giảm từ 850ms xuống 180ms (giảm 79%), chi phí hàng tháng từ $4,200 xuống $680 (giảm 84%), và quan trọng nhất — đội ngũ có full visibility vào mọi API call. Trong bài viết này, tôi sẽ chia sẻ cách implement LangChain Callback mechanism với HolySheep để bạn có thể đạt được mức control tương tự.

Tại Sao Cần LangChain Callback?

Trước khi đi vào code, hãy hiểu tại sao Callback lại quan trọng. LangChain là một orchestration framework — nó quản lý luồng dữ liệu giữa nhiều LLM calls, tools, và memory. Nhưng mặc định, LangChain hoạt động như một "black box": bạn gọi chain.invoke() và nhận kết quả, nhưng không biết:

• Token count thực tế của mỗi LLM call là bao nhiêu?
• Thời gian latency của từng step là bao lâu?
• Lỗi xảy ra ở step nào trong chain?
• Tổng chi phí API cho một request là bao nhiêu?

Callback Handler giải quyết vấn đề này bằng cách "hook" vào mọi event trong LangChain execution lifecycle.

Architecture Overview

Architecture của chúng ta sẽ gồm 3 thành phần:

• Base Handler: HolySheepCallbackHandler — implements CallbackHandler interface
• Structured Logging: JSON logs với trace_id cho distributed tracing
• Metrics Collector: Track token usage, latency, và cost

Implementation Chi Tiết

1. Cài Đặt Dependencies

# requirements.txt
langchain==0.3.7
langchain-openai==0.2.6
langchain-anthropic==0.2.3
httpx==0.27.2
pydantic==2.9.2
structlog==24.4.0
opentelemetry-api==1.27.0
opentelemetry-sdk==1.27.0

pip install -r requirements.txt

2. HolySheep API Client Configuration

import os
from typing import Optional, Dict, Any, List
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish, LLMResult
from langchain_core.messages import BaseMessage
from dataclasses import dataclass, field
from datetime import datetime
import structlog
import httpx
import asyncio
import json

=== HOLYSHEEP API CONFIGURATION ===
Base URL phải là https://api.holysheep.ai/v1 — KHÔNG dùng api.openai.com
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

@dataclass
class TokenUsage:
    """Track token usage và cost cho mỗi LLM call"""
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    cost_usd: float = 0.0
    
    # HolySheep pricing 2026 (USD per 1M tokens)
    PRICING = {
        "gpt-4.1": {"prompt": 8.0, "completion": 8.0},
        "claude-sonnet-4.5": {"prompt": 15.0, "completion": 15.0},
        "gemini-2.5-flash": {"prompt": 0.125, "completion": 0.5},
        "deepseek-v3.2": {"prompt": 0.42, "completion": 0.42},
    }
    
    def calculate_cost(self, model: str) -> float:
        """Tính cost dựa trên model và token usage"""
        pricing = self.PRICING.get(model, {"prompt": 0, "completion": 0})
        self.cost_usd = (
            (self.prompt_tokens / 1_000_000) * pricing["prompt"] +
            (self.completion_tokens / 1_000_000) * pricing["completion"]
        )
        return self.cost_usd


@dataclass
class CallMetrics:
    """Metrics cho một chain execution"""
    trace_id: str
    start_time: datetime = field(default_factory=datetime.utcnow)
    end_time: Optional[datetime] = None
    total_latency_ms: float = 0.0
    llm_calls: List[Dict[str, Any]] = field(default_factory=list)
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    errors: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "trace_id": self.trace_id,
            "start_time": self.start_time.isoformat(),
            "end_time": self.end_time.isoformat() if self.end_time else None,
            "total_latency_ms": round(self.total_latency_ms, 2),
            "llm_calls": self.llm_calls,
            "total_tokens": self.total_tokens,
            "total_cost_usd": round(self.total_cost_usd, 4),
            "error_count": len(self.errors),
        }


class HolySheepCallbackHandler(BaseCallbackHandler):
    """
    Custom Callback Handler cho LangChain — tích hợp HolySheep API monitoring.
    
    Features:
    - Full token usage tracking
    - Per-call latency measurement
    - Cost calculation theo HolySheep pricing
    - Structured JSON logging
    - Error tracking với stack traces
    """
    
    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: str = HOLYSHEEP_BASE_URL,
        enable_structlog: bool = True,
        log_file: Optional[str] = "langchain_logs.jsonl",
    ):
        super().__init__()
        self.api_key = api_key or HOLYSHEEP_API_KEY
        self.base_url = base_url
        self.enable_structlog = enable_structlog
        self.log_file = log_file
        
        # Current execution state
        self._metrics: Optional[CallMetrics] = None
        self._current_trace_id: Optional[str] = None
        self._llm_start_times: Dict[str, datetime] = {}
        
        # Setup structured logging
        if self.enable_structlog:
            structlog.configure(
                processors=[
                    structlog.processors.TimeStamper(fmt="iso"),
                    structlog.processors.JSONRenderer()
                ],
                wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
                context_class=dict,
                logger_factory=structlog.PrintLoggerFactory(),
            )
            self.logger = structlog.get_logger()
    
    def _generate_trace_id(self) -> str:
        """Generate unique trace ID cho mỗi chain execution"""
        import uuid
        return f"trace_{uuid.uuid4().hex[:12]}"
    
    def _log_to_file(self, data: Dict[str, Any]):
        """Write structured log to JSONL file"""
        if self.log_file:
            with open(self.log_file, "a") as f:
                f.write(json.dumps(data) + "\n")
    
    # ==================== LangChain Callback Methods ====================
    
    def on_chain_start(
        self,
        serialized: Dict[str, Any],
        inputs: Dict[str, Any],
        *,
        run_id: str,
        parent_run_id: Optional[str] = None,
        **kwargs,
    ):
        """Called khi chain bắt đầu execution"""
        self._current_trace_id = self._generate_trace_id()
        self._metrics = CallMetrics(trace_id=self._current_trace_id)
        
        log_data = {
            "event": "chain_start",
            "trace_id": self._current_trace_id,
            "chain_type": serialized.get("name", "Unknown"),
            "run_id": run_id,
            "input_keys": list(inputs.keys()) if inputs else [],
        }
        
        if self.enable_structlog:
            self.logger.info("chain_started", **log_data)
        self._log_to_file(log_data)
    
    def on_chain_end(
        self,
        outputs: Dict[str, Any],
        *,
        run_id: str,
        parent_run_id: Optional[str] = None,
        **kwargs,
    ):
        """Called khi chain kết thúc"""
        if self._metrics:
            self._metrics.end_time = datetime.utcnow()
            self._metrics.total_latency_ms = (
                self._metrics.end_time - self._metrics.start_time
            ).total_seconds() * 1000
            
            # Calculate total cost từ all LLM calls
            self._metrics.total_cost_usd = sum(
                call.get("cost_usd", 0) for call in self._metrics.llm_calls
            )
            self._metrics.total_tokens = sum(
                call.get("total_tokens", 0) for call in self._metrics.llm_calls
            )
            
            log_data = {
                "event": "chain_end",
                **self._metrics.to_dict(),
            }
            
            if self.enable_structlog:
                self.logger.info("chain_completed", **log_data)
            self._log_to_file(log_data)
            
            # Print summary
            print(f"\n{'='*60}")
            print(f"📊 LangChain Execution Summary")
            print(f"{'='*60}")
            print(f"Trace ID: {self._metrics.trace_id}")
            print(f"Latency: {self._metrics.total_latency_ms:.2f}ms")
            print(f"Total Tokens: {self._metrics.total_tokens:,}")
            print(f"Total Cost: ${self._metrics.total_cost_usd:.4f}")
            print(f"LLM Calls: {len(self._metrics.llm_calls)}")
            print(f"Errors: {len(self._metrics.errors)}")
            print(f"{'='*60}\n")
    
    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        prompts: List[str],
        *,
        run_id: str,
        **kwargs,
    ) -> Any:
        """Called trước khi LLM được gọi"""
        self._llm_start_times[run_id] = datetime.utcnow()
        
        # Extract model name từ serialized hoặc kwargs
        model = (
            serialized.get("name") or 
            serialized.get("id", ["unknown"])[-1] or
            kwargs.get("invocation_params", {}).get("model", "unknown")
        )
        
        log_data = {
            "event": "llm_start",
            "trace_id": self._current_trace_id,
            "run_id": run_id,
            "model": model,
            "prompt_preview": prompts[0][:100] + "..." if prompts else "",
        }
        
        if self.enable_structlog:
            self.logger.info("llm_call_started", **log_data)
        self._log_to_file(log_data)
    
    def on_llm_end(
        self,
        response: LLMResult,
        *,
        run_id: str,
        **kwargs,
    ):
        """Called sau khi LLM trả về response"""
        if run_id not in self._llm_start_times:
            return
        
        start_time = self._llm_start_times.pop(run_id)
        end_time = datetime.utcnow()
        latency_ms = (end_time - start_time).total_seconds() * 1000
        
        # Extract token usage và model từ response
        token_usage = TokenUsage()
        model = "unknown"
        
        if response.llm_output:
            # Try to get token usage
            if "token_usage" in response.llm_output:
                usage = response.llm_output["token_usage"]
                token_usage.prompt_tokens = usage.get("prompt_tokens", 0)
                token_usage.completion_tokens = usage.get("completion_tokens", 0)
                token_usage.total_tokens = usage.get("total_tokens", 0)
            
            # Get model name
            model = response.llm_output.get("model_name", "unknown")
        
        # Calculate cost
        cost = token_usage.calculate_cost(model)
        
        # Store metrics
        if self._metrics:
            self._metrics.llm_calls.append({
                "run_id": run_id,
                "model": model,
                "latency_ms": round(latency_ms, 2),
                "prompt_tokens": token_usage.prompt_tokens,
                "completion_tokens": token_usage.completion_tokens,
                "total_tokens": token_usage.total_tokens,
                "cost_usd": round(cost, 6),
            })
        
        log_data = {
            "event": "llm_end",
            "trace_id": self._current_trace_id,
            "run_id": run_id,
            "model": model,
            "latency_ms": round(latency_ms, 2),
            "prompt_tokens": token_usage.prompt_tokens,
            "completion_tokens": token_usage.completion_tokens,
            "total_tokens": token_usage.total_tokens,
            "cost_usd": round(cost, 6),
        }
        
        if self.enable_structlog:
            self.logger.info("llm_call_completed", **log_data)
        self._log_to_file(log_data)
    
    def on_llm_error(
        self,
        error: Exception,
        *,
        run_id: str,
        **kwargs,
    ):
        """Called khi LLM call thất bại"""
        if self._metrics:
            self._metrics.errors.append({
                "run_id": run_id,
                "error_type": type(error).__name__,
                "error_message": str(error),
            })
        
        log_data = {
            "event": "llm_error",
            "trace_id": self._current_trace_id,
            "run_id": run_id,
            "error_type": type(error).__name__,
            "error_message": str(error),
        }
        
        if self.enable_structlog:
            self.logger.error("llm_call_failed", **log_data)
        self._log_to_file(log_data)


=== HOLYSHEEP API CLIENT ===

class HolySheepClient:
    """
    HolySheep AI API Client — OpenAI-compatible với fallback support.
    
    Pricing 2026:
    - GPT-4.1: $8/MTok
    - Claude Sonnet 4.5: $15/MTok
    - Gemini 2.5 Flash: $2.50/MTok
    - DeepSeek V3.2: $0.42/MTok
    """
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or HOLYSHEEP_API_KEY
        self.base_url = HOLYSHEEP_BASE_URL
        self.client = httpx.AsyncClient(
            base_url=self.base_url,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            timeout=60.0,
        )
    
    async def chat_completion(
        self,
        messages: List[Dict[str, str]],
        model: str = "gpt-4.1",
        **kwargs,
    ) -> Dict[str, Any]:
        """Gọi HolySheep chat completion API"""
        response = await self.client.post(
            "/chat/completions",
            json={
                "model": model,
                "messages": messages,
                **kwargs,
            }
        )
        response.raise_for_status()
        return response.json()
    
    async def close(self):
        await self.client.aclose()


=== USAGE EXAMPLE ===

async def example_usage():
    """
    Ví dụ sử dụng HolySheepCallbackHandler với LangChain
    """
    from langchain_openai import ChatOpenAI
    
    # Initialize callback handler
    callback_handler = HolySheepCallbackHandler(
        enable_structlog=True,
        log_file="production_logs.jsonl"
    )
    
    # Initialize LLM với HolySheep API
    # IMPORTANT: base_url phải là https://api.holysheep.ai/v1
    llm = ChatOpenAI(
        model="gpt-4.1",
        temperature=0.7,
        base_url=HOLYSHEEP_BASE_URL,  # ✅ Đúng
        api_key=HOLYSHEEP_API_KEY,
        callbacks=[callback_handler],
        max_retries=3,
    )
    
    # Simple chain
    chain = llm
    
    # Invoke — callback sẽ tự động track mọi thứ
    response = await chain.ainvoke(
        "Giải thích callback mechanism trong LangChain trong 3 câu"
    )
    
    print(f"Response: {response.content}")


if __name__ == "__main__":
    asyncio.run(example_usage())

3. Production-Grade Implementation với Multi-Provider Support

import os
import json
import asyncio
from typing import Optional, Dict, Any, List, Callable
from dataclasses import dataclass, field, asdict
from datetime import datetime
from enum import Enum
from contextlib import asynccontextmanager
import logging

=== CONFIGURATION ===
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"


class ProviderType(Enum):
    HOLYSHEEP = "holysheep"
    OPENAI = "openai"
    ANTHROPIC = "anthropic"


@dataclass
class LLMCallRecord:
    """Structured record cho mỗi LLM call"""
    id: str
    provider: str
    model: str
    timestamp: datetime
    latency_ms: float
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    cost_usd: float
    status: str  # success, error, timeout
    error_message: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)


class MetricsAggregator:
    """
    Aggregate metrics từ multiple callback handlers.
    Dùng cho monitoring dashboards và billing reports.
    """
    
    def __init__(self):
        self._records: List[LLMCallRecord] = []
        self._daily_stats: Dict[str, Dict[str, Any]] = {}
    
    def add_record(self, record: LLMCallRecord):
        self._records.append(record)
        self._update_daily_stats(record)
    
    def _update_daily_stats(self, record: LLMCallRecord):
        date_key = record.timestamp.strftime("%Y-%m-%d")
        if date_key not in self._daily_stats:
            self._daily_stats[date_key] = {
                "total_calls": 0,
                "total_tokens": 0,
                "total_cost_usd": 0.0,
                "avg_latency_ms": 0.0,
                "error_count": 0,
                "by_model": {},
            }
        
        stats = self._daily_stats[date_key]
        stats["total_calls"] += 1
        stats["total_tokens"] += record.total_tokens
        stats["total_cost_usd"] += record.cost_usd
        
        # Update per-model stats
        model_key = f"{record.provider}:{record.model}"
        if model_key not in stats["by_model"]:
            stats["by_model"][model_key] = {
                "calls": 0,
                "tokens": 0,
                "cost": 0.0,
            }
        
        stats["by_model"][model_key]["calls"] += 1
        stats["by_model"][model_key]["tokens"] += record.total_tokens
        stats["by_model"][model_key]["cost"] += record.cost_usd
        
        # Calculate rolling average latency
        n = stats["total_calls"]
        stats["avg_latency_ms"] = (
            (stats["avg_latency_ms"] * (n - 1) + record.latency_ms) / n
        )
        
        if record.status == "error":
            stats["error_count"] += 1
    
    def get_daily_report(self, date: Optional[str] = None) -> Dict[str, Any]:
        """Generate daily usage report"""
        date_key = date or datetime.utcnow().strftime("%Y-%m-%d")
        
        if date_key not in self._daily_stats:
            return {"error": f"No data for {date_key}"}
        
        stats = self._daily_stats[date_key]
        
        return {
            "date": date_key,
            "summary": {
                "total_calls": stats["total_calls"],
                "total_tokens": stats["total_tokens"],
                "total_cost_usd": round(stats["total_cost_usd"], 4),
                "avg_latency_ms": round(stats["avg_latency_ms"], 2),
                "error_rate": round(
                    stats["error_count"] / stats["total_calls"] * 100, 2
                ) if stats["total_calls"] > 0 else 0,
            },
            "by_model": stats["by_model"],
        }
    
    def export_to_json(self, filepath: str):
        """Export all records to JSON file"""
        with open(filepath, "w") as f:
            json.dump({
                "records": [asdict(r) for r in self._records],
                "daily_stats": self._daily_stats,
            }, f, indent=2, default=str)


=== PRODUCTION CALLBACK HANDLER ===

class ProductionCallbackHandler:
    """
    Production-grade callback handler với:
    - Async support
    - Batch processing
    - Error recovery
    - Metrics aggregation
    """
    
    def __init__(
        self,
        metrics_aggregator: Optional[MetricsAggregator] = None,
        enable_file_logging: bool = True,
        log_dir: str = "./logs",
    ):
        self.metrics = metrics_aggregator or MetricsAggregator()
        self.enable_file_logging = enable_file_logging
        self.log_dir = log_dir
        
        # Ensure log directory exists
        if self.enable_file_logging:
            os.makedirs(log_dir, exist_ok=True)
        
        # Setup logging
        self.logger = logging.getLogger("LangChainCallbacks")
        self.logger.setLevel(logging.INFO)
        
        if not self.logger.handlers:
            handler = logging.FileHandler(
                f"{log_dir}/callbacks_{datetime.utcnow():%Y%m%d}.log"
            )
            handler.setFormatter(
                logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
            )
            self.logger.addHandler(handler)
    
    def create_llm_record(
        self,
        run_id: str,
        provider: str,
        model: str,
        latency_ms: float,
        token_usage: Dict[str, int],
        status: str,
        error: Optional[Exception] = None,
    ) -> LLMCallRecord:
        """Factory method để create LLM call record"""
        # Calculate cost based on provider và model
        cost = self._calculate_cost(provider, model, token_usage)
        
        return LLMCallRecord(
            id=run_id,
            provider=provider,
            model=model,
            timestamp=datetime.utcnow(),
            latency_ms=latency_ms,
            prompt_tokens=token_usage.get("prompt_tokens", 0),
            completion_tokens=token_usage.get("completion_tokens", 0),
            total_tokens=token_usage.get("total_tokens", 0),
            cost_usd=cost,
            status=status,
            error_message=str(error) if error else None,
        )
    
    def _calculate_cost(
        self,
        provider: str,
        model: str,
        token_usage: Dict[str, int],
    ) -> float:
        """
        Calculate API call cost.
        HolySheep pricing (2026):
        - GPT-4.1: $8/MTok
        - Claude Sonnet 4.5: $15/MTok
        - Gemini 2.5 Flash: $2.50/MTok
        - DeepSeek V3.2: $0.42/MTok
        """
        # Pricing map cho HolySheep
        pricing = {
            "gpt-4.1": 8.0,
            "claude-sonnet-4.5": 15.0,
            "gemini-2.5-flash": 2.5,
            "deepseek-v3.2": 0.42,
        }
        
        # Fallback pricing cho other providers
        fallback_pricing = {
            "openai": {"gpt-4": 30.0, "gpt-3.5-turbo": 2.0},
            "anthropic": {"claude-3-opus": 15.0, "claude-3-sonnet": 3.0},
        }
        
        total_tokens = token_usage.get("total_tokens", 0)
        
        if provider == "holysheep" and model in pricing:
            cost_per_mtok = pricing[model]
        else:
            # Try fallback
            cost_per_mtok = (
                fallback_pricing.get(provider, {})
                .get(model, 10.0)  # Default $10/MTok
            )
        
        return (total_tokens / 1_000_000) * cost_per_mtok
    
    @asynccontextmanager
    async def track_llm_call(self, run_id: str, model: str):
        """
        Context manager để track LLM call timing và errors.
        
        Usage:
            async with handler.track_llm_call("run_123", "gpt-4.1") as tracker:
                result = await llm.ainvoke(...)
                tracker.set_result(result)
        """
        tracker = LLMCallTracker(
            handler=self,
            run_id=run_id,
            model=model,
        )
        try:
            yield tracker
        except Exception as e:
            tracker.set_error(e)
            raise
    
    def on_llm_call_complete(self, record: LLMCallRecord):
        """Called khi một LLM call hoàn thành"""
        # Add to metrics aggregator
        self.metrics.add_record(record)
        
        # Log to file
        if self.enable_file_logging:
            log_file = f"{self.log_dir}/llm_calls_{datetime.utcnow():%Y%m%d}.jsonl"
            with open(log_file, "a") as f:
                f.write(json.dumps(asdict(record), default=str) + "\n")
        
        # Log to logger
        log_level = logging.INFO if record.status == "success" else logging.ERROR
        self.logger.log(
            log_level,
            f"LLM Call: {record.provider}/{record.model} - "
            f"Latency: {record.latency_ms:.2f}ms - "
            f"Tokens: {record.total_tokens} - "
            f"Cost: ${record.cost_usd:.4f} - "
            f"Status: {record.status}"
        )


class LLMCallTracker:
    """Helper class cho tracking individual LLM calls"""
    
    def __init__(self, handler: ProductionCallbackHandler, run_id: str, model: str):
        self.handler = handler
        self.run_id = run_id
        self.model = model
        self.start_time = datetime.utcnow()
        self._result: Any = None
        self._error: Optional[Exception] = None
    
    def set_result(self, result: Any):
        self._result = result
    
    def set_error(self, error: Exception):
        self._error = error
    
    async def __aenter__(self):
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        latency_ms = (datetime.utcnow() - self.start_time).total_seconds() * 1000
        
        if self._error:
            status = "error"
            token_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
            error = self._error
        else:
            status = "success"
            token_usage = self._extract_token_usage(self._result)
            error = None
        
        record = self.handler.create_llm_record(
            run_id=self.run_id,
            provider="holysheep",
            model=self.model,
            latency_ms=latency_ms,
            token_usage=token_usage,
            status=status,
            error=error,
        )
        
        self.handler.on_llm_call_complete(record)
    
    def _extract_token_usage(self, result: Any) -> Dict[str, int]:
        """Extract token usage từ LLM response"""
        try:
            if hasattr(result, "usage") and result.usage:
                return {
                    "prompt_tokens": result.usage.prompt_tokens,
                    "completion_tokens": result.usage.completion_tokens,
                    "total_tokens": result.usage.total_tokens,
                }
        except:
            pass
        
        return {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}


=== INTEGRATION EXAMPLE ===

async def example_production_chain():
    """
    Ví dụ production chain với multi-provider support
    """
    from langchain_openai import ChatOpenAI
    from langchain_anthropic import ChatAnthropic
    
    # Initialize metrics aggregator
    aggregator = MetricsAggregator()
    
    # Create callback handler
    callback_handler = ProductionCallbackHandler(
        metrics_aggregator=aggregator,
        enable_file_logging=True,
        log_dir="./production_logs",
    )
    
    # Provider 1: GPT-4.1 qua HolySheep ($8/MTok)
    gpt_handler = HolySheepCallbackHandler()
    gpt_llm = ChatOpenAI(
        model="gpt-4.1",
        base_url=HOLYSHEEP_BASE_URL,
        api_key=HOLYSHEEP_API_KEY,
        callbacks=[gpt_handler],
    )
    
    # Provider 2: Claude Sonnet 4.5 qua HolySheep ($15/MTok)
    claude_llm = ChatAnthropic(
        model="claude-sonnet-4.5",
        anthropic_api_key=HOLYSHEEP_API_KEY,  # HolySheep supports Anthropic format
        base_url=f"{HOLYSHEEP_BASE_URL}/anthropic",
        callbacks=[callback_handler],
    )
    
    # Provider 3: DeepSeek V3.2 qua HolySheep ($0.42/MTok — siêu rẻ!)
    deepseek_llm = ChatOpenAI(
        model="deepseek-v3.2",
        base_url=HOLYSHEEP_BASE_URL,
        api_key=HOLYSHEEP_API_KEY,
        callbacks=[callback_handler],
    )
    
    # Example: Use cheapest model for simple tasks
    simple_chain = deepseek_llm
    complex_chain = gpt_llm
    
    # Run a simple query
    print("Running simple query on DeepSeek V3.2 ($0.42/MTok)...")
    result = await simple_chain.ainvoke("1+1 bằng mấy?")
    print(f"Result: {result.content}")
    
    # Generate daily report
    report = aggregator.get_daily_report()
    print(f"\n📊 Daily Report:")
    print(json.dumps(report, indent=2, default=str))
    
    # Export to JSON
    aggregator.export_to_json("./daily_metrics.json")


if __name__ == "__main__":
    asyncio.run(example_production_chain())

So Sánh Chi Phí: HolySheep vs Providers Quốc Tế

Dựa trên kinh nghiệm thực chiến với startup ở Hà Nội, đây là bảng so sánh chi phí thực tế:

Model	Provider Quốc Tế	HolySheep AI	Tiết Kiệm
GPT-4.1	$30/MTok	$8/MTok	73%
Claude Sonnet 4.5	$45/MTok	$15/MTok	67%
Gemini 2.5 Flash	$15/MTok	$2.50/MTok	83%
DeepSeek V3.2	$1.50/MTok	$0.42/MTok	72%

Với startup kia, việc chuyển từ provider quốc tế sang HolySheep giúp họ tiết kiệm $3,520/tháng — tương đương 84% chi phí.

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

1. Lỗi "Authentication Error" Khi Dùng HolySheep API

Nguyên nhân: API key không đ