التمثيلات عبر النماذج: ربط نماذج الذكاء الاصطناعي

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

# تحميل النموذج المدرب مسبقًا
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)

# الحصول على درجات التشابه المتقاطعة
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):
        """بناء فايس لمؤشر للبحث السريع بالتشابه"""
        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

# تكميم ديناميكي للاستنتاج على وحدة المعالجة المركزية
quantized_model = quantize_dynamic(
    model, 
    {torch.nn.Linear}, 
    dtype=torch.qint8
)

import torch.onnx

dummy_input = processor(text=["مثال"], 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("الصور", 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"صورة لـ {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 = ["قطة", "كلب", "طائر", "أسد", " حصان"]
predictions = zero_shot_classify(image, labels, model, processor)
print(f"التنبؤ الأعلى: {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))
        
        # إرجاع الأعلى 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"السياق:\n{context}\n\nالسؤال: {query}\n\nالإجابة:"
        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 = [
            "المحتوى العنيف",
            "المحتوى البذيء",
            "الصور الكراهية",
            "العنف المتصور",
            "المحتوى الصريح"
        ]
        
        self.safe_categories = [
            "آمن للعمل",
            "مناسب للأسرة",
            "المحتوى التعليمي"
        ]
    
    def moderate_image(self, image, threshold: float = 0.3):
        """التحقق من وجود محتوى غير آمن في الصورة"""
        # تجميع جميع الفئات
        all_categories = self.unsafe_categories + self.safe_categories
        text_inputs = [f"صورة تحتوي على {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}