写在前面

对于很多人来说，编程难在不知道如何起步
对于一个几乎完全陌生的领域，我们要如何应对呢?
如何通过 AI 来解决不擅长的内容呢

Classify Images with OpenCLIP

July 8, 2024
https://chatgpt.com/share/dc8e2ac6-a918-44ab-b967-284373acb624

Classify Images with OpenCLIP

🧑‍🦰User

如何使用 open_clip 进行分类

import torch
from PIL import Image
import open_clip

model, _, preprocess = open_clip.create_model_and_transforms('ViT-g-14', pretrained='laion2b_s12b_b42k')
model.eval()  # model in train mode by default, impacts some models with BatchNorm or stochastic depth active                                                                                                
tokenizer = open_clip.get_tokenizer('ViT-g-14')

image = preprocess(Image.open("test/ts.jpg")).unsqueeze(0)
text = tokenizer(["a picture of taylor swift", "a picture of AOC"])

with torch.no_grad(), torch.cuda.amp.autocast():
    image_features = model.encode_image(image)
    text_features = model.encode_text(text)
    image_features /= image_features.norm(dim=-1, keepdim=True)
    text_features /= text_features.norm(dim=-1, keepdim=True)

    text_probs = (100.0 * image_features @ text_features.T).softmax(dim=-1)

print("Label probs:", text_probs)  # prints: [[1., 0., 0.]]

编写一个响应 HTTP 请求的服务，并接收给定的一组提示词和一组图像，将每张图像与每个提示词进行分类。

🤖 Assistant

要创建一个响应 HTTP 请求的服务，使用 open_clip 模型对给定的提示词进行图像分类，可以使用像 Flask 或 FastAPI 这样的 Web 服务器框架。以下是使用 FastAPI 的示例，它是一个现代的、高性能的、易于使用的构建 API 的框架。

此服务将执行以下任务：

接收包含图像和提示词的 HTTP POST 请求。
使用 open_clip 模型对图像进行提示词分类。
返回分类概率。

步骤 1：设置环境

你需要安装 Python，然后使用 pip 安装所需的包：

1	pip install fastapi[all] torch pillow open_clip

步骤 2：创建 FastAPI 应用程序

创建一个名为 clip_server.py 的文件，并添加以下代码：

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
import torch
from PIL import Image
import open_clip
from io import BytesIO
import numpy as np

app = FastAPI()

# Load the CLIP model
model, _, preprocess = open_clip.create_model_and_transforms('ViT-g-14', pretrained='laion2b_s12b_b42k')
model.eval()
tokenizer = open_clip.get_tokenizer('ViT-g-14')

@app.post("/classify/")
async def classify_images(images: list[UploadFile], prompts: list[str]):
    if not images or not prompts:
        raise HTTPException(status_code=400, detail="Images and prompts cannot be empty.")

    try:
        # Process each image and prompt
        text_tokens = tokenizer(prompts)
        results = []
        for image in images:
            pil_image = Image.open(BytesIO(await image.read()))
            image_tensor = preprocess(pil_image).unsqueeze(0)
            
            with torch.no_grad(), torch.cuda.amp.autocast():
                image_features = model.encode_image(image_tensor)
                text_features = model.encode_text(text_tokens)
                image_features /= image_features.norm(dim=-1, keepdim=True)
                text_features /= text_features.norm(dim=-1, keepdim=True)

                text_probs = (100.0 * image_features @ text_features.T).softmax(dim=-1)
                results.append(text_probs.tolist())

        return JSONResponse(content={"results": results})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

步骤 3：运行服务

在终端中执行以下命令以运行你的 FastAPI 应用程序：

1	python clip_server. Py

这将启动位于 localhost 上的服务器，端口为 8000。你现在可以向 http://localhost:8000/classify/ 发送带有图像和提示词的 POST 请求。

步骤 4：测试服务

你可以使用 Postman 或编写一个简单的 python 脚本，利用 python 的 requests 库来测试 API：

import requests

url = 'http://localhost:8000/classify/'
files = {'images': open('path_to_image.jpg', 'rb')}
data = {'prompts': ['a picture of taylor swift', 'a picture of AOC']}

response = requests.post(url, files=files, data=data)
print(response.json())

此基本设置展示了如何使用 open_clip 和 FastAPI 创建服务。你可以通过添加更强大的错误处理、日志记录和其他功能来扩展此示例。

🧑‍🦰User

1
2
3

图像应该是图像列表。重写它

🤖 Assistant

当然，让我们调整 FastAPI 服务，使其能够在单个请求中处理多个图像。在更新的服务中，你可以在 POST 请求中发送一组图像文件。我将修改图像上传处理，以相应地处理每个图像文件。

这是修订后的 clip_server.py 版本：

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
import torch
from PIL import Image
import open_clip
from io import BytesIO
import numpy as np

app = FastAPI()

# Load the CLIP model
model, _, preprocess = open_clip.create_model_and_transforms('ViT-g-14', pretrained='laion2b_s12b_b42k')
model.eval()
tokenizer = open_clip.get_tokenizer('ViT-g-14')

@app.post("/classify/")
async def classify_images(images: list[UploadFile] = File(...), prompts: list[str]):
    if not images or not prompts:
        raise HTTPException(status_code=400, detail="Images and prompts cannot be empty.")

    try:
        # Process each image and prompt
        text_tokens = tokenizer(prompts)
        results = []
        for image_file in images:
            pil_image = Image.open(BytesIO(await image_file.read()))
            image_tensor = preprocess(pil_image).unsqueeze(0)
            
            with torch.no_grad(), torch.cuda.amp.autocast():
                image_features = model.encode_image(image_tensor)
                text_features = model.encode_text(text_tokens)
                image_features /= image_features.norm(dim=-1, keepdim=True)
                text_features /= text_features.norm(dim=-1, keepdim=True)

                text_probs = (100.0 * image_features @ text_features.T).softmax(dim=-1)
                results.append(text_probs.tolist())

        return JSONResponse(content={"results": results})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

主要调整：

classify_images 函数中的 images 参数现在显式地期望一个使用 File(...) 注解的文件列表。这告诉 FastAPI 在请求中期望多个文件。
其余部分处理每个上传的文件，应用模型的图像预处理，然后根据提供的提示词进行分类。

测试更新的服务：

你仍然可以使用之前相同的方式测试此更新的服务，但需使用多个图像：

import requests

url = 'http://localhost:8000/classify/'
files = [
    ('images', ('image1.jpg', open('path_to_image1.jpg', 'rb'), 'image/jpeg')),
    ('images', ('image2.jpg', open('path_to_image2.jpg', 'rb'), 'image/jpeg'))
]
data = {'prompts': ['a picture of taylor swift', 'a picture of AOC']}

response = requests.post(url, files=files, data=data)
print(response.json())

此脚本将两个图像与提示词一起发送到 FastAPI 服务，你应该会看到每个图像相对于每个提示词的分类概率。

1
2
3

不，批量处理传递给模型的图像。不要使用 unsqueeze，改为 tensor

🤖 Assistant

为了修改 FastAPI 服务，使其在传递给 open_clip 模型之前批量处理所有图像，我们需要调整图像的预处理方式并将它们传递给模型。我们将合并所有图像到一个单一的 tensor 批量中，然后将该批量传递给模型。这种方法更有效，尤其是在处理多个图像时。

以下是你可以调整 clip_server.py 文件以批量处理图像的方法：

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
import torch
from PIL import Image
import open_clip
from io import BytesIO
import numpy as np

app = FastAPI()

# Load the CLIP model
model, _, preprocess = open_clip.create_model_and_transforms('ViT-g-14', pretrained='laion2b_s12b_b42k')
model.eval()
tokenizer = open_clip.get_tokenizer('ViT-g-14')

@app.post("/classify/")
async def classify_images(images: list[UploadFile] = File(...), prompts: list[str]):
    if not images or not prompts:
        raise HTTPException(status_code=400, detail="Images and prompts cannot be empty.")

    try:
        # Tokenize the prompts once for all images
        text_tokens = tokenizer(prompts)

        # Load and preprocess all images, then stack into a single tensor
        image_tensors = []
        for image_file in images:
            pil_image = Image.open(BytesIO(await image_file.read()))
            processed_image = preprocess(pil_image)
            image_tensors.append(processed_image)
        image_batch = torch.stack(image_tensors)

        # Perform inference in a single pass
        with torch.no_grad(), torch.cuda.amp.autocast():
            image_features = model.encode_image(image_batch)
            text_features = model.encode_text(text_tokens)
            image_features /= image_features.norm(dim=-1, keepdim=True)
            text_features /= text_features.norm(dim=-1, keepdim=True)

            text_probs = (100.0 * image_features @ text_features.T).softmax(dim=-1)
            results = text_probs.tolist()

        return JSONResponse(content={"results": results})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

关键调整：

图像批量准备：所有图像将预处理并存储在一个列表中。然后使用 torch.stack 将这些预处理后的图像堆叠为单个 tensor，从而形成一个传递给模型的批次。
模型推理：模型一次性处理整个图像批次。这通过利用 GPU 的批量处理能力来提高性能（如果可用）。

测试更新的服务：

你可以使用之前的测试脚本，确保它需要时修改以发送多个图像：

import requests

url = 'http://localhost:8000/classify/'
files = [
    ('images', ('image1.jpg', open('path_to_image1.jpg', 'rb'), 'image/jpeg')),
    ('images', ('image2.jpg', open('path_to_image2.jpg', 'rb'), 'image/jpeg'))
]
data = {'prompts': ['a picture of taylor swift', 'a picture of AOC']}

response = requests.post(url, files=files, data=data)
print(response.json())

此服务现在将一次性批量处理所有上传的图像，使其在处理多个图像输入时更高效和可扩展。

🧑‍🦰User


让我们做得更简单。只需将图像设为 JSON 编码的文件列表。也以这种方式发送。

🤖 Assistant

为了简化服务并允许图像以 JSON 编码的字符串形式发送，你需要修改图像的接收和处理方式。在这种设置中，图像可以通过 JSON 作为 base 64 编码的字符串发送。这种方法适用于无法直接上传文件的情况，或需要将数据以 JSON 格式发送的系统。

步骤 1：修改 FastAPI 应用程序

以下是如何调整 FastAPI 应用程序，使其接受 base 64 编码的图像字符串和 JSON 负载中的提示词列表：

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
import torch
from PIL import Image
import open_clip
from io import BytesIO
import base64
import numpy as np

app = FastAPI()

# Load the CLIP model
model, _, preprocess = open_clip.create_model_and_transforms('ViT-g-14', pretrained='laion2b_s12b_b42k')
model.eval()
tokenizer = open_clip.get_tokenizer('ViT-g-14')

@app.post("/classify/")
async def classify_images(data: dict):
    images_encoded = data.get("images")
    prompts = data.get("prompts")

    if not images_encoded or not prompts:
        raise HTTPException(status_code=400, detail="Images and prompts cannot be empty.")

    try:
        # Decode images and prepare batch
        image_tensors = []
        for image_data in images_encoded:
            image_bytes = base64.b64decode(image_data)
            pil_image = Image.open(BytesIO(image_bytes))
            processed_image = preprocess(pil_image)
            image_tensors.append(processed_image)
        image_batch = torch.stack(image_tensors)

        # Tokenize the prompts
        text_tokens = tokenizer(prompts)

        # Perform inference
        with torch.no_grad(), torch.cuda.amp.autocast():
            image_features = model.encode_image(image_batch)
            text_features = model.encode_text(text_tokens)
            image_features /= image_features.norm(dim=-1, keepdim=True)
            text_features /= text_features.norm(dim=-1, keepdim=True)

            text_probs = (100.0 * image_features @ text_features.T).softmax(dim=-1)
            results = text_probs.tolist()

        return JSONResponse(content={"results": results})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

步骤 2：如何测试修改后的服务

要测试该服务，你可以编写一个脚本，将 base 64 编码的图像与提示词一起作为 JSON 发送。以下是如何使用 python 的 requests 库进行操作：

import requests
import base64

url = 'http://localhost:8000/classify/'

# Encode the image to base64
def encode_image_to_base64(filepath):
    with open(filepath, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode('utf-8')

# Prepare data
data = {
    "images": [encode_image_to_base64("path_to_image1.jpg"), encode_image_to_base64("path_to_image2.jpg")],
    "prompts": ["a picture of taylor swift", "a picture of AOC"]
}

response = requests.post(url, json=data)
print(response.json())

此测试脚本将图像编码为 base 64 字符串，并将其与提示词一起捆绑在 JSON 结构中。服务将解码图像，预处理它们，然后使用 open_clip 模型进行分类。此方法为在 JSON 结构中处理图像数据提供了一种灵活的方式，使其更容易与需要 JSON 格式的各种系统集成。

🧑‍🦰User

1
2
3


让发送代码处理已经加载的 PIL 图像，而不是磁盘上的

🤖 Assistant

为了调整 python 脚本，以便从已经加载的 PIL 图像（无需保存到磁盘）发送 base 64 编码的图像，你可以修改编码功能以处理内存中的图像。这种方法允许你直接将 PIL 图像对象转换为 base 64 编码的字符串。

以下是你如何调整脚本的方法：

用于发送内存中的 PIL 图像的修改脚本

import requests
import base64
from PIL import Image
from io import BytesIO

url = 'http://localhost:8000/classify/'

# Function to encode a PIL image to base64
def encode_pil_image_to_base64(pil_img):
    buffered = BytesIO()
    pil_img.save(buffered, format="JPEG")  # You can change the format to PNG or other supported types depending on the image
    return base64.b64encode(buffered.getvalue()).decode('utf-8')

# Load or create a PIL image (example)
image1 = Image.open("path_to_image1.jpg")  # Assume this is how you might have loaded the image
image2 = Image.open("path_to_image2.jpg")  # Another example image

# Prepare data with the images already loaded into PIL
data = {
    "images": [encode_pil_image_to_base64(image1), encode_pil_image_to_base64(image2)],
    "prompts": ["a picture of taylor swift", "a picture of AOC"]
}

response = requests.post(url, json=data)
print(response.json())

脚本中的关键元素

encode_pil_image_to_base64 函数：此函数接受一个 PIL Image 对象，将其保存到 BytesIO 缓冲区（模拟文件），然后将该缓冲区编码为 base 64 字符串。示例中的图像格式设置为 JPEG，但你可以根据图像内容和所需质量进行调整。
内存中的图像加载：示例显示了从磁盘加载图像的方式，但你可以将其替换为任何获取 PIL Image 对象的方法，如从网络、不同格式或生成图像。
数据准备和请求：图像被编码为 base 64 字符串，并与提示词一起打包在 JSON 结构中。然后，这个 JSON 作为 POST 请求发送到服务器。

这种方法为图像在程序内被操作或生成的应用程序提供了灵活性，消除了在传输前需要将其暂时存储到磁盘的需求。这在动态图像处理应用程序或处理不应写入磁盘的敏感数据时特别有用。