多模态 AI 完全指南：掌握 CLIP、LLaVA、GPT-4V 与 Gemini Vision

2021: CLIP (OpenAI) - 通过对比学习连接图像与文本
2022: BLIP - 统一图像描述生成与 VQA
2023: BLIP-2 - 通过 Q-Former 实现高效的多模态学习
2023: LLaVA - 开源视觉-语言助手
2023: GPT-4V - 商用多模态 LLM
2023: Gemini - 谷歌的多模态基础模型
2024: Claude 3 Vision - Anthropic 的多模态模型
2024: LLaVA-1.6、InternVL2、Qwen-VL2 - 开源模型的改进
2025: 扩展至视频理解与 3D 理解

图像 → [图像编码器 (ViT/ResNet)] → 图像嵌入 (512 维)
                                                    ↕ 相似度度量
文本 → [文本编码器 (Transformer)]→ 文本嵌入 (512 维)

import torch
import torch.nn.functional as F
from PIL import Image
import requests
from transformers import CLIPProcessor, CLIPModel

# 加载 CLIP 模型
model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")

def clip_zero_shot_classification(
    image: Image.Image,
    candidate_labels: list[str]
) -> dict[str, float]:
    """使用 CLIP 进行零样本图像分类。"""

    # 生成文本提示词（CLIP 推荐的格式）
    text_prompts = [f"a photo of a {label}" for label in candidate_labels]

    # 预处理
    inputs = processor(
        text=text_prompts,
        images=image,
        return_tensors="pt",
        padding=True
    )

    # 推理
    with torch.no_grad():
        outputs = model(**inputs)
        logits_per_image = outputs.logits_per_image
        probs = logits_per_image.softmax(dim=1)

    # 返回结果
    return {
        label: prob.item()
        for label, prob in zip(candidate_labels, probs[0])
    }

# 使用示例
image_url = "https://example.com/sample_image.jpg"
image = Image.open(requests.get(image_url, stream=True).raw)

labels = ["cat", "dog", "bird", "fish", "rabbit"]
results = clip_zero_shot_classification(image, labels)

# 按概率排序
sorted_results = sorted(results.items(), key=lambda x: x[1], reverse=True)
for label, prob in sorted_results:
    print(f"{label}: {prob:.4f} ({prob*100:.1f}%)")

import numpy as np
from typing import Union

class CLIPSearchEngine:
    """基于 CLIP 的图像-文本检索引擎。"""

    def __init__(self, model_name: str = "openai/clip-vit-large-patch14"):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.image_embeddings = []
        self.text_embeddings = []
        self.image_metadata = []
        self.text_metadata = []

    def encode_images(self, images: list[Image.Image]) -> torch.Tensor:
        """将一批图像转换为嵌入向量。"""
        inputs = self.processor(
            images=images,
            return_tensors="pt",
            padding=True
        )
        with torch.no_grad():
            image_features = self.model.get_image_features(**inputs)
            # L2 归一化
            image_features = F.normalize(image_features, p=2, dim=-1)
        return image_features

    def encode_texts(self, texts: list[str]) -> torch.Tensor:
        """将一批文本转换为嵌入向量。"""
        inputs = self.processor(
            text=texts,
            return_tensors="pt",
            padding=True,
            truncation=True
        )
        with torch.no_grad():
            text_features = self.model.get_text_features(**inputs)
            text_features = F.normalize(text_features, p=2, dim=-1)
        return text_features

    def index_images(
        self,
        images: list[Image.Image],
        metadata: list[dict] = None
    ):
        """对图像建立索引。"""
        embeddings = self.encode_images(images)
        self.image_embeddings.append(embeddings)
        if metadata:
            self.image_metadata.extend(metadata)

    def text_to_image_search(
        self,
        query: str,
        top_k: int = 5
    ) -> list[dict]:
        """用文本查询检索图像。"""
        if not self.image_embeddings:
            return []

        # 合并所有图像嵌入
        all_embeddings = torch.cat(self.image_embeddings, dim=0)

        # 编码查询
        query_embedding = self.encode_texts([query])

        # 计算余弦相似度（已完成 L2 归一化）
        similarities = (all_embeddings @ query_embedding.T).squeeze(-1)

        # 选取 Top-K 结果
        top_indices = similarities.argsort(descending=True)[:top_k]

        results = []
        for idx in top_indices:
            idx = idx.item()
            result = {
                "index": idx,
                "similarity": similarities[idx].item()
            }
            if self.image_metadata:
                result.update(self.image_metadata[idx])
            results.append(result)

        return results

# OpenCLIP：支持多种架构与训练数据集
# pip install open_clip_torch

import open_clip
import torch
from PIL import Image

# 查看可用模型列表
available_models = open_clip.list_pretrained()
print("可用模型:", available_models[:5])

# 加载在 LAION-2B 上训练的大型模型
model, preprocess_train, preprocess_val = open_clip.create_model_and_transforms(
    'ViT-H-14',
    pretrained='laion2b_s32b_b79k'
)
tokenizer = open_clip.get_tokenizer('ViT-H-14')

def compute_clip_similarity(
    image: Image.Image,
    texts: list[str],
    model=model,
    preprocess=preprocess_val,
    tokenizer=tokenizer
) -> list[float]:
    """计算图像与文本列表之间的 CLIP 相似度。"""
    model.eval()

    # 图像预处理
    image_input = preprocess(image).unsqueeze(0)

    # 文本分词
    text_input = tokenizer(texts)

    with torch.no_grad(), torch.cuda.amp.autocast():
        image_features = model.encode_image(image_input)
        text_features = model.encode_text(text_input)

        # 归一化
        image_features /= image_features.norm(dim=-1, keepdim=True)
        text_features /= text_features.norm(dim=-1, keepdim=True)

        # 计算相似度
        similarity = (100.0 * image_features @ text_features.T).softmax(dim=-1)

    return similarity[0].tolist()

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
from transformers import CLIPModel, CLIPProcessor
import torch.optim as optim

class ImageTextDataset(Dataset):
    """图像-文本对数据集。"""

    def __init__(
        self,
        image_paths: list[str],
        texts: list[str],
        processor: CLIPProcessor
    ):
        self.image_paths = image_paths
        self.texts = texts
        self.processor = processor

    def __len__(self):
        return len(self.image_paths)

    def __getitem__(self, idx):
        image = Image.open(self.image_paths[idx]).convert("RGB")
        text = self.texts[idx]

        inputs = self.processor(
            images=image,
            text=text,
            return_tensors="pt",
            padding="max_length",
            max_length=77,
            truncation=True
        )

        return {
            "pixel_values": inputs["pixel_values"].squeeze(0),
            "input_ids": inputs["input_ids"].squeeze(0),
            "attention_mask": inputs["attention_mask"].squeeze(0)
        }

class CLIPFineTuner:
    """CLIP 模型微调类。"""

    def __init__(self, model_name: str = "openai/clip-vit-base-patch32"):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model.to(self.device)

    def contrastive_loss(
        self,
        image_features: torch.Tensor,
        text_features: torch.Tensor,
        temperature: float = 0.07
    ) -> torch.Tensor:
        """对比学习损失函数。"""
        # 归一化
        image_features = F.normalize(image_features, dim=-1)
        text_features = F.normalize(text_features, dim=-1)

        # 相似度矩阵
        logits = torch.matmul(image_features, text_features.T) / temperature

        # 对角线为正确答案（第 i 个图像与第 i 个文本配对）
        labels = torch.arange(len(logits)).to(self.device)

        # 双向交叉熵
        loss_i = F.cross_entropy(logits, labels)
        loss_t = F.cross_entropy(logits.T, labels)

        return (loss_i + loss_t) / 2

    def train(
        self,
        train_dataset: ImageTextDataset,
        num_epochs: int = 10,
        batch_size: int = 32,
        learning_rate: float = 1e-5
    ):
        """CLIP 微调训练。"""
        dataloader = DataLoader(
            train_dataset,
            batch_size=batch_size,
            shuffle=True,
            num_workers=4
        )

        optimizer = optim.AdamW(
            self.model.parameters(),
            lr=learning_rate,
            weight_decay=0.01
        )

        scheduler = optim.lr_scheduler.CosineAnnealingLR(
            optimizer,
            T_max=num_epochs
        )

        for epoch in range(num_epochs):
            total_loss = 0
            self.model.train()

            for batch in dataloader:
                pixel_values = batch["pixel_values"].to(self.device)
                input_ids = batch["input_ids"].to(self.device)
                attention_mask = batch["attention_mask"].to(self.device)

                # 前向传播
                outputs = self.model(
                    pixel_values=pixel_values,
                    input_ids=input_ids,
                    attention_mask=attention_mask
                )

                image_features = outputs.image_embeds
                text_features = outputs.text_embeds

                # 计算损失
                loss = self.contrastive_loss(image_features, text_features)

                # 反向传播
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()

                total_loss += loss.item()

            scheduler.step()
            avg_loss = total_loss / len(dataloader)
            print(f"Epoch {epoch+1}/{num_epochs}, Loss: {avg_loss:.4f}")

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

# BLIP 图像描述生成
class BLIPCaptioner:
    def __init__(self):
        self.processor = BlipProcessor.from_pretrained(
            "Salesforce/blip-image-captioning-large"
        )
        self.model = BlipForConditionalGeneration.from_pretrained(
            "Salesforce/blip-image-captioning-large",
            torch_dtype=torch.float16
        )
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model.to(self.device)

    def caption(
        self,
        image: Image.Image,
        conditional_text: str = None,
        max_new_tokens: int = 50
    ) -> str:
        """为图像生成描述文字。"""
        if conditional_text:
            # 条件式描述生成
            inputs = self.processor(
                image,
                conditional_text,
                return_tensors="pt"
            ).to(self.device, torch.float16)
        else:
            # 无条件描述生成
            inputs = self.processor(
                image,
                return_tensors="pt"
            ).to(self.device, torch.float16)

        with torch.no_grad():
            output = self.model.generate(
                **inputs,
                max_new_tokens=max_new_tokens,
                num_beams=4,
                early_stopping=True
            )

        return self.processor.decode(output[0], skip_special_tokens=True)

# BLIP VQA（视觉问答）
class BLIPVisualQA:
    def __init__(self):
        self.processor = BlipProcessor.from_pretrained(
            "Salesforce/blip-vqa-base"
        )
        self.model = BlipForQuestionAnswering.from_pretrained(
            "Salesforce/blip-vqa-base"
        )

    def answer(self, image: Image.Image, question: str) -> str:
        """回答关于图像的问题。"""
        inputs = self.processor(image, question, return_tensors="pt")

        with torch.no_grad():
            output = self.model.generate(**inputs, max_new_tokens=50)

        return self.processor.decode(output[0], skip_special_tokens=True)

# 使用示例
captioner = BLIPCaptioner()
vqa = BLIPVisualQA()

image = Image.open("sample.jpg")

# 生成描述
caption = captioner.caption(image)
print(f"描述: {caption}")

# 条件式描述
cond_caption = captioner.caption(image, "a photo of")
print(f"条件式描述: {cond_caption}")

# VQA
answer = vqa.answer(image, "What color is the sky?")
print(f"答案: {answer}")

from transformers import Blip2Processor, Blip2ForConditionalGeneration
import torch

