LangChain Callback กลไกสำคัญสำหรับการมอนิเตอร์และบันทึก Log การเรียก LLM

บทนำ: เมื่อระบบ AI ลูกค้าสัมพันธ์ของร้านค้าออนไลน์ล่มกลางคัน

สมมติว่าคุณเป็น Tech Lead ของร้านค้าออนไลน์ขนาดใหญ่แห่งหนึ่ง คุณเพิ่ง deploy ระบบ AI Chatbot สำหรับตอบคำถามลูกค้า 24/7 โดยใช้ LangChain ต่อกับ HolySheep AI เป็น backend วันแรกทำงานราบรื่น แต่สัปดาห์ต่อมา ลูกค้าจำนวนมากเริ่มบ่นว่า bot ตอบช้า บางคนได้รับคำตอบที่ไม่เกี่ยวข้อง ทีม DevOps ต้องการรู้ว่าเกิดอะไรขึ้น แต่ log เดิมไม่มีข้อมูลเพียงพอ นี่คือจุดที่ LangChain Callback Mechanism เข้ามาช่วยได้ บทความนี้จะพาคุณเจาะลึกกลไก callback ใน LangChain ตั้งแต่พื้นฐานจนถึงการ implement ระบบ monitoring ที่พร้อมใช้งานจริง โดยใช้ HolySheep AI เป็น LLM provider ที่มี latency เฉลี่ยต่ำกว่า 50ms ทำให้การ tracking ทำได้แม่นยำและรวดเร็ว

LangChain Callback คืออะไรและทำงานอย่างไร

LangChain Callback เป็นระบบ event-driven ที่ช่วยให้คุณ "สอดส่อง" ทุกขั้นตอนของ chain ตั้งแต่เริ่มเรียก LLM ไปจนถึงได้ response กลับมา คุณสามารถ intercept event ต่างๆ เช่น on_llm_start, on_llm_end, on_chain_start, on_chain_end เพื่อบันทึก metrics, วัดความเร็ว, หรือแม้แต่ส่ง alert เมื่อเกิดข้อผิดพลาด การใช้งาน callback ทำให้คุณมองเห็นภาพรวมของ LLM usage ได้อย่างชัดเจน รู้ว่า token ใช้ไปเท่าไหร่, latency เท่าไหร่, prompt ที่ส่งไปมีขนาดเท่าไหร่ ข้อมูลเหล่านี้สำคัญมากสำหรับการ optimize ต้นทุน โดยเฉพาะเมื่อใช้ HolySheep AI ที่มีราคาคุ้มค่าอย่าง DeepSeek V3.2 อยู่ที่ $0.42/MTok เท่านั้น

การติดตั้งและ Setup เบื้องต้น

ก่อนจะเข้าสู่ส่วน implementation มาดู dependencies ที่ต้องเตรียมกันก่อน:

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

จากนั้นสร้างไฟล์ .env สำหรับเก็บ API key:

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1

Custom Callback Handler สำหรับ Monitoring

มาเริ่มสร้าง callback handler ที่จะบันทึกข้อมูลทุกการเรียก LLM กันเลย:

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

class LLMMonitoringCallback(BaseCallbackHandler):
    """Callback handler สำหรับ monitor การเรียก LLM แบบละเอียด"""
    
    def __init__(self):
        super().__init__()
        self.llm_calls = []
        self.total_tokens = 0
        self.total_latency_ms = 0.0
        self.error_count = 0
        
    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs
    ) -> None:
        """เริ่มเรียก LLM - บันทึก timestamp และ prompt"""
        self.llm_calls.append({
            "event": "start",
            "timestamp": datetime.now().isoformat(),
            "prompt_count": len(prompts),
            "total_prompt_chars": sum(len(p) for p in prompts),
            "model": serialized.get("name", "unknown")
        })
        
    def on_llm_end(
        self, response: LLMResult, **kwargs
    ) -> None:
        """LLM ตอบกลับมาแล้ว - บันทึก response และ token usage"""
        if not self.llm_calls:
            return
            
        start_event = self.llm_calls[-1]
        end_time = datetime.now()
        
        # คำนวณ latency
        start_time = datetime.fromisoformat(start_event["timestamp"])
        latency_ms = (end_time - start_time).total_seconds() * 1000
        self.total_latency_ms += latency_ms
        
        # ดึง token usage จาก response
        token_usage = {}
        if response.llm_output and "token_usage" in response.llm_output:
            tu = response.llm_output["token_usage"]
            token_usage = {
                "prompt_tokens": tu.get("prompt_tokens", 0),
                "completion_tokens": tu.get("completion_tokens", 0),
                "total_tokens": tu.get("total_tokens", 0)
            }
            self.total_tokens += token_usage.get("total_tokens", 0)
        
        # อัพเดท event สุดท้าย
        self.llm_calls[-1].update({
            "event": "end",
            "latency_ms": round(latency_ms, 2),
            "token_usage": token_usage,
            "response_preview": response.generations[0][0].text[:200] if response.generations else ""
        })
        
    def on_llm_error(
        self, error: Exception, **kwargs
    ) -> None:
        """เกิด error - บันทึกเพื่อ debug"""
        self.error_count += 1
        if self.llm_calls:
            self.llm_calls[-1].update({
                "event": "error",
                "error_type": type(error).__name__,
                "error_message": str(error)[:500]
            })
            
    def get_summary(self) -> Dict[str, Any]:
        """สรุปผลการ monitoring"""
        return {
            "total_calls": len([c for c in self.llm_calls if c["event"] == "end"]),
            "total_tokens": self.total_tokens,
            "avg_latency_ms": round(
                self.total_latency_ms / max(len([c for c in self.llm_calls if c["event"] == "end"]), 1), 
                2
            ),
            "error_count": self.error_count,
            "cost_estimate_usd": round(self.total_tokens / 1_000_000 * 8, 6)  # ประมาณการด้วย GPT-4.1 rate
        }

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

ต่อไปมาดูวิธีการสร้าง LLM chain ที่ใช้ HolySheep API โดยผ่าน LangChain:

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

สร้าง LLM instance ที่เชื่อมต่อกับ HolySheep AI
llm = ChatOpenAI(
    model="gpt-4.1",
    base_url="https://api.holysheep.ai/v1",
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    timeout=60,
    max_retries=3
)

สร้าง simple chain สำหรับทดสอบ
prompt = ChatPromptTemplate.from_template(
    "ตอบคำถามต่อไปนี้ในภาษาไทย: {question}"
)

chain = prompt | llm | StrOutputParser()

ทดสอบการเรียกพร้อม callback
monitor_callback = LLMMonitoringCallback()

question = "บริการขนส่งของร้านค้าออนไลน์มีค่าธรรมเนียมเท่าไหร่?"
response = chain.invoke(
    {"question": question},
    config={"callbacks": [monitor_callback]}
)

print(f"Response: {response}")
print(f"Summary: {monitor_callback.get_summary()}")

ระบบ RAG Monitoring สำหรับ Enterprise

สำหรับโปรเจกต์ที่ใช้ RAG (Retrieval-Augmented Generation) การมอนิเตอร์ยิ่งสำคัญกว่า เพราะมีหลายขั้นตอนที่ต้อง track:

from langchain_core.documents import Document
from langchain_community.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

class RAGMonitoringCallback(BaseCallbackHandler):
    """Enhanced callback สำหรับ RAG system monitoring"""
    
    def __init__(self, log_file: str = "rag_metrics.jsonl"):
        super().__init__()
        self.log_file = log_file
        self.metrics = {
            "retrieval_times": [],
            "llm_times": [],
            "total_costs": {},
            "errors": []
        }
        
    def on_retriever_start(self, query: str) -> None:
        """เริ่มค้นหา document"""
        self.retrieval_start = time.time()
        
    def on_retriever_end(
        self, documents: List[Document], **kwargs
    ) -> None:
        """ได้ผลลัพธ์การค้นหา"""
        retrieval_time = (time.time() - self.retrieval_start) * 1000
        self.metrics["retrieval_times"].append(retrieval_time)
        print(f"[RETRIEVAL] Found {len(documents)} docs in {retrieval_time:.2f}ms")
        
    def on_llm_start(self, serialized: Dict, prompts: List[str], **kwargs) -> None:
        """เริ่มเรียก LLM"""
        self.llm_start = time.time()
        
    def on_llm_end(self, response: LLMResult, **kwargs) -> None:
        """LLM ตอบกลับ"""
        llm_time = (time.time() - self.llm_start) * 1000
        self.metrics["llm_times"].append(llm_time)
        
        # คำนวณ cost
        if response.llm_output and "token_usage" in response.llm_output:
            tu = response.llm_output["token_usage"]
            total_tokens = tu.get("total_tokens", 0)
            model = serialized.get("name", "gpt-4.1")
            cost_per_mtok = {
                "gpt-4.1": 8.0,
                "gpt-4.1-mini": 2.0,
                "deepseek-v3.2": 0.42
            }.get(model, 8.0)
            cost = (total_tokens / 1_000_000) * cost_per_mtok
            
            if model not in self.metrics["total_costs"]:
                self.metrics["total_costs"][model] = 0.0
            self.metrics["total_costs"][model] += cost
            
    def log_metrics(self) -> None:
        """บันทึก metrics ลงไฟล์"""
        with open(self.log_file, "a") as f:
            f.write(json.dumps({
                "timestamp": datetime.now().isoformat(),
                "metrics": self.metrics
            }) + "\n")
            
    def get_report(self) -> str:
        """สร้างรายงานสรุป"""
        avg_retrieval = sum(self.metrics["retrieval_times"]) / max(len(self.metrics["retrieval_times"]), 1)
        avg_llm = sum(self.metrics["llm_times"]) / max(len(self.metrics["llm_times"]), 1)
        total_cost = sum(self.metrics["total_costs"].values())
        
        return f"""
╔══════════════════════════════════════════════════════╗
║              RAG SYSTEM MONITORING REPORT            ║
╠══════════════════════════════════════════════════════╣
║  Total Retrievals: {len(self.metrics["retrieval_times"]):>33} ║
║  Avg Retrieval Time: {avg_retrieval:>28.2f} ms ║
║  Total LLM Calls: {len(self.metrics["llm_times"]):>34} ║
║  Avg LLM Latency: {avg_llm:>33.2f} ms ║
║  Total Cost: ${total_cost:>37.2f} ║
║  Total Errors: {len(self.metrics["errors"]):>36} ║
╚══════════════════════════════════════════════════════╝
"""

การส่ง Alert เมื่อเกิดปัญหา

นอกจากการบันทึก log แล้ว คุณควรมีระบบ alert เมื่อ metrics เกิน threshold ที่กำหนด:

import smtplib
from email.mime.text import MIMEText
from dataclasses import dataclass

@dataclass
class AlertConfig:
    latency_threshold_ms: float = 5000.0  # 5 วินาที
    error_rate_threshold: float = 0.05  # 5%
    token_budget_warning: int = 1_000_000  # 1M tokens

class AlertCallback(LLMMonitoringCallback):
    """Callback ที่ส่ง alert เมื่อเกิดปัญหา"""
    
    def __init__(self, alert_config: AlertConfig = None):
        super().__init__()
        self.config = alert_config or AlertConfig()
        self.alert_history = []
        
    def _send_alert(self, title: str, message: str, severity: str = "WARNING"):
        """ส่ง alert notification"""
        alert = {
            "timestamp": datetime.now().isoformat(),
            "severity": severity,
            "title": title,
            "message": message
        }
        self.alert_history.append(alert)
        # ใน production ควรส่งไปยัง Slack, PagerDuty, หรือ email
        print(f"[ALERT][{severity}] {title}: {message}")
        
    def on_llm_end(self, response: LLMResult, **kwargs) -> None:
        """ตรวจสอบ latency หลัง LLM ตอบกลับ"""
        super().on_llm_end(response, kwargs)
        
        # หา event ล่าสุด
        start_event = None
        for i in range(len(self.llm_calls) - 1, -1, -1):
            if self.llm_calls[i].get("event") == "start":
                start_event = self.llm_calls[i]
                break
                
        if start_event and "latency_ms" in start_event:
            if start_event["latency_ms"] > self.config.latency_threshold_ms:
                self._send_alert(
                    title="High LLM Latency Detected",
                    message=f"Latency: {start_event['latency_ms']}ms (threshold: {self.config.latency_threshold_ms}ms)",
                    severity="WARNING"
                )
                
    def check_token_budget(self, current_tokens: int):
        """ตรวจสอบ token budget"""
        if current_tokens > self.config.token_budget_warning:
            self._send_alert(
                title="Token Budget Warning",
                message=f"Used {current_tokens:,} tokens (warning at {self.config.token_budget_warning:,})",
                severity="INFO"
            )

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

กรณีที่ 1: API Key ไม่ถูกต้องหรือหมดอายุ

# ❌ ข้อผิดพลาดที่พบบ่อย
AuthenticationError: Incorrect API key provided

✅ วิธีแก้ไข: ตรวจสอบ environment variable
import os
print(f"API Key loaded: {os.getenv('HOLYSHEEP_API_KEY')[:10]}...")

หรือใช้ LangChain environment variable
os.environ["OPENAI_API_KEY"] = os.getenv("HOLYSHEEP_API_KEY")

ตรวจสอบว่า key ขึ้นต้นด้วย pattern ที่ถูกต้อง
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key or len(api_key) < 20:
    raise ValueError("Invalid API key format. Please check your HolySheep AI key.")

กรรมที่ 2: Callback ไม่ทำงานเมื่อใช้ .invoke() แทน .astream()

# ❌ ข้อผิดพลาด: Callback ไม่ถูกเรียก
เพราะบาง method ไม่ส่ง callback ไปยัง LLM

✅ วิธีแก้ไข: ใช้ config parameter อย่างถูกต้อง
from langchain_core.runnables import RunnableConfig

วิธีที่ 1: ส่งผ่าน config dict
response = chain.invoke(
    {"question": "What is SEO?"},
    config={"callbacks": [monitor_callback]}
)

วิธีที่ 2: ใช้ with_config method
response = chain.with_config(callbacks=[monitor_callback]).invoke({"question": "What is SEO?"})

วิธีที่ 3: สำหรับ async operations ใช้ ainvoke
response = await chain.ainvoke(
    {"question": "What is SEO?"},
    config={"callbacks": [monitor_callback]}
)

กรณีที่ 3: Token Usage ไม่ถูกบันทึกจาก API Response

# ❌ ข้อผิดพลาด: token_usage เป็น None เสมอ
เกิดจาก response format ที่ไม่ตรงกับที่ LangChain คาดหวัง

✅ วิธีแก้ไข: ตรวจสอบและ parse response อย่างถูกต้อง
def safe_get_token_usage(response: LLMResult) -> dict:
    """ดึง token usage อย่างปลอดภัย"""
    if not response.llm_output:
        return {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
    
    # HolySheep API ส่ง token_usage ในรูปแบบนี้
    tu = response.llm_output.get("token_usage", {})
    
    # บางครั้ง model ส่งค่ามาเป็น string
    return {
        "prompt_tokens": int(tu.get("prompt_tokens", 0)),
        "completion_tokens": int(tu.get("completion_tokens", 0)),
        "total_tokens": int(tu.get("total_tokens", 0))
    }

ทดสอบดึงค่า
usage = safe_get_token_usage(response)
print(f"Tokens used: {usage['total_tokens']}")

กรณีที่ 4: Memory Leak จาก Callback ที่เก็บข้อมูลมากเกินไป

# ❌ ข้อผิดพลาด: memory usage เพิ่มขึ้นเรื่อยๆ
เพราะเก็บทุก call ไว้ใน memory โดยไม่มี limit

✅ วิธีแก้ไข: ใช้ bounded storage
from collections import deque

class BoundedMonitoringCallback(BaseCallbackHandler):
    """Callback ที่จำกัดขนาดของ stored data"""
    
    def __init__(self, max_calls: int = 1000):
        super().__init__()
        self.max_calls = max_calls
        self.llm_calls = deque(maxlen=max_calls)  # ลบของเก่าอัตโนมัติ
        
    def on_llm_end(self, response: LLMResult, **kwargs) -> None:
        # ... logic บันทึก ...
        
        # เมื่อถึง limit ให้ flush ลง disk
        if len(self.llm_calls) >= self.max_calls:
            self._flush_to_disk()
            
    def _flush_to_disk(self):
        """บันทึกข้อมูลลง disk แล้ว clear memory"""
        with open("llm_calls_archive.jsonl", "a") as f:
            for call in self.llm_calls:
                f.write(json.dumps(call) + "\n")
        self.llm_calls.clear()
        print(f"[INFO] Flushed {self.max_calls} calls to disk")

Best Practices สำหรับ Production

เมื่อนำ callback ไปใช้ใน production มีสิ่งที่ควรคำนึงถึงดังนี้:

• Async Support - หากระบบของคุณใช้ async operations ควร implement on_llm_new_token ด้วยเพื่อ stream tokens ออกไปได้ทันที
• Structured Logging - ใช้ format ที่เป็นมาตรฐาน เช่น JSON lines เพื่อให้ parse ด้วย log aggregation tools ได้ง่าย
• Correlation ID - เพิ่ม request_id ให้ทุก call เพื่อ trace การทำงานข้าม services ได้
• Sample Rate - หาก traffic สูงมาก อาจใช้ sampling แทนการ monitor ทุก request
• Cost Alerting - ตั้ง threshold สำหรับค่าใช้จ่ายรายวัน/รายเดือน เพื่อป้องกัน bill shock

สรุป

LangChain Callback Mechanism เป็นเครื่องมือทรงพลังสำหรับการมอนิเตอร์และ debug ระบบ LLM ช่วยให้คุณเห็นภาพรวมของ token usage, latency, และ error patterns ได้อย่างชัดเจน การ implement callback ที่ดีไม่เพียงช่วยแก้ปัญหา production แต่ยังเป็นพื้นฐานสำหรับการ optimize ต้นทุนและปรับปรุงประสิทธิภาพอีกด้วย หากคุณกำลังมองหา LLM provider ที่มีความเสถียรและราคาคุ้มค่า HolySheep AI เป็นตัวเลือกที่น่าสนใจ โดยมีราคาที่ประหยัดกว่า OpenAI ถึง 85%+ เช่น DeepSeek V3.2 อยู่ที่ $0.42/MTok เท่านั้น รองรับ WeChat และ Alipay พร้อม latency เฉลี่ยต่ำกว่า 50ms และเครดิตฟรีเมื่อลงทะเบียน 👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน