Pràctica PR5073/02: API d'IA en Producció amb FastAPI i Docker

Objectius

• Dissenyar una API REST per a serveis d'IA seguint bones pràctiques
• Implementar endpoints de classificació i generació de text
• Integrar models de Hugging Face Transformers
• Conteneiritzar l'API amb Docker i docker-compose
• Implementar tests automàtics amb pytest
• Configurar health checks i documentació Swagger automàtica
• Personalitzar tots els endpoints i respostes amb el nom de l'alumne

Prerequisits

Requisit	Detall
Temps estimat	8-10 hores
RAM mínima	4 GB (8 GB recomanat per als models Transformers)
Espai en disc	5 GB lliures
Docker Desktop	v4.25+ instal·lat i funcionant
Coneixements	Python intermedi, REST APIs bàsic, conceptes HTTP

---

Introducció

Per quin motiu FastAPI per a IA?

FastAPI es el framework web de Python més popular per a APIs d'IA en producció el 2025. Les raons:

1. Velocitat: basat en Starlette i Uvicorn (ASGI), es 2-3x més ràpid que Flask en benchmark estàndard
2. Tipat automàtic: usa anotacions de tipus Python + Pydantic per validació automàtica
3. Documentació automàtica: genera Swagger UI i ReDoc sense cap configuració addicional
4. Async nativa: suporta operacions asíncrones per a operacions I/O-bound (cridades a APIs, base de dades)
5. Estàndard OpenAPI: la documentació generada es compatible amb l'estàndard OpenAPI 3.0

Comparativa de frameworks per a APIs d'IA:

Framework	Velocitat	IA ecosystem	Async	Documentacio auto
FastAPI	Alta	Excellent	Si	Si (Swagger)
Flask	Mitja	Bona	Parcial	No
Django REST	Mitja	Bona	Parcial	Parcial
Tornado	Alta	Limitada	Si	No
Sanic	Alta	Limitada	Si	No

Arquitectura de l'API

flowchart LR
    Client[Client HTTP] --> |Request JSON| API[FastAPI]
    API --> |Validacio Pydantic| Router[Routers]
    Router --> Classify[classify.py]
    Router --> Generate[generate.py]
    Classify --> Model1[Sentiment Model\nHugging Face]
    Generate --> Model2[Text Gen Model\nOllama API]
    API --> |Response JSON| Client
    API --> |/healthz| Health[Health Check]
    API --> |/docs| Swagger[Swagger UI]

---

Part 1: Estructura del projecte

Creeu l'estructura de directoris:

mkdir api-ia-joan-garcia
cd api-ia-joan-garcia

# Crear estructura de directoris
mkdir -p app/routers tests

# Crear fitxers buits
touch app/__init__.py
touch app/main.py
touch app/models.py
touch app/config.py
touch app/routers/__init__.py
touch app/routers/classify.py
touch app/routers/generate.py
touch tests/__init__.py
touch tests/test_api.py
touch Dockerfile
touch docker-compose.yml
touch requirements.txt
touch .env.example

L'estructura final hauria de ser:

api-ia-joan-garcia/
├── app/
│   ├── __init__.py
│   ├── main.py          # Punt d'entrada FastAPI
│   ├── models.py        # Schemas Pydantic (validació de dades)
│   ├── config.py        # Variables d'entorn i configuració
│   └── routers/
│       ├── __init__.py
│       ├── classify.py  # Endpoints de classificació
│       └── generate.py  # Endpoints de generació
├── tests/
│   ├── __init__.py
│   └── test_api.py      # Tests amb pytest
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env.example

---

Part 2: Configuració i models de dades

requirements.txt

fastapi==0.115.4
uvicorn[standard]==0.32.0
pydantic==2.9.2
pydantic-settings==2.6.1
transformers==4.46.2
torch==2.5.1
requests==2.32.3
httpx==0.27.2
pytest==8.3.3
pytest-asyncio==0.24.0
httpx==0.27.2
python-multipart==0.0.12
slowapi==0.1.9

app/config.py

# app/config.py
# Pràctica PR5073/02 - Configuracio de l'API
# Alumne: Joan Garcia

from pydantic_settings import BaseSettings
from typing import Optional

class Configuracio(BaseSettings):
    """
    Configuracio de l'API carregada des de variables d'entorn.
    Pydantic BaseSettings llegeix automaticament les variables d'entorn.
    """
    # Identificacio de l'alumne (apareix a tots els endpoints)
    alumne: str = "joan_garcia"
    alumne_nom_complet: str = "Joan Garcia"

    # Configuracio del servidor
    host: str = "0.0.0.0"
    port: int = 8000
    workers: int = 1
    log_level: str = "info"

    # APIs externes (opcionals)
    openai_api_key: Optional[str] = None
    anthropic_api_key: Optional[str] = None

    # Ollama (per a generacio de text local)
    ollama_url: str = "http://ollama:11434"
    ollama_model: str = "llama3.2:3b"

    # Configuracio dels models Hugging Face
    model_classificacio: str = "nlptown/bert-base-multilingual-uncased-sentiment"
    model_cache_dir: str = "/app/model_cache"

    # Rate limiting
    rate_limit_per_minut: int = 60

    class Config:
        env_file = ".env"
        case_sensitive = False

# Instancia global de configuracio
config = Configuracio()

app/models.py

# app/models.py
# Pràctica PR5073/02 - Schemas Pydantic per a l'API
# Alumne: Joan Garcia

from pydantic import BaseModel, Field, field_validator
from typing import Optional, List, Dict, Any
from datetime import datetime
from app.config import config

ALUMNE = config.alumne

# ============================================================
# MODELS DE CLASSIFICACIO
# ============================================================

class ClassifyRequest(BaseModel):
    """Schema per a peticions de classificació de text."""
    text: str = Field(
        ...,
        min_length=1,
        max_length=5000,
        description="Text a classificar (1-5000 caracters)",
        examples=["M'encanta aquest producte, funciona molt be!"]
    )
    model: str = Field(
        default="sentiment",
        description="Model a usar: 'sentiment' o 'zero-shot'",
        examples=["sentiment"]
    )
    idioma: Optional[str] = Field(
        default=None,
        description="Codi ISO 639-1 de l'idioma (ca, es, en). Auto-detectat si no s'especifica."
    )

    @field_validator("text")
    @classmethod
    def text_no_buit(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("El text no pot estar buit o ser sols espais")
        return v.strip()

class ClassifyResponse(BaseModel):
    """Schema per a respostes de classificació."""
    etiqueta: str = Field(description="Etiqueta predita (ex: 'positiu', 'negatiu')")
    puntuacio: float = Field(description="Probabilitat de l'etiqueta (0-1)")
    totes_les_etiquetes: Optional[Dict[str, float]] = Field(
        default=None,
        description="Probabilitats per a totes les etiquetes"
    )
    text_entrada: str = Field(description="Text analitzat (primeres 100 chars)")
    model_usat: str = Field(description="Model usat per a la prediccio")
    alumne: str = Field(default=ALUMNE, description="Identificador de l'alumne")
    timestamp: datetime = Field(default_factory=datetime.utcnow)

# ============================================================
# MODELS DE GENERACIO
# ============================================================

class GenerateRequest(BaseModel):
    """Schema per a peticions de generacio de text."""
    prompt: str = Field(
        ...,
        min_length=5,
        max_length=2000,
        description="Prompt per a la generacio de text",
        examples=["Explica en catala que es la intel·ligencia artificial"]
    )
    max_tokens: int = Field(
        default=500,
        ge=50,
        le=2000,
        description="Maxim de tokens a generar (50-2000)"
    )
    temperatura: float = Field(
        default=0.7,
        ge=0.0,
        le=2.0,
        description="Temperatura de la generacio (0=determinista, 2=creatiu)"
    )
    sistema: Optional[str] = Field(
        default=None,
        description="Prompt del sistema que defineix el comportament del model"
    )

class GenerateResponse(BaseModel):
    """Schema per a respostes de generacio."""
    text_generat: str = Field(description="Text generat pel model")
    tokens_usats: Optional[int] = Field(default=None)
    model_usat: str = Field(description="Model usat per a la generacio")
    prompt_original: str = Field(description="Prompt original (primeres 100 chars)")
    alumne: str = Field(default=ALUMNE)
    timestamp: datetime = Field(default_factory=datetime.utcnow)

# ============================================================
# MODELS DE SALUT I ERROR
# ============================================================

class HealthResponse(BaseModel):
    """Schema per al health check."""
    estat: str = Field(description="'ok' o 'error'")
    alumne: str = Field(default=ALUMNE)
    version: str = Field(default="1.0.0")
    models_carregats: List[str] = Field(default_factory=list)
    timestamp: datetime = Field(default_factory=datetime.utcnow)

class ErrorResponse(BaseModel):
    """Schema per a respostes d'error."""
    error: str
    detall: Optional[str] = None
    codi_error: int
    alumne: str = Field(default=ALUMNE)

---

Part 3: Implementació dels routers

app/routers/classify.py

# app/routers/classify.py
# Pràctica PR5073/02 - Endpoints de classificació
# Alumne: Joan Garcia

from fastapi import APIRouter, HTTPException, BackgroundTasks
from transformers import pipeline, Pipeline
from typing import Optional
import logging
import time

from app.models import ClassifyRequest, ClassifyResponse
from app.config import config

logger = logging.getLogger(__name__)

router = APIRouter(
    prefix="/api/v1",
    tags=["Classificacio de Text"],
    responses={
        422: {"description": "Error de validació de dades"},
        503: {"description": "Model no disponible"}
    }
)

# Cache dels models (carregats un cop, reutilitzats)
_models: dict[str, Pipeline] = {}

def obtenir_model_sentiment() -> Pipeline:
    """
    Carrega el model de sentiment si no esta ja en memoria.
    Singleton pattern: el model es carrega una sola vegada.
    """
    model_key = "sentiment"
    if model_key not in _models:
        logger.info(f"Carregant model de sentiment: {config.model_classificacio}")
        t_inici = time.time()

        # nlptown/bert-base-multilingual-uncased-sentiment:
        # - Multilingüe (suporta catala)
        # - 5 etiquetes: 1-5 estrelles
        # - Model relativament lleuger (669 MB)
        _models[model_key] = pipeline(
            "sentiment-analysis",
            model=config.model_classificacio,
            tokenizer=config.model_classificacio,
            device=-1,  # CPU (-1) o GPU (0)
            truncation=True,
            max_length=512
        )

        temps = time.time() - t_inici
        logger.info(f"Model carregat en {temps:.2f}s")

    return _models[model_key]

@router.post(
    "/classify/sentiment",
    response_model=ClassifyResponse,
    summary="Classificació de sentiment",
    description="""
    Analitza el sentiment d'un text (positiu/negatiu/neutral).
    Suporta català, castellà, anglès i altres idiomes europeus.

    Model usat: nlptown/bert-base-multilingual-uncased-sentiment
    Escala: 1 estrella (molt negatiu) → 5 estrelles (molt positiu)
    """
)
async def classificar_sentiment(
    peticio: ClassifyRequest,
    background_tasks: BackgroundTasks
) -> ClassifyResponse:
    """Endpoint principal de classificació de sentiment."""

    try:
        model = obtenir_model_sentiment()
        t_inici = time.time()

        # Inferencia
        resultat = model(peticio.text)[0]
        temps_inferencia = (time.time() - t_inici) * 1000

        # Convertir etiqueta numica a text llegible
        etiqueta_num = resultat["label"]  # Ex: "4 stars"
        num_estrelles = int(etiqueta_num.split()[0])

        if num_estrelles >= 4:
            etiqueta_llegible = "molt_positiu" if num_estrelles == 5 else "positiu"
        elif num_estrelles == 3:
            etiqueta_llegible = "neutral"
        else:
            etiqueta_llegible = "molt_negatiu" if num_estrelles == 1 else "negatiu"

        logger.info(
            f"Classificat en {temps_inferencia:.0f}ms: "
            f"'{peticio.text[:50]}...' -> {etiqueta_llegible} ({resultat['score']:.3f})"
        )

        # Log en background (sense bloquejar la resposta)
        background_tasks.add_task(
            registrar_estadistica,
            endpoint="classify/sentiment",
            temps_ms=temps_inferencia,
            resultat=etiqueta_llegible
        )

        return ClassifyResponse(
            etiqueta=etiqueta_llegible,
            puntuacio=round(resultat["score"], 4),
            text_entrada=peticio.text[:100],
            model_usat=config.model_classificacio
        )

    except Exception as e:
        logger.error(f"Error en classificació: {e}")
        raise HTTPException(
            status_code=503,
            detail=f"Error al model de classificació: {str(e)}"
        )

@router.post(
    "/classify/zero-shot",
    response_model=ClassifyResponse,
    summary="Classificació Zero-Shot",
    description="""
    Classifica text en categories personalitzades sense entrenament específic.
    Util quan les categories no son fixes o canvien freqüentment.
    """
)
async def classificar_zero_shot(
    peticio: ClassifyRequest,
    categories: list[str] = ["positiu", "negatiu", "neutral", "urgent"]
) -> ClassifyResponse:
    """Endpoint de classificació zero-shot amb categories personalitzades."""

    if "zero-shot" not in _models:
        logger.info("Carregant model zero-shot...")
        _models["zero-shot"] = pipeline(
            "zero-shot-classification",
            model="facebook/bart-large-mnli",
            device=-1
        )

    model = _models["zero-shot"]
    resultat = model(peticio.text, candidate_labels=categories)

    return ClassifyResponse(
        etiqueta=resultat["labels"][0],
        puntuacio=round(resultat["scores"][0], 4),
        totes_les_etiquetes=dict(zip(resultat["labels"], resultat["scores"])),
        text_entrada=peticio.text[:100],
        model_usat="facebook/bart-large-mnli"
    )

async def registrar_estadistica(endpoint: str, temps_ms: float, resultat: str):
    """Registra estadistiques d'us en background."""
    logger.info(f"ESTADISTICA | endpoint={endpoint} | temps={temps_ms:.0f}ms | resultat={resultat}")

app/routers/generate.py

# app/routers/generate.py
# Pràctica PR5073/02 - Endpoints de generació de text
# Alumne: Joan Garcia

from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
import requests
import json
import logging
from typing import AsyncGenerator

from app.models import GenerateRequest, GenerateResponse
from app.config import config

logger = logging.getLogger(__name__)

router = APIRouter(
    prefix="/api/v1",
    tags=["Generacio de Text"],
    responses={
        503: {"description": "Model de generacio no disponible"}
    }
)

def generar_amb_ollama(prompt: str, sistema: str, max_tokens: int, temperatura: float) -> str:
    """
    Genera text usant Ollama (model local).
    Es fa una crida HTTP síncrona a l'API d'Ollama.
    """
    payload = {
        "model": config.ollama_model,
        "messages": [
            {"role": "system", "content": sistema},
            {"role": "user", "content": prompt}
        ],
        "options": {
            "num_predict": max_tokens,
            "temperature": temperatura
        },
        "stream": False
    }

    try:
        response = requests.post(
            f"{config.ollama_url}/api/chat",
            json=payload,
            timeout=60
        )
        response.raise_for_status()
        data = response.json()
        return data["message"]["content"]

    except requests.exceptions.ConnectionError:
        raise HTTPException(
            status_code=503,
            detail=f"No es pot connectar a Ollama a {config.ollama_url}. "
                   "Comprova que el servei esta en marxa."
        )
    except requests.exceptions.Timeout:
        raise HTTPException(
            status_code=504,
            detail="Timeout: el model tarda massa a respondre. Redueix max_tokens."
        )

@router.post(
    "/generate",
    response_model=GenerateResponse,
    summary="Generació de text amb LLM",
    description="""
    Genera text en catala usando un LLM local via Ollama.
    El prompt del sistema esta configurat per defecte per a respondre en catala.

    Model per defecte: llama3.2:3b (local, sense cost d'API)
    """
)
async def generar_text(peticio: GenerateRequest) -> GenerateResponse:
    """Endpoint principal de generació de text."""

    # Prompt del sistema per defecte
    sistema = peticio.sistema or (
        f"Ets un assistent expert creat per {config.alumne_nom_complet}. "
        f"Respon sempre en català, de manera clara i precisa. "
        f"Ets rigorós amb els fets i honest quan no coneixes la resposta."
    )

    logger.info(f"Generant text: '{peticio.prompt[:50]}...' (max_tokens={peticio.max_tokens})")

    text_generat = generar_amb_ollama(
        prompt=peticio.prompt,
        sistema=sistema,
        max_tokens=peticio.max_tokens,
        temperatura=peticio.temperatura
    )

    return GenerateResponse(
        text_generat=text_generat,
        model_usat=config.ollama_model,
        prompt_original=peticio.prompt[:100]
    )

@router.post(
    "/generate/stream",
    summary="Generació de text en streaming",
    description="Igual que /generate però retorna el text progressivament (Server-Sent Events)."
)
async def generar_text_streaming(peticio: GenerateRequest):
    """Endpoint de generació amb streaming (tokens progressius)."""

    sistema = peticio.sistema or f"Respon sempre en català. Creat per {config.alumne_nom_complet}."

    async def generador_tokens() -> AsyncGenerator[str, None]:
        payload = {
            "model": config.ollama_model,
            "messages": [
                {"role": "system", "content": sistema},
                {"role": "user", "content": peticio.prompt}
            ],
            "options": {"num_predict": peticio.max_tokens, "temperature": peticio.temperatura},
            "stream": True
        }

        try:
            with requests.post(
                f"{config.ollama_url}/api/chat",
                json=payload,
                stream=True,
                timeout=120
            ) as response:
                for linia in response.iter_lines():
                    if linia:
                        try:
                            chunk = json.loads(linia)
                            if "message" in chunk and "content" in chunk["message"]:
                                token = chunk["message"]["content"]
                                yield f"data: {json.dumps({'token': token})}\n\n"
                        except json.JSONDecodeError:
                            continue

            yield f"data: {json.dumps({'finalitzat': True, 'alumne': config.alumne})}\n\n"

        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"

    return StreamingResponse(
        generador_tokens(),
        media_type="text/event-stream",
        headers={"X-Alumne": config.alumne}
    )

---

Part 4: Punt d'entrada principal

app/main.py

# app/main.py
# Pràctica PR5073/02 - Punt d'entrada FastAPI
# Alumne: Joan Garcia

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
import logging
import time
from datetime import datetime

from app.config import config
from app.models import HealthResponse, ErrorResponse
from app.routers import classify, generate

# Configuracio del logging
logging.basicConfig(
    level=getattr(logging, config.log_level.upper()),
    format="%(asctime)s | %(name)s | %(levelname)s | %(message)s"
)
logger = logging.getLogger(__name__)

# Rate limiter
limiter = Limiter(key_func=get_remote_address)

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Lifecycle de l'aplicacio: startup i shutdown."""
    # STARTUP
    logger.info(f"=== Arrancant API Intel·ligencia Artificial - {config.alumne} ===")
    logger.info(f"Alumne: {config.alumne_nom_complet}")
    logger.info(f"Versio: 1.0.0")
    logger.info(f"Ollama URL: {config.ollama_url}")
    logger.info(f"Model generacio: {config.ollama_model}")
    logger.info(f"Model classificacio: {config.model_classificacio}")

    yield

    # SHUTDOWN
    logger.info(f"=== Tancant API - {config.alumne} ===")

# Crear aplicacio FastAPI
app = FastAPI(
    title=f"API Intel·ligencia Artificial - {config.alumne}",
    description=f"""
    API REST per a inferencia de models d'IA.

    Creat per: **{config.alumne_nom_complet}**
    Curs: IABD - Programació IA (5073)
    Institut: Sa Palomera, Blanes

    ## Endpoints disponibles

    ### Classificacio
    - `POST /api/v1/classify/sentiment` — Analisi de sentiment multilingüe
    - `POST /api/v1/classify/zero-shot` — Classificacio amb categories personalitzades

    ### Generacio
    - `POST /api/v1/generate` — Generacio de text amb LLM local
    - `POST /api/v1/generate/stream` — Generacio en streaming (SSE)

    ### Sistema
    - `GET /healthz` — Health check
    - `GET /` — Informacio de l'API
    """,
    version="1.0.0",
    contact={
        "name": config.alumne_nom_complet,
        "email": f"{config.alumne}@sapalomera.cat"
    },
    lifespan=lifespan
)

# Rate limiting
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)

# Middleware per a logging de peticions
@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    t_inici = time.time()
    response = await call_next(request)
    temps_ms = (time.time() - t_inici) * 1000

    logger.info(
        f"{request.method} {request.url.path} "
        f"| Status: {response.status_code} "
        f"| Temps: {temps_ms:.0f}ms "
        f"| IP: {request.client.host if request.client else 'unknown'}"
    )

    # Afegir headers personalitzats
    response.headers["X-Alumne"] = config.alumne
    response.headers["X-Temps-Resposta-Ms"] = str(round(temps_ms))

    return response

# Registrar routers
app.include_router(classify.router)
app.include_router(generate.router)

# ============================================================
# ENDPOINTS PRINCIPALS
# ============================================================

@app.get("/", tags=["Sistema"])
async def arrel():
    """Endpoint arrel amb informacio de l'API."""
    return {
        "api": "API Intel·ligencia Artificial",
        "alumne": config.alumne,
        "alumne_nom_complet": config.alumne_nom_complet,
        "versio": "1.0.0",
        "documentacio": "/docs",
        "healthcheck": "/healthz",
        "endpoints_disponibles": [
            "/api/v1/classify/sentiment",
            "/api/v1/classify/zero-shot",
            "/api/v1/generate",
            "/api/v1/generate/stream"
        ]
    }

@app.get(
    "/healthz",
    response_model=HealthResponse,
    tags=["Sistema"],
    summary="Health Check",
    description="Comprova l'estat de salut de l'API i dels models carregats."
)
async def health_check():
    """Endpoint de health check per a Docker i load balancers."""
    models_disponibles = []

    # Comprovar Ollama
    import requests
    try:
        resp = requests.get(f"{config.ollama_url}/api/tags", timeout=5)
        if resp.status_code == 200:
            models_disponibles.append(f"ollama:{config.ollama_model}")
    except Exception:
        pass

    return HealthResponse(
        estat="ok",
        models_carregats=models_disponibles
    )

---

Part 5: Dockerfile i docker-compose

Dockerfile

# Dockerfile
# Pràctica PR5073/02 - API IA en Producció
# Alumne: Joan Garcia

# --- FASE 1: Builder ---
# Instal·lem les dependències en una imatge temporal
FROM python:3.11-slim AS builder

WORKDIR /app

# Copiar i instal·lar dependències (en una capa separada per cache)
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# --- FASE 2: Runtime ---
# Imatge final lleugera amb només el necessari
FROM python:3.11-slim AS runtime

WORKDIR /app

# Metadades
LABEL maintainer="joan_garcia@sapalomera.cat"
LABEL alumne="joan_garcia"
LABEL practica="PR5073/02"
LABEL version="1.0.0"

# Copiar dependències instal·lades del builder
COPY --from=builder /root/.local /root/.local

# Copiar codi de l'aplicació
COPY app/ ./app/

# Variables d'entorn per defecte
ENV ALUMNE=joan_garcia
ENV ALUMNE_NOM_COMPLET="Joan Garcia"
ENV LOG_LEVEL=info
ENV OLLAMA_URL=http://ollama:11434
ENV OLLAMA_MODEL=llama3.2:3b
ENV PATH=/root/.local/bin:$PATH

# Exposar port
EXPOSE 8000

# Health check intern
HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:8000/healthz', timeout=5)" || exit 1

# Usuari no-root per seguretat
RUN adduser --disabled-password --gecos '' apiuser
USER apiuser

# Punt d'entrada
CMD ["uvicorn", "app.main:app", \
     "--host", "0.0.0.0", \
     "--port", "8000", \
     "--workers", "1", \
     "--log-level", "info"]

docker-compose.yml

# docker-compose.yml
# Pràctica PR5073/02 - API IA en Producció
# Alumne: Joan Garcia

services:

  # Servei Ollama (model LLM local)
  ollama-joan-garcia:
    image: ollama/ollama:latest
    container_name: ollama-joan-garcia
    volumes:
      - ollama-joan-garcia-data:/root/.ollama
    ports:
      - "11434:11434"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:11434/api/tags"]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 30s
    restart: unless-stopped

  # Descàrrega del model Ollama (s'executa una vegada)
  ollama-pull-joan-garcia:
    image: ollama/ollama:latest
    container_name: ollama-pull-joan-garcia
    volumes:
      - ollama-joan-garcia-data:/root/.ollama
    depends_on:
      ollama-joan-garcia:
        condition: service_healthy
    entrypoint: >
      sh -c "ollama pull llama3.2:3b && echo 'Model descarregat correctament'"
    restart: "no"

  # API FastAPI principal
  api-ia-joan-garcia:
    build:
      context: .
      dockerfile: Dockerfile
      target: runtime
    container_name: api-ia-joan-garcia
    ports:
      - "8000:8000"
    environment:
      - ALUMNE=joan_garcia
      - ALUMNE_NOM_COMPLET=Joan Garcia
      - LOG_LEVEL=info
      - OLLAMA_URL=http://ollama-joan-garcia:11434
      - OLLAMA_MODEL=llama3.2:3b
      - OPENAI_API_KEY=${OPENAI_API_KEY:-}
    volumes:
      - model-cache-joan-garcia:/app/model_cache
    depends_on:
      ollama-joan-garcia:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/healthz"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 60s
    restart: unless-stopped
    labels:
      - "alumne=joan_garcia"
      - "practica=PR5073/02"

volumes:
  ollama-joan-garcia-data:
    name: ollama-joan-garcia-data
  model-cache-joan-garcia:
    name: model-cache-joan-garcia

---

Part 6: Tests automàtics

tests/test_api.py

# tests/test_api.py
# Pràctica PR5073/02 - Tests automatics de l'API
# Alumne: Joan Garcia

import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch, MagicMock
from app.main import app

# Client de test (no fa servir el servidor real)
client = TestClient(app)

ALUMNE = "joan_garcia"

# ============================================================
# TESTS DE SISTEMA
# ============================================================

class TestSistema:
    """Tests dels endpoints de sistema."""

    def test_arrel_endpoint(self):
        """L'endpoint arrel ha de retornar info de l'API."""
        response = client.get("/")
        assert response.status_code == 200
        data = response.json()
        assert data["alumne"] == ALUMNE
        assert "documentacio" in data
        assert "/api/v1/classify/sentiment" in data["endpoints_disponibles"]

    def test_health_check_retorna_ok(self):
        """El health check ha de retornar estat 'ok'."""
        response = client.get("/healthz")
        assert response.status_code == 200
        data = response.json()
        assert data["estat"] == "ok"
        assert data["alumne"] == ALUMNE
        assert "versio" in data
        assert "timestamp" in data

    def test_header_alumne_present(self):
        """Totes les respostes han d'incloure el header X-Alumne."""
        response = client.get("/healthz")
        assert "x-alumne" in response.headers
        assert response.headers["x-alumne"] == ALUMNE

    def test_documentacio_swagger_disponible(self):
        """La documentació Swagger ha d'estar disponible."""
        response = client.get("/docs")
        assert response.status_code == 200

    def test_openapi_json_disponible(self):
        """L'OpenAPI JSON ha d'estar disponible."""
        response = client.get("/openapi.json")
        assert response.status_code == 200
        data = response.json()
        assert ALUMNE in data["info"]["title"]

# ============================================================
# TESTS DE CLASSIFICACIO
# ============================================================

class TestClassificacio:
    """Tests dels endpoints de classificació."""

    @patch("app.routers.classify.obtenir_model_sentiment")
    def test_classificar_text_positiu(self, mock_model):
        """Ha de classificar correctament text positiu."""
        # Mock del model de Hugging Face
        mock_model.return_value = MagicMock(
            return_value=[{"label": "5 stars", "score": 0.92}]
        )

        response = client.post(
            "/api/v1/classify/sentiment",
            json={"text": "M'encanta aquest producte, es fantàstic!"}
        )

        assert response.status_code == 200
        data = response.json()
        assert data["etiqueta"] == "molt_positiu"
        assert data["puntuacio"] > 0.9
        assert data["alumne"] == ALUMNE

    @patch("app.routers.classify.obtenir_model_sentiment")
    def test_classificar_text_negatiu(self, mock_model):
        """Ha de classificar correctament text negatiu."""
        mock_model.return_value = MagicMock(
            return_value=[{"label": "1 stars", "score": 0.88}]
        )

        response = client.post(
            "/api/v1/classify/sentiment",
            json={"text": "Terrible, no funciona per res."}
        )

        assert response.status_code == 200
        data = response.json()
        assert data["etiqueta"] == "molt_negatiu"

    def test_classificar_text_buit_retorna_422(self):
        """Text buit ha de retornar error 422."""
        response = client.post(
            "/api/v1/classify/sentiment",
            json={"text": ""}
        )
        assert response.status_code == 422

    def test_classificar_text_massa_llarg_retorna_422(self):
        """Text de més de 5000 caracters ha de retornar error 422."""
        text_llarg = "a" * 5001
        response = client.post(
            "/api/v1/classify/sentiment",
            json={"text": text_llarg}
        )
        assert response.status_code == 422

    def test_classificar_sense_body_retorna_422(self):
        """Petició sense body ha de retornar error 422."""
        response = client.post("/api/v1/classify/sentiment")
        assert response.status_code == 422

# ============================================================
# TESTS DE GENERACIO
# ============================================================

class TestGeneracio:
    """Tests dels endpoints de generació."""

    @patch("app.routers.generate.requests.post")
    def test_generar_text_basic(self, mock_post):
        """Ha de generar text correctament."""
        mock_post.return_value = MagicMock(
            status_code=200,
            json=lambda: {
                "message": {"content": "La IA es la simulació de processos intel·ligents per ordinadors."}
            }
        )

        response = client.post(
            "/api/v1/generate",
            json={"prompt": "Explica que es la intel·ligencia artificial"}
        )

        assert response.status_code == 200
        data = response.json()
        assert len(data["text_generat"]) > 10
        assert data["alumne"] == ALUMNE
        assert "prompt_original" in data

    def test_generar_prompt_massa_curt_retorna_422(self):
        """Prompt de menys de 5 caracters ha de fallar."""
        response = client.post(
            "/api/v1/generate",
            json={"prompt": "Hola"}
        )
        assert response.status_code == 422

    def test_generar_temperatura_fora_de_rang_retorna_422(self):
        """Temperatura fora del rang [0, 2] ha de fallar."""
        response = client.post(
            "/api/v1/generate",
            json={"prompt": "Pregunta de prova", "temperatura": 5.0}
        )
        assert response.status_code == 422

---

Part 7: Execució i verificació

Construir i arrancar

# Build complet i arrancar tots els serveis
cd api-ia-joan-garcia
docker compose up --build -d

# Verificar que tots els serveis estan en marxa
docker compose ps

# Veure els logs en temps real
docker compose logs -f api-ia-joan-garcia

Provar l'API manualment

# 1. Verificar health check
curl http://localhost:8000/healthz | python -m json.tool

# 2. Classificar sentiment (text en catala)
curl -X POST http://localhost:8000/api/v1/classify/sentiment \
  -H "Content-Type: application/json" \
  -d '{"text": "Aquest curs m'\''ha semblat molt interessant i he après molt!"}' \
  | python -m json.tool

# 3. Generar text
curl -X POST http://localhost:8000/api/v1/generate \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Explica en catala els avantatges de FastAPI", "max_tokens": 200}' \
  | python -m json.tool

# 4. Documentació Swagger
open http://localhost:8000/docs

Executar tests

# Executar tots els tests
docker exec api-ia-joan-garcia pytest tests/ -v --tb=short

# Executar amb cobertura de codi
docker exec api-ia-joan-garcia pytest tests/ -v --cov=app --cov-report=html

# Veure el report de cobertura
# (el report es genera a htmlcov/index.html dins del contenidor)

Aturar els serveis

# Aturar i eliminar contenidors (manté els volums)
docker compose down

# Aturar, eliminar contenidors I volums (elimina els models descarregats!)
docker compose down -v

---

Preguntes de reflexió

Preguntes de reflexió

Responeu aquestes preguntes al document de memòria:

1. Pydantic vs dict: Per quin motiu usar schemas Pydantic en lloc de diccionaris simples per a validar les dades d'entrada i sortida de l'API?

2. Multi-stage Dockerfile: El Dockerfile usa dues fases (builder i runtime). Per quin motiu es una bona pràctica? Quina es la diferència de mida entre una imatge amb una sola fase i la imatge multi-stage?

3. Tests amb mocks: Per quin motiu els tests usen @patch per a simular els models d'IA en lloc de carregar els models reals?

4. Health checks: Per quin motiu es important tenir un endpoint /healthz? Qui el crida i quan?

5. Async vs sync: L'API usa async def per als endpoints. En quins casos l'ús d'async millora el rendiment i en quins casos no?

---

Lliurament

Que heu de lliurar

1. Codi font complet:
2. Tots els fitxers .py de app/ i tests/

3. Dockerfile, docker-compose.yml, requirements.txt

4. Captures de pantalla (5 obligatòries):

5. docker compose ps mostrant tots els serveis en marxa
6. Swagger UI (/docs) amb els endpoints documentats
7. Resultat d'una crida curl a /api/v1/classify/sentiment
8. Resultat d'una crida curl a /api/v1/generate

9. Resultat del pytest amb tots els tests passats

10. Memòria tècnica (PDF, 3-5 pàgines):

11. Descripció de l'arquitectura (diagrama inclòs)
12. Decisions de disseny: per quin motiu aquests endpoints, aquests models, aquesta estructura
13. Resultats dels tests (nombre, cobertura)
14. Respostes a les preguntes de reflexió

15. Millores futures proposades

16. Rúbrica emplenada (vegeu rubriques/rubrica_api_ia.md)

Format de lliurament

• Comprimir tot en un ZIP: PR5073/02_Joan_Garcia.zip
• Lliurar a la plataforma Moodle del curs

Rúbrica de correcció: PR5073/02 - Rúbrica