自然语言处理（十一）：多模态大模型

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from transformers import CLIPProcessor, CLIPModel
from PIL import Image
import torch

model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32").eval()
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

image = Image.open("cat.jpg")
labels = ["a photo of a cat", "a photo of a dog",
          "a photo of a car", "a photo of a bird"]

with torch.no_grad():
    inputs = processor(text=labels, images=image,
                       return_tensors="pt", padding=True)
    out = model(**inputs)

# logits_per_image 已经按 1/τ 缩放
probs = out.logits_per_image.softmax(dim=-1)[0]
for label, p in zip(labels, probs):
    print(f"{p.item():>6.2%}  {label}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from transformers import Blip2Processor, Blip2ForConditionalGeneration
from PIL import Image
import torch

processor = Blip2Processor.from_pretrained("Salesforce/blip2-opt-2.7b")
model = Blip2ForConditionalGeneration.from_pretrained(
    "Salesforce/blip2-opt-2.7b",
    torch_dtype=torch.float16, device_map="auto",
)

image = Image.open("street.jpg")
prompt = "Question: What is happening in this photo? Answer:"
inputs = processor(image, text=prompt, return_tensors="pt").to("cuda", torch.float16)

with torch.no_grad():
    ids = model.generate(**inputs, max_new_tokens=60)
print(processor.batch_decode(ids, skip_special_tokens=True)[0].strip())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# LLaVA-1.5 推理（Transformers 版）
from transformers import LlavaForConditionalGeneration, AutoProcessor
from PIL import Image
import torch

model_id = "llava-hf/llava-1.5-7b-hf"
processor = AutoProcessor.from_pretrained(model_id)
model = LlavaForConditionalGeneration.from_pretrained(
    model_id, torch_dtype=torch.float16, device_map="auto",
)

image = Image.open("ui_screenshot.png")
prompt = ("USER: <image>\n这个 UI 是干什么的？是否存在 bug？请具体说明。"
          "ASSISTANT:")

inputs = processor(text=prompt, images=image,
                   return_tensors="pt").to("cuda", torch.float16)
with torch.no_grad():
    out = model.generate(**inputs, max_new_tokens=200, do_sample=False)
print(processor.batch_decode(out, skip_special_tokens=True)[0])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from transformers import BlipProcessor, BlipForConditionalGeneration
from PIL import Image
import torch

class ImageCaptioner:
    def __init__(self, model_id="Salesforce/blip-image-captioning-large"):
        self.proc = BlipProcessor.from_pretrained(model_id)
        self.model = BlipForConditionalGeneration.from_pretrained(model_id).eval()

    @torch.no_grad()
    def caption(self, image_path, prefix=None,
                max_length=50, num_beams=3):
        img = Image.open(image_path).convert("RGB")
        # `prefix` 用于条件生成，例如 "A photo of"
        inputs = self.proc(img, text=prefix, return_tensors="pt") \
            if prefix else self.proc(img, return_tensors="pt")
        out = self.model.generate(
            **inputs, max_length=max_length,
            num_beams=num_beams, length_penalty=1.0, early_stopping=True,
        )
        return self.proc.decode(out[0], skip_special_tokens=True)

cap = ImageCaptioner()
print(cap.caption("dog.jpg"))                        # 无条件生成
print(cap.caption("dog.jpg", prefix="A photo of"))   # 条件生成

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from transformers import BlipProcessor, BlipForQuestionAnswering
from PIL import Image
import torch

class VQA:
    def __init__(self, model_id="Salesforce/blip-vqa-base"):
        self.proc = BlipProcessor.from_pretrained(model_id)
        self.model = BlipForQuestionAnswering.from_pretrained(model_id).eval()

    @torch.no_grad()
    def answer(self, image_path, question, max_length=30):
        img = Image.open(image_path).convert("RGB")
        inputs = self.proc(img, question, return_tensors="pt")
        out = self.model.generate(**inputs, max_length=max_length)
        return self.proc.decode(out[0], skip_special_tokens=True)

vqa = VQA()
print(vqa.answer("market.jpg", "How many apples are on the table?"))
print(vqa.answer("market.jpg", "What color is the basket?"))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

import numpy as np
import torch
from PIL import Image
from transformers import CLIPProcessor, CLIPModel

class CLIPRetrieval:
    """基于 CLIP 的轻量级图像检索工具。超过百万条目时建议改用 FAISS-IVF。"""

    def __init__(self, model_id="openai/clip-vit-base-patch32",
                 device=None):
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        self.model = CLIPModel.from_pretrained(model_id).to(self.device).eval()
        self.proc = CLIPProcessor.from_pretrained(model_id)
        self.embeds = []   # 存储形状为 (512,) 的 np.ndarray 列表
        self.paths = []

    @torch.no_grad()
    def add_images(self, paths, batch_size=32):
        for i in range(0, len(paths), batch_size):
            batch_paths = paths[i:i + batch_size]
            imgs = [Image.open(p).convert("RGB") for p in batch_paths]
            inputs = self.proc(images=imgs, return_tensors="pt").to(self.device)
            emb = self.model.get_image_features(**inputs)
            emb = emb / emb.norm(dim=-1, keepdim=True)
            self.embeds.extend(emb.cpu().numpy())
            self.paths.extend(batch_paths)

    @torch.no_grad()
    def search_text(self, query, top_k=5):
        inputs = self.proc(text=[query], return_tensors="pt",
                           padding=True).to(self.device)
        q = self.model.get_text_features(**inputs)
        q = q / q.norm(dim=-1, keepdim=True)
        sims = np.vstack(self.embeds) @ q.cpu().numpy().T
        sims = sims.flatten()
        idx = np.argpartition(-sims, top_k)[:top_k]
        idx = idx[np.argsort(-sims[idx])]
        return [(self.paths[i], float(sims[i])) for i in idx]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import whisper

class WhisperASR:
    def __init__(self, size="base", device="cuda"):
        self.model = whisper.load_model(size, device=device)

    def transcribe(self, audio_path, language=None, task="transcribe"):
        # task 可选值为 {"transcribe", "translate"}
        return self.model.transcribe(
            audio_path,
            language=language,           # None 表示自动检测语种
            task=task,
            word_timestamps=True,        # 输出每个单词的时间戳
            condition_on_previous_text=False,  # 长音频更稳定
            no_speech_threshold=0.6,
            verbose=False,
        )

asr = WhisperASR(size="small")
result = asr.transcribe("meeting.mp3")
print(f"识别语种: {result['language']}")
for seg in result["segments"]:
    print(f"[{seg['start']:6.2f} - {seg['end']:6.2f}] {seg['text'].strip()}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from openai import OpenAI
import base64, mimetypes

client = OpenAI()

def analyse_image(path: str, prompt: str, model: str = "gpt-4o") -> str:
    mime, _ = mimetypes.guess_type(path)
    with open(path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()
    resp = client.chat.completions.create(
        model=model,
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url",
                 "image_url": {"url": f"data:{mime};base64,{b64}",
                               "detail": "high"}},
            ],
        }],
        max_tokens=500,
        temperature=0.2,
    )
    return resp.choices[0].message.content

print(analyse_image("dashboard.png",
                    "总结关键指标，并指出是否存在异常。"))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

import cv2
import numpy as np

def sample_frames(video_path, num_frames=8):
    """均匀采样视频帧，返回 PIL.Image 对象列表。"""
    from PIL import Image
    cap = cv2.VideoCapture(video_path)
    total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    if total <= 0:
        cap.release()
        raise RuntimeError(f"无法读取视频文件 {video_path}")
    indices = np.linspace(0, total - 1, num_frames).astype(int)
    frames = []
    for idx in indices:
        cap.set(cv2.CAP_PROP_POS_FRAMES, int(idx))
        ok, frame = cap.read()
        if not ok:
            continue
        frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        frames.append(Image.fromarray(frame))
    cap.release()
    return frames