llm_call.py

import requests
import base64 
import cv2 as cv
import ollama

from openai import OpenAI

def send_to_gpt(img, prompt, examples=[]):
        api = "YOUR API KEY"

        # Complete chat for given examples
        example_chat = []

        if len(examples) != 0:
            for i in range(len(examples)):
                example_chat.append({
                    "role": "user",
                    "content": [
                        {
                        "type": "text",
                        "text": f"{prompt}"
                        },
                        {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{examples[i][0]}"
                        }
                        }
                    ]
                })
                example_chat.append({
                    "role": "assistant",
                    "content": f"{str(examples[i][1])}"
                })

        # Convert the frame to b64
        _, buffer = cv.imencode('.jpg', img)
        img_as_text = base64.b64encode(buffer).decode("UTF-8")

        headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api}"
        }

        payload = {
            "model": "gpt-4o-mini",
            "messages": [
                *example_chat,
                {
                    "role": "user",
                    "content": [
                        {
                        "type": "text",
                        "text": f"{prompt}"
                        },
                        {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{img_as_text}"
                        }
                        }
                    ]
                }
            ]
            #"max_tokens": 50
        }

        try:
            response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)
            return response.json()["choices"][0]["message"]["content"]

        except Exception as ex:
            return "Failed with exception: " + str(ex), "gpt-4o-mini"

def send_to_ollama(model, img, prompt, examples=[]):
    
    # Complete chat for given examples
    example_chat = []

    if len(examples) != 0:
        for i in range(len(examples)):
            example_chat.append({"role" : "user", "content" : prompt, "images" : [examples[i][0]]})
            example_chat.append({"role" : "assistant", "content" : str(examples[i][1])})

    # Encode the image to b64
    _, buffer = cv.imencode('.jpg', img)
    img_as_text = base64.b64encode(buffer).decode("UTF-8")

    # Send the frame to llava and catch any exception
    try:
        client = ollama.Client(
            host='http://host.docker.internal:11434',
            #headers={'x-some-header': 'some-value'}
        )
        response = client.chat(
            model=model, 
            messages=[
                *example_chat,
                {"role" : "user", "content" : prompt, "images" : [img_as_text]},
            ]
        )

        return response["message"]["content"]
    
    except Exception as ex:
        return "Failed with exception: " + str(ex)
    

def send_to_vllm(model, img, prompt):
    openai_api_key = "EMPTY"
    openai_api_base = "http://localhost:8000/v1"
    #openai_api_base = "https://notified-shoe-concern-digital.trycloudflare.com/v1"

    client = OpenAI(
        api_key=openai_api_key,
        base_url=openai_api_base,
    )

    _, buffer = cv.imencode('.jpg', img)
    img_as_text = base64.b64encode(buffer).decode("UTF-8")


    try:
        chat_completion_from_base64 = client.chat.completions.create(
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": prompt},
                        {
                            "type": "image_url",
                            "image_url": {"url": f"data:image/jpeg;base64,{img_as_text}"},
                        },
                    ],
                }
            ],
            model=model,
            max_completion_tokens=128,
        )

        result = chat_completion_from_base64.choices[0].message.content
        return result

    except Exception as ex:
        return "Failed with exception: " + str(ex)

def prompt_llm(llm, img, prompt, examples=[]):

    match llm:
        case "gpt-4o-mini":
            return send_to_gpt(img, prompt, examples)
        case "llava-latest":
            return send_to_ollama("llava", img, prompt, examples)
        case "gemma3-4b":
            return send_to_ollama("gemma3:4b", img, prompt, examples)
        case "llama3.2-vision-latest":
            return send_to_ollama("llama3.2-vision:latest", img, prompt, examples)
        case "Qwen2.5-VL-3B-Instruct":
            return send_to_vllm("Qwen/Qwen2.5-VL-3B-Instruct", img, prompt)
        case "Qwen2.5-VL-3B-Instruct-quantized.w8a8":
            return send_to_vllm("RedHatAI/Qwen2.5-VL-3B-Instruct-quantized.w8a8", img, prompt)
        case "Qwen2.5-VL-72B-Instruct":
            return send_to_vllm("Qwen/Qwen2.5-VL-72B-Instruct", img, prompt)
        case "Qwen2.5-VL-72B-Instruct-quantized.w8a8":
            return send_to_vllm("RedHatAI/Qwen2.5-VL-72B-Instruct-quantized.w8a8", img, prompt)

    return "Unknown LLM"

    #return send_to_ollama("llava", img, prompt, examples)
    #return send_to_ollama("bakllava", img, prompt, examples) # Poor results
    #return send_to_ollama("llava-llama3", img, prompt, examples)
    #return send_to_ollama("gemma3:4b", img, prompt, examples)
    #return send_to_ollama("llama3.2-vision", img, prompt, examples)
    #return send_to_ollama("mistral-small3.1", img, prompt, examples) # Min 14.5G Memory