ในฐานะนักพัฒนาที่ทำงานกับระบบเวชระเบียนอิเล็กทรอนิกส์มานานกว่า 5 ปี ผมเข้าใจดีว่าการจัดการข้อมูลส่วนตัวของผู้ป่วย (PHI — Protected Health Information) เป็นเรื่องที่ซับซ้อนและต้องระวังเป็นพิเศษ โดยเฉพาะเมื่อต้องนำข้อมูลไปใช้ในงานวิจัยหรือพัฒนา AI ในบทความนี้ ผมจะสอนคุณตั้งแต่เริ่มต้นจนสามารถสร้างระบบลบข้อมูลส่วนตัวที่ผ่านมาตรฐาน ด้วย HolySheep AI ซึ่งมีค่าบริการที่ประหยัดกว่าถึง 85% เมื่อเทียบกับบริการอื่น
PHI คืออะไร และทำไมต้องลบออก
ข้อมูล PHI หรือ Protected Health Information คือข้อมูลส่วนบุคคลของผู้ป่วยที่เปิดเผยได้เฉพาะผู้ที่เกี่ยวข้องเท่านั้น ตัวอย่างเช่น ชื่อ-นามสกุล หมายเลขบัตรประจำตัวประชาชน วันเกิด ที่อยู่ เบอร์โทรศัพท์ ผลการตรวจ และข้อมูลการรักษา
กฎหมายในหลายประเทศกำหนดให้ต้องลบข้อมูลเหล่านี้ออกก่อนนำไปใช้ มิฉะนั้นอาจถูกปรับหรือจำคุกได้ ระบบที่เราจะสร้างนี้ใช้หลักการ "去标识化" หรือ De-identification ซึ่งหมายถึงการลบข้อมูลระบุตัวตนออกจนไม่สามารถระบุได้ว่าเป็นของใคร
สถาปัตยกรรมระบบ HolySheep Medical PHI
ระบบที่เราจะสร้างประกอบด้วย 3 ส่วนหลักที่ทำงานร่วมกัน
- ชั้นรับข้อมูล (Ingestion Layer) — รับข้อมูลเวชระเบียนจากหลายแหล่ง
- ชั้นประมวลผล (Processing Layer) — ใช้ AI วิเคราะห์และลบข้อมูลส่วนตัว
- ชั้นตรวจสอบ (Verification Layer) — ตรวจสอบซ้ำด้วยหลายโมเดลเพื่อความแม่นยำ
การติดตั้งและเตรียมความพร้อม
ก่อนเริ่มต้น คุณต้องมี API Key จาก สมัครสมาชิก HolySheep AI ซึ่งจะได้รับเครดิตฟรีเมื่อลงทะเบียน ระบบรองรับการชำระเงินผ่าน WeChat และ Alipay ทำให้สะดวกมากสำหรับผู้ใช้ในไทย
ขั้นตอนที่ 1: ติดตั้ง Python และไลบรารีที่จำเป็น
ดาวน์โหลด Python 3.10 ขึ้นไปจากเว็บ python.org แล้วติดตั้ง จากนั้นเปิด Terminal หรือ Command Prompt แล้วพิมพ์คำสั่งติดตั้งไลบรารี
pip install requests openai pandas openpyxl python-dotenv
ขั้นตอนที่ 2: สร้างไฟล์กำหนดค่า
สร้างไฟล์ชื่อ config.py เพื่อเก็บค่าต่างๆ ที่ต้องใช้ในการเชื่อมต่อ
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
ความหน่วงตอบสนองน้อยกว่า 50ms พร้อม HolySheep
BASE_URL = "https://api.holysheep.ai/v1"
ราคาประหยัดมาก: 1 ดอลลาร์ = ¥1 (ประหยัด 85%+)
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
เลือกโมเดลสำหรับงาน PHI De-identification
MODEL_DEIDENTIFY = "gpt-4.1" # แม่นยำสูงสำหรับการลบข้อมูล
MODEL_VERIFY = "claude-sonnet-4.5" # ตรวจสอบซ้ำด้วยโมเดลที่สอง
MODEL_FALLBACK = "deepseek-v3.2" # ใช้เมื่อโมเดลหลักไม่ทำงาน
กำหนดราคาต่อพัน token (USD)
MODEL_PRICES = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
สร้างฟังก์ชันลบข้อมูล PHI ด้วย HolySheep API
ต่อไปจะเป็นการสร้างฟังก์ชันหลักที่ใช้ AI วิเคราะห์และลบข้อมูลส่วนตัว โค้ดนี้ใช้โมเดล gpt-4.1 ซึ่งมีความแม่นยำสูงในการตรวจจับและแทนที่ข้อมูลที่เป็นความลับ
# phi_deidentifier.py
import requests
import json
import time
from config import BASE_URL, HOLYSHEEP_API_KEY, MODEL_DEIDENTIFY
def deidentify_phi(medical_text, api_key=HOLYSHEEP_API_KEY):
"""
ลบข้อมูล PHI ออกจากข้อความเวชระเบียน
ราคาจริง: $8.00/MTok กับ gpt-4.1
ความหน่วง: <50ms กับ HolySheep
"""
prompt = f"""คุณคือผู้เชี่ยวชาญด้านการลบข้อมูลส่วนตัวทางการแพทย์
จงวิเคราะห์ข้อความต่อไปนี้และลบข้อมูล PHI ออกให้หมด
ข้อมูลที่ต้องลบ:
- ชื่อ-นามสกุล
- หมายเลขบัตรประจำตัวประชาชน
- วันเกิดและอายุ
- ที่อยู่และเบอร์โทรศัพท์
- หมายเลขโรงพยาบาลหรือ HN
- วันที่รับบริการ
กฎการลบ:
- ชื่อบุคคล → [ผู้ป่วย]
- หมายเลขบัตร → [หมายเลขบัตร]
- วันที่ → [วันที่] (ระบุเดือน/ปี ถ้าไม่จำเป็นต้องระบุวัน)
- อายุ → [อายุ XX ปี]
ข้อความต้นฉบับ:
{medical_text}
คืนค่าเป็น JSON ดังนี้:
{{
"deidentified_text": "...(ข้อความที่ลบข้อมูลแล้ว)...",
"detected_phi": [...(รายการสิ่งที่พบและลบออก)...],
"confidence_score": 0.0-1.0
}}"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": MODEL_DEIDENTIFY,
"messages": [
{"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญการลบข้อมูล PHI ที่แม่นยำและเชื่อถือได้"},
{"role": "user", "content": prompt}
],
"temperature": 0.1, # ค่าต่ำเพื่อความสม่ำเสมอ
"max_tokens": 2000
}
start_time = time.time()
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
result = response.json()
content = result["choices"][0]["message"]["content"]
# แปลง JSON string เป็น dict
try:
return json.loads(content), latency_ms
except json.JSONDecodeError:
# ถ้า AI คืนค่าไม่เป็น JSON ให้ลองหา JSON ในข้อความ
import re
match = re.search(r'\{.*\}', content, re.DOTALL)
if match:
return json.loads(match.group()), latency_ms
return {"error": "ไม่สามารถอ่านผลลัพธ์"}, latency_ms
else:
raise Exception(f"API Error {response.status_code}: {response.text}")
def deidentify_batch(records, api_key=HOLYSHEEP_API_KEY):
"""ประมวลผลข้อมูลหลายรายการพร้อมกัน"""
results = []
for i, record in enumerate(records):
print(f"กำลังประมวลผลรายการ {i+1}/{len(records)}...")
try:
deidentified, latency = deidentify_phi(record, api_key)
results.append({
"original": record,
"result": deidentified,
"latency_ms": latency
})
except Exception as e:
results.append({
"original": record,
"result": {"error": str(e)},
"latency_ms": 0
})
return results
ทดสอบการทำงาน
if __name__ == "__main__":
sample_text = """
ผู้ป่วยชาย อายุ 65 ปี นายสมชาย ใจดี ที่อยู่ 123/45 ถนนสุขุมวิท
แขวงคลองเตย เขตคลองเตย กรุงเทพ 10110 เบอร์ 0891234567
เข้ารับการรักษาวันที่ 15 มีนาคม 2567 หมายเลข HN: H123456
บัตรประชาชน: 1-2345-67890-12-3
"""
result, latency = deidentify_phi(sample_text)
print(f"ความหน่วง: {latency:.2f}ms")
print(json.dumps(result, ensure_ascii=False, indent=2))
ระบบตรวจสอบข้อมูลซ้ำด้วยหลายโมเดล
เพื่อให้มั่นใจว่าข้อมูล PHI ถูกลบออกหมดและถูกต้องตามมาตรฐาน ระบบจะใช้เทคนิค "多模型协同审核" หรือ Multi-Model Verification โดยให้โมเดลหลายตัวตรวจสอบผลลัพธ์ซ้ำ ทำให้ความแม่นยำเพิ่มขึ้นอย่างมาก
# phi_verifier.py
import requests
import json
from config import BASE_URL, HOLYSHEEP_API_KEY, MODEL_VERIFY, MODEL_FALLBACK
def verify_deidentification(original_text, deidentified_text, api_key=HOLYSHEEP_API_KEY):
"""
ตรวจสอบว่าข้อมูล PHI ถูกลบออกหมดแล้วหรือไม่
ราคา: $15.00/MTok กับ claude-sonnet-4.5 (ตรวจสอบละเอียด)
"""
verification_prompt = f"""ตรวจสอบข้อความที่ผ่านการลบข้อมูล PHI แล้ว
ข้อความต้นฉบับ:
{original_text}
ข้อความที่ลบข้อมูลแล้ว:
{deidentified_text}
ให้คะแนนความถูกต้อง (0-100) และระบุปัญหาที่พบ (ถ้ามี)
คืนค่า JSON:
{{
"score": 0-100,
"issues_found": [...],
"phi_remaining": [...],
"is_safe_for_research": true/false,
"recommendation": "..."
}}"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": MODEL_VERIFY,
"messages": [
{"role": "system", "content": "คุณคือผู้ตรวจสอบคุณภาพการลบข้อมูล PHI ที่เข้มงวด"},
{"role": "user", "content": verification_prompt}
],
"temperature": 0.1,
"max_tokens": 1500
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
result = response.json()
content = result["choices"][0]["message"]["content"]
try:
return json.loads(content)
except json.JSONDecodeError:
return {"error": "ไม่สามารถอ่านผลตรวจสอบ", "raw_response": content}
else:
# ใช้โมเดลสำรองถ้าโมเดลหลักไม่ทำงาน
payload["model"] = MODEL_FALLBACK
# ราคา fallback: $0.42/MTok กับ deepseek-v3.2
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
result = response.json()
return json.loads(result["choices"][0]["message"]["content"])
raise Exception(f"ทั้งโมเดลหลักและโมเดลสำรองไม่ทำงาน: {response.text}")
def comprehensive_phi_pipeline(text, api_key=HOLYSHEEP_API_KEY):
"""
ท่อประมวลผล PHI แบบครบวงจร
ขั้นที่ 1: ลบข้อมูล
ขั้นที่ 2: ตรวจสอบด้วยโมเดลที่ 2
ขั้นที่ 3: คืนค่าพร้อมรายงาน
"""
from phi_deidentifier import deidentify_phi
# ขั้นที่ 1: ลบข้อมูล
deid_result, latency = deidentify_phi(text, api_key)
if "error" in deid_result:
return {
"success": False,
"error": deid_result["error"],
"stage": "deidentification"
}
# ขั้นที่ 2: ตรวจสอบด้วยโมเดลที่ 2
verify_result = verify_deidentification(
text,
deid_result.get("deidentified_text", ""),
api_key
)
# ขั้นที่ 3: ตัดสินใจ
is_complete = verify_result.get("is_safe_for_research", False)
return {
"success": True,
"deidentified_text": deid_result.get("deidentified_text"),
"detected_phi": deid_result.get("detected_phi", []),
"verification": verify_result,
"latency_ms": latency,
"is_safe": is_complete,
"can_proceed": is_complete and verify_result.get("score", 0) >= 80
}
ทดสอบระบบ
if __name__ == "__main__":
test_record = """
นางสาวมาลี รักสงบ อายุ 42 ปี
บัตรประชาชน 3-2121-43434-56-7
เข้าพบแพทย์วันที่ 20 เมษายน 2567
วินิจฉัยว่าเป็นเบาหวานชนิดที่ 2
"""
result = comprehensive_phi_pipeline(test_record)
print(json.dumps(result, ensure_ascii=False, indent=2))
บันทึกผลลัพธ์ลงไฟล์ Excel
# save_results.py
import pandas as pd
import json
from datetime import datetime
def save_to_excel(results, filename="phi_deidentification_results.xlsx"):
"""
บันทึกผลลัพธ์การลบ PHI ลงไฟล์ Excel
พร้อมสถิติและรายงานสรุป
"""
# เตรียมข้อมูลสำหรับ DataFrame
data_rows = []
for item in results:
if item.get("success"):
result = item.get("result", {})
verification = result.get("verification", {})
row = {
"ข้อความต้นฉบับ": item.get("original", ""),
"ข้อความที่ลบข้อมูลแล้ว": result.get("deidentified_text", ""),
"ความเชื่อมั่น": result.get("confidence_score", 0),
"คะแนนตรวจสอบ": verification.get("score", 0),
"ปัญหาที่พบ": ", ".join(verification.get("issues_found", [])),
"PHI ที่เหลือ": ", ".join(verification.get("phi_remaining", [])),
"ปลอดภัยสำหรับวิจัย": verification.get("is_safe_for_research", False),
"ความหน่วง (ms)": item.get("latency_ms", 0),
"สถานะ": "✓ ผ่าน" if verification.get("is_safe_for_research") else "✗ ไม่ผ่าน"
}
else:
row = {
"ข้อความต้นฉบับ": item.get("original", ""),
"ข้อความที่ลบข้อมูลแล้ว": f"ข้อผิดพลาด: {item.get('result', {}).get('error', 'ไม่ทราบ')}",
"ความเชื่อมั่น": 0,
"คะแนนตรวจสอบ": 0,
"ปัญหาที่พบ": "การประมวลผลล้มเหลว",
"PHI ที่เหลือ": "",
"ปลอดภัยสำหรับวิจัย": False,
"ความหน่วง (ms)": 0,
"สถานะ": "✗ ผิดพลาด"
}
data_rows.append(row)
# สร้าง DataFrame
df = pd.DataFrame(data_rows)
# บันทึกเป็น Excel พร้อมจัดรูปแบบ
with pd.ExcelWriter(filename, engine='openpyxl') as writer:
df.to_excel(writer, sheet_name='ผลลัพธ์', index=False)
# สร้าง sheet สรุป
summary_data = {
"รายการ": [
"จำนวนทั้งหมด",
"ผ่านการตรวจสอบ",
"ไม่ผ่านการตรวจสอบ",
"อัตราความสำเร็จ",
"ความหน่วงเฉลี่ย (ms)",
"คะแนนเฉลี่ย"
],
"ค่า": [
len(results),
sum(1 for r in results if r.get("success")),
sum(1 for r in results if not r.get("success")),
f"{(sum(1 for r in results if r.get('success')) / len(results) * 100):.1f}%",
f"{sum(r.get('latency_ms', 0) for r in results) / len(results):.2f}",
f"{sum(r.get('result', {}).get('confidence_score', 0) for r in results) / len(results):.2f}"
]
}
summary_df = pd.DataFrame(summary_data)
summary_df.to_excel(writer, sheet_name='สรุป', index=False)
# ปรับความกว้างคอลัมน์
worksheet = writer.sheets['ผลลัพธ์']
for idx, col in enumerate(df.columns):
max_length = max(df[col].astype(str).map(len).max(), len(col)) + 2
worksheet.column_dimensions[chr(65 + idx)].width = min(max_length, 50)
return filename
def generate_audit_report(results, output_file="audit_report.txt"):
"""สร้างรายงานตรวจสอบย้อนกลับตามมาตรฐาน"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with open(output_file, 'w', encoding='utf-8') as f:
f.write("=" * 60 + "\n")
f.write("รายงานตรวจสอบย้อนกลับ PHI De-identification\n")
f.write("=" * 60 + "\n")
f.write(f"วันที่สร้าง: {timestamp}\n")
f.write(f"จำนวนรายการ: {len(results)}\n\n")
f.write("-" * 60 + "\n")
f.write("รายละเอียดแต่ละรายการ\n")
f.write("-" * 60 + "\n\n")
for i, item in enumerate(results, 1):
f.write(f"[{i}] ")
if item.get("success"):
result = item.get("result", {})
verification = result.get("verification", {})
f.write(f"ผ่าน (คะแนน: {verification.get('score', 0)})\n")
f.write(f" PHI ที่ตรวจพบ: {len(result.get('detected_phi', []))} รายการ\n")
f.write(f" PHI ที่เหลือ: {len(verification.get('phi_remaining', []))} รายการ\n")
else:
f.write("ไม่ผ่าน\n")
f.write(f" ข้อผิดพลาด: {item.get('result', {}).get('error', 'ไม่ทราบ')}\n")
f.write("\n")
f.write("