12장: AI 개발에서의 Python 활용

1. 생태계의 깊이
   - NumPy, SciPy: 수치 계산의 기반
   - pandas: 데이터 처리
   - PyTorch, TensorFlow: 딥러닝 프레임워크
   - Hugging Face: 사전 학습 모델 생태계
 
2. 접착제 언어(Glue Language)
   - C/C++/CUDA로 작성된 고성능 라이브러리를 Python으로 조합
   - Python은 느리지만, 실제 연산은 최적화된 네이티브 코드가 수행
 
3. 프로토타이핑 효율
   - 대화형 환경 (Jupyter, PyREPL)
   - 짧은 코드로 실험 가능
   - 시각화 도구와의 긴밀한 통합

# 프로젝트 생성
uv init ml-project
cd ml-project
 
# Python 3.12 설정
uv python install 3.12
echo "3.12" > .python-version
 
# 핵심 의존성 설치
uv add torch torchvision
uv add transformers datasets
uv add numpy pandas scikit-learn
 
# 개발 도구
uv add --dev pytest jupyter ruff mypy
uv add --dev matplotlib seaborn
 
# 실험 추적
uv add wandb mlflow

[project]
name = "ml-project"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "torch>=2.2",
    "transformers>=4.40",
    "datasets>=2.18",
    "numpy>=1.26",
    "pandas>=2.2",
]
 
[project.optional-dependencies]
gpu = ["torch[cuda]"]
serving = ["fastapi", "uvicorn", "pydantic>=2.0"]
 
[tool.ruff]
target-version = "py312"
 
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]

from typing import TypedDict, Protocol
from dataclasses import dataclass
 
# 데이터 스키마 정의
class RawSample(TypedDict):
    text: str
    label: str
    metadata: dict[str, str]
 
class ProcessedSample(TypedDict):
    input_ids: list[int]
    attention_mask: list[int]
    label_id: int
 
class BatchData(TypedDict):
    input_ids: list[list[int]]
    attention_mask: list[list[int]]
    labels: list[int]
 
# 파이프라인 단계 Protocol
class DataProcessor[TIn, TOut](Protocol):
    def process(self, data: TIn) -> TOut: ...
 
class TextCleaner:
    def process(self, data: RawSample) -> RawSample:
        cleaned_text = data["text"].strip().lower()
        return RawSample(
            text=cleaned_text,
            label=data["label"],
            metadata=data["metadata"],
        )
 
class Tokenizer:
    def __init__(self, vocab: dict[str, int], max_length: int = 512) -> None:
        self.vocab = vocab
        self.max_length = max_length
 
    def process(self, data: RawSample) -> ProcessedSample:
        tokens = data["text"].split()
        input_ids = [self.vocab.get(t, 0) for t in tokens]
        input_ids = input_ids[:self.max_length]
        attention_mask = [1] * len(input_ids)
 
        # 패딩
        pad_length = self.max_length - len(input_ids)
        input_ids = input_ids + [0] * pad_length
        attention_mask = attention_mask + [0] * pad_length
 
        return ProcessedSample(
            input_ids=input_ids,
            attention_mask=attention_mask,
            label_id=hash(data["label"]) % 10,
        )

from typing import TypedDict, NotRequired
from dataclasses import dataclass
 
@dataclass
class TrainingConfig:
    model_name: str
    learning_rate: float = 3e-5
    batch_size: int = 32
    num_epochs: int = 3
    max_length: int = 512
    warmup_ratio: float = 0.1
    weight_decay: float = 0.01
    gradient_accumulation_steps: int = 1
    fp16: bool = True
    seed: int = 42
 
    def validate(self) -> None:
        assert self.learning_rate > 0, "Learning rate must be positive"
        assert self.batch_size > 0, "Batch size must be positive"
        assert 0.0 <= self.warmup_ratio <= 1.0, "Warmup ratio must be in [0, 1]"
 
@dataclass
class EvalConfig:
    eval_batch_size: int = 64
    eval_steps: int = 500
    metric_for_best_model: str = "f1"
    greater_is_better: bool = True
 
class ExperimentConfig(TypedDict):
    name: str
    training: TrainingConfig
    evaluation: EvalConfig
    tags: NotRequired[list[str]]
    notes: NotRequired[str]

from dataclasses import dataclass
 
@dataclass
class ModelConfig:
    model_name: str
    task: str
    device: str = "auto"
    max_length: int = 512
 
def load_model(config: ModelConfig):
    """모델 로드 (타입 안전)"""
    from transformers import AutoTokenizer, AutoModelForSequenceClassification
 
    tokenizer = AutoTokenizer.from_pretrained(config.model_name)
    model = AutoModelForSequenceClassification.from_pretrained(
        config.model_name
    )
    return tokenizer, model
 
def predict(
    text: str,
    tokenizer,
    model,
    max_length: int = 512,
) -> dict[str, float]:
    """추론 수행"""
    inputs = tokenizer(
        text,
        return_tensors="pt",
        max_length=max_length,
        truncation=True,
        padding=True,
    )
    outputs = model(**inputs)
    probabilities = outputs.logits.softmax(dim=-1)
 
    labels = model.config.id2label
    result: dict[str, float] = {}
    for i, prob in enumerate(probabilities[0]):
        result[labels[i]] = round(float(prob), 4)
    return result

from typing import Protocol, TypedDict
from dataclasses import dataclass, field
 
class Message(TypedDict):
    role: str
    content: str
 
class CompletionResponse(TypedDict):
    content: str
    model: str
    usage: dict[str, int]
 
class LLMClient(Protocol):
    def complete(
        self,
        messages: list[Message],
        temperature: float = 0.7,
        max_tokens: int = 1000,
    ) -> CompletionResponse: ...
 
@dataclass
class ConversationManager:
    client: LLMClient
    system_prompt: str = ""
    history: list[Message] = field(default_factory=list)
    max_history: int = 20
 
    def __post_init__(self) -> None:
        if self.system_prompt:
            self.history.append(
                Message(role="system", content=self.system_prompt)
            )
 
    def send(self, user_message: str) -> str:
        self.history.append(
            Message(role="user", content=user_message)
        )
 
        # 히스토리 길이 제한
        messages = self.history[-self.max_history:]
 
        response = self.client.complete(
            messages=messages,
            temperature=0.7,
        )
 
        assistant_message = response["content"]
        self.history.append(
            Message(role="assistant", content=assistant_message)
        )
 
        return assistant_message

from dataclasses import dataclass
 
@dataclass
class DataPoint:
    value: float | int | str | None
    source: str
    quality: str = "unknown"
 
def clean_data_point(point: DataPoint) -> DataPoint | None:
    match point:
        case DataPoint(value=None):
            return None  # null 값 제거
        case DataPoint(value=str(s)) if len(s.strip()) == 0:
            return None  # 빈 문자열 제거
        case DataPoint(value=float(v)) if v != v:  # NaN 체크
            return None
        case DataPoint(value=float(v), source=src) if v < 0:
            return DataPoint(
                value=abs(v),
                source=src,
                quality="corrected",
            )
        case DataPoint(value=str(s), source=src):
            return DataPoint(
                value=s.strip(),
                source=src,
                quality="cleaned",
            )
        case _:
            return DataPoint(
                value=point.value,
                source=point.source,
                quality="valid",
            )

import threading
from concurrent.futures import ThreadPoolExecutor
 
def preprocess_chunk(
    data: list[dict],
    start_idx: int,
    results: list,
) -> None:
    """데이터 청크 전처리"""
    processed = []
    for item in data:
        # CPU 바운드 전처리 작업
        text = item.get("text", "")
        tokens = text.lower().split()
        tokens = [t for t in tokens if len(t) > 2]
        processed.append(tokens)
    results[start_idx] = processed
 
def parallel_preprocess(
    data: list[dict],
    num_workers: int = 4,
) -> list:
    """Free-threaded 모드에서 병렬 전처리"""
    chunk_size = len(data) // num_workers
    chunks = []
    for i in range(num_workers):
        start = i * chunk_size
        end = start + chunk_size if i < num_workers - 1 else len(data)
        chunks.append(data[start:end])
 
    results: list = [None] * num_workers
 
    with ThreadPoolExecutor(max_workers=num_workers) as executor:
        futures = []
        for i, chunk in enumerate(chunks):
            future = executor.submit(preprocess_chunk, chunk, i, results)
            futures.append(future)
 
        for future in futures:
            future.result()  # 완료 대기
 
    # 결과 병합
    merged = []
    for result in results:
        if result is not None:
            merged.extend(result)
    return merged

# Jupyter 커널 설정
# uv로 3.12 환경에서 jupyter 설치
# uv add jupyter ipykernel
# uv run python -m ipykernel install --user --name ml-312
 
# Jupyter 노트북에서 PEP 695 활용
# (Jupyter는 Python 3.12+ 커널에서 새 문법을 지원)
 
# 타입 별칭
type Embedding = list[float]
type BatchEmbeddings = list[Embedding]
 
# 제네릭 함수
def cosine_similarity[T: (list[float],)](a: T, b: T) -> float:
    dot_product = sum(x * y for x, y in zip(a, b))
    norm_a = sum(x * x for x in a) ** 0.5
    norm_b = sum(x * x for x in b) ** 0.5
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot_product / (norm_a * norm_b)

[project]
name = "ml-project"
requires-python = ">=3.12"
dependencies = [
    "numpy>=1.26",
    "pandas>=2.2",
    "scikit-learn>=1.4",
    "transformers>=4.40",
    "datasets>=2.18",
]
 
[project.optional-dependencies]
# CPU 전용
cpu = ["torch>=2.2"]
 
# GPU (CUDA 12.1)
gpu = ["torch>=2.2"]
 
# 서빙
serve = [
    "fastapi>=0.110",
    "uvicorn>=0.28",
    "pydantic>=2.6",
]
 
# 개발
dev = [
    "pytest>=8.0",
    "ruff>=0.3",
    "jupyter>=1.0",
    "matplotlib>=3.8",
]

# CPU 개발 환경
uv sync --extra cpu --extra dev
 
# GPU 학습 환경
uv sync --extra gpu
 
# 서빙 환경
uv sync --extra serve