AI-Powered Serverless Image Processing Pipeline — Part 4: Integrating AI Models for Image Enhancement (Python)

⏱ 8 min read  |  ~1628 words

AI‑Powered Serverless Image Processing Pipeline — Part 4: Integrating AI Models for Image Enhancement (Python)

In Part 1 we set up the Puter.com AI Gateway and provisioned a serverless bucket for raw uploads; Part 2 showed how to spin up a Lambda‑based thumbnail generator and push the results to a CDN. This fourth installment ties everything together by pulling a state‑of‑the‑art image‑enhancement model from Puter’s model zoo and running it inside a serverless GPU worker (Modal) so that every uploaded photo can be automatically upscaled, denoised, and color‑corrected before it lands in the final storage tier.

Why a dedicated AI model matters now

The 2026 “Ultimate Guide – The Best Serverless AI Deployment Solutions” from SiliconFlow highlights that Modal has become the de‑facto platform for on‑demand GPU workloads because it eliminates the need for a persistent EC2 fleet while still offering p4d‑class acceleration at a per‑invocation cost of <$0.10 per minute. combined with puter’s AI Model Marketplace, we can select a pre‑trained Super‑Resolution Transformer (SRT‑v2) that delivers 4× upscaling with research‑grade perceptual quality. The result is a fully serverless pipeline that rivals a traditional on‑prem GPU farm but scales to zero when there are no uploads.

Architecture Overview

Prerequisites

• A Puter.com account with API keys for Storage, AI Gateway, and Database.
• A Modal account (free tier works; just enable GPU).
• Python 3.11+ locally (or in a virtual environment).
• Basic familiarity with FastAPI, async programming, and PyTorch.

Step 1 – Install the SDKs

# Core Puter SDK (covers storage, AI gateway, DB)
pip install puter-sdk==2.4.1

# Modal Python client (GPU‑enabled runtimes)
pip install modal==0.62.0

# Image utilities
pip install pillow==10.2.0 tqdm==4.66.1

# PyTorch – pull the CUDA‑12.2 build (Modal handles the driver)
pip install torch==2.3.0+cu122 torchvision==0.18.0 --extra-index-url https://download.pytorch.org/whl/cu122

Step 2 – Register the SRT‑v2 Model in Puter

Navigate to Puter’s Model Marketplace and locate Super‑Resolution Transformer v2 (SRT‑v2). Click “Add to Project” and copy the generated MODEL_ID and ACCESS_TOKEN. For this tutorial we’ll store them as environment variables:

export PUTER_API_KEY="pk_live_XXXXXXXXXXXXXXXX"
export PUTER_MODEL_ID="srt-v2-4x"
export PUTER_MODEL_TOKEN="mt_XXXXXXXXXXXXXXXX"

Step 3 – Define the Modal GPU Function

Below is the heart of the pipeline. The function is annotated with @modal.function and requests a gpu runtime. It pulls the model via the Puter AI Gateway, runs inference, and writes the enhanced image back to a second bucket called processed-images. All I/O happens over Puter’s signed URLs, so we never expose credentials inside the worker.

import os
import json
import base64
import hashlib
from pathlib import Path
from io import BytesIO

import torch
from torchvision import transforms
from PIL import Image
import requests
import modal

# ----------------------------------------------------------------------
# Modal configuration – GPU‑enabled container
# ----------------------------------------------------------------------
gpu_image = modal.Image.debian_slim().pip_install(
    "torch==2.3.0+cu122",
    "torchvision==0.18.0",
    "pillow==10.2.0",
    "requests",
    "tqdm"
)

# Declare the Modal app
app = modal.App("image-enhance-pipeline", image=gpu_image)

# ----------------------------------------------------------------------
# Helper: fetch a signed URL from Puter for a given object key
# ----------------------------------------------------------------------
def get_signed_url(bucket: str, key: str, method: str = "GET", expires: int = 300) -> str:
    api_key = os.getenv("PUTER_API_KEY")
    endpoint = f"https://api.puter.com/v2/storage/{bucket}/{key}"
    payload = {
        "method": method,
        "expires_in": expires
    }
    headers = {"Authorization": f"Bearer {api_key}"}
    resp = requests.post(f"{endpoint}/sign", json=payload, headers=headers)
    resp.raise_for_status()
    return resp.json()["signed_url"]

# ----------------------------------------------------------------------
# Load the model via Puter AI Gateway (streaming to avoid large memory spikes)
# ----------------------------------------------------------------------
def load_puter_model():
    model_id = os.getenv("PUTER_MODEL_ID")
    token = os.getenv("PUTER_MODEL_TOKEN")
    gateway_url = f"https://api.puter.com/v2/ai/models/{model_id}/download"

    headers = {"Authorization": f"Bearer {token}"}
    # Streaming download – Modal’s container has enough disk (10 GB) for the ~2 GB checkpoint
    with requests.get(gateway_url, headers=headers, stream=True) as r:
        r.raise_for_status()
        tmp_path = Path("/tmp") / f"{model_id}.pt"
        with open(tmp_path, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)
    # Load into torch (map_location='cpu' first, then .to('cuda') after GPU is ready)
    checkpoint = torch.load(tmp_path, map_location="cpu")
    model = checkpoint["model"]
    model.eval()
    return model

# ----------------------------------------------------------------------
# Core enhancement function – this is what Modal will invoke
# ----------------------------------------------------------------------
@app.function(
    gpu="any",               # Any GPU that Modal can provision
    timeout=300,            # 5‑minute max per image (plenty for 4× upscaling)
    secrets=[
        modal.Secret.from_name("puter-api-key"),   # contains PUTER_API_KEY
        modal.Secret.from_name("puter-model-token") # contains PUTER_MODEL_TOKEN
    ],
    mounts=[
        modal.Mount.from_local_dir(".", remote_path="/root")
    ]
)
def enhance_image(event_payload: dict) -> dict:
    """
    event_payload – dict received from Puter webhook:
    {
        "bucket": "raw-images",
        "key": "uploads/2024/09/awesome-shot.jpg",
        "event_id": "evt_12345"
    }
    Returns a JSON‑serialisable dict for downstream logging.
    """
    # 1️⃣ Resolve signed URLs for input & output
    raw_url = get_signed_url(event_payload["bucket"], event_payload["key"], method="GET")
    processed_bucket = "processed-images"
    processed_key = f"enhanced/{event_payload['key']}"
    upload_url = get_signed_url(processed_bucket, processed_key, method="PUT")

    # 2️⃣ Download the raw image (stream into memory)
    resp = requests.get(raw_url, stream=True)
    resp.raise_for_status()
    raw_bytes = resp.content
    img = Image.open(BytesIO(raw_bytes)).convert("RGB")

    # 3️⃣ Prepare tensor & run model
    preprocess = transforms.Compose([
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.5]*3, std=[0.5]*3)  # model expects [-1, 1]
    ])
    input_tensor = preprocess(img).unsqueeze(0).to("cuda")   # shape: (1, 3, H, W)

    # Lazy‑load the model the first time the container boots
    if not hasattr(enhance_image, "model"):
        enhance_image.model = load_puter_model().to("cuda")
    with torch.no_grad():
        output_tensor = enhance_image.model(input_tensor)

    # 4️⃣ Post‑process back to Pillow image
    postprocess = transforms.Compose([
        transforms.Normalize(mean=[-1]*3, std=[2]*3),  # invert the earlier norm
        transforms.ToPILImage()
    ])
    enhanced_img = postprocess(output_tensor.squeeze(0).cpu())

    # 5️⃣ Encode to JPEG (quality 92) and upload via signed PUT URL
    out_buf = BytesIO()
    enhanced_img.save(out_buf, format="JPEG", quality=92)
    out_buf.seek(0)

    upload_resp = requests.put(upload_url, data=out_buf, headers={"Content-Type": "image/jpeg"})
    upload_resp.raise_for_status()

    # 6️⃣ Persist metadata (timestamp, model version, PSNR)
    metadata = {
        "event_id": event_payload["event_id"],
        "source_key": event_payload["key"],
        "dest_key": processed_key,
        "model_id": os.getenv("PUTER_MODEL_ID"),
        "processed_at": modal.get_current_time().isoformat(),
        "psnr": compute_psnr(img, enhanced_img)   # helper defined later
    }
    # Store in Puter DB (simple INSERT)
    store_metadata(metadata)

    return metadata

# ----------------------------------------------------------------------
# Helper: simple PSNR for sanity‑checking quality
# ----------------------------------------------------------------------
def compute_psnr(original: Image.Image, enhanced: Image.Image) -> float:
    orig = torch.tensor(np.array(original)).float()
    enh = torch.tensor(np.array(enhanced)).float()
    mse = torch.mean((orig - enh) ** 2)
    if mse == 0:
        return float("inf")
    max_pixel = 255.0
    psnr = 20 * torch.log10(max_pixel / torch.sqrt(mse))
    return round(psnr.item(), 2)

# ----------------------------------------------------------------------
# Helper: store processing metadata in Puter DB (PostgreSQL‑compatible)
# ----------------------------------------------------------------------
def store_metadata(meta: dict):
    db_url = os.getenv("PUTER_DATABASE_URL")  # e.g. postgres://user:pass@db.puter.com/dbname
    import sqlalchemy as sa
    engine = sa.create_engine(db_url)
    meta_table = sa.Table(
        "image_processing_log",
        sa.MetaData(),
        sa.Column("event_id", sa.String, primary_key=True),
        sa.Column("source_key", sa.String),
        sa.Column("dest_key", sa.String),
        sa.Column("model_id", sa.String),
        sa.Column("processed_at", sa.DateTime),
        sa.Column("psnr", sa.Float)
    )
    with engine.begin() as conn:
        conn.execute(
            meta_table.insert().values(**meta)
        )

Step 4 – Wire the Puter Webhook to Modal

Puter can push an HTTP POST to any public endpoint when a new object arrives. Modal makes this trivial by exposing a modal.WebEndpoint that runs inside the same app.

# Continue in the same file (app definition)

@app.function(
    cpu=1,
    timeout=60,
    secrets=[modal.Secret.from_name("puter-api-key")]
)
@modal.web_endpoint(method="POST")
def webhook_handler(request):
    """
    Expected payload (simplified):
    {
        "bucket": "raw-images",
        "key": "uploads/2024/09/awesome-shot.jpg",
        "event_id": "evt_12345"
    }
    """
    payload = request.json()
    # Fire‑and‑forget the enhancement – Modal returns 202 Accepted
    enhance_image.spawn(payload)  # asynchronous background call
    return modal.Response.json({"status": "accepted"}, status_code=202)

Deploy the entire app with a single command:

modal deploy image_enhance.py

Modal will output a public HTTPS URL (e.g. https://xyz123-modal.app). Copy that URL and register it in the Puter console under Storage → Webhooks → New Webhook. Choose the PUT event for the raw-images bucket and paste the Modal endpoint.

Step 5 – (Optional) Slack Notification on Completion

If you want real‑time visibility, add a tiny wrapper around store_metadata that posts to a Slack Incoming Webhook. The code below is deliberately isolated so you can toggle it with an environment flag SLACK_WEBHOOK_URL.

import os

SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL")

def notify_slack(meta: dict):
    if not SLACK_WEBHOOK:
        return
    message = {
        "text": f":art: Image enhancement complete! *{meta['source_key']}* → *{meta['dest_key']}*",
        "attachments": [
            {
                "fields": [
                    {"title": "Model", "value": meta["model_id"], "short": True},
                    {"title": "PSNR", "value": f"{meta['psnr']} dB", "short": True},
                    {"title": "Processed At", "value": meta["processed_at"], "short": True}
                ]
            }
        ]
    }
    requests.post(SLACK_WEBHOOK, json=message, timeout=5)

# Update store_metadata to call notify_slack
def store_metadata(meta: dict):
    # ... existing DB insert ...
    notify_slack(meta)

Step 6 – Testing Locally (Dry‑Run)

Before you push a production image, you can invoke the function locally using Modal’s modal run command. This spins up a temporary GPU container on Modal’s dev cluster, which is perfect for debugging.

modal run image_enhance.py::enhance_image --payload '{"bucket":"raw-images","key":"test/sample.jpg","event_id":"local-test"}'

The CLI will print the returned JSON metadata. Verify that:

• The psnr is > 30 dB (good quality for 4× upscaling).
• The destination object exists in processed-images and can be opened in any viewer.

Step 7 – Cost Monitoring & Scaling Tips

• Cold‑start mitigation: Puter’s model download (~2 GB) dominates the first invocation. To keep latency low for high‑traffic use‑cases, enable Modal’s prewarm flag (adds a warm container that lives for 15 min).
• GPU selection: For batch processing of > 5 MP images, request a gpu="A100" runtime; Modal automatically upgrades the instance.
• Budget alerts: Use Modal’s built‑in usage dashboard and set a monthly ceiling of $50 – the SiliconFlow guide notes that a single 4× upscaling costs roughly $0.018 per image on a p4d worker.

Full End‑to‑End Flow Recap

1. Client uploads a raw JPEG to raw-images via Puter’s signed URL.
2. Puter emits a webhook to the Modal /webhook_handler endpoint.
3. The webhook spawns enhance_image in a GPU container.
4. The worker streams the original image, loads the SRT‑v2 model from the AI Gateway, runs inference, and streams the enhanced result back to processed-images.
5. Metadata (including PSNR) is persisted to Puter DB; optional Slack notification is sent.

Debugging Checklist

<