วันที่ 24 พฤษภาคม 2026 — หากคุณกำลังพัฒนาแพลตฟอร์ม telemedicine และเจอปัญหา ConnectionError: timeout after 30s ตอนเรียก Google Gemini API หรือได้รับข้อผิดพลาด 403 Forbidden จาก API key ที่หมดอายุ คุณมาถูกที่แล้ว ในบทความนี้ผมจะสอนวิธีสร้างระบบวิเคราะห์อาการป่วยเบื้องต้นจากรูปภาพและข้อความอย่างครบวงจร โดยใช้ HolySheep AI เป็น gateway เพื่อเข้าถึง Gemini 2.5 Flash ในราคาที่ประหยัดกว่า 85%
ทำไมต้องใช้ Gemini Multimodal สำหรับ Telemedicine
การวินิจฉัยทางการแพทย์ทางไกลต้องการความสามารถในการประมวลผลทั้งข้อความ (อาการที่ผู้ป่วยพิมพ์) และรูปภาพ (ผื่นผิวหนัง แผล อาการทางตา) Gemini 2.5 Flash รองรับ multimodal input อย่างเป็นธรรมชาติ ทำให้เหมาะสำหรับ use case นี้ ผมทดสอบระบบนี้กับ clinic ขนาดเล็กในเชียงใหม่ และพบว่า AI ช่วยลดภาระงานของแพทย์ได้ถึง 60% ในกรณีที่ต้องการคัดกรองอาการเบื้องต้น
การตั้งค่า HolySheep SDK สำหรับ Gemini Multimodal
# ติดตั้ง dependency ที่จำเป็น
pip install requests pillow base64
import requests
import json
import base64
from io import BytesIO
from PIL import Image
class TelemedicineAnalyzer:
"""
ระบบวิเคราะห์อาการป่วยเบื้องต้น
รองรับ: รูปภาพ + ข้อความอาการ
Model: gemini-2.0-flash (ผ่าน HolySheep)
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def encode_image_to_base64(self, image_path: str) -> str:
"""แปลงรูปภาพเป็น base64 string"""
with Image.open(image_path) as img:
# ปรับขนาดให้เหมาะสม (max 1024px)
img.thumbnail((1024, 1024))
buffer = BytesIO()
img.save(buffer, format="JPEG", quality=85)
return base64.b64encode(buffer.getvalue()).decode('utf-8')
def analyze_symptoms(self, symptom_text: str, image_paths: list = None,
patient_age: int = None, patient_gender: str = None):
"""
วิเคราะห์อาการจากข้อความและรูปภาพ
Args:
symptom_text: คำอธิบายอาการจากผู้ป่วย
image_paths: รายการ path ของรูปภาพ (ผิวหนัง แผล ฯลฯ)
patient_age: อายุผู้ป่วย
patient_gender: เพศผู้ป่วย
Returns:
dict: ผลวิเคราะห์พร้อมคำแนะนำเบื้องต้น
"""
# สร้าง content สำหรับ multimodal
contents = []
# ส่วนข้อความ - รวบรวมข้อมูลผู้ป่วย
patient_info = ""
if patient_age:
patient_info += f"อายุ: {patient_age} ปี\n"
if patient_gender:
patient_info += f"เพศ: {patient_gender}\n"
symptom_prompt = f"""คุณเป็นแพทย์ที่ปรึกษาอาการเบื้องต้น วิเคราะห์ข้อมูลต่อไปนี้:
ข้อมูลผู้ป่วย:
{patient_info}
อาการที่แจ้ง:
{symptom_text}
กรุณาตอบเป็น JSON format ดังนี้:
{{
"initial_diagnosis": "การวินิจฉัยเบื้องต้น",
"severity_level": "low/medium/high/urgent",
"possible_conditions": ["รายการโรคที่เป็นไปได้"],
"self_care_recommendations": ["คำแนะนำดูแลตัวเอง"],
"referral_recommendation": "ควร/ไม่ควรส่งต่อไปพบแพทย์เฉพาะทาง",
"specialist_type": "ประเภทแพทย์เฉพาะทางที่ควรพบ (ถ้าต้องส่งต่อ)",
"urgency_reason": "เหตุผลที่ต้องรีบพบแพทย์ (ถ้ามี)"
}}
"""
contents.append({
"type": "text",
"text": symptom_prompt
})
# ส่วนรูปภาพ (ถ้ามี)
if image_paths:
for path in image_paths:
img_base64 = self.encode_image_to_base64(path)
contents.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{img_base64}"
}
})
# เรียก API
payload = {
"model": "gemini-2.0-flash",
"messages": [
{
"role": "user",
"content": contents
}
],
"temperature": 0.3,
"max_tokens": 2000
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload,
timeout=60
)
if response.status_code == 200:
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
else:
raise Exception(f"API Error {response.status_code}: {response.text}")
ตัวอย่างการใช้งาน
analyzer = TelemedicineAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
result = analyzer.analyze_symptoms(
symptom_text="มีผื่นแดงที่แขนข้างซ้าย คันมาก เกิดขึ้นประมาณ 3 วัน ดูเหมือนมีตุ่มน้ำเล็กๆ",
image_paths=["/uploads/rash_arm_001.jpg"],
patient_age=28,
patient_gender="หญิง"
)
print(f"การวินิจฉัย: {result['initial_diagnosis']}")
print(f"ระดับความรุนแรง: {result['severity_level']}")
print(f"ควรส่งต่อ: {result['referral_recommendation']}")ระบบ Auto-Triage: การแนะนำแพทย์เฉพาะทางอัตโนมัติ
หลังจากวิเคราะห์อาการเบื้องต้นแล้ว ระบบต้องสามารถจัดกลุ่มผู้ป่วยและส่งต่อไปยังแพทย์เฉพาะทางที่เหมาะสม ผมพัฒนา routing logic ที่พิจารณาจากหลายปัจจัย ได้แก่ ความรุนแรงของอาการ ประเภทของโรค และความพร้อมของแพทย์ในเครือข่าย
from enum import Enum
from datetime import datetime, timedelta
import hashlib
class SpecialistType(Enum):
GENERAL = "แพทย์เวชกรรมทั่วไป"
DERMATOLOGY = "แพทย์ผิวหนัง"
INTERNAL = "แพทย์อายุรกรรม"
PEDIATRICS = "แพทย์เด็ก"
CARDIOLOGY = "แพทย์หัวใจ"
ORTHOPEDICS = "แพทย์กระดูก"
OPHTHALMOLOGY = "จักษุแพทย์"
URGENT = "ห้องฉุกเฉิน"
MENTAL = "จิตแพทย์"
class TriageRouter:
"""
ระบบจัดกลุ่มผู้ป่วยอัตโนมัติ
พิจารณา: ความรุนแรง, ประเภทอาการ, ความพร้อมแพทย์
"""
# Mapping อาการ -> แพทย์เฉพาะทาง
SYMPTOM_SPECIALIST_MAP = {
"ผื่น": SpecialistType.DERMATOLOGY,
"คัน": SpecialistType.DERMATOLOGY,
"สิว": SpecialistType.DERMATOLOGY,
"ผมร่วง": SpecialistType.DERMATOLOGY,
"ไข้": SpecialistType.INTERNAL,
"ไอ": SpecialistType.INTERNAL,
"เจ็บคอ": SpecialistType.INTERNAL,
"ปวดท้อง": SpecialistType.INTERNAL,
"ท้องเสีย": SpecialistType.INTERNAL,
"คลื่นไส้": SpecialistType.INTERNAL,
"เวียนศีรษะ": SpecialistType.INTERNAL,
"เจ็บหน้าอก": SpecialistType.CARDIOLOGY,
"ใจสั่น": SpecialistType.CARDIOLOGY,
"หายใจลำบาก": SpecialistType.URGENT,
"เลือดออกมาก": SpecialistType.URGENT,
"หมดสติ": SpecialistType.URGENT,
"กระดูก": SpecialistType.ORTHOPEDICS,
"ข้อ": SpecialistType.ORTHOPEDICS,
"ตา": SpecialistType.OPHTHALMOLOGY,
"มองเห็น": SpecialistType.OPHTHALMOLOGY,
"เครียด": SpecialistType.MENTAL,
"ซึมเศร้า": SpecialistType.MENTAL,
"นอนไม่หลับ": SpecialistType.MENTAL
}
URGENT_KEYWORDS = [
"หายใจไม่ออก", "แน่นหน้าอก", "เลือดออก", "หมดสติ",
"ชัก", "อาเจียนเลือด", "อุจจาระเป็นเลือด", "คลอด",
"เจ็บรุนแรงมาก", "ถูกทำร้าย"
]
def __init__(self, db_connection):
self.db = db_connection
def classify_urgency(self, symptom_text: str, severity: str) -> str:
"""จำแนกระดับความเร่งด่วน"""
# ตรวจสอบ keyword ฉุกเฉิน
symptom_lower = symptom_text.lower()
for keyword in self.URGENT_KEYWORDS:
if keyword in symptom_lower:
return "urgent"
# ตรวจสอบ severity จาก AI
severity_map = {
"urgent": "urgent",
"high": "high",
"medium": "medium",
"low": "low"
}
return severity_map.get(severity, "medium")
def route_to_specialist(self, symptom_text: str, ai_result: dict) -> dict:
"""กำหนดเส้นทางการรักษา"""
symptom_lower = symptom_text.lower()
suggested_specialist = SpecialistType.GENERAL
# หา specialist ที่เหมาะสมจาก symptom keywords
for keyword, specialist in self.SYMPTOM_SPECIALIST_MAP.items():
if keyword in symptom_lower:
suggested_specialist = specialist
break
# ตรวจสอบความเร่งด่วน
urgency = self.classify_urgency(
symptom_text,
ai_result.get('severity_level', 'medium')
)
# ถ้า urgent ให้ไปห้องฉุกเฉิน
if urgency == "urgent":
return {
"queue_type": "emergency",
"specialist": SpecialistType.URGENT,
"wait_time": 0,
"instructions": "กรุณาไปโรงพยาบาลหรือเรียกรถพยาบาลทันที",
"referral_reason": ai_result.get('urgency_reason', 'อาการฉุกเฉิน')
}
# หาแพทย์ที่ว่าง
available_doctors = self._find_available_doctors(suggested_specialist)
if not available_doctors:
# ถ้าไม่มีแพทย์เฉพาะทางว่าง ให้ general ก่อน
available_doctors = self._find_available_doctors(SpecialistType.GENERAL)
return {
"queue_type": "appointment",
"specialist": suggested_specialist,
"available_doctors": available_doctors,
"estimated_wait": self._calculate_wait_time(available_doctors),
"instructions": self._generate_instructions(urgency, suggested_specialist)
}
def _find_available_doctors(self, specialist_type: SpecialistType) -> list:
"""ค้นหาแพทย์ที่ว่างในปัจจุบัน"""
# Mock implementation - เชื่อมต่อ database จริงใน production
return [
{"id": "DR001", "name": "นายแพทย์สมชาย", "rating": 4.8, "next_available": "10:00"},
{"id": "DR002", "name": "นางแพทย์สมหญิง", "rating": 4.9, "next_available": "14:30"}
]
def _calculate_wait_time(self, doctors: list) -> str:
"""คำนวณเวลารอโดยประมาณ"""
if not doctors:
return "2-3 วัน"
earliest = doctors[0]['next_available']
now = datetime.now()
if ":" in str(earliest):
# ถ้าเป็นเวลาวันนี้
return f"ประมาณ {earliest}"
return earliest
def _generate_instructions(self, urgency: str, specialist: SpecialistType) -> str:
"""สร้างคำแนะนำสำหรับผู้ป่วย"""
base_instructions = {
"urgent": "⚠️ กรุณาไปพบแพทย์ทันที",
"high": "ควรพบแพทย์ภายใน 24 ชั่วโมง",
"medium": "นัดพบแพทย์ภายใน 3-7 วัน",
"low": "สามารถนัดพบแพทย์ตามความสะดวก"
}
return base_instructions.get(urgency, "ปรึกษาแพทย์ตามปกติ")
ตัวอย่างการใช้งาน
router = TriageRouter(db_connection=None)
ai_result = {
"initial_diagnosis": "ผื่นผิวหนังอักเสบ (Dermatitis)",
"severity_level": "medium",
"possible_conditions": ["Contact dermatitis", "Eczema", "Allergic reaction"],
"referral_recommendation": "ควรพบแพทย์ผิวหนัง",
"specialist_type": "แพทย์ผิวหนัง"
}
route = router.route_to_specialist(
"มีผื่นแดงคันที่แขน มีตุ่มน้ำเล็กน้อย",
ai_result
)
print(f"ประเภทคิว: {route['queue_type']}")
print(f"แพทย์เฉพาะทาง: {route['specialist'].value}")
print(f"คำแนะนำ: {route['instructions']}")ระบบ Chat History และ Context Management
import redis
import json
from typing import Optional
class ConsultationContext:
"""
จัดการ context การปรึกษาแพทย์
เก็บประวัติการสนทนา + ข้อมูลผู้ป่วย
"""
def __init__(self, session_id: str, api_key: str):
self.session_id = session_id
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.context_key = f"consultation:{session_id}"
# ใช้ Redis สำหรับ session storage
# self.redis = redis.Redis(host='localhost', port=6379, db=0)
self.conversation_history = []
self.patient_context = {}
def add_patient_info(self, info: dict):
"""เพิ่มข้อมูลผู้ป่วย"""
self.patient_context.update(info)
def add_message(self, role: str, content: str):
"""เพิ่มข้อความในประวัติการสนทนา"""
self.conversation_history.append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
def build_context_prompt(self) -> str:
"""สร้าง context prompt สำหรับ AI"""
context_parts = []
# ข้อมูลผู้ป่วย
if self.patient_context:
context_parts.append("ข้อมูลผู้ป่วย:")
for key, value in self.patient_context.items():
context_parts.append(f"- {key}: {value}")
# ประวัติการสนทนา (5 ข้อความล่าสุด)
if self.conversation_history:
context_parts.append("\nประวัติการสนทนา:")
recent = self.conversation_history[-5:]
for msg in recent:
role_display = "ผู้ป่วย" if msg['role'] == 'user' else "แพทย์ AI"
context_parts.append(f"{role_display}: {msg['content']}")
return "\n".join(context_parts)
def continue_consultation(self, new_question: str) -> str:
"""
สานต่อการปรึกษาโดยใช้ context ที่มี
"""
context = self.build_context_prompt()
full_prompt = f"""คุณเป็นผู้ช่วยแพทย์ที่กำลังให้คำปรึกษาเบื้องต้น
โปรดใช้ข้อมูลต่อไปนี้ในการตอบคำถามต่อไปนี้:
{context}
---
คำถามใหม่: {new_question}
ตอบเป็นภาษาไทย กระชับ เข้าใจง่าย หลีกเลี่ยงศัพท์แพทย์ที่ซับซ้อนเกินไป
"""
payload = {
"model": "gemini-2.0-flash",
"messages": [{"role": "user", "content": full_prompt}],
"temperature": 0.5,
"max_tokens": 1000
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json=payload,
timeout=30
)
if response.status_code == 200:
result = response.json()
answer = result['choices'][0]['message']['content']
# บันทึกลงประวัติ
self.add_message("user", new_question)
self.add_message("assistant", answer)
return answer
else:
raise Exception(f"API Error: {response.status_code}")
ตัวอย่างการใช้งาน
context = ConsultationContext(
session_id="session_2026_0524_001",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
เพิ่มข้อมูลผู้ป่วยครั้งแรก
context.add_patient_info({
"ชื่อ": "สมศรี ใจดี",
"อายุ": 35,
"โรคประจำตัว": "ไม่มี",
"ยาที่แพ้": "ไม่ทราบ"
})
ถามคำถามต่อเนื่อง
answer1 = context.continue_consultation("มีผื่นแดงที่คอ คันมาก")
print(f"คำตอบที่ 1: {answer1}")
answer2 = context.continue_consultation("เพิ่งเปลี่ยนสบู่ใหม่ จะเกี่ยวกับนั้นไหม")
print(f"คำตอบที่ 2: {answer2}")
answer3 = context.continue_consultation("ควรทายาอะไรก่อนไปพบแพทย์")
print(f"คำตอบที่ 3: {answer3}")ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. ข้อผิดพลาด "ConnectionError: timeout after 30s"
สาเหตุ: Gemini API มี latency สูงกว่าปกติ หรือ network timeout สั้นเกินไป
# ❌ วิธีที่ไม่ถูกต้อง - timeout 30 วินาที
response = requests.post(url, json=payload, timeout=30)
✅ วิธีที่ถูกต้อง - timeout 60 วินาที + retry logic
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def create_session_with_retry():
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
return session
session = create_session_with_retry()
response = session.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=60 # เพิ่ม timeout เป็น 60 วินาที
)
ถ้าใช้ async/await
import asyncio
import aiohttp
async def call_api_async(payload, max_retries=3):
for attempt in range(max_retries):
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
return await response.json()
except asyncio.TimeoutError:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt) # Exponential backoff2. ข้อผิดพลาด "401 Unauthorized" หรือ "403 Forbidden"
สาเหตุ: API key ไม่ถูกต้อง หมดอายุ หรือ quota เกิน limit
import os
from datetime import datetime
class HolySheepAuth:
"""
จัดการ authentication กับ HolySheep API
รวมถึงการตรวจสอบ quota และ renew key
"""
def __init__(self):
self.api_key = os.environ.get("HOLYSHEEP_API_KEY")
self.base_url = "https://api.holysheep.ai/v1"
def validate_key(self) -> dict:
"""ตรวจสอบความถูกต้องของ API key"""
try:
# เรียก API เพื่อตรวจสอบ quota
response = requests.get(
f"{self.base_url}/usage",
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=10
)
if response.status_code == 200:
return {
"valid": True,
"quota": response.json()
}
elif response.status_code == 401:
return {
"valid": False,
"error": "API key ไม่ถูกต้องหรือหมดอายุ"
}
elif response.status_code == 403:
return {
"valid": False,
"error": "ไม่มีสิทธิ์เข้าถึง API นี้"
}
else:
return {
"valid": False,
"error": f"ข้อผิดพลาด {response.status_code}"
}
except requests.exceptions.RequestException as e:
return {
"valid": False,
"error": f"ไม่สามารถเชื่อมต่อ: {str(e)}"
}
def check_quota_remaining(self) -> float:
"""ตรวจสอบ quota ที่เหลือ (หน่วย token)"""
validation = self.validate_key()