การกรองเนื้อหาภาษาไทยใน AI Output: ระบบตรวจจับคำหยาบและกลยุทธ์ความปลอดภัย

ในฐานะวิศวกรที่ดูแลระบบ AI มาหลายปี ผมพบว่าการสร้างระบบกรองเนื้อหาที่เชื่อถือได้เป็นสิ่งจำเป็นอย่างยิ่งสำหรับ production environment บทความนี้จะพาคุณสำรวจสถาปัตยกรรมที่ใช้งานได้จริง พร้อมโค้ดที่พร้อม deploy

ทำไมต้องมีระบบ Content Safety Filter

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

สำหรับทีมที่กำลังมองหา AI API ราคาประหยัด ที่รองรับภาษาไทยและมี built-in safety features อย่าง HolySheep AI มีอัตราแลกเปลี่ยนที่คุ้มค่ามากที่ ¥1=$1 ซึ่งประหยัดได้ถึง 85%+ เมื่อเทียบกับบริการอื่น รองรับการชำระเงินผ่าน WeChat และ Alipay พร้อม latency ที่ต่ำกว่า 50ms

สถาปัตยกรรมระบบกรองเนื้อหา

1. การติดตั้ง Client Library

# ติดตั้ง required packages
pip install openai-annotated-types httpx aiohttp regex

หรือใช้ poetry
poetry add openai annotated-types httpx

2. สร้าง Content Safety Service หลัก

import httpx
import re
from typing import Optional
from dataclasses import dataclass
from enum import Enum

class ContentCategory(Enum):
    PROFANITY = "profanity"
    SEXUAL = "sexual"
    VIOLENCE = "violence"
    PERSONAL_INFO = "personal_info"
    HATE_SPEECH = "hate_speech"

@dataclass
class SafetyResult:
    is_safe: bool
    categories: list[ContentCategory]
    confidence: float
    filtered_text: Optional[str] = None

class ContentSafetyFilter:
    """
    Production-ready content safety filter
    รองรับภาษาไทยและหลายภาษา
    """
    
    # คำหยาบภาษาไทยที่พบบ่อย (ตัวอย่าง subset)
    THAI_PROFANITY = [
        r'ควย', r'เหี้ย', r'สัด', r'มึง', r'กู', r'มันส์',
        r'เย็ด', r'ขี้', r'ห่า', r'ชาติ', r'ไอ้'
    ]
    
    # Pattern สำหรับข้อมูลส่วนบุคคล
    PERSONAL_INFO_PATTERNS = [
        (r'\b\d{13}\b', 'เลขบัตรประชาชน'),  # Thai ID
        (r'\b\d{10,11}\b', 'เบอร์โทรศัพท์'),
        (r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', 'อีเมล'),
    ]
    
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.client = httpx.AsyncClient(timeout=30.0)
        
        # Compile patterns สำหรับประสิทธิภาพ
        self._profanity_pattern = re.compile(
            '|'.join(self.THAI_PROFANITY), 
            re.IGNORECASE
        )
        
    async def check_with_ai(
        self, 
        text: str, 
        categories: list[str] = None
    ) -> SafetyResult:
        """
        ใช้ AI model ตรวจสอบเนื้อหา
        ราคา DeepSeek V3.2: $0.42/MTok - ประหยัดมากสำหรับ batch processing
        """
        if categories is None:
            categories = ["violence", "sexual", "hate_speech"]
            
        messages = [
            {
                "role": "system",
                "content": f"""คุณเป็นระบบตรวจสอบความปลอดภัยเนื้อหา
                ตรวจสอบข้อความและระบุหมวดหมู่ที่มีปัญหา: {', '.join(categories)}
                ตอบกลับเป็น JSON format: {{"is_safe": bool, "categories": [], "confidence": float}}"""
            },
            {
                "role": "user", 
                "content": text
            }
        ]
        
        response = await self.client.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "deepseek-v3.2",
                "messages": messages,
                "temperature": 0.1,  # ความแม่นยำสูง
                "max_tokens": 200
            }
        )
        
        result = response.json()
        content = result['choices'][0]['message']['content']
        
        # Parse JSON response
        import json
        try:
            parsed = json.loads(content)
            return SafetyResult(
                is_safe=parsed.get('is_safe', True),
                categories=[ContentCategory(c) for c in parsed.get('categories', [])],
                confidence=parsed.get('confidence', 0.0)
            )
        except json.JSONDecodeError:
            return SafetyResult(is_safe=True, categories=[], confidence=0.0)
    
    def check_profanity(self, text: str) -> tuple[bool, list[str]]:
        """ตรวจจับคำหยาบแบบ rule-based - เร็วมาก"""
        matches = self._profanity_pattern.findall(text)
        return len(matches) == 0, matches
    
    def check_personal_info(self, text: str) -> list[tuple[str, str]]:
        """ตรวจจับข้อมูลส่วนบุคคล"""
        findings = []
        for pattern, info_type in self.PERSONAL_INFO_PATTERNS:
            matches = re.findall(pattern, text)
            for match in matches:
                findings.append((match, info_type))
        return findings
    
    async def full_audit(self, text: str) -> SafetyResult:
        """
        ทำการตรวจสอบแบบครบวงจร
        ใช้ multi-layer approach เพื่อประสิทธิภาพ
        """
        # Layer 1: Fast rule-based check
        is_safe_profanity, profanity_matches = self.check_profanity(text)
        personal_info = self.check_personal_info(text)
        
        # ถ้าพบคำหยาบหรือข้อมูลส่วนบุคคล ให้ block ทันที
        if not is_safe_profanity or personal_info:
            return SafetyResult(
                is_safe=False,
                categories=[ContentCategory.PROFANITY] if not is_safe_profanity else [ContentCategory.PERSONAL_INFO],
                confidence=1.0,
                filtered_text=self._mask_profanity(text, profanity_matches)
            )
        
        # Layer 2: AI-powered deep check (optional - สำหรับ edge cases)
        ai_result = await self.check_with_ai(text)
        
        return ai_result
    
    def _mask_profanity(self, text: str, matches: list[str]) -> str:
        """แทนที่คำหยาบด้วย mask"""
        result = text
        for match in set(matches):
            result = re.sub(match, '*' * len(match), result, flags=re.IGNORECASE)
        return result
    
    async def close(self):
        await self.client.aclose()

การใช้งานใน Production

3. Middleware Integration สำหรับ FastAPI

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager

app = FastAPI()

Initialize filter
safety_filter = ContentSafetyFilter(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

@app.middleware("http")
async def content_safety_middleware(request: Request, call_next):
    """
    Middleware สำหรับตรวจสอบทุก request
    ประมวลผล async ไม่กระทบ latency
    """
    if request.method in ["POST", "PUT", "PATCH"]:
        body = await request.body()
        text = body.decode('utf-8')
        
        # Async check
        result = await safety_filter.full_audit(text)
        
        if not result.is_safe:
            return JSONResponse(
                status_code=400,
                content={
                    "error": "Content violation detected",
                    "categories": [c.value for c in result.categories],
                    "filtered_preview": result.filtered_text[:100] if result.filtered_text else None
                }
            )
    
    return await call_next(request)

@app.post("/generate")
async def generate_text(prompt: str):
    """
    Endpoint สำหรับสร้าง text พร้อม safety check
    """
    client = httpx.AsyncClient()
    
    try:
        response = await client.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
            json={
                "model": "gpt-4.1",  # $8/MTok - premium quality
                "messages": [{"role": "user", "content": prompt}],
                "max_tokens": 1000
            }
        )
        
        raw_output = response.json()['choices'][0]['message']['content']
        
        # Post-generation safety check
        safety_result = await safety_filter.full_audit(raw_output)
        
        if not safety_result.is_safe:
            return {"output": safety_result.filtered_text, "warning": "Content was filtered"}
        
        return {"output": raw_output}
        
    finally:
        await client.aclose()

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await safety_filter.close()

app.router.lifespan_context = lifespan

Benchmark และการเปรียบเทียบประสิทธิภาพ

จากการทดสอบของผม DeepSeek V3.2 เหมาะสำหรับ batch processing ที่ต้องการประหยัดต้นทุน ในขณะที่ Claude Sonnet 4.5 เหมาะกับงานที่ต้องการความแม่นยำสูงสุด

การ Optimizing สำหรับ Scale

import asyncio
from typing import List
from collections import defaultdict

class BatchSafetyChecker:
    """
    รองรับการตรวจสอบพร้อมกันหลายรายการ
    เหมาะสำหรับ batch processing
    """
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self._semaphore = asyncio.Semaphore(10)  # rate limit
    
    async def check_batch(
        self, 
        texts: List[str],
        batch_size: int = 20
    ) -> List[SafetyResult]:
        """
        ตรวจสอบหลายข้อความพร้อมกัน
        ใช้ batching เพื่อลด API calls
        """
        results = []
        
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            
            # Prepare batch request
            tasks = [
                self._check_single(text, idx)
                for idx, text in enumerate(batch)
            ]
            
            batch_results = await asyncio.gather(*tasks)
            results.extend(batch_results)
            
            # Rate limiting
            await asyncio.sleep(0.1)
        
        return results
    
    async def _check_single(self, text: str, idx: int) -> SafetyResult:
        async with self._semaphore:
            # Implement single check logic
            pass

Usage example
async def main():
    checker = BatchSafetyChecker("YOUR_HOLYSHEEP_API_KEY")
    
    texts = [
        "ข้อความที่ต้องการตรวจสอบที่ 1",
        "ข้อความที่ต้องการตรวจสอบที่ 2",
        # ... หลายพันรายการ
    ]
    
    results = await checker.check_batch(texts, batch_size=50)
    
    # Statistics
    unsafe_count = sum(1 for r in results if not r.is_safe)
    print(f"พบเนื้อหา
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
การรักษาความปลอดภัย Function Calling จากการโจมตีแบบ Maliciou
คู่มือฉบับสมบูรณ์: การ Implement AI API Key Rotation และ Sec
คู่มือย้ายระบบ AI เซิร์ฟเวอร์: จาก OpenAI API สู่ HolySheep