บทความนี้เป็นบันทึกจากประสบการณ์ตรงของทีมวิศวกร HolySheep AI ในการสร้างระบบ Multi-Modal Model Routing ที่รองรับการประมวลผลภาพ (images), วิดีโอ (videos), และข้อความ (text) พร้อมกัน พร้อมทั้งระบบ Cost Attribution ที่ช่วยให้องค์กรติดตามค่าใช้จ่ายได้ละเอียดระดับ Request-level ผ่าน แพลตฟอร์ม HolySheep AI
ทำไมต้องสร้าง Multi-Modal Router
จากการวิเคราะห์ Log การใช้งานจริงในระบบ Production ของเรา พบว่า Request ที่เข้ามาแต่ละประเภทมี "ลักษณะการใช้งาน" ที่ต่างกันมาก:
- Text-only requests — คิดเป็น 65% ของ Total requests แต่ใช้ Cost เพียง 15%
- Image + Text requests — คิดเป็น 28% ของ Total requests ใช้ Cost 55%
- Video + Text requests — คิดเป็น 7% ของ Total requests ใช้ Cost 30%
หากเราส่ง Request ทุกประเภทไปยัง Model เดียวกันโดยไม่มีการ Route จะเกิด "ความสิ้นเปลือง" อย่างมหาศาล ดังนั้นเราจึงออกแบบระบบ Routing ที่แยก Request ตาม Modality ตามความเหมาะสม
สถาปัตยกรรม Multi-Modal Router
ระบบที่เราสร้างประกอบด้วย 4 ชั้นหลัก:
+------------------------------------------+
| API Gateway (FastAPI) |
+------------------------------------------+
| | |
v v v
+------+ +--------+ +-----------+
| Text | | Image | | Video |
|Router| | Router | | Router |
+------+ +--------+ +-----------+
| | |
v v v
+------+ +--------+ +-----------+
|Gemini| |Gemini 2.| |Gemini 2.5|
|Flash | |Flash | |Pro |
+------+ +--------+ +-----------+
| | |
v v v
+------------------------------------------+
| Cost Attribution Service |
+------------------------------------------+การตรวจจับประเภท Request
ขั้นตอนแรกคือการตรวจจับว่า Request ที่เข้ามาเป็นประเภทใด เราใช้ Python Class ที่ออกแบบมาเพื่อการนี้โดยเฉพาะ:
import base64
import re
from enum import Enum
from typing import Dict, List, Union, Optional
from dataclasses import dataclass
class ModalityType(Enum):
TEXT_ONLY = "text_only"
IMAGE_TEXT = "image_text"
VIDEO_TEXT = "video_text"
AUDIO_TEXT = "audio_text"
MIXED = "mixed" # หลายประเภทพร้อมกัน
@dataclass
class ContentPart:
type: str
data: str
mime_type: Optional[str] = None
class ModalityDetector:
"""
ตรวจจับประเภท Modality ของ Request สำหรับ Gemini 2.5 Pro
รองรับ: text, image (base64), video, audio
"""
IMAGE_MIME_PATTERNS = [
'image/jpeg', 'image/png', 'image/gif', 'image/webp',
'image/heic', 'image/heif'
]
VIDEO_MIME_PATTERNS = [
'video/mp4', 'video/mpeg', 'video/webm',
'video/quicktime', 'video/x-msvideo'
]
AUDIO_MIME_PATTERNS = [
'audio/mpeg', 'audio/wav', 'audio/ogg',
'audio/webm', 'audio/mp3'
]
# Pattern สำหรับตรวจจับ Base64 Image
BASE64_IMAGE_PATTERN = re.compile(
r'data:image/[^;]+;base64,([A-Za-z0-9+/=]+)'
)
@classmethod
def detect(cls, contents: List[Dict]) -> ModalityType:
"""
ตรวจจับประเภท Modality จาก contents list
Args:
contents: List of content objects ตาม format ของ Gemini API
[{"type": "text", "text": "..."},
{"type": "image", "source": {"type": "base64", ...}}]
Returns:
ModalityType enum value
"""
has_text = False
has_image = False
has_video = False
has_audio = False
for content in contents:
content_type = content.get("type", "")
if content_type == "text":
has_text = True
elif content_type == "image" or content_type == "inline_data":
source = content.get("source", content)
mime_type = source.get("mime_type", "")
# ตรวจจับจาก MIME type
if any(mime in mime_type.lower() for mime in cls.IMAGE_MIME_PATTERNS):
has_image = True
elif any(mime in mime_type.lower() for mime in cls.VIDEO_MIME_PATTERNS):
has_video = True
elif any(mime in mime_type.lower() for mime in cls.AUDIO_MIME_PATTERNS):
has_audio = True
else:
# ลองตรวจจับจาก data URI pattern
data = source.get("data", "")
if cls.BASE64_IMAGE_PATTERN.search(data):
has_image = True
elif content_type == "video":
has_video = True
elif content_type == "audio":
has_audio = True
# ตัดสินใจประเภท
modalities = [has_text, has_image, has_video, has_audio]
count = sum(modalities)
if count == 0:
return ModalityType.TEXT_ONLY
elif count > 1:
return ModalityType.MIXED
elif has_image:
return ModalityType.IMAGE_TEXT
elif has_video:
return ModalityType.VIDEO_TEXT
elif has_audio:
return ModalityType.AUDIO_TEXT
elif has_text:
return ModalityType.TEXT_ONLY
return ModalityType.TEXT_ONLY
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# Test 1: Text only
text_request = [{"type": "text", "text": "สวัสดีครับ"}]
result = ModalityDetector.detect(text_request)
print(f"Text Request: {result.value}") # Output: text_only
# Test 2: Image + Text
image_request = [
{"type": "text", "text": "อธิบายภาพนี้"},
{"type": "image", "source": {"type": "base64", "mime_type": "image/jpeg", "data": "..."}}
]
result = ModalityDetector.detect(image_request)
print(f"Image+Text: {result.value}") # Output: image_text
# Test 3: Video + Text
video_request = [
{"type": "text", "text": "วิเคราะห์วิดีโอนี้"},
{"type": "video", "source": {"type": "base64", "mime_type": "video/mp4", "data": "..."}}
]
result = ModalityDetector.detect(video_request)
print(f"Video+Text: {result.value}") # Output: video_textการเลือก Model ตามประเภท Modality
หลังจากตรวจจับประเภท Request ได้แล้ว ขั้นตอนถัดไปคือการ Route ไปยัง Model ที่เหมาะสม โดยคำนึงถึง:
- ความสามารถของ Model — บาง Model รองรับเฉพาะ Text หรือ Image เท่านั้น
- ต้นทุน — ราคาต่อ Million Tokens แตกต่างกันมาก
- Latency — บางงานต้องการ Response ที่รวดเร็ว
- คุณภาพ — บางงานต้องการความแม่นยำสูงสุด
from enum import Enum
from typing import Dict, Optional
import hashlib
import time
class ModelType(Enum):
# Model ที่รองรับ Multi-Modal บน HolySheep
GEMINI_2_5_PRO = "gemini-2.5-pro" # ราคา $2.50/MTok
GEMINI_2_5_FLASH = "gemini-2.5-flash" # ราคา $0.25/MTok
GEMINI_FLASH = "gemini-1.5-flash" # ราคา $0.075/MTok
GPT_4_1 = "gpt-4.1" # ราคา $8/MTok
CLAUDE_SONNET = "claude-sonnet-4-5" # ราคา $15/MTok
@dataclass
class ModelConfig:
model_id: str
supports_vision: bool
supports_video: bool
cost_per_mtok_input: float
cost_per_mtok_output: float
avg_latency_ms: float
max_tokens: int
class MultiModalRouter:
"""
Router หลักสำหรับเลือก Model ที่เหมาะสม
อิงตามโครงสร้างราคาของ HolySheep AI
"""
# กำหนด Model Config (อ้างอิงจากราคา HolySheep 2026)
MODEL_CONFIGS: Dict[str, ModelConfig] = {
"gemini-2.5-pro": ModelConfig(
model_id="gemini-2.5-pro",
supports_vision=True,
supports_video=True,
cost_per_mtok_input=2.50,
cost_per_mtok_output=10.00,
avg_latency_ms=1200,
max_tokens=32768
),
"gemini-2.5-flash": ModelConfig(
model_id="gemini-2.5-flash",
supports_vision=True,
supports_video=True,
cost_per_mtok_input=0.25,
cost_per_mtok_output=1.00,
avg_latency_ms=350,
max_tokens=32768
),
"gemini-1.5-flash": ModelConfig(
model_id="gemini-1.5-flash",
supports_vision=True,
supports_video=False,
cost_per_mtok_input=0.075,
cost_per_mtok_output=0.30,
avg_latency_ms=180,
max_tokens=8192
),
"gpt-4.1": ModelConfig(
model_id="gpt-4.1",
supports_vision=True,
supports_video=False,
cost_per_mtok_input=8.00,
cost_per_mtok_output=24.00,
avg_latency_ms=800,
max_tokens=16384
),
"claude-sonnet-4-5": ModelConfig(
model_id="claude-sonnet-4-5",
supports_vision=True,
supports_video=False,
cost_per_mtok_input=15.00,
cost_per_mtok_output=75.00,
avg_latency_ms=1500,
max_tokens=8192
),
}
# Cost Threshold สำหรับ Fallback
COST_THRESHOLDS = {
ModalityType.TEXT_ONLY: 0.10, # < $0.10 → Flash
ModalityType.IMAGE_TEXT: 0.50, # < $0.50 → Flash vision
ModalityType.VIDEO_TEXT: 2.00, # < $2.00 → 2.5 Flash
# สูงกว่านี้ → 2.5 Pro
}
# Quality Threshold
QUALITY_REQUIRED_TASKS = [
"code generation", "complex reasoning", "analysis",
"translation", "summarization", "creative writing"
]
def __init__(self, base_url: str = "https://api.holysheep.ai/v1"):
self.base_url = base_url
def estimate_cost(self, model_id: str, contents: List[Dict],
estimated_output_tokens: int = 500) -> float:
"""
ประมาณค่าใช้จ่ายของ Request
Args:
model_id: Model ที่จะใช้
contents: Request contents
estimated_output_tokens: ประมาณการจำนวน Output tokens
Returns:
ค่าใช้จ่ายโดยประมาณในหน่วย USD
"""
config = self.MODEL_CONFIGS.get(model_id)
if not config:
return 0.0
# นับ Input tokens (สมมติ 4 tokens ต่อคำภาษาไทย, 1 token ต่อ 4 chars ภาษาอังกฤษ)
input_text = ""
image_count = 0
video_count = 0
for content in contents:
if content.get("type") == "text":
input_text += content.get("text", "")
elif content.get("type") in ["image", "inline_data"]:
image_count += 1
elif content.get("type") == "video":
video_count += 1
# Rough estimation
input_tokens = len(input_text) / 4 # สมมติ avg 4 chars/token
input_tokens += image_count * 300 # 300 tokens/image
input_tokens += video_count * 5000 # 5000 tokens/video
cost_input = (input_tokens / 1_000_000) * config.cost_per_mtok_input
cost_output = (estimated_output_tokens / 1_000_000) * config.cost_per_mtok_output
return cost_input + cost_output
def route(self, modality: ModalityType,
estimated_cost_budget: Optional[float] = None,
latency_priority: bool = False,
quality_priority: bool = False) -> str:
"""
เลือก Model ที่เหมาะสมตามเงื่อนไข
Args:
modality: ประเภทของ Request
estimated_cost_budget: งบประมาณสูงสุด (USD)
latency_priority: ต้องการ Latency ต่ำ
quality_priority: ต้องการคุณภาพสูงสุด
Returns:
Model ID ที่เหมาะสม
"""
# Priority 1: Video → Gemini 2.5 Pro (รองรับเท่านั้น)
if modality == ModalityType.VIDEO_TEXT:
if quality_priority:
return "gemini-2.5-pro"
return "gemini-2.5-flash" # ประหยัดกว่า 90%
# Priority 2: Quality Priority → ใช้ Model ดีที่สุด
if quality_priority:
return "gemini-2.5-pro"
# Priority 3: Latency Priority → ใช้ Flash
if latency_priority:
if modality == ModalityType.IMAGE_TEXT:
return "gemini-1.5-flash" # ถูกที่สุดและเร็วที่สุด
return "gemini-2.5-flash"
# Priority 4: Budget Check → เลือกตามงบ
if estimated_cost_budget is not None:
if modality == ModalityType.TEXT_ONLY:
if estimated_cost_budget < 0.10:
return "gemini-1.5-flash"
return "gemini-2.5-flash"
elif modality == ModalityType.IMAGE_TEXT:
if estimated_cost_budget < 0.50:
return "gemini-1.5-flash"
return "gemini-2.5-flash"
# Default: เลือกตาม Modality
if modality == ModalityType.TEXT_ONLY:
return "gemini-2.5-flash" # ถูกกว่า GPT-4.1 ถึง 97%
elif modality == ModalityType.IMAGE_TEXT:
return "gemini-2.5-flash" # ราคา $0.25 vs $8
elif modality == ModalityType.MIXED:
return "gemini-2.5-pro"
return "gemini-2.5-flash"
ตัวอย่างการใช้งาน
router = MultiModalRouter()
Route ตาม Modality
test_cases = [
(ModalityType.TEXT_ONLY, "Text summarization"),
(ModalityType.IMAGE_TEXT, "OCR + Analysis"),
(ModalityType.VIDEO_TEXT, "Video understanding"),
]
for modality, task in test_cases:
model = router.route(modality)
print(f"{task}: {model}")ระบบ Cost Attribution
สำหรับองค์กรที่ต้องการ Track ค่าใช้จ่ายระดับ Team/Project/User ระบบ Cost Attribution ของ HolySheep ช่วยให้สามารถ:
- แบ่งค่าใช้จ่ายตาม API Key หรือ Metadata
- สร้างรายงานแยกตาม Department
- ตั้ง Budget Alert ระดับ Project
import json
from datetime import datetime
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field, asdict
from collections import defaultdict
import hashlib
@dataclass
class CostRecord:
"""บันทึกค่าใช้จ่ายของแต่ละ Request"""
request_id: str
timestamp: datetime
model_id: str
modality_type: str
input_tokens: int
output_tokens: int
cost_usd: float
# Attribution Fields
project_id: Optional[str] = None
user_id: Optional[str] = None
department: Optional[str] = None
api_key_id: Optional[str] = None
# Request Metadata
latency_ms: float = 0.0
cache_hit: bool = False
success: bool = True
error_message: Optional[str] = None
class CostAttributionService:
"""
บริการติดตามและ Attribution ค่าใช้จ่าย
ใช้ Metadata จาก Request Header หรือ Body
"""
def __init__(self):
self.records: List[CostRecord] = []
self._project_budgets: Dict[str, float] = {}
self._department_costs: Dict[str, float] = defaultdict(float)
def create_request_id(self, prefix: str = "req") -> str:
"""สร้าง Request ID ที่ไม่ซ้ำกัน"""
timestamp = datetime.now().isoformat()
hash_input = f"{prefix}:{timestamp}"
return f"{prefix}_{hashlib.md5(hash_input.encode()).hexdigest()[:12]}"
def extract_attribution(self, request_data: Dict[str, Any]) -> Dict[str, Optional[str]]:
"""
แยก Attribution metadata จาก Request
รองรับ 3 วิธี:
1. X-Attribution-* Headers (แนะนำ)
2. metadata object ใน request body
3. Default values
"""
attribution = {
"project_id": None,
"user_id": None,
"department": None,
"api_key_id": None
}
# วิธีที่ 1: จาก Headers (假设 FastAPI)
headers = request_data.get("headers", {})
attribution["project_id"] = headers.get("x-project-id")
attribution["user_id"] = headers.get("x-user-id")
attribution["department"] = headers.get("x-department")
attribution["api_key_id"] = headers.get("x-api-key-id")
# วิธีที่ 2: จาก Request Body
metadata = request_data.get("metadata", {})
if not attribution["project_id"]:
attribution["project_id"] = metadata.get("project_id")
if not attribution["user_id"]:
attribution["user_id"] = metadata.get("user_id")
if not attribution["department"]:
attribution["department"] = metadata.get("department")
# วิธีที่ 3: Default จาก API Key
if not attribution["api_key_id"]:
api_key = headers.get("authorization", "")
if api_key.startswith("Bearer "):
api_key = api_key[7:]
if api_key:
attribution["api_key_id"] = self._hash_api_key(api_key)
return attribution
def _hash_api_key(self, api_key: str) -> str:
"""Hash API Key เพื่อไม่เก็บ Key จริงใน Log"""
return hashlib.sha256(api_key.encode()).hexdigest()[:8]
def calculate_cost(self, model_id: str, input_tokens: int,
output_tokens: int, cache_hit: bool = False) -> float:
"""
คำนวณค่าใช้จ่ายตาม Model และ Token count
ราคาอ้างอิงจาก HolySheep (2026):
- Gemini 2.5 Pro: $2.50/M input, $10/M output
- Gemini 2.5 Flash: $0.25/M input, $1/M output
- Gemini 1.5 Flash: $0.075/M input, $0.30/M output
"""
pricing = {
"gemini-2.5-pro": {"input": 2.50, "output": 10.00},
"gemini-2.5-flash": {"input": 0.25, "output": 1.00},
"gemini-1.5-flash": {"input": 0.075, "output": 0.30},
"gpt-4.1": {"input": 8.00, "output": 24.00},
"claude-sonnet-4-5": {"input": 15.00, "output": 75.00},
}
model_pricing = pricing.get(model_id, pricing["gemini-2.5-flash"])
input_cost = (input_tokens / 1_000_000) * model_pricing["input"]
output_cost = (output_tokens / 1_000_000) * model_pricing["output"]
total_cost = input_cost + output_cost
# Cache discount (ถ้ามี)
if cache_hit:
total_cost *= 0.1 # ลด 90% สำหรับ cache hit
return round(total_cost, 6) # 6 decimal places
def record_request(self,
request_data: Dict[str, Any],
response_data: Optional[Dict[str, Any]] = None,
error: Optional[str] = None) -> CostRecord:
"""
บันทึกค่าใช้จ่ายของ Request
Args:
request_data: ข้อมูล Request ที่ส่งเข้ามา
response_data: ข้อมูล Response (สำหรับนับ tokens)
error: ข้อความ error (ถ้ามี)
Returns:
CostRecord object
"""
# ดึง Attribution
attribution = self.extract_attribution(request_data)
# ดึง Model และ Modality
model_id = request_data.get("model", "gemini-2.5-flash")
modality = request_data.get("modality", "text_only")
# นับ Tokens
input_tokens = self._count_input_tokens(request_data.get("contents", []))
output_tokens = 0
cache_hit = False
if response_data:
usage = response_data.get("usage", {})
input_tokens = usage.get("prompt_tokens", input_tokens)
output_tokens = usage.get("completion_tokens", 0)
cache_hit = usage.get("cache_hit", False)
# คำนวณค่าใช้จ่าย
cost = self.calculate_cost(model_id, input_tokens, output_tokens, cache_hit)
# สร้าง Record
record = CostRecord(
request_id=self.create_request_id(),
timestamp=datetime.now(),
model_id=model_id,
modality_type=modality,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=cost,
latency_ms=request_data.get("latency_ms", 0),
cache_hit=cache_hit,
success=error is None,
error_message=error,
**attribution
)
self.records.append(record)
# อัพเดท Budget tracking
self._update_budget_tracking(record)
return record
def _count_input_tokens(self, contents: List[Dict]) -> int:
"""นับ Input tokens โดยประมาณ"""
total = 0
for content in contents:
if content.get("type") == "text":
total += len(content.get("text", "")) // 4
elif content.get("type") == "image":
total += 300 # ~300 tokens per image
elif content.get("type") == "video":
total += 5000 # ~5000 tokens per video
return total
def _update_budget_tracking(self, record: CostRecord):
"""อัพเดท Budget tracking"""
if record.department:
self._department_costs[record.department] += record.cost_usd
if record.project_id:
if record.project_id not in self._project_budgets:
self._project_budgets[record.project_id] = 0
self._project_budgets[record.project_id] += record.cost_usd
def get_cost_report(self,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
department: Optional[str] = None,
project_id: Optional[str] = None) -> Dict[str, Any]:
"""
สร้างรายงานค่าใช้จ่าย
Returns:
Dictionary ที่มี breakdown ของค่าใช้จ่าย
"""
filtered = self.records
if start_date:
filtered = [r for r in filtered if r.timestamp >= start_date]
if end_date:
filtered = [r for r in filtered if r.timestamp <= end_date]
if department:
filtered = [r for r in filtered if r.department == department]
if project_id:
filtered = [r for r in filtered if r.project_id == project_id]
# Calculate totals
total_cost = sum(r.cost_usd for r in filtered)
total_requests = len(filtered)
success_requests = sum(1 for r in filtered if r.success)
failed_requests = total_requests - success_requests
total_input_tokens = sum(r.input_tokens for r in filtered)
total_output_tokens = sum(r.output_tokens for r in filtered)
# Breakdown by model
model_breakdown = defaultdict(lambda: {"cost": 0, "requests": 0})
for r in filtered:
model_breakdown[r.model_id]["cost"] += r.cost_usd
model_breakdown[r.model_id]["requests"] += 1
# Breakdown by modality
modality_breakdown = defaultdict(lambda: {"cost": 0, "requests": 0})
for r in filtered:
modality_breakdown[r.modality_type]["cost"] += r.cost_usd
modality_breakdown[r.modality_type]["requests"] += 1
# Breakdown by department
dept_breakdown = {}
for r in filtered:
if r.department:
if r.department not in dept_breakdown:
dept_breakdown[r.department] = {"cost": 0, "requests": 0}
dept_breakdown[r.department]["cost"] += r.cost_usd
dept_breakdown[r.department]["requests"] += 1
return {
"summary": {
"total_cost_usd": round(total_cost, 4),
"total_requests": total_requests,
"success_rate": f"{success_requests/total