LangChain Callbacks: Event Tracking và Logging — Hướng Dẫn Toàn Diện 2026

Là một kỹ sư đã triển khai LangChain cho hơn 20 dự án AI production, tôi hiểu rằng callbacks là trái tim của hệ thống monitoring. Bài viết này sẽ hướng dẫn bạn từ cơ bản đến nâng cao về cách track events và logging trong LangChain với HolySheep AI — nền tảng tiết kiệm đến 85% chi phí API với độ trễ dưới 50ms.

Bảng So Sánh Chi Phí API 2026

Trước khi đi vào chi tiết kỹ thuật, hãy xem dữ liệu giá thực tế từ các nhà cung cấp hàng đầu:

Model	Output ($/MTok)	10M tokens/tháng ($)
GPT-4.1	$8.00	$80
Claude Sonnet 4.5	$15.00	$150
Gemini 2.5 Flash	$2.50	$25
DeepSeek V3.2	$0.42	$4.20

Với HolySheep AI, bạn có thể truy cập tất cả các model này qua một endpoint duy nhất. Tỷ giá ¥1 = $1 giúp doanh nghiệp Việt Nam tiết kiệm đáng kể khi thanh toán qua WeChat hoặc Alipay.

Callbacks Trong LangChain Là Gì?

Callbacks là cơ chế cho phép bạn "lắng nghe" các sự kiện trong chain của LangChain. Khi tôi bắt đầu với LangChain, việc không có visibility vào internal operations là một trong những thách thức lớn nhất. Callbacks giải quyết vấn đề này bằng cách cho phép bạn:

• Theo dõi token usage theo thời gian thực
• Log các bước trung gian trong chain
• Đo đạc độ trễ của từng component
• Debug và audit các LLM calls
• Tích hợp với hệ thống monitoring bên ngoài

BaseCallbackHandler — Nền Tảng Của Mọi Callbacks

Trong kinh nghiệm thực chiến của tôi, việc tạo một custom handler tập trung giúp đơn giản hóa rất nhiều việc quản lý logging. Dưới đây là implementation hoàn chỉnh sử dụng HolySheep AI:

"""
LangChain Callback Handler cho HolySheep AI
Tác giả: HolySheep AI Technical Team
Version: 2.0.0
"""

import json
import time
from datetime import datetime
from typing import Any, Dict, List, Optional
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish, LLMResult

class HolySheepCallbackHandler(BaseCallbackHandler):
    """
    Custom callback handler cho LangChain
    - Track token usage với chi phí cụ thể
    - Logging chi tiết từng event
    - Integration với monitoring systems
    """
    
    def __init__(self, model_name: str = "gpt-4.1"):
        super().__init__()
        self.model_name = model_name
        self.events: List[Dict[str, Any]] = []
        self.total_tokens = 0
        self.total_cost = 0.0
        self.start_time = None
        
        # HolySheep pricing 2026 (output price per MTok)
        self.pricing = {
            "gpt-4.1": 8.00,           # $8/MTok
            "claude-sonnet-4.5": 15.00, # $15/MTok
            "gemini-2.5-flash": 2.50,   # $2.50/MTok
            "deepseek-v3.2": 0.42,     # $0.42/MTok
        }
    
    def _calculate_cost(self, tokens: int) -> float:
        """Tính chi phí dựa trên số tokens"""
        price_per_token = self.pricing.get(self.model_name, 8.00) / 1_000_000
        return tokens * price_per_token
    
    def _log_event(self, event_type: str, data: Dict[str, Any]):
        """Log event với timestamp và metadata"""
        event = {
            "timestamp": datetime.now().isoformat(),
            "type": event_type,
            "model": self.model_name,
            "data": data
        }
        self.events.append(event)
        print(f"[HolySheep] {event_type}: {json.dumps(data, indent=2, ensure_ascii=False)}")
    
    def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs):
        """Called when LLM starts processing"""
        self.start_time = time.time()
        self._log_event("llm_start", {
            "serialized": serialized,
            "prompts_count": len(prompts),
            "first_prompt_length": len(prompts[0]) if prompts else 0
        })
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        """Called when LLM finishes processing"""
        if self.start_time:
            latency_ms = (time.time() - self.start_time) * 1000
        
        # Extract token information from response
        for gen in response.generations:
            for generation in gen:
                if hasattr(generation, 'generation_info') and generation.generation_info:
                    token_info = generation.generation_info.get('token_usage', {})
                    
                    prompt_tokens = token_info.get('prompt_tokens', 0)
                    completion_tokens = token_info.get('completion_tokens', 0)
                    total_tokens = prompt_tokens + completion_tokens
                    cost = self._calculate_cost(total_tokens)
                    
                    self.total_tokens += total_tokens
                    self.total_cost += cost
                    
                    self._log_event("llm_end", {
                        "latency_ms": round(latency_ms, 2),
                        "prompt_tokens": prompt_tokens,
                        "completion_tokens": completion_tokens,
                        "total_tokens": total_tokens,
                        "cost_usd": round(cost, 6),
                        "total_cost_usd": round(self.total_cost, 6)
                    })
    
    def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs):
        """Called when chain starts"""
        self._log_event("chain_start", {
            "chain_name": serialized.get('name', 'unknown'),
            "input_keys": list(inputs.keys())
        })
    
    def on_chain_end(self, outputs: Dict[str, Any], **kwargs):
        """Called when chain ends"""
        self._log_event("chain_end", {
            "output_keys": list(outputs.keys())
        })
    
    def get_summary(self) -> Dict[str, Any]:
        """Trả về tổng kết sử dụng"""
        return {
            "total_events": len(self.events),
            "total_tokens": self.total_tokens,
            "total_cost_usd": round(self.total_cost, 6),
            "events": self.events
        }

