LangChain Callback: ระบบติดตามและมอนิเตอร์การทำงานแบบละเอียด

บทนำ: ทำไมต้องใช้ Callback Mechanism

เมื่อพัฒนาแอปพลิเคชัน LLM ด้วย LangChain การติดตามการทำงานภายใน Chain หรือ Agent เป็นสิ่งจำเป็นอย่างยิ่ง Callback Mechanism ช่วยให้เราสามารถรับ Events ต่างๆ ระหว่างการทำงาน เช่น การเริ่มต้น การสิ้นสุด ข้อผิดพลาด และ Token ที่ใช้งาน ซึ่งข้อมูลเหล่านี้สำคัญมากสำหรับการควบคุมต้นทุนและการ Debug สำหรับการใช้งาน LLM API ที่คุ้มค่า ปี 2026 มีราคาเป็นดังนี้:

• GPT-4.1: $8/ล้าน tokens
• Claude Sonnet 4.5: $15/ล้าน tokens
• Gemini 2.5 Flash: $2.50/ล้าน tokens
• DeepSeek V3.2: $0.42/ล้าน tokens

สำหรับ 10 ล้าน tokens/เดือน ต้นทุนจะแตกต่างกันมาก: DeepSeek ใช้เพียง $4.20 ในขณะที่ Claude Sonnet 4.5 ใช้ถึง $150 คุณสามารถประหยัดได้ถึง 85%+ เมื่อใช้ บริการจาก HolySheep AI ที่รวม Models หลากหลายในราคาพิเศษ พร้อมระบบ WeChat/Alipay และความหน่วงต่ำกว่า 50ms

โครงสร้างพื้นฐานของ Callback Handler

LangChain มี BaseCallbackHandler ที่สามารถ Override methods ต่างๆ ได้ตามต้องการ:

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

class CostTrackingHandler(BaseCallbackHandler):
    """Handler สำหรับติดตามต้นทุนและประสิทธิภาพ"""
    
    def __init__(self):
        super().__init__()
        self.total_tokens = 0
        self.prompt_tokens = 0
        self.completion_tokens = 0
        self.start_time = None
        self.errors = []
        
    def on_llm_start(self, serialized, prompts, **kwargs):
        self.start_time = datetime.now()
        print(f"[LLM START] - Prompts length: {len(prompts[0]) if prompts else 0}")
        
    def on_llm_end(self, response: LLMResult, **kwargs):
        if self.start_time:
            elapsed = (datetime.now() - self.start_time).total_seconds()
            print(f"[LLM END] - Elapsed: {elapsed:.3f}s")
        
        if response.llm_output:
            token_usage = response.llm_output.get('token_usage', {})
            self.prompt_tokens += token_usage.get('prompt_tokens', 0)
            self.completion_tokens += token_usage.get('completion_tokens', 0)
            self.total_tokens += token_usage.get('total_tokens', 0)
            print(f"[TOKENS] - Prompt: {self.prompt_tokens}, Completion: {self.completion_tokens}")
            
    def on_llm_error(self, error, **kwargs):
        error_msg = str(error)
        self.errors.append(error_msg)
        print(f"[LLM ERROR] - {error_msg}")
        
    def on_tool_start(self, serialized, input_str, **kwargs):
        tool_name = serialized.get('name', 'unknown')
        print(f"[TOOL START] - {tool_name}")
        
    def on_tool_end(self, output, **kwargs):
        print(f"[TOOL END] - Output length: {len(output)}")
        
    def get_summary(self):
        return {
            'total_tokens': self.total_tokens,
            'prompt_tokens': self.prompt_tokens,
            'completion_tokens': self.completion_tokens,
            'error_count': len(self.errors),
            'errors': self.errors
        }

การเชื่อมต่อกับ HolySheep AI API

ต่อไปนี้คือตัวอย่างการใช้งาน Callback กับ HolySheep AI ซึ่งให้บริการ Models หลากหลายในราคาประหยัด:

import os
from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.tools import Tool

ตั้งค่า HolySheep AI API
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

สร้าง Callback Handler
tracking_handler = CostTrackingHandler()

เลือก Model ตามความต้องการ
DeepSeek V3.2 - ประหยัดที่สุด $0.42/MTok
llm_deepseek = ChatOpenAI(
    model="deepseek-chat-v3.2",
    temperature=0.7,
    callbacks=[tracking_handler]
)

Gemini 2.5 Flash - สมดุลระหว่างคุณภาพและราคา
llm_gemini = ChatOpenAI(
    model="gemini-2.0-flash",
    temperature=0.7,
    callbacks=[tracking_handler]
)

ตัวอย่างการคำนวณต้นทุน
def calculate_cost(tokens, model):
    pricing = {
        "deepseek-chat-v3.2": 0.42,
        "gemini-2.0-flash": 2.50,
        "gpt-4.1": 8.00,
        "claude-sonnet-4.5": 15.00
    }
    price_per_mtok = pricing.get(model, 1.0)
    return (tokens / 1_000_000) * price_per_mtok

ทดสอบการใช้งาน
test_prompt = "อธิบายหลักการของ Callback Pattern ในภาษา Python"
response = llm_deepseek.call([{"role": "user", "content": test_prompt}])

summary = tracking_handler.get_summary()
cost = calculate_cost(summary['total_tokens'], "deepseek-chat-v3.2")
print(f"Total Cost: ${cost:.4f}")
print(f"Summary: {summary}")

การสร้าง Callback แบบ Streaming

สำหรับแอปพลิเคชันที่ต้องการ Streaming Response การใช้ Callback ช่วยให้ติดตาม Token ที่ส่งมาได้แบบ Real-time:

import os
from langchain.chat_models import ChatOpenAI
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import HumanMessage

class StreamingCallback(BaseCallbackHandler):
    """Handler สำหรับ Streaming Response"""
    
    def __init__(self):
        super().__init__()
        self.received_tokens = []
        self.token_count = 0
        
    def on_llm_new_token(self, token, **kwargs):
        self.received_tokens.append(token)
        self.token_count += 1
        # แสดงผลแบบ Streaming (ลบ end='' เพื่อเห็นการไหลของข้อความ)
        print(token, end='', flush=True)
        
    def on_llm_end(self, response, **kwargs):
        print("\n--- Streaming Complete ---")
        print(f"Total tokens received: {self.token_count}")
        return self.received_tokens

ตั้งค่า API
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"

สร้าง Streaming Handler
streaming_handler = StreamingCallback()

ใช้งาน Streaming Chat
llm_streaming = ChatOpenAI(
    model="deepseek-chat-v3.2",
    temperature=0.7,
    streaming=True,
    callbacks=[streaming_handler]
)

ทดสอบ Streaming
messages = [HumanMessage(content="เล่าสรุปเกี่ยวกับ AI ในปี 2026")]
response = llm_streaming(messages)
print(f"\nFinal response length: {len(response.content)} characters")

การรวม Callback หลายตัว

LangChain รองรับการใช้ Callback หลายตัวพร้อมกันผ่าน CallbackManager:

from langchain.callbacks.manager import CallbackManager
from langchain.schema import HumanMessage

สร้าง Handlers หลายตัว
cost_handler = CostTrackingHandler()
streaming_handler = StreamingCallback()

รวมเข้าด้วยกัน
callback_manager = CallbackManager(
    callbacks=[cost_handler, streaming_handler]
)

ใช้งานกับ Chat Model
llm = ChatOpenAI(
    model="gemini-2.0-flash",
    temperature=0.5,
    callbacks=callback_manager
)

ทดสอบ
messages = [HumanMessage(content="อธิบายเรื่อง LangChain อย่างละเอียด")]
response = llm(messages)

ดึงข้อมูลจาก Handler
print("\n=== Cost Summary ===")
cost_summary = cost_handler.get_summary()
print(f"Tokens used: {cost_summary['total_tokens']}")
print(f"Errors: {cost_summary['error_count']}")

print("\n=== Streaming Summary ===")
print(f"Tokens received: {streaming_handler.token_count}")

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

กรณีที่ 1: AttributeError: 'NoneType' object has no attribute 'get'

สาเหตุ: เมื่อ LLM Response ไม่มี token_usage information (บางครั้ง Model อาจไม่ส่งข้อมูลนี้กลับมา) วิธีแก้ไข: ตรวจสอบก่อนเข้าถึง dictionary:

def on_llm_end(self, response: LLMResult, **kwargs):
    if self.start_time:
        elapsed = (datetime.now() - self.start_time).total_seconds()
        print(f"Elapsed: {elapsed:.3f}s")
    
    # เพิ่มการตรวจสอบความปลอดภัย
    if response.llm_output and response.llm_output.get('token_usage'):
        token_usage = response.llm_output['token_usage']
        self.prompt_tokens += token_usage.get('prompt_tokens', 0)
        self.completion_tokens += token_usage.get('completion_tokens', 0)
        self.total_tokens += token_usage.get('total_tokens', 0)
    else:
        # สำหรับ Model ที่ไม่ส่ง token_usage กลับมา
        print("Token usage information not available from this model")

กรณีที่ 2: RateLimitError หรือ 429 Too Many Requests

สาเหตุ: เกินโควต้าการใช้งานหรือ Rate Limit ของ API วิธีแก้ไข: ใช้ Retry Handler และ Exponential Backoff:

from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult
import time
import random

class RetryCallbackHandler(BaseCallbackHandler):
    """Handler ที่จัดการ Retry อัตโนมัติเมื่อเกิด Rate Limit"""
    
    def __init__(self, max_retries=3, base_delay=1.0):
        super().__init__()
        self.max_retries = max_retries
        self.base_delay = base_delay
        
    def on_llm_error(self, error, **kwargs):
        error_str = str(error).lower()
        
        if 'rate' in error_str or '429' in error_str or 'limit' in error_str:
            print("Rate limit detected. Implementing retry...")
            for attempt in range(self.max_retries):
                # Exponential backoff with jitter
                delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"Retry attempt {attempt + 1}/{self.max_retries} after {delay:.2f}s")
                time.sleep(delay)
                return True  # Return True to indicate retry should be attempted
        return False

ใช้งานร่วมกับ CallbackManager
callback_manager = CallbackManager(
    callbacks=[CostTrackingHandler(), RetryCallbackHandler(max_retries=3)]
)

กรณีที่ 3: Invalid URL หรือ Connection Error

สาเหตุ: base_url ไม่ถูกต้อง หรือ API Key ไม่ถูกต้อง วิธีแก้ไข: ตรวจสอบ Environment Variables และใช้ try-except:

import os
from langchain.chat_models import ChatOpenAI

def initialize_llm_with_validation(model_name: str):
    """Initialize LLM พร้อมการตรวจสอบความถูกต้อง"""
    
    # ตรวจสอบ API Key
    api_key = os.environ.get("HOLYSHEEP_API_KEY") or os.environ.get("OPENAI_API_KEY")
    if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
        raise ValueError(
            "กรุณาตั้งค่า API Key ที่ถูกต้อง\n"
            "ลงทะเบียนที่: https://www.holysheep.ai/register"
        )
    
    # ตรวจสอบ base_url
    base_url = os.environ.get("OPENAI_API_BASE", "https://api.holysheep.ai/v1")
    if not base_url.startswith("https://"):
        raise ValueError("base_url ต้องใช้ https:// เท่านั้น")
    
    try:
        llm = ChatOpenAI(
            model=model_name,
            temperature=0.7,
            callbacks=[CostTrackingHandler()]
        )
        # ทดสอบการเชื่อมต่อ
        test_response = llm.call([{"role": "user", "content": "test"}])
        return llm
    except Exception as e:
        error_msg = str(e)
        if "401" in error_msg or "authentication" in error_msg.lower():
            raise PermissionError("API Key ไม่ถูกต้อง กรุณาตรวจสอบที่ https://www.holysheep.ai/register")
        elif "connection" in error_msg.lower():
            raise ConnectionError("ไม่สามารถเชื่อมต่อ API ได้ กรุณาตรวจสอบการเชื่อมต่ออินเทอร์เน็ต")
        else:
            raise

ตัวอย่างการใช้งาน
try:
    llm = initialize_llm_with_validation("deepseek-chat-v3.2")
    print("เชื่อมต่อสำเร็จ!")
except ValueError as e:
    print(f"Configuration Error: {e}")
except PermissionError as e:
    print(f"Authentication Error: {e}")
except ConnectionError as e:
    print(f"Connection Error: {e}")

สรุป

LangChain Callback Mechanism เป็นเครื่องมือทรงพลังสำหรับการมอนิเตอร์และติดตามการทำงานของ LLM ช่วยให้เราสามารถ:

• ติดตามจำนวน Token ที่ใช้และคำนวณต้นทุนได้อย่างแม่นยำ
• Debug ปัญหาที่เกิดขึ้นระหว่างการทำงาน
• จัดการ Streaming Response แบบ Real-time
• Implement Retry Logic อัตโนมัติสำหรับ Rate Limit

การใช้ HolySheep AI เป็น API Provider ช่วยให้ประหยัดต้นทุนได้มากถึง 85%+ เมื่อเทียบกับการใช้งานโดยตรงจาก OpenAI หรือ Anthropic โดยเฉพาะเมื่อใช้ DeepSeek V3.2 ที่มีราคาเพียง $0.42/ล้าน tokens พร้อมความหน่วงต่ำกว่า 50ms และระบบชำระเงินที่สะดวกผ่าน WeChat/Alipay 👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน