Skip to content

Code examples

Python examples for calling peach-models. They use httpx and, for the OpenAI-style routes, the openai package. Install them with pip install httpx openai.

All examples read your credentials from the environment:

import os

BASE_URL = "https://api.models.peach.ebu.io"
CODOPS = os.environ["CODOPS"]
TOKEN = os.environ[f"CODOPS_{CODOPS.upper()}_SECRET"]
HEADERS = {"Authorization": f"Bearer {CODOPS}:{TOKEN}"}

On the Peach cluster and in PEACH Lab, CODOPS and CODOPS_<CODOPS>_SECRET are already set. Never put a token in your code or in a notebook you share.

Inside a Peach task or endpoint, the pipe_algorithms_lib.deep_serve helpers do most of this for you; see the overview.

Embeddings

Send texts as a list, up to 2048 per call:

import httpx

texts = ["Election results in Sweden", "Weather forecast for Geneva"]

resp = httpx.post(
    f"{BASE_URL}/v1/embeddings",
    headers=HEADERS,
    json={"model": "labse-embedding", "input": texts},
    timeout=70,
)
resp.raise_for_status()
vectors = [item["embedding"] for item in resp.json()["data"]]
print(len(vectors), "vectors of size", len(vectors[0]))

The same call with the OpenAI client:

from openai import OpenAI

client = OpenAI(base_url=f"{BASE_URL}/v1", api_key=f"{CODOPS}:{TOKEN}")
result = client.embeddings.create(model="labse-embedding", input=texts)
vectors = [item.embedding for item in result.data]

Chat LLM

from openai import OpenAI

client = OpenAI(base_url=f"{BASE_URL}/v1", api_key=f"{CODOPS}:{TOKEN}", timeout=900)

models = [m.id for m in client.models.list().data]
print(models)  # pick the LLM name from this list

answer = client.chat.completions.create(
    model="<llm model name>",
    messages=[{"role": "user", "content": "Summarise this article in two sentences: ..."}],
    max_tokens=300,
)
print(answer.choices[0].message.content)

Translation

The translation route takes one text per request:

import httpx

resp = httpx.post(
    f"{BASE_URL}/v1/translations",
    headers=HEADERS,
    json={
        "model": "translategemma-4b",
        "text": "The match was postponed because of heavy snow.",
        "source_lang": "en",
        "target_lang": "fr",
    },
    timeout=310,
)
resp.raise_for_status()
print(resp.json()["translated_text"])
print("server time:", resp.headers.get("Server-Timing"))

To translate a long document quickly, split it into paragraphs and send them in parallel. This is much faster than sending the whole document at once (see How models run). The next example does that.

Parallel requests with retries

This helper sends many requests in parallel, keeps no more than max_parallel in flight, and retries the errors that are worth retrying (see Limits and errors).

import asyncio
import random

import httpx

RETRY_STATUS = {429, 503, 504}


async def post_with_retry(client, sem, path, payload, attempts=5):
    """POST with a parallel-request limit and retries for 429/503/504."""
    for attempt in range(attempts):
        async with sem:  # hold a slot only while the request is in flight
            try:
                resp = await client.post(path, json=payload)
            except httpx.TransportError:
                resp = None  # network error: retry below
        if resp is not None and resp.status_code not in RETRY_STATUS:
            resp.raise_for_status()  # 400/401/404/422/500: don't loop on these
            return resp.json()
        if attempt == attempts - 1:
            break
        # Wait for the longer of the server's Retry-After and our own backoff.
        delay = 2**attempt
        retry_after = resp.headers.get("Retry-After") if resp is not None else None
        if retry_after and retry_after.isdigit():
            delay = max(delay, int(retry_after))
        await asyncio.sleep(delay + random.uniform(0, 1))
    if resp is not None:
        resp.raise_for_status()
    raise RuntimeError(f"{path} failed after {attempts} attempts")


async def translate_document(paragraphs, source_lang, target_lang,
                             model="translategemma-4b", max_parallel=16):
    sem = asyncio.Semaphore(max_parallel)
    async with httpx.AsyncClient(base_url=BASE_URL, headers=HEADERS, timeout=310) as client:
        tasks = [
            post_with_retry(client, sem, "/v1/translations", {
                "model": model,
                "text": paragraph,
                "source_lang": source_lang,
                "target_lang": target_lang,
            })
            for paragraph in paragraphs
        ]
        results = await asyncio.gather(*tasks)
    return [r["translated_text"] for r in results]


document = open("article.txt", encoding="utf-8").read()
paragraphs = [p.strip() for p in document.split("\n\n") if p.strip()]
translated = asyncio.run(translate_document(paragraphs, "en", "de"))
print("\n\n".join(translated))

Choose max_parallel to match the model:

Model max_parallel
translategemma-4b 16 to 20
translategemma-12b 8
models that take one request at a time (embeddings, speech to text, ...) 2 to 4

In a notebook, where an event loop is already running, use translated = await translate_document(...) instead of asyncio.run(...).

The same helper works for any route; for example, embeddings in chunks of 256 texts:

async def embed_all(texts, model="labse-embedding", chunk=256, max_parallel=3):
    sem = asyncio.Semaphore(max_parallel)
    async with httpx.AsyncClient(base_url=BASE_URL, headers=HEADERS, timeout=70) as client:
        chunks = [texts[i:i + chunk] for i in range(0, len(texts), chunk)]
        results = await asyncio.gather(*[
            post_with_retry(client, sem, "/v1/embeddings", {"model": model, "input": c})
            for c in chunks
        ])
    return [item["embedding"] for r in results for item in r["data"]]

Bulk work with priority and a webhook

The native routes accept priority and webhook. With a webhook, a request that is not finished in time returns "status": "pending" and the result is posted to your URL later:

import base64

import httpx

with open("interview.mp3", "rb") as f:
    audio_b64 = base64.b64encode(f.read()).decode()

resp = httpx.post(
    f"{BASE_URL}/models/whisper-large-v3/transcribe",
    json={
        "codops": CODOPS,
        "token": TOKEN,
        "audio_data": audio_b64,  # or "audio_url": "https://..." for a file we can download
        "priority": "batch",
        "webhook": "https://your-service.example/peach-models/callback",
    },
    timeout=330,
)
resp.raise_for_status()
body = resp.json()
if body["status"] == "pending":
    print("result will be posted to the webhook, request id", body["request_id"])
else:
    print(body)

The webhook receives a JSON POST with the request_id, a status of completed or error, and the result fields of that model type. The URL must be reachable from the Peach cluster. Check each route's exact fields in the interactive reference.

Measuring your own calls

To tune chunk size and parallel requests, log the Server-Timing header next to your own timing:

import time

start = time.perf_counter()
resp = httpx.post(f"{BASE_URL}/v1/translations", headers=HEADERS, json=payload, timeout=310)
client_ms = (time.perf_counter() - start) * 1000
print(f"client {client_ms:.0f} ms | server {resp.headers.get('Server-Timing')} | chars {len(payload['text'])}")

Limits and errors explains how to read the numbers.