Khởi tạo handler
handler = HolySheepCallbackHandler(model_name="deepseek-v3.2")
print(f"Handler initialized for model: {handler.model_name}")
print(f"Pricing: ${handler.pricing[handler.model_name]}/MTok")

Tích Hợp Với HolySheep API

Điều quan trọng là tất cả code mẫu trong bài viết này sử dụng HolySheep AI với base_url chuẩn. Dưới đây là cách setup ChatOpenAI với HolySheep endpoint:

"""
LangChain + HolySheep AI Integration
base_url: https://api.holysheep.ai/v1
Supports: GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2
"""

import os
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from your_callback_module import HolySheepCallbackHandler

=== CẤU HÌNH HOLYSHEEP ===
API Key từ HolySheep AI Dashboard
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

Model selection - so sánh chi phí:
- GPT-4.1: $8/MTok → 10M tokens = $80
- Claude Sonnet 4.5: $15/MTok → 10M tokens = $150
- Gemini 2.5 Flash: $2.50/MTok → 10M tokens = $25
- DeepSeek V3.2: $0.42/MTok → 10M tokens = $4.20 ⭐ BEST VALUE

MODEL_CONFIG = {
    "model": "deepseek-v3.2",  # Tiết kiệm 95% so với Claude
    "temperature": 0.7,
    "max_tokens": 2000,
    "request_timeout": 60,
    "max_retries": 3,
    "streaming": True
}

=== KHỞI TẠO LLM VỚI HOLYSHEEP ===
llm = ChatOpenAI(
    **MODEL_CONFIG,
    api_key=os.environ["HOLYSHEEP_API_KEY"],
    base_url="https://api.holysheep.ai/v1"  # ✅ HolySheep endpoint
)

=== KHỞI TẠO CALLBACK HANDLER ===
callback_handler = HolySheepCallbackHandler(model_name=MODEL_CONFIG["model"])

=== TẠO CHAIN VỚI CALLBACKS ===
prompt = PromptTemplate(
    input_variables=["topic"],
    template="Giải thích {topic} theo cách đơn giản nhất, phù hợp với người mới bắt đầu."
)

chain = LLMChain(
    llm=llm,
    prompt=prompt,
    callbacks=[callback_handler]  # ✅ Attach callback
)

=== CHẠY CHAIN VÀ XEM RESULTS ===
print("=" * 60)
print("🔄 Running LangChain với HolySheep AI Callbacks")
print("=" * 60)

result = chain.run(
    topic="LangChain callbacks và event tracking",
    callbacks=[callback_handler]
)

print("\n" + "=" * 60)
print("📊 SUMMARY")
print("=" * 60)

summary = callback_handler.get_summary()
print(f"Total Events: {summary['total_events']}")
print(f"Total Tokens: {summary['total_tokens']:,}")
print(f"Total Cost: ${summary['total_cost_usd']:.6f}")

=== SO SÁNH CHI PHÍ ===
print("\n📈 Cost Comparison (10M tokens/month):")
models = ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
for model in models:
    cost_10m = (1_000_000 / 1_000_000) * 10 * 8.00 if model == "gpt-4.1" else \
               (1_000_000 / 1_000_000) * 10 * 15.00 if model == "claude-sonnet-4.5" else \
               (1_000_000 / 1_000_000) * 10 * 2.50 if model == "gemini-2.5-flash" else \
               (1_000_000 / 1_000_000) * 10 * 0.42
    print(f"  {model}: ${cost_10m:.2f}/tháng")

Advanced Callback Patterns

Trong production, tôi thường sử dụng nhiều callback handlers kết hợp. Dưới đây là pattern nâng cao với async support và real-time metrics:

"""
Advanced LangChain Callbacks - Multi-Handler Pattern
Author: HolySheep AI Technical Team
Features: Async support, Real-time metrics, Cost optimization
"""

import asyncio
import json
from typing import AsyncIterator
from datetime import datetime
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult
from collections import defaultdict
import time

class MetricsCollector(BaseCallbackHandler):
    """
    Async callback handler cho real-time metrics collection
    - Tính toán chi phí theo thời gian thực
    - Tracking latency distribution
    - Performance monitoring
    """
    
    def __init__(self):
        self.metrics = defaultdict(list)
        self.cost_tracker = defaultdict(float)
        self.latency_tracker = []
        self.request_count = 0
        self.error_count = 0
        
        # HolySheep 2026 Pricing
        self.pricing_per_1k = {
            "deepseek-v3.2": 0.00042,      # $0.42/MTok = $0.00042/1K tokens
            "gemini-2.5-flash": 0.00250,   # $2.50/MTok
            "gpt-4.1": 0.00800,            # $8.00/MTok
            "claude-sonnet-4.5": 0.01500,  # $15.00/MTok
        }
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        self.metrics["llm_starts"].append({
            "timestamp": datetime.now().isoformat(),
            "model": serialized.get("name", "unknown")
        })
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        self.request_count += 1
        start_time = time.time()
        
        for generation_group in response.generations:
            for generation in generation_group:
                if hasattr(generation, 'generation_info') and generation.generation_info:
                    token_usage = generation.generation_info.get('token_usage', {})
                    model = response.llm_output.get('model_name', 'unknown') if response.llm_output else 'unknown'
                    
                    total_tokens = token_usage.get('total_tokens', 0)
                    prompt_tokens = token_usage.get('prompt_tokens', 0)
                    completion_tokens = token_usage.get('completion_tokens', 0)
                    
                    # Calculate cost
                    price_per_token = self.pricing_per_1k.get(model, 0.008) / 1000
                    cost = total_tokens * price_per_token
                    
                    self.cost_tracker[model] += cost
                    
                    self.metrics["tokens"].append({
                        "total": total_tokens,
                        "prompt": prompt_tokens,
                        "completion": completion_tokens,
                        "cost_usd": cost
                    })
        
        latency = (time.time() - start_time) * 1000
        self.latency_tracker.append(latency)
    
    def on_llm_error(self, error, **kwargs):
        self.error_count += 1
        self.metrics["errors"].append({
            "timestamp": datetime.now().isoformat(),
            "error": str(error)
        })
    
    def get_metrics(self) -> dict:
        """Tính toán metrics tổng hợp"""
        total_tokens = sum(m["total"] for m in self.metrics["tokens"])
        total_cost = sum(self.cost_tracker.values())
        
        avg_latency = sum(self.latency_tracker) / len(self.latency_tracker) if self.latency_tracker else 0
        p95_latency = sorted(self.latency_tracker)[int(len(self.latency_tracker) * 0.95)] if self.latency_tracker else 0
        
        return {
            "request_count": self.request_count,
            "error_count": self.error_count,
            "total_tokens": total_tokens,
            "total_cost_usd": round(total_cost, 6),
            "cost_by_model": dict(self.cost_tracker),
            "latency": {
                "avg_ms": round(avg_latency, 2),
                "p95_ms": round(p95_latency, 2),
                "min_ms": round(min(self.latency_tracker), 2) if self.latency_tracker else 0,
                "max_ms": round(max(self.latency_tracker), 2) if self.latency_tracker else 0
            },
            "cost_projection_10m_tokens": {
                model: round(cost / total_tokens * 10_000_000, 2) if total_tokens > 0 else 0
                for model, cost in self.cost_tracker.items()
            }
        }

