Hallucination Guardrails

1from decimal import Decimal
2from datetime import datetime
3from pydantic import BaseModel, ConfigDict, Field, field_validator
4
5class SlipExtraction(BaseModel):
6    # strict=True: ห้าม coercion ข้ามชนิด, extra='forbid': ห้าม field เกิน
7    model_config = ConfigDict(strict=True, extra="forbid")
8
9    amount: Decimal = Field(gt=0, le=Decimal("10000000"))
10    ref_code: str = Field(min_length=6, max_length=40)
11    sender_account: str = Field(pattern=r"^[0-9xX*-]{6,20}$")
12    transferred_at: datetime
13    bank_name: str
14    confidence: float = Field(ge=0, le=1)
15
16    @field_validator("ref_code")
17    @classmethod
18    def normalize_ref(cls, v: str) -> str:
19        cleaned = v.strip().upper()
20        if not cleaned:
21            raise ValueError("ref_code ว่างไม่ได้")
22        return cleaned

1import json
2from pydantic import ValidationError
3from anthropic import Anthropic
4from .schema import SlipExtraction
5
6client = Anthropic()
7
8def extract_slip(image_b64: str) -> SlipExtraction:
9    tool = {
10        "name": "report_slip",
11        "description": "รายงานข้อมูลที่อ่านได้จากสลิปโอนเงิน",
12        "input_schema": SlipExtraction.model_json_schema(),
13    }
14    resp = client.messages.create(
15        model="claude-sonnet-4-5",
16        max_tokens=1024,
17        tools=[tool],
18        tool_choice={"type": "tool", "name": "report_slip"},
19        messages=[{
20            "role": "user",
21            "content": [
22                {"type": "image", "source": {
23                    "type": "base64", "media_type": "image/jpeg", "data": image_b64,
24                }},
25                {"type": "text", "text":
26                    "อ่านค่าจากสลิป ถ้าช่องไหนอ่านไม่ชัด ให้ลด confidence อย่าเดา"},
27            ],
28        }],
29    )
30    block = next(b for b in resp.content if b.type == "tool_use")
31    # validate ซ้ำเสมอ - schema ที่ส่งให้โมเดลไม่ได้การันตีว่ามันจะทำตาม
32    return SlipExtraction.model_validate(block.input)

1def levenshtein(a: str, b: str) -> int:
2    if a == b:
3        return 0
4    if not a:
5        return len(b)
6    if not b:
7        return len(a)
8
9    prev = list(range(len(b) + 1))
10    for i, ca in enumerate(a, start=1):
11        cur = [i]
12        for j, cb in enumerate(b, start=1):
13            cost = 0 if ca == cb else 1
14            cur.append(min(
15                prev[j] + 1,       # ลบ
16                cur[j - 1] + 1,    # เพิ่ม
17                prev[j - 1] + cost # แทน
18            ))
19        prev = cur
20    return prev[-1]
21
22def similarity_ratio(a: str, b: str) -> float:
23    if not a and not b:
24        return 1.0
25    return 1 - levenshtein(a, b) / max(len(a), len(b))

1from dataclasses import dataclass
2from .levenshtein import similarity_ratio
3from .schema import SlipExtraction
4
5KNOWN_BANKS = ["ธนาคารกสิกรไทย", "ธนาคารไทยพาณิชย์", "ธนาคารกรุงเทพ",
6               "ธนาคารกรุงไทย", "ธนาคารกรุงศรีอยุธยา", "พร้อมเพย์"]
7
8@dataclass
9class GuardResult:
10    ok: bool
11    reason: str | None = None
12
13def guard(extracted: SlipExtraction, expected_ref: str) -> GuardResult:
14    # 1) confidence ต่ำ = ส่งคนรีวิว ไม่ auto-approve
15    if extracted.confidence < 0.75:
16        return GuardResult(False, "confidence ต่ำเกินไป ต้องให้คนตรวจ")
17
18    # 2) เลขอ้างอิงต้องเกือบตรงกับที่ลงทะเบียนไว้ (กัน OCR สลับเลข)
19    ref_sim = similarity_ratio(extracted.ref_code, expected_ref.upper())
20    if ref_sim < 0.9:
21        return GuardResult(False, f"เลขอ้างอิงไม่ตรง (ใกล้เคียง {ref_sim:.0%})")
22
23    # 3) ชื่อธนาคารต้อง map เข้ารายการจริงได้
24    best = max(similarity_ratio(extracted.bank_name, b) for b in KNOWN_BANKS)
25    if best < 0.8:
26        return GuardResult(False, "ชื่อธนาคารไม่ตรงกับรายการที่รู้จัก")
27
28    return GuardResult(True)