Workflow examples

Te kompletne skrypty używają Workflow API z zainstalowanej paczki cgcsdk. Utwórz pliki we własnym projekcie; dostęp do repozytoriów źródłowych CGC albo repozytorium dokumentacji nie jest wymagany.

Przygotowanie

python -m venv venv

Aktywuj środowisko i zainstaluj SDK:

# PowerShell
venv\Scripts\Activate.ps1
pip install cgcsdk

# Linux/macOS
source venv/bin/activate
pip install cgcsdk

Potwierdź wybrany kontekst CGC:

cgc context list

Każdy przykład poniżej tworzy prawdziwe zasoby CGC. Joby kończą się automatycznie. Trwałe wolumeny nie są usuwane automatycznie.

Jeden samodzielny job CSV

Jeden job tworzy CSV w kontenerze, przetwarza trzy rekordy i sprawdza sumę 4600 groszy. Nie wymaga pliku wejściowego, GPU ani PVC. Wynik jest wypisywany w logach, a pliki tymczasowe są usuwane.

job_workflow.py

job_workflow.py

#!/usr/bin/env python3
"""Run one CSV-processing job; temporary files are created inside its container.

Usage:
    python job_workflow.py --dry-run
    python job_workflow.py

No input file, PVC, GPU or extra packages are required. Results are printed in
job logs; temporary CSV files are removed before the job exits. A successful
job is deleted after its logs are retrieved. Failed jobs are left for diagnosis
and remain subject to the backend's finished-job retention policy.
"""

import argparse
import json

from cgc.sdk import App, Image, ShellCommand
from cgc.sdk.exceptions import SDKException


PROCESSING_CODE = r'''
import csv
import json
import tempfile
from pathlib import Path

with tempfile.TemporaryDirectory(prefix="cgc-csv-") as workdir:
    input_path = Path(workdir) / "input.csv"
    output_path = Path(workdir) / "processed.csv"
    rows = [
        {"product": "alpha", "quantity": 2, "unit_price_cents": 1000},
        {"product": "beta", "quantity": 3, "unit_price_cents": 700},
        {"product": "gamma", "quantity": 1, "unit_price_cents": 500},
    ]
    with input_path.open("w", newline="", encoding="utf-8") as stream:
        writer = csv.DictWriter(stream, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)

    with input_path.open(newline="", encoding="utf-8") as source:
        processed = [
            {
                "product": row["product"],
                "total_cents": int(row["quantity"]) * int(row["unit_price_cents"]),
            }
            for row in csv.DictReader(source)
        ]
    with output_path.open("w", newline="", encoding="utf-8") as stream:
        writer = csv.DictWriter(stream, fieldnames=["product", "total_cents"])
        writer.writeheader()
        writer.writerows(processed)

    with output_path.open(newline="", encoding="utf-8") as stream:
        saved = list(csv.DictReader(stream))
    expected = [("alpha", 2000), ("beta", 2100), ("gamma", 500)]
    actual = [(row["product"], int(row["total_cents"])) for row in saved]
    if actual != expected:
        raise RuntimeError("Processed CSV failed verification: " + repr(actual))
    summary = {"rows": len(saved), "total_cents": sum(total for _, total in actual)}

print("PROCESSING_RESULT " + json.dumps(summary), flush=True)
print("PROCESSING_OK", flush=True)
'''


app = App("csv-job-example", image=Image.from_registry("python:3.12-slim"))


@app.function(name="csv-process", cpu=1, memory=1, gpu=0)
def process_data():
    return ShellCommand(executable=("python", "-u", "-c", PROCESSING_CODE)).render()


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--dry-run", action="store_true", help="Print the plan without contacting CGC")
    parser.add_argument("--timeout", type=int, default=600, help="Job wait timeout in seconds")
    parser.add_argument("--poll-seconds", type=int, default=5)
    args = parser.parse_args(argv)
    if args.timeout <= 0 or args.poll_seconds <= 0:
        parser.error("--timeout and --poll-seconds must be positive")

    if args.dry_run:
        print(json.dumps(process_data.submit(dry_run=True), indent=2))
        return 0

    print("Creating one CSV job in the active CGC context; no GPU or PVC.")
    run = process_data.submit()
    print("Job:", run.name)
    try:
        status = run.wait(timeout_seconds=args.timeout, poll_seconds=args.poll_seconds)
        logs = run.logs(timestamps=True)
        text = logs if isinstance(logs, str) else json.dumps(logs)
        print(text)
        if status != "Succeeded" or "PROCESSING_OK" not in text:
            raise SDKException(-1, f"Job '{run.name}' did not pass verification. Status: {status}.")
    except Exception:
        print("Job left for diagnosis:", run.name)
        raise

    run.delete()
    print("CSV processing verified; successful job deleted:", run.name)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python job_workflow.py --timeout 600

Po sprawdzeniu wyniku skrypt usuwa własny job. Przy błędzie lub timeoutcie pozostawia go do diagnostyki. Do trwałego zapisania plików użyj wolumenu.

Job przetwarzający plik

Ten przykład domyślnie używa publicznego CSV z listą państw i przyjmuje opcjonalne argumenty:

Argument	Znaczenie
--input-url	Publiczny URL pliku tekstowego albo CSV, którego wiersze zostaną policzone. Domyślnie używa małego publicznego CSV z państwami.
--volume	Opcjonalna nazwa wolumenu CGC używanego do trwałego zapisania pliku z liczbą wierszy.
--volume-size	Pojemność wolumenu w GB, gdy brakujący wolumen ma zostać utworzony. Domyślnie 20.
--storage-class	Klasa storage dla nowego wolumenu; bez tej opcji używana jest skonfigurowana klasa domyślna.
--mount-path	Ścieżka w kontenerze, pod którą montowany jest wolumen. Domyślnie /data.
--output	Opcjonalna ścieżka pliku wyniku. Z --volume musi znajdować się wewnątrz --mount-path.
--delete-job	Usuwa wyłącznie job zakończony sukcesem po pobraniu logów.
--timeout-seconds	Maksymalny czas oczekiwania na job w sekundach. Domyślnie 300.

Skrypt tworzy krótko działający job z jednym rdzeniem CPU i 2 GB pamięci.

python process_file_workflow.py

Zapis wyniku na wolumenie CGC:

python process_file_workflow.py \
  --volume file-processing-data \
  --storage-class ssd-rwx-ec \
  --output /data/line-count.txt

--storage-class musi wskazywać klasę dostępną w Twoim środowisku; podaj ją, jeśli nie ustawiono klasy domyślnej.

Przekaż --input-url tylko wtedy, gdy chcesz użyć innego publicznego pliku tekstowego albo CSV.

process_file_workflow.py

process_file_workflow.py

#!/usr/bin/env python3
"""Download a text/CSV file in a CGC workflow job and count its lines."""

import argparse
import json
import shlex
import time
from pathlib import PurePosixPath

from cgc.sdk import App, Image, Volume


DEFAULT_INPUT_URL = "https://raw.githubusercontent.com/cs109/2014_data/master/countries.csv"


def wait_for_job(run, timeout_seconds: int, poll_seconds: int = 5):
    deadline = time.time() + timeout_seconds
    last_status = None

    while time.time() < deadline:
        status = run.status()
        if status != last_status:
            print("Job status:", status)
            last_status = status
        if status in {"Succeeded", "Failed"}:
            return status
        time.sleep(poll_seconds)

    return "Timeout"


def print_job_logs(run):
    response = run.logs(timestamps=True)
    logs = response.get("details", {}).get("logs", {})

    if not logs:
        print("No logs returned.")
        return

    print("Job logs:")
    for pod_name, containers in logs.items():
        for container_name, text in containers.items():
            print(f"[{pod_name}/{container_name}]")
            print(text.rstrip() or "<empty>")


def create_process_file_task(volume=None):
    app = App(
        name="file-processing-workflow",
        image=Image.from_registry("python:3.11-slim"),
        volumes=[volume] if volume else [],
    )


    @app.function(
        name="count-file-lines",
        cpu=1,
        memory=2,
        ttl_seconds_after_finished=3600,
    )
    def process_file(input_url: str, output_path: str = ""):
        code = f"""
from pathlib import Path
from urllib.request import urlopen

input_url = {json.dumps(input_url)}
output_path = {json.dumps(output_path)}

with urlopen(input_url, timeout=60) as response:
    text = response.read().decode("utf-8")

line_count = len(text.splitlines())
print(f"Downloaded: {{input_url}}")
print(f"Line count: {{line_count}}")

if output_path:
    path = Path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(str(line_count), encoding="utf-8")
    print(f"Saved line count to: {{path}}")
"""
        return f"python -c {shlex.quote(code)}"

    return process_file


def wait_for_volume(volume, timeout_seconds=300):
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        if volume.ready():
            return
        time.sleep(5)
    raise RuntimeError(f"Volume '{volume.name}' did not become ready")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Count lines in a remote text or CSV file using a CGC job."
    )
    parser.add_argument(
        "--input-url",
        default=DEFAULT_INPUT_URL,
        help="Public URL of the text or CSV file processed by the CGC job.",
    )
    parser.add_argument(
        "--volume",
        help="CGC volume name used to persist the result file.",
    )
    parser.add_argument(
        "--volume-size",
        type=int,
        default=20,
        help="Volume capacity in GB when a missing volume is created.",
    )
    parser.add_argument(
        "--storage-class",
        help="Storage class for a new volume; omit to use the configured default.",
    )
    parser.add_argument(
        "--mount-path",
        default="/data",
        help="Path where the volume is mounted inside the job container.",
    )
    parser.add_argument(
        "--output",
        help="Result file path. With a volume, it must be inside --mount-path.",
    )
    parser.add_argument(
        "--delete-job",
        action="store_true",
        help="Delete this job after successful completion and log retrieval.",
    )
    parser.add_argument(
        "--timeout-seconds",
        type=int,
        default=300,
        help="Maximum number of seconds to wait for the job.",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    if args.timeout_seconds <= 0:
        raise ValueError("--timeout-seconds must be positive")
    if args.volume:
        mount = PurePosixPath(args.mount_path)
        if not mount.is_absolute() or ".." in mount.parts:
            raise ValueError("--mount-path must be an absolute container path without '..'")
    volume = None
    output_path = args.output or ""

    if args.volume:
        volume = Volume.from_name(
            name=args.volume,
            mount_path=args.mount_path,
            size=args.volume_size,
            storage_class=args.storage_class,
            create_if_missing=True,
        )
        if not output_path:
            output_path = str(PurePosixPath(args.mount_path) / "line-count.txt")
        if ".." in PurePosixPath(output_path).parts or not PurePosixPath(output_path).is_relative_to(
            PurePosixPath(args.mount_path)
        ):
            raise ValueError("--output must be inside --mount-path")
        volume.ensure()
        wait_for_volume(volume, args.timeout_seconds)
    elif output_path:
        print(
            "Warning: no volume was selected. The output file will disappear "
            "when the job container is removed."
        )

    process_file = create_process_file_task(volume)
    run = process_file.submit(args.input_url, output_path)
    print("Created job:", run.name)

    status = wait_for_job(run, timeout_seconds=args.timeout_seconds)

    print("Final status:", status)

    if status == "Timeout":
        print("The job did not finish before the timeout.")
        print("Check platform events with:")
        print(f"cgc events {run.name}")
    else:
        print_job_logs(run)

    if status != "Succeeded":
        raise RuntimeError(f"Job '{run.name}' ended with status {status}; left for diagnosis")

    if args.delete_job:
        run.delete()
        print("Deleted job:", run.name)


if __name__ == "__main__":
    main()

Job przetwarzający obrazy w PyTorch

Ten workflow używa dwóch plików:

• process_image.py jest przesyłany na wolumen image-processing-data.
• run_image_workflow.py tworzy joby uruchamiające przesłany skrypt.

process_image.py przyjmuje:

Argument	Znaczenie
--input	Ścieżka do źródłowego JPEG albo PNG wewnątrz kontenera joba.
--output	Ścieżka, pod którą zostanie zapisany przetworzony obraz.

run_image_workflow.py tworzy albo ponownie używa wolumenu, montuje go do Filebrowser, prosi o przesłanie plików i uruchamia process_image.py.

python run_image_workflow.py

Uruchomienie dla wielu obrazów:

python run_image_workflow.py \
  --input /data/photo-a.jpg \
  --output /data/photo-a-processed.jpg \
  --input /data/photo-b.jpg \
  --output /data/photo-b-processed.jpg

process_image.py

process_image.py

import argparse

import numpy as np
import torch
from PIL import Image


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input",
        required=True,
        help="Path to the source JPEG or PNG inside the job container.",
    )
    parser.add_argument(
        "--output",
        required=True,
        help="Path where the processed image will be saved.",
    )
    args = parser.parse_args()

    image = Image.open(args.input).convert("RGB")
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    pixels = torch.from_numpy(np.array(image)).to(device)
    processed = 255 - pixels

    output = Image.fromarray(processed.cpu().numpy().astype(np.uint8))
    output.save(args.output)

    print(f"Processed image on {device} and saved it to {args.output}")


if __name__ == "__main__":
    main()

run_image_workflow.py

run_image_workflow.py

import argparse
import shlex
import time
from pathlib import PurePosixPath

from cgc.sdk import App, Image, Volume, volume as volume_api
from cgc.sdk.exceptions import SDKException


VOLUME_NAME = "image-processing-data"
MOUNT_PATH = "/data"
FILEBROWSER_NAME = "filebrowser"


def is_mount_details(value):
    return (
        isinstance(value, dict)
        and value.get("pvc_name") == VOLUME_NAME
        and value.get("target_template_name") == FILEBROWSER_NAME
        and "mount_path" in value
    )


def get_mount_details_from_error(error):
    for value in error.args:
        if is_mount_details(value):
            return value
        if isinstance(value, dict) and is_mount_details(value.get("details")):
            return value["details"]
    return None


def prepare_volume():
    data = Volume.from_name(
        name=VOLUME_NAME,
        mount_path=MOUNT_PATH,
        size=2,
        storage_class="ssd-rwx-ec",
        create_if_missing=True,
    )
    data.ensure()
    deadline = time.monotonic() + 300
    while not data.ready():
        if time.monotonic() >= deadline:
            raise RuntimeError("Volume did not become ready: " + VOLUME_NAME)
        time.sleep(5)

    try:
        response = volume_api.volume_mount(
            name=VOLUME_NAME,
            target=FILEBROWSER_NAME,
        )
    except SDKException as error:
        details = get_mount_details_from_error(error)
        if details is None and error.reason == "PVC_MOUNT_ALREADY_MOUNTED":
            print("Volume is already mounted to Filebrowser.")
        elif details is not None:
            print(
                "Volume is mounted to Filebrowser. "
                f"Filebrowser mount path: {details['mount_path']}"
            )
        else:
            raise
    else:
        details = response.get("details", {})
        if response["code"] != 200 and not is_mount_details(details):
            raise RuntimeError(f"Could not mount volume to Filebrowser: {response}")
        if "mount_path" in details:
            print("Filebrowser mount path:", details["mount_path"])

    return data


def create_process_image_task(data):
    app = App(
        name="image-processing-workflow",
        image=Image.from_registry("pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime"),
        volumes=[data],
    )

    @app.function(
        name="process-image",
        cpu=4,
        memory=16,
        gpu=1,
        gpu_type="nvidia-rtx-a5000",
    )
    def process_image(input_path, output_path):
        return (
            "test -f /data/process_image.py"
            f" && test -f {shlex.quote(input_path)}"
            " && python -m pip install --quiet Pillow"
            " && python /data/process_image.py"
            f" --input {shlex.quote(input_path)}"
            f" --output {shlex.quote(output_path)}"
        )

    return process_image

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input",
        action="append",
        help="Input image path inside the job container. Repeat for multiple images.",
    )
    parser.add_argument(
        "--output",
        action="append",
        help="Output image path inside the job container. Repeat for multiple images.",
    )
    args = parser.parse_args()
    input_paths = args.input or ["/data/input.jpg"]
    output_paths = args.output or ["/data/output.jpg"]

    if len(input_paths) != len(output_paths):
        parser.error("Provide the same number of --input and --output values.")

    for path in input_paths + output_paths:
        candidate = PurePosixPath(path)
        if ".." in candidate.parts or not candidate.is_relative_to(MOUNT_PATH):
            parser.error("Input and output paths must be inside " + MOUNT_PATH)

    data = prepare_volume()
    process_image = create_process_image_task(data)

    print("Volume is ready and mounted to Filebrowser:", VOLUME_NAME)
    print("Open Filebrowser, open the mounted volume, and upload to its root:")
    print("- process_image.py")
    for input_path in input_paths:
        print(f"- {input_path.removeprefix(MOUNT_PATH + '/')}")
    print()
    print("Use this command to find the Filebrowser URL and app token:")
    print("cgc compute list -d")
    print("Sign in with username 'admin' and the displayed app token.")
    input("Press Enter after all files are uploaded...")

    runs = process_image.for_each(input_paths, output_paths)
    failed = []

    for run in runs:
        status = run.wait(timeout_seconds=600)
        print("Job name:", run.name)
        print("Final status:", status)
        print(run.logs(timestamps=True))

        if status != "Succeeded":
            failed.append(run.name)
        else:
            run.delete()
            print("Deleted successful job:", run.name)

    if failed:
        raise RuntimeError(f"Failed jobs: {', '.join(failed)}")


if __name__ == "__main__":
    main()

Długodziałający serwis HTTP

Skrypt deklaruje text-analysis-service przez @app.inference(). Zwraca ustrukturyzowany ShellCommand, wdraża zasób i jego port api, czeka na gotowość HTTP, wysyła request do /analyze i opcjonalnie usuwa zasób. Aplikacja obsługuje zarówno /health, jak i /api/health, ponieważ URL ingress może już zawierać nazwę portu jako prefiks ścieżki.

python text_inference.py --delete

text_inference.py

text_inference.py

import argparse

from cgc.sdk import App, Image, ShellCommand


RESOURCE_NAME = "text-analysis-service"
PORT = 8000
PORT_NAME = "api"

app = App(
    name="http-service-workflow",
    image=Image.from_registry("python:3.11-slim"),
)


@app.inference(
    entity="custom",
    name=RESOURCE_NAME,
    port=PORT,
    port_name=PORT_NAME,
    cpu=1,
    memory=2,
)
def text_analysis_service():
    code = r"""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json


class Handler(BaseHTTPRequestHandler):
    def _send_json(self, status_code, payload):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status_code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path in {"/health", "/api/health"}:
            self._send_json(200, {"status": "ok"})
            return
        self._send_json(404, {"error": "not found", "path": self.path})

    def do_POST(self):
        if self.path not in {"/analyze", "/api/analyze"}:
            self._send_json(404, {"error": "not found", "path": self.path})
            return

        length = int(self.headers.get("Content-Length", "0"))
        payload = json.loads(self.rfile.read(length) or b"{}")
        text = payload.get("text", "")
        words = text.split()
        self._send_json(
            200,
            {
                "input": text,
                "characters": len(text),
                "words": len(words),
                "uppercase": text.upper(),
                "first_five_words": words[:5],
            },
        )

    def log_message(self, format, *args):
        print(format % args, flush=True)


ThreadingHTTPServer(("0.0.0.0", 8000), Handler).serve_forever()
"""
    return ShellCommand(executable=("python", "-c", code))


def parse_args():
    parser = argparse.ArgumentParser(
        description="Deploy and call a long-running HTTP service on CGC."
    )
    parser.add_argument(
        "--text",
        default="CGC SDK workflows can run long-running HTTP services.",
        help="Text sent to the /analyze endpoint.",
    )
    parser.add_argument(
        "--wait-timeout",
        type=int,
        default=600,
        help="How long to wait for the resource and HTTP endpoint, in seconds.",
    )
    parser.add_argument(
        "--delete",
        action="store_true",
        help="Delete the CGC resource at the end of the script.",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    endpoint = None
    if args.wait_timeout <= 0:
        raise ValueError("--wait-timeout must be positive")

    try:
        endpoint = text_analysis_service.deploy(
            wait=True,
            reuse_existing=False,
            resource_timeout_seconds=args.wait_timeout,
        )
        print("Endpoint:", endpoint.url)

        health = endpoint.wait_http(
            "/health",
            timeout_seconds=args.wait_timeout,
        ).json()
        print("Health:", health)

        payload = {"text": args.text}
        print("Sending text for processing:", payload)
        result = endpoint.post_json("/analyze", payload, timeout=30)
        print("Processed response:", result)
    except Exception:
        if endpoint is not None:
            print("Inspect resource logs with endpoint.logs(timestamps=True):", endpoint.name)
        raise
    finally:
        if args.delete and endpoint is not None:
            endpoint.delete()
            print(f"Deleted resource: {RESOURCE_NAME}")
        elif endpoint is not None:
            print(
                f"Resource left running: {RESOURCE_NAME}. "
                "Rerun with --delete when you want to remove it."
            )


if __name__ == "__main__":
    main()

Pobieranie datasetu Roboflow

Przed uruchomieniem ustaw ROBOFLOW_API_KEY. Klucz nie jest przyjmowany jako argument CLI i nie powinien być wpisywany bezpośrednio w skrypcie.

Argument	Znaczenie
--workspace-id	Identyfikator workspace Roboflow.
--project-id	Identyfikator projektu Roboflow.
--version	Numeryczna wersja datasetu.
--format	Format eksportu; domyślnie yolov11.

Skrypt tworzy albo ponownie używa wolumenu object-detection-data i zachowuje go po pobraniu.

python roboflow_dataset.py \
  --workspace-id your-workspace \
  --project-id your-project \
  --version 1

roboflow_dataset.py

roboflow_dataset.py

import argparse
import os
import time

from cgc.sdk import Image, Volume
from cgc.sdk.recipes.datasets import RoboflowDataset
from cgc.sdk.recipes.yolo import YoloConfig, YoloWorkflow


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--workspace-id", required=True)
    parser.add_argument("--project-id", required=True)
    parser.add_argument("--version", required=True, type=int)
    parser.add_argument("--format", default="yolov11")
    args = parser.parse_args()

    try:
        api_key = os.environ["ROBOFLOW_API_KEY"]
    except KeyError as error:
        raise RuntimeError(
            "Set the ROBOFLOW_API_KEY environment variable before running."
        ) from error

    data = Volume.from_name(
        name="object-detection-data",
        mount_path="/root/data",
        size=50,
        storage_class="ssd-rwx-ec",
        create_if_missing=True,
    )
    data.ensure()
    deadline = time.monotonic() + 300
    while not data.ready():
        if time.monotonic() >= deadline:
            raise RuntimeError("Dataset volume did not become ready")
        time.sleep(5)

    dataset = RoboflowDataset(
        workspace_id=args.workspace_id,
        project_id=args.project_id,
        version=args.version,
        format=args.format,
        api_key=api_key,
    )

    workflow = YoloWorkflow(
        name="yolo-training",
        image=Image.from_registry("ultralytics/ultralytics:latest"),
        volume=data,
        config=YoloConfig(),
    )

    run = workflow.download(dataset).submit()
    status = run.wait(timeout_seconds=1800)
    print("Download status:", status)
    print(run.logs(timestamps=True))

    if status != "Succeeded":
        raise RuntimeError(f"Dataset download ended with status: {status}; job left for diagnosis: {run.name}")
    run.delete()
    print("Deleted successful download job:", run.name)


if __name__ == "__main__":
    main()

Pretrained model inference

Skrypty YOLO i RT-DETR używają publicznego URL obrazu, wypisują detekcje i zapisują opisany obraz lokalnie. Usuń endpoint, kiedy nie jest już potrzebny.

Skrypt	Model	Plik wyniku
yolo_inference.py	yolo11n.pt	yolo_prediction.jpg
rtdetr_inference.py	rtdetr-l.pt	rtdetr_prediction.jpg

Oba przykłady proszą o jedno GPU nvidia-rtx-a5000, cztery rdzenie CPU i 16 GB pamięci.

python yolo_inference.py
python rtdetr_inference.py

yolo_inference.py

yolo_inference.py

import base64
from pathlib import Path

from cgc.sdk import Image
from cgc.sdk.recipes.yolo import YoloConfig, YoloWorkflow


SAMPLE_IMAGE_URL = "https://ultralytics.com/images/bus.jpg"
OUTPUT_IMAGE_PATH = Path("yolo_prediction.jpg")
RESOURCE_NAME = "sdk-yolo-inference-demo"


def create_workflow():
    return YoloWorkflow(
        name=RESOURCE_NAME,
        image=Image.from_registry("ultralytics/ultralytics:latest"),
        config=YoloConfig(
            model_name="yolo11n.pt",
            inference_cpu=4,
            inference_memory=16,
            inference_gpu=1,
            inference_gpu_type="nvidia-rtx-a5000",
        ),
    )


def save_prediction_image(prediction: dict, path: Path = OUTPUT_IMAGE_PATH):
    image_bytes = base64.b64decode(prediction["annotated_image_base64"])
    path.write_bytes(image_bytes)
    return path


if __name__ == "__main__":
    workflow = create_workflow()
    print("Creating or reusing CGC resource and waiting for it to start...")
    endpoint = workflow.deploy_pretrained_inference(wait=True, reuse_existing=True)
    print("Endpoint:", endpoint.url)

    print("Waiting for inference service to become ready...")
    endpoint.wait_http("/health")

    print("Sending prediction request...")
    prediction = endpoint.post_json(
        "/predict",
        {"image_url": SAMPLE_IMAGE_URL},
        timeout=120,
    )

    print("Model:", prediction["model"])
    print("Detections:")
    for detection in prediction["detections"]:
        print(
            "- {class_name}: {confidence:.2f}".format(
                class_name=detection["class_name"],
                confidence=detection["confidence"],
            )
        )
    saved_path = save_prediction_image(prediction)
    print("Saved annotated image:", saved_path.resolve())

rtdetr_inference.py

rtdetr_inference.py

import base64
from pathlib import Path

from cgc.sdk import Image
from cgc.sdk.recipes.rtdetr import RTDetrConfig, RTDetrWorkflow


SAMPLE_IMAGE_URL = "https://ultralytics.com/images/bus.jpg"
OUTPUT_IMAGE_PATH = Path("rtdetr_prediction.jpg")
RESOURCE_NAME = "sdk-rtdetr-inference-demo"


def create_workflow():
    return RTDetrWorkflow(
        name=RESOURCE_NAME,
        image=Image.from_registry("ultralytics/ultralytics:latest"),
        config=RTDetrConfig(
            model_name="rtdetr-l.pt",
            inference_cpu=4,
            inference_memory=16,
            inference_gpu=1,
            inference_gpu_type="nvidia-rtx-a5000",
        ),
    )


def save_prediction_image(prediction: dict, path: Path = OUTPUT_IMAGE_PATH):
    image_bytes = base64.b64decode(prediction["annotated_image_base64"])
    path.write_bytes(image_bytes)
    return path


if __name__ == "__main__":
    workflow = create_workflow()
    print("Creating or reusing CGC resource and waiting for it to start...")
    endpoint = workflow.deploy_pretrained_inference(wait=True, reuse_existing=True)
    print("Endpoint:", endpoint.url)

    print("Waiting for inference service to become ready...")
    endpoint.wait_http("/health")

    print("Sending prediction request...")
    prediction = endpoint.post_json(
        "/predict",
        {"image_url": SAMPLE_IMAGE_URL},
        timeout=120,
    )

    print("Model:", prediction["model"])
    print("Detections:")
    for detection in prediction["detections"]:
        print(
            "- {class_name}: {confidence:.2f}".format(
                class_name=detection["class_name"],
                confidence=detection["confidence"],
            )
        )
    saved_path = save_prediction_image(prediction)
    print("Saved annotated image:", saved_path.resolve())

Szczegółowe wyjaśnienie wszystkich argumentów Workflow API znajdziesz w Workflow overview.