ในยุคที่ AI API กลายเป็นหัวใจสำคัญของแอปพลิเคชันสมัยใหม่ การจัดการโครงสร้างพื้นฐานอย่างมีประสิทธิภาพเป็นสิ่งจำเป็นอย่างยิ่ง บทความนี้จะพาคุณสำรวจวิธีใช้ Terraform ในการ provision และจัดการ AI API infrastructure พร้อมตัวอย่างจริงจาก 3 กรณีศึกษา

ทำไมต้องใช้ IaC กับ AI API

การใช้ Infrastructure as Code กับ AI API ช่วยให้เราสามารถ:

กรณีศึกษาที่ 1: ระบบ AI บริการลูกค้าอีคอมเมิร์ซ - รับมือ Traffic Spike

ร้านค้าออนไลน์ขนาดใหญ่ต้องเผชิญกับช่วง Flash Sale ที่ traffic พุ่งสูงถึง 10 เท่า การใช้ Terraform ช่วยให้สามารถ auto-scale infrastructure ได้อย่างราบรื่น

โครงสร้างโปรเจกต์

ecommerce-ai-terraform/
├── main.tf
├── variables.tf
├── outputs.tf
├── modules/
│   ├── api-gateway/
│   ├── redis-cache/
│   └── monitoring/
└── environments/
    ├── production/
    └── staging/

main.tf - การกำหนด Infrastructure หลัก

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
  required_version = ">= 1.5.0"
}

provider "aws" {
  region = var.aws_region
}


API Gateway สำหรับ AI Chat

module "ai_api_gateway" {
  source = "./modules/api-gateway"
  
  environment = var.environment
  api_key     = var.holysheep_api_key
  
  scaling_config = {
    min_capacity = var.min_capacity
    max_capacity = var.max_capacity
    target_cpu   = 70
  }
  
  tags = {
    Project     = "ecommerce-ai"
    Environment = var.environment
  }
}


Redis Cache สำหรับ Session Management

module "redis_cache" {
  source = "./modules/redis-cache"
  
  node_type           = "cache.r6g.large"
  num_cache_nodes     = var.redis_nodes
  engine_version      = "7.0"
  port                = 6379
  
  automatic_failover_enabled = var.environment == "production" ? true : false
  
  parameters = [
    {
      name  = "maxmemory-policy"
      value = "allkeys-lru"
    }
  ]
}


CloudWatch Monitoring

module "monitoring" {
  source = "./modules/monitoring"
  
  service_name = "ecommerce-ai-chat"
  alert_threshold = 5000  # requests per minute
}

AI Service - Chat Handler with HolySheep

#!/usr/bin/env python3
"""
E-commerce AI Customer Service Chat Handler
ใช้ HolySheep AI API สำหรับ response generation
"""

import os
import json
import redis
import httpx
from typing import Dict, Optional
from datetime import datetime

class EcommerceAIChat:
    def __init__(self):
        self.api_key = os.environ.get("HOLYSHEEP_API_KEY")
        self.base_url = "https://api.holysheep.ai/v1"
        
        # Redis connection for session management
        self.redis_client = redis.from_url(
            os.environ.get("REDIS_URL", "redis://localhost:6379"),
            decode_responses=True
        )
        
        self.http_client = httpx.AsyncClient(timeout=30.0)
    
    async def chat(self, session_id: str, user_message: str) -> Dict:
        """Process customer chat with AI"""
        
        # Get conversation history from Redis
        history = self._get_conversation_history(session_id)
        
        # Build context with product info
        system_prompt = """คุณคือพนักงานบริการลูกค้าอีคอมเมิร์ซชื่อ "ติ๊ก"
        - ให้บริการด้วยความเป็นมิตร
        - แนะนำสินค้าตามความต้องการ
        - ตอบสั้น กระชับ ไม่เกิน 3 ประโยค
        - ถ้าไม่แน่ใจ ให้บอกว่าจะตรวจสอบและตอบกลับ"""
        
        # Call HolySheep AI API
        response = await self._call_holysheep(
            system_prompt=system_prompt,
            messages=history + [{"role": "user", "content": user_message}]
        )
        
        # Save to conversation history
        self._save_message(session_id, "user", user_message)
        self._save_message(session_id, "assistant", response["content"])
        
        return {
            "response": response["content"],
            "tokens_used": response["usage"]["total_tokens"],
            "session_id": session_id,
            "timestamp": datetime.utcnow().isoformat()
        }
    
    async def _call_holysheep(self, system_prompt: str, messages: list) -> Dict:
        """เรียก HolySheep AI API - ราคาประหยัด 85%+"""
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "system", "content": system_prompt},
                *messages
            ],
            "temperature": 0.7,
            "max_tokens": 500
        }
        
        response = self.http_client.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload
        )
        response.raise_for_status()
        return response.json()
    
    def _get_conversation_history(self, session_id: str) -> list:
        """ดึงประวัติการสนทนาจาก Redis"""
        key = f"chat:session:{session_id}"
        history_str = self.redis_client.get(key)
        
        if history_str:
            return json.loads(history_str)
        return []
    
    def _save_message(self, session_id: str, role: str, content: str):
        """บันทึกข้อความลง Redis"""
        key = f"chat:session:{session_id}"
        history = self._get_conversation_history(session_id)
        
        history.append({"role": role, "content": content})
        
        # Keep last 10 messages
        if len(history) > 10:
            history = history[-10:]
        
        self.redis_client.setex(key, 3600, json.dumps(history))  # 1 hour TTL



AWS Lambda Handler

async def handler(event, context):
    chat = EcommerceAIChat()
    
    body = json.loads(event["body"])
    session_id = body.get("session_id", "anonymous")
    user_message = body.get("message", "")
    
    result = await chat.chat(session_id, user_message)
    
    return {
        "statusCode": 200,
        "body": json.dumps(result),
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*"
        }
    }

กรณีศึกษาที่ 2: RAG System ขนาดองค์กร

บริษัทที่ปรึกษาต้องการระบบ RAG (Retrieval-Augmented Generation) สำหรับค้นหาเอกสารภายใน 120,000 ฉบับ รองรับพนักงาน 500 คน

Terraform Module สำหรับ Vector Database

# modules/vector-db/main.tf
variable "environment" {}
variable "instance_class" {}

resource "aws_opensearch_domain" "vector_db" {
  domain_name    = "enterprise-rag-${var.environment}"
  engine_version = "OpenSearch_2.11"
  
  cluster_config {
    instance_type            = var.instance_class
    instance_count           = var.environment == "production" ? 3 : 1
    dedicated_master_enabled = var.environment == "production"
    dedicated_master_type    = "r6g.large.search"
    dedicated_master_count   = 3
    warm_enabled             = true
    warm_type                = "ultrawarm.medium.search"
  }
  
  ebs_options {
    ebs_enabled = true
    volume_type = "gp3"
    volume_size = var.environment == "production" ? 500 : 100
    throughput  = 250
  }
  
  node_to_node_encryption_options {
    enabled = true
  }
  
  encryption_at_rest_options {
    enabled    = true
    kms_key_id = aws_kms_key.opensearch.key_id
  }
  
  access_policies = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Principal = {
          AWS = "*"
        }
        Action = [
          "es:*"
        ]
        Resource = "${aws_opensearch_domain.vector_db.arn}/*"
        Condition = {
          IpAddress = {
            "aws:SourceIp" = var.allowed_ip_ranges
          }
        }
      }
    ]
  })
  
  log_publishing_options {
    log_type = "INDEX_SLOW_LOGS"
    cloudwatch_log_group_arn = aws_cloudwatch_log_group.opensearch_logs.arn
  }
}

resource "aws_kms_key" "opensearch" {
  description             = "KMS key for OpenSearch encryption"
  deletion_window_in_days = 10
  enable_key_rotation     = true
  
  tags = {
    Purpose = "RAG Vector Database Encryption"
  }
}

output "vector_db_endpoint" {
  value = aws_opensearch_domain.vector_db.endpoint
}

output "vector_db_arn" {
  value = aws_opensearch_domain.vector_db.arn
}

RAG Pipeline with HolySheep

#!/usr/bin/env python3
"""
Enterprise RAG Pipeline
ดึงเอกสารที่เกี่ยวข้อง + สร้างคำตอบด้วย HolySheep AI
"""

import boto3
import httpx
import json
from opensearchpy import OpenSearch
from typing import List, Dict
import os

class EnterpriseRAGPipeline:
    def __init__(self):
        self.opensearch = OpenSearch(
            hosts=[{
                'host': os.environ['OPENSEARCH_HOST'],
                'port': 443,
                'use_ssl': True
            }],
            http_auth=(os.environ['OPENSEARCH_USER'], 
                      os.environ['OPENSEARCH_PASS'])
        )
        
        self.api_key = os.environ['HOLYSHEEP_API_KEY']
        self.base_url = "https://api.holysheep.ai/v1"
        
        # Embedding model (ใช้ OpenSearch built-in หรือ external)
        self.embedding_url = f"{self.base_url}/embeddings"
    
    async def query(self, question: str, filters: Dict = None) -> Dict:
        """Query RAG system"""
        
        # 1. Generate embedding for question
        question_embedding = await self._get_embedding(question)
        
        # 2. Search OpenSearch for relevant documents
        search_results = self._search_documents(
            query_vector=question_embedding,
            filters=filters,
            max_results=5
        )
        
        # 3. Build context from retrieved documents
        context = self._build_context(search_results)
        
        # 4. Generate answer with HolySheep
        answer = await self._generate_answer(question, context)
        
        return {
            "answer": answer["content"],
            "sources": [
                {
                    "doc_id": doc["_id"],
                    "title": doc["_source"].get("title", "Untitled"),
                    "score": doc["_score"],
                    "snippet": doc["_source"].get("content", "")[:200]
                }
                for doc in search_results["hits"]["hits"]
            ],
            "metadata": {
                "model_used": "gpt-4.1",
                "total_tokens": answer["usage"]["total_tokens"],
                "retrieval_time_ms": 0  # วัดจริงใน production
            }
        }
    
    async def _get_embedding(self, text: str) -> List[float]:
        """Get text embedding from HolySheep"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "text-embedding-3-small",
            "input": text
        }
        
        async with httpx.AsyncClient() as client:
            response = await client.post(
                self.embedding_url,
                headers=headers,
                json=payload,
                timeout=10.0
            )
            response.raise_for_status()
            data = response.json()
            return data["data"][0]["embedding"]
    
    async def _generate_answer(self, question: str, context: str) -> Dict:
        """Generate answer using retrieved context"""
        
        system_prompt = """คุณคือผู้ช่วยค้นหาข้อมูลเอกสารองค์กร
        - ตอบคำถามโดยอ้างอิงจากเอกสารที่ให้มาเท่านั้น
        - ถ้าไม่มีข้อมูลในเอกสาร ให้ตอบว่า "ไม่พบข้อมูลที่เกี่ยวข้อง"
        - อ้างอิงแหล่งที่มาทุกครั้ง"""
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "gpt-4.1",
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": f"เอกสารที่เกี่ยวข้อง:\n{context}\n\nคำถาม: {question}"}
            ],
            "temperature": 0.3,  # ลด temperature สำหรับ factual QA
            "max_tokens": 1000
        }
        
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30.0
            )
            response.raise_for_status()
            return response.json()
    
    def _search_documents(self, query_vector: List[float], 
                         filters: Dict, max_results: int) -> Dict:
        """Search OpenSearch kNN"""
        
        query = {
            "size": max_results,
            "query": {
                "knn": {
                    "embedding": {
                        "vector": query_vector,
                        "k": max_results
                    }
                }
            },
            "_source": ["title", "content", "metadata", "created_at"]
        }
        
        # Add filters if provided
        if filters:
            query["query"] = {
                "bool": {
                    "must": [
                        {"knn": {"embedding": {"vector": query_vector, "k": max_results}}}
                    ],
                    "filter": [
                        {"term": {k: v}} for k, v in filters.items()
                    ]
                }
            }
        
        return self.opensearch.search(
            index="enterprise-docs",
            body=query
        )
    
    def _build_context(self, search_results: Dict) -> str:
        """Build context string from search results"""
        context_parts = []
        
        for i, hit in enumerate(search_results["hits"]["hits"], 1):
            source = hit["_source"]
            context_parts.append(
                f"[เอกสาร {i}] {source.get('title', 'Untitled')}\n"
                f"{source.get('content', '')}\n"
            )
        
        return "\n---\n".join(context_parts)

กรณีศึกษาที่ 3: โปรเจกต์นักพัฒนาอิสระ - SaaS AI Writing Tool

นักพัฒนาอิสระสร้าง SaaS สำหรับ AI Writing Assistant ใช้ HolySheep AI ร่วมกับ multi-tenant architecture ประหยัดค่าใช้จ่ายได้มาก

Multi-tenant Terraform Setup

# environments/shared/modules/multi-tenant-api/main.tf
variable "tenant_id" {}
variable "plan_tier" {}

locals {
  instance_sizes = {
    "free"     = { memory = 512, cpu = 0.5 }
    "starter"  = { memory = 1024, cpu = 1 }
    "pro"      = { memory = 2048, cpu = 2 }
    "enterprise" = { memory = 4096, cpu = 4 }
  }
  
  selected_plan = local.instance_sizes[var.plan_tier]
}

resource "aws_lambda_function" "tenant_api" {
  function_name = "ai-writing-${var.tenant_id}"
  
  runtime = "python3.11"
  handler = "handler.handler"
  
  filename

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง