Кросс-модальные вложения: объединение модулей ИИ

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

# Загрузка предобученной модели CLIP
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

# Подготовка входных данных
image = Image.open("example.jpg")
texts = ["фото кошки", "фото собаки", "фото птицы"]

# Обработка и получение вложений
inputs = processor(text=texts, images=image, return_tensors="pt", padding=True)

# Получение оценок кросc-модальной схожести
outputs = model(**inputs)
logits_per_image = outputs.logits_per_image
probs = logits_per_image.softmax(dim=1)

print(f"Вероятности: {probs}")

import open_clip

model, _, preprocess = open_clip.create_model_and_transforms(
    'ViT-L-14',
    pretrained='laion2b_s32b_b82k'
)
tokenizer = open_clip.get_tokenizer('ViT-L-14')

import numpy as np
from typing import List, Tuple
import faiss
from transformers import CLIPModel, CLIPProcessor

class CrossModalSearchEngine:
    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.image_index = None
        self.image_metadata = []

    def encode_images(self, images: List) -> np.ndarray:
        """Кодирование изображений в вложения"""
        inputs = self.processor(images=images, return_tensors="pt", padding=True)
        with torch.no_grad():
            embeddings = self.model.get_image_features(**inputs)
        return embeddings.cpu().numpy()

    def encode_text(self, texts: List[str]) -> np.ndarray:
        """Кодирование текстовых запросов в вложения"""
        inputs = self.processor(text=texts, return_tensors="pt", padding=True)
        with torch.no_grad():
            embeddings = self.model.get_text_features(**inputs)
        return embeddings.cpu().numpy()

    def build_index(self, image_embeddings: np.ndarray):
        """Создание индекса FAISS для эффективного поиска схожести"""
        dimension = image_embeddings.shape[1]

        # Нормализация вложений для косинусной схожести
        faiss.normalize_L2(image_embeddings)

        # Создание индекса (использование HNSW для масштабируемого развертывания)
        self.image_index = faiss.IndexHNSWFlat(dimension, 32)
        self.image_index.hnsw.efConstruction = 40
        self.image_index.add(image_embeddings)

    def search(self, query: str, k: int = 10) -> List[Tuple[int, float]]:
        """Поиск изображений по текстовым запросам"""
        query_embedding = self.encode_text([query])
        faiss.normalize_L2(query_embedding)

        distances, indices = self.image_index.search(query_embedding, k)
        return list(zip(indices[0], distances[0]))

# Пример использования
engine = CrossModalSearchEngine()

# Построение индекса из коллекции изображений
image_embeddings = engine.encode_images(image_collection)
engine.build_index(image_embeddings)

# Поиск по тексту
results = engine.search("закат над горами", k=5)

from transformers import CLIPModel, CLIPProcessor, AdamW
import torch.nn as nn

class FineTuneCLIP:
    def __init__(self, model_name: str, num_epochs: int = 10):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.num_epochs = num_epochs

    def contrastive_loss(self, image_embeddings, text_embeddings, temperature=0.07):
        """Вычисление контрастной потери InfoNCE"""
        # Нормализация вложений
        image_embeddings = nn.functional.normalize(image_embeddings, dim=1)
        text_embeddings = nn.functional.normalize(text_embeddings, dim=1)

        # Вычисление матрицы схожести
        logits = torch.matmul(image_embeddings, text_embeddings.T) / temperature

        # Метки - это диагональ (соответствующие пары)
        labels = torch.arange(len(logits), device=logits.device)

        # Симметричная потеря
        loss_i = nn.functional.cross_entropy(logits, labels)
        loss_t = nn.functional.cross_entropy(logits.T, labels)

        return (loss_i + loss_t) / 2

    def train(self, dataloader, learning_rate=5e-6):
        """Тонкая настройка на специфичных для домена данных"""
        optimizer = AdamW(self.model.parameters(), lr=learning_rate)

        self.model.train()
        for epoch in range(self.num_epochs):
            total_loss = 0
            for batch in dataloader:
                images, texts = batch['images'], batch['texts']

                # Обработка входных данных
                inputs = self.processor(
                    text=texts,
                    images=images,
                    return_tensors="pt",
                    padding=True
                )

                # Прямой проход
                outputs = self.model(**inputs, return_loss=True)
                loss = outputs.loss

                # Обратный проход
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()

                total_loss += loss.item()

            avg_loss = total_loss / len(dataloader)
            print(f"Эпоха {epoch+1}/{self.num_epochs}, Потеря: {avg_loss:.4f}")

import torch
from torch.quantization import quantize_dynamic

# Динамическое квантование для инференса на CPU
quantized_model = quantize_dynamic(
    model,
    {torch.nn.Linear},
    dtype=torch.qint8
)

import torch.onnx

dummy_input = processor(text=["sample"], return_tensors="pt")
torch.onnx.export(
    model,
    tuple(dummy_input.values()),
    "clip_model.onnx",
    input_names=['input_ids', 'attention_mask'],
    output_names=['output'],
    dynamic_axes={
        'input_ids': {0: 'batch_size'},
        'attention_mask': {0: 'batch_size'}
    }
)

def batch_encode(items: List, batch_size: int = 32):
    """Обработка элементов пакетами для повышения эффективности"""
    embeddings = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i+batch_size]
        batch_embeddings = encode_batch(batch)
        embeddings.append(batch_embeddings)
    return np.concatenate(embeddings, axis=0)

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# Определение схемы
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=512),
    FieldSchema(name="metadata", dtype=DataType.JSON)
]
schema = CollectionSchema(fields, description="Эмбеддинги изображений")

# Создание коллекции
collection = Collection("images", schema)

# Создание индекса
index_params = {
    "metric_type": "IP",  # Внутреннее произведение (косинусное сходство)
    "index_type": "IVF_FLAT",
    "params": {"nlist": 1024}
}
collection.create_index("embedding", index_params)

def zero_shot_classify(image, candidate_labels: List[str], model, processor):
    """Классификация изображения в произвольные категории"""
    # Создание текстовых запросов
    text_inputs = [f"a photo of a {label}" for label in candidate_labels]

    # Получение эмбеддингов
    inputs = processor(
        text=text_inputs,
        images=image,
        return_tensors="pt",
        padding=True
    )
    outputs = model(**inputs)

    # Вычисление вероятностей
    logits = outputs.logits_per_image
    probs = logits.softmax(dim=1)

    # Возврат ранжированных предсказаний
    sorted_indices = probs.argsort(descending=True)[0]
    return [(candidate_labels[idx], probs[0][idx].item()) for idx in sorted_indices]

# Пример использования
labels = ["cat", "dog", "bird", "fish", "horse"]
predictions = zero_shot_classify(image, labels, model, processor)
print(f"Top prediction: {predictions[0]}")

class MultimodalRAG:
    def __init__(self, clip_model, llm_model):
        self.clip = clip_model
        self.llm = llm_model
        self.knowledge_base = []

    def add_documents(self, images, texts, metadata):
        """Добавление мультимодальных документов в базу знаний"""
        image_embeds = self.clip.encode_images(images)
        text_embeds = self.clip.encode_text(texts)

        # Хранение объединенной информации
        for i, (img_emb, txt_emb, meta) in enumerate(
            zip(image_embeds, text_embeds, metadata)
        ):
            self.knowledge_base.append({
                'image_embedding': img_emb,
                'text_embedding': txt_emb,
                'metadata': meta,
                'index': i
            })

    def retrieve(self, query: str, k: int = 5):
        """Извлечение релевантного мультимодального контента"""
        query_embedding = self.clip.encode_text([query])[0]

        # Вычисление сходства
        similarities = []
        for doc in self.knowledge_base:
            # Среднее сходство по модальностям
            img_sim = np.dot(query_embedding, doc['image_embedding'])
            txt_sim = np.dot(query_embedding, doc['text_embedding'])
            combined_sim = (img_sim + txt_sim) / 2
            similarities.append((combined_sim, doc))

        # Возврат top-k
        similarities.sort(reverse=True)
        return [doc for _, doc in similarities[:k]]

    def answer_query(self, query: str):
        """Ответ на запрос с использованием извлеченного мультимодального контекста"""
        retrieved_docs = self.retrieve(query)

        # Конструирование контекста из извлеченных документов
        context = "\n".join([doc['metadata']['text'] for doc in retrieved_docs])

        # Генерация ответа с LLM
        prompt = f"Context:\n{context}\n\nQuestion: {query}\n\nAnswer:"
        answer = self.llm.generate(prompt)

        return answer, retrieved_docs

class ContentModerator:
    def __init__(self, model, processor):
        self.model = model
        self.processor = processor

        # Определение категорий безопасности
        self.unsafe_categories = [
            "violent content",
            "adult content",
            "hateful imagery",
            "graphic violence",
            "explicit material"
        ]

        self.safe_categories = [
            "safe for work",
            "family friendly",
            "educational content"
        ]

    def moderate_image(self, image, threshold: float = 0.3):
        """Проверка изображения на наличие небезопасного контента"""
        # Объединение всех категорий
        all_categories = self.unsafe_categories + self.safe_categories
        text_inputs = [f"image containing {cat}" for cat in all_categories]

        # Получение предсказаний
        inputs = self.processor(
            text=text_inputs,
            images=image,
            return_tensors="pt"
        )
        outputs = self.model(**inputs)
        probs = outputs.logits_per_image.softmax(dim=1)[0]

        # Проверка небезопасных категорий
        unsafe_scores = probs[:len(self.unsafe_categories)]
        max_unsafe_score = unsafe_scores.max().item()

        return {
            'is_safe': max_unsafe_score < threshold,
            'confidence': 1 - max_unsafe_score,
            'flagged_categories': [
                self.unsafe_categories[i]
                for i, score in enumerate(unsafe_scores)
                if score > threshold
            ]
        }

from torchvision import transforms

# Стандартная предобработка для CLIP
clip_transform = transforms.Compose([
    transforms.Resize(224, interpolation=transforms.InterpolationMode.BICUBIC),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(
        mean=[0.48145466, 0.4578275, 0.40821073],
        std=[0.26862954, 0.26130258, 0.27577711]
    )
])

def assess_embedding_quality(embeddings: np.ndarray):
    """Вычисление метрик для качества эмбеддингов"""
    # Вычисление среднего парного расстояния
    distances = np.linalg.norm(
        embeddings[:, None] - embeddings[None, :],
        axis=2
    )
    avg_distance = distances.mean()

    # Проверка на кластеризацию (низкое внутрикластерное расстояние)
    from sklearn.cluster import KMeans
    kmeans = KMeans(n_clusters=10)
    labels = kmeans.fit_predict(embeddings)

    # Вычисление коэффициента силуэта
    from sklearn.metrics import silhouette_score
    score = silhouette_score(embeddings, labels)

    return {
        'avg_pairwise_distance': avg_distance,
        'silhouette_score': score
    }

FROM nvidia/cuda:11.8.0-cudnn8-runtime-ubuntu22.04

# Установка Python и зависимостей
RUN apt-get update && apt-get install -y python3-pip

WORKDIR /app

# Установка требований
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt

# Копирование кода приложения
COPY . .

# Открытие порта API
EXPOSE 8000

# Запуск сервера API
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

# api.py - FastAPI сервер для кроссмодальных эмбеддингов
from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel
import torch
from PIL import Image
import io

app = FastAPI()

# Загрузка модели при старте
@app.on_event("startup")
async def load_model():
    global model, processor
    model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
    processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
    if torch.cuda.is_available():
        model = model.cuda()

class TextQuery(BaseModel):
    text: str

@app.post("/embed/text")
async def embed_text(query: TextQuery):
    inputs = processor(text=[query.text], return_tensors="pt", padding=True)
    if torch.cuda.is_available():
        inputs = {k: v.cuda() for k, v in inputs.items()}

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

    return {"embedding": embeddings.cpu().numpy().tolist()}

@app.post("/embed/image")
async def embed_image(file: UploadFile = File(...)):
    image = Image.open(io.BytesIO(await file.read()))
    inputs = processor(images=image, return_tensors="pt")
    if torch.cuda.is_available():
        inputs = {k: v.cuda() for k, v in inputs.items()}

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

    return {"embedding": embeddings.cpu().numpy().tolist()}

@app.post("/similarity")
async def compute_similarity(
    text: TextQuery,
    file: UploadFile = File(...)
):
    # Получение обоих эмбеддингов
    image = Image.open(io.BytesIO(await file.read()))
    inputs = processor(text=[text.text], images=image, return_tensors="pt", padding=True)

    if torch.cuda.is_available():
        inputs = {k: v.cuda() for k, v in inputs.items()}

    outputs = model(**inputs)
    similarity = outputs.logits_per_image[0][0].item()

    return {"similarity": similarity}