Batch processing

https://raw.githubusercontent.com/perceptron-ai-inc/perceptron/main/cookbook/_shared/assets/capabilities/qna/studio_scene.webp
https://raw.githubusercontent.com/perceptron-ai-inc/perceptron/main/cookbook/_shared/assets/capabilities/detection/ppe_line.webp

import json
import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from itertools import islice
from pathlib import Path
from time import perf_counter

from perceptron import Client, SDKError, image


WORKERS = 4


def process_image(client, line_number, url):
    started = perf_counter()
    record = {"line": line_number}
    try:
        response = client.chat.completions.create(
            model="perceptron-mk1.5",
            messages=[{
                "role": "user",
                "content": [
                    image(url),
                    {"type": "text", "text": "Describe the visible objects in two sentences."},
                ],
            }],
            max_completion_tokens=512,
        )
        if not response.choices:
            raise ValueError("The response contained no completion")
        choice = response.choices[0]
        record["finish_reason"] = choice.finish_reason
        record["usage"] = response.usage.to_dict() if response.usage else None
        if choice.finish_reason != "stop" or choice.message.tool_calls:
            raise ValueError(f"Incomplete answer: {choice.finish_reason}")
        record.update(status="ok", text=choice.message.content or "")
    except (SDKError, ValueError) as error:
        record.update(status="error", error={
            "type": type(error).__name__,
            "message": str(error),
            "status_code": getattr(error, "status_code", None),
        })
    record["elapsed_seconds"] = round(perf_counter() - started, 3)
    return record


def run(path):
    client = Client(
        api_key=os.environ["PERCEPTRON_API_KEY"],
        provider="perceptron",
        timeout=60.0,
    )
    with path.open() as source, ThreadPoolExecutor(max_workers=WORKERS) as pool:
        items = ((number, line.strip()) for number, line in enumerate(source, 1) if line.strip())
        while group := list(islice(items, WORKERS)):
            futures = [pool.submit(process_image, client, number, url) for number, url in group]
            for future in as_completed(futures):
                print(json.dumps(future.result()), flush=True)


if __name__ == "__main__":
    if len(sys.argv) != 2:
        raise SystemExit("Usage: python batch_questions.py images.txt")
    run(Path(sys.argv[1]))

python batch_questions.py images.txt > results.jsonl