class StreamingCostOptimizer(BaseCallbackHandler):
    """
    Streaming callback với token counting theo thời gian thực
    Giúp tối ưu chi phí streaming
    """
    
    def __init__(self):
        self.current_tokens = 0
        self.stream_start_time = None
        self.chunks = []
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        self.stream_start_time = time.time()
        print(f"[Streaming] Starting LLM call...")
    
    def on_llm_new_token(self, token: str, **kwargs):
        """Called for each streaming token"""
        self.current_tokens += 1
        self.chunks.append(token)
        
        # Print streaming output (limit to prevent spam)
        if len(self.chunks) % 50 == 0:
            print(f"[Streaming] Tokens received: {self.current_tokens}")
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        if self.stream_start_time:
            total_time = time.time() - self.stream_start_time
            tokens_per_second = self.current_tokens / total_time if total_time > 0 else 0
            
            print(f"\n[Streaming] Complete!")
            print(f"  Total tokens: {self.current_tokens}")
            print(f"  Time: {total_time:.2f}s")
            print(f"  Speed: {tokens_per_second:.2f} tokens/s")
            
            # Cost với DeepSeek V3.2 ($0.42/MTok)
            cost = self.current_tokens * (0.42 / 1_000_000)
            print(f"  Cost (DeepSeek V3.2): ${cost:.6f}")

=== DEMO USAGE ===
async def demo_advanced_callbacks():
    """Demo với multiple callbacks"""
    from langchain_openai import ChatOpenAI
    from langchain.chains import LLMChain
    from langchain.prompts import PromptTemplate
    import os
    
    os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
    
    # Initialize LLM với HolySheep
    llm = ChatOpenAI(
        model="deepseek-v3.2",
        api_key=os.environ["OPENAI_API_KEY"],
        base_url="https://api.holysheep.ai/v1",
        streaming=True
    )
    
    # Multiple callbacks
    metrics_handler = MetricsCollector()
    streaming_handler = StreamingCostOptimizer()
    
    chain = LLMChain(
        llm=llm,
        prompt=PromptTemplate.from_template("Viết một đoạn văn ngắn về: {topic}"),
        callbacks=[metrics_handler, streaming_handler]
    )
    
    # Run
    result = await chain.arun(topic="HolySheep AI - Tiết kiệm 85% chi phí API")
    
    # Get metrics
    print("\n" + "=" * 60)
    print("📊 ADVANCED METRICS SUMMARY")
    print("=" * 60)
    
    metrics = metrics_handler.get_metrics()
    print(json.dumps(metrics, indent=2, ensure_ascii=False))
    
    return metrics

Chạy demo
if __name__ == "__main__":
    asyncio.run(demo_advanced_callbacks())

Streaming Callback Với Token Counting

Streaming là tính năng quan trọng để giảm perceived latency. Callback handler theo dõi từng token nhận được, giúp bạn đánh giá chi phí và tốc độ thực tế:

"""
Streaming Callback với Real-time Token Counting
Optimized cho HolySheep AI với <50ms latency
"""

from typing import Optional, Iterator
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult
import time

class StreamingTokenCounter(BaseCallbackHandler):
    """
    Callback đếm tokens trong streaming mode
    - Real-time token counting
    - Chi phí tính theo từng token
    - Độ trễ theo dõi
    """
    
    def __init__(self, model: str = "deepseek-v3.2"):
        self.model = model
        self.tokens = 0
        self.chars = 0
        self.start_time = None
        self.first_token_time = None
        
        # HolySheep 2026 pricing
        self.cost_per_token = {
            "deepseek-v3.2": 0.42 / 1_000_000,      # $0.42/MTok
            "gemini-2.5-flash": 2.50 / 1_000_000,
            "gpt-4.1": 8.00 / 1_000_000,
        }
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        self.start_time = time.time()
        self.first_token_time = None
        print(f"[TokenCounter] Model: {self.model}")
        print(f"[TokenCounter] Waiting for response...")
    
    def on_llm_new_token(self, token: str, **kwargs):
        if self.first_token_time is None:
            self.first_token_time = time.time()
            time_to_first_token = (self.first_token_time - self.start_time) * 1000
            print(f"[TokenCounter] First token received in {time_to_first_token:.2f}ms")
        
        self.tokens += 1
        self.chars += len(token)
    
    def on_llm_end(self, response: LLMResult, **kwargs):
        total_time = time.time() - self.start_time
        cost = self.tokens * self.cost_per_token.get(self.model, 0.42 / 1_000_000)
        
        print(f"\n[TokenCounter] === STREAMING COMPLETE ===")
        print(f"  Total tokens: {self.tokens}")
        print(f"  Total characters: {self.chars}")
        print(f"  Total time: {total_time:.2f}s")
        print(f"  Tokens/second: {self.tokens / total_time:.2f}")
        print(f"  Model: {self.model}")
        print(f"  Cost: ${cost:.8f}")
        print(f"  ====================================")

Sử dụng với HolySheep
from langchain_openai import ChatOpenAI
import os

os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

llm = ChatOpenAI(
    model="deepseek-v3.2",
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://api.holysheep.ai/v1",
    streaming=True,
    callbacks=[StreamingTokenCounter(model="deepseek-v3.2")]
)

Test streaming
for chunk in llm.stream("Explain why DeepSeek V3.2 is cost-effective for production:"):
    print(chunk.content, end="", flush=True)

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

Qua kinh nghiệm triển khai LangChain callbacks cho nhiều dự án, tôi đã gặp và xử lý nhiều lỗi phổ biến. Dưới đây là 5 trường hợp điển hình nhất:

1. Lỗi "No module named 'langchain'" hoặc Import Errors

Nguyên nhân: Phiên bản LangChain không tương thích hoặc package chưa được cài đặt đúng cách.

# ❌ CÁCH SAI - Cài đặt không đầy đủ
pip install langchain

✅ CÁCH ĐÚNG - Cài đặt đầy đủ với tất cả dependencies
pip install --upgrade langchain langchain-openai langchain-community
pip install langchain-core langchain-experimental

Hoặc sử dụng requirements.txt:
langchain>=0.1.0
langchain-openai>=0.0.5
langchain-community>=0.0.10
openai>=1.0.0

Kiểm tra version
import langchain
print(f"LangChain version: {langchain.__version__}")

2. Lỗi "Invalid API Key" hoặc Authentication Failed

Nguyên nhân: API key không đúng hoặc base_url bị sai. Lưu ý quan trọng: KHÔNG sử dụng api.openai.com hoặc api.anthropic.com khi dùng HolySheep.

# ❌ CÁCH SAI - Sử dụng sai base_url
llm = ChatOpenAI(
    model="gpt-4.1",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.openai.com/v1"  # ❌ SAI!
)

✅ CÁCH ĐÚNG - Sử dụng HolySheep AI endpoint
import os

Đảm bảo biến môi trường được set đúng
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

llm = ChatOpenAI(
    model="deepseek-v3.2",  # Hoặc "gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash"
    api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"  # ✅ ĐÚNG!
)

Verify connection
try:
    response = llm.invoke("Test connection")
    print("✅ Connection successful!")
except Exception as e:
    print(f"❌ Connection failed: {e}")
    # Kiểm tra lại API key tại: https://www.holysheep.ai/register

3. Lỗi "Callback handler not being called" hoặc Events Not Tracked

Nguyên nhân: Callback được truyền sai vị trí hoặc không được gắn vào đúng chain.

from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from your_callback_module import HolySheepCallbackHandler

Khởi tạo handler
callback_handler = HolySheepCallbackHandler(model_name="deepseek-v3.2")

❌ CÁCH SAI - Callback không được truyền đúng cách
chain = LLMChain(
    llm=llm,
    prompt=prompt
)
result = chain.run("query")  # ❌ Không có callback!

❌ CÁCH SAI 2 - Chỉ truyền callback khi run
result = chain.run("query", callbacks=[callback_handler])  # ❌ Vẫn có thể miss events

✅ CÁCH ĐÚNG - Gắn callback ngay khi khởi tạo chain
chain = LLMChain(
    llm=llm,
    prompt=prompt,
    callbacks=[callback_handler]  # ✅ Gắn callback vào chain
)

Truyền thêm khi run để đảm bảo
result = chain.run(
    "query",
    callbacks=[callback_handler]  # ✅ Callback được đảm bảo
)

Verify events
print(f"Events tracked: {len(callback_handler.events)}")
print(f"Total tokens: {callback_handler.total_tokens}")

Nếu vẫn không có events, kiểm tra:
1. Handler có inherit từ BaseCallbackHandler không
2. Methods có đúng signature không (serialized, prompts, **kwargs)
assert hasattr(callback_handler, 'on_llm_start'), "Missing on_llm_start method"
assert hasattr(callback_handler, 'on_llm_end'), "Missing on_llm_end method"

4. Lỗi "Streaming callback not working" hoặc Token Count = 0

Nguyên nhơn: Model không support streaming hoặc streaming mode chưa được bật.

from langchain_openai import ChatOpenAI

❌ CÁCH SAI - Không bật streaming
llm = ChatOpenAI(
    model="deepseek-v3.2",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    # streaming=False  # Mặc định là False!
)

❌ CÁCH SAI 2 - Callback không có on_llm_new_token
class BrokenCallback:
    def on_llm_start(self, *args, **kwargs):
        print("Started")
    def on_llm_end(self, *args, **kwargs):
        print("Ended")
    # Thiếu on_llm_new_token!

✅ CÁCH ĐÚNG - Bật streaming và implement đúng method
class WorkingStreamingCallback(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs):
        """Method BẮT BUỘC phải có cho streaming"""
        print(f"Token: {token}", end="", flush=True)
    
    def on_llm_start(self, *args, **kwargs):
        print("Starting...")
    
    def on_llm_end(self, *args, **kwargs):
        print("\nDone!")

Initialize với streaming=True
llm = ChatOpenAI(
    model="deepseek-v3.2",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    streaming=True  # ✅ BẬT STREAMING
)

Sử dụng stream() thay vì invoke()
for chunk in llm.stream("Count from 1 to 5:"):
    print(chunk.content, end="", flush=True)

5. Lỗi "Cost calculation incorrect" hoặc Token Usage = None

Nguyên nhân: Response không chứa token usage information hoặc cách trích xuất không đúng.

# ❌ CÁCH SAI - Không kiểm tra token usage structure
def on_llm_end(self, response, **kwargs):
    # Giả định token_usage luôn tồn tại
    tokens = response.generations[0][0].generation_info['token_usage']['total_tokens']
    # ❌ Sẽ crash nếu generation_info = None

✅ CÁCH ĐÚNG - Safe extraction với fallback
def on_llm_end(self, response: LLMResult, **kwargs):
    total_tokens = 0
    cost = 0.0
    
    try:
        # Method 1: Từ generation_info
        for generation_group in response.generations:
            for generation in generation_group:
                if hasattr(generation, 'generation_info') and generation.generation_info:
                    token_info = generation.generation_info.get('token_usage', {})
                    total_tokens = token_info.get('total_tokens', 0)
                    break
        
        # Method 2: Từ llm_output (OpenAI format)
        if total_tokens == 0 and hasattr(response, 'llm_output'):
            if response.llm_output and 'token_usage' in response.llm_output:
                total_tokens = response.llm_output['token_usage'].get('total_tokens', 0)
        
        # Method 3: Estimate từ text length (fallback)
        if total_tokens == 0:
            # Ước tính: ~4 characters per token cho tiếng Anh
            text = response.generations[0][0].text if response.generations else ""
            total_tokens = len(text) // 4
            print(f"[Warning] Using estimated tokens: {total_tokens}")
        
        # Tính cost v
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí
Bài viết liên quan
Hướng Dẫn Sử Dụng Hệ Thống Ticket AI API Toàn Diện Từ HolySh
LangChain Memory: Hướng Dẫn Toàn Diện Về Conversation Contex
Hyperliquid Cấu Trúc Vi Mô Sổ Lệnh: Đo Lường Và Tối Ưu Hóa T