class BLIP2Assistant:
    """基于 BLIP-2 的视觉问答助手。"""

    def __init__(
        self,
        model_name: str = "Salesforce/blip2-opt-2.7b"
    ):
        self.processor = Blip2Processor.from_pretrained(model_name)
        self.model = Blip2ForConditionalGeneration.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            device_map="auto"
        )

    def generate_response(
        self,
        image: Image.Image,
        prompt: str = None,
        max_new_tokens: int = 200,
        temperature: float = 1.0
    ) -> str:
        """为图像和（可选的）提示词生成响应。"""

        if prompt:
            inputs = self.processor(
                images=image,
                text=prompt,
                return_tensors="pt"
            ).to("cuda", torch.float16)
        else:
            inputs = self.processor(
                images=image,
                return_tensors="pt"
            ).to("cuda", torch.float16)

        with torch.no_grad():
            generated_ids = self.model.generate(
                **inputs,
                max_new_tokens=max_new_tokens,
                temperature=temperature,
                do_sample=temperature > 0
            )

        generated_text = self.processor.batch_decode(
            generated_ids,
            skip_special_tokens=True
        )[0].strip()

        return generated_text

    def batch_caption(
        self,
        images: list[Image.Image],
        batch_size: int = 8
    ) -> list[str]:
        """为一批图像批量生成描述。"""
        all_captions = []

        for i in range(0, len(images), batch_size):
            batch = images[i:i + batch_size]

            inputs = self.processor(
                images=batch,
                return_tensors="pt",
                padding=True
            ).to("cuda", torch.float16)

            with torch.no_grad():
                generated_ids = self.model.generate(
                    **inputs,
                    max_new_tokens=50
                )

            captions = self.processor.batch_decode(
                generated_ids,
                skip_special_tokens=True
            )

            all_captions.extend([c.strip() for c in captions])

        return all_captions

# 使用示例
assistant = BLIP2Assistant("Salesforce/blip2-flan-t5-xxl")
image = Image.open("document.png")

# 自由格式提问
response = assistant.generate_response(
    image,
    "Question: What is the main topic of this document? Answer:"
)
print(response)

# 对话式会话
conversation_history = []
questions = [
    "图像中能看到什么？",
    "颜色是什么样的？",
    "背景是怎样的？"
]

for q in questions:
    history_text = "\n".join(conversation_history)
    prompt = f"{history_text}\nQuestion: {q} Answer:"
    answer = assistant.generate_response(image, prompt)
    print(f"Q: {q}")
    print(f"A: {answer}")
    conversation_history.append(f"Q: {q} A: {answer}")

图像 → [CLIP ViT-L/14] → 图像特征 (1024 维)
                              ↓
                      [线性投影层]
                              ↓
                      [视觉 token]
                              ↓
           [LLM (LLaMA/Vicuna)] ← [文本 token]
                              ↓
                          最终响应

from transformers import LlavaNextProcessor, LlavaNextForConditionalGeneration
import torch
from PIL import Image

class LLaVAAssistant:
    """基于 LLaVA-1.6 的视觉助手。"""

    def __init__(
        self,
        model_name: str = "llava-hf/llava-v1.6-mistral-7b-hf"
    ):
        self.processor = LlavaNextProcessor.from_pretrained(model_name)
        self.model = LlavaNextForConditionalGeneration.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            low_cpu_mem_usage=True,
            device_map="auto"
        )

    def chat(
        self,
        image: Image.Image,
        message: str,
        max_new_tokens: int = 500,
        temperature: float = 0.7
    ) -> str:
        """结合图像进行对话。"""

        # LLaVA-1.6 的对话格式
        conversation = [
            {
                "role": "user",
                "content": [
                    {"type": "image"},
                    {"type": "text", "text": message}
                ]
            }
        ]

        prompt = self.processor.apply_chat_template(
            conversation,
            add_generation_prompt=True
        )

        inputs = self.processor(
            prompt,
            image,
            return_tensors="pt"
        ).to("cuda")

        with torch.no_grad():
            output = self.model.generate(
                **inputs,
                max_new_tokens=max_new_tokens,
                temperature=temperature,
                do_sample=temperature > 0,
                pad_token_id=self.processor.tokenizer.eos_token_id
            )

        # 排除输入部分，只提取生成的文本
        generated = output[0][inputs["input_ids"].shape[1]:]
        return self.processor.decode(generated, skip_special_tokens=True)

    def analyze_chart(self, chart_image: Image.Image) -> dict:
        """分析图表图像。"""
        analysis_prompts = [
            "这张图表的标题是什么？",
            "x 轴和 y 轴分别代表什么？",
            "最高值和最低值分别是多少？",
            "请描述整体趋势。",
            "这份数据中最重要的洞察是什么？"
        ]

        results = {}
        for prompt in analysis_prompts:
            response = self.chat(chart_image, prompt)
            results[prompt] = response

        return results

    def extract_text_from_image(self, image: Image.Image) -> str:
        """从图像中提取文字（OCR）。"""
        return self.chat(
            image,
            "请准确提取这张图像中的所有文字。"
            "只返回文字内容，不要附加其他说明。"
        )


# 实战应用：文档分析流水线
class DocumentAnalysisPipeline:
    """使用 LLaVA 的文档分析流水线。"""

    def __init__(self):
        self.llava = LLaVAAssistant()

    def analyze_document(self, document_image: Image.Image) -> dict:
        """对文档图像进行综合分析。"""

        # 1. 识别文档类型
        doc_type = self.llava.chat(
            document_image,
            "这份文档属于什么类型？（发票、合同、报告、表单等）"
        )

        # 2. 提取文字
        extracted_text = self.llava.extract_text_from_image(document_image)

        # 3. 提取关键信息
        key_info = self.llava.chat(
            document_image,
            f"请以 JSON 格式从这份 {doc_type} 中提取以下信息："
            "日期、发件人、收件人、金额（如有）、主要内容摘要"
        )

        # 4. 识别待办事项
        action_items = self.llava.chat(
            document_image,
            "如果这份文档中有需要处理的事项，请以列表形式列出。"
        )

        return {
            "document_type": doc_type,
            "extracted_text": extracted_text,
            "key_information": key_info,
            "action_items": action_items
        }

from transformers import InstructBlipProcessor, InstructBlipForConditionalGeneration
import torch
from PIL import Image

class InstructBLIPAssistant:
    """基于 InstructBLIP 的指令跟随助手。"""

    def __init__(self, model_name: str = "Salesforce/instructblip-vicuna-7b"):
        self.processor = InstructBlipProcessor.from_pretrained(model_name)
        self.model = InstructBlipForConditionalGeneration.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            device_map="auto"
        )

    def instruct(
        self,
        image: Image.Image,
        instruction: str,
        max_new_tokens: int = 300
    ) -> str:
        """针对图像执行具体的指令。"""
        inputs = self.processor(
            images=image,
            text=instruction,
            return_tensors="pt"
        ).to("cuda", torch.float16)

        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                do_sample=False,
                num_beams=5,
                max_new_tokens=max_new_tokens,
                min_length=1,
                top_p=0.9,
                repetition_penalty=1.5,
                length_penalty=1.0,
                temperature=1.0
            )

        generated_text = self.processor.batch_decode(
            outputs,
            skip_special_tokens=True
        )[0].strip()

        return generated_text

# 多种使用示例
assistant = InstructBLIPAssistant()
image = Image.open("complex_diagram.png")

# 复杂图表的说明
description = assistant.instruct(
    image,
    "请详细说明这张图表，包括每个组成部分的作用及其相互关系。"
)

# 特定对象检测
objects = assistant.instruct(
    image,
    "请以列表形式列出图像中发现的所有对象，并说明每个对象的位置。"
)

# 情感分析
emotion = assistant.instruct(
    image,
    "请分析图像中人物的情绪状态，并说明依据。"
)

# 对比分析
if len([image]) > 1:  # 存在多张图像的情况
    comparison = assistant.instruct(
        image,
        "请详细描述图像的特征，并说明与类似图像相比的差异之处。"
    )

import openai
import base64
from pathlib import Path
import httpx

client = openai.OpenAI()

def encode_image_to_base64(image_path: str) -> str:
    """将图像文件编码为 Base64。"""
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode('utf-8')

def image_url_to_base64(url: str) -> str:
    """从 URL 下载图像并编码为 Base64。"""
    response = httpx.get(url)
    return base64.b64encode(response.content).decode('utf-8')

class GPT4VisionAnalyzer:
    """基于 GPT-4 Vision 的图像分析器。"""

    def __init__(self, model: str = "gpt-4o"):
        self.client = openai.OpenAI()
        self.model = model

    def analyze_image(
        self,
        image_source: str,  # 文件路径或 URL
        prompt: str,
        is_url: bool = True,
        detail: str = "high",  # "low", "high", "auto"
        max_tokens: int = 1000
    ) -> str:
        """分析单张图像。"""

        if is_url:
            image_content = {
                "type": "image_url",
                "image_url": {
                    "url": image_source,
                    "detail": detail
                }
            }
        else:
            # 本地文件
            base64_image = encode_image_to_base64(image_source)
            ext = Path(image_source).suffix.lower()
            media_type_map = {
                ".jpg": "image/jpeg",
                ".jpeg": "image/jpeg",
                ".png": "image/png",
                ".gif": "image/gif",
                ".webp": "image/webp"
            }
            media_type = media_type_map.get(ext, "image/jpeg")

            image_content = {
                "type": "image_url",
                "image_url": {
                    "url": f"data:{media_type};base64,{base64_image}",
                    "detail": detail
                }
            }

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": [
                        image_content,
                        {"type": "text", "text": prompt}
                    ]
                }
            ],
            max_tokens=max_tokens
        )

        return response.choices[0].message.content

    def analyze_multiple_images(
        self,
        image_sources: list[dict],  # [{"source": "...", "is_url": True}]
        prompt: str,
        max_tokens: int = 2000
    ) -> str:
        """同时分析多张图像。"""
        content = []

        for img_info in image_sources:
            source = img_info["source"]
            is_url = img_info.get("is_url", True)

            if is_url:
                content.append({
                    "type": "image_url",
                    "image_url": {"url": source, "detail": "high"}
                })
            else:
                base64_image = encode_image_to_base64(source)
                content.append({
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/jpeg;base64,{base64_image}"
                    }
                })

        content.append({"type": "text", "text": prompt})

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": content}],
            max_tokens=max_tokens
        )

        return response.choices[0].message.content

    def analyze_chart_or_graph(self, image_source: str) -> dict:
        """将图表或图形分析为结构化格式。"""
        prompt = """请分析这张图表/图形，并以以下 JSON 格式返回：
{
  "chart_type": "柱状图/折线图/饼图/散点图等",
  "title": "图表标题",
  "x_axis": {"label": "X 轴标签", "unit": "单位"},
  "y_axis": {"label": "Y 轴标签", "unit": "单位"},
  "data_series": [{"name": "系列名称", "trend": "上升/下降/持平"}],
  "key_findings": ["发现1", "发现2"],
  "data_range": {"min": 0, "max": 0},
  "anomalies": ["异常值说明"]
}"""

        response = self.analyze_image(
            image_source,
            prompt,
            detail="high",
            max_tokens=1500
        )

        import json
        try:
            # 尝试解析 JSON
            start = response.find('{')
            end = response.rfind('}') + 1
            if start >= 0 and end > start:
                return json.loads(response[start:end])
        except json.JSONDecodeError:
            pass

        return {"raw_response": response}

    def extract_structured_data_from_document(
        self,
        document_image_path: str
    ) -> dict:
        """从文档图像中提取结构化数据。"""
        prompt = """请以 JSON 格式从这份文档中提取以下信息：
1. 文档类型
2. 日期（如有）
3. 发件人/作者
4. 收件人（如有）
5. 主要内容摘要（3-5 句）
6. 主要数值数据（表格、金额等）
7. 是否有签名/批准

JSON 格式：
{
  "document_type": "",
  "date": "",
  "author": "",
  "recipient": "",
  "summary": "",
  "numerical_data": [],
  "signature_present": false
}"""

        return self.analyze_chart_or_graph.__func__(self, document_image_path)


# 实战应用：电商商品分析
def analyze_product_images(image_urls: list[str]) -> dict:
    """分析多张商品图像。"""
    analyzer = GPT4VisionAnalyzer()

    image_sources = [{"source": url, "is_url": True} for url in image_urls]

    result = analyzer.analyze_multiple_images(
        image_sources,
        prompt="""请分析这些商品图像，并以以下 JSON 格式返回：
{
  "product_name": "推测的商品名称",
  "category": "商品类别",
  "color_options": ["颜色列表"],
  "key_features": ["主要特征"],
  "condition": "全新商品/二手等",
  "quality_score": 0-10,
  "marketing_description": "营销文案（100 字）",
  "seo_keywords": ["SEO 关键词"]
}"""
    )

    return result

import google.generativeai as genai
import PIL.Image
from pathlib import Path
import base64

# 设置 API 密钥
genai.configure(api_key="YOUR_GEMINI_API_KEY")

class GeminiVisionAnalyzer:
    """基于 Gemini Vision 的分析器。"""

    def __init__(self, model_name: str = "gemini-1.5-pro"):
        self.model = genai.GenerativeModel(model_name)
        self.vision_model = genai.GenerativeModel("gemini-1.5-flash")

    def analyze_image(
        self,
        image_path: str,
        prompt: str
    ) -> str:
        """分析图像。"""
        image = PIL.Image.open(image_path)
        response = self.model.generate_content([prompt, image])
        return response.text

    def analyze_with_url(self, image_url: str, prompt: str) -> str:
        """分析 URL 中的图像。"""
        import httpx
        image_data = httpx.get(image_url).content

        image_part = {
            "mime_type": "image/jpeg",
            "data": base64.b64encode(image_data).decode('utf-8')
        }

        response = self.model.generate_content([
            {"text": prompt},
            image_part
        ])
        return response.text

    def analyze_video(
        self,
        video_path: str,
        questions: list[str]
    ) -> dict:
        """分析视频。（Gemini 1.5 Pro 的强项）"""

        # 上传视频文件
        print(f"正在上传视频: {video_path}")
        video_file = genai.upload_file(
            path=video_path,
            display_name="analysis_video"
        )

        # 等待上传完成
        import time
        while video_file.state.name == "PROCESSING":
            print("处理中...")
            time.sleep(10)
            video_file = genai.get_file(video_file.name)

        if video_file.state.name == "FAILED":
            raise ValueError("视频处理失败")

        print(f"视频上传完成: {video_file.uri}")

        # 逐个问题分析
        results = {}
        for question in questions:
            response = self.model.generate_content(
                [video_file, question],
                request_options={"timeout": 600}
            )
            results[question] = response.text

        # 删除文件（可选）
        genai.delete_file(video_file.name)

        return results

    def analyze_multiple_images_interleaved(
        self,
        image_text_pairs: list[dict]  # [{"image": PIL.Image, "text": str}]
    ) -> str:
        """处理图像与文本交叉排列的复合查询。"""
        content = []

        for pair in image_text_pairs:
            if "text" in pair:
                content.append(pair["text"])
            if "image" in pair:
                content.append(pair["image"])

        response = self.model.generate_content(content)
        return response.text

    def process_document_batch(
        self,
        document_images: list[PIL.Image.Image],
        extraction_schema: str
    ) -> list[dict]:
        """批量处理多份文档（利用 Gemini 的长上下文）。"""
        import json

        # 将所有图像放入一个请求中处理
        content = [f"请分析以下 {len(document_images)} 份文档：\n"]

        for i, img in enumerate(document_images, 1):
            content.append(f"\n--- 文档 {i} ---")
            content.append(img)

        content.append(f"\n请针对每份文档，按以下 JSON schema 提取数据：\n{extraction_schema}")

        response = self.model.generate_content(content)

        # 解析 JSON
        try:
            text = response.text
            # 提取 JSON 数组
            start = text.find('[')
            end = text.rfind(']') + 1
            if start >= 0 and end > start:
                return json.loads(text[start:end])
        except json.JSONDecodeError:
            return [{"raw_response": response.text}]


# 使用示例：视频分析
analyzer = GeminiVisionAnalyzer()

video_questions = [
    "请总结这段视频的整体内容。",
    "请列出主要场景及其对应的时间戳。",
    "视频中提到的主要关键词或概念是什么？",
    "这段视频的主题和目的是什么？"
]

results = analyzer.analyze_video("lecture_video.mp4", video_questions)
for question, answer in results.items():
    print(f"\n问题: {question}")
    print(f"答案: {answer}")

import anthropic
import base64
import httpx
from pathlib import Path

client = anthropic.Anthropic()

class ClaudeVisionAnalyzer:
    """基于 Claude Vision 的图像分析器。"""

    def __init__(self, model: str = "claude-3-5-sonnet-20241022"):
        self.client = anthropic.Anthropic()
        self.model = model

    def _prepare_image_content(
        self,
        image_source: str,
        is_url: bool = True
    ) -> dict:
        """将图像内容准备为 Claude API 所需的格式。"""
        if is_url:
            return {
                "type": "image",
                "source": {
                    "type": "url",
                    "url": image_source
                }
            }
        else:
            # 将本地文件编码为 Base64
            with open(image_source, "rb") as f:
                image_data = base64.standard_b64encode(f.read()).decode("utf-8")

            ext = Path(image_source).suffix.lower()
            media_type_map = {
                ".jpg": "image/jpeg",
                ".jpeg": "image/jpeg",
                ".png": "image/png",
                ".gif": "image/gif",
                ".webp": "image/webp"
            }
            media_type = media_type_map.get(ext, "image/jpeg")

            return {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": media_type,
                    "data": image_data
                }
            }

    def analyze(
        self,
        image_source: str,
        prompt: str,
        is_url: bool = True,
        system_prompt: str = None,
        max_tokens: int = 1000
    ) -> str:
        """分析图像。"""
        image_content = self._prepare_image_content(image_source, is_url)

        messages = [
            {
                "role": "user",
                "content": [
                    image_content,
                    {"type": "text", "text": prompt}
                ]
            }
        ]

        kwargs = {
            "model": self.model,
            "max_tokens": max_tokens,
            "messages": messages
        }

        if system_prompt:
            kwargs["system"] = system_prompt

        response = self.client.messages.create(**kwargs)
        return response.content[0].text

    def analyze_code_screenshot(
        self,
        screenshot_path: str
    ) -> dict:
        """分析代码截图并提取代码。"""
        system_prompt = """你是一名代码分析专家。
请从截图中准确提取代码并进行分析。"""

        extraction_prompt = """针对这张代码截图：
1. 准确提取代码（包含缩进）
2. 识别编程语言
3. 说明代码的主要功能
4. 指出潜在的 bug 或改进建议

请以以下 JSON 格式回复：
{
  "language": "编程语言",
  "code": "提取的代码",
  "description": "代码说明",
  "potential_issues": ["问题1", "问题2"],
  "improvements": ["改进建议1", "改进建议2"]
}"""

        response = self.analyze(
            screenshot_path,
            extraction_prompt,
            is_url=False,
            system_prompt=system_prompt,
            max_tokens=2000
        )

        import json
        try:
            start = response.find('{')
            end = response.rfind('}') + 1
            return json.loads(response[start:end])
        except json.JSONDecodeError:
            return {"raw_response": response}

    def compare_images(
        self,
        image_sources: list[tuple[str, bool]],  # (source, is_url) 对
        comparison_prompt: str
    ) -> str:
        """对多张图像进行比较分析。"""
        content = []

        for source, is_url in image_sources:
            content.append(self._prepare_image_content(source, is_url))

        content.append({"type": "text", "text": comparison_prompt})

        response = self.client.messages.create(
            model=self.model,
            max_tokens=2000,
            messages=[{"role": "user", "content": content}]
        )

        return response.content[0].text

    def analyze_ui_design(self, ui_screenshot_path: str) -> dict:
        """分析 UI 设计截图。"""
        prompt = """请以 UX/UI 专家的视角分析这张 UI 截图：

分析项目：
1. 布局结构
2. 配色方案
3. 排版
4. 可用性（Usability）评估
5. 无障碍（Accessibility）问题
6. 改进建议

请以 JSON 格式返回：
{
  "layout": "布局说明",
  "color_palette": ["主要颜色"],
  "typography": "排版评估",
  "usability_score": 0-10,
  "usability_issues": ["问题列表"],
  "accessibility_issues": ["无障碍问题"],
  "improvements": ["改进建议"]
}"""

        return self.analyze(
            ui_screenshot_path,
            prompt,
            is_url=False,
            max_tokens=1500
        )

# 使用示例
analyzer = ClaudeVisionAnalyzer()

# 图像分析
result = analyzer.analyze(
    "https://example.com/product.jpg",
    "请详细说明这款商品的特点，并推荐潜在的目标客群。",
    is_url=True
)
print(result)

import torch
import numpy as np
from PIL import Image
from transformers import CLIPModel, CLIPProcessor
import chromadb
from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction
import base64
import io

class MultimodalRAGSystem:
    """多模态 RAG 系统。"""

    def __init__(self):
        # 初始化 CLIP 模型
        self.clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
        self.clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

        # 初始化 ChromaDB
        self.chroma_client = chromadb.Client()
        self.image_collection = self.chroma_client.get_or_create_collection(
            name="images",
            metadata={"hnsw:space": "cosine"}
        )
        self.text_collection = self.chroma_client.get_or_create_collection(
            name="texts"
        )

    def get_image_embedding(self, image: Image.Image) -> np.ndarray:
        """将图像转换为 CLIP 嵌入。"""
        inputs = self.clip_processor(
            images=image,
            return_tensors="pt"
        )
        with torch.no_grad():
            features = self.clip_model.get_image_features(**inputs)
            features = torch.nn.functional.normalize(features, p=2, dim=-1)
        return features.numpy()[0]

    def get_text_embedding(self, text: str) -> np.ndarray:
        """将文本转换为 CLIP 嵌入。"""
        inputs = self.clip_processor(
            text=[text],
            return_tensors="pt",
            padding=True,
            truncation=True
        )
        with torch.no_grad():
            features = self.clip_model.get_text_features(**inputs)
            features = torch.nn.functional.normalize(features, p=2, dim=-1)
        return features.numpy()[0]

    def index_image(
        self,
        image: Image.Image,
        image_id: str,
        metadata: dict = None
    ):
        """对图像建立索引。"""
        embedding = self.get_image_embedding(image)

        # 将图像保存为 Base64
        buffer = io.BytesIO()
        image.save(buffer, format="PNG")
        image_b64 = base64.b64encode(buffer.getvalue()).decode('utf-8')

        doc_metadata = {"image_b64": image_b64}
        if metadata:
            doc_metadata.update(metadata)

        self.image_collection.add(
            embeddings=[embedding.tolist()],
            ids=[image_id],
            metadatas=[doc_metadata]
        )

    def search_images_by_text(
        self,
        query: str,
        n_results: int = 5
    ) -> list[dict]:
        """用文本检索图像。"""
        query_embedding = self.get_text_embedding(query)

        results = self.image_collection.query(
            query_embeddings=[query_embedding.tolist()],
            n_results=n_results,
            include=["metadatas", "distances", "ids"]
        )

        retrieved = []
        for i in range(len(results['ids'][0])):
            metadata = results['metadatas'][0][i]
            image_b64 = metadata.pop('image_b64', None)

            image = None
            if image_b64:
                image_bytes = base64.b64decode(image_b64)
                image = Image.open(io.BytesIO(image_bytes))

            retrieved.append({
                "id": results['ids'][0][i],
                "distance": results['distances'][0][i],
                "metadata": metadata,
                "image": image
            })

        return retrieved

    def multimodal_rag_query(
        self,
        question: str,
        vision_model_fn,  # GPT-4V、Claude Vision 等
        n_image_results: int = 3
    ) -> str:
        """执行多模态 RAG 查询。"""

        # 检索相关图像
        relevant_images = self.search_images_by_text(question, n_image_results)

        if not relevant_images:
            return vision_model_fn(question=question, images=[])

        # 用检索到的图像生成响应
        retrieved_images = [r["image"] for r in relevant_images if r["image"]]
        metadata_info = [
            f"图像 {i+1}: {r['metadata']}"
            for i, r in enumerate(relevant_images)
        ]

        enhanced_prompt = f"""
问题: {question}

相关图像信息:
{chr(10).join(metadata_info)}

请参考以上图像回答问题。
并具体引用每张图像中的相关内容。
"""

        return vision_model_fn(question=enhanced_prompt, images=retrieved_images)

# ColPali：使用视觉语言模型直接检索 PDF 页面
# pip install colpali-engine

from colpali_engine.models import ColPali, ColPaliProcessor
import torch

class ColPaliPDFSearch:
    """使用 ColPali 进行 PDF 页面检索。"""

    def __init__(self, model_name: str = "vidore/colpali-v1.2"):
        self.model = ColPali.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            device_map="cuda"
        )
        self.processor = ColPaliProcessor.from_pretrained(model_name)

    def index_pdf_pages(
        self,
        page_images: list[Image.Image]
    ) -> torch.Tensor:
        """对 PDF 页面图像建立索引。"""
        all_embeddings = []

        batch_size = 4
        for i in range(0, len(page_images), batch_size):
            batch = page_images[i:i + batch_size]
            inputs = self.processor.process_images(batch)
            inputs = {k: v.to("cuda") for k, v in inputs.items()}

            with torch.no_grad():
                embeddings = self.model(**inputs)

            all_embeddings.append(embeddings)

        return torch.cat(all_embeddings, dim=0)

    def search(
        self,
        query: str,
        page_embeddings: torch.Tensor,
        top_k: int = 3
    ) -> list[int]:
        """用查询检索相关的 PDF 页面。"""
        # 查询嵌入
        query_inputs = self.processor.process_queries([query])
        query_inputs = {k: v.to("cuda") for k, v in query_inputs.items()}

        with torch.no_grad():
            query_embedding = self.model(**query_inputs)

        # 计算 MaxSim 分数（ColPali 的核心机制）
        scores = self.processor.score_multi_vector(
            query_embedding,
            page_embeddings
        )

        # 返回 Top-K 页面索引
        top_indices = scores[0].argsort(descending=True)[:top_k]
        return top_indices.tolist()

from transformers import AutoModelForCausalLM, AutoProcessor
import torch
from PIL import Image

class Phi3VisionModel:
    """Microsoft Phi-3 Vision 模型。"""

    def __init__(self):
        model_id = "microsoft/Phi-3-vision-128k-instruct"

        self.model = AutoModelForCausalLM.from_pretrained(
            model_id,
            device_map="cuda",
            trust_remote_code=True,
            torch_dtype=torch.bfloat16,
            _attn_implementation='flash_attention_2'  # 需要 CUDA
        )

        self.processor = AutoProcessor.from_pretrained(
            model_id,
            trust_remote_code=True
        )

    def analyze(self, image: Image.Image, prompt: str) -> str:
        """分析图像。"""
        messages = [
            {"role": "user", "content": f"<|image_1|>\n{prompt}"}
        ]

        prompt_text = self.processor.tokenizer.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=True
        )

        inputs = self.processor(
            prompt_text,
            [image],
            return_tensors="pt"
        ).to("cuda")

        with torch.no_grad():
            output = self.model.generate(
                **inputs,
                max_new_tokens=500,
                eos_token_id=self.processor.tokenizer.eos_token_id
            )

        generated = output[0][inputs['input_ids'].shape[1]:]
        return self.processor.decode(generated, skip_special_tokens=True)

from transformers import Qwen2VLForConditionalGeneration, AutoProcessor
from qwen_vl_utils import process_vision_info
import torch

class QwenVLModel:
    """Qwen2-VL 多模态模型。"""

    def __init__(self, model_name: str = "Qwen/Qwen2-VL-7B-Instruct"):
        self.model = Qwen2VLForConditionalGeneration.from_pretrained(
            model_name,
            torch_dtype=torch.bfloat16,
            attn_implementation="flash_attention_2",
            device_map="auto"
        )
        self.processor = AutoProcessor.from_pretrained(
            model_name,
            min_pixels=256*28*28,
            max_pixels=1280*28*28
        )

    def analyze_image(
        self,
        image_path: str,
        question: str
    ) -> str:
        """分析图像。"""
        messages = [
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "image": image_path
                    },
                    {
                        "type": "text",
                        "text": question
                    }
                ]
            }
        ]

        text = self.processor.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=True
        )

        image_inputs, video_inputs = process_vision_info(messages)

        inputs = self.processor(
            text=[text],
            images=image_inputs,
            videos=video_inputs,
            padding=True,
            return_tensors="pt"
        ).to("cuda")

        with torch.no_grad():
            output_ids = self.model.generate(**inputs, max_new_tokens=512)

        generated_ids = [
            output_ids[len(input_ids):]
            for input_ids, output_ids in zip(inputs.input_ids, output_ids)
        ]

        return self.processor.batch_decode(
            generated_ids,
            skip_special_tokens=True,
            clean_up_tokenization_spaces=False
        )[0]

# 使用 Ollama 在本地运行多模态模型
# 从 ollama.ai 安装 Ollama

# 下载并运行 LLaVA 模型
ollama pull llava:13b

# 结合图像运行模型
ollama run llava:13b

import ollama
from pathlib import Path

class OllamaVisionModel:
    """使用 Ollama 的本地视觉模型。"""

    def __init__(self, model: str = "llava:13b"):
        self.model = model

    def analyze(
        self,
        image_path: str,
        prompt: str
    ) -> str:
        """用本地模型分析图像。"""
        response = ollama.chat(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": prompt,
                    "images": [image_path]
                }
            ]
        )
        return response["message"]["content"]

    def batch_analyze(
        self,
        image_paths: list[str],
        prompt: str
    ) -> list[str]:
        """依次分析多张图像。"""
        results = []
        for path in image_paths:
            result = self.analyze(path, prompt)
            results.append(result)
        return results

# 使用示例
model = OllamaVisionModel("llava:13b")
result = model.analyze(
    "/path/to/image.jpg",
    "这张图像中有什么？请详细描述。"
)
print(result)

from transformers import VideoMAEImageProcessor, VideoMAEModel
import torch
import numpy as np

class VideoFeatureExtractor:
    """使用 VideoMAE 的视频特征提取器。"""

    def __init__(self, model_name: str = "MCG-NJU/videomae-base"):
        self.processor = VideoMAEImageProcessor.from_pretrained(model_name)
        self.model = VideoMAEModel.from_pretrained(model_name)

    def extract_video_features(
        self,
        video_frames: list,  # PIL 图像或 numpy 数组列表
        num_frames: int = 16  # VideoMAE 通常使用 16 帧
    ) -> torch.Tensor:
        """从视频帧中提取特征。"""

        # 均匀采样帧
        total_frames = len(video_frames)
        indices = np.linspace(0, total_frames - 1, num_frames, dtype=int)
        sampled_frames = [video_frames[i] for i in indices]

        # 预处理
        inputs = self.processor(sampled_frames, return_tensors="pt")

        with torch.no_grad():
            outputs = self.model(**inputs)

        # [batch, num_patches, hidden_size] 形状
        return outputs.last_hidden_state

# 用 OpenCV 提取视频帧
import cv2

def extract_frames_from_video(
    video_path: str,
    target_fps: int = 1
) -> list:
    """从视频中提取帧。"""
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_interval = int(fps / target_fps)

    frames = []
    frame_count = 0

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        if frame_count % frame_interval == 0:
            # 将 BGR 转换为 RGB
            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            from PIL import Image
            pil_frame = Image.fromarray(frame_rgb)
            frames.append(pil_frame)

        frame_count += 1

    cap.release()
    return frames

import google.generativeai as genai
import time

class LongVideoUnderstanding:
    """使用 Gemini 1.5 Pro 的长视频理解系统。"""

    def __init__(self):
        self.model = genai.GenerativeModel("gemini-1.5-pro")

    def analyze_long_video(
        self,
        video_path: str,
        analysis_tasks: list[str]
    ) -> dict:
        """分析长达 1 小时的视频。"""

        print("正在上传视频...")
        video_file = genai.upload_file(
            path=video_path,
            display_name="long_video_analysis"
        )

        # 等待处理完成
        while video_file.state.name == "PROCESSING":
            print(f"处理中... (状态: {video_file.state.name})")
            time.sleep(15)
            video_file = genai.get_file(video_file.name)

        if video_file.state.name != "ACTIVE":
            raise RuntimeError(f"视频处理失败: {video_file.state.name}")

        print(f"上传完成 (URI: {video_file.uri})")

        results = {}

        for task in analysis_tasks:
            print(f"正在分析: {task}")
            response = self.model.generate_content(
                [video_file, task],
                request_options={"timeout": 900}
            )
            results[task] = response.text

        # 清理已上传的文件
        genai.delete_file(video_file.name)
        print("文件清理完成")

        return results

    def create_video_summary(self, video_path: str) -> dict:
        """生成视频的综合摘要。"""
        tasks = [
            "请用 3-5 句话总结这段视频的整体内容。",
            "请列出主要场景及其时间戳。格式：MM:SS - 说明",
            "请以列表形式列出视频中出现的主要人物、物体、地点。",
            "这段视频中强调的核心信息或结论是什么？",
            "这段视频的目标受众和目的是什么？"
        ]

        return self.analyze_long_video(video_path, tasks)

# 视频理解系统的使用
video_analyzer = LongVideoUnderstanding()
summary = video_analyzer.create_video_summary("lecture.mp4")

for task, result in summary.items():
    print(f"\n{'='*50}")
    print(f"问题: {task}")
    print(f"答案: {result}")