เมื่อพัฒนาแอปพลิเคชันที่ใช้ AI API ในระดับ Production ปัญหาหนึ่งที่นักพัฒาต้องเจอคือ Content Filter หรือการกรองเนื้อหาที่ไม่เหมาะสม บทความนี้จะพาคุณเข้าใจกลไกการทำงาน วิธีจัดการเมื่อเนื้อหาถูก Filter และเทคนิคการ Optimize สำหรับ Production Environment

Content Filter คืออะไรและทำงานอย่างไร

Content Filter เป็นระบบความปลอดภัยที่ทำงานขณะประมวลผล Prompt และ Response มันจะสแกนเนื้อหาทั้ง Input และ Output หากตรวจพบเนื้อหาที่ละเอียดอ่อน เช่น ความรุนแรง เนื้อหาทางเพศ หรือข้อมูลที่เป็นอันตราย ระบบจะส่งคืน Error Response แทนที่จะเป็นข้อมูลจริง

โครงสร้าง Response เมื่อถูก Filter

เมื่อเนื้อหาถูก Filter คุณจะได้รับ Response ที่มีโครงสร้างเฉพาะ ซึ่งแตกต่างจาก Error ปกติ

# Response ปกติ (สำเร็จ)
{
  "id": "chatcmpl-xxx",
  "object": "chat.completion",
  "created": 1735689600,
  "model": "gpt-4",
  "choices": [{
    "index": 0,
    "message": {
      "role": "assistant",
      "content": "ผลลัพธ์ปกติ"
    },
    "finish_reason": "stop"
  }],
  "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}
}


Response เมื่อถูก Filter

{
  "error": {
    "message": "Content filter triggered",
    "type": "content_filter",
    "code": "content_filter",
    "param": null,
    "error": {
      "code": "content_filter",
      "innererror": {
        "content_filter_reason": "violence",
        "filtered_reason": "violence",
        "prompt_filter_results": [...]
      }
    }
  }
}

การ Implement ระบบตรวจจับและจัดการ Content Filter

Architecture สำหรับ Production

การออกแบบระบบที่ดีต้องมี Layer การตรวจสอบหลายชั้น เริ่มจาก Pre-filter ฝั่ง Client ไปจนถึง Error Handling ฝั่ง Server

import openai
from typing import Optional, Dict, Any, Callable
from dataclasses import dataclass
from enum import Enum
import time
import logging


Configuration

client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",  # ใช้ HolySheep API เท่านั้น
    timeout=30.0,
    max_retries=3
)

class ContentFilterReason(Enum):
    VIOLENCE = "violence"
    SEXUAL = "sexual"
    HATE = "hate"
    SELF_HARM = "self-harm"
    ILLEGAL = "illegal"
    UNKNOWN = "unknown"

@dataclass
class FilterResult:
    is_filtered: bool
    reason: Optional[ContentFilterReason]
    error_message: str
    retry_allowed: bool
    fallback_message: str

class ContentFilterHandler:
    """
    Handler สำหรับจัดการ Content Filter ในระดับ Production
    รองรับ: Detection, Retry Logic, Fallback Strategy, Logging
    """
    
    def __init__(self, logger: Optional[logging.Logger] = None):
        self.logger = logger or logging.getLogger(__name__)
        self.fallback_responses = {
            ContentFilterReason.VIOLENCE: "ขออภัย ฉันไม่สามารถตอบคำถามเกี่ยวกับความรุนแรงได้",
            ContentFilterReason.SEXUAL: "ขออภัย เนื้อหานี้ไม่เหมาะสมสำหรับการสนทนา",
            ContentFilterReason.HATE: "ขออภัย ฉันไม่สนับสนุนเนื้อหาที่เกลียดชัง",
            ContentFilterReason.SELF_HARM: "หากคุณกำลังมีความคิดที่จะทำร้ายตัวเอง กรุณาติดต่อสายด่วนสุขภาพจิต",
            ContentFilterReason.ILLEGAL: "ขออภัย ฉันไม่สามารถช่วยในเรื่องที่ผิดกฎหมายได้",
            ContentFilterReason.UNKNOWN: "ขออภัย เกิดข้อผิดพลาดในการประมวลผล กรุณาลองใหม่อีกครั้ง"
        }
    
    def parse_filter_error(self, error: Exception) -> FilterResult:
        """แยกวิเคราะห์ Error Response จาก Content Filter"""
        error_dict = vars(error)
        error_msg = str(error)
        
        # ตรวจหา Content Filter Error
        if "content_filter" in error_msg.lower() or "content filter" in error_msg.lower():
            # ดึง Filter Reason จาก Response
            reason = self._extract_filter_reason(error_dict)
            return FilterResult(
                is_filtered=True,
                reason=reason,
                error_message=error_msg,
                retry_allowed=False,  # Content Filter ไม่ควร Retry
                fallback_message=self.fallback_responses.get(
                    reason, 
                    self.fallback_responses[ContentFilterReason.UNKNOWN]
                )
            )
        
        # Network Error หรือ Server Error - สามารถ Retry ได้
        return FilterResult(
            is_filtered=False,
            reason=None,
            error_message=error_msg,
            retry_allowed=True,
            fallback_message="เกิดข้อผิดพลาดในการเชื่อมต่อ กรุณาลองใหม่อีกครั้ง"
        )
    
    def _extract_filter_reason(self, error_dict: Dict[str, Any]) -> ContentFilterReason:
        """ดึง Filter Reason จาก Error Structure"""
        try:
            # ลองหาใน nested error structure
            if 'error' in error_dict:
                inner = error_dict['error']
                if isinstance(inner, dict):
                    if 'innererror' in inner:
                        reason_str = inner['innererror'].get('content_filter_reason', '')
                    else:
                        reason_str = inner.get('content_filter_reason', '')
                    
                    for reason in ContentFilterReason:
                        if reason.value in reason_str.lower():
                            return reason
        except Exception:
            pass
        
        return ContentFilterReason.UNKNOWN

async def chat_with_filter_handling(
    prompt: str,
    model: str = "gpt-4o",
    system_prompt: Optional[str] = None,
    max_retries: int = 3
) -> Dict[str, Any]:
    """
    Function หลักสำหรับ Chat พร้อมระบบจัดการ Content Filter
    
    Features:
    - Automatic Filter Detection
    - Retry with Exponential Backoff (สำหรับ Network Error เท่านั้น)
    - Graceful Fallback
    - Comprehensive Logging
    """
    handler = ContentFilterHandler()
    messages = []
    
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": prompt})
    
    last_error = None
    
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                temperature=0.7,
                max_tokens=1000
            )
            
            # สำเร็จ - คืนค่า Response
            return {
                "success": True,
                "content": response.choices[0].message.content,
                "usage": response.usage.model_dump() if response.usage else None,
                "filter_triggered": False
            }
            
        except openai.APIError as e:
            last_error = e
            filter_result = handler.parse_filter_error(e)
            
            if filter_result.is_filtered:
                # Content Filter - ไม่ Retry แต่ Fallback
                return {
                    "success": False,
                    "content": filter_result.fallback_message,
                    "error_type": "content_filter",
                    "filter_reason": filter_result.reason.value if filter_result.reason else None,
                    "filter_triggered": True
                }
            
            # Network Error - Retry ด้วย Exponential Backoff
            if filter_result.retry_allowed and attempt < max_retries - 1:
                wait_time = 2 ** attempt  # 1s, 2s, 4s
                logging.warning(f"Retry {attempt + 1}/{max_retries} after {wait_time}s")
                time.sleep(wait_time)
                continue
            
            # Error อื่นๆ
            return {
                "success": False,
                "content": filter_result.fallback_message,
                "error_type": "api_error",
                "filter_triggered": False
            }
    
    # หลังจาก Retry หมดแล้วยังไม่สำเร็จ
    return {
        "success": False,
        "content": "เกิดข้อผิดพลาดหลังจากพยายามหลายครั้ง กรุณาลองใหม่ภายหลัง",
        "error_type": "max_retries_exceeded",
        "filter_triggered": False
    }

การ Implement Retry Strategy ที่ชาญฉลาด

สำหรับ Production คุณต้องแยกแยะระหว่าง Error ที่สามารถ Retry ได้ (Network Timeout, Rate Limit) กับ Content Filter ที่ไม่ควร Retry เพราะจะได้ผลลัพธ์เดิมเสมอ

import asyncio
from typing import TypeVar, Callable, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import hashlib

T = TypeVar('T')

@dataclass
class RetryConfig:
    max_retries: int = 3
    base_delay: float = 1.0
    max_delay: float = 60.0
    exponential_base: float = 2.0
    jitter: bool = True
    
@dataclass 
class RequestMetrics:
    request_id: str
    timestamp: datetime
    model: str
    prompt

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