HolySheep 智慧供应链异常预警 Agent：ระบบวิเคราะห์คำสั่งซื้อด้วย DeepSeek และ Gemini

ในยุคที่ห่วงโซ่อุปทานทั่วโลกต้องเผชิญกับความผันผวนสูง การมีระบบ ตรวจจับความผิดปกติแบบเรียลไทม์ จึงกลายเป็นความจำเป็นเชิงกลยุทธ์ บทความนี้จะพาคุณสร้าง Supply Chain Anomaly Detection Agent ที่ใช้ DeepSeek V3.2 สำหรับวิเคราะห์คำสั่งซื้อ และ Gemini 2.5 Flash สำหรับสร้างรายงานอัตโนมัติ พร้อมระบบ multi-model fallback ที่ทำงานได้อย่างเสถียร 24/7

ตารางเปรียบเทียบ: HolySheep vs API อย่างเป็นทางการ vs บริการรีเลย์อื่นๆ

ระบบ Supply Chain Anomaly Detection คืออะไร

ระบบตรวจจับความผิดปกติในห่วงโซ่อุปทาน ทำหน้าที่วิเคราะห์ข้อมูลคำสั่งซื้อแบบเรียลไทม์ ตรวจจับรูปแบบที่ผิดปกติ เช่น การสั่งซื้อที่ผิดปกติ การเปลี่ยนแปลงราคาที่ผิดสังเกต หรือดีเลย์ที่ไม่คาดคิด แล้วส่งการแจ้งเตือนพร้อมรายงานโดยอัตโนมัติ

ในระบบนี้เราจะใช้:

• DeepSeek V3.2 — วิเคราะห์รูปแบบคำสั่งซื้อและตรวจจับความผิดปกติ (ต้นทุนต่ำมาก $0.42/MTok)
• Gemini 2.5 Flash — สร้างรายงานสรุปและ Dashboard อัตโนมัติ
• Multi-Model Fallback — ระบบสำรองเมื่อโมเดลหลักล่ม ทำให้ระบบทำงานได้ตลอด 24/7

การตั้งค่าโครงสร้างโปรเจกต์

# requirements.txt
requests>=2.28.0
python-dotenv>=1.0.0
pandas>=2.0.0
numpy>=1.24.0
schedule>=1.2.0

import os
import json
import time
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from enum import Enum

import requests
import pandas as pd
import numpy as np

============================================
การตั้งค่า HolySheep API
============================================
⚠️ สำคัญ: base_url ต้องเป็น https://api.holysheep.ai/v1 เท่านั้น
key: YOUR_HOLYSHEEP_API_KEY — ใส่ API key ของคุณที่นี่

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # 👈 แทนที่ด้วย API key จริง

กำหนดโมเดลหลักและโมเดลสำรอง
MODELS = {
    "deepseek": {
        "primary": "deepseek-v3.2",
        "fallback": ["deepseek-v3.1", "deepseek-chat"],
        "cost_per_1k_tokens": 0.42  # USD/MTok
    },
    "gemini": {
        "primary": "gemini-2.5-flash",
        "fallback": ["gemini-2.0-flash", "gemini-pro"],
        "cost_per_1k_tokens": 2.50  # USD/MTok
    }
}

class AlertSeverity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class OrderRecord:
    order_id: str
    customer_id: str
    product_id: str
    quantity: int
    unit_price: float
    total_amount: float
    order_date: datetime
    expected_delivery: datetime
    supplier_id: str
    region: str
    status: str

@dataclass
class AnomalyResult:
    order_id: str
    anomaly_type: str
    severity: AlertSeverity
    description: str
    confidence_score: float
    recommended_action: str
    detected_at: datetime = field(default_factory=datetime.now)

ตั้งค่า Logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("SupplyChainAnomalyAgent")

คลาส HolySheep API Client พร้อมระบบ Fallback

class HolySheepAIClient:
    """
    HolySheep AI API Client พร้อมระบบ Multi-Model Fallback
    รองรับ DeepSeek และ Gemini
    
    📌 ข้อดีของ HolySheep:
       - ความหน่วงต่ำกว่า 50ms
       - ประหยัด 85%+ เมื่อเทียบกับ API อย่างเป็นทางการ
       - รองรับ WeChat/Alipay สำหรับผู้ใช้ในประเทศจีน
    """
    
    def __init__(self, api_key: str, base_url: str = HOLYSHEEP_BASE_URL):
        self.api_key = api_key
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        })
        self.cost_tracker = {"total_tokens": 0, "total_cost_usd": 0}
    
    def _call_model(
        self, 
        model_name: str, 
        messages: List[Dict],
        temperature: float = 0.7,
        max_tokens: int = 2048
    ) -> Dict[str, Any]:
        """
        เรียกใช้โมเดลผ่าน HolySheep API
        """
        endpoint = f"{self.base_url}/chat/completions"
        
        payload = {
            "model": model_name,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        try:
            response = self.session.post(
                endpoint, 
                json=payload, 
                timeout=30
            )
            response.raise_for_status()
            result = response.json()
            
            # ติดตามค่าใช้จ่าย
            usage = result.get("usage", {})
            prompt_tokens = usage.get("prompt_tokens", 0)
            completion_tokens = usage.get("completion_tokens", 0)
            total_tokens = prompt_tokens + completion_tokens
            
            self._track_cost(model_name, total_tokens)
            
            return {
                "success": True,
                "content": result["choices"][0]["message"]["content"],
                "usage": usage,
                "model_used": model_name
            }
            
        except requests.exceptions.Timeout:
            logger.error(f"⏱️ Timeout calling {model_name}")
            return {"success": False, "error": "timeout", "model_used": model_name}
        except requests.exceptions.RequestException as e:
            logger.error(f"❌ Request failed for {model_name}: {str(e)}")
            return {"success": False, "error": str(e), "model_used": model_name}
        except (KeyError, IndexError) as e:
            logger.error(f"⚠️ Invalid response from {model_name}: {str(e)}")
            return {"success": False, "error": "invalid_response", "model_used": model_name}
    
    def _track_cost(self, model_name: str, tokens: int):
        """ติดตามค่าใช้จ่ายสะสม"""
        model_config = self._find_model_config(model_name)
        if model_config:
            cost = (tokens / 1000) * model_config["cost_per_1k_tokens"]
            self.cost_tracker["total_tokens"] += tokens
            self.cost_tracker["total_cost_usd"] += cost
    
    def _find_model_config(self, model_name: str) -> Optional[Dict]:
        """ค้นหาข้อมูลค่าใช้จ่ายของโมเดล"""
        for model_type, config in MODELS.items():
            if model_name in [config["primary"]] + config["fallback"]:
                return config
        return None
    
    def call_with_fallback(
        self,
        model_type: str,
        messages: List[Dict],
        temperature: float = 0.7,
        max_tokens: int = 2048
    ) -> Dict[str, Any]:
        """
        เรียกใช้โมเดลพร้อมระบบ Fallback อัตโนมัติ
        
        หากโมเดลหลักล่ม ระบบจะทดลองโมเดลสำรองตามลำดับ
        ทำให้ระบบทำงานได้อย่างต่อเนื่อง 99.9% Uptime
        """
        if model_type not in MODELS:
            return {"success": False, "error": f"Unknown model type: {model_type}"}
        
        config = MODELS[model_type]
        models_to_try = [config["primary"]] + config["fallback"]
        
        last_error = None
        for model_name in models_to_try:
            logger.info(f"🔄 ลองใช้โมเดล: {model_name} ({model_type})")
            
            result = self._call_model(
                model_name, 
                messages, 
                temperature, 
                max_tokens
            )
            
            if result["success"]:
                logger.info(f"✅ ใช้โมเดล {model_name} สำเร็จ")
                return result
            else:
                last_error = result.get("error", "unknown")
                logger.warning(f"⚠️ โมเดล {model_name} ล้มเหลว: {last_error}")
                time.sleep(0.5)  # รอสักครู่ก่อนลองโมเดลถัดไป
        
        return {
            "success": False, 
            "error": f"All models failed. Last error: {last_error}",
            "models_tried": models_to_try
        }
    
    def get_cost_report(self) -> Dict[str, Any]:
        """ดึงรายงานค่าใช้จ่าย"""
        return {
            "total_tokens_used": self.cost_tracker["total_tokens"],
            "total_cost_usd": round(self.cost_tracker["total_cost_usd"], 4),
            "breakdown_by_model": self._get_cost_breakdown()
        }
    
    def _get_cost_breakdown(self) -> Dict[str, float]:
        """คำนวณค่าใช้จ่ายแยกตามประเภทโมเดล"""
        # โค้ดสำหรับคำนวณค่าใช้จ่ายแยกตามโมเดล
        return {
            "deepseek-v3.2": self.cost_tracker["total_cost_usd"] * 0.7,
            "gemini-2.5-flash": self.cost_tracker["total_cost_usd"] * 0.3
        }

ระบบวิเคราะห์คำสั่งซื้อด้วย DeepSeek

class SupplyChainAnalyzer:
    """
    ระบบวิเคราะห์คำสั่งซื้อและตรวจจับความผิดปกติ
    ใช้ DeepSeek V3.2 สำหรับการวิเคราะห์
    """
    
    def __init__(self, ai_client: HolySheepAIClient):
        self.ai_client = ai_client
        self.historical_patterns = {}
    
    def analyze_order_anomaly(
        self, 
        order: OrderRecord,
        recent_orders: List[OrderRecord]
    ) -> Optional[AnomalyResult]:
        """
        วิเคราะห์คำสั่งซื้อเดี่ยวเพื่อหาความผิดปกติ
        ใช้ DeepSeek V3.2 ซึ่งมีต้นทุนเพียง $0.42/MTok
        """
        # เตรียมข้อมูลสำหรับวิเคราะห์
        order_data = self._prepare_order_context(order, recent_orders)
        
        system_prompt = """คุณเป็นผู้เชี่ยวชาญด้านการวิเคราะห์ห่วงโซ่อุปทาน
        วิเคราะห์คำสั่งซื้อและตรวจจับความผิดปกติ
        คืนค่า JSON ที่มี:
        - anomaly_detected: true/false
        - anomaly_type: "quantity_spike", "price_anomaly", "timing_anomaly", "pattern_break", "none"
        - severity: "low", "medium", "high", "critical"
        - description: คำอธิบายปัญหา
        - confidence_score: 0.0-1.0
        - recommended_action: การดำเนินการที่แนะนำ"""
        
        user_prompt = f"""วิเคราะห์คำสั่งซื้อนี้:

ข้อมูลคำสั่งซื้อปัจจุบัน:
{json.dumps(order_data['current_order'], indent=2, default=str)}

คำสั่งซื้อล่าสุด 10 รายการ:
{json.dumps(order_data['recent_orders'], indent=2, default=str)}

สถิติของลูกค้า:
{json.dumps(order_data['customer_stats'], indent=2)}

ตอบกลับเป็น JSON เท่านั้น"""
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]
        
        # เรียกใช้ DeepSeek ผ่าน HolySheep พร้อม Fallback
        response = self.ai_client.call_with_fallback(
            model_type="deepseek",
            messages=messages,
            temperature=0.3,  # ความแม่นยำสูง
            max_tokens=1024
        )
        
        if not response["success"]:
            logger.error(f"❌ ไม่สามารถวิเคราะห์คำสั่งซื้อ: {response['error']}")
            return None
        
        # แปลผลลัพธ์
        try:
            analysis = json.loads(response["content"])
            
            if analysis.get("anomaly_detected"):
                return AnomalyResult(
                    order_id=order.order_id,
                    anomaly_type=analysis.get("anomaly_type", "unknown"),
                    severity=AlertSeverity(analysis.get("severity", "medium")),
                    description=analysis.get("description", ""),
                    confidence_score=analysis.get("confidence_score", 0.0),
                    recommended_action=analysis.get("recommended_action", "")
                )
        except (json.JSONDecodeError, KeyError) as e:
            logger.error(f"⚠️ ไม่สามารถแปลผลลัพธ์: {e}")
        
        return None
    
    def _prepare_order_context(
        self, 
        order: OrderRecord, 
        recent_orders: List[OrderRecord]
    ) -> Dict[str, Any]:
        """เตรียมข้อมูลบริบทสำหรับการวิเคราะห์"""
        # คำนวณสถิติของลูกค้า
        customer_orders = [o for o in recent_orders if o.customer_id == order.customer_id]
        customer_stats = {
            "avg_order_value": np.mean([o.total_amount for o in customer_orders]) if customer_orders else 0,
            "avg_quantity": np.mean([o.quantity for o in customer_orders]) if customer_orders else 0,
            "order_count": len(customer_orders),
            "last_order_date": customer_orders[0].order_date.isoformat() if customer_orders else None
        }
        
        return {
            "current_order": {
                "order_id": order.order_id,
                "customer_id": order.customer_id,
                "quantity": order.quantity,
                "unit_price": order.unit_price,
                "total_amount": order.total_amount,
                "order_date": order.order_date.isoformat(),
                "region": order.region
            },
            "recent_orders": [
                {
                    "order_id": o.order_id,
                    "quantity": o.quantity,
                    "total_amount": o.total_amount,
                    "order_date": o.order_date.isoformat()
                }
                for o in recent_orders[:10]
            ],
            "customer_stats": customer_stats
        }
    
    def batch_analyze_orders(
        self, 
        orders: List[OrderRecord]
    ) -> List[AnomalyResult]:
        """วิเคราะห์คำสั่งซื้อหลายรายการพร้อมกัน"""
        results = []
        
        for i, order in enumerate(orders):
            logger.info(f"📊 วิเคราะห์คำสั่งซื้อ {i+1}/{len(orders)}: {order.order_id}")
            
            result = self.analyze_order_anomaly(
                order, 
                orders[max(0, i-50):i]  # ใช้ 50 คำสั่งซื้อล่าสุดเป็นบริบท
            )
            
            if result:
                results.append(result)
            
            # หน่วงเวลาเล็กน้อยเพื่อหลีกเลี่ยง Rate Limit
            time.sleep(0.1)
        
        return results

การสร้างรายงานด้วย Gemini 2.5 Flash

class ReportGenerator:
    """
    ระบบสร้างรายงานอัตโนมัติ
    ใช้ Gemini 2.5 Flash ซึ่งเร็วและถูก
    """
    
    def __init__(self, ai_client: HolySheepAIClient):
        self.ai_client = ai_client
    
    def generate_anomaly_report(
        self, 
        anomalies: List[AnomalyResult],
        summary_stats: Dict[str, Any]
    ) -> str:
        """
        สร้างรายงานสรุปความผิดปกติแบบครบถ้วน
        ใช้ Gemini 2.5 Flash ซึ่งมีความเร็วสูงและต้นทุน $2.50/MTok
        """
        if not anomalies:
            return "✅ ไม่พบความผิดปกติในช่วงเวลาที่วิเคราะห์"
        
        # เตรียมข้อมูลรายงาน
        anomalies_data = [
            {
                "order_id": a.order_id,
                "type": a.anomaly_type,
                "severity": a.severity.value,
                "description": a.description,
                "confidence": a.confidence_score,
                "action": a.recommended_action
            }
            for a in anomalies
        ]
        
        system_prompt = """คุณเป็นผู้เชี่ยวชาญด้านการจัดการห่วงโซ่อุปทาน
        สร้างรายงานสรุปที่ชัดเจน กระชับ และมีประโยชน์
        รายงานควรประกอบด้วย:
        1. สรุปภาพรวม (Executive Summary)
        2. รายละเอียดความผิดปกติแต่ละรายการ
        3. ความเสี่ยงและผลกระทบ
        4. ข้อเสนอแนะเชิงปฏิบัติ
        
        เขียนเป็นภาษาไทยที่เป็นทางการ เหมาะสำหรับผู้บริหาร"""
        
        user_prompt = f"""สร้างรายงานสรุปความผิดปกติในห่วงโซ่อุปทาน

สถิติภาพรวม:
{json.dumps(summary_stats, indent=2)}

รายการความผิดปกติ:
{json.dumps(anomalies_data, indent=2, ensure_ascii=False)}

วันที่รายงาน: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"""
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]
        
        # เรียกใช้ Gemini ผ่าน HolySheep พร้อม Fallback
        response = self.ai_client.call_with_fallback(
            model_type="gemini",
            messages=messages,
            temperature=0.5,
            max_tokens=2048
        )
        
        if response["success"]:
            return response["content"]
        else:
            return f"❌ ไม่สามารถสร้างรายงาน: {response['error']}"
    
    def generate_dashboard_data(
        self,
        anomalies: List[AnomalyResult],
        orders: List[OrderRecord]
    ) -> Dict[str, Any]:
        """
        สร้างข้อมูล Dashboard ในรูปแบบ JSON
        """
        # คำนวณ Metrics
        total_orders = len(orders)
        anomaly_count = len(anomalies)
        anomaly_rate = (anomaly_count / total_orders * 100) if total_orders > 0 else 0
        
        # นับตามระดับความรุนแรง
        severity_counts = {
            "critical": 0,
            "high": 0,
            "medium": 0,
            "low": 0
        }
        for a in anomalies:
            severity_counts[a.severity.value] += 1
        
        return {
            "generated_at": datetime.now().isoformat(),
            "metrics": {
                "total_orders_analyzed": total_orders,
                "anomalies_detected": anomaly_count,
                "anomaly_rate_percent": round(anomaly_rate, 2),
                "severity_distribution": severity_counts
            },
            "top_anomalies": [
                {
                    "order_id": a.order_id,
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
รีวิว HolySheep AI: คู่มือฉบับสมบูรณ์สำหรับ Claude Code ในปร
เปรียบเทียบ AI ยังไงให้คุ้ม! Claude vs GPT-4o vs DeepSeek: ว
คู่มือย้ายระบบ API สำหรับการตรวจสอบสัญญา: จาก OpenAI/Anthrop