Multi-Modal LLM Integration: Beyond Text in Enterprise Applications

  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
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584

import asyncio
import torch
import numpy as np
from typing import Dict, List, Optional, Union, Any
from dataclasses import dataclass
from enum import Enum
import cv2
import librosa
import pytesseract
from PIL import Image
import whisper

class ModalityType(Enum):
    TEXT = "text"
    IMAGE = "image"
    AUDIO = "audio"
    DOCUMENT = "document"
    VIDEO = "video"

@dataclass
class ModalInput:
    """Structured input for multi-modal processing"""
    data: Any
    modality: ModalityType
    metadata: Dict[str, Any]
    priority: int = 1
    processing_hints: Optional[Dict[str, Any]] = None

@dataclass
class ProcessedInput:
    """Processed input ready for model consumption"""
    features: torch.Tensor
    embeddings: Optional[torch.Tensor]
    modality: ModalityType
    attention_mask: Optional[torch.Tensor]
    metadata: Dict[str, Any]

class MultiModalPreprocessor:
    """Production-grade multi-modal preprocessing pipeline"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        
        # Initialize modality-specific processors
        self.text_processor = TextPreprocessor(config['text'])
        self.image_processor = ImagePreprocessor(config['image'])
        self.audio_processor = AudioPreprocessor(config['audio'])
        self.document_processor = DocumentPreprocessor(config['document'])
        
        # Initialize feature extractors
        self._initialize_feature_extractors()
        
    def _initialize_feature_extractors(self):
        """Initialize pre-trained feature extractors"""
        from transformers import AutoTokenizer, AutoModel
        from torchvision import models, transforms
        
        # Text feature extractor
        self.text_tokenizer = AutoTokenizer.from_pretrained(
            self.config['text']['model_name']
        )
        self.text_encoder = AutoModel.from_pretrained(
            self.config['text']['model_name']
        ).to(self.device)
        
        # Vision feature extractor
        self.vision_encoder = models.resnet50(pretrained=True).to(self.device)
        self.vision_encoder.eval()
        
        # Remove final classification layer
        self.vision_encoder = torch.nn.Sequential(
            *(list(self.vision_encoder.children())[:-1])
        )
        
        # Vision preprocessing
        self.vision_transform = transforms.Compose([
            transforms.Resize(256),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize(
                mean=[0.485, 0.456, 0.406],
                std=[0.229, 0.224, 0.225]
            )
        ])
        
        # Audio feature extractor
        if self.config.get('audio', {}).get('enabled', False):
            self.audio_encoder = whisper.load_model("base")
    
    async def process_inputs(self, inputs: List[ModalInput]) -> List[ProcessedInput]:
        """Process multiple inputs in parallel"""
        tasks = []
        for modal_input in inputs:
            task = self._process_single_input(modal_input)
            tasks.append(task)
        
        processed_inputs = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Filter out exceptions and log errors
        valid_inputs = []
        for i, result in enumerate(processed_inputs):
            if isinstance(result, Exception):
                logger.error(f"Failed to process input {i}: {result}")
            else:
                valid_inputs.append(result)
        
        return valid_inputs
    
    async def _process_single_input(self, modal_input: ModalInput) -> ProcessedInput:
        """Process a single modal input"""
        try:
            if modal_input.modality == ModalityType.TEXT:
                return await self._process_text(modal_input)
            elif modal_input.modality == ModalityType.IMAGE:
                return await self._process_image(modal_input)
            elif modal_input.modality == ModalityType.AUDIO:
                return await self._process_audio(modal_input)
            elif modal_input.modality == ModalityType.DOCUMENT:
                return await self._process_document(modal_input)
            elif modal_input.modality == ModalityType.VIDEO:
                return await self._process_video(modal_input)
            else:
                raise ValueError(f"Unsupported modality: {modal_input.modality}")
        except Exception as e:
            logger.error(f"Processing error for {modal_input.modality}: {e}")
            raise
    
    async def _process_text(self, modal_input: ModalInput) -> ProcessedInput:
        """Process text input with tokenization and embedding"""
        text = modal_input.data
        
        # Clean and normalize text
        cleaned_text = self.text_processor.clean_text(text)
        
        # Tokenize
        tokens = self.text_tokenizer(
            cleaned_text,
            padding='max_length',
            truncation=True,
            max_length=self.config['text']['max_length'],
            return_tensors='pt'
        )
        
        # Generate embeddings
        with torch.no_grad():
            embeddings = self.text_encoder(**tokens.to(self.device))
            pooled_embeddings = torch.mean(embeddings.last_hidden_state, dim=1)
        
        return ProcessedInput(
            features=tokens['input_ids'].squeeze(),
            embeddings=pooled_embeddings.cpu(),
            modality=ModalityType.TEXT,
            attention_mask=tokens['attention_mask'].squeeze(),
            metadata={
                **modal_input.metadata,
                'text_length': len(cleaned_text),
                'token_count': tokens['input_ids'].shape[-1]
            }
        )
    
    async def _process_image(self, modal_input: ModalInput) -> ProcessedInput:
        """Process image input with feature extraction"""
        image_data = modal_input.data
        
        # Handle different image input formats
        if isinstance(image_data, str):  # File path
            image = Image.open(image_data).convert('RGB')
        elif isinstance(image_data, np.ndarray):  # NumPy array
            image = Image.fromarray(image_data)
        elif isinstance(image_data, Image.Image):  # PIL Image
            image = image_data.convert('RGB')
        else:
            raise ValueError(f"Unsupported image format: {type(image_data)}")
        
        # Apply preprocessing
        processed_image = self.vision_transform(image).unsqueeze(0)
        
        # Extract features
        with torch.no_grad():
            features = self.vision_encoder(processed_image.to(self.device))
            features = features.flatten(start_dim=1)
        
        # Additional image analysis
        analysis = await self._analyze_image(image, modal_input.processing_hints)
        
        return ProcessedInput(
            features=processed_image.squeeze(),
            embeddings=features.cpu(),
            modality=ModalityType.IMAGE,
            attention_mask=None,
            metadata={
                **modal_input.metadata,
                'image_size': image.size,
                'analysis': analysis
            }
        )
    
    async def _analyze_image(self, image: Image.Image, hints: Optional[Dict]) -> Dict:
        """Perform additional image analysis based on processing hints"""
        analysis = {}
        
        if hints and 'detect_text' in hints:
            # OCR text detection
            try:
                extracted_text = pytesseract.image_to_string(image)
                analysis['extracted_text'] = extracted_text.strip()
            except Exception as e:
                analysis['ocr_error'] = str(e)
        
        if hints and 'detect_objects' in hints:
            # Object detection (would integrate with YOLO or similar)
            analysis['objects'] = await self._detect_objects(image)
        
        # Basic image properties
        analysis['format'] = image.format
        analysis['mode'] = image.mode
        analysis['has_transparency'] = image.mode in ('RGBA', 'LA') or 'transparency' in image.info
        
        return analysis
    
    async def _process_audio(self, modal_input: ModalInput) -> ProcessedInput:
        """Process audio input with transcription and feature extraction"""
        audio_data = modal_input.data
        
        # Load audio file
        if isinstance(audio_data, str):  # File path
            audio, sr = librosa.load(audio_data, sr=self.config['audio']['sample_rate'])
        else:
            audio, sr = audio_data, self.config['audio']['sample_rate']
        
        # Extract audio features
        features = await self._extract_audio_features(audio, sr)
        
        # Transcribe audio if requested
        transcript = None
        if modal_input.processing_hints and modal_input.processing_hints.get('transcribe', True):
            try:
                result = self.audio_encoder.transcribe(audio_data if isinstance(audio_data, str) else audio)
                transcript = result['text']
            except Exception as e:
                logger.error(f"Transcription error: {e}")
        
        return ProcessedInput(
            features=torch.tensor(features),
            embeddings=torch.tensor(features[:512]),  # Take first 512 features for embedding
            modality=ModalityType.AUDIO,
            attention_mask=None,
            metadata={
                **modal_input.metadata,
                'duration': len(audio) / sr,
                'sample_rate': sr,
                'transcript': transcript
            }
        )
    
    async def _extract_audio_features(self, audio: np.ndarray, sr: int) -> np.ndarray:
        """Extract comprehensive audio features"""
        features = []
        
        # MFCC features
        mfccs = librosa.feature.mfcc(y=audio, sr=sr, n_mfcc=13)
        features.extend(np.mean(mfccs, axis=1))
        
        # Spectral features
        spectral_centroids = librosa.feature.spectral_centroid(y=audio, sr=sr)
        features.append(np.mean(spectral_centroids))
        
        spectral_rolloff = librosa.feature.spectral_rolloff(y=audio, sr=sr)
        features.append(np.mean(spectral_rolloff))
        
        # Zero crossing rate
        zcr = librosa.feature.zero_crossing_rate(audio)
        features.append(np.mean(zcr))
        
        # Chroma features
        chroma = librosa.feature.chroma_stft(y=audio, sr=sr)
        features.extend(np.mean(chroma, axis=1))
        
        return np.array(features)

class MultiModalFusionEngine:
    """Advanced fusion engine for multi-modal inputs"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.fusion_strategy = config.get('fusion_strategy', 'late')
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        
        # Initialize fusion networks
        self._initialize_fusion_networks()
    
    def _initialize_fusion_networks(self):
        """Initialize different fusion architectures"""
        
        # Early fusion network (feature-level concatenation)
        self.early_fusion_net = nn.Sequential(
            nn.Linear(self.config['total_feature_dim'], 1024),
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Linear(1024, 512),
            nn.ReLU(),
            nn.Linear(512, self.config['output_dim'])
        ).to(self.device)
        
        # Late fusion network (decision-level combination)
        self.modality_weights = nn.Parameter(
            torch.ones(len(ModalityType)) / len(ModalityType)
        )
        
        # Cross-modal attention for hybrid fusion
        self.cross_modal_attention = nn.MultiheadAttention(
            embed_dim=self.config['embedding_dim'],
            num_heads=8,
            dropout=0.1,
            batch_first=True
        ).to(self.device)
        
        # Modality-specific projectors
        self.modality_projectors = nn.ModuleDict({
            modality.value: nn.Linear(
                self.config['modality_dims'][modality.value],
                self.config['embedding_dim']
            )
            for modality in ModalityType
        }).to(self.device)
    
    async def fuse_inputs(self, processed_inputs: List[ProcessedInput]) -> torch.Tensor:
        """Fuse multi-modal inputs using configured strategy"""
        
        if self.fusion_strategy == 'early':
            return await self._early_fusion(processed_inputs)
        elif self.fusion_strategy == 'late':
            return await self._late_fusion(processed_inputs)
        elif self.fusion_strategy == 'hybrid':
            return await self._hybrid_fusion(processed_inputs)
        else:
            raise ValueError(f"Unknown fusion strategy: {self.fusion_strategy}")
    
    async def _early_fusion(self, processed_inputs: List[ProcessedInput]) -> torch.Tensor:
        """Early fusion: concatenate features before processing"""
        
        # Project all modalities to common feature space
        projected_features = []
        
        for inp in processed_inputs:
            if inp.embeddings is not None:
                projector = self.modality_projectors[inp.modality.value]
                projected = projector(inp.embeddings.to(self.device))
                projected_features.append(projected)
        
        if not projected_features:
            raise ValueError("No valid inputs for fusion")
        
        # Concatenate features
        concatenated = torch.cat(projected_features, dim=-1)
        
        # Process through fusion network
        fused_output = self.early_fusion_net(concatenated)
        
        return fused_output
    
    async def _late_fusion(self, processed_inputs: List[ProcessedInput]) -> torch.Tensor:
        """Late fusion: combine decisions from individual modalities"""
        
        modality_outputs = {}
        
        for inp in processed_inputs:
            if inp.embeddings is not None:
                # Process each modality independently
                projector = self.modality_projectors[inp.modality.value]
                modality_output = projector(inp.embeddings.to(self.device))
                modality_outputs[inp.modality] = modality_output
        
        # Weighted combination of modality outputs
        weighted_outputs = []
        modality_indices = []
        
        for modality, output in modality_outputs.items():
            modality_idx = list(ModalityType).index(modality)
            weight = torch.softmax(self.modality_weights, dim=0)[modality_idx]
            weighted_outputs.append(weight * output)
            modality_indices.append(modality_idx)
        
        # Sum weighted outputs
        fused_output = torch.sum(torch.stack(weighted_outputs), dim=0)
        
        return fused_output
    
    async def _hybrid_fusion(self, processed_inputs: List[ProcessedInput]) -> torch.Tensor:
        """Hybrid fusion: cross-modal attention + weighted combination"""
        
        # Project inputs to common embedding space
        embeddings = []
        
        for inp in processed_inputs:
            if inp.embeddings is not None:
                projector = self.modality_projectors[inp.modality.value]
                projected = projector(inp.embeddings.to(self.device))
                embeddings.append(projected.unsqueeze(0))  # Add batch dimension
        
        if len(embeddings) < 2:
            # Fall back to single modality processing
            return embeddings[0].squeeze(0) if embeddings else torch.zeros(self.config['embedding_dim']).to(self.device)
        
        # Stack embeddings
        stacked_embeddings = torch.cat(embeddings, dim=0).unsqueeze(0)  # [1, num_modalities, embed_dim]
        
        # Apply cross-modal attention
        attended_embeddings, attention_weights = self.cross_modal_attention(
            stacked_embeddings, stacked_embeddings, stacked_embeddings
        )
        
        # Global pooling
        fused_output = torch.mean(attended_embeddings.squeeze(0), dim=0)
        
        return fused_output

class ProductionMultiModalLLM:
    """Production-ready multi-modal LLM integration"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.preprocessor = MultiModalPreprocessor(config['preprocessing'])
        self.fusion_engine = MultiModalFusionEngine(config['fusion'])
        
        # Initialize the base LLM
        self.llm = self._initialize_llm()
        
        # Performance monitoring
        self.metrics = MultiModalMetrics()
        
        # Caching for repeated inputs
        self.cache = MultiModalCache(config['caching'])
    
    def _initialize_llm(self):
        """Initialize the base language model"""
        from transformers import AutoTokenizer, AutoModelForCausalLM
        
        model_name = self.config['llm']['model_name']
        
        tokenizer = AutoTokenizer.from_pretrained(model_name)
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16 if self.config['llm'].get('use_fp16', False) else torch.float32,
            device_map="auto" if self.config['llm'].get('use_device_map', False) else None
        )
        
        return {'tokenizer': tokenizer, 'model': model}
    
    async def process_multimodal_query(self, 
                                      query: str,
                                      modal_inputs: List[ModalInput]) -> Dict[str, Any]:
        """Process a multi-modal query with context"""
        
        start_time = asyncio.get_event_loop().time()
        
        try:
            # Check cache for repeated inputs
            cache_key = self._generate_cache_key(query, modal_inputs)
            cached_result = await self.cache.get(cache_key)
            if cached_result:
                self.metrics.record_cache_hit()
                return cached_result
            
            # Preprocess all modal inputs
            processed_inputs = await self.preprocessor.process_inputs(modal_inputs)
            
            if not processed_inputs:
                return {'error': 'No valid inputs processed', 'response': None}
            
            # Fuse multi-modal inputs
            fused_features = await self.fusion_engine.fuse_inputs(processed_inputs)
            
            # Prepare context for LLM
            context = self._prepare_llm_context(query, processed_inputs, fused_features)
            
            # Generate response
            response = await self._generate_response(context)
            
            # Prepare result
            result = {
                'response': response,
                'processed_modalities': [inp.modality.value for inp in processed_inputs],
                'context_length': len(context),
                'processing_time': asyncio.get_event_loop().time() - start_time
            }
            
            # Cache result
            await self.cache.set(cache_key, result)
            
            # Record metrics
            self.metrics.record_successful_query(
                modalities=[inp.modality for inp in processed_inputs],
                processing_time=result['processing_time']
            )
            
            return result
            
        except Exception as e:
            self.metrics.record_failed_query(str(e))
            return {'error': str(e), 'response': None}
    
    def _prepare_llm_context(self, 
                           query: str,
                           processed_inputs: List[ProcessedInput],
                           fused_features: torch.Tensor) -> str:
        """Prepare context string for LLM consumption"""
        
        context_parts = [f"User Query: {query}\n"]
        
        # Add information about processed inputs
        context_parts.append("Available Context:")
        
        for inp in processed_inputs:
            context_parts.append(f"- {inp.modality.value.title()}:")
            
            if inp.modality == ModalityType.TEXT:
                # Include text content directly
                if 'original_text' in inp.metadata:
                    context_parts.append(f"  Content: {inp.metadata['original_text'][:500]}...")
            
            elif inp.modality == ModalityType.IMAGE:
                # Include image analysis results
                if 'analysis' in inp.metadata:
                    analysis = inp.metadata['analysis']
                    if 'extracted_text' in analysis:
                        context_parts.append(f"  Extracted Text: {analysis['extracted_text']}")
                    if 'objects' in analysis:
                        context_parts.append(f"  Detected Objects: {', '.join(analysis['objects'])}")
                context_parts.append(f"  Image Size: {inp.metadata.get('image_size', 'Unknown')}")
            
            elif inp.modality == ModalityType.AUDIO:
                # Include transcription and audio properties
                if 'transcript' in inp.metadata and inp.metadata['transcript']:
                    context_parts.append(f"  Transcript: {inp.metadata['transcript']}")
                context_parts.append(f"  Duration: {inp.metadata.get('duration', 0):.2f}s")
            
            elif inp.modality == ModalityType.DOCUMENT:
                # Include document content and structure
                if 'extracted_content' in inp.metadata:
                    content = inp.metadata['extracted_content'][:800]
                    context_parts.append(f"  Content Preview: {content}...")
                if 'structure' in inp.metadata:
                    context_parts.append(f"  Document Structure: {inp.metadata['structure']}")
        
        # Add fusion information
        context_parts.append(f"\nMulti-modal Analysis: The provided inputs have been analyzed and integrated.")
        context_parts.append("Please provide a comprehensive response based on all available context.\n")
        
        return "\n".join(context_parts)
    
    async def _generate_response(self, context: str) -> str:
        """Generate response using the base LLM"""
        
        tokenizer = self.llm['tokenizer']
        model = self.llm['model']
        
        # Tokenize input
        inputs = tokenizer(
            context,
            return_tensors="pt",
            max_length=self.config['llm']['max_context_length'],
            truncation=True
        )
        
        # Generate response
        with torch.no_grad():
            outputs = model.generate(
                inputs['input_ids'].to(model.device),
                attention_mask=inputs['attention_mask'].to(model.device),
                max_new_tokens=self.config['llm']['max_new_tokens'],
                do_sample=True,
                temperature=self.config['llm']['temperature'],
                pad_token_id=tokenizer.eos_token_id
            )
        
        # Decode response
        response = tokenizer.decode(
            outputs[0][inputs['input_ids'].shape[1]:],  # Skip input tokens
            skip_special_tokens=True
        )
        
        return response.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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

class CrossModalAttentionLayer(nn.Module):
    """Advanced cross-modal attention for fine-grained interaction modeling"""
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__()
        self.config = config
        self.embed_dim = config['embed_dim']
        self.num_heads = config['num_heads']
        
        # Modality-specific query, key, value projections
        self.modality_projections = nn.ModuleDict()
        for modality in ['text', 'vision', 'audio']:
            self.modality_projections[modality] = nn.ModuleDict({
                'query': nn.Linear(self.embed_dim, self.embed_dim),
                'key': nn.Linear(self.embed_dim, self.embed_dim),
                'value': nn.Linear(self.embed_dim, self.embed_dim)
            })
        
        # Cross-modal attention layers
        self.cross_attention = nn.MultiheadAttention(
            embed_dim=self.embed_dim,
            num_heads=self.num_heads,
            dropout=config.get('dropout', 0.1),
            batch_first=True
        )
        
        # Modality fusion gates
        self.fusion_gates = nn.ModuleDict()
        for source in ['text', 'vision', 'audio']:
            for target in ['text', 'vision', 'audio']:
                if source != target:
                    self.fusion_gates[f"{source}_to_{target}"] = nn.Sequential(
                        nn.Linear(self.embed_dim * 2, self.embed_dim),
                        nn.Sigmoid()
                    )
    
    def forward(self, modality_embeddings: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
        """Apply cross-modal attention between all modality pairs"""
        
        attended_embeddings = {}
        
        for target_modality, target_embed in modality_embeddings.items():
            # Initialize with original embeddings
            attended_embed = target_embed.clone()
            
            for source_modality, source_embed in modality_embeddings.items():
                if source_modality == target_modality:
                    continue
                
                # Project embeddings
                query = self.modality_projections[target_modality]['query'](target_embed)
                key = self.modality_projections[source_modality]['key'](source_embed)
                value = self.modality_projections[source_modality]['value'](source_embed)
                
                # Cross-modal attention
                attended_values, attention_weights = self.cross_attention(
                    query, key, value
                )
                
                # Fusion gate
                gate_key = f"{source_modality}_to_{target_modality}"
                if gate_key in self.fusion_gates:
                    concat_features = torch.cat([attended_embed, attended_values], dim=-1)
                    gate = self.fusion_gates[gate_key](concat_features)
                    attended_embed = gate * attended_values + (1 - gate) * attended_embed
            
            attended_embeddings[target_modality] = attended_embed
        
        return attended_embeddings

 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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

class AdaptiveModalityWeighting(nn.Module):
    """Dynamically weight modalities based on content and context"""
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__()
        self.config = config
        
        # Content analyzers for each modality
        self.content_analyzers = nn.ModuleDict({
            'text': self._build_text_analyzer(),
            'vision': self._build_vision_analyzer(),
            'audio': self._build_audio_analyzer()
        })
        
        # Context-aware weighting network
        self.weighting_network = nn.Sequential(
            nn.Linear(config['total_features'], 256),
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, len(config['modalities'])),
            nn.Softmax(dim=-1)
        )
    
    def _build_text_analyzer(self) -> nn.Module:
        """Build text content analyzer"""
        return nn.Sequential(
            nn.Linear(768, 256),  # Assuming BERT-like embeddings
            nn.ReLU(),
            nn.Linear(256, 64)
        )
    
    def _build_vision_analyzer(self) -> nn.Module:
        """Build vision content analyzer"""
        return nn.Sequential(
            nn.Linear(2048, 256),  # ResNet features
            nn.ReLU(),
            nn.Linear(256, 64)
        )
    
    def _build_audio_analyzer(self) -> nn.Module:
        """Build audio content analyzer"""
        return nn.Sequential(
            nn.Linear(128, 64),  # Audio features
            nn.ReLU(),
            nn.Linear(64, 32)
        )
    
    def forward(self, 
               modality_embeddings: Dict[str, torch.Tensor],
               context_features: Optional[torch.Tensor] = None) -> torch.Tensor:
        """Compute adaptive weights and fuse modalities"""
        
        # Analyze content quality for each modality
        content_features = []
        available_modalities = []
        
        for modality, embeddings in modality_embeddings.items():
            if modality in self.content_analyzers:
                analyzed = self.content_analyzers[modality](embeddings)
                content_features.append(analyzed)
                available_modalities.append(modality)
        
        if not content_features:
            raise ValueError("No valid modalities for weighting")
        
        # Combine content analysis
        combined_features = torch.cat(content_features, dim=-1)
        
        # Add context features if available
        if context_features is not None:
            combined_features = torch.cat([combined_features, context_features], dim=-1)
        
        # Compute adaptive weights
        modality_weights = self.weighting_network(combined_features)
        
        # Apply weights to embeddings
        weighted_embeddings = []
        for i, modality in enumerate(available_modalities):
            weight = modality_weights[:, i:i+1]
            weighted_embedding = weight * modality_embeddings[modality]
            weighted_embeddings.append(weighted_embedding)
        
        # Sum weighted embeddings
        fused_embedding = torch.sum(torch.stack(weighted_embeddings), dim=0)
        
        return fused_embedding

  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
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111

class MultiModalIntelligentCache:
    """Intelligent caching system for multi-modal processing"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.redis_client = redis.Redis.from_url(config['redis_url'])
        self.cache_stats = CacheStats()
        
        # Content-based cache keys
        self.content_hasher = ContentHasher()
        
        # Cache policies
        self.ttl_policies = {
            ModalityType.TEXT: config['text_ttl'],
            ModalityType.IMAGE: config['image_ttl'],
            ModalityType.AUDIO: config['audio_ttl'],
            ModalityType.DOCUMENT: config['document_ttl']
        }
    
    async def get_processed_input(self, modal_input: ModalInput) -> Optional[ProcessedInput]:
        """Get cached processed input if available"""
        
        # Generate content-based cache key
        cache_key = self._generate_cache_key(modal_input)
        
        try:
            # Check cache
            cached_data = await self.redis_client.get(cache_key)
            if cached_data:
                self.cache_stats.record_hit(modal_input.modality)
                return pickle.loads(cached_data)
            else:
                self.cache_stats.record_miss(modal_input.modality)
                return None
        except Exception as e:
            logger.error(f"Cache retrieval error: {e}")
            return None
    
    async def cache_processed_input(self, 
                                   modal_input: ModalInput,
                                   processed_input: ProcessedInput):
        """Cache processed input with intelligent TTL"""
        
        cache_key = self._generate_cache_key(modal_input)
        
        # Determine TTL based on content analysis
        ttl = self._compute_intelligent_ttl(modal_input, processed_input)
        
        try:
            # Serialize and cache
            serialized_data = pickle.dumps(processed_input)
            await self.redis_client.setex(cache_key, ttl, serialized_data)
            
            self.cache_stats.record_cache_set(modal_input.modality, len(serialized_data))
            
        except Exception as e:
            logger.error(f"Cache storage error: {e}")
    
    def _compute_intelligent_ttl(self, 
                               modal_input: ModalInput,
                               processed_input: ProcessedInput) -> int:
        """Compute TTL based on content characteristics"""
        
        base_ttl = self.ttl_policies[modal_input.modality]
        
        # Adjust based on content complexity
        complexity_score = self._compute_content_complexity(processed_input)
        
        # More complex content gets longer TTL (expensive to recompute)
        complexity_multiplier = 1 + (complexity_score * 2)
        
        # Adjust based on processing cost
        if processed_input.metadata.get('processing_time', 0) > 1.0:
            complexity_multiplier *= 1.5
        
        return int(base_ttl * complexity_multiplier)
    
    def _compute_content_complexity(self, processed_input: ProcessedInput) -> float:
        """Compute content complexity score (0-1)"""
        
        if processed_input.modality == ModalityType.TEXT:
            # Text complexity based on length and vocabulary
            text_length = processed_input.metadata.get('text_length', 0)
            return min(text_length / 10000, 1.0)  # Normalize by max expected length
        
        elif processed_input.modality == ModalityType.IMAGE:
            # Image complexity based on size and analysis results
            image_size = processed_input.metadata.get('image_size', (0, 0))
            pixel_count = image_size[0] * image_size[1]
            analysis = processed_input.metadata.get('analysis', {})
            
            complexity = min(pixel_count / (1920 * 1080), 1.0)  # Normalize by HD resolution
            
            # Increase complexity if OCR or object detection was performed
            if analysis.get('extracted_text') or analysis.get('objects'):
                complexity *= 1.5
            
            return min(complexity, 1.0)
        
        elif processed_input.modality == ModalityType.AUDIO:
            # Audio complexity based on duration and transcription
            duration = processed_input.metadata.get('duration', 0)
            complexity = min(duration / 300, 1.0)  # Normalize by 5 minutes
            
            # Increase if transcription was performed
            if processed_input.metadata.get('transcript'):
                complexity *= 1.3
            
            return min(complexity, 1.0)
        
        return 0.5  # Default complexity

  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
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

class MultiModalBatchProcessor:
    """Optimized batch processing for multi-modal inputs"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        
        # Batch size optimizations per modality
        self.optimal_batch_sizes = {
            ModalityType.TEXT: config['text_batch_size'],
            ModalityType.IMAGE: config['image_batch_size'],
            ModalityType.AUDIO: config['audio_batch_size']
        }
        
        # Processing queues
        self.processing_queues = {
            modality: asyncio.Queue() 
            for modality in ModalityType
        }
        
        # Batch processing tasks
        self.batch_processors = {}
        
    async def start_batch_processors(self):
        """Start background batch processing tasks"""
        
        for modality in ModalityType:
            processor_task = asyncio.create_task(
                self._batch_processor_worker(modality)
            )
            self.batch_processors[modality] = processor_task
    
    async def _batch_processor_worker(self, modality: ModalityType):
        """Background worker for batch processing specific modality"""
        
        queue = self.processing_queues[modality]
        batch_size = self.optimal_batch_sizes[modality]
        batch_timeout = self.config.get('batch_timeout', 0.1)  # 100ms
        
        while True:
            batch = []
            batch_futures = []
            
            try:
                # Collect batch within timeout
                start_time = asyncio.get_event_loop().time()
                
                while (len(batch) < batch_size and 
                       (asyncio.get_event_loop().time() - start_time) < batch_timeout):
                    
                    try:
                        # Try to get item with short timeout
                        item, future = await asyncio.wait_for(
                            queue.get(), 
                            timeout=batch_timeout - (asyncio.get_event_loop().time() - start_time)
                        )
                        batch.append(item)
                        batch_futures.append(future)
                    except asyncio.TimeoutError:
                        break
                
                if batch:
                    # Process batch
                    results = await self._process_modality_batch(modality, batch)
                    
                    # Return results via futures
                    for future, result in zip(batch_futures, results):
                        if not future.cancelled():
                            future.set_result(result)
                
            except Exception as e:
                logger.error(f"Batch processing error for {modality}: {e}")
                
                # Set exceptions on futures
                for future in batch_futures:
                    if not future.cancelled():
                        future.set_exception(e)
            
            # Small delay to prevent busy waiting
            await asyncio.sleep(0.001)
    
    async def process_input_async(self, modal_input: ModalInput) -> ProcessedInput:
        """Add input to batch processing queue and wait for result"""
        
        # Create future for result
        result_future = asyncio.Future()
        
        # Add to appropriate queue
        await self.processing_queues[modal_input.modality].put((modal_input, result_future))
        
        # Wait for result
        return await result_future
    
    async def _process_modality_batch(self, 
                                     modality: ModalityType,
                                     batch: List[ModalInput]) -> List[ProcessedInput]:
        """Process a batch of inputs for specific modality"""
        
        if modality == ModalityType.TEXT:
            return await self._process_text_batch(batch)
        elif modality == ModalityType.IMAGE:
            return await self._process_image_batch(batch)
        elif modality == ModalityType.AUDIO:
            return await self._process_audio_batch(batch)
        else:
            # Fallback to individual processing
            return [await self._process_single_input(inp) for inp in batch]
    
    async def _process_image_batch(self, batch: List[ModalInput]) -> List[ProcessedInput]:
        """Optimized batch processing for images"""
        
        # Collect and preprocess all images
        images = []
        metadata_list = []
        
        for modal_input in batch:
            try:
                # Load and preprocess image
                image = self._load_image(modal_input.data)
                preprocessed = self.vision_transform(image)
                images.append(preprocessed)
                metadata_list.append(modal_input.metadata)
            except Exception as e:
                logger.error(f"Failed to preprocess image: {e}")
                # Add placeholder for failed image
                images.append(torch.zeros(3, 224, 224))
                metadata_list.append({'error': str(e)})
        
        if not images:
            return []
        
        # Batch process through vision encoder
        image_batch = torch.stack(images).to(self.device)
        
        with torch.no_grad():
            batch_features = self.vision_encoder(image_batch)
            batch_features = batch_features.flatten(start_dim=1)
        
        # Create ProcessedInput objects
        results = []
        for i, (features, metadata) in enumerate(zip(batch_features, metadata_list)):
            result = ProcessedInput(
                features=images[i],
                embeddings=features.cpu().unsqueeze(0),
                modality=ModalityType.IMAGE,
                attention_mask=None,
                metadata=metadata
            )
            results.append(result)
        
        return results