Modal 서버리스 GPU 컴퓨팅 — 인프라 없이 ML 파이프라인 실행

import modal

# PyTorch와 CUDA가 사전 설치된 컨테이너 이미지 정의
stub = modal.Stub("my-modal-app")

image = modal.Image.debian_slim().pip_install(
    "torch",
    "transformers",
    "accelerate"
)

# Modal Python SDK 설치
pip install modal-client

# Modal 계정으로 인증
modal setup

import modal
from transformers import AutoModelForCausalLM, AutoTokenizer

stub = modal.Stub("llm-inference")

# 컨테이너 시작 시 모델 한 번만 로드
@stub.cls(
    image=modal.Image.debian_slim().pip_install("transformers", "torch", "accelerate"),
    gpu="A10G",
    memory=8192
)
class LLMEndpoint:
    @modal.enter()
    def load_model(self):
        self.model = AutoModelForCausalLM.from_pretrained(
            "meta-llama/Llama-3.2-3B-Instruct",
            torch_dtype="auto",
            device_map="auto"
        )
        self.tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3.2-3B-Instruct")

    @modal.method()
    def generate(self, prompt: str, max_tokens: int = 512) -> str:
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
        outputs = self.model.generate(**inputs, max_new_tokens=max_tokens)
        return self.tokenizer.decode(outputs[0], skip_special_tokens=True)

# 앱을 Modal 클라우드에 배포
modal deploy my_app.py

# 명령줄에서 테스트
modal run my_app::LLMEndpoint.generate --prompt "양자 컴퓨팅 설명해줘" --max_tokens 256

@stub.cls(
    gpu="L4",
    concurrency_limit=20,
    allow_concurrent_inputs=10,
    keep_warm=2  # 최소 2개 컨테이너 웜 상태로 유지
)
class ProductionLLM:
    @modal.enter()
    def load_model(self):
        self.model = load_optimized_model()
        self.tokenizer = AutoTokenizer.from_pretrained("your-model")

    @modal.web_endpoint(method="POST")
    def infer(self, req: dict):
        prompt = req.get("prompt", "")
        result = self.model.generate(prompt, max_tokens=req.get("max_tokens", 256))
        return {"response": result}

@stub.function(
    image=image,
    gpu="A100-80GB",
    timeout=3600,  # 최대 1시간
    retries=2
)
def batch_embed(docs: list[str]) -> list[list[float]]:
    """배치 문서 처리 후 임베딩 반환."""
    model = get_embedding_model()
    return model.encode(docs, batch_size=64).tolist()

# 배치 작업 실행
results = batch_embed.remote([f"문서 {i}" for i in range(10000)])

@stub.function(
    gpu="H100-80GB",
    memory=16384,
    timeout=14400  # 4시간
)
def run_finetune(dataset_path: str, output_dir: str):
    """데이터셋에서 LoRA 파인튜닝 실행."""
    from trl import SFTTrainer
    from peft import LoraConfig

    model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3.2-3B")
    tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3.2-3B")

    peft_config = LoraConfig(
        r=16, lora_alpha=32, target_modules=["q_proj", "v_proj"],
        lora_dropout=0.05, bias="none", task_type="CAUSAL_LM"
    )

    trainer = SFTTrainer(
        model=model, tokenizer=tokenizer,
        train_dataset=load_dataset(dataset_path),
        peft_config=peft_config,
        args=TrainingArguments(output_dir=output_dir, num_train_epochs=3)
    )
    trainer.train()
    trainer.save_model(output_dir)

# 3B 파라미터 모델에 H100 사용하지 마세요
# 대신 A10G 사용 — 비용 75% 절감
@stub.function(gpu="A10G", memory=4096)
def light_inference(prompt: str):
    model = load_small_model()  # 3B 파라미터는 쉽게 수용
    return model.generate(prompt)

# 대규모 파인튜닝에만 H100 예약
@stub.function(gpu="H100-80GB", memory=32768)
def heavy_finetune(config: dict):
    return run_large_scale_training(config)

# 예측 가능한 트래픽: 비즈니스 시간 동안만 웜 상태 유지
@stub.function(gpu="L4", keep_warm=1)
def production_endpoint():
    ...

# 버스트 트래픽: 더 높은 concurrency_limit 사용
@stub.function(gpu="L4", concurrency_limit=50, keep_warm=3)
def bursty_endpoint():
    ...

# ❌ 나쁨: 매번 모델을 로드
@stub.function(gpu="A10G")
def bad_approach(prompt: str):
    model = load_model()  # 호출마다 다시 로드!
    return model.generate(prompt)

# ✅ 좋음: 한 번 로드하고 요청 간 재사용
@stub.cls(gpu="A10G")
class GoodApproach:
    @modal.enter()
    def setup(self):
        self.model = load_model()  # 시작 시 한 번만 로드
    
    @modal.method()
    def generate(self, prompt: str):
        return self.model.generate(prompt)  # 로드된 모델 재사용

import modal

stub = modal.Stub("secret-demo")

@stub.function(
    secrets=[
        modal.Secret.from_name("huggingface-token"),
        modal.Secret.from_name("openai-key"),
    ]
)
def secure_inference(prompt: str):
    import os
    hf_token = os.environ["HF_TOKEN"]  # 시크릿에서 주입
    openai_key = os.environ["OPENAI_API_KEY"]
    return call_api(prompt, hf_token, openai_key)

modal secret create huggingface-token HF_TOKEN=your_token_here
modal secret create openai-key OPENAI_API_KEY=sk-...

# 모델 체크포인트용 볼륨 생성
checkpoint_volume = modal.Volume.from_name("model-checkpoints", create_if_missing=True)

@stub.function(
    gpu="A100-80GB",
    volumes={"/checkpoints": checkpoint_volume},
    timeout=7200
)
def fine_tune_and_save(dataset_url: str):
    dataset = load_dataset(dataset_url)
    
    trainer.train()
    trainer.save_model("/checkpoints/final-model")
    
    print(f"체크포인트 볼륨에 저장됨. 크기: {os.path.getsize('/checkpoints/final-model')}")

@stub.function(volumes={"/checkpoints": checkpoint_volume})
def load_and_infer(prompt: str):
    model = AutoModelForCausalLM.from_pretrained("/checkpoints/final-model")
    return model.generate(prompt)

@stub.function(
    gpu="L4",
    network_mounts={"/etc/resolv.conf": modal.NetworkMount()},
    blocked_subnets=["169.254.0.0/16"],  # 메타데이터 서비스 차단
    allowed_domains=["api.openai.com"]   # 특정 도메인만 허용
)
def restricted_inference(prompt: str):
    return call_openai(prompt)

custom_image = (
    modal.Image.from_dockerhub("nvidia/cuda:12.2.0-devel-ubuntu22.04")
    .apt_install("git", "cmake", "build-essential")
    .pip_install("torch", "transformers", "bitsandbytes")
    .copy_local_dir("./my-custom-model", "/app/model")
)

@stub.function(image=custom_image, gpu="A100-80GB")
def custom_model_inference(request: dict):
    model = torch.load("/app/model/best.pt")
    return model.predict(request["input"])

@stub.cls(
    gpu="A100-80GB",
    memory=32768,  # 대형 모델용 32GB RAM
    ephemeral_disk=100_000  # 모델 가중치용 100GB 디스크
)
class LargeModel:
    @modal.enter()
    def load(self):
        self.model = AutoModel.from_pretrained(
            "big-model",
            torch_dtype=torch.float16,  # 반정밀도 사용
            device_map="auto"
        )

@stub.cls(
    gpu="A10G",
    keep_warm=3,  # 항상 3개의 웜 컨테이너 유지
    timeout=600
)
class WarmEndpoint:
    @modal.enter()
    def load(self):
        self.model = load_model()
        print("모델 로드 완료")

@stub.function(
    gpu="H100-80GB",
    timeout=28800,  # 8시간
    volumes={"/data": modal.Volume.from_name("training-data")}
)
def long_training_job(config_path: str):
    for epoch in range(10):
        train_epoch(config_path)
        if epoch % 2 == 0:
            save_checkpoint(f"/data/checkpoint-{epoch}")

@stub.cls(
    gpu="L4",
    concurrency_limit=100,       # 최대 컨테이너 수
    allow_concurrent_inputs=20,  # 컨테이너당 요청 수
    keep_warm=5                  # 웜 풀 크기
)
class ScalableEndpoint:
    @modal.method()
    def handle(self, request: dict):
        return process(request)