ในโลกของ Healthcare AI ยุคใหม่ การพัฒนาระบบถามตอบทางการแพทย์ที่เชื่อมต่อกับ LLM API นั้นมีความซับซ้อนและท้าทายเป็นอย่างมาก โดยเฉพาะด้านความปลอดภัยที่ต้องคำนึงถึงข้อมูลผู้ป่วยที่มีความละเอียดอ่อน

สถานการณ์ข้อผิดพลาดจริง: ConnectionError: timeout ที่โรงพยาบาล

เมื่อเดือนที่แล้ว ทีมพัฒนาของเราได้รับเคสเร่งด่วนจากโรงพยาบาลแห่งหนึ่งที่ระบบถามตอบ AI ล่มในช่วงเวลาเร่งด่วน ข้อผิดพลาดที่พบคือ:

ConnectionError: timeout exceeded while connecting to api.holysheep.ai/v1/chat/completions
Request ID: req_3847xh9d
Retry attempt 3/5 failed - aborting
Patient data partially exposed in error log: {"name": "สมชาย", "symptoms": "เจ็บหน้าอก"}

ปัญหานี้เกิดจากการไม่มี timeout handling ที่ดี และข้อมูลผู้ป่วยรั่วไหลใน log file ซึ่งเป็นการละเมิด PDPA อย่างร้ายแรง บทความนี้จะสอนวิธีแก้ไขอย่างครอบคลุม

หลักการพื้นฐานความปลอดภัย API สำหรับ Medical AI

ระบบถามตอบทางการแพทย์ต้องปฏิบัติตามมาตรฐาน HIPAA และ PDPA ของไทย โดยมีองค์ประกอบสำคัญ:

การตั้งค่า Secure API Client สำหรับ Medical Consultation

การใช้ สมัครที่นี่ HolySheep AI ช่วยให้คุณได้รับ API ที่มีความปลอดภัยสูง พร้อม latency ต่ำกว่า 50ms ราคาประหยัดกว่า 85% เมื่อเทียบกับ OpenAI โดยเริ่มต้นด้วยการตั้งค่า client ที่ปลอดภัย:

import httpx
import asyncio
import json
import hashlib
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass

@dataclass
class MedicalAPIConfig:
    """การตั้งค่าความปลอดภัยสำหรับ Medical AI API"""
    api_key: str
    base_url: str = "https://api.holysheep.ai/v1"
    timeout: float = 30.0
    max_retries: int = 3
    rate_limit: int = 100  # requests per minute

class SecureMedicalAPIClient:
    """
    Secure API Client สำหรับระบบถามตอบทางการแพทย์
    - รองรับ Retry with exponential backoff
    - ป้องกันข้อมูลรั่วไหลใน logs
    - Rate limiting ในตัว
    """
    
    def __init__(self, config: MedicalAPIConfig):
        self.config = config
        self._request_count = 0
        self._window_start = time.time()
        self._lock = asyncio.Lock()
        
        # สร้าง HTTP client ที่ปลอดภัย
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(
                connect=10.0,
                read=config.timeout,
                write=10.0,
                pool=5.0
            ),
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=5),
            headers={
                "Authorization": f"Bearer {config.api_key}",
                "Content-Type": "application/json",
                "X-Request-Timestamp": str(int(time.time()))
            }
        )
    
    async def _check_rate_limit(self):
        """ตรวจสอบ Rate Limit อย่างปลอดภัย"""
        async with self._lock:
            current_time = time.time()
            if current_time - self._window_start >= 60:
                self._request_count = 0
                self._window_start = current_time
            
            if self._request_count >= self.config.rate_limit:
                wait_time = 60 - (current_time - self._window_start)
                raise RateLimitError(f"Rate limit exceeded. Wait {wait_time:.1f}s")
            
            self._request_count += 1
    
    async def _make_request(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """ส่ง request พร้อม error handling ที่ปลอดภัย"""
        try:
            response = await self.client.post(
                f"{self.config.base_url}/chat/completions",
                json=payload
            )
            response.raise_for_status()
            return response.json()
        except httpx.TimeoutException as e:
            # ไม่เปิดเผยข้อมูลภายในใน error message
            raise APIError("Request timeout. Please try again.") from e
        except httpx.HTTPStatusError as e:
            raise APIError(f"API error: {e.response.status_code}") from e

    async def close(self):
        await self.client.aclose()

class APIError(Exception):
    """Custom Exception ที่ไม่เปิดเผยข้อมูลภายใน"""
    pass

class RateLimitError(Exception):
    pass


การใช้งาน

config = MedicalAPIConfig(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    timeout=30.0,
    max_retries=3
)
client = SecureMedicalAPIClient(config)

การ Mask ข้อมูลผู้ป่วยก่อนส่งไปยัง AI

ข้อสำคัญที่สุดในการออกแบบ Medical AI คือการปกป้องข้อมูลส่วนบุคคลของผู้ป่วย เราต้องทำ PII (Personally Identifiable Information) Masking ก่อนส่งข้อมูลไปยัง API:

import re
from typing import Dict, Any, List, Tuple
import logging

class PatientDataMasker:
    """
    ระบบ Mask ข้อมูลส่วนตัวผู้ป่วยตามมาตรฐาน PDPA
    - ชื่อ-นามสกุล -> [PATIENT_NAME]
    - เบอร์โทรศัพท์ -> [PHONE]
    - หมายเลขบัตรประจำตัวประชาชน -> [NATIONAL_ID]
    - ที่อยู่ -> [ADDRESS]
    - วันเกิด -> [DOB]
    """
    
    PATTERNS: List[Tuple[str, str]] = [
        # ชื่อไทย 2-3 คำ (สันนิษฐานว่าเป็นชื่อ-นามสกุล)
        (r'[ก-๙]{2,}\s+[ก-๙]{2,}(?:\s+[ก-๙]{2,})?', '[PATIENT_NAME]'),
        # เบอร์โทรไทย
        (r'0[6-8]\d-\d{3}-\d{4}', '[PHONE]'),
        (r'0[6-8]\d{9}', '[PHONE]'),
        # หมายเลขบัตรประจำตัวประชาชน 13 หลัก
        (r'\d{1}\s?\d{4}\s?\d{5}\s?\d{2}\s?\d', '[NATIONAL_ID]'),
        # Email
        (r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', '[EMAIL]'),
        # ที่อยู่ (คำขึ้นต้นที่พบบ่อย)
        (r'(บ้านเลขที่|ที่อยู่|ตึก|อาคาร)\s+[^\n]{10,}', '[ADDRESS]'),
    ]
    
    @classmethod
    def mask_prompt(cls, text: str) -> str:
        """Mask ข้อมูลใน prompt อย่างปลอดภัย"""
        masked = text
        for pattern, replacement in cls.PATTERNS:
            masked = re.sub(pattern, replacement, masked, flags=re.IGNORECASE)
        return masked
    
    @classmethod
    def mask_response(cls, text: str) -> str:
        """Mask ข้อมูลใน response จาก AI ด้วย"""
        return cls.mask_prompt(text)

class MedicalPromptBuilder:
    """สร้าง prompt ที่ปลอดภัยสำหรับ Medical AI"""
    
    SYSTEM_PROMPT = """คุณเป็นผู้ช่วยแพทย์ AI ที่ได้รับการรับรอง
- ตอบคำถามด้านสุขภาพเบื้องต้นเท่านั้น
- ไม่สามารถวินิจฉัยโรคแทนแพทย์ได้
- แนะนำให้ผู้ป่วยไปพบแพทย์เมื่ออาการรุนแรง
- ข้อมูลที่ได้รับเป็นข้อมูลที่ถูก mask แล้ว"""
    
    @staticmethod
    def build_consultation_prompt(
        patient_query: str,
        conversation_history: List[Dict] = None
    ) -> Dict[str, Any]:
        """
        สร้าง prompt สำหรับการปรึกษาทางการแพทย์
        - Mask ข้อมูลผู้ป่วยอัตโนมัติ
        - รวม history อย่างปลอดภัย
        """
        # Mask ข้อมูลก่อนส่ง
        masked_query = PatientDataMasker.mask_prompt(patient_query)
        
        messages = [
            {"role": "system", "content": MedicalPromptBuilder.SYSTEM_PROMPT}
        ]
        
        # เพิ่ม history ที่ mask แล้ว
        if conversation_history:
            for msg in conversation_history[-5:]:  # เก็บแค่ 5 ข้อความล่าสุด
                masked_content = PatientDataMasker.mask_prompt(msg.get("content", ""))
                messages.append({
                    "role": msg.get("role", "user"),
                    "content": masked_content
                })
        
        messages.append({"role": "user", "content": masked_query})
        
        return {
            "model": "gpt-4.1",
            "messages": messages,
            "temperature": 0.3,  # ความแม่นยำสูง ลด hallucination
            "max_tokens": 1000,
            "metadata": {
                "query_type": "medical_consultation",
                "pii_masked": True
            }
        }


การใช้งาน

query = "สวัสดีครับ ผมชื่อนายสมชาย มีอาการเจ็บหน้าอกด้านซ้ายมา 3 วัน โทร 0891234567"
masked = PatientDataMasker.mask_prompt(query)
print(masked)

Output: สวัสดีครับ ผมชื่อ [PATIENT_NAME] มีอาการเจ็บหน้าอกด้านซ้ายมา 3 วัน โทร [PHONE]

การ Implement Retry Logic ที่ปลอดภัย

ในระบบ Medical AI การ retry request ต้องทำอย่างระมัดระวัง เพื่อไม่ให้เกิด duplicate medical advice หรือ data inconsistency:

import asyncio
import logging
from typing import TypeVar, Callable, Any
from functools import wraps

T = TypeVar('T')

class SafeRetryHandler:
    """
    Retry Handler สำหรับ Medical API
    - Exponential backoff
    - Jitter เพื่อป้องกัน Thundering Herd
    - Maximum retry limit
    - Logging ที่ไม่เปิดเผยข้อมูล
    """
    
    def __init__(
        self,
        max_retries: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 30.0,
        jitter: float = 0.5
    ):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.jitter = jitter
    
    async def retry_with_backoff(
        self,
        func: Callable[..., Any],
        *args,
        operation_name: str = "API call",
        **kwargs
    ) -> Any:
        """Execute function with exponential backoff retry"""
        last_exception = None
        
        for attempt in range(self.max_retries + 1):
            try:
                result = await func(*args, **kwargs)
                
                if attempt > 0:
                    logging.info(f"{operation_name} succeeded on attempt {attempt + 1}")
                
                return result
                
            except RateLimitError:
                # Rate limit ให้รอนานกว่า retry ปกติ
                delay = min(self.max_delay, self.base_delay * (2 ** attempt))
                delay += asyncio.uniform(-self.jitter, self.jitter) * delay
                
                logging.warning(f"Rate limited. Retrying in {delay:.2f}s")
                await asyncio.sleep(delay)
                last_exception = None
                
            except httpx.TimeoutException as e:
                delay = min(self.max_delay, self.base_delay * (2 ** attempt))
                delay += asyncio.uniform(-self.jitter, self.jitter) * delay
                
                logging.warning(
                    f"Timeout on attempt {attempt + 1}/{self.max_retries + 1}"
                )
                await asyncio.sleep(delay)
                last_exception = e
                
            except httpx.HTTPStatusError as e:
                # 5xx errors เท่านั้นที่ควร retry
                if 500 <= e.response.status_code < 600:
                    delay = min(self.max_delay, self.base_delay * (2 ** attempt))
                    await asyncio.sleep(delay)
                    last_exception = e
                else:
                    # 4xx errors ไม่ควร retry
                    raise APIError(f"Request failed: {e.response.status_code}") from e
        
        # ถ้า retry หมดแล้วยังล้มเหลว
        raise APIError(
            f"{operation_name} failed after {self.max_retries + 1} attempts"
        ) from last_exception


การใช้งาน

async def medical_consultation(query: str):
    retry_handler = SafeRetryHandler(max_retries=3, base_delay=2.0

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