HuggingFace to Livepeer (advanced paths)

export LP_AI_MODELS_DIR=/data/livepeer-ai-models
mkdir -p "$LP_AI_MODELS_DIR"

[
  {
    "pipeline": "text-to-image",
    "model_id": "SG161222/RealVisXL_V4.0_Lightning",
    "price_per_unit": 4768371,
    "pixels_per_unit": 1,
    "currency": "wei",
    "warm": true
  }
]

git clone https://github.com/livepeer/ai-worker.git
cd ai-worker

docker run --rm \
  -v "$LP_AI_MODELS_DIR:/models" \
  -v "$(pwd)/runner:/runner" \
  -e MODEL_DIR=/models \
  -e PIPELINE=text-to-image \
  -e MODEL_ID=SG161222/RealVisXL_V4.0_Lightning \
  livepeer/ai-runner:latest \
  bash /runner/dl_checkpoints.sh

ls -la "$LP_AI_MODELS_DIR/SG161222/RealVisXL_V4.0_Lightning/"

go-livepeer \
  -orchestrator \
  -transcoder \
  -nvidia all \
  -aiWorker \
  -aiModels /path/to/aiModels.json \
  -aiModelsDir "$LP_AI_MODELS_DIR" \
  -ethUrl <your-arbitrum-rpc> \
  -serviceAddr <your-public-host>:<port> \
  -pricePerUnit 0

curl -X POST http://localhost:<runner-port>/text-to-image \
  -H "Content-Type: application/json" \
  -d '{
    "model_id": "SG161222/RealVisXL_V4.0_Lightning",
    "prompt": "a quiet harbour at dawn, photo realistic",
    "width": 1024,
    "height": 1024,
    "num_inference_steps": 4,
    "guidance_scale": 2.0
  }' \
  --output result.json

go-livepeer \
  -gateway \
  -httpAddr 0.0.0.0:8935 \
  -orchAddr <your-orch-host>:<port> \
  -ethUrl <your-arbitrum-rpc>

curl -X POST http://localhost:8935/text-to-image \
  -H "Content-Type: application/json" \
  -d '{
    "model_id": "SG161222/RealVisXL_V4.0_Lightning",
    "prompt": "a quiet harbour at dawn, photo realistic",
    "width": 1024,
    "height": 1024,
    "num_inference_steps": 4,
    "guidance_scale": 2.0
  }' \
  --output gateway-result.json

mkdir my-pipeline
cd my-pipeline
uv init --lib

[project]
name = "my-pipeline"
version = "0.1.0"
requires-python = ">=3.10.12,<3.11"
dependencies = [
    "ai-runner[realtime]",
]

[project.scripts]
my-pipeline = "my_pipeline.main:main"

[tool.uv.sources]
ai-runner = { git = "https://github.com/livepeer/ai-worker.git", rev = "v0.14.0", subdirectory = "runner" }

[tool.uv]
package = true

[tool.setuptools.packages.find]
where = ["src"]

my-pipeline/
├── pyproject.toml
├── Dockerfile
└── src/
    └── my_pipeline/
        ├── __init__.py
        ├── main.py
        └── pipeline/
            ├── __init__.py
            ├── pipeline.py
            └── params.py

from runner.live.pipelines import BaseParams

class MyPipelineParams(BaseParams):
    prompt: str = "default prompt"
    # add fields your pipeline accepts at runtime

import asyncio
import logging
import os
from pathlib import Path

from runner.live.pipelines import Pipeline
from runner.live.trickle import VideoFrame, VideoOutput

from my_pipeline.pipeline.params import MyPipelineParams

class MyPipeline(Pipeline):
    name: str = "my-pipeline"

    def __init__(self):
        # initialise model state, load weights, set up CUDA streams
        ...

    @classmethod
    def prepare_models(cls):
        """Download and prepare model weights. Called when PREPARE_MODELS=1."""
        from huggingface_hub import snapshot_download

        models_dir = Path(os.environ.get("MODEL_DIR", "/models")) / "MyPipeline--models"
        models_dir.mkdir(parents=True, exist_ok=True)

        snapshot_download(
            "your-org/your-model",
            local_dir=models_dir / "your-model",
            local_dir_use_symlinks=False,
        )

    async def put_video_frame(self, frame: VideoFrame) -> None:
        # accept incoming frames, queue for inference
        ...

    async def get_processed_video_frame(self) -> VideoOutput:
        # return processed frames
        ...

    async def update_params(self, params: MyPipelineParams) -> None:
        # apply runtime parameter updates
        ...

    async def stop(self) -> None:
        # release GPU memory, close streams
        ...

from runner.app import start_app
from runner.live.pipelines import PipelineSpec

pipeline_spec = PipelineSpec(
    name="my-pipeline",  # MUST match the model_id used in go-livepeer
    pipeline_cls="my_pipeline.pipeline.pipeline:MyPipeline",
    params_cls="my_pipeline.pipeline.params:MyPipelineParams",
    initial_params={"prompt": "default prompt"},
)

def main():
    start_app(pipeline=pipeline_spec)

if __name__ == "__main__":
    main()

ARG BASE_IMAGE=livepeer/ai-runner:live-base-57efd92
FROM ${BASE_IMAGE}

WORKDIR /app

COPY pyproject.toml uv.lock ./
RUN mkdir -p src/my_pipeline/pipeline && \
    touch src/my_pipeline/__init__.py && \
    touch src/my_pipeline/pipeline/__init__.py
RUN uv sync --locked --no-install-project

COPY src/my_pipeline/ ./src/my_pipeline/
RUN uv sync --locked

ENV HF_HUB_OFFLINE=1

ARG GIT_SHA
ARG VERSION="undefined"
ENV GIT_SHA="${GIT_SHA}" \
    VERSION="${VERSION}"

CMD ["uv", "run", "--frozen", "my-pipeline"]

docker build -t my-org/my-pipeline:dev .

mkdir -p ./models
docker run --rm --gpus all \
  -v "$(pwd)/models:/models" \
  -e MODEL_DIR=/models \
  -e PREPARE_MODELS=1 \
  my-org/my-pipeline:dev

docker run --rm --gpus all \
  -p 8000:8000 \
  -v "$(pwd)/models:/models" \
  -e MODEL_DIR=/models \
  my-org/my-pipeline:dev

curl http://localhost:8000/health

AI_RUNNER_MY_PIPELINE_IMAGE=${AI_RUNNER_MY_PIPELINE_IMAGE:-my-org/my-pipeline}

function download_live_models() {
  case "$PIPELINE" in
  # existing cases...
  "my-pipeline")
    printf "\nPreparing my-pipeline models...\n"
    prepare_my_pipeline_models
    ;;
  "all")
    # existing code...
    prepare_my_pipeline_models
    ;;
  esac
}

function prepare_my_pipeline_models() {
  printf "\nPreparing my-pipeline models...\n"
  run_pipeline_prepare "my-pipeline" "$AI_RUNNER_MY_PIPELINE_IMAGE"
}

var livePipelineToImage = map[string]string{
    // existing entries...
    "my-pipeline": "my-org/my-pipeline",
}

[
  {
    "pipeline": "my-pipeline",
    "model_id": "your-org/your-model",
    "price_per_unit": 1,
    "pixels_per_unit": 1,
    "currency": "wei",
    "warm": true
  }
]

from fastapi import FastAPI
from pydantic import BaseModel
from huggingface_hub import snapshot_download
import os
from pathlib import Path

app = FastAPI()
model = None
model_loaded = False

class JobRequest(BaseModel):
    inputs: dict

class JobResponse(BaseModel):
    outputs: dict

def load_model():
    global model, model_loaded
    models_dir = Path(os.environ.get("MODEL_DIR", "/models"))
    models_dir.mkdir(parents=True, exist_ok=True)

    snapshot_download(
        "your-org/your-model",
        local_dir=models_dir / "your-model",
        local_dir_use_symlinks=False,
    )

    # actually load weights into VRAM
    model = ...  # your loaded model
    model_loaded = True

@app.on_event("startup")
async def startup():
    load_model()

@app.get("/health")
async def health():
    if not model_loaded:
        return {"status": "loading"}, 503
    return {"status": "ok"}

@app.post("/infer", response_model=JobResponse)
async def infer(request: JobRequest) -> JobResponse:
    outputs = model(request.inputs)
    return JobResponse(outputs=outputs)

FROM nvidia/cuda:12.1.0-cudnn8-runtime-ubuntu22.04

RUN apt-get update && apt-get install -y python3.10 python3-pip
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .

ENV PORT=8000
EXPOSE 8000

CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t my-org/byoc-pipeline:dev .

docker run --rm --gpus all -p 8000:8000 \
  -v "$(pwd)/models:/models" \
  my-org/byoc-pipeline:dev

# In another terminal:
curl http://localhost:8000/health
curl -X POST http://localhost:8000/infer \
  -H "Content-Type: application/json" \
  -d '{"inputs": {"your": "input"}}'

services:
  byoc-pipeline:
    image: my-org/byoc-pipeline:dev
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    volumes:
      - ./models:/models
    environment:
      - MODEL_DIR=/models
    ports:
      - "8000:8000"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

go-livepeer \
  -gateway \
  -httpAddr 0.0.0.0:8935 \
  -orchAddr <your-orch-host>:<port> \
  -ethUrl <your-arbitrum-rpc>