ในยุคที่ AI API กลายเป็นโครงสร้างพื้นฐานหลักของแอปพลิเคชันสมัยใหม่ การจัดการต้นทุนและการติดตามการใช้งานอย่างมีประสิทธิภาพเป็นสิ่งจำเป็นอย่างยิ่ง บทความนี้จะพาคุณสร้างแดชบอร์ดสถิติการใช้งาน AI API พร้อมระบบแบ่งต้นทุนระหว่างหลายโปรเจกต์ ออกแบบมาสำหรับ production environment ที่รองรับ high-throughput workloads

ทำไมต้องมีระบบจัดการต้นทุน AI API

จากประสบการณ์ในการดูแลระบบ AI ขนาดใหญ่ พบว่าปัญหาหลัก 3 ประการที่ทีม development พบเจอ:

สถาปัตยกรรมระบบแดชบอร์ดต้นทุน

สถาปัตยกรรมที่แนะนำประกอบด้วย 4 ชั้นหลัก:

การตั้งค่า Base Client สำหรับ HolySheep API

เริ่มต้นด้วยการสร้าง base client ที่รวม logging และ cost tracking ไว้ในที่เดียว ผมเลือกใช้ HolySheep AI เพราะมี latency เฉลี่ยต่ำกว่า 50ms และราคาประหยัดกว่า OpenAI ถึง 85%+

import asyncio
import aiohttp
import time
import json
from datetime import datetime
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from collections import defaultdict
import hashlib

@dataclass
class TokenUsage:
    """โครงสร้างข้อมูลสำหรับเก็บ token usage"""
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    model: str = ""
    timestamp: datetime = field(default_factory=datetime.utcnow)
    project_id: str = ""
    request_id: str = ""

@dataclass
class CostEntry:
    """ข้อมูลต้นทุนของแต่ละ request"""
    project_id: str
    model: str
    input_tokens: int
    output_tokens: int
    cost: float
    latency_ms: float
    timestamp: datetime
    endpoint: str

