ในยุคที่ AI API กลายเป็นหัวใจสำคัญของแอปพลิเคชันสมัยใหม่ การจัดการโครงสร้างพื้นฐานอย่างมีประสิทธิภาพเป็นสิ่งจำเป็นอย่างยิ่ง บทความนี้จะพาคุณสำรวจการใช้ Pulumi IaC เพื่อสร้าง ปรับใช้ และจัดการ AI API infrastructure อย่างเป็นระบบ พร้อมโค้ดตัวอย่างระดับ production ที่ใช้งานได้จริง รวมถึงการเชื่อมต่อกับ HolySheep AI ผู้ให้บริการ API คุณภาพสูงราคาประหยัด

Pulumi IaC คืออะไรและทำไมต้องใช้กับ AI API

Pulumi เป็น Infrastructure as Code (IaC) tool ที่ใช้ภาษาโปรแกรมทั่วไป เช่น Python, TypeScript, Go ในการกำหนดและจัดการโครงสร้างพื้นฐาน ต่างจาก Terraform ที่ใช้ HCL ทำให้สามารถใช้ความสามารถของภาษาโปรแกรมได้อย่างเต็มที่ สำหรับ AI API infrastructure การใช้ Pulumi ช่วยให้การ deploy, scale, และ monitor ระบบทำได้อย่างเป็นระบบและทำซ้ำได้

สถาปัตยกรรมระบบ AI API ด้วย Pulumi

1. โครงสร้างโปรเจกต์พื้นฐาน

ai-api-infra/
├── Pulumi.yaml
├── Pulumi.dev.yaml
├── Pulumi.prod.yaml
├── __main__.py
├── config.py
├── requirements.txt
└── modules/
    ├── __init__.py
    ├── api_gateway.py
    ├── compute.py
    ├── database.py
    ├── monitoring.py
    └── holyseep_client.py

2. การตั้งค่า Configuration และ API Client

import pulumi
import pulumi_aws as aws
from pulumi import Config, Output, export

class HolySheepConfig:
    """Configuration สำหรับ HolySheep AI API"""
    
    def __init__(self):
        self.config = Config()
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = self.config.require_secret("holyseep-api-key")
        self.models = {
            "gpt41": "gpt-4.1",
            "claude_sonnet": "claude-sonnet-4.5",
            "gemini_flash": "gemini-2.5-flash",
            "deepseek": "deepseek-v3.2"
        }
    
    def get_endpoint_url(self, model: str) -> str:
        return f"{self.base_url}/chat/completions"
    
    def get_pricing(self, model: str) -> dict:
        """ราคาเป็น USD ต่อ Million Tokens"""
        pricing = {
            "gpt-4.1": 8.0,
            "claude-sonnet-4.5": 15.0,
            "gemini-2.5-flash": 2.50,
            "deepseek-v3.2": 0.42
        }
        return pricing.get(model, 0.0)

config = HolySheepConfig()

3. การสร้าง API Gateway และ Load Balancer

import pulumi_aws as aws
from pulumi import ResourceOptions, export

class APIGateway:
    """API Gateway พร้อม rate limiting และ caching"""
    
    def __init__(self, name: str, vpc, tags: dict):
        self.name = name
        self.vpc = vpc
        self.tags = tags
        self._create_gateway()
    
    def _create_gateway(self):
        # Application Load Balancer
        self.alb = aws.lb.LoadBalancer(
            f"{self.name}-alb",
            name=f"{self.name}-alb",
            internal=False,
            load_balancer_type="application",
            security_groups=[self.vpc.sg.id],
            subnets=self.vpc.public_subnets,
            tags=self.tags
        )
        
        # Target Group สำหรับ API servers
        self.target_group = aws.lb.TargetGroup(
            f"{self.name}-tg",
            name=f"{self.name}-tg",
            port=8000,
            protocol="HTTP",
            target_type="ip",
            vpc_id=self.vpc.vpc.id,
            health_check=aws.lb.TargetGroupHealthCheckArgs(
                path="/health",
                interval=30,
                timeout=5,
                healthy_threshold=2,
                unhealthy_threshold=3
            ),
            deregistration_delay=30
        )
        
        # Listener พร้อม SSL
        self.listener = aws.lb.Listener(
            f"{self.name}-listener",
            load_balancer_arn=self.alb.arn,
            port=443,
            protocol="HTTPS",
            ssl_policy="ELBSecurityPolicy-TLS13-1-2-2021-06",
            certificate_arn=self._get_certificate(),
            default_actions=[aws.lb.ListenerDefaultActionArgs(
                type="forward",
                target_group_arn=self.target_group.arn
            )]
        )
        
        export("api_gateway_url", self.alb.dns_name)
        export("api_gateway_arn", self.alb.arn)
    
    def _get_certificate(self) -> str:
        # ACM Certificate สำหรับ domain
        cert = aws.acm.Certificate(
            f"{self.name}-cert",
            domain_name="api.yourdomain.com",
            validation_method="DNS",
            subject_alternative_names=["*.api.yourdomain.com"]
        )
        return cert.arn


การใช้งาน

api_gateway = APIGateway("production", vpc, tags)

4. Auto Scaling Group สำหรับ AI API Server

class AIAPIServer:
    """Auto Scaling Group สำหรับ AI API Server พร้อม GPU support"""
    
    def __init__(self, name: str, vpc, target_group, config: HolySheepConfig):
        self.name = name
        self.vpc = vpc
        self.target_group = target_group
        self.config = config
        self._create_launch_template()
        self._create_asg()
    
    def _create_launch_template(self):
        # IAM Role สำหรับ EC2
        self.role = aws.iam.Role(
            f"{self.name}-role",
            name=f"{self.name}-role",
            assume_role_policy="""{
                "Version": "2012-10-17",
                "Statement": [{
                    "Action": "sts:AssumeRole",
                    "Effect": "Allow",
                    "Principal": {"Service": "ec2.amazonaws.com"}
                }]
            }"""
        )
        
        # Instance Profile
        self.instance_profile = aws.iam.InstanceProfile(
            f"{self.name}-profile",
            name=f"{self.name}-profile",
            role=self.role.name
        )
        
        # Launch Template
        self.launch_template = aws.ec2.LaunchTemplate(
            f"{self.name}-lt",
            name=f"{self.name}-lt",
            image_id="ami-0c55b159cbfafe1f0",  # Amazon Linux 2023
            instance_type="t3.medium",
            iam_instance_profile=aws.ec2.LaunchTemplateIamInstanceProfileArgs(
                name=self.instance_profile.name
            ),
            vpc_security_group_ids=[self.vpc.sg.id],
            user_data=self._get_user_data(),
            monitoring=aws.ec2.LaunchTemplateMonitoringArgs(
                enabled=True
            ),
            metadata_options=aws.ec2.LaunchTemplateMetadataOptionsArgs(
                http_endpoint="enabled",
                http_tokens="required"
            ),
            tag_specifications=[
                aws.ec2.LaunchTemplateTagSpecificationArgs(
                    resource_type="instance",
                    tags={"Name": f"{self.name}-instance"}
                )
            ]
        )
    
    def _get_user_data(self) -> str:
        return """#!/bin/bash
yum update -y
yum install -y docker python3-pip
systemctl start docker
systemctl enable docker


ติดตั้ง Python dependencies

pip3 install fastapi uvicorn boto3 aiohttp redis
pip3 install pulumi pulumi-aws


สร้าง application directory

mkdir -p /app
cd /app


ดาวน์โหลด application code

aws s3 cp s3://your-bucket/ai-api-latest.tar.gz /tmp/
tar -xzf /tmp/ai-api-latest.tar.gz -C /app/


ตั้งค่า environment

cat > /app/.env << 'EOF'
HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
REDIS_HOST=${REDIS_HOST}
LOG_LEVEL=info
EOF


Start container

docker run -d --name ai-api \\
  --restart unless-stopped \\
  -p 8000:8000 \\
  --env-file /app/.env \\
  ai-api:latest
"""
    
    def _create_asg(self):
        # Auto Scaling Group
        self.asg = aws.autoscaling.Group(
            f"{self.name}-asg",
            name=f"{self.name}-asg",
            vpc_zone_identifiers=self.vpc.private_subnets,
            launch_template=aws.autoscaling.GroupLaunchTemplateArgs(
                id=self.launch_template.id,
                version="$Latest"
            ),
            min_size=2,
            max_size=10,
            desired_capacity=4,
            health_check_type="ELB",
            health_check_grace_period=60,
            default_cooldown=300,
            suspension_states=[],
            tags=[
                aws.autoscaling.GroupTagArgs(
                    key="Name",
                    value=f"{self.name}-instance",
                    propagate_at_launch=True
                ),
                aws.autoscaling.GroupTagArgs(
                    key="Environment",
                    value="production",
                    propagate_at_launch=True
                )
            ],
            lifecycle_hook_specifications=[aws.autoscaling.GroupLifecycleHookSpecificationArgs(
                name="healthcheck-hook",
                lifecycle_transition="autoscaling:EC2_INSTANCE_LAUNCHING",
                heartbeat_timeout=300,
                notification_target=aws.autoscaling.GroupLifecycleHookSpecificationNotificationTargetArgs(
                    target_arn=self.sns_topic.arn
                )
            )]
        )
        
        # Scaling Policy
        self.scale_up_policy = aws.autoscaling.Policy(
            f"{self.name}-scale-up",
            name=f"{self.name}-scale-up",
            scaling_adjustment=2,
            adjustment_type="ChangeInCapacity",
            cooldown=300,
            autoscaling_group_name=self.asg.name
        )
        
        # CloudWatch Alarm สำหรับ Scale Up
        self.cpu_alarm_high = aws.cloudwatch.MetricAlarm(
            f"{self.name}-cpu-high",
            name=f"{self.name}-cpu-high",
            comparison_operator="GreaterThanThreshold",
            evaluation_periods=2,
            metric_name="CPUUtilization",
            namespace="AWS/EC2",
            period=60,
            statistic="Average",
            threshold=70,
            alarm_description="Scale up when CPU > 70%",
            alarm_actions=[self.scale_up_policy.arn]
        )
        
        export("asg_name", self.asg.name)
        export("desired_capacity", self.asg.desired_capacity)

การจัดการ Concurrent Requests และ Rate Limiting

import asyncio
import aiohttp
from typing import List, Dict, Optional
from dataclasses import dataclass
import time
import redis.asyncio as redis

@dataclass
class RateLimiterConfig:
    """Configuration สำหรับ Rate Limiter"""
    requests_per_minute: int = 60
    requests_per_second: int = 10
    burst_size: int = 20
    model_limits: Dict[str, Dict[str, int]] = None

class HolySheepAPIClient:
    """Async client สำหรับ HolySheep AI พร้อม rate limiting"""
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        rate_config: RateLimiterConfig = None
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.rate_config = rate_config or RateLimiterConfig()
        self.redis_client: Optional[redis.Redis] = None
        self._semaphore = asyncio.Semaphore(self.rate_config.burst_size)
        self._last_request_time = 0
        self._min_interval = 1.0 / self.rate_config.requests_per_second
        self._request_counts: Dict[str, List[float]] = {}
    
    async def __aenter__(self):
        self.redis_client = redis.from_url(
            "redis://localhost:6379",
            encoding="utf-8",
            decode_responses=True

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