เมื่อวานผมเจอปัญหาที่ทำให้หงุดหงิดมาก — เมื่อรัน production pipeline บน LangChain มี API call หายไป 3 ครั้งโดยไม่มี error message แสดงเลย แค่เห็นแค่ ConnectionError: timeout after 30s ตอนเรียกใช้ GPT-4.1 model ผ่าน HolySheheep AI พอผมเปิด debug mode ด้วย Callback mechanism ถึงรู้ว่า request ที่หายไปมันไปติดที่ rate limit แต่ไม่มี retry logic เลย

บทความนี้ผมจะสอนวิธีใช้ LangChain Callback อย่างละเอียด ครอบคลุมตั้งแต่พื้นฐานจนถึง advanced monitoring พร้อมโค้ดที่ใช้งานได้จริง ประหยัดค่าใช้จ่ายได้มากเพราะ HolySheep AI มีราคาถูกกว่าวงในถึง 85%+ และ latency ต่ำกว่า 50ms

LangChain Callback คืออะไรและทำไมต้องใช้

Callback ใน LangChain คือ event system ที่ช่วยให้เราตรวจสอบทุกขั้นตอนของ LLM chain — ตั้งแต่เริ่ม request, ได้ response, เกิด error, จนถึงจบการทำงาน ถ้าไม่มี callback คุณจะบอกได้แค่ว่า "มันพัง" แต่ไม่รู้ว่า "ทำไม" หรือ "เกิดตรงไหน"

ใน production environment การมี monitoring ที่ดีจะช่วยลดค่าใช้จ่ายได้มหาศาล — ผมเคยเสียเงินไปกับ retry loop ที่ผิดพลาดเกือบ 200 ดอลลาร์ก่อนจะติดตั้ง callback เต็มรูปแบบ

การติดตั้งและเริ่มต้นใช้งาน

ก่อนอื่นติดตั้ง dependencies ที่จำเป็น:

pip install langchain langchain-community langchain-core holyhsheep-ai python-dotenv

สร้างไฟล์ .env เพื่อเก็บ API key อย่างปลอดภัย:

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
LOG_LEVEL=DEBUG

Custom Callback Handler แบบเต็มรูปแบบ

นี่คือโค้ด callback handler ที่ผมใช้ใน production จริง — มันจะ log ทุกอย่างตั้งแต่ token usage, latency, errors, จนถึง cost estimation:

import os
import time
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.messages import BaseMessage
from langchain_core.outputs import LLMResult

class ProductionCallbackHandler(BaseCallbackHandler):
    """Custom callback handler สำหรับ monitoring และ logging"""
    
    def __init__(self, project_name: str = "default"):
        self.project_name = project_name
        self.start_time = None
        self.request_count = 0
        self.error_count = 0
        self.total_tokens = 0
        self.cost_tracker = {}
        
    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs
    ) -> None:
        self.start_time = time.time()
        self.request_count += 1
        print(f"[{datetime.now().isoformat()}] 🔄 LLM Request #{self.request_count}")
        print(f"    Model: {serialized.get('name', 'unknown')}")
        print(f"    Prompt length: {len(prompts[0]) if prompts else 0} chars")
        
    def on_llm_end(self, response: LLMResult, **kwargs) -> None:
        elapsed = time.time() - self.start_time
        print(f"[{datetime.now().isoformat()}] ✅ LLM Response")
        print(f"    Latency: {elapsed*1000:.2f}ms")
        
        if response.llm_output:
            token_usage = response.llm_output.get('token_usage', {})
            prompt_tokens = token_usage.get('prompt_tokens', 0)
            completion_tokens = token_usage.get('completion_tokens', 0)
            total_tokens = token_usage.get('total_tokens', 0)
            
            self.total_tokens += total_tokens
            print(f"    Tokens: {total_tokens} (prompt={prompt_tokens}, completion={completion_tokens})")
            
            # คำนวณค่าใช้จ่าย (ตัวอย่างสำหรับ GPT-4.1)
            cost = self._calculate_cost(prompt_tokens, completion_tokens, "gpt-4.1")
            print(f"    Estimated cost: ${cost:.4f}")
            
    def on_llm_error(
        self, error: Exception, **kwargs
    ) -> None:
        self.error_count += 1
        elapsed = time.time() - self.start_time if self.start_time else 0
        print(f"[{datetime.now().isoformat()}] ❌ LLM Error (#{self.error_count})")
        print(f"    Error type: {type(error).__name__}")
        print(f"    Error message: {str(error)}")
        print(f"    Time before failure: {elapsed*1000:.2f}ms")
        
        # Log error ไปยัง file สำหรับวิเคราะห์
        self._log_error(error, elapsed)
        
    def _calculate_cost(self, prompt_tokens: int, completion_tokens: int, model: str) -> float:
        """คำนวณค่าใช้จ่ายตาม model (ราคาจาก HolySheep AI)"""
        rates = {
            "gpt-4.1": (0.000008, 0.000032),       # $8/MTok in, $32/MTok out
            "claude-sonnet-4.5": (0.000015, 0.000075),  # $15/MTok
            "gemini-2.5-flash": (0.00000125, 0.000005),  # $2.50/MTok
            "deepseek-v3.2": (0.00000042, 0.00000168),   # $0.42/MTok
        }
        
        if model in rates:
            in_rate, out_rate = rates[model]
            return (prompt_tokens * in_rate) + (completion_tokens * out_rate)
        return 0.0
        
    def _log_error(self, error: Exception, elapsed: float) -> None:
        """บันทึก error ลง log file"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "project": self.project_name,
            "error_type": type(error).__name__,
            "error_message": str(error),
            "latency_ms": elapsed * 1000,
            "request_count": self.request_count
        }
        
        with open(f"error_log_{self.project_name}.jsonl", "a") as f:
            f.write(json.dumps(log_entry) + "\n")
            
    def get_summary(self) -> Dict[str, Any]:
        """สรุปสถิติทั้งหมด"""
        return {
            "total_requests": self.request_count,
            "total_errors": self.error_count,
            "total_tokens": self.total_tokens,
            "error_rate": self.error_count / self.request_count if self.request_count > 0 else 0
        }

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

ต่อไปคือการสร้าง LLM instance ที่เชื่อมต่อกับ HolySheep API ซึ่งมี latency เฉลี่ยต่ำกว่า 50ms และราคาถูกกว่าที่อื่นมาก:

import os
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage
from langchain_core.callbacks import StdOutCallbackHandler


ตั้งค่า HolySheep AI base URL

os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"


สร้าง ChatOpenAI instance พร้อม callback handler

callback_handler = ProductionCallbackHandler(project_name="holyhsheep-monitoring")
callbacks = [callback_handler]


เลือก model ตามความต้องการ (GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2)

llm = ChatOpenAI(
    model_name="gpt-4.1",  # $8/MTok (ประหยัด 85%+ เทียบกับ OpenAI)
    temperature=0.7,
    max_tokens=2000,
    callbacks=callbacks
)


ทดสอบการทำงาน

try:
    response = llm.invoke([HumanMessage(content="อธิบายเรื่อง LangChain Callback")])
    print(f"\n📝 Response: {response.content[:100]}...")
except Exception as e:
    print(f"\n💥 Error occurred: {type(e).__name__}: {e}")


แสดงสรุปสถิติ

summary = callback_handler.get_summary()
print(f"\n📊 Summary: {summary}")

การใช้งาน Callback ใน Chain และ Agent

สำหรับ LangChain Expression Language (LCEL) chains คุณสามารถส่ง callback ผ่าน method chaining ได้เลย:

from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser
from langchain_core.callbacks import BaseCallbackHandler

class DebugCallback(BaseCallbackHandler):
    """Lightweight callback สำหรับ chain debugging"""
    
    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"🔗 Chain started with inputs: {list(inputs.keys())}")
        
    def on_chain_end(self, outputs, **kwargs):
        print(f"🔗 Chain completed, outputs: {list(outputs.keys())}")


สร้าง chain พร้อม debug callback

prompt = ChatPromptTemplate.from_messages([
    ("system", "คุณคือผู้ช่วย AI ที่ตอบเป็นภาษาไทย"),
    ("human", "{question}")
])

chain = prompt | ChatOpenAI(
    model_name="gemini-2.5-flash",  # $2.50/MTok — ราคาถูกมาก!
    callbacks=[DebugCallback()]
) | StrOutputParser()


รัน chain

result = chain.invoke(
    {"question": "LangChain Callback ทำงานอย่างไร?"},
    config={"callbacks": [ProductionCallbackHandler()]}
)
print(f"\n✨ Result: {result}")

การติดตาม Token Usage และ Cost

นี่คือสคริปต์ที่ผมใช้ติดตามค่าใช้จ่ายจริงในโปรเจกต์ — มันจะบันทึก usage ทุกครั้งและสรุปเป็นรายวัน:

import csv
from datetime import datetime, timedelta
from collections import defaultdict

class CostTracker:
    """ติดตามค่าใช้จ่ายจริงและวิเคราะห์การใช้งาน"""
    
    # ราคาจาก HolySheep AI (อัปเดต 2026)
    MODEL_RATES = {
        "gpt-4.1": {"prompt": 0.000008, "completion": 0.000032},
        "claude-sonnet-4.5": {"prompt": 0.000015, "completion": 0.000075},
        "gemini-2.5-flash": {"prompt": 0.00000125, "completion": 0.000005},
        "deepseek-v3.2": {"prompt": 0.00000042, "completion": 0.00000168},
    }
    
    def __init__(self, csv_file: str = "cost_report.csv"):
        self.csv_file = csv_file
        self.daily_usage = defaultdict(lambda: {
            "requests": 0, "prompt_tokens": 0, "completion_tokens": 0, "cost": 0
        })
        self._init_csv()
        
    def _init_csv(self):
        with open(self.csv_file, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow([
                "timestamp", "model", "prompt_tokens", "completion_tokens",
                "total_tokens", "cost_usd", "latency_ms"
            ])
    
    def record_usage(
        self, model: str, prompt_tokens: int, completion_tokens: int,
        cost_usd: float, latency_ms: float
    ):
        today = datetime.now().strftime("%Y-%m-%d")
        
        self.daily_usage[today]["requests"] += 1
        self.daily_usage[today]["prompt_tokens"] += prompt_tokens
        self.daily_usage[today]["completion_tokens"] += completion_tokens
        self.daily_usage[today]["cost"] += cost_usd
        
        with open(self.csv_file, 'a', newline='') as f:
            writer = csv.writer(f)
            writer.writerow([
                datetime.now().isoformat(), model, prompt_tokens,
                completion_tokens, prompt_tokens + completion_tokens,
                cost_usd, latency_ms
            ])
    
    def get_daily_report(self, days: int = 7) -> str:
        report_lines = ["📊 Cost Report (Last 7 days)"]
        report_lines.append("=" * 50)
        
        total_cost = 0
        for i in range(days):
            date = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d")
            usage = self.daily_usage.get(date, {})
            if usage["requests"] > 0:
                report_lines.append(f"\n📅 {date}")
                report_lines.append(f"   Requests: {usage['requests']}")
                report_lines.append(f"   Tokens: {usage['prompt_tokens'] + usage['completion_tokens']:,}")
                report_lines.append(f"   Cost: ${usage['cost']:.4f}")
                total_cost += usage["cost"]
                
        report_lines.append(f"\n💰 Total (7 days): ${total_cost:.4f}")
        return "\n".join(report_lines)


ใช้งาน

tracker = CostTracker()


บันทึก usage หลังจากเรียก API

tracker.record_usage(
    model="deepseek-v3.2",
    prompt_tokens=150,
    completion_tokens=280,
    cost_usd=0.000218,  # $0.42/MTok = ประหยัดมาก!
    latency_ms=38.5
)

print(tracker.get_daily_report())

Retry Logic พร้อม Callback

นี่คือ pattern ที่ผมแนะนำสำหรับ handling rate limit และ timeout — มันใช้ exponential backoff และ callback สำหรับ logging:

import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from langchain_core.callbacks import BaseCallbackHandler

class RetryCallback(BaseCallbackHandler):
    """Callback สำหรับติดตาม retry attempts"""
    
    def __init__(self):
        self.attempts = {}
        
    def on_llm_retry(self, attempt_number: int, error: Exception, **kwargs):
        wait_time = 2 ** attempt_number  # exponential backoff
        print(f"🔁 Retry #{attempt_number} after {wait_time}s")
        print(f"   Error: {type(error).__name__}: {str(error)[:100]}")
        
        if attempt_number not in self.attempts:
            self.attempts[attempt_number] = []
        self.attempts[attempt_number].append({
            "error": str(error),
            "timestamp": datetime.now().isoformat()
        })


Retry decorator

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((ConnectionError, TimeoutError, RateLimitError)),
    before_sleep=lambda retry_state: print(f"⏳ Waiting {retry_state.next_action.sleep}s before retry...")
)
def call_llm_with_retry(prompt: str, model: str = "gpt-4.1") -> str:
    llm = ChatOpenAI(
        model_name=model,
        callbacks=[RetryCallback()],
        request_timeout=30
    )
    response = llm.invoke([HumanMessage(content=prompt)])
    return response.content


ทดสอบ

try:
    result = call_llm_with_retry("ทดสอบ retry logic")
    print(f"✅ Success: {result[:50]}...")
except Exception as e:
    print(f"❌ All retries failed: {type(e).__name__}: {e}")

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

1. ConnectionError: timeout after 30s

สาเหตุ: Default timeout ของ LangChain อยู่ที่ 30 วินาที ซึ่งอาจไม่พอสำหรับ cold start หรือ high load

# วิธีแก้: เพิ่ม timeout และใช้ retry
llm = ChatOpenAI(
    model_name="gpt-4.1",
    max_retries=3,
    request_timeout=120,  # เพิ่ม timeout เป็น 120 วินาที
    callbacks=[ProductionCallbackHandler()]
)


หรือตั้งค่าผ่าน environment variable

os.environ["OPENAI_TIMEOUT"] = "120"

2. 401 Unauthorized / Invalid API Key

สาเหตุ: API key ไม่ถูกต้องหรือไม่ได้ตั้งค่า environment variable

# วิธีแก้: ตรวจสอบว่า base_url และ api_key ถูกต้อง
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"  # ตรวจสอบว่าไม่มีช่องว่าง


หรือส่ง trực tiếp ใน constructor

llm = ChatOpenAI(
    openai_api_base="https://api.holysheep.ai/v1",
    openai_api_key="YOUR_HOLYSHEEP_API_KEY",
    model_name="gpt-4.1"
)


ตรวจสอบว่าใช้งานได้

import os
print(f"API Base: {os.environ.get('OPENAI_API_BASE')}")
print(f"API Key set: {'Yes' if os.environ.get('OPENAI_API_KEY') else 'No'}")

3. RateLimitError: 429 Too Many Requests

สาเหตุ: เรียก API เร็วเกินไปเกิน rate limit ของ plan

# วิธีแก้: ใช้ rate limiter และ retry with backoff
import asyncio
from collections import defaultdict
from time import time

class RateLimiter:
    def __init__(self, max_requests: int = 100, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests = defaultdict(list)
        
    async def acquire(self):
        now = time()
        # ลบ requests ที่เก่ากว่า window
        self.requests["default"] = [
            r for r in self.requests["default"] if now - r < self.window
        ]
        
        if len(self.requests["default"]) >= self.max_requests:
            sleep_time = self.window - (now - self.requests["default"][0])
            print(f"⏳ Rate limit reached, sleeping {sleep_time:.1f}s")
            await asyncio.sleep(sleep_time)
            
        self.requests["default"].append(time())


ใช้งาน

rate_limiter = RateLimiter(max_requests=100, window_seconds=60)

async def bounded_llm_call(prompt: str):
    await rate_limiter.acquire()
    llm = ChatOpenAI(model_name="deepseek-v3.2")  # ประหยัดที่สุด: $0.42/MTok
    return await llm.ainvoke([HumanMessage(content=prompt)])


รัน multiple requests

asyncio.run(bounded_llm_call("ทดสอบ rate limiting"))

สรุป

LangChain Callback mechanism เป็นเครื่องมือที่จำเป็นสำหรับทุกคนที่ใช้ LLM ใน production — ช่วยให้ตรวจสอบได้ว่าเกิดอะไรขึ้น เกิดที่ไหน และทำไม ลดค่าใช้จ่ายจากการ retry ที่ผิดพลาด และช่วยวิเคราะห์ปัญหาได้รวดเร็ว

สำหรับการใช้งานจริง ผมแนะนำให้ใช้ HolySheep AI เพราะมี latency เฉลี่ยต่ำกว่า 50ms และราคาประหยัดกว่าถึง 85%+ เทียบกับ OpenAI โดยเฉพาะ DeepSeek V3.2 ที่ราคาเพียง $0.42/MTok เหมาะมากสำหรับ high-volume applications

อย่าลืมติดตั้ง callback handler และ retry logic ก่อนนำไปใช้ใน production — มันจะช่วยประหยัดเวลาและเงินได้มากในระยะยาว

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง