เมื่อเดือนที่แล้ว ทีมงานของผมเจอปัญหาใหญ่หลวง — ระบบ Content Moderation เดิมที่ใช้ Claude Sonnet ตรวจจับรูปภาพและข้อความแยกกัน เริ่มมีปัญหาเมื่อผู้ใช้งานส่งคอนเทนต์ที่ผสมผสานทั้งสองอย่าง เช่น รูปภาพที่มีข้อความซ้อนอยู่ข้างใน หรือ caption ที่อ้างอิงถึงรูปภาพ ระบบเดิมมองข้าม context ระหว่างรูปภาพและข้อความ ทำให้เนื้อหาที่ไม่เหมาะสมบางส่วนหลุดรอดไปได้
หลังจากทดสอบหลายโมเดล พบว่า Gemini 2.5 Flash ที่มีความสามารถ multi-modal ในตัว สามารถวิเคราะห์รูปภาพและข้อความพร้อมกันได้อย่างมีประสิทธิภาพ แถมราคาถูกกว่า Claude Sonnet 4.5 ถึง 6 เท่า ($2.50 vs $15 ต่อล้าน token) ในบทความนี้จะสอนวิธีพัฒนา API สำหรับตรวจจับเนื้อหาที่ไม่เหมาะสมแบบครบวงจร
ปัญหาจริงที่พบ: Multi-Modal Content หลุดรอดการตรวจจับ
ระบบ Content Moderation แบบเดิมที่ใช้โมเดลเดียวตรวจจับรูปภาพ และอีกโมเดลตรวจจับข้อความ มีข้อจำกัดสำคัญ:
- ขาด Context ข้าม Modal: รูปภาพอาจดูปกติ แต่เมื่อรวมกับข้อความ caption แล้วกลับเป็นเนื้อหาที่ไม่เหมาะสม
- ค่าใช้จ่ายสูง: เรียก API 2 ครั้งต่อ 1 คอนเทนต์ ทำให้ต้นทุนเพิ่มขึ้นเป็นเท่าตัว
- Latency สูง: รอผลจากทั้งสอง API ทำให้ response time เพิ่มขึ้นเกือบ 2 เท่า
ด้วย Gemini 2.5 Flash multi-modal เราสามารถส่งรูปภาพและข้อความใน request เดียว ระบบจะวิเคราะห์ทั้งสองส่วนพร้อมกัน และส่งผลลัพธ์กลับมาพร้อม confidence score สำหรับแต่ละประเภทเนื้อหา
การตั้งค่า Environment และการติดตั้ง Dependencies
# สร้าง virtual environment (Python 3.10+)
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
ติดตั้ง packages ที่จำเป็น
pip install openai httpx python-dotenv pillow aiofiles
หรือใช้ Poetry
poetry add openai httpx python-dotenv pillow aiofilesสร้างไฟล์ .env สำหรับเก็บ API key:
# .env
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
หากยังไม่มี API key สามารถสมัครได้ที่ https://www.holysheep.ai/register
โค้ดพื้นฐาน: Multi-Modal Content Moderation API
import os
import base64
from io import BytesIO
from openai import OpenAI
from PIL import Image
from dotenv import load_dotenv
โหลด environment variables
load_dotenv()
สร้าง client สำหรับ HolySheep AI
⚠️ สำคัญ: base_url ต้องเป็น https://api.holysheep.ai/v1 เท่านั้น
client = OpenAI(
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1" # ไม่ใช่ api.openai.com!
)
def encode_image_to_base64(image_path: str) -> str:
"""แปลงรูปภาพเป็น base64 string"""
with Image.open(image_path) as img:
# แปลง RGBA เป็น RGB (หากมี alpha channel)
if img.mode == 'RGBA':
img = img.convert('RGB')
# Resize รูปภาพให้เล็กลงเพื่อลดขนาด (max 1024px)
max_size = 1024
if max(img.size) > max_size:
ratio = max_size / max(img.size)
new_size = tuple(int(dim * ratio) for dim in img.size)
img = img.resize(new_size, Image.Resampling.LANCZOS)
buffer = BytesIO()
img.save(buffer, format="JPEG", quality=85)
return base64.b64encode(buffer.getvalue()).decode("utf-8")
def moderate_content(
image_path: str,
text: str = None,
user_id: str = None
) -> dict:
"""
ตรวจจับเนื้อหาที่ไม่เหมาะสมจากรูปภาพและข้อความร่วมกัน
Args:
image_path: พาธของรูปภาพ
text: ข้อความ caption/description (optional)
user_id: ID ของผู้ใช้ (สำหรับ logging)
Returns:
dict ที่มีผลการตรวจจับพร้อม confidence scores
"""
# เตรียมรูปภาพ
base64_image = encode_image_to_base64(image_path)
# สร้าง prompt สำหรับ content moderation
moderation_prompt = """คุณเป็นระบบ Content Moderation ที่ตรวจจับเนื้อหาที่ไม่เหมาะสม
วิเคราะห์รูปภาพและข้อความที่ให้มา แล้วตอบกลับเป็น JSON format ดังนี้:
{
"is_safe": true/false,
"categories": {
"violence": {"detected": true/false, "confidence": 0.0-1.0, "details": "..."},
"adult": {"detected": true/false, "confidence": 0.0-1.0, "details": "..."},
"hate_speech": {"detected": true/false, "confidence": 0.0-1.0, "details": "..."},
"harassment": {"detected": true/false, "confidence": 0.0-1.0, "details": "..."},
"dangerous_content": {"detected": true/false, "confidence": 0.0-1.0, "details": "..."},
"spam": {"detected": true/false, "confidence": 0.0-1.0, "details": "..."}
},
"overall_confidence": 0.0-1.0,
"summary": "สรุปการตรวจจับภาษาไทย",
"recommended_action": "allow/warn/block"
}
หากพบเนื้อหาที่ไม่เหมาะสม ให้ระบุ confidence สูงสุดเท่ากับ 1.0
หากไม่พบ ให้ confidence เป็น 0.0-0.3"""
# เตรียม messages
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": moderation_prompt
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
]
# เพิ่มข้อความ text หากมี
if text:
messages[0]["content"].insert(1, {
"type": "text",
"text": f"ข้อความที่ต้องวิเคราะห์เพิ่มเติม: {text}"
})
# เรียก API
response = client.chat.completions.create(
model="gemini-2.0-flash",
messages=messages,
response_format={"type": "json_object"},
temperature=0.1 # ความแม่นยำสูง ลดความสุ่ม
)
import json
result = json.loads(response.choices[0].message.content)
# เพิ่ม metadata
result["metadata"] = {
"user_id": user_id,
"image_path": image_path,
"text_length": len(text) if text else 0,
"tokens_used": response.usage.total_tokens,
"latency_ms": response.response_ms if hasattr(response, 'response_ms') else None
}
return result
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# ทดสอบกับรูปภาพตัวอย่าง
result = moderate_content(
image_path="test_image.jpg",
text="ช่วงเทศกาลลดราคา คลิกเลย!",
user_id="user_12345"
)
print(f"Safe: {result['is_safe']}")
print(f"Action: {result['recommended_action']}")
print(f"Confidence: {result['overall_confidence']}")
print(f"Tokens: {result['metadata']['tokens_used']}")เวอร์ชัน Async สำหรับ High-Traffic Application
import asyncio
import os
import base64
from io import BytesIO
from openai import AsyncOpenAI
from PIL import Image
from dotenv import load_dotenv
from typing import List, Optional
load_dotenv()
Async client สำหรับ HolySheep AI
async_client = AsyncOpenAI(
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
class AsyncContentModerator:
"""Async Content Moderation รองรับ high-traffic"""
def __init__(self, max_concurrent: int = 10):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.categories = [
"violence", "adult", "hate_speech",
"harassment", "dangerous_content", "spam"
]
async def moderate_single(
self,
image_bytes: bytes,
text: Optional[str] = None
) -> dict:
"""ตรวจจับเนื้อหาจากรูปภาพเดียว"""
async with self.semaphore:
# เตรียมรูปภาพ
image_base64 = base64.b64encode(image_bytes).decode("utf-8")
# สร้าง prompt
prompt = """ตรวจจับเนื้อหาที่ไม่เหมาะสมจากรูปภาพและข้อความ
ตอบเป็น JSON พร้อม categories ต่อไปนี้: violence, adult, hate_speech, harassment, dangerous_content, spam
แต่ละ category มี detected (boolean), confidence (0-1), details (string)
{
"is_safe": boolean,
"categories": {
"violence": {"detected": boolean, "confidence": float, "details": string},
"adult": {"detected": boolean, "confidence": float, "details": string},
...
},
"overall_confidence": float,
"summary": string,
"recommended_action": "allow|warn|block"
}"""
# เตรียม content list
content = [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}
}
]
if text:
content.insert(1, {"type": "text", "text": text})
# เรียก API พร้อมวัดเวลา
import time
start_time = time.perf_counter()
response = await async_client.chat.completions.create(
model="gemini-2.0-flash",
messages=[{"role": "user", "content": content}],
response_format={"type": "json_object"},
temperature=0.1
)
latency_ms = (time.perf_counter() - start_time) * 1000
import json
result = json.loads(response.choices[0].message.content)
result["_meta"] = {
"latency_ms": round(latency_ms, 2),
"tokens": response.usage.total_tokens
}
return result
async def moderate_batch(
self,
items: List[dict]
) -> List[dict]:
"""
ตรวจจับเนื้อหาหลายรายการพร้อมกัน
Args:
items: List of {"image_bytes": bytes, "text": str|None, "id": str}
"""
tasks = [
self.moderate_single(item["image_bytes"], item.get("text"))
for item in items
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# เพิ่ม ID กลับเข้าไปในผลลัพธ์
final_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
final_results.append({
"error": str(result),
"id": items[i].get("id")
})
else:
result["id"] = items[i].get("id")
final_results.append(result)
return final_results
ตัวอย่างการใช้งาน async version
async def main():
moderator = AsyncContentModerator(max_concurrent=10)
# ตัวอย่าง: ตรวจจับหลายรูปภาพพร้อมกัน
with open("image