การปฏิรูประบบเบิกจ่ายค่ารักษาพยาบาลตามหลัก DRG (Diagnosis Related Groups) และ DIP (Big Data Diagnosis-Intervention Packet) กำลังเปลี่ยนแปลงวงการ Healthcare ของจีนอย่างรวดเร็ว หน่วยงานประกันสังคม สถานพยาบาล และบริษัทประกันสุขภาพ ต่างต้องการระบบ AI ที่สามารถตรวจสอบความถูกต้องของการเบิกจ่ายได้อย่างแม่นยำและรวดเร็ว
บทความนี้จะอธิบายวิธีการสร้าง DRG/DIP Intelligent Review Agent โดยใช้ HolySheep AI ซึ่งเป็นแพลตฟอร์มที่รวม LLM หลายรุ่นเข้าด้วยกัน รองรับการ fallback อัตโนมัติ พร้อมระบบ quota governance ที่ชาญฉลาด ช่วยให้องค์กรประหยัดค่าใช้จ่ายได้มากกว่า 85% เมื่อเทียบกับการใช้งานผ่านช่องทางอื่นโดยตรง
DRG/DIP Intelligent Review คืออะไร
DRG (Diagnosis Related Groups) คือระบบจัดกลุ่มการวินิจฉัยโรคที่ใช้ในการกำหนดอัตราค่าบริการแบบครอบคลุม (Package Pricing) ส่วน DIP (Big Data Diagnosis-Intervention Packet) เป็นระบบที่ใช้ข้อมูล大数据 วิเคราะห์ patterns ของการรักษาเพื่อกำหนดค่าบริการ
ระบบ Intelligent Review Agent ทำหน้าที่:
- ตรวจสอบความถูกต้องของ ICD codes ที่ใช้ในการเบิกจ่าย
- วิเคราะห์ความสอดคล้องระหว่าง diagnosis, procedure และ treatment plan
- ตรวจจับกรณี over-coding หรือ under-coding
- ตรวจสอบ compliance กับ local medical insurance policies
- จัดการ dispute ระหว่างผู้ให้บริการและผู้จ่ายเงิน
สถาปัตยกรรม Multi-Model Fallback สำหรับ Healthcare
ในระบบ Healthcare ที่ต้องการความเสถียรสูง การใช้ LLM เพียงตัวเดียวไม่เพียงพอ สถาปัตยกรรมที่แนะนำคือการใช้ Multi-Model Fallback ที่เรียงลำดับความสำคัญดังนี้:
- DeepSeek V3.2 - สำหรับงาน triage เบื้องต้น (ราคาถูกที่สุด $0.42/MTok)
- Gemini 2.5 Flash - สำหรับงาน review ทั่วไป (ราคาประหยัด $2.50/MTok)
- Claude Sonnet 4.5 - สำหรับงาน complex analysis ($15/MTok)
- GPT-4.1 - สำหรับงาน final verification ($8/MTok)
ระบบจะเริ่มจาก model ที่ถูกที่สุดก่อน และ fallback ไปยัง model ที่แพงกว่าเมื่อความ confidence ต่ำกว่าเกณฑ์ที่กำหนด วิธีนี้ช่วยประหยัดค่าใช้จ่ายได้ถึง 70% โดยยังคงรักษา accuracy ในระดับสูง
โค้ดตัวอย่าง: Multi-Model DRG Review Agent
ด้านล่างคือโค้ด Python ที่ใช้ HolySheep API สำหรับสร้างระบบ DRG/DIP Intelligent Review พร้อมระบบ fallback อัตโนมัติ
import openai
import json
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
กำหนดค่า HolySheep API
openai.api_key = "YOUR_HOLYSHEEP_API_KEY"
openai.api_base = "https://api.holysheep.ai/v1"
class ReviewPriority(Enum):
ROUTINE = "routine" # งานทั่วไป
COMPLEX = "complex" # งานซับซ้อน
URGENT = "urgent" # งานเร่งด่วน
FINAL = "final" # งานตรวจสอบขั้นสุดท้าย
@dataclass
class ModelConfig:
name: str
confidence_threshold: float
max_tokens: int
cost_per_mtok: float
priority: ReviewPriority
กำหนดโมเดลที่ใช้งาน (ราคาจาก HolySheep 2026)
MODEL_CONFIGS = {
ReviewPriority.ROUTINE: ModelConfig(
name="deepseek-v3.2",
confidence_threshold=0.85,
max_tokens=2048,
cost_per_mtok=0.42,
priority=ReviewPriority.ROUTINE
),
ReviewPriority.COMPLEX: ModelConfig(
name="gemini-2.5-flash",
confidence_threshold=0.90,
max_tokens=4096,
cost_per_mtok=2.50,
priority=ReviewPriority.COMPLEX
),
ReviewPriority.URGENT: ModelConfig(
name="claude-sonnet-4.5",
confidence_threshold=0.95,
max_tokens=8192,
cost_per_mtok=15.0,
priority=ReviewPriority.URGENT
),
ReviewPriority.FINAL: ModelConfig(
name="gpt-4.1",
confidence_threshold=0.98,
max_tokens=16384,
cost_per_mtok=8.0,
priority=ReviewPriority.FINAL
),
}
class DRGIntelligentReviewer:
def __init__(self):
self.review_history = []
self.quota_usage = {priority: {"used": 0, "limit": 100000}
for priority in ReviewPriority}
def analyze_case_complexity(self, case_data: Dict) -> ReviewPriority:
"""วิเคราะห์ความซับซ้อนของเคสเพื่อเลือกโมเดลที่เหมาะสม"""
complexity_score = 0
# ปัจจัยที่เพิ่มความซับซ้อน
if case_data.get("multiple_diagnoses", False):
complexity_score += 2
if case_data.get("surgical_procedure", False):
complexity_score += 3
if case_data.get("icu_admission", False):
complexity_score += 2
if case_data.get("complications", False):
complexity_score += 3
if case_data.get("total_cost") > 50000: # CNY
complexity_score += 2
# ตรวจสอบ dispute flag
if case_data.get("dispute_flag", False):
return ReviewPriority.FINAL
# ตรวจสอบ urgent flag
if case_data.get("urgent_review", False):
return ReviewPriority.URGENT
# ตรวจสอบ quota ก่อนเลือกโมเดล
if complexity_score >= 7:
return ReviewPriority.URGENT
elif complexity_score >= 4:
return ReviewPriority.COMPLEX
else:
return ReviewPriority.ROUTINE
def review_with_fallback(self, case_data: Dict) -> Dict:
"""ทำการ review พร้อมระบบ fallback อัตโนมัติ"""
# วิเคราะห์ความซับซ้อน
priority = self.analyze_case_complexity(case_data)
# ลำดับ fallback ตาม priority
fallback_order = {
ReviewPriority.ROUTINE: [
ReviewPriority.ROUTINE,
ReviewPriority.COMPLEX
],
ReviewPriority.COMPLEX: [
ReviewPriority.COMPLEX,
ReviewPriority.URGENT
],
ReviewPriority.URGENT: [
ReviewPriority.URGENT,
ReviewPriority.FINAL
],
ReviewPriority.FINAL: [ReviewPriority.FINAL]
}
last_error = None
for fallback_priority in fallback_order[priority]:
config = MODEL_CONFIGS[fallback_priority]
# ตรวจสอบ quota
if self.quota_usage[fallback_priority]["used"] >= \
self.quota_usage[fallback_priority]["limit"]:
print(f"⚠️ Quota exceeded for {config.name}, trying fallback")
continue
try:
result = self._call_model(config, case_data)
# บันทึกการใช้งาน
self.quota_usage[fallback_priority]["used"] += 1
self.review_history.append({
"case_id": case_data.get("case_id"),
"model": config.name,
"priority": priority.value,
"timestamp": time.time()
})
return {
"success": True,
"result": result,
"model_used": config.name,
"confidence": result.get("confidence", 0)
}
except Exception as e:
last_error = str(e)
print(f"❌ Model {config.name} failed: {e}, trying fallback...")
continue
return {
"success": False,
"error": f"All models failed: {last_error}",
"model_used": None
}
def _call_model(self, config: ModelConfig, case_data: Dict) -> Dict:
"""เรียกใช้โมเดลผ่าน HolySheep API"""
prompt = self._build_drg_prompt(case_data)
response = openai.ChatCompletion.create(
model=config.name,
messages=[
{"role": "system", "content": self._get_medical_system_prompt()},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=config.max_tokens
)
result_text = response.choices[0].message.content
# ตรวจสอบ confidence
if config.priority in [ReviewPriority.ROUTINE, ReviewPriority.COMPLEX]:
# สำหรับโมเดลถูกๆ ต้องตรวจสอบ confidence
if "CONFIDENCE_LOW" in result_text:
raise ValueError("Confidence below threshold")
return json.loads(result_text)
def _build_drg_prompt(self, case_data: Dict) -> str:
"""สร้าง prompt สำหรับ DRG review"""
return f"""
ทำการตรวจสอบกรณี DRG/DIP ต่อไปนี้:
Case ID: {case_data.get('case_id')}
การวินิจฉัยหลัก (ICD-10): {case_data.get('primary_diagnosis')}
การวินิจฉัยรอง: {case_data.get('secondary_diagnoses', [])}
Procedure (ICD-9-CM): {case_data.get('procedures', [])}
ค่าใช้จ่ายรวม: {case_data.get('total_cost')} CNY
ระยะเวลา admitted: {case_data.get('length_of_stay')} วัน
ตรวจสอบ:
1. ความถูกต้องของ ICD codes
2. ความสอดคล้องระหว่าง diagnosis และ procedure
3. การจัดกลุ่ม DRG ที่เหมาะสม
4. การ over-coding หรือ under-coding
5. ข้อผิดพลาดที่พบ (ถ้ามี)
Return JSON format with fields:
- drg_code: รหัส DRG ที่แนะนำ
- confidence: ความมั่นใจ (0-1)
- issues: รายการปัญหาที่พบ
- recommendation: คำแนะนำ
- requires_human_review: boolean
"""
def _get_medical_system_prompt(self) -> str:
"""System prompt สำหรับงาน medical review"""
return """คุณเป็นผู้เชี่ยวชาญด้านการตรวจสอบการเบิกจ่ายประกันสุขภาพตามระบบ DRG/DIP
คุณมีความรู้เกี่ยวกับ:
- มาตรฐาน ICD-10 และ ICD-9-CM
- กฎเกณฑ์การจัดกลุ่ม DRG
- นโยบายประกันสุขภาพของจีน
- Medical coding standards
ให้คำตอบที่แม่นยำ รัดกุม และเป็นไปตามมาตรฐานวิชาชีพเสมอ
หากไม่แน่ใจให้แนะนำให้มีการตรวจสอบโดยมนุษย์"""
def get_quota_status(self) -> Dict:
"""ดึงสถานะ quota ปัจจุบัน"""
return self.quota_usage.copy()
def estimate_cost(self, num_cases: int, priority_mix: Dict) -> Dict:
"""ประมาณการค่าใช้จ่าย"""
avg_tokens_per_case = {
ReviewPriority.ROUTINE: 1500,
ReviewPriority.COMPLEX: 3000,
ReviewPriority.URGENT: 5000,
ReviewPriority.FINAL: 8000
}
total_cost = 0
details = {}
for priority, ratio in priority_mix.items():
cases = int(num_cases * ratio)
tokens = cases * avg_tokens_per_case[priority]
mtok = tokens / 1_000_000
cost = mtok * MODEL_CONFIGS[priority].cost_per_mtok
total_cost += cost
details[priority.value] = {
"cases": cases,
"tokens": tokens,
"cost_usd": cost
}
# HolySheep อัตรา ¥1=$1 (ประหยัด 85%+)
return {
"total_cost_usd": total_cost,
"total_cost_cny": total_cost, # อัตรา 1:1
"savings_vs_direct": total_cost * 5.67, # ประหยัด ~85%
"details": details
}
ตัวอย่างการใช้งาน
if __name__ == "__main__":
reviewer = DRGIntelligentReviewer()
# ทดสอบกรณี routine
routine_case = {
"case_id": "DRG-2026-001",
"primary_diagnosis": "J18.9 - Pneumonia, unspecified",
"secondary_diagnoses": ["J44.1", "E11.9"],
"procedures": ["96.04"],
"total_cost": 8500,
"length_of_stay": 5
}
result = reviewer.review_with_fallback(routine_case)
print(f"Review Result: {json.dumps(result, indent=2)}")
# ประมาณการค่าใช้จ่าย
cost_estimate = reviewer.estimate_cost(
num_cases=10000,
priority_mix={
ReviewPriority.ROUTINE: 0.7,
ReviewPriority.COMPLEX: 0.2,
ReviewPriority.URGENT: 0.08,
ReviewPriority.FINAL: 0.02
}
)
print(f"Cost Estimate: {json.dumps(cost_estimate, indent=2)}")ระบบ Quota Governance สำหรับองค์กรขนาดใหญ่
องค์กร Healthcare มักต้องจัดการ quota ของหลาย department เช่น ฝ่ายclaims, ฝ่ายaudit, ฝ่ายcompliance ระบบ quota governance ช่วยให้สามารถ:
- กำหนด monthly quota ตาม department
- ติดตามการใช้งานแบบ real-time
- ตั้ง alert เมื่อใกล้ถึง limit
- รองรับการ cross-charge ระหว่าง departments
import hashlib
from datetime import datetime, timedelta
from typing import Optional, List
class QuotaManager:
"""ระบบจัดการ Quota สำหรับองค์กร Healthcare"""
def __init__(self):
# โครงสร้าง quota: {department_id: {monthly_limit, used, reset_date}}
self.department_quotas = {}
self.transaction_logs = []
def setup_department(
self,
dept_id: str,
monthly_limit_mtok: float,
priority_tier: str = "standard"
):
"""ตั้งค่า quota สำหรับ department"""
self.department_quotas[dept_id] = {
"monthly_limit_mtok": monthly_limit_mtok,
"used_mtok": 0,
"reset_date": self._get_next_reset_date(),
"tier": priority_tier,
"alert_threshold": 0.8, # alert เมื่อใช้ไป 80%
"budget_usd": monthly_limit_mtok * 0.42 # ประมาณค่าใช้จ่าย
}
print(f"✅ Department {dept_id} quota setup: {monthly_limit_mtok} MTok")
def check_and_consume(
self,
dept_id: str,
tokens: int,
model_name: str
) -> dict:
"""ตรวจสอบ quota และบันทึกการใช้งาน"""
if dept_id not in self.department_quotas:
return {
"allowed": False,
"reason": f"Department {dept_id} not found"
}
quota = self.department_quotas[dept_id]
mtok_used = tokens / 1_000_000
# ตรวจสอบการ reset รายเดือน
if datetime.now() >= quota["reset_date"]:
self._reset_quota(dept_id)
quota = self.department_quotas[dept_id]
# ตรวจสอบ quota
if quota["used_mtok"] + mtok_used > quota["monthly_limit_mtok"]:
return {
"allowed": False,
"reason": "Monthly quota exceeded",
"used": quota["used_mtok"],
"limit": quota["monthly_limit_mtok"],
"remaining": quota["monthly_limit_mtok"] - quota["used_mtok"]
}
# บันทึกการใช้งาน
quota["used_mtok"] += mtok_used
self._log_transaction(dept_id, tokens, model_name, mtok_used)
# ตรวจสอบ alert threshold
usage_ratio = quota["used_mtok"] / quota["monthly_limit_mtok"]
alert = None
if usage_ratio >= quota["alert_threshold"]:
alert = {
"level": "warning" if usage_ratio < 0.95 else "critical",
"message": f"Department {dept_id} has used {usage_ratio*100:.1f}% of quota",
"action_required": True
}
return {
"allowed": True,
"used": quota["used_mtok"],
"limit": quota["monthly_limit_mtok"],
"remaining": quota["monthly_limit_mtok"] - quota["used_mtok"],
"alert": alert
}
def get_department_report(self, dept_id: str) -> dict:
"""ดึงรายงานการใช้งานของ department"""
if dept_id not in self.department_quotas:
return {"error": "Department not found"}
quota = self.department_quotas[dept_id]
transactions = [t for t in self.transaction_logs if t["dept_id"] == dept_id]
# วิเคราะห์การใช้งานตาม model
model_usage = {}
for t in transactions:
model = t["model"]
if model not in model_usage:
model_usage[model] = {"count": 0, "mtok": 0}
model_usage[model]["count"] += 1
model_usage[model]["mtok"] += t["mtok_used"]
return {
"department_id": dept_id,
"tier": quota["tier"],
"period": {
"start": quota["reset_date"] - timedelta(days=30),
"end": quota["reset_date"]
},
"usage": {
"used_mtok": quota["used_mtok"],
"limit_mtok": quota["monthly_limit_mtok"],
"percentage": (quota["used_mtok"] / quota["monthly_limit_mtok"]) * 100,
"remaining_mtok": quota["monthly_limit_mtok"] - quota["used_mtok"]
},
"cost": {
"estimated_usd": quota["used_mtok"] * 0.42,
"budget_usd": quota["budget_usd"]
},
"by_model": model_usage,
"transaction_count": len(transactions)
}
def allocate_shared_pool(
self,
dept_ids: List[str],
shared_limit_mtok: float
):
"""แบ่ง shared pool quota ให้ departments"""
share_per_dept = shared_limit_mtok / len(dept_ids)
for dept_id in dept_ids:
if dept_id in self.department_quotas:
current = self.department_quotas[dept_id]
current["monthly_limit_mtok"] += share_per_dept
current["budget_usd"] = current["monthly_limit_mtok"] * 0.42
print(f"✅ Added {share_per_dept:.2f} MTok to {dept_id}")
def _reset_quota(self, dept_id: str):
"""reset quota รายเดือน"""
quota = self.department_quotas[dept_id]
old_used = quota["used_mtok"]
quota["used_mtok"] = 0
quota["reset_date"] = self._get_next_reset_date()
print(f"🔄 Department {dept_id} quota reset. Was used: {old_used:.3f} MTok")
def _get_next_reset_date(self) -> datetime:
"""คำนวณวัน reset ถัดไป (วันที่ 1 ของเดือน)"""
today = datetime.now()
if today.month == 12:
return datetime(today.year + 1, 1, 1)
else:
return datetime(today.year, today.month + 1, 1)
def _log_transaction(self, dept_id: str, tokens: int, model: str, mtok: float):
"""บันทึก transaction log"""
self.transaction_logs.append({
"timestamp": datetime.now().isoformat(),
"dept_id": dept_id,
"tokens": tokens,
"model": model,
"mtok_used": mtok,
"cost_usd": mtok * 0.42
})
class EnterpriseInvoiceManager:
"""ระบบจัดการ invoice สำหรับองค์กร"""
def __init__(self, quota_manager: QuotaManager):
self.quota_manager = quota_manager
self.invoices = []
self.tax_rate = 0.06 # VAT 6%
def generate_monthly_invoice(self, dept_id: str) -> dict:
"""สร้าง invoice รายเดือนสำหรับ department"""
report = self.quota_manager.get_department_report(dept_id)
if "error" in report:
return report
subtotal = report["cost"]["estimated_usd"]
tax = subtotal * self.tax_rate
total = subtotal + tax
invoice = {
"invoice_id": f"INV-{dept_id}-{datetime.now().strftime('%Y%m')}",
"department_id": dept_id,
"billing_period": report["period"],
"line_items": [
{
"description": f"AI Review Usage - {report['usage']['used_mtok']:.3f} MTok",
"amount_cny": subtotal
}
],
"subtotal_cny": subtotal,
"tax_cny": tax,
"total_cny": total,
"payment_methods": ["WeChat Pay", "Alipay", "Bank Transfer"],
"status": "pending",
"due_date": datetime.now() + timedelta(days=30)
}
self.invoices.append(invoice)
return invoice
def export_all_invoices(self, year: int, month: int) -> List[dict]:
"""export invoices ทั้งหมดของเดือน"""
period_str = f"{year}-{month:02d}"
filtered = [
inv for inv in self.invoices
if inv["invoice_id"].endswith(period_str)
]
return {
"period": period_str,
"invoices": filtered,
"total_amount_cny": sum(inv["total_cny"] for inv in filtered),
"total_amount_usd": sum(inv["total_cny"] for inv in filtered) # อัตรา 1:1
}
ตัวอย่างการใช้งาน Quota Manager
if __name__ == "__main__":
quota_mgr = QuotaManager()
invoice_mgr = EnterpriseInvoiceManager(quota_mgr)
# ตั้งค่า departments
quota_mgr.setup_department("claims-dept", monthly_limit_mtok=500, tier="premium")
quota_mgr.setup_department("audit-dept", monthly_limit_mtok=200, tier="standard")
quota_mgr.setup_department("compliance-dept", monthly_limit_mtok=100, tier="standard")
# จัดสรร shared pool
quota_mgr.allocate_shared_pool(
dept_ids=["claims-dept", "audit-dept"],
shared_limit_mtok=100
)
# ทดสอบการใช้งาน
result = quota_mgr.check_and_consume(
dept_id="claims-dept",
tokens=500000, # 500K tokens
model_name="deepseek-v3.2"
)
print(f"Quota check: {result}")
# ดึงรายงาน
report = quota_mgr.get_department_report("claims-dept")
print(f"\nDepartment Report:\n{json.dumps(report, indent=2, default=str)}")