class HolySheepAIClient:
    """Base client สำหรับ HolySheep API พร้อม cost tracking"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    # ราคา per million tokens (USD) - updated 2026
    PRICING = {
        "gpt-4.1": {"input": 8.0, "output": 8.0},
        "claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
        "gemini-2.5-flash": {"input": 2.50, "output": 2.50},
        "deepseek-v3.2": {"input": 0.42, "output": 0.42},
    }
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self._session: Optional[aiohttp.ClientSession] = None
        self._usage_log: List[CostEntry] = []
        self._project_costs: Dict[str, Dict[str, float]] = defaultdict(lambda: defaultdict(float))
        self._rate_limits: Dict[str, float] = {}
        self._last_request_times: Dict[str, float] = {}
        
    async def __aenter__(self):
        self._session = aiohttp.ClientSession(
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            timeout=aiohttp.ClientTimeout(total=60)
        )
        return self
        
    async def __aexit__(self, *args):
        if self._session:
            await self._session.close()
            
    def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """คำนวณต้นทุนจาก token usage"""
        prices = self.PRICING.get(model, {"input": 8.0, "output": 8.0})
        input_cost = (input_tokens / 1_000_000) * prices["input"]
        output_cost = (output_tokens / 1_000_000) * prices["output"]
        return round(input_cost + output_cost, 6)
    
    def _generate_request_id(self, project_id: str, prompt_hash: str) -> str:
        """สร้าง request ID แบบ unique"""
        timestamp = str(time.time())
        return hashlib.sha256(f"{project_id}:{prompt_hash}:{timestamp}".encode()).hexdigest()[:16]
    
    async def chat_completion(
        self,
        project_id: str,
        model: str,
        messages: List[Dict[str, str]],
        temperature: float = 0.7,
        max_tokens: int = 2048
    ) -> Dict[str, Any]:
        """ส่ง request ไปยัง HolySheep API พร้อม tracking"""
        
        start_time = time.perf_counter()
        prompt_hash = hashlib.md5(json.dumps(messages, sort_keys=True).encode()).hexdigest()
        request_id = self._generate_request_id(project_id, prompt_hash)
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        try:
            async with self._session.post(
                f"{self.BASE_URL}/chat/completions",
                json=payload
            ) as response:
                response_data = await response.json()
                end_time = time.perf_counter()
                latency_ms = (end_time - start_time) * 1000
                
                if response.status == 200:
                    usage = response_data.get("usage", {})
                    input_tokens = usage.get("prompt_tokens", 0)
                    output_tokens = usage.get("completion_tokens", 0)
                    cost = self._calculate_cost(model, input_tokens, output_tokens)
                    
                    # บันทึก cost entry
                    entry = CostEntry(
                        project_id=project_id,
                        model=model,
                        input_tokens=input_tokens,
                        output_tokens=output_tokens,
                        cost=cost,
                        latency_ms=latency_ms,
                        timestamp=datetime.utcnow(),
                        endpoint="/chat/completions"
                    )
                    self._usage_log.append(entry)
                    self._project_costs[project_id][model] += cost
                    
                    return {
                        "success": True,
                        "data": response_data,
                        "request_id": request_id,
                        "cost": cost,
                        "latency_ms": latency_ms
                    }
                else:
                    return {
                        "success": False,
                        "error": response_data,
                        "status": response.status
                    }
                    
        except aiohttp.ClientError as e:
            return {"success": False, "error": str(e)}


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

async def example_usage():
    async with HolySheepAIClient("YOUR_HOLYSHEEP_API_KEY") as client:
        # โปรเจกต์ A - Chatbot
        result_a = await client.chat_completion(
            project_id="project-a-chatbot",
            model="deepseek-v3.2",
            messages=[{"role": "user", "content": "สวัสดี บอกข้อมูลเกี่ยวกับ AI"}]
        )
        print(f"Project A Result: {result_a}")
        
        # โปรเจกต์ B - Content Generation
        result_b = await client.chat_completion(
            project_id="project-b-content",
            model="gpt-4.1",
            messages=[{"role": "user", "content": "เขียนบทความเกี่ยวกับ SEO"}]
        )
        print(f"Project B Result: {result_b}")
        
        # ดูสรุปต้นทุน
        print(f"Total Cost by Project: {dict(client._project_costs)}")

if __name__ == "__main__":
    asyncio.run(example_usage())

ระบบ Cost Aggregation และ Real-time Dashboard

หลังจากมี base client แล้ว ต่อไปจะสร้างระบบ aggregation ที่รวมข้อมูลและแสดงผลแบบ real-time ผ่าน WebSocket สำหรับ monitoring dashboard

import asyncio
import websockets
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
import sqlite3
from threading import Lock

@dataclass
class ProjectSummary:
    """สรุปต้นทุนรายโปรเจกต์"""
    project_id: str
    total_requests: int
    total_input_tokens: int
    total_output_tokens: int
    total_cost: float
    avg_latency_ms: float
    last_request: datetime
    cost_this_month: float
    cost_today: float
    cost_this_hour: float

class CostAggregator:
    """ระบบรวมข้อมูลต้นทุนแบบ real-time"""
    
    def __init__(self, db_path: str = "cost_tracking.db"):
        self.db_path = db_path
        self._lock = Lock()
        self._project_cache: Dict[str, ProjectSummary] = {}
        self._setup_database()
        
    def _setup_database(self):
        """สร้าง database schema"""
        with self._lock:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS cost_entries (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    project_id TEXT NOT NULL,
                    model TEXT NOT NULL,
                    input_tokens INTEGER,
                    output_tokens INTEGER,
                    cost REAL,
                    latency_ms REAL,
                    endpoint TEXT,
                    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
                )
            """)
            cursor.execute("""
                CREATE INDEX IF NOT EXISTS idx_project_timestamp 
                ON cost_entries(project_id, timestamp)
            """)
            cursor.execute("""
                CREATE INDEX IF NOT EXISTS idx_timestamp 
                ON cost_entries(timestamp)
            """)
            conn.commit()
            conn.close()
            
    def record_cost(self, entry: CostEntry):
        """บันทึก cost entry ลง database"""
        with self._lock:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO cost_entries 
                (project_id, model, input_tokens, output_tokens, cost, latency_ms, endpoint, timestamp)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                entry.project_id,
                entry.model,
                entry.input_tokens,
                entry.output_tokens,
                entry.cost,
                entry.latency_ms,
                entry.endpoint,
                entry.timestamp
            ))
            conn.commit()
            conn.close()
            
    def get_project_summary(self, project_id: str) -> ProjectSummary:
        """ดึงข้อมูลสรุปของโปรเจกต์"""
        with self._lock:
            conn = sqlite3.connect(self.db_path)
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            
            # Total all time
            cursor.execute("""
                SELECT 
                    COUNT(*) as total_requests,
                    COALESCE(SUM(input_tokens), 0) as total_input,
                    COALESCE(SUM(output_tokens), 0) as total_output,
                    COALESCE(SUM(cost), 0) as total_cost,
                    COALESCE(AVG(latency_ms), 0) as avg_latency,
                    MAX(timestamp) as last_request
                FROM cost_entries
                WHERE project_id = ?
            """, (project_id,))
            row = cursor.fetchone()
            
            # This month
            cursor.execute("""
                SELECT COALESCE(SUM(cost), 0) as monthly_cost
                FROM cost_entries
                WHERE project_id = ?
                AND timestamp >= date('now', 'start of month')
            """, (project_id,))
            monthly = cursor.fetchone()
            
            # Today
            cursor.execute("""
                SELECT COALESCE(SUM(cost), 0) as daily_cost
                FROM cost_entries
                WHERE project_id = ?
                AND date(timestamp) = date('now')
            """, (project_id,))
            daily = cursor.fetchone()
            
            # This hour
            cursor.execute("""
                SELECT COALESCE(SUM(cost), 0) as hourly_cost
                FROM cost_entries
                WHERE project_id = ?
                AND timestamp >= datetime('now', '-1 hour')
            """, (project_id,))
            hourly = cursor.fetchone()
            
            conn.close()
            
            return ProjectSummary(
                project_id=project_id,
                total_requests=row["total_requests"],
                total_input_tokens=row["total_input"],
                total_output_tokens=row["total_output"],
                total_cost=row["total_cost"],
                avg_latency_ms=row["avg_latency"],
                last_request=datetime.fromisoformat(row["last_request"]) if row["last_request"] else None,
                cost_this_month=monthly["monthly_cost"],
                cost_today=daily["daily_cost"],
                cost_this_hour=hourly["hourly_cost"]
            )
            
    def get_all_projects(self) -> List[str]:
        """ดึงรายชื่อโปรเจกต์ทั้งหมด"""
        with self._lock:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            cursor.execute("SELECT DISTINCT project_id FROM cost_entries")
            projects = [row[0] for row in cursor.fetchall()]
            conn.close()
            return projects
            
    def get_cost_by_model(self, project_id: Optional[str] = None) -> Dict[str, float]:
        """ดึงต้นทุนแยกตาม model"""
        with self._lock:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            if project_id:
                cursor.execute("""
                    SELECT model, SUM(cost) as total_cost
                    FROM cost_entries
                    WHERE project_id = ?
                    GROUP BY model
                """, (project_id,))
            else:
                cursor.execute("""
                    SELECT model, SUM(cost) as total_cost
                    FROM cost_entries
                    GROUP BY model
                """)
            results = {row[0]: row[1] for row in cursor.fetchall()}
            conn.close()
            return results

class WebSocketDashboard:
    """WebSocket server สำหรับ real-time dashboard"""
    
    def __init__(self, aggregator: CostAggregator, port: int = 8765):
        self.aggregator = aggregator
        self.port = port
        self.clients: set = set()
        
    async def register(self, websocket):
        self.clients.add(websocket)
        try:
            await websocket.send(json.dumps({
                "type": "connected",
                "message": "Connected to Cost Dashboard"
            }))
        finally:
            self.clients.discard(websocket)
            
    async def broadcast_update(self):
        """ส่งข้อมูลอัปเดตไปยัง clients ทั้งหมด"""
        if not self.clients:
            return
            
        projects = self.aggregator.get_all_projects()
        dashboard_data = {
            "type": "dashboard_update",
            "timestamp": datetime.utcnow().isoformat(),
            "projects": {}
        }
        
        for project_id in projects:
            summary = self.aggregator.get_project_summary(project_id)
            dashboard_data["projects"][project_id] = asdict(summary)
            
        dashboard_data["cost_by_model"] = self.aggregator.get_cost_by_model()
        
        # คำนวณ total
        dashboard_data["total"] = {
            "all_time_cost": sum(
                p.total_cost for p in dashboard_data["projects"].values()
            ),
            "today_cost": sum(
                p["cost_today"] for p in dashboard_data["projects"].values()
            ),
            "monthly_cost": sum(
                p["cost_this_month"] for p in dashboard_data["projects"].values()
            )
        }
        
        message = json.dumps(dashboard_data)
        await asyncio.gather(
            *[client.send(message) for client in self.clients],
            return_exceptions=True
        )
        
    async def start_server(self):
        """เริ่ม WebSocket server"""
        async with websockets.serve(self.register, "localhost", self.port):
            print(f"Dashboard server running on ws://localhost:{self.port}")
            await asyncio.Future()  # run forever


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

async def main():
    aggregator = CostAggregator()
    dashboard = WebSocketDashboard(aggregator)
    
    # สมมติว่ามี entries ใน database แล้ว
    projects = aggregator.get_all_projects()
    print(f"Projects: {projects}")
    
    for project_id in projects:
        summary = aggregator.get_project_summary(project_id)
        print(f"\n{project_id}:")
        print(f"  Total Cost: ${summary.total_cost:.6f}")
        print(f"  Today: ${summary.cost_today:.6f}")
        print(f"  This Month: ${summary.cost_this_month:.6f}")
        
    # เริ่ม dashboard server
    await dashboard.start_server()

if __name__ == "__main__":
    asyncio.run(main())

Benchmark: Performance ของระบบ Cost Tracking

จากการทดสอบใน production environment พบผลลัพธ์ดังนี้ (ทดสอบบน server ที่มี 8 vCPUs, 16GB RAM):

การแบ่งต้นทุนระหว่างหลายโปรเจกต์

ระบบแบ่งต้นทุนที่ดีต้องรองรับหลายรูปแบบการจัดสรร:

from enum import Enum
from dataclasses import dataclass
from typing import Dict, Callable

class AllocationType(Enum):
    FIXED_BUDGET = "fixed_budget"
    PERCENTAGE = "percentage"
    PAY_AS_YOU_GO = "pay_as_you_go"
    TIERED = "tiered"

@dataclass
class ProjectBudget:
    project_id: str
    allocation_type: AllocationType
    limit: float  # งบประมาณหรือเปอร์เซ็นต์
    tiered_thresholds: list = None  # สำหรับ tiered pricing
    
@dataclass  
class BudgetAlert:
    project_id: str
    threshold_percent: float
    current_cost: float
    budget_limit: float
    severity: str  # info, warning, critical

class CostAllocator:
    """ระบบจัดสรรและควบคุมงบประมาณ"""
    
    TIERED_DISCOUNTS = {
        100_000: 0.10,   # >100K tokens: 10% off
        500_000: 0.20,   # >500K tokens: 20% off  
        1_000_000: 0.30,  # >1M tokens: 30% off
    }
    
    def __init__(self, aggregator: CostAggregator):
        self.aggregator = aggregator
        self.budgets: Dict[str, ProjectBudget] = {}
        self.alerts: list[BudgetAlert] = []
        
    def set_budget(self, project_id: str, budget: ProjectBudget):
        """กำหนดงบประมาณสำหรับโปรเจกต์"""
        self.budgets[project_id] = budget
        
    def calculate_effective_cost(self, project_id: str, raw_cost: float) -> float:
        """คำนวณต้นทุนหลังจากส่วนลด tiered"""
        summary = self.aggregator.get_project_summary(project_id)
        total_tokens = summary.total_input_tokens + summary.total_output_tokens
        
        discount = 0.0
        for threshold, tier_discount in sorted(self.TIERED_DISCOUNTS.items()):
            if total_tokens >= threshold:
                discount = tier_discount
                
        return raw_cost * (1 - discount)
    
    def check_budget_status(self, project_id: str) -> Dict:
        """ตรวจสอบสถานะงบประมาณของโปรเจกต์"""
        if project_id not in self.budgets:
            return {"status": "unlimited", "remaining": None}
            
        budget = self.budgets[project_id]
        summary = self.aggregator.get_project_summary(project_id)
        
        if budget.allocation_type == AllocationType.PERCENTAGE:
            # คำนวณจาก total monthly budget
            all_projects = self.aggregator.get_all_projects()
            total_monthly = sum(
                self.aggregator.get_project_summary(p).cost_this_month 
                for p in all_projects
            )
            allocated = total_monthly * (budget.limit / 100)
            remaining = allocated - summary.cost_this_month
            pct = (summary.cost_this_month / allocated * 100) if allocated > 0 else 0
            
        elif budget.allocation_type == AllocationType.FIXED_BUDGET:
            remaining = budget.limit - summary.cost_this_month
            pct = (summary.cost_this_month / budget.limit * 100) if budget.limit > 0 else 0
            
        elif budget.allocation_type == AllocationType.TIERED:
            effective_cost = self.calculate_effective_cost(project_id, summary.total_cost)
            remaining = budget.limit - effective_cost
            pct = (effective_cost / budget.limit * 100) if budget.limit > 0 else 0
            
        else:  # PAY_AS_YOU_GO
            return {"status": "pay_as_you_go", "remaining": None}
            
        # ตรวจสอบ alert thresholds
        if pct >= 100:
            severity = "critical"
        elif pct >= 80:
            severity = "warning"
        elif pct >= 50:
            severity = "info"
        else:
            severity = "ok"
            
        return {
            "status": severity,
            "remaining": max(0, remaining),
            "spent": summary.cost_this_month,
            "budget": budget.limit,
            "percent_used": pct,
            "daily_cost": summary.cost_today,
            "hourly_cost": summary.cost_this_hour
        }
        
    def should_block_request(self, project_id: str) -> tuple[bool, str]:
        """ตรวจสอบว่าควร block request หรือไม่"""
        if project_id not in self.budgets:
            return False, ""
            
        budget = self.budgets[project_id]
        
        # Pay-as-you-go ไม่ block
        if budget.allocation_type == AllocationType.PAY_AS_YOU_GO:
            return False, ""
            
        status = self.check_budget_status(project_id)
        
        if status["status"] == "critical":
            return True, f"Budget exceeded for project {project_id}"
        elif status["status"] == "warning":
            # แจ้งเตือนแต่ไม่ block
            return False, f"Budget warning: {status['percent_used']:.1f}% used"
            
        return False, ""
        
    def get_all_budget_status(self) -> Dict[str, Dict]:
        """ดึงสถานะงบประมาณทั้งหมด"""
        all_projects = self.aggregator.get_all_projects()
        return {p: self.check_budget_status(p) for p in all_projects}
    
    def generate_report(self) -> Dict:
        """สร้างรายงานสรุปการใช้งาน"""
        all_status = self.get_all_budget_status()
        
        total_spent = sum(
            s.get("spent", 0) for s in all_status.values()
        )
        total_budget = sum(
            self.budgets[p].limit for p in self.budgets.keys() 
            if self.budgets[p].allocation_type != AllocationType.PAY_AS_YOU_GO
        )
        
        return {
            "period": "current_month",
            "total_spent": total_spent,
            "total_budget": total_budget,
            "utilization_percent": (total_spent / total_budget * 100) if total_budget > 0 else 0,
            "project_breakdown": all_status,
            "cost

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง