Chào các bạn developer! Mình là Minh, tech lead của một startup AI tại TP.HCM. Hôm nay mình muốn chia sẻ hành trình chuyển đổi hạ tầng AI của đội mình từ việc sử dụng trực tiếp OpenAI API sang HolySheep AI - và đặc biệt là cách chúng tôi xây dựng hệ thống xử lý lỗi Function Calling bất bại.

Bối Cảnh: Vì Sao Chúng Tôi Cần Function Calling?

Trong dự án chatbot chăm sóc khách hàng của mình, chúng tôi cần AI có khả năng:

Function Calling là giải pháp hoàn hảo - cho phép AI "gọi" các function đã định nghĩa sẵn thay vì chỉ trả text. Tuy nhiên, khi production với hơn 50,000 request/ngày, chúng tôi gặp phải những vấn đề nghiêm trọng về error handling và parameter validation.

Kiến Trúc Function Calling Trên HolySheep AI

HolySheep cung cấp endpoint tương thích hoàn toàn với OpenAI格式, nhưng với độ trễ thấp hơn 85% và chi phí chỉ bằng 1/6 so với API chính thức.

Cấu Hình Base URL và API Key

# Cài đặt thư viện
pip install openai httpx pydantic

Cấu hình client

import os from openai import OpenAI

Quan trọng: Sử dụng endpoint HolySheep thay vì OpenAI

client = OpenAI( api_key=os.getenv("YOUR_HOLYSHEEP_API_KEY"), base_url="https://api.holysheep.ai/v1" # Endpoint chính thức của HolySheep ) print("✅ Kết nối HolySheep AI thành công!") print(f"📡 Base URL: {client.base_url}")

Định Nghĩa Functions với JSON Schema

# Định nghĩa functions cho hệ thống đặt hàng
from typing import List, Optional
from pydantic import BaseModel, Field, field_validator

functions = [
    {
        "name": "get_order_status",
        "description": "Lấy thông tin trạng thái đơn hàng theo mã vận đơn",
        "parameters": {
            "type": "object",
            "properties": {
                "tracking_code": {
                    "type": "string",
                    "description": "Mã vận đơn 10-13 ký tự (format: VH-XXXXX)",
                    "pattern": "^VH-[A-Z0-9]{5,8}$"
                },
                "include_timeline": {
                    "type": "boolean",
                    "description": "Bao gồm timeline chi tiết",
                    "default": False
                }
            },
            "required": ["tracking_code"],
            "additionalProperties": False
        }
    },
    {
        "name": "calculate_shipping_fee",
        "description": "Tính phí vận chuyển dựa trên địa chỉ và trọng lượng",
        "parameters": {
            "type": "object",
            "properties": {
                "weight_kg": {
                    "type": "number",
                    "description": "Trọng lượng kiện hàng (kg)",
                    "minimum": 0.1,
                    "maximum": 70.0
                },
                "from_district": {
                    "type": "string",
                    "enum": ["quan_1", "quan_3", "quan_5", "tan_binh", "phu_nhuan"]
                },
                "to_province": {
                    "type": "string",
                    "enum": ["hcm", "hanoi", "danang", "cantho", "hue", "other"]
                },
                "service_type": {
                    "type": "string",
                    "enum": ["standard", "express", "COD"],
                    "default": "standard"
                }
            },
            "required": ["weight_kg", "from_district", "to_province"]
        }
    },
    {
        "name": "cancel_order",
        "description": "Hủy đơn hàng (chỉ áp dụng khi trạng thái 'pending' hoặc 'confirmed')",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "Mã đơn hàng 6 chữ số"
                },
                "reason": {
                    "type": "string",
                    "description": "Lý do hủy đơn",
                    "minLength": 10,
                    "maxLength": 500
                },
                "customer_confirmed": {
                    "type": "boolean",
                    "description": "Xác nhận từ khách hàng"
                }
            },
            "required": ["order_id", "reason", "customer_confirmed"]
        }
    }
]

print(f"📋 Đã định nghĩa {len(functions)} functions cho hệ thống")

Hệ Thống Validation Tham Số Tự Động

Đây là phần quan trọng nhất - mình đã xây dựng một lớp validation đa tầng để đảm bảo mọi tham số đều hợp lệ trước khi gọi function thực tế.

import re
from typing import Any, Dict, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ValidationError(Exception):
    """Custom exception cho validation errors"""
    def __init__(self, field: str, message: str, value: Any = None):
        self.field = field
        self.message = message
        self.value = value
        super().__init__(f"Validation Error [{field}]: {message}")

class FunctionCallError(Exception):
    """Exception khi gọi function thất bại"""
    def __init__(self, function_name: str, original_error: Exception, retry_count: int):
        self.function_name = function_name
        self.original_error = original_error
        self.retry_count = retry_count
        super().__init__(f"Function '{function_name}' failed after {retry_count} retries: {original_error}")

@dataclass
class ValidationRule:
    """Quy tắc validation cho một trường"""
    field_name: str
    field_type: type
    required: bool = True
    min_value: Optional[float] = None
    max_value: Optional[float] = None
    min_length: Optional[int] = None
    max_length: Optional[int] = None
    pattern: Optional[str] = None
    enum_values: Optional[List[Any]] = None
    custom_validator: Optional[Callable] = None

class ParameterValidator:
    """
    Validator mạnh mẽ cho Function Calling parameters
    Hỗ trợ: type checking, range validation, pattern matching, enum, custom rules
    """
    
    def __init__(self, function_schema: Dict[str, Any]):
        self.schema = function_schema
        self.rules = self._parse_schema_to_rules(function_schema)
    
    def _parse_schema_to_rules(self, schema: Dict) -> List[ValidationRule]:
        """Parse JSON schema thành danh sách ValidationRule"""
        rules = []
        properties = schema.get("parameters", {}).get("properties", {})
        required_fields = schema.get("parameters", {}).get("required", [])
        
        for field_name, field_def in properties.items():
            rule = ValidationRule(
                field_name=field_name,
                field_type=self._get_python_type(field_def.get("type")),
                required=field_name in required_fields,
                min_value=field_def.get("minimum"),
                max_value=field_def.get("maximum"),
                min_length=field_def.get("minLength"),
                max_length=field_def.get("maxLength"),
                pattern=field_def.get("pattern"),
                enum_values=field_def.get("enum"),
                custom_validator=None
            )
            rules.append(rule)
        
        return rules
    
    def _get_python_type(self, json_type: str) -> type:
        """Map JSON type sang Python type"""
        type_map = {
            "string": str,
            "number": float,
            "integer": int,
            "boolean": bool,
            "array": list,
            "object": dict
        }
        return type_map.get(json_type, str)
    
    def validate(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """
        Validate tất cả parameters
        Returns: Dict đã được sanitize
        Raises: ValidationError nếu có lỗi
        """
        validated = {}
        errors = []
        
        for rule in self.rules:
            value = params.get(rule.field_name)
            
            # Check required
            if value is None:
                if rule.required:
                    errors.append(ValidationError(
                        rule.field_name,
                        "Trường bắt buộc nhưng không được cung cấp"
                    ))
                continue
            
            # Type checking
            if not isinstance(value, rule.field_type):
                try:
                    # Auto-convert nếu có thể
                    if rule.field_type == int and isinstance(value, float):
                        value = int(value)
                    elif rule.field_type == float and isinstance(value, int):
                        value = float(value)
                    else:
                        errors.append(ValidationError(
                            rule.field_name,
                            f"Type không hợp lệ: expected {rule.field_type.__name__}, got {type(value).__name__}",
                            value
                        ))
                        continue
                except (ValueError, TypeError):
                    errors.append(ValidationError(
                        rule.field_name,
                        f"Không thể convert sang {rule.field_type.__name__}",
                        value
                    ))
                    continue
            
            # Range validation (numbers)
            if rule.min_value is not None and value < rule.min_value:
                errors.append(ValidationError(
                    rule.field_name,
                    f"Giá trị {value} nhỏ hơn minimum {rule.min_value}",
                    value
                ))
                continue
            
            if rule.max_value is not None and value > rule.max_value:
                errors.append(ValidationError(
                    rule.field_name,
                    f"Giá trị {value} lớn hơn maximum {rule.max_value}",
                    value
                ))
                continue
            
            # Length validation (strings)
            if rule.min_length is not None and len(value) < rule.min_length:
                errors.append(ValidationError(
                    rule.field_name,
                    f"Độ dài {len(value)} nhỏ hơn minimum {rule.min_length}",
                    value
                ))
                continue
            
            if rule.max_length is not None and len(value) > rule.max_length:
                errors.append(ValidationError(
                    rule.field_name,
                    f"Độ dài {len(value)} lớn hơn maximum {rule.max_length}",
                    value
                ))
                continue
            
            # Pattern validation
            if rule.pattern and isinstance(value, str):
                if not re.match(rule.pattern, value):
                    errors.append(ValidationError(
                        rule.field_name,
                        f"Pattern không khớp: '{value}' không match '{rule.pattern}'",
                        value
                    ))
                    continue
            
            # Enum validation
            if rule.enum_values is not None and value not in rule.enum_values:
                errors.append(ValidationError(
                    rule.field_name,
                    f"Giá trị '{value}' không nằm trong danh sách cho phép: {rule.enum_values}",
                    value
                ))
                continue
            
            # Custom validator
            if rule.custom_validator:
                try:
                    rule.custom_validator(value)
                except Exception as e:
                    errors.append(ValidationError(
                        rule.field_name,
                        f"Custom validation failed: {str(e)}",
                        value
                    ))
                    continue
            
            validated[rule.field_name] = value
        
        if errors:
            error_summary = "; ".join([f"{e.field}: {e.message}" for e in errors])
            raise ValidationError("multiple", error_summary, errors)
        
        return validated
    
    def validate_with_defaults(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Validate và apply default values"""
        # Apply defaults
        for rule in self.rules:
            if rule.field_name not in params:
                default = self.schema.get("parameters", {}).get("properties", {}).get(rule.field_name, {}).get("default")
                if default is not None:
                    params[rule.field_name] = default
        
        return self.validate(params)

Test validator

validator = ParameterValidator(functions[0]) # get_order_status test_params = { "tracking_code": "VH-ABC123", "include_timeline": True } try: result = validator.validate_with_defaults(test_params) print(f"✅ Validation thành công: {result}") except ValidationError as e: print(f"❌ Validation thất bại: {e}")

Chiến Lược Retry và Fallback Toàn Diện

Đây là mấu chốt của hệ thống production. Mình đã implement một bộ retry logic với exponential backoff và circuit breaker pattern.

import asyncio
import time
from typing import Any, Callable, Dict, List, Optional, TypeVar, Union
from functools import wraps
from enum import Enum
import json

T = TypeVar('T')

class RetryStrategy(Enum):
    """Các chiến lược retry khả dụng"""
    FIXED = "fixed"                    # Fixed delay giữa các lần retry
    LINEAR = "linear"                  # Delay tăng tuyến tính
    EXPONENTIAL = "exponential"        # Delay tăng theo cấp số nhân
    EXPONENTIAL_WITH_JITTER = "exp_jitter"  # Exponential + random jitter

class CircuitState(Enum):
    """Circuit Breaker states"""
    CLOSED = "closed"      # Bình thường, cho phép request
    OPEN = "open"          # Block request do lỗi liên tục
    HALF_OPEN = "half_open"  # Thử nghiệm một request

@dataclass
class RetryConfig:
    """Cấu hình retry"""
    max_attempts: int = 3
    initial_delay: float = 1.0      # seconds
    max_delay: float = 30.0         # seconds
    strategy: RetryStrategy = RetryStrategy.EXPONENTIAL_WITH_JITTER
    retryable_exceptions: tuple = (ConnectionError, TimeoutError, RuntimeError)
    enable_circuit_breaker: bool = True
    circuit_failure_threshold: int = 5    # Số lỗi để open circuit
    circuit_recovery_timeout: float = 60.0 # seconds trước khi thử lại

class CircuitBreaker:
    """Circuit Breaker implementation để ngăn cascade failures"""
    
    def __init__(self, config: RetryConfig):
        self.config = config
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time: Optional[float] = None
        self.success_count = 0
    
    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time >= self.config.circuit_recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                logger.info("🔄 Circuit chuyển sang HALF_OPEN")
                return True
            return False
        
        # HALF_OPEN: cho phép một request thử nghiệm
        return True
    
    def record_success(self):
        self.success_count += 1
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
            logger.info("✅ Circuit chuyển sang CLOSED sau khi phục hồi")
        elif self.state == CircuitState.CLOSED:
            self.success_count = 0
    
    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.config.circuit_failure_threshold:
            self.state = CircuitState.OPEN
            logger.warning(f"⚠️ Circuit OPEN do {self.failure_count} lỗi liên tiếp")

def calculate_delay(attempt: int, config: RetryConfig) -> float:
    """Tính toán delay với strategy được chọn"""
    if config.strategy == RetryStrategy.FIXED:
        delay = config.initial_delay
    elif config.strategy == RetryStrategy.LINEAR:
        delay = config.initial_delay * attempt
    elif config.strategy == RetryStrategy.EXPONENTIAL:
        delay = config.initial_delay * (2 ** (attempt - 1))
    elif config.strategy == RetryStrategy.EXPONENTIAL_WITH_JITTER:
        base_delay = config.initial_delay * (2 ** (attempt - 1))
        jitter = base_delay * 0.2 * (hash(str(time.time())) % 10 / 10)
        delay = base_delay + jitter
    else:
        delay = config.initial_delay
    
    return min(delay, config.max_delay)

async def async_retry_with_fallback(
    func: Callable[..., T],
    fallback_func: Optional[Callable[..., T]] = None,
    config: Optional[RetryConfig] = None,
    function_name: str = "unknown",
    **kwargs
) -> T:
    """
    Async retry với fallback support
    - Thử gọi func với retry logic
    - Nếu thất bại sau max_attempts, gọi fallback_func
    - Circuit breaker để tránh cascade failure
    """
    if config is None:
        config = RetryConfig()
    
    circuit_breaker = CircuitBreaker(config) if config.enable_circuit_breaker else None
    last_exception: Optional[Exception] = None
    
    for attempt in range(1, config.max_attempts + 1):
        # Kiểm tra circuit breaker
        if circuit_breaker and not circuit_breaker.can_execute():
            logger.warning(f"⛔ Circuit breaker OPEN, chuyển sang fallback")
            break
        
        try:
            logger.info(f"🔄 Attempt {attempt}/{config.max_attempts} cho {function_name}")
            
            if asyncio.iscoroutinefunction(func):
                result = await func(**kwargs)
            else:
                result = func(**kwargs)
            
            if circuit_breaker:
                circuit_breaker.record_success()
            
            logger.info(f"✅ {function_name} thành công ở attempt {attempt}")
            return result
            
        except config.retryable_exceptions as e:
            last_exception = e
            logger.warning(f"⚠️ Attempt {attempt} thất bại: {type(e).__name__}: {str(e)}")
            
            if circuit_breaker:
                circuit_breaker.record_failure()
            
            if attempt < config.max_attempts:
                delay = calculate_delay(attempt, config)
                logger.info(f"⏳ Chờ {delay:.2f}s trước retry...")
                await asyncio.sleep(delay)
            
        except Exception as e:
            # Non-retryable exception - throw ngay
            logger.error(f"❌ Non-retryable error: {type(e).__name__}: {str(e)}")
            raise
    
    # Tất cả attempts đều thất bại
    logger.error(f"❌ {function_name} thất bại sau {config.max_attempts} attempts")
    
    if fallback_func:
        logger.info(f"🔄 Chuyển sang fallback function...")
        try:
            if asyncio.iscoroutinefunction(fallback_func):
                result = await fallback_func(**kwargs)
            else:
                result = fallback_func(**kwargs)
            logger.info(f"✅ Fallback thành công")
            return result
        except Exception as fallback_error:
            logger.error(f"❌ Fallback cũng thất bại: {fallback_error}")
            raise FunctionCallError(function_name, fallback_error, config.max_attempts)
    
    raise FunctionCallError(
        function_name, 
        last_exception or Exception("Unknown error"), 
        config.max_attempts
    )

============ DEMO: Tích hợp HolySheep Function Calling ============

async def call_holysheep_function_calling( messages: List[Dict], functions: List[Dict], function_name: str, params: Dict[str, Any] ) -> Dict[str, Any]: """ Gọi HolySheep Function Calling với đầy đủ error handling """ # Validate parameters trước validator = ParameterValidator( functions[[f["name"] for f in functions].index(function_name)] ) validated_params = validator.validate_with_defaults(params) async def call_api(): response = client.chat.completions.create( model="gpt-4o", messages=messages, tools=[{"type": "function", "function": f} for f in functions], tool_choice={"type": "function", "function": {"name": function_name}}, temperature=0.1 ) return response # Fallback: Trả về mock data nếu API fail def fallback_mock_call(): logger.info(f"📦 Fallback mock data cho {function_name}") if function_name == "get_order_status": return { "status": "pending", "message": "Hệ thống tạm thời unavailable, hiển thị dữ liệu cache", "last_update": "2026-01-15T10:30:00Z" } elif function_name == "calculate_shipping_fee": return { "fee": 25000, "currency": "VND", "estimated_days": 3, "message": "Ước tính dựa trên cache" } return {"status": "fallback", "message": "Using cached data"} retry_config = RetryConfig( max_attempts=3, initial_delay=1.0, max_delay=10.0, strategy=RetryStrategy.EXPONENTIAL_WITH_JITTER, enable_circuit_breaker=True, circuit_failure_threshold=3, circuit_recovery_timeout=30.0 ) result = await async_retry_with_fallback( call_api, fallback_func=fallback_mock_call, config=retry_config, function_name=function_name ) return result

Chạy demo

async def main(): messages = [{"role": "user", "content": "Kiểm tra đơn hàng VH-ABC123"}] result = await call_holysheep_function_calling( messages=messages, functions=functions, function_name="get_order_status", params={"tracking_code": "VH-ABC123", "include_timeline": True} ) print(f"🎉 Kết quả: {result}")

asyncio.run(main())

Webhook Handler với Error Recovery

Trong production, chúng ta cần một webhook handler để xử lý function call results một cách an toàn.

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Optional, List, Dict, Any, Union
import traceback

app = FastAPI(title="HolySheep Function Calling Service")

class FunctionCallRequest(BaseModel):
    """Request format cho function call"""
    messages: List[Dict[str, Any]]
    function_name: Optional[str] = None
    user_id: str
    session_id: str
    fallback_enabled: bool = True
    metadata: Optional[Dict[str, Any]] = None

class FunctionCallResponse(BaseModel):
    """Response format"""
    success: bool
    function_name: str
    result: Optional[Dict[str, Any]]
    error: Optional[Dict[str, Any]]
    fallback_used: bool = False
    processing_time_ms: float

@app.post("/v1/function-call")
async def handle_function_call(
    request: FunctionCallRequest,
    background_tasks: BackgroundTasks
) -> JSONResponse:
    """
    Main endpoint cho HolySheep Function Calling
    - Auto-validation
    - Retry with exponential backoff
    - Circuit breaker
    - Fallback to cache/mock data
    """
    start_time = time.time()
    
    try:
        # Tìm function schema
        target_function = None
        for func in functions:
            if func["name"] == request.function_name:
                target_function = func
                break
        
        if not target_function:
            raise ValueError(f"Function '{request.function_name}' not found")
        
        # Build messages for API
        messages_with_tools = request.messages.copy()
        messages_with_tools.append({
            "role": "system",
            "content": f"Bạn cần gọi function '{request.function_name}' để xử lý yêu cầu này."
        })
        
        # Gọi với đầy đủ error handling
        result = await call_holysheep_function_calling(
            messages=messages_with_tools,
            functions=functions,
            function_name=request.function_name,
            params=request.metadata or {}
        )
        
        processing_time = (time.time() - start_time) * 1000
        
        return JSONResponse({
            "success": True,
            "function_name": request.function_name,
            "result": result,
            "error": None,
            "fallback_used": False,
            "processing_time_ms": round(processing_time, 2)
        })
        
    except ValidationError as e:
        logger.error(f"Validation Error: {e}")
        return JSONResponse({
            "success": False,
            "function_name": request.function_name,
            "result": None,
            "error": {
                "type": "validation_error",
                "message": str(e),
                "field": e.field
            },
            "fallback_used": False,
            "processing_time_ms": round((time.time() - start_time) * 1000, 2)
        }, status_code=400)
        
    except FunctionCallError as e:
        logger.error(f"Function Call Error: {e}")
        if request.fallback_enabled:
            # Trigger fallback
            fallback_result = get_cached_fallback_data(request.function_name)
            return JSONResponse({
                "success": True,
                "function_name": request.function_name,
                "result": fallback_result,
                "error": None,
                "fallback_used": True,
                "warning": f"Using fallback due to: {str(e.original_error)}",
                "processing_time_ms": round((time.time() - start_time) * 1000, 2)
            })
        return JSONResponse({
            "success": False,
            "function_name": request.function_name,
            "result": None,
            "error": {
                "type": "function_call_error",
                "message": str(e),
                "retry_count": e.retry_count
            },
            "fallback_used": False,
            "processing_time_ms": round((time.time() - start_time) * 1000, 2)
        }, status_code=503)
        
    except Exception as e:
        logger.error(f"Unexpected Error: {traceback.format_exc()}")
        return JSONResponse({
            "success": False,
            "function_name": request.function_name,
            "result": None,
            "error": {
                "type": "internal_error",
                "message": str(e)
            },
            "fallback_used": False,
            "processing_time_ms": round((time.time() - start_time) * 1000, 2)
        }, status_code=500)

def get_cached_fallback_data(function_name: str) -> Dict[str, Any]:
    """Fallback data từ cache hoặc default values"""
    cache = {
        "get_order_status": {"status": "unknown", "cached": True},
        "calculate_shipping_fee": {"fee": 30000, "currency": "VND", "cached": True},
        "cancel_order": {"success": False, "message": "Service unavailable", "cached": True}
    }
    return cache.get(function_name, {"status": "fallback"})

Health check endpoint

@app.get("/health") async def health_check(): return {"status": "healthy", "service": "HolySheep Function Calling"}

Logging background tasks

@app.on_event("startup") async def startup(): logger.info("🚀 HolySheep Function Calling Service started") logger.info(f"📡 Endpoint: https://api.holysheep.ai/v1")

Lỗi Thường Gặp và Cách Khắc Phục

1. Lỗi "Invalid parameter type" - Type Mismatch

Mô tả: AI trả về tham số với type không đúng với JSON schema định nghĩa.

Nguyên nhân:

Giải pháp:

# Auto-type conversion trong validator
def _auto_convert_type(self, value: Any, target_type: type) -> Any:
    """Tự động convert type khi có thể"""
    if isinstance(value, target_type):
        return value
    
    try:
        if target_type == int:
            return int(float(value))  # Xử lý "42.0" -> 42
        elif target_type == float:
            if isinstance(value, str):
                value = value.replace(",", ".")
            return float(value)
        elif target_type == bool:
            if isinstance(value, str):
                return value.lower() in ("true", "1", "yes", "có")
            return bool(value)
        elif target_type == list:
            if isinstance(value, str):
                return [item.strip() for item in value.split(",")]
            return list(value)
    except (ValueError, TypeError):
        pass
    
    raise ValidationError(
        self.field_name,
        f"Không thể convert '{value}' ({type(value).__name__}) sang {target_type.__name__}"
    )

2. Lỗi "Pattern validation failed" - Regex Không Khớp

Mô tả: Giá trị string không match với regex pattern định nghĩa.

Nguyên nhân:

Giải pháp:

# Smart pattern validation với auto-fix
def _smart_validate_pattern(self, value: str, pattern: str) -> str:
    """Validate và tự động fix common issues"""
    # Strip whitespace
    value = value.strip()
    
    # Extract pattern flags từ regex
    flags = ""
    if pattern.endswith("(?i)"):
        pattern = pattern[:-4]
        value = value.lower()
    elif pattern.endswith("(?i)"):
        pattern = pattern[:-4]
        value = value.upper()
    
    # Thử match trực tiếp
    if re.match(pattern, value):
        return value
    
    # Thử normalize và match lại
    normalized = re.sub(r'[\s\-_/]+', '', value)  # Remove separators
    pattern_normalized = re.sub(r'[\s\-_/]+', '', pattern)
    
    if re.match(pattern_normalized, normalized):
        # Return original value, chỉ fix nếu pattern cho phép
        return value
    
    raise ValidationError(
        self.field_name,
        f"'{value}' không khớp pattern '{pattern}'. "
        f"Ví dụ hợp lệ: {self._generate_example(pattern)}",
        value
    )

Whitelist approach - an toàn hơn

ALLOWED_TRACKING_PREFIXES = ["VH-", "GH-", "NJV-"] def validate_tracking_code(code: str) -> str: """Validation với whitelist prefix""" code = code.strip().upper() for prefix in ALLOWED_TRACKING_PREFIXES: if code.startswith(prefix): return code raise ValidationError( "tracking_code", f"Mã vận đơn phải bắt đầu bằng: {', '.join(ALLOWED_TRACKING_PREFIXES)}", code )

3. Lỗi "Required parameter missing" - Thiếu Tham Số Bắt Buộc

Mô tả: AI không trả