ในยุคที่ LLM API กลายเป็นหัวใจสำคัญของแอปพลิเคชัน AI การทำ Contract Testing จึงไม่ใช่ทางเลือกอีกต่อไป แต่เป็นความจำเป็นเชิงกลยุทธ์ บทความนี้จะพาคุณเจาะลึกว่า Contract Testing คืออะไร ทำไมมันสำคัญสำหรับ LLM Services และแนวทางปฏิบัติจริงที่ใช้งานได้ทันทีกับ HolySheep AI

ทำไมต้องสนใจราคา API ก่อนเริ่ม Project

ก่อนจะลงลึกเรื่องเทคนิค มาดูตัวเลขสำคัญที่จะกำหนด Architecture ของคุณ:

โมเดลราคา Output (2026)ต้นทุน 10M tokens/เดือน
GPT-4.1$8/MTok$80
Claude Sonnet 4.5$15/MTok$150
Gemini 2.5 Flash$2.50/MTok$25
DeepSeek V3.2$0.42/MTok$4.20

ข้อค้นพบสำคัญ: DeepSeek V3.2 มีต้นทุนต่ำกว่า GPT-4.1 ถึง 19 เท่า สำหรับปริมาณงาน 10M tokens/เดือน นี่คือเหตุผลว่าทำไมการทำ Contract Testing กับ Multi-Provider จึงคุ้มค่าอย่างยิ่ง — คุณสามารถสลับ Provider ตาม Use Case และประหยัดงบประมาณได้มหาศาล

Contract Testing คืออะไร และทำไม LLM ถึงต้องการมัน

Contract Testing คือเทคนิคการทดสอบที่ตรวจสอบว่า API ระหว่าง Consumer และ Provider ยังคง "ตกลงกัน" อยู่ โดยไม่ต้องรัน Integration Test ทั้งระบบ ในบริบทของ LLM Services มันหมายถึง:

การตั้งค่า Environment สำหรับ HolySheep AI

ก่อนเริ่มเขียน Test เราต้องตั้งค่า Client ที่รองรับ Contract Testing อย่างถูกต้อง:

# requirements.txt

openapi-schema-validator==0.6.2

pact-python==1.4.4

jsonschema==4.21.1

httpx==0.27.0

pytest==8.0.0

pytest-asyncio==0.23.4

import os class HolySheepConfig: """Configuration สำหรับ HolySheep AI API""" BASE_URL = "https://api.holysheep.ai/v1" API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY") # Timeout settings (ms) CONNECT_TIMEOUT = 5000 # 5 วินาที READ_TIMEOUT = 30000 # 30 วินาที # Retry settings MAX_RETRIES = 3 RETRY_DELAY = 1000 # มิลลิวินาที # Model defaults DEFAULT_MODEL = "deepseek-v3.2" TEMPERATURE_RANGE = (0.0, 2.0) MAX_TOKENS_RANGE = (1, 128000) @classmethod def to_dict(cls) -> dict: return { "base_url": cls.BASE_URL, "timeout": { "connect": cls.CONNECT_TIMEOUT, "read": cls.READ_TIMEOUT }, "max_retries": cls.MAX_RETRIES }

ตรวจสอบ Configuration

print(f"HolySheep Base URL: {HolySheepConfig.BASE_URL}") print(f"Default Model: {HolySheepConfig.DEFAULT_MODEL}") print(f"Latency Target: <{HolySheepConfig.CONNECT_TIMEOUT}ms")

Schema Contract สำหรับ LLM Responses

สร้าง Schema ที่กำหนด Contract ระหว่าง Provider กับ Consumer:

"""
LLM API Contract Schema Definitions
Compatible กับ OpenAI API format (ซึ่ง HolySheep รองรับ)
"""

from pydantic import BaseModel, Field, field_validator
from typing import Optional, List, Dict, Any
from enum import Enum

class FinishReason(str, Enum):
    STOP = "stop"
    LENGTH = "length"
    CONTENT_FILTER = "content_filter"
    TOOL_CALLS = "tool_calls"

class Message(BaseModel):
    """Chat Message Structure"""
    role: str = Field(..., pattern="^(system|user|assistant|tool)$")
    content: str
    name: Optional[str] = None
    tool_call_id: Optional[str] = None

class Usage(BaseModel):
    """Token Usage Information"""
    prompt_tokens: int = Field(ge=0)
    completion_tokens: int = Field(ge=0)
    total_tokens: int = Field(ge=0)
    
    # Provider-specific fields (optional)
    prompt_cache_hits: Optional[int] = None
    prompt_cache_misses: Optional[int] = None
    
    @field_validator('total_tokens')
    @classmethod
    def total_must_match_sum(cls, v, info):
        # Allow some tolerance for rounding
        if 'prompt_tokens' in info.data and 'completion_tokens' in info.data:
            expected = info.data['prompt_tokens'] + info.data['completion_tokens']
            if abs(v - expected) > 5:  # 5 token tolerance
                raise ValueError(f"total_tokens ({v}) should approximately equal sum of prompt ({info.data['prompt_tokens']}) and completion ({info.data['completion_tokens']})")
        return v

class ChatCompletionResponse(BaseModel):
    """Standard Chat Completion Response Contract"""
    id: str
    object: str = "chat.completion"
    created: int = Field(..., ge=0)  # Unix timestamp
    model: str
    choices: List[Dict[str, Any]] = Field(min_length=1)
    usage: Usage
    service_tier: Optional[str] = None
    system_fingerprint: Optional[str] = None
    
    # Validate at least one choice exists
    @field_validator('choices')
    @classmethod
    def must_have_at_least_one_choice(cls, v):
        if not v:
            raise ValueError("choices must not be empty")
        return v

class StreamChunk(BaseModel):
    """Streaming Response Chunk Structure"""
    id: str
    object: str = "chat.completion.chunk"
    created: int = Field(..., ge=0)
    model: str
    choices: List[Dict[str, Any]]
    
    @field_validator('choices')
    @classmethod
    def validate_chunk_choice(cls, v):
        if not v:
            raise ValueError("Stream chunk must have at least one choice")
        choice = v[0]
        if 'index' not in choice:
            raise ValueError("Choice must have 'index' field")
        return v

class ErrorResponse(BaseModel):
    """Standard Error Response Contract"""
    error: Dict[str, Any] = Field(
        ...,
        description="Error object containing code, message, and optional details"
    )
    
    @field_validator('error')
    @classmethod
    def validate_error_structure(cls, v):
        required_fields = ['message', 'type']
        for field in required_fields:
            if field not in v:
                raise ValueError(f"Error must contain '{field}' field")
        return v

Export schemas for use in tests

LLM_CONTRACT_SCHEMAS = { "chat_completion": ChatCompletionResponse, "stream_chunk": StreamChunk, "error": ErrorResponse, "usage": Usage, "message": Message } def validate_response(response_data: dict, schema_type: str) -> tuple[bool, Optional[str]]: """ Validate response against contract schema Returns: (is_valid, error_message) """ schema = LLM_CONTRACT_SCHEMAS.get(schema_type) if not schema: return False, f"Unknown schema type: {schema_type}" try: schema(**response_data) return True, None except Exception as e: return False, str(e)

การเขียน Contract Test ด้วย pytest

นี่คือหัวใจของบทความ — Contract Test ที่ใช้งานได้จริง:

"""
LLM API Contract Testing Suite
สำหรับ HolySheep AI และ Multi-Provider Support
"""

import pytest
import httpx
import asyncio
import json
from typing import Generator, Dict, Any
from pact import Consumer, Provider, Term, Like, EachLike
import time

from llm_contract_schema import (
    HolySheepConfig,
    LLM_CONTRACT_SCHEMAS,
    validate_response,
    ChatCompletionResponse,
    StreamChunk
)

class HolySheepClient:
    """Client สำหรับ HolySheep AI API พร้อม Contract Validation"""
    
    def __init__(self, api_key: str = None):
        self.base_url = HolySheepConfig.BASE_URL
        self.api_key = api_key or HolySheepConfig.API_KEY
        self.timeout = httpx.Timeout(
            connect=HolySheepConfig.CONNECT_TIMEOUT / 1000,
            read=HolySheepConfig.READ_TIMEOUT / 1000
        )
    
    def _get_headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
    
    async def chat_completion(
        self,
        messages: list,
        model: str = None,
        **kwargs
    ) -> Dict[str, Any]:
        """ส่ง Chat Completion Request และ Return Response"""
        model = model or HolySheepConfig.DEFAULT_MODEL
        
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers=self._get_headers(),
                json={
                    "model": model,
                    "messages": messages,
                    **kwargs
                }
            )
            
            if response.status_code != 200:
                # Validate error response contract
                is_valid, error = validate_response(
                    response.json(),
                    "error"
                )
                if not is_valid:
                    pytest.fail(f"Invalid error response contract: {error}")
                return {"error": response.json(), "status_code": response.status_code}
            
            return response.json()
    
    async def stream_chat(
        self,
        messages: list,
        model: str = None,
        **kwargs
    ) -> Generator[Dict[str, Any], None, None]:
        """Streaming Chat Completion"""
        model = model or HolySheepConfig.DEFAULT_MODEL
        
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            async with client.stream(
                "POST",
                f"{self.base_url}/chat/completions",
                headers=self._get_headers(),
                json={
                    "model": model,
                    "messages": messages,
                    "stream": True,
                    **kwargs
                }
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        data = line[6:]  # Remove "data: " prefix
                        if data == "[DONE]":
                            break
                        
                        chunk = json.loads(data)
                        
                        # Validate stream chunk contract
                        is_valid, error = validate_response(chunk, "stream_chunk")
                        if not is_valid:
                            pytest.fail(f"Invalid stream chunk contract: {error}")
                        
                        yield chunk


==================== PACT Contract Tests ====================

@pytest.fixture(scope="module") def pact_config(): """Pact Configuration สำหรับ HolySheep AI""" return { "consumer": {"name": "llm-app-consumer"}, "provider": {"name": "holysheep-ai"}, "broker_url": "http://localhost:8000", "publish_verification_results": False } @pytest.fixture def holy_sheep_client(): """Client fixture for testing""" return HolySheepClient() class TestHolySheepContract: """Contract Tests สำหรับ HolySheep AI API""" @pytest.mark.asyncio async def test_chat_completion_response_structure(self, holy_sheep_client): """ Test: Response ต้องมีโครงสร้างตาม Contract Contract: ChatCompletionResponse schema """ messages = [{"role": "user", "content": "Hello, test message"}] response = await holy_sheep_client.chat_completion( messages=messages, max_tokens=50, temperature=0.7 ) # ตรวจสอบว่าไม่มี error assert "error" not in response, f"API Error: {response.get('error')}" # Validate contract is_valid, error = validate_response(response, "chat_completion") assert is_valid, f"Contract violation: {error}" # ตรวจสอบ required fields assert "id" in response assert "choices" in response assert "usage" in response assert len(response["choices"]) > 0 # ตรวจสอบ usage calculation usage = response["usage"] assert usage["total_tokens"] == usage["prompt_tokens"] + usage["completion_tokens"] print(f"✅ Response structure valid: {response['id']}") print(f" Tokens used: {usage['total_tokens']}") @pytest.mark.asyncio async def test_streaming_response_contract(self, holy_sheep_client): """ Test: Streaming chunks ต้องมีโครงสร้างถูกต้อง Contract: StreamChunk schema """ messages = [{"role": "user", "content": "Count to 3"}] chunks_received = [] async for chunk in holy_sheep_client.stream_chat( messages=messages, max_tokens=30 ): chunks_received.append(chunk) # ทุก chunk ต้องมีโครงสร้างที่ถูกต้อง assert "choices" in chunk assert len(chunk["choices"]) > 0 # delta content อาจว่างเปล่าได้ แต่ต้องมี field choice = chunk["choices"][0] assert "delta" in choice assert "index" in choice # ตรวจสอบว่าได้รับ chunks assert len(chunks_received) > 0, "No streaming chunks received" # ตรวจสอบว่า chunk แรกมี finish_reason เป็น None first_chunk = chunks_received[0] assert first_chunk["choices"][0].get("finish_reason") is None # ตรวจสอบ chunk สุดท้ายต้องมี finish_reason last_chunk = chunks_received[-1] assert last_chunk["choices"][0].get("finish_reason") is not None print(f"✅ Received {len(chunks_received)} valid stream chunks") @pytest.mark.asyncio async def test_model_selection_affects_response(self, holy_sheep_client): """ Test: ต่าง model ต้องให้ response ที่แตกต่างกัน """ messages = [{"role": "user", "content": "What is 2+2?"}] # Test DeepSeek V3.2 response_deepseek = await holy_sheep_client.chat_completion( messages=messages, model="deepseek-v3.2", max_tokens=20 ) # Test Gemini Flash response_gemini = await holy_sheep_client.chat_completion( messages=messages, model="gemini-2.5-flash", max_tokens=20 ) # ทั้งสองต้อง valid contract is_valid_ds, _ = validate_response(response_deepseek, "chat_completion") is_valid_gm, _ = validate_response(response_gemini, "chat_completion") assert is_valid_ds, "DeepSeek response violates contract" assert is_valid_gm, "Gemini response violates contract" # model field ต้องตรงกับที่ request assert response_deepseek["model"] == "deepseek-v3.2" assert response_gemini["model"] == "gemini-2.5-flash" print(f"✅ Multi-model support validated") print(f" DeepSeek model: {response_deepseek['model']}") print(f" Gemini model: {response_gemini['model']}") class TestProviderCompatibility: """Tests สำหรับตรวจสอบ Compatibility ระหว่าง Providers""" @pytest.mark.asyncio async def test_openai_compatible_format(self, holy_sheep_client): """ Test: HolySheep ต้องรองรับ OpenAI-compatible format """ # OpenAI-style request payload = { "model": "deepseek-v3.2", "messages": [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Explain quantum computing."} ], "temperature": 0.8, "max_tokens": 100, "top_p": 0.95, "frequency_penalty": 0.5, "presence_penalty": 0.3 } async with httpx.AsyncClient(timeout=30) as client: response = await client.post( f"{holy_sheep_client.base_url}/chat/completions", headers=holy_sheep_client._get_headers(), json=payload ) assert response.status_code == 200 data = response.json() # Validate OpenAI-compatible response assert "id" in data assert "object" in data assert data["object"] == "chat.completion" assert "choices" in data assert "usage" in data print(f"✅ OpenAI-compatible format validated") print(f" Response ID: {data['id']}") print(f" Object type: {data['object']}")

==================== Performance Contract Tests ====================

class TestPerformanceContracts: """Tests สำหรับ Performance Requirements""" @pytest.mark.asyncio async def test_latency_requirement(self, holy_sheep_client): """ Test: Response time ต้อง < 50ms (ตาม SLA ของ HolySheep) """ messages = [{"role": "user", "content": "Hi"}] start_time = time.time() response = await holy_sheep_client.chat_completion( messages=messages, max_tokens=10 ) end_time = time.time() latency_ms = (end_time - start_time) * 1000 # HolySheep รับประกัน < 50ms assert latency_ms < 50, f"Latency {latency_ms:.2f}ms exceeds 50ms SLA" print(f"✅ Latency requirement met: {latency_ms:.2f}ms (< 50ms)") @pytest.mark.asyncio async def test_cost_per_token_accuracy(self, holy_sheep_client): """ Test: Token count ต้องถูกต้องสำหรับ Cost Calculation """ messages = [{"role": "user", "content": "Say 'test'"}] response = await holy_sheep_client.chat_completion( messages=messages, max_tokens=5 ) usage = response["usage"] # ตรวจสอบความถูกต้องของ token count assert usage["total_tokens"] > 0 assert usage["total_tokens"] >= usage["prompt_tokens"] assert usage["total_tokens"] >= usage["completion_tokens"] # คำนวณค่าใช้จ่าย (DeepSeek V3.2: $0.42/MTok) cost_per_token = 0.42 / 1_000_000 # $ per token estimated_cost = usage["total_tokens"] * cost_per_token print(f"✅ Token count validated") print(f" Prompt tokens: {usage['prompt_tokens']}") print(f" Completion tokens: {usage['completion_tokens']}") print(f" Total tokens: {usage['total_tokens']}") print(f" Estimated cost (DeepSeek V3.2): ${estimated_cost:.6f}")

การรัน Contract Tests

สร้าง Makefile หรือ run script สำหรับการรัน tests:

#!/bin/bash

run_contract_tests.sh

set -e echo "==========================================" echo "LLM API Contract Testing Suite" echo "=========================================="

Set environment variables

export HOLYSHEEP_API_KEY="${HOLYSHEEP_API_KEY:-YOUR_HOLYSHEEP_API_KEY}" export PACT_BROKER_URL="http://localhost:8000"

Run pytest with coverage

echo "" echo "Running Contract Tests..." pytest \ --verbose \ --tb=short \ --asyncio-mode=auto \ tests/test_llm_contract.py \ -v \ --cov=llm_contract \ --cov-report=term-missing

Generate Pact contract files

echo "" echo "Publishing Contracts to Broker..." pact-broker publish \ --consumer-app-version="$(git rev-parse --short HEAD)" \ --build-url="${CI_BUILD_URL}" \ ./pacts echo "" echo "==========================================" echo "Contract Tests Complete!" echo "=========================================="

Run specific test categories

run_unit_tests() { echo "Running Unit Contract Tests..." pytest tests/ -k "test_chat_completion" -v } run_performance_tests() { echo "Running Performance Contract Tests..." pytest tests/ -k "test_latency or test_cost" -v } run_compatibility_tests() { echo "Running Provider Compatibility Tests..." pytest tests/test_provider_compatibility.py -v }

Execute all test suites

run_unit_tests run_performance_tests run_compatibility_tests

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. Error 401 Unauthorized — API Key ไม่ถูกต้อง

สัญญาณ: {"error":{"message":"Incorrect API key provided","type":"invalid_request_error","code":"invalid_api_key"}}

สาเหตุ: API Key ไม่ตรงกับที่ลงทะเบียนไว้ หรือใส่ Base URL ผิด

# ❌ วิธีผิด - ใช้ OpenAI URL แทน HolySheep
client = OpenAI(
    api_key="your-key",
    base_url="https://api.openai.com/v1"  # ผิด!
)

✅ วิธีถูก - ใช้ HolySheep URL

client = OpenAI( api_key="YOUR_HOLYSHEEP_API_KEY", base_url="https://api.holysheep.ai/v1" # ถูกต้อง! )

หรือใช้ HolySheepClient โดยตรง

from llm_client import HolySheepClient client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")

ตรวจสอบ API Key

import os assert os.environ.get("HOLYSHEEP_API_KEY"), "HOLYSHEEP_API_KEY not set"

2. Error 400 Bad Request — Schema Mismatch

สัญญาณ: {"error":{"message":"Invalid request","type":"invalid_request_error","param":null}}

สาเหตุ: Request Body ไม่ตรงตาม Schema ที่กำหนด เช่น temperature เกินขอบเขต

# ❌ วิธีผิด - Temperature เกิน 2.0
response = await client.chat.completions.create(
    model="deepseek-v3.2",
    messages=messages,
    temperature=3.0  # เกิน max (2.0)
)

✅ วิธีถูก - ใช้ค่าที่อยู่ในขอบเขต

response = await client.chat.completions.create( model="deepseek-v3.2", messages=messages, temperature=0.7, # ค่าปกติ max_tokens=1000 # ต้อง <= 128000 สำหรับ DeepSeek )

ตรวจสอบ Schema ก่อนส่ง request

from pydantic import BaseModel, Field, field_validator class ChatRequest(BaseModel): model: str messages: list temperature: float = Field(default=0.7, ge=0.0, le=2.0) max_tokens: int = Field(default=100, ge=1, le=128000) @field_validator('messages') @classmethod def validate_messages(cls, v): if not v: raise ValueError("messages cannot be empty") for msg in v: if msg.get('role') not in ['system', 'user', 'assistant', 'tool']: raise ValueError(f"Invalid role: {msg.get('role')}") return v def validate_request(messages: list, **kwargs) -> bool: """Validate request before sending""" try: ChatRequest(messages=messages, **kwargs) return True except Exception as e: print(f"Request validation failed: {e}") return False

3. Timeout Error — Response ใช้เวลานานเกินไป

สัญญาณ: httpx.ReadTimeout: HTTPX read timeout exceeded

สาเหตุ: Default timeout สั้นเกินไปสำหรับ Complex Requests หรือ High Load

# ❌ วิธีผิด - Timeout เป็น 5 วินาที (สำหรับ complex request ไม่พอ)
client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    timeout=5.0  # น้อยเกินไป!
)

✅ วิธีถูก - แบ่ง Timeout ตามประเภท Request

import httpx

Simple request (สั้น)

simple_timeout = httpx.Timeout( connect=5.0, read=10.0, write=5.0, pool=5.0 )

Complex request (ยาว)

complex_timeout = httpx.Timeout( connect=10.0, read=60.0, # 60 วินาทีสำหรับ long response write=10.0, pool=10.0 )

Streaming request

stream_timeout = httpx.Timeout( connect=10.0, read=120.0, # Long-running stream write=10.0 )

เลือก timeout ตาม request type

def get_client_for_request(is_streaming: bool, is_complex: bool): if is_streaming: timeout = stream_timeout elif is_complex: timeout = complex_timeout else: timeout = simple_timeout return OpenAI( api_key="YOUR_HOLYSHEEP_API_KEY", base_url="https://api.holysheep.ai/v1", timeout=timeout )

หรือ Retry with exponential backoff

from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10) ) async def chat_with_retry(messages, model="deepseek-v3.2"): async with httpx.AsyncClient(timeout=complex_timeout) as client: response = await client.post( "https://api.holysheep.ai/v1/chat/completions", headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"}, json={"model": model, "messages": messages} ) return response

4. Streaming Parse Error — อ่าน Response ไม่ได้

สัญญาณ: json.JSONDecodeError: Expecting value: line