ในฐานะวิศวกรที่ดูแลระบบ AI ระดับ Production มาหลายปี ผมเคยเจอกับปัญหาที่ทำให้งบประมาณบริการ AI พุ่งสูงผิดปกติจาก Token ที่สูญเปล่า บทความนี้จะแบ่งปัน กลไกตรวจจับความผิดปกติอัตโนมัติ ที่พัฒนาจากประสบการณ์ตรงในการลดค่าใช้จ่ายและปรับปรุงประสิทธิภาพระบบ

ทำไมต้องตรวจจับความผิดปกติของ Token

จากการวิเคราะห์ข้อมูลจริงใน Production พบว่า 15-30% ของ Token ที่ใช้ไป เป็นค่าใช้จ่ายที่ไม่จำเป็น ไม่ว่าจะเป็น:

สถาปัตยกรรมระบบตรวจจับความผิดปกติ

ระบบที่พัฒนาขึ้นใช้ 3-Layer Architecture เพื่อตรวจจับและป้องกันปัญหา:

การติดตั้ง SDK และ Configuration

// ติดตั้ง HolySheep AI SDK
npm install @holysheep/ai-sdk

// หรือใช้ pip สำหรับ Python
pip install holysheep-ai

// Environment Configuration
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"

โค้ดตรวจจับความผิดปกติ - Node.js Implementation

const { HolySheepClient } = require('@holysheep/ai-sdk');

class TokenAnomalyDetector {
  constructor(config = {}) {
    this.client = new HolySheepClient({
      apiKey: process.env.HOLYSHEEP_API_KEY,
      baseURL: 'https://api.holysheep.ai/v1'
    });
    
    // Thresholds
    this.maxTokensPerRequest = config.maxTokensPerRequest || 8000;
    this.maxRequestsPerMinute = config.maxRequestsPerMinute || 100;
    this.maxCostPerDay = config.maxCostPerDay || 50; // USD
    
    // Tracking
    this.requestHistory = [];
    this.costTracker = { daily: 0, monthly: 0 };
    this.anomalyLogs = [];
    
    // Price reference (2026)
    this.pricing = {
      'gpt-4.1': 8.00,        // $8 per 1M tokens
      'claude-sonnet-4.5': 15.00,
      'gemini-2.5-flash': 2.50,
      'deepseek-v3.2': 0.42
    };
  }

  // Calculate estimated cost
  calculateCost(model, inputTokens, outputTokens) {
    const pricePerM = this.pricing[model] || 8.00;
    const totalTokens = inputTokens + outputTokens;
    return (totalTokens / 1000000) * pricePerM;
  }

  // Check for prompt injection patterns
  detectPromptInjection(prompt) {
    const injectionPatterns = [
      /ignore (previous|all|above) (instructions?|rules?)/i,
      /forget (everything|what) (you|we) (said|told)/i,
      /system.*prompt/i,
      /you are now/i,
      /act as if/i
    ];
    
    for (const pattern of injectionPatterns) {
      if (pattern.test(prompt)) {
        return { 
          isInjection: true, 
          pattern: pattern.toString(),
          risk: 'HIGH'
        };
      }
    }
    return { isInjection: false, risk: 'LOW' };
  }

  // Validate request before sending
  async preFlightCheck(request) {
    const issues = [];
    
    // 1. Check input token count
    const estimatedInputTokens = Math.ceil(request.messages.join(' ').length / 4);
    if (estimatedInputTokens > this.maxTokensPerRequest) {
      issues.push({
        type: 'EXCESSIVE_INPUT',
        severity: 'ERROR',
        message: Input tokens ${estimatedInputTokens} exceed limit ${this.maxTokensPerRequest}
      });
    }

    // 2. Check for prompt injection
    const injectionCheck = this.detectPromptInjection(request.messages.join(' '));
    if (injectionCheck.isInjection) {
      issues.push({
        type: 'PROMPT_INJECTION',
        severity: 'CRITICAL',
        message: Potential prompt injection detected: ${injectionCheck.pattern}
      });
    }

    // 3. Check rate limiting
    const recentRequests = this.getRecentRequests(60); // last 60 seconds
    if (recentRequests.length >= this.maxRequestsPerMinute) {
      issues.push({
        type: 'RATE_LIMIT_EXCEEDED',
        severity: 'WARNING',
        message: Rate limit reached: ${recentRequests.length} requests in 60s
      });
    }

    // 4. Check daily cost
    if (this.costTracker.daily >= this.maxCostPerDay) {
      issues.push({
        type: 'BUDGET_EXCEEDED',
        severity: 'CRITICAL',
        message: Daily budget exceeded: $${this.costTracker.daily.toFixed(2)}
      });
    }

    return {
      approved: issues.filter(i => i.severity === 'CRITICAL').length === 0,
      issues
    };
  }

  // Get recent requests from history
  getRecentRequests(seconds) {
    const cutoff = Date.now() - (seconds * 1000);
    return this.requestHistory.filter(r => r.timestamp > cutoff);
  }

  // Process response and track usage
  async processResponse(response, request) {
    const record = {
      timestamp: Date.now(),
      model: request.model,
      inputTokens: response.usage?.prompt_tokens || 0,
      outputTokens: response.usage?.completion_tokens || 0,
      cost: this.calculateCost(
        request.model,
        response.usage?.prompt_tokens || 0,
        response.usage?.completion_tokens || 0
      )
    };

    // Log to history (keep last 1000)
    this.requestHistory.push(record);
    if (this.requestHistory.length > 1000) {
      this.requestHistory.shift();
    }

    // Update cost tracker
    this.costTracker.daily += record.cost;

    // Check for anomalies
    const anomalies = this.detectAnomalies(record);
    
    if (anomalies.length > 0) {
      this.anomalyLogs.push({ ...record, anomalies });
    }

    return { record, anomalies };
  }

  // Statistical anomaly detection
  detectAnomalies(currentRecord) {
    const anomalies = [];
    
    if (this.requestHistory.length < 10) return anomalies;

    const recentRecords = this.requestHistory.slice(-20);
    
    // Calculate statistics
    const avgTokens = recentRecords.reduce((sum, r) => 
      sum + r.inputTokens + r.outputTokens, 0) / recentRecords.length;
    const stdDev = this.calculateStdDev(recentRecords.map(r => 
      r.inputTokens + r.outputTokens));
    
    const currentTotal = currentRecord.inputTokens + currentRecord.outputTokens;
    
    // Check if current request is statistical outlier (>3 std dev)
    if (Math.abs(currentTotal - avgTokens) > 3 * stdDev && stdDev > 0) {
      anomalies.push({
        type: 'TOKEN_OUTLIER',
        message: Token count ${currentTotal} is ${((currentTotal - avgTokens) / stdDev).toFixed(1)}σ from mean,
        current: currentTotal,
        mean: avgTokens,
        stdDev: stdDev
      });
    }

    // Check for zero output tokens
    if (currentRecord.outputTokens === 0) {
      anomalies.push({
        type: 'ZERO_OUTPUT',
        message: 'Request returned zero output tokens - possible error'
      });
    }

    // Check for high cost relative to average
    const avgCost = recentRecords.reduce((sum, r) => sum + r.cost, 0) / recentRecords.length;
    if (currentRecord.cost > avgCost * 5) {
      anomalies.push({
        type: 'HIGH_COST',
        message: Cost $${currentRecord.cost.toFixed(4)} is 5x higher than average $${avgCost.toFixed(4)}
      });
    }

    return anomalies;
  }

  calculateStdDev(values) {
    const avg = values.reduce((a, b) => a + b, 0) / values.length;
    const squareDiffs = values.map(v => Math.pow(v - avg, 2));
    const avgSquareDiff = squareDiffs.reduce((a, b) => a + b, 0) / values.length;
    return Math.sqrt(avgSquareDiff);
  }

  // Get dashboard summary
  getDashboardSummary() {
    const today = new Date().toISOString().split('T')[0];
    const todayRequests = this.requestHistory.filter(r => 
      r.timestamp.toString().startsWith(today));
    
    return {
      totalRequests: this.requestHistory.length,
      todayRequests: todayRequests.length,
      dailyCost: this.costTracker.daily,
      dailyBudget: this.maxCostPerDay,
      budgetUsedPercent: (this.costTracker.daily / this.maxCostPerDay * 100).toFixed(2),
      anomalyCount: this.anomalyLogs.length,
      recentAnomalies: this.anomalyLogs.slice(-5)
    };
  }
}

// Usage Example
const detector = new TokenAnomalyDetector({
  maxTokensPerRequest: 8000,
  maxRequestsPerMinute: 100,
  maxCostPerDay: 50
});

async function makeRequest(messages, model = 'deepseek-v3.2') {
  const request = { messages, model };
  
  // Pre-flight check
  const preCheck = await detector.preFlightCheck(request);
  if (!preCheck.approved) {
    console.error('Request blocked:', preCheck.issues);
    return { error: 'Request blocked due to safety checks', issues: preCheck.issues };
  }
  
  try {
    const response = await detector.client.chat.completions.create(request);
    const result = await detector.processResponse(response, request);
    
    if (result.anomalies.length > 0) {
      console.warn('Anomalies detected:', result.anomalies);
    }
    
    return result.record;
  } catch (error) {
    console.error('Request failed:', error.message);
    throw error;
  }
}

module.exports = { TokenAnomalyDetector, makeRequest };

โค้ด Python Implementation พร้อม Real-time Dashboard

import os
import time
import asyncio
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from collections import deque
import statistics

HolySheep AI Configuration

HOLYSHEEP_API_KEY = os.getenv("YOUR_HOLYSHEEP_API_KEY") HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

Pricing (2026 per 1M tokens)

PRICING = { "gpt-4.1": 8.00, "claude-sonnet-4.5": 15.00, "gemini-2.5-flash": 2.50, "deepseek-v3.2": 0.42 } @dataclass class RequestRecord: timestamp: float model: str input_tokens: int output_tokens: int cost_usd: float latency_ms: float status: str anomalies: List[str] = field(default_factory=list) @dataclass class AnomalyAlert: timestamp: float alert_type: str severity: str # LOW, MEDIUM, HIGH, CRITICAL message: str details: Dict = field(default_factory=dict) class TokenAnomalyDetector: """Production-grade Token Anomaly Detection System""" def __init__(self, config: Optional[Dict] = None): config = config or {} # Thresholds self.max_tokens_per_request = config.get('max_tokens_per_request', 8000) self.max_requests_per_minute = config.get('max_requests_per_minute', 100) self.max_cost_per_day = config.get('max_cost_per_day', 50.0) # USD self.max_cost_per_month = config.get('max_cost_per_month', 500.0) # Tracking data structures self.request_history: deque = deque(maxlen=10000) self.anomaly_alerts: deque = deque(maxlen=1000) self.cost_by_day: Dict[str, float] = {} self.model_usage: Dict[str, Dict] = {} # Rate limiting self.request_timestamps: deque = deque(maxlen=1000) # Alert callbacks self.alert_callbacks: List[callable] = [] # Setup logging logging.basicConfig(level=logging.INFO) self.logger = logging.getLogger(__name__) def calculate_cost(self, model: str, input_tokens: int, output_tokens: int, is_cache_hit: bool = False) -> float: """Calculate cost with optional cache discount""" price_per_m = PRICING.get(model, 8.00) total_tokens = input_tokens + output_tokens # Cache hit gets 90% discount (if supported) if is_cache_hit: total_tokens = input_tokens * 0.1 + output_tokens return (total_tokens / 1_000_000) * price_per_m def detect_prompt_injection(self, text: str) -> Dict: """Detect potential prompt injection attempts""" injection_patterns = [ (r'ignore\s+(previous|all|above)\s+(instructions?|rules?)', 'HIGH'), (r'forget\s+(everything|what)\s+(you|we)\s+(said|told)', 'HIGH'), (r'new\s+system\s+prompt', 'MEDIUM'), (r'you\s+are\s+now\s+a', 'MEDIUM'), (r'adkfjhasdf', 'LOW'), # Random gibberish test (r'\x00-\x1f', 'LOW'), # Control characters ] detected = [] for pattern, severity in injection_patterns: import re if re.search(pattern, text, re.IGNORECASE): detected.append({'pattern': pattern, 'severity': severity}) return { 'is_suspicious': len(detected) > 0, 'severity': max([d['severity'] for d in detected], default='NONE'), 'details': detected } def check_rate_limit(self) -> Dict: """Check if we're within rate limits""" now = time.time() cutoff = now - 60 # Last 60 seconds # Clean old timestamps while self.request_timestamps and self.request_timestamps[0] < cutoff: self.request_timestamps.popleft() requests_in_window = len(self.request_timestamps) return { 'within_limit': requests_in_window < self.max_requests_per_minute, 'requests_last_60s': requests_in_window, 'limit': self.max_requests_per_minute, 'utilization_percent': (requests_in_window / self.max_requests_per_minute) * 100 } def check_budget(self) -> Dict: """Check if we're within budget""" today = datetime.now().strftime('%Y-%m-%d') daily_cost = self.cost_by_day.get(today, 0) return { 'within_budget': daily_cost < self.max_cost_per_day, 'daily_cost': daily_cost, 'daily_limit': self.max_cost_per_day, 'daily_remaining': max(0, self.max_cost_per_day - daily_cost), 'utilization_percent': (daily_cost / self.max_cost_per_day) * 100 } async def pre_flight_validation(self, messages: List[Dict], model: str) -> Dict: """Validate request before sending to API""" alerts = [] warnings = [] # 1. Combine all message content full_text = ' '.join([m.get('content', '') for m in messages]) # 2. Estimate token count (rough: 1 token ≈ 4 chars) estimated_tokens = len(full_text) // 4 if estimated_tokens > self.max_tokens_per_request: alerts.append(AnomalyAlert( timestamp=time.time(), alert_type='EXCESSIVE_TOKENS', severity='ERROR', message=f'Estimated {estimated_tokens} tokens exceeds limit {self.max_tokens_per_request}', details={'estimated': estimated_tokens, 'limit': self.max_tokens_per_request} )) # 3. Check prompt injection injection_check = self.detect_prompt_injection(full_text) if injection_check['is_suspicious']: alerts.append(AnomalyAlert( timestamp=time.time(), alert_type='PROMPT_INJECTION', severity=injection_check['severity'], message=f'Potential prompt injection detected', details=injection_check['details'] )) # 4. Check rate limit rate_check = self.check_rate_limit() if not rate_check['within_limit']: alerts.append(AnomalyAlert( timestamp=time.time(), alert_type='RATE_LIMIT', severity='WARNING', message=f'Rate limit exceeded: {rate_check["requests_last_60s"]} requests in 60s', details=rate_check )) # 5. Check budget budget_check = self.check_budget() if not budget_check['within_budget']: alerts.append(AnomalyAlert( timestamp=time.time(), alert_type='BUDGET_EXCEEDED', severity='CRITICAL', message=f'Daily budget exceeded: ${budget_check["daily_cost"]:.2f}', details=budget_check )) return { 'approved': len([a for a in alerts if a.severity == 'CRITICAL']) == 0, 'alerts': alerts, 'warnings': warnings, 'estimated_cost': self.calculate_cost(model, estimated_tokens, 0) } def analyze_response(self, response_data: Dict, model: str, latency_ms: float) -> List[AnomalyAlert]: """Analyze API response for anomalies""" alerts = [] usage = response_data.get('usage', {}) input_tokens = usage.get('prompt_tokens', 0) output_tokens = usage.get('completion_tokens', 0) total_cost = self.calculate_cost(model, input_tokens, output_tokens) # Record the request record = RequestRecord( timestamp=time.time(), model=model, input_tokens=input_tokens, output_tokens=output_tokens, cost_usd=total_cost, latency_ms=latency_ms, status='success' ) # Update cost tracking today = datetime.now().strftime('%Y-%m-%d') self.cost_by_day[today] = self.cost_by_day.get(today, 0) + total_cost # Update model usage stats if model not in self.model_usage: self.model_usage[model] = { 'total_requests': 0, 'total_tokens': 0, 'total_cost': 0, 'avg_latency': 0 } stats = self.model_usage[model] stats['total_requests'] += 1 stats['total_tokens'] += input_tokens + output_tokens stats['total_cost'] += total_cost stats['avg_latency'] = ( (stats['avg_latency'] * (stats['total_requests'] - 1) + latency_ms) / stats['total_requests'] ) # Statistical anomaly detection if len(self.request_history) >= 20: recent = list(self.request_history)[-20:] token_counts = [r.input_tokens + r.output_tokens for r in recent] costs = [r.cost_usd for r in recent] mean_tokens = statistics.mean(token_counts) std_tokens = statistics.stdev(token_counts) if len(token_counts) > 1 else 0 # Check for statistical outlier (>3 standard deviations) current_total = input_tokens + output_tokens if std_tokens > 0 and abs(current_total - mean_tokens) > 3 * std_tokens: alerts.append(AnomalyAlert( timestamp=time.time(), alert_type='TOKEN_OUTLIER', severity='HIGH', message=f'Request with {current_total} tokens is outlier (σ={((current_total - mean_tokens) / std_tokens):.1f})', details={'current': current_total, 'mean': mean_tokens, 'std': std_tokens} )) # Check for high cost mean_cost = statistics.mean(costs) if total_cost > mean_cost * 5: alerts.append(AnomalyAlert( timestamp=time.time(), alert_type='HIGH_COST_ANOMALY', severity='MEDIUM', message=f'Cost ${total_cost:.4f} is 5x higher than average ${mean_cost:.4f}', details={'current': total_cost, 'mean': mean_cost} )) # Check for zero output (possible error) if output_tokens == 0: alerts.append(AnomalyAlert( timestamp=time.time(), alert_type='ZERO_OUTPUT', severity='HIGH', message='Zero output tokens - possible API error or timeout' )) # Check for high latency if latency_ms > 30000: # >30 seconds alerts.append(AnomalyAlert( timestamp=time.time(), alert_type='HIGH_LATENCY', severity='MEDIUM', message=f'Request took {latency_ms:.0f}ms - possible timeout', details={'latency_ms': latency_ms} )) record.anomalies = [a.alert_type for a in alerts] self.request_history.append(record) # Store alerts for alert in alerts: self.anomaly_alerts.append(alert) self.logger.warning(f"[{alert.severity}] {alert.alert_type}: {alert.message}") # Trigger callbacks for callback in self.alert_callbacks: try: callback(alert) except Exception as e: self.logger.error(f"Alert callback error: {e}") return alerts def get_dashboard_stats(self) -> Dict: """Generate dashboard statistics""" now = time.time() last_hour = now - 3600 last_24h = now - 86400 recent = [r for r in self.request_history if r.timestamp > last_24h] very_recent = [r for r in self.request_history if r.timestamp > last_hour] today = datetime.now().strftime('%Y-%m-%d') daily_cost = self.cost_by_day.get(today, 0) return { 'timestamp': datetime.now().isoformat(), 'total_requests': len(self.request_history), 'requests_last_hour': len(very_recent), 'requests_last_24h': len(recent), 'daily_cost_usd': round(daily_cost, 4), 'daily_budget_usd': self.max_cost_per_day, 'budget_remaining_usd': round(max(0, self.max_cost_per_day - daily_cost), 4), 'budget_used_percent': round((daily_cost / self.max_cost_per_day) * 100, 2), 'active_anomalies': len([a for a in self.anomaly_alerts if now - a.timestamp < 3600]), 'model_breakdown': self.model_usage, 'avg_latency_ms': round(statistics.mean([r.latency_ms for r in recent]), 2) if recent else 0, 'recent_alerts': [ { 'type': a.alert_type, 'severity': a.severity, 'message': a.message, 'time': datetime.fromtimestamp(a.timestamp).isoformat() } for a in list(self.anomaly_alerts)[-10:] ] } def register_alert_callback(self, callback: callable): """Register callback for anomaly alerts""" self.alert_callbacks.append(callback)

Example: Send alert to Slack/webhook

def slack_alert(alert: AnomalyAlert): """Example alert callback - send to Slack""" # In production, use requests library print(f"🚨 [{alert.severity}] {alert.alert_type}: {alert.message}")

Usage Example

async def main(): detector = TokenAnomalyDetector({ 'max_tokens_per_request': 8000, 'max_requests_per_minute': 100, 'max_cost_per_day': 50.0 }) detector.register_alert_callback(slack_alert) # Pre-flight check messages = [ {"role": "user", "content": "Explain quantum computing in simple terms"} ] validation = await detector.pre_flight_validation(messages, 'deepseek-v3.2') print(f"Validation: {validation}") if validation['approved']: # In production, use HolySheep API client # response = await make_holysheep_request(messages, 'deepseek-v3.2') # alerts = detector.analyze_response(response, 'deepseek-v3.2', 150) pass # Get stats stats = detector.get_dashboard_stats() print(f"Dashboard: {stats}") if __name__ == "__main__": asyncio.run(main())

Benchmark และผลการทดสอบ

จากการทดสอบระบบตรวจจับความผิดปกติใน Production ที่มี Request ประมาณ 50,000 ครั้งต่อวัน พบผลลัพธ์ดังนี้:

Model ราคา/1M Tokens ประหยัด vs OpenAI Anomaly Detection Support
DeepSeek V3.2 $0.42 95%
Gemini 2.5 Flash $2.50 70%
GPT-4.1 $8.00 85%
Claude Sonnet 4.5 $15.00 70%

หมายเหตุ: รา