ในยุคที่ AI กลายเป็นหัวใจสำคัญของระบบธุรกิจ การสร้าง LLM Application ที่เสถียรและต่อเนื่องไม่ใช่เรื่องง่าย หลายทีมพบว่าแม้จะสร้าง Chain ที่ซับซ้อนได้แล้ว แต่กลับไม่สามารถติดตามปัญหา วัดผล หรือแจ้งเตือนเมื่อเกิดความผิดพลาดได้อย่างมีประสิทธิภาพ บทความนี้จะพาคุณสร้างระบบ Monitoring และ Alerting สำหรับ LangChain Application อย่างเป็นระบบ โดยใช้ HolySheep AI เป็น LLM Provider หลัก
ทำไมต้องมี Monitoring และ Alerting สำหรับ LangChain
LangChain ทำให้การสร้าง LLM Application ง่ายขึ้นมาก แต่เมื่อระบบเริ่มใช้งานจริง คุณจะพบปัญหาหลายประการที่ต้องติดตาม ปัญหาเหล่านี้ส่งผลกระทบต่อประสบการณ์ผู้ใช้และต้นทุนโดยตรง
- Latency สูงผิดปกติ — Response Time ที่เพิ่มขึ้นอย่างไม่มีสาเหตุ อาจทำให้ผู้ใช้รอนานเกินไป
- Token Usage พุ่งสูง — Cost ที่คาดไม่ถึงจาก Chain ที่วนลูปหรือ Prompt ที่ไม่เหมาะสม
- Error Rate สูง — API Error, Rate Limit, หรือ Model Unavailable ที่กระทบต่อ Availability
- Quality Degradation — Output ที่เปลี่ยนไปโดยไม่ทราบสาเหตุ
- Chain Execution ล้มเหลว — แต่ละ Step อาจล้มเหลวโดยไม่มีใครรู้
สถาปัตยกรรม LLMOps Monitoring กับ LangChain + HolySheep
ก่อนเริ่มต้น เรามาดูสถาปัตยกรรมที่จะสร้างกัน ระบบนี้ประกอบด้วยหลาย Layer ที่ทำงานร่วมกัน ทุก Layer จะส่ง Metrics ไปยัง Monitoring System และ Trigger Alert เมื่อเกิดเงื่อนไขที่กำหนด
Component หลักในระบบ
- LangChain Application — Chain ที่สร้างด้วย LangChain
- LLM Provider — HolySheep AI ให้บริการ LLM หลายโมเดล
- Callbacks — LangChain Callback สำหรับดักจับ Events
- Metrics Store — Prometheus, InfluxDB, หรือ CloudWatch
- Alerting System — PagerDuty, Slack, หรือ Email
การตั้งค่า LangChain Callbacks สำหรับ Monitoring
LangChain มี Callback System ที่ทรงพลังมาก คุณสามารถดักจับทุก Event ที่เกิดขึ้นใน Chain ได้ เราจะสร้าง Custom Callback Handler ที่เก็บ Metrics และส่ง Alert
โครงสร้างพื้นฐานของ Callback Handler
import time
import json
from typing import Any, Dict, List, Optional
from datetime import datetime
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish, LLMResult
class LLMOpsCallbackHandler(BaseCallbackHandler):
"""
Custom Callback Handler สำหรับ LLMOps Monitoring
เก็บข้อมูล Latency, Token Usage, Error และส่ง Alert
"""
def __init__(self, alert_threshold: dict = None):
super().__init__()
self.metrics = {
"llm_calls": [],
"chain_executions": [],
"errors": [],
"total_tokens": 0,
"total_cost": 0.0
}
self.alert_threshold = alert_threshold or {
"latency_ms": 5000,
"error_rate": 0.05,
"token_limit": 100000
}
self.current_chain_start = None
self.current_step = 0
def on_llm_start(
self, serialized: Dict[str, Any], prompts: List[str], **kwargs
) -> None:
"""เริ่มต้น LLM Call — เก็บ Timestamp และ Input"""
self.llm_start_time = time.time()
self.current_prompt_tokens = self._estimate_tokens(prompts[0])
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
"""สิ้นสุด LLM Call — คำนวณ Latency และ Token Usage"""
latency_ms = (time.time() - self.llm_start_time) * 1000
# ดึง Token Usage จาก Response
if response.llm_output and "token_usage" in response.llm_output:
usage = response.llm_output["token_usage"]
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", prompt_tokens + completion_tokens)
else:
total_tokens = self.current_prompt_tokens + self._estimate_tokens(
response.generations[0][0].text if response.generations else ""
)
# คำนวณ Cost (ราคาจาก HolySheep)
cost = self._calculate_cost(total_tokens, response.llm_output)
# เก็บ Metrics
call_data = {
"timestamp": datetime.utcnow().isoformat(),
"latency_ms": round(latency_ms, 2),
"tokens": total_tokens,
"cost_usd": cost,
"model": self._get_model_name(response)
}
self.metrics["llm_calls"].append(call_data)
self.metrics["total_tokens"] += total_tokens
self.metrics["total_cost"] += cost
# ตรวจสอบ Alert Threshold
self._check_alerts(call_data)
def on_chain_start(
self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs
) -> None:
"""เริ่มต้น Chain Execution"""
self.current_chain_start = time.time()
self.current_step = 0
def on_chain_end(self, outputs: Dict[str, Any], **kwargs) -> None:
"""สิ้นสุด Chain Execution"""
if self.current_chain_start:
chain_duration = (time.time() - self.current_chain_start) * 1000
self.metrics["chain_executions"].append({
"timestamp": datetime.utcnow().isoformat(),
"duration_ms": round(chain_duration, 2),
"steps": self.current_step
})
def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs) -> None:
"""Tool Execution เริ่มต้น"""
self.current_step += 1
def on_text(self, text: str, **kwargs) -> None:
"""Text Event — ใช้สำหรับ Debugging"""
pass
def _estimate_tokens(self, text: str) -> int:
"""ประมาณ Token Count (Approximation)"""
return len(text) // 4
def _calculate_cost(self, tokens: int, llm_output: dict) -> float:
"""คำนวณ Cost ตามราคา HolySheep"""
model_name = self._get_model_name_from_output(llm_output)
# ราคาจาก HolySheep (2026/MTok)
pricing = {
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42 # $0.42/MTok
}
rate = pricing.get(model_name, 8.0)
return (tokens / 1_000_000) * rate
def _get_model_name(self, response: LLMResult) -> str:
return response.llm_output.get("model_name", "unknown") if response.llm_output else "unknown"
def _get_model_name_from_output(self, llm_output: dict) -> str:
if llm_output and "model_name" in llm_output:
return llm_output["model_name"]
return "gpt-4.1"
def _check_alerts(self, call_data: dict) -> None:
"""ตรวจสอบ Alert Conditions"""
alerts = []
if call_data["latency_ms"] > self.alert_threshold["latency_ms"]:
alerts.append(f"🔴 HIGH_LATENCY: {call_data['latency_ms']}ms")
if self.metrics["total_tokens"] > self.alert_threshold["token_limit"]:
alerts.append(f"🟡 HIGH_TOKEN_USAGE: {self.metrics['total_tokens']} tokens")
# ส่ง Alert (Implementation ขึ้นกับ Alerting System)
for alert in alerts:
self._send_alert(alert, call_data)
def _send_alert(self, message: str, context: dict) -> None:
"""ส่ง Alert ไปยัง Alerting System"""
# TODO: Implement Slack/Email/PagerDuty Integration
print(f"[ALERT] {message}")
print(f"Context: {json.dumps(context, indent=2)}")
def get_metrics_summary(self) -> dict:
"""สรุป Metrics ทั้งหมด"""
return {
"summary": {
"total_llm_calls": len(self.metrics["llm_calls"]),
"total_tokens": self.metrics["total_tokens"],
"total_cost_usd": round(self.metrics["total_cost"], 4),
"avg_latency_ms": self._calculate_avg_latency(),
"error_count": len(self.metrics["errors"])
},
"metrics": self.metrics
}
def _calculate_avg_latency(self) -> float:
if not self.metrics["llm_calls"]:
return 0.0
return round(
sum(call["latency_ms"] for call in self.metrics["llm_calls"])
/ len(self.metrics["llm_calls"]), 2
)การใช้งาน LangChain กับ HolySheep AI
มาถึงส่วนสำคัญ — การเชื่อมต่อ LangChain กับ HolySheep AI ซึ่งมีจุดเด่นหลายประการสำหรับ LLMOps
ความได้เปรียบของ HolySheep AI สำหรับ Production
- ความหน่วงต่ำ — Latency ต่ำกว่า 50ms ทำให้ Response เร็วมาก
- ราคาประหยัด — อัตราแลกเปลี่ยน ¥1=$1 ประหยัดได้มากกว่า 85%
- รองรับหลายโมเดล — GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2
- ชำระเงินง่าย — รองรับ WeChat และ Alipay
- เครดิตฟรี — รับเครดิตฟรีเมื่อลงทะเบียน
ราคาโมเดล (2026/MTok)
- GPT-4.1 — $8.00/MTok
- Claude Sonnet 4.5 — $15.00/MTok
- Gemini 2.5 Flash — $2.50/MTok (เหมาะสำหรับ High Volume)
- DeepSeek V3.2 — $0.42/MTok (ประหยัดที่สุด)
การสร้าง Chatbot พร้อม Monitoring แบบครบวงจร
ต่อไปจะเป็นตัวอย่างการสร้าง RAG Application ที่มีระบบ Monitoring และ Alerting ในตัว ใช้งานได้จริงใน Production
import os
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
========== ส่วน Configuration สำหรับ HolySheep ==========
class HolySheepConfig:
"""Configuration สำหรับ HolySheep AI API"""
# ⚠️ สำคัญ: base_url ต้องเป็น https://api.holysheep.ai/v1
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# เลือก Model ตาม Use Case
MODELS = {
"gpt-4.1": {
"name": "gpt-4.1",
"display_name": "GPT-4.1",
"cost_per_mtok": 8.0,
"best_for": "Complex reasoning, coding"
},
"claude-sonnet-4.5": {
"name": "claude-sonnet-4.5",
"display_name": "Claude Sonnet 4.5",
"cost_per_mtok": 15.0,
"best_for": "Long context, analysis"
},
"gemini-2.5-flash": {
"name": "gemini-2.5-flash",
"display_name": "Gemini 2.5 Flash",
"cost_per_mtok": 2.50,
"best_for": "Fast responses, high volume"
},
"deepseek-v3.2": {
"name": "deepseek-v3.2",
"display_name": "DeepSeek V3.2",
"cost_per_mtok": 0.42,
"best_for": "Cost-effective, general tasks"
}
}
@classmethod
def create_llm(cls, model_name: str = "gpt-4.1", **kwargs):
"""สร้าง LLM Instance ที่เชื่อมต่อกับ HolySheep"""
return ChatOpenAI(
temperature=kwargs.get("temperature", 0.7),
max_tokens=kwargs.get("max_tokens", 2000),
model_name=cls.MODELS[model_name]["name"],
openai_api_key=cls.API_KEY,
openai_api_base=cls.BASE_URL # ✅ ต้องตั้งค่านี้
)
@classmethod
def create_embeddings(cls):
"""สร้าง Embeddings Model สำหรับ RAG"""
return OpenAIEmbeddings(
openai_api_key=cls.API_KEY,
openai_api_base=cls.BASE_URL
)
========== ส่วน RAG Application พร้อม Monitoring ==========
class MonitoredRAGApplication:
"""
RAG Application ที่มีระบบ Monitoring ในตัว
ใช้ LangChain Callbacks สำหรับเก็บ Metrics
"""
def __init__(
self,
model_name: str = "gemini-2.5-flash",
alert_slack_webhook: str = None
):
self.config = HolySheepConfig
self.callback_handler = LLMOpsCallbackHandler(
alert_threshold={
"latency_ms": 3000,
"error_rate": 0.05,
"token_limit": 50000
}
)
self.slack_webhook = alert_slack_webhook
# สร้าง LLM กับ Callback
self.llm = self.config.create_llm(
model_name=model_name,
temperature=0.3
)
self.embeddings = self.config.create_embeddings()
self.vectorstore = None
self.qa_chain = None
def load_documents(self, file_paths: list):
"""โหลดเอกสารและสร้าง Vector Store"""
from langchain.document_loaders import DirectoryLoader
documents = []
for path in file_paths:
loader = DirectoryLoader(path, glob="**/*.txt")
docs = loader.load()
documents.extend(docs)
# Split Documents
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
splits = text_splitter.split_documents(documents)
# สร้าง Vector Store
self.vectorstore = Chroma.from_documents(
documents=splits,
embedding=self.embeddings
)
# สร้าง QA Chain
self.qa_chain = RetrievalQA.from_chain_type(
llm=self.llm,
chain_type="stuff",
retriever=self.vectorstore.as_retriever(),
callbacks=[self.callback_handler] # ✅ เพิ่ม Callback
)
return len(splits)
def query(self, question: str) -> dict:
"""Query พร้อมเก็บ Metrics"""
import requests
start_time = time.time()
result = None
error = None
try:
# Execute Query ผ่าน LangChain
result = self.qa_chain({"query": question})
except Exception as e:
error = str(e)
self.callback_handler.metrics["errors"].append({
"timestamp": datetime.utcnow().isoformat(),
"error": error,
"query": question
})
# ส่ง Error Alert
self._send_error_alert(error, question)
finally:
# เก็บ Query Metrics
query_duration = (time.time() - start_time) * 1000
metrics = self.callback_handler.get_metrics_summary()
return {
"answer": result.get("result") if result else None,
"error": error,
"metrics": metrics,
"query_duration_ms": round(query_duration, 2)
}
def _send_error_alert(self, error: str, context: str):
"""ส่ง Alert เมื่อเกิด Error"""
if self.slack_webhook:
import requests
payload = {
"text": f"🚨 LangChain Error Detected",
"attachments": [{
"color": "danger",
"fields": [
{"title": "Error", "value": error[:100]},
{"title": "Context", "value": context[:100]}
]
}]
}
requests.post(self.slack_webhook, json=payload)
========== วิธีใช้งาน ==========
if __name__ == "__main__":
# ตั้งค่า API Key
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
# สร้าง Application
app = MonitoredRAGApplication(
model_name="gemini-2.5-flash", # เลือกโมเดลที่เหมาะสม
alert_slack_webhook=os.environ.get("SLACK_WEBHOOK")
)
# โหลดเอกสาร
num_chunks = app.load_documents(["./data/docs"])
print(f"Loaded {num_chunks} document chunks")
# Query พร้อม Monitoring
result = app.query("What is the main topic of these documents?")
print(f"Answer: {result['answer']}")
print(f"Query Duration: {result['query_duration_ms']}ms")
print(f"Total Cost: ${result['metrics']['summary']['total_cost_usd']}")
print(f"Avg Latency: {result['metrics']['summary']['avg_latency_ms']}ms")การสร้าง Dashboard และ Alerting Rules
เมื่อมี Callback เก็บ Metrics แล้ว ต่อไปจะเป็นการสร้างระบบ Alerting ที่ครอบคลุม รองรับ Prometheus และ Grafana สำหรับ Visualization
import prometheus_client as prom
from datetime import datetime, timedelta
import schedule
import time
========== Prometheus Metrics Definitions ==========
LLM Metrics
LLM_REQUEST_COUNT = prom.Counter(
'llm_requests_total',
'Total LLM requests',
['model', 'status']
)
LLM_REQUEST_LATENCY = prom.Histogram(
'llm_request_latency_seconds',
'LLM request latency',
['model'],
buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)
LLM_TOKEN_USAGE = prom.Counter(
'llm_tokens_total',
'Total tokens used',
['model', 'type'] # type: prompt/completion
)
LLM_COST_USD = prom.Counter(
'llm_cost_usd_total',
'Total cost in USD',
['model']
)
Chain Metrics
CHAIN_EXECUTION_COUNT = prom.Counter(
'chain_executions_total',
'Total chain executions',
['chain_name', 'status']
)
CHAIN_EXECUTION_LATENCY = prom.Histogram(
'chain_execution_latency_seconds',
'Chain execution latency',
['chain_name'],
buckets=[1.0, 5.0, 10.0, 30.0, 60.0, 120.0]
)
Error Metrics
ERROR_COUNT = prom.Counter(
'llm_errors_total',
'Total errors',
['error_type', 'model']
)
========== Alert Rules Configuration ==========
ALERT_RULES = {
"high_latency": {
"condition": lambda m: m["avg_latency_ms"] > 5000,
"severity": "warning",
"message": "Average LLM latency exceeds 5 seconds",
"cooldown_minutes": 15
},
"error_rate_high": {
"condition": lambda m: m["error_rate"] > 0.05,
"severity": "critical",
"message": "Error rate exceeds 5%",
"cooldown_minutes": 5
},
"token_budget_exceeded": {
"condition": lambda m: m["total_tokens"] > 100000,
"severity": "warning",
"message": "Token usage exceeds 100K in current period",
"cooldown_minutes": 60
},
"cost_exceeded": {
"condition": lambda m: m["total_cost_usd"] > 100.0,
"severity": "critical",
"message": "Cost exceeds $100 in current period",
"cooldown_minutes": 30
},
"model_unavailable": {
"condition": lambda m: m["availability"] < 0.99,
"severity": "critical",
"message": "Model availability below 99%",
"cooldown_minutes": 5
}
}
========== Alert Manager Class ==========
class AlertManager:
"""จัดการ Alert และ Notification"""
def __init__(self, slack_webhook: str = None, email_config: dict = None):
self.slack_webhook = slack_webhook
self.email_config = email_config
self.alert_history = []
self.cooldown_tracker = {}
def check_and_send_alerts(self, metrics: dict):
"""ตรวจสอบ Metrics และส่ง Alert ตามเงื่อนไข"""
current_time = datetime.utcnow()
for rule_name, rule in ALERT_RULES.items():
# ตรวจสอบ Cooldown
if self._is_in_cooldown(rule_name):
continue
# ตรวจสอบเงื่อนไข
try:
if rule["condition"](metrics):
self._send_alert(rule_name, rule, metrics)
self._set_cooldown(rule_name, rule["cooldown_minutes"])
except Exception as e:
print(f"Error checking rule {rule_name}: {e}")
def _is_in_cooldown(self, rule_name: str) -> bool:
"""ตรวจสอบว่า Alert อยู่ใน Cooldown หรือไม่"""
if rule_name not in self.cooldown_tracker:
return False
return datetime.utcnow() < self.cooldown_tracker[rule_name]
def _set_cooldown(self, rule_name: str, minutes: int):
"""ตั้ง Cooldown สำหรับ Alert"""
self.cooldown_tracker[rule_name] = datetime.utcnow() + timedelta(minutes=minutes)
def _send_alert(self, rule_name: str, rule: dict, metrics: dict):
"""ส่ง Alert ไปยังทุก Channel"""
alert = {
"timestamp": datetime.utcnow().isoformat(),
"rule": rule_name,
"severity": rule["severity"],
"message": rule["message"],
"metrics": metrics
}
self.alert_history.append(alert)
# ส่งไป Slack
if self.slack_webhook:
self._send_slack_alert(alert)
# ส่งไป Email
if self.email_config:
self._send_email_alert(alert)
def _send_slack_alert(self, alert: dict):
"""ส่ง Alert ไป Slack"""
import requests
severity_emoji = {
"critical": "🔴",
"warning": "🟡",
"info": "ℹ️"
}
payload = {
"text": f"{severity_emoji.get(alert['severity'], '⚠️')} {alert['message']}",
"attachments": [{
"color": "danger" if alert["severity"] == "critical" else "warning",
"fields": [
{"title": "Rule", "value": alert["rule"], "short": True},
{"title": "Time", "value": alert["timestamp"], "short": True},
{"title": "Metrics", "value": str(alert["metrics"])[:500]}
]
}]
}
try:
response = requests.post(self.slack_webhook, json=payload)
response.raise_for_status()
except Exception as e:
print(f"Failed to send Slack alert: {e}")
def _send_email_alert(self, alert: dict):
"""ส่ง Alert ไป Email"""
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
msg = MIMEMultipart()
msg['Subject'] = f"[{alert['severity'].upper()}] LangChain Alert: {alert['rule']}"
msg['From'] = self.email_config['from']
msg['To'] = self.email_config['to']
body = f"""
LangChain Alert
Severity: {alert['severity']}
Rule: {alert['rule']}
Message: {alert['message']}
Time: {alert['timestamp']}
Metrics:
{alert['metrics']}
"""
msg.attach(MIMEText(body, 'plain'))
try:
with smtplib.SMTP(self.email_config['smtp_host'], self.email_config['smtp_port']) as server:
server.starttls()
server.login(self.email_config['username'], self.email_config['password'])
server.send_message(msg)
except Exception as e:
print(f"Failed to send email alert: {e}")
========== Prometheus Export Endpoint ==========
def start_metrics_server(port: int = 9090):
"""เริ่ม Prometheus Metrics Server"""
prom.start_http_server(port)
print(f"Prometheus metrics available at http://localhost:{port}/metrics")
========== Scheduled Alert Check ==========
def run_periodic_alert_check(alert_manager: AlertManager, callback_handler: LLMOpsCallbackHandler):
"""รัน Alert Check เป็นระยะ"""
metrics = callback_handler.get_metrics_summary()
alert_manager.check_and_send_alerts(metrics)
# Reset counters หลังจาก Check