Hugging face hub VLM

hubchatv.py

import asyncio
from huggingface_hub import AsyncInferenceClient
import base64
import os

async def process_image(image_path, client, question):
    """Process a single image with the Hugging Face Async Inference Client."""
    with open(image_path, "rb") as f:
        base64_image = base64.b64encode(f.read()).decode("utf-8")
    image_url = f"data:image/jpeg;base64,{base64_image}"
    
    output = await client.chat.completions.create(
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {"url": image_url},
                    },
                    {
                        "type": "text",
                        "text": question,
                    },
                ],
            },
        ],
        max_tokens=4096,
    )
    
    print(f"--- Response for {image_path} ---")
    print(output.choices[0].message.content)
    print()  

async def main():
    # Input: Directory Path
    directory_path = input("Please enter the path to a directory containing images: ")
    
    # Input: Question
    question = input("Please enter your question about the image(s): ")
    
    # Hugging Face Async Inference Client
    client = AsyncInferenceClient("meta-llama/Llama-3.2-11B-Vision-Instruct")
    
    # Process images in the specified directory
    if os.path.isdir(directory_path):
        print(f"Processing directory: {directory_path}")
        tasks = []
        for filename in os.listdir(directory_path):
            file_path = os.path.join(directory_path, filename)
            if file_path.lower().endswith(('.jpg', '.jpeg', '.png', '.bmp', '.gif')):
                task = asyncio.create_task(process_image(file_path, client, question))
                tasks.append(task)
            else:
                print(f"Skipping non-image file: {filename}")
        await asyncio.gather(*tasks)
    else:
        print("Invalid directory path. Please ensure the path is correct and try again.")

asyncio.run(main())