วิธีการตรวจสอบ Prompt ก่อนส่งไปยัง AI API — คู่มือฉบับสมบูรณ์

สวัสดีครับ ผมเป็นวิศวกรที่ใช้งาน AI API มาหลายปี วันนี้จะมาแชร์ประสบการณ์ตรงในการตรวจสอบ Prompt ก่อนส่งไปยัง AI API ว่าทำไมต้องทำ ทำอย่างไร และมีข้อผิดพลาดอะไรบ้างที่พบบ่อย

ทำไมต้อง Validate Prompt ก่อนส่ง?

จากประสบการณ์ที่ผมเคยเจอปัญหา API cost พุ่งสูงผิดปกติจาก prompt ที่มีข้อผิดพลาด การตรวจสอบ prompt ก่อนส่งช่วยประหยัดค่าใช้จ่ายได้มากถึง 30-40% และยังช่วยลดการเรียก API ที่ไม่จำเป็น ลดความล่าช้า และป้องกันข้อผิดพลาดที่ไม่คาดคิด

เปรียบเทียบบริการ AI API

HolySheep AI มีอัตราแลกเปลี่ยนที่คุ้มค่ามาก ¥1=$1 ช่วยประหยัดได้ถึง 85%+ จากราคา Official พร้อมความล่าช้าน้อยกว่า 50ms สมัครที่นี่

โครงสร้างพื้นฐานของ Prompt Validation

การตรวจสอบ prompt ที่ดีควรครอบคลุมหลายระดับ ตั้งแต่การตรวจสอบความยาว ความปลอดภัย ไปจนถึงการจำกัด token และการจัดรูปแบบ

การติดตั้งและตั้งค่า Client

# ติดตั้งไลบรารีที่จำเป็น
pip install httpx tiktoken jsonschema

ไลบรารีสำหรับนับ token
tiktoken ใช้กับ model GPT series
สำหรับ Claude ใช้ anthropic SDK

import httpx
import tiktoken
from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field
from enum import Enum

class ValidationError(Exception):
    """Exception สำหรับข้อผิดพลาดการตรวจสอบ"""
    pass

class PromptValidator:
    """
    คลาสตรวจสอบ Prompt ก่อนส่งไปยัง AI API
    รองรับทั้ง Chat Completions และ Claude
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        max_tokens: int = 4096,
        max_prompt_length: int = 100000
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.max_tokens = max_tokens
        self.max_prompt_length = max_prompt_length
        
        # โมเดลสำหรับนับ token
        self.encoders = {
            "gpt-4": tiktoken.get_encoding("cl100k_base"),
            "gpt-3.5-turbo": tiktoken.get_encoding("cl100k_base"),
            "default": tiktoken.get_encoding("cl100k_base")
        }
    
    def count_tokens(self, text: str, model: str = "gpt-4") -> int:
        """นับจำนวน token ในข้อความ"""
        encoder = self.encoders.get(model, self.encoders["default"])
        return len(encoder.encode(text))
    
    def validate_prompt(self, prompt: str, model: str = "gpt-4") -> Dict[str, Any]:
        """
        ตรวจสอบ prompt ทั้งหมด
        คืนค่า dict ที่มีผลการตรวจสอบและข้อมูลเพิ่มเติม
        """
        result = {
            "valid": True,
            "errors": [],
            "warnings": [],
            "token_count": 0,
            "char_count": len(prompt)
        }
        
        # ตรวจสอบความว่างเปล่า
        if not prompt or not prompt.strip():
            result["valid"] = False
            result["errors"].append("Prompt ห้ามว่างเปล่า")
        
        # ตรวจสอบความยาวตัวอักษร
        if len(prompt) > self.max_prompt_length:
            result["valid"] = False
            result["errors"].append(
                f"Prompt ยาวเกินกำหนด: {len(prompt)} ตัวอักษร "
                f"(สูงสุด: {self.max_prompt_length})"
            )
        
        # นับ token และตรวจสอบ
        token_count = self.count_tokens(prompt, model)
        result["token_count"] = token_count
        
        if token_count > self.max_tokens:
            result["valid"] = False
            result["errors"].append(
                f"Token จำนวนเกินกำหนด: {token_count} tokens "
                f"(สูงสุด: {self.max_tokens})"
            )
        elif token_count > self.max_tokens * 0.8:
            result["warnings"].append(
                f"Prompt ใช้ token ไป {token_count} ({token_count/self.max_tokens*100:.1f}%) "
                f"ใกล้ถึงขีดจำกัด"
            )
        
        # ตรวจสอบ prompt injection
        if self._detect_prompt_injection(prompt):
            result["valid"] = False
            result["errors"].append("ตรวจพบ prompt injection ที่อาจเป็นอันตราย")
        
        return result
    
    def _detect_prompt_injection(self, prompt: str) -> bool:
        """ตรวจจับ prompt injection patterns"""
        dangerous_patterns = [
            "ignore previous instructions",
            "disregard your instructions",
            " ignore your guidelines",
            "new instructions:",
            "/ignore",
            "override system",
        ]
        prompt_lower = prompt.lower()
        return any(pattern in prompt_lower for pattern in dangerous_patterns)

ตัวอย่างการใช้งาน
validator = PromptValidator(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    max_tokens=8192
)

test_prompt = "อธิบายเรื่อง AI สั้นๆ"
result = validator.validate_prompt(test_prompt, model="gpt-4")
print(f"Token count: {result['token_count']}")
print(f"Valid: {result['valid']}")

ระบบ Rate Limiting และ Retry Logic

import time
import asyncio
from typing import Callable, Any, Optional
from dataclasses import dataclass
from collections import defaultdict
import httpx

@dataclass
class RateLimitConfig:
    """การตั้งค่า Rate Limiting"""
    max_requests_per_minute: int = 60
    max_tokens_per_minute: int = 100000
    backoff_base: float = 1.0
    max_retries: int = 3

class AIClientWithValidation:
    """
    AI Client ที่มีการตรวจสอบ Prompt และจัดการ Rate Limit
    ใช้ base_url ของ HolySheep AI
    """
    
    def __init__(
        self,
        api_key: str,
        rate_config: Optional[RateLimitConfig] = None
    ):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.rate_config = rate_config or RateLimitConfig()
        
        # ตัวนับ requests
        self.request_timestamps = []
        self.token_usage = []
        
        # HTTP client
        self.client = httpx.AsyncClient(timeout=60.0)
        
        # ตัวตรวจสอบ
        self.validator = PromptValidator(api_key, max_tokens=8192)
    
    async def send_request(
        self,
        messages: List[Dict[str, str]],
        model: str = "gpt-4",
        temperature: float = 0.7,
        max_response_tokens: int = 1024
    ) -> Dict[str, Any]:
        """
        ส่ง request ไปยัง API พร้อมตรวจสอบและ validation
        """
        # รวม prompt จาก messages
        full_prompt = "\n".join([m.get("content", "") for m in messages])
        
        # ตรวจสอบ prompt
        validation = self.validator.validate_prompt(full_prompt, model)
        if not validation["valid"]:
            raise ValidationError(
                f"Prompt ไม่ผ่านการตรวจสอบ: {validation['errors']}"
            )
        
        # ตรวจสอบ Rate Limit
        await self._check_rate_limit(validation["token_count"])
        
        # คำนวณ total tokens
        total_tokens = validation["token_count"] + max_response_tokens
        if total_tokens > self.rate_config.max_tokens_per_minute:
            raise ValidationError(
                f"จำนวน tokens ({total_tokens}) เกินขีดจำกัด "
                f"({self.rate_config.max_tokens_per_minute}/นาที)"
            )
        
        # ส่ง request
        for retry in range(self.rate_config.max_retries):
            try:
                response = await self._make_request(
                    messages=messages,
                    model=model,
                    temperature=temperature,
                    max_tokens=max_response_tokens
                )
                return response
                
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 429:
                    # Rate limited - รอแล้ว retry
                    wait_time = self.rate_config.backoff_base * (2 ** retry)
                    await asyncio.sleep(wait_time)
                    continue
                raise
            except httpx.TimeoutException:
                if retry < self.rate_config.max_retries - 1:
                    await asyncio.sleep(1)
                    continue
                raise
    
    async def _check_rate_limit(self, token_count: int):
        """ตรวจสอบและบังคับ rate limit"""
        now = time.time()
        current_minute = int(now / 60)
        
        # ลบ timestamp ที่เก่ากว่า 1 นาที
        self.request_timestamps = [
            (ts, tokens) for ts, tokens in self.request_timestamps
            if int(ts / 60) >= current_minute - 1
        ]
        
        # นับ requests ในนาทีปัจจุบัน
        current_requests = [
            ts for ts in self.request_timestamps
            if int(ts / 60) == current_minute
        ]
        
        if len(current_requests) >= self.rate_config.max_requests_per_minute:
            oldest = min(current_requests)
            wait_time = 60 - (now - oldest)
            if wait_time > 0:
                await asyncio.sleep(wait_time)
        
        # บันทึก usage
        self.request_timestamps.append((now, token_count))
    
    async def _make_request(
        self,
        messages: List[Dict[str, str]],
        model: str,
        temperature: float,
        max_tokens: int
    ) -> Dict[str, Any]:
        """ส่ง request ไปยัง HolySheep API"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        response = await self.client.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload
        )
        response.raise_for_status()
        return response.json()
    
    async def close(self):
        """ปิด connection"""
        await self.client.aclose()

ตัวอย่างการใช้งาน
async def main():
    client = AIClientWithValidation(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        rate_config=RateLimitConfig(
            max_requests_per_minute=60,
            max_tokens_per_minute=150000
        )
    )
    
    messages = [
        {"role": "user", "content": "ทบทวนโค้ด Python นี้ให้หน่อย: def hello(): pass"}
    ]
    
    try:
        result = await client.send_request(
            messages=messages,
            model="gpt-4",
            temperature=0.3
        )
        print(f"Response: {result['choices'][0]['message']['content']}")
    except ValidationError as e:
        print(f"Validation failed: {e}")
    finally:
        await client.close()

รัน asyncio
asyncio.run(main())

ระบบ Prompt Caching และ Deduplication

import hashlib
import json
from typing import Dict, Optional, Any, List
from datetime import datetime, timedelta

class PromptCache:
    """
    ระบบ Cache สำหรับเก็บ prompt ที่ซ้ำกัน
    ช่วยลดค่าใช้จ่ายและเพิ่มความเร็ว
    """
    
    def __init__(self, ttl_minutes: int = 60, max_size: int = 10000):
        self.ttl = timedelta(minutes=ttl_minutes)
        self.max_size = max_size
        self._cache: Dict[str, Dict[str, Any]] = {}
        self._hit_count = 0
        self._miss_count = 0
    
    def _generate_key(self, prompt: str, model: str) -> str:
        """สร้าง cache key จาก prompt และ model"""
        content = json.dumps({
            "prompt": prompt,
            "model": model
        }, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()
    
    def get(self, prompt: str, model: str) -> Optional[Dict[str, Any]]:
        """ดึงข้อมูลจาก cache"""
        key = self._generate_key(prompt, model)
        
        if key in self._cache:
            entry = self._cache[key]
            # ตรวจสอบ TTL
            if datetime.now() - entry["cached_at"] < self.ttl:
                self._hit_count += 1
                entry["hits"] += 1
                return entry["response"]
            else:
                # ลบ entry ที่หมดอายุ
                del self._cache[key]
        
        self._miss_count += 1
        return None
    
    def set(self, prompt: str, model: str, response: Dict[str, Any]):
        """เก็บข้อมูลลง cache"""
        key = self._generate_key(prompt, model)
        
        # ลบ entry เก่าถ้า cache เต็ม
        if len(self._cache) >= self.max_size:
            self._evict_oldest()
        
        self._cache[key] = {
            "response": response,
            "cached_at": datetime.now(),
            "hits": 0
        }
    
    def _evict_oldest(self):
        """ลบ entry เก่าที่สุด"""
        if not self._cache:
            return
        
        oldest_key = min(
            self._cache.keys(),
            key=lambda k: self._cache[k]["cached_at"]
        )
        del self._cache[oldest_key]
    
    def get_stats(self) ->
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
AI Semantic Search ใช้งานอย่างไร: คู่มือ Fine-tune Embedding
How to Monitor Multi-Model API Health ด้วย Prometheus Metric
การใช้ Parallel Function Calling เพิ่มประสิทธิภาพ AI API สู่