ในยุคที่ LLM-powered application กลายเป็นหัวใจหลักของธุรกิจดิจิทัล การมีระบบติดตาม event และ logging ที่แข็งแกร่งไม่ใช่ทางเลือกอีกต่อไป แต่เป็นความจำเป็น บทความนี้จะพาคุณเจาะลึกการใช้งาน LangChain callbacks ตั้งแต่พื้นฐานจนถึง production-ready implementation พร้อมตัวอย่างโค้ดที่พร้อมใช้งานจริง

กรณีศึกษา: ทีมสตาร์ทอัพ AI ในกรุงเทพฯ

บริบทธุรกิจ

ทีมพัฒนา AI chatbot สำหรับธุรกิจอีคอมเมิร์ซรายใหญ่ในกรุงเทพฯ มีผู้ใช้งาน active กว่า 500,000 รายต่อเดือน แชทบอทต้องตอบคำถามลูกค้าเกี่ยวกับสินค้า สถานะคำสั่งซื้อ และการจัดการคืนสินค้าได้อย่างรวดเร็วและแม่นยำ

จุดเจ็บปวดของระบบเดิม

ก่อนหน้านี้ ทีมใช้ OpenAI API โดยตรงและพบปัญหาหลายประการ:

การย้ายมาใช้ HolySheep AI

หลังจากทดสอบและเปรียบเทียบหลายผู้ให้บริการ ทีมตัดสินใจย้ายมาใช้ HolySheep AI เนื่องจาก:

ขั้นตอนการย้ายระบบ

การย้ายระบบทำผ่าน 3 ขั้นตอนหลัก:

1. เปลี่ยน base_url — แก้ไข configuration จาก api.openai.com เป็น https://api.holysheep.ai/v1

2. หมุนคีย์ API — สร้าง API key ใหม่จาก HolySheep dashboard และอัพเดท environment variables

3. Canary Deploy — ทดสอบกับ traffic 10% ก่อนขยายเป็น 100%

ผลลัพธ์ 30 วันหลังการย้าย

ตัวชี้วัดหลังการย้ายแสดงการปรับปรุงอย่างเห็นได้ชัด:

ทำความเข้าใจ LangChain Callbacks พื้นฐาน

LangChain callbacks เป็นระบบ event-driven ที่ช่วยให้คุณสามารถ "listen" เหตุการณ์ต่างๆ ที่เกิดขึ้นระหว่างการทำงานของ LLM chain ได้ ก่อนจะเข้าสู่โค้ดจริง มาทำความเข้าใจ callback events หลักๆ ที่คุณควรรู้จัก:

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

มาเริ่มต้นด้วยตัวอย่างที่ง่ายที่สุด — การสร้าง callback handler สำหรับ tracking LLM calls:

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


ตั้งค่า logger

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProductionCallbackHandler(BaseCallbackHandler):
    """
    Callback handler สำหรับ production environment
    - บันทึก log ทุก event
    - Track token usage
    - วัดความล่าช้า
    - จับ error cases
    """
    
    def __init__(self):
        super().__init__()
        self.start_time: Optional[datetime] = None
        self.token_count: int = 0
        self.prompt_tokens: int = 0
        self.completion_tokens: int = 0
        self.events: List[Dict] = []
    
    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        prompts: List[str],
        **kwargs: Any
    ) -> None:
        """เริ่มต้น LLM call — บันทึกเวลาและ prompt"""
        self.start_time = datetime.now()
        self.events.append({
            "event": "llm_start",
            "timestamp": self.start_time.isoformat(),
            "prompt_length": len(prompts[0]) if prompts else 0
        })
        logger.info(f"🔄 LLM call started at {self.start_time}")
    
    def on_llm_end(
        self,
        response: LLMResult,
        **kwargs: Any
    ) -> None:
        """LMM call เสร็จสิ้น — คำนวณ latency และ token usage"""
        if self.start_time:
            end_time = datetime.now()
            latency_ms = (end_time - self.start_time).total_seconds() * 1000
            
            # ดึง token usage จาก response
            if response.llm_output and "token_usage" in response.llm_output:
                usage = response.llm_output["token_usage"]
                self.prompt_tokens = usage.get("prompt_tokens", 0)
                self.completion_tokens = usage.get("completion_tokens", 0)
                self.token_count = usage.get("total_tokens", 0)
            
            self.events.append({
                "event": "llm_end",
                "timestamp": end_time.isoformat(),
                "latency_ms": latency_ms,
                "token_count": self.token_count,
                "prompt_tokens": self.prompt_tokens,
                "completion_tokens": self.completion_tokens
            })
            
            logger.info(
                f"✅ LLM call completed in {latency_ms:.2f}ms | "
                f"Tokens: {self.token_count} (prompt: {self.prompt_tokens}, "
                f"completion: {self.completion_tokens})"
            )
    
    def on_llm_error(
        self,
        error: BaseException,
        **kwargs: Any
    ) -> None:
        """จับ error ที่เกิดขึ้น"""
        self.events.append({
            "event": "llm_error",
            "error_type": type(error).__name__,
            "error_message": str(error),
            "timestamp": datetime.now().isoformat()
        })
        logger.error(f"❌ LLM Error: {error}")
    
    def get_summary(self) -> Dict[str, Any]:
        """สรุป metrics ทั้งหมด"""
        return {
            "total_events": len(self.events),
            "total_tokens": self.token_count,
            "prompt_tokens": self.prompt_tokens,
            "completion_tokens": self.completion_tokens,
            "events": self.events
        }

การติดตั้ง Callback กับ Chain

หลังจากสร้าง callback handler แล้ว ต่อไปคือการนำไปใช้กับ LangChain chain จริง:

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


ตั้งค่า HolySheep API

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


สร้าง callback handler

callback_handler = ProductionCallbackHandler()


สร้าง LLM instance พร้อม callback

llm = ChatOpenAI(
    model="gpt-4.1",
    temperature=0.7,
    callbacks=[callback_handler]  # ผูก callback ที่นี่
)


สร้าง chain

prompt = PromptTemplate(
    input_variables=["topic"],
    template="อธิบายเกี่ยวกับ {topic} อย่างกระชับ 3 ประโยค"
)

chain = LLMChain(llm=llm, prompt=prompt)


รัน chain

result = chain.run(topic="LangChain callbacks")
print(f"Result: {result}")


ดู summary

summary = callback_handler.get_summary()
print(f"\n📊 Summary: {json.dumps(summary, indent=2)}")

Callback สำหรับ Agent และ Tools

สำหรับ application ที่ซับซ้อนกว่า เช่น agent ที่ใช้ tools หลายตัว คุณต้อง track events มากขึ้น:

from langchain.agents import AgentExecutor, ZeroShotAgent
from langchain.tools import Tool
from langchain.callbacks.base import BaseCallbackHandler
from typing import Union, Text, Any
import time

class AgentCallbackHandler(BaseCallbackHandler):
    """Callback handler สำหรับ Agent — track tools usage"""
    
    def __init__(self):
        super().__init__()
        self.tool_calls: List[Dict] = []
        self.agent_thoughts: List[str] = []
    
    def on_tool_start(
        self,
        serialized: Dict[str, Any],
        input_str: str,
        **kwargs: Any
    ) -> None:
        """เมื่อเริ่มใช้ tool"""
        tool_name = serialized.get("name", "unknown")
        self.tool_calls.append({
            "tool": tool_name,
            "input": input_str[:200],  # ตัดเฉพาะ 200 ตัวอักษรแรก
            "started_at": time.time()
        })
        print(f"🔧 Using tool: {tool_name}")
    
    def on_tool_end(
        self,
        output: str,
        **kwargs: Any
    ) -> None:
        """เมื่อ tool เสร็จสิ้น"""
        if self.tool_calls:
            self.tool_calls[-1]["output"] = output[:200]
            self.tool_calls[-1]["duration_ms"] = (
                time.time() - self.tool_calls[-1]["started_at"]
            ) * 1000
            print(f"✅ Tool completed in {self.tool_calls[-1]['duration_ms']:.2f}ms")
    
    def on_agent_action(
        self,
        action: Union[AgentAction, Any],
        **kwargs: Any
    ) -> None:
        """จับ agent thoughts/actions"""
        if hasattr(action, "tool"):
            thought = f"Agent thinking: use {action.tool} with input {action.tool_input}"
            self.agent_thoughts.append(thought)
            print(f"🤔 {thought}")


ตัวอย่างการใช้งานกับ agent

from langchain.agents import initialize_agent, AgentType


สร้าง tools

def search_knowledge_base(query: str) -> str:
    """ค้นหาข้อมูลจาก knowledge base"""
    # Implement การค้นหาจริงที่นี่
    return f"ผลลัพธ์การค้นหา '{query}' จาก knowledge base"

tools = [
    Tool(
        name="SearchKnowledgeBase",
        func=search_knowledge_base,
        description="ค้นหาข้อมูลจาก knowledge base"
    )
]


สร้าง agent พร้อม callback

agent_callback = AgentCallbackHandler()

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    callbacks=[agent_callback],  # ผูก callback
    verbose=True
)


รัน agent

result = agent.run("ค้นหาข้อมูลเกี่ยวกับนโยบายการคืนสินค้า")
print(f"\nAgent Result: {result}")
print(f"Tool calls: {agent_callback.tool_calls}")

การบันทึก Event ไปยัง Database

สำหรับ production environment คุณควรบันทึก event ไปยัง database เพื่อวิเคราะห์ในภายหลัง:

import sqlite3
from datetime import datetime
from langchain.callbacks.base import BaseCallbackHandler
from typing import Any, Dict, List
from contextlib import contextmanager

class DatabaseCallbackHandler(BaseCallbackHandler):
    """บันทึก events ไปยัง SQLite database"""
    
    def __init__(self, db_path: str = "langchain_events.db"):
        super().__init__()
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        """สร้างตารางถ้ายังไม่มี"""
        with self._get_connection() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS llm_events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT,
                    event_type TEXT,
                    model TEXT,
                    prompt_length INTEGER,
                    completion_length INTEGER,
                    latency_ms REAL,
                    token_usage INTEGER,
                    error_message TEXT,
                    metadata TEXT
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS tool_events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT,
                    tool_name TEXT,
                    input_preview TEXT,
                    output_preview TEXT,
                    duration_ms REAL
                )
            """)
    
    @contextmanager
    def _get_connection(self):
        """Context manager สำหรับ connection"""
        conn = sqlite3.connect(self.db_path)
        try:
            yield conn
            conn.commit()
        finally:
            conn.close()
    
    def on_llm_end(self, response: LRMResult, **kwargs: Any) -> None:
        """บันทึก LLM event"""
        with self._get_connection() as conn:
            token_usage = 0
            if response.llm_output and "token_usage" in response.llm_output:
                token_usage = response.llm_output["token_usage"].get("total_tokens", 0)
            
            conn.execute("""
                INSERT INTO llm_events 
                (timestamp, event_type, latency_ms, token_usage)
                VALUES (?, ?, ?, ?)
            """, (
                datetime.now().isoformat(),
                "llm_end",
                kwargs.get("latency_ms", 0),
                token_usage
            ))
    
    def on_tool_end(self, output: str, **kwargs: Any) -> None:
        """บันทึก tool event"""
        with self._get_connection() as conn:
            conn.execute("""
                INSERT INTO tool_events
                (timestamp, output_preview, duration_ms)
                VALUES (?, ?, ?)
            """, (
                datetime.now().isoformat(),
                output[:500] if output else "",
                kwargs.get("duration_ms", 0)
            ))
    
    def get_cost_summary(self) -> Dict[str, Any]:
        """ดึงสรุปค่าใช้จ่าย"""
        with self._get_connection() as conn:
            cursor = conn.execute("""
                SELECT 
                    COUNT(*) as total_calls,
                    SUM(token_usage) as total_tokens,
                    AVG(latency_ms) as avg_latency
                FROM llm_events
                WHERE timestamp > datetime('now', '-30 days')
            """)
            row = cursor.fetchone()
            return {
                "total_calls": row[0],
                "total_tokens": row[1] or 0,
                "avg_latency_ms": row[2] or 0,
                # คำนวณค่าใช้จ่ายตามราคา HolySheep
                "estimated_cost_usd": (row[1] or 0) / 1_000_000 * 8  # GPT-4.1 = $8/MTok
            }

ราคา HolySheep AI 2026

สำหรับการคำนวณค่าใช้จ่ายและวางแผน budget นี่คือราคาของ HolySheep AI:

ด้วยอัตราแลกเปลี่ยน ¥1 = $1 และความล่าช้าต่ำกว่า 50ms คุณสามารถ optimize cost ได้อย่างมีประสิทธิภาพ

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

1. Callback ไม่ทำงาน — Memory Leak จาก Event Queue

ปัญหา: ใน production พบว่า callback ทำงานช้าลงเรื่อยๆ และสุดท้ายค้าง

สาเหตุ: events list ใน callback handler โตขึ้นเรื่อยๆ โดยไม่มีการ clear

# ❌ โค้ดที่มีปัญหา
class BadCallbackHandler(BaseCallbackHandler):
    def __init__(self):
        super().__init__()
        self.events = []  # โตขึ้นเรื่อยๆ ไม่มีที่สิ้นสุด


✅ แก้ไข: ใช้ deque ที่มี maxlen หรือ flush เป็นระยะ

from collections import deque
from typing import Optional

class GoodCallbackHandler(BaseCallbackHandler):
    def __init__(self, max_events: int = 1000):
        super().__init__()
        self.events = deque(maxlen=max_events)  # จำกัดขนาด
        self._last_flush = datetime.now()
        self._flush_interval = 300  # flush ทุก 5 นาที
    
    def _should_flush(self) -> bool:
        """ตรวจสอบว่าควร flush หรือยัง"""
        return (datetime.now() - self._last_flush).seconds > self._flush_interval
    
    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        if self._should_flush():
            self._flush_to_database()
            self.events.clear()
            self._last_flush = datetime.now()
        self.events.append({"event": "llm_end", "timestamp": datetime.now()})

2. Token Count ไม่ถูกต้อง — Missing Metadata

ปัญหา: token usage ไม่ตรงกับที่ API ตอบกลับจริง

สาเหตุ: ไม่ได้ parse response จาก llm_output อย่างถูกต้อง

# ❌ โค้ดที่มีปัญหา
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
    # พยายามเข้าถึง token_usage โดยตรง (ผิด)
    tokens = response.token_usage  # AttributeError!


✅ แก้ไข: เข้าถึงผ่าน llm_output

def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
    tokens = 0
    if response.llm_output:
        # HolySheep API ส่ง token_usage ใน llm_output
        token_usage = response.llm_output.get("token_usage", {})
        if isinstance(token_usage, dict):
            tokens = token_usage.get("total_tokens", 0)
            prompt_tokens = token_usage.get("prompt_tokens", 0)
            completion_tokens = token_usage.get("completion_tokens", 0)
        elif isinstance(token_usage, int):
            # บางครั้ง API ส่งมาเป็น int โดยตรง
            tokens = token_usage
    
    logger.info(f"Token usage: {tokens}")

3. Async/Sync Callback ปนกัน — Runtime Error

ปัญหา: ใช้ async callback กับ sync chain หรือในทางกลับกัน แล้วเกิด runtime error

สาเหตุ: LangChain มีทั้ง sync และ async callback interfaces

# ❌ โค้ดที่มีปัญหา: async callback กับ sync chain
class AsyncCallback(BaseCallbackHandler):
    async def on_llm_start(self, *args, **kwargs):
        await some_async_operation()


ใช้กับ chain ปกติ (sync)

chain.invoke({"input": "test"}, callbacks=[AsyncCallback()])  # Error!


✅ แก้ไข: สร้าง callback ที่รองรับทั้งสองแบบ

from typing import Optional
from asyncio import iscoroutinefunction

class UniversalCallbackHandler(BaseCallbackHandler):
    async def on_llm_start(self, *args, **kwargs):
        # ตรวจสอบว่า context ปัจจุบันเป็น async หรือไม่
        try:
            import asyncio
            loop = asyncio.get_running_loop()
            # ถ้ามี loop กำลังทำงาน ใช้ async version
            await self._async_on_llm_start(*args, **kwargs)
        except RuntimeError:
            # ไม่มี loop ใช้ sync version
            self._sync_on_llm_start(*args, **kwargs)
    
    def _sync_on_llm_start(self, *args, **kwargs):
        logger.info("Sync LLM start")
    
    async def _async_on_llm_start(self, *args, **kwargs):
        logger.info("Async LLM start")
        await asyncio.sleep(0)  # yield control


หรือใช้วิธีง่ายกว่า: ใช้ AsyncCallbackHandler กับ async chain

from langchain.callbacks.base import AsyncCallbackHandler

async_chain = chain.with_config({"callbacks": [AsyncCallbackHandler()]})

Best Practices สำหรับ Production

จากประสบการณ์ในการ deploy callback system หลายโปรเจกต์ นี่คือแนวทางปฏิบัติที่แนะนำ: