Multi-Modal Fusion: Combining Text, Image, and Audio with Ollama

# Install Ollama (if not already installed)
curl -fsSL https://ollama.ai/install.sh | sh

# Pull a multi-modal model
ollama pull llava:13b

# Verify installation
ollama list

# requirements.txt
ollama>=0.1.7
pillow>=10.0.0
librosa>=0.10.0
numpy>=1.24.0
requests>=2.31.0

pip install -r requirements.txt

import ollama
from PIL import Image
import base64
import io

class TextImageProcessor:
    def __init__(self, model_name="llava:13b"):
        self.model = model_name
        self.client = ollama.Client()
    
    def encode_image(self, image_path):
        """Convert image to base64 for Ollama processing"""
        with open(image_path, 'rb') as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')
    
    def analyze_image_with_text(self, image_path, text_prompt):
        """Analyze image with accompanying text prompt"""
        image_data = self.encode_image(image_path)
        
        response = self.client.chat(
            model=self.model,
            messages=[
                {
                    'role': 'user',
                    'content': text_prompt,
                    'images': [image_data]
                }
            ]
        )
        
        return response['message']['content']

# Example usage
processor = TextImageProcessor()

# Analyze a product image with description
result = processor.analyze_image_with_text(
    "product_image.jpg",
    "This is a smartphone product listing. What are the key features visible in the image? How does it match the description: 'Premium flagship with triple camera system'?"
)

print(result)

class AdvancedImageTextFusion:
    def __init__(self):
        self.processor = TextImageProcessor()
    
    def compare_images_with_context(self, image1_path, image2_path, context_text):
        """Compare two images within a specific context"""
        
        # Analyze first image
        analysis1 = self.processor.analyze_image_with_text(
            image1_path,
            f"Context: {context_text}\nAnalyze this image and describe key elements."
        )
        
        # Analyze second image
        analysis2 = self.processor.analyze_image_with_text(
            image2_path,
            f"Context: {context_text}\nAnalyze this image and describe key elements."
        )
        
        # Compare both analyses
        comparison = self.processor.analyze_image_with_text(
            image1_path,  # Use first image as reference
            f"""
            Context: {context_text}
            
            First image analysis: {analysis1}
            Second image analysis: {analysis2}
            
            Compare these two analyses and highlight key differences or similarities.
            """
        )
        
        return {
            'image1_analysis': analysis1,
            'image2_analysis': analysis2,
            'comparison': comparison
        }

# Example: Compare before/after renovation photos
fusion = AdvancedImageTextFusion()
result = fusion.compare_images_with_context(
    "kitchen_before.jpg",
    "kitchen_after.jpg",
    "Home renovation project focusing on kitchen modernization"
)

import librosa
import numpy as np
from typing import Dict, List

class AudioProcessor:
    def __init__(self):
        self.sample_rate = 16000
    
    def extract_audio_features(self, audio_path):
        """Extract basic audio features for analysis"""
        # Load audio file
        y, sr = librosa.load(audio_path, sr=self.sample_rate)
        
        # Extract features
        features = {
            'duration': len(y) / sr,
            'tempo': librosa.beat.tempo(y=y, sr=sr)[0],
            'spectral_centroid': np.mean(librosa.feature.spectral_centroid(y=y, sr=sr)),
            'zero_crossing_rate': np.mean(librosa.feature.zero_crossing_rate(y)),
            'mfcc': np.mean(librosa.feature.mfcc(y=y, sr=sr), axis=1)
        }
        
        return features
    
    def audio_to_text_description(self, audio_path):
        """Convert audio characteristics to text description"""
        features = self.extract_audio_features(audio_path)
        
        # Create text description based on audio features
        description = f"""
        Audio Analysis:
        - Duration: {features['duration']:.2f} seconds
        - Tempo: {features['tempo']:.1f} BPM
        - Spectral Centroid: {features['spectral_centroid']:.2f} Hz
        - Zero Crossing Rate: {features['zero_crossing_rate']:.4f}
        - MFCC characteristics: {features['mfcc'][:5]}  # First 5 coefficients
        """
        
        return description.strip()

# Example usage
audio_processor = AudioProcessor()
audio_description = audio_processor.audio_to_text_description("speech_sample.wav")
print(audio_description)

class MultiModalFusion:
    def __init__(self):
        self.text_image_processor = TextImageProcessor()
        self.audio_processor = AudioProcessor()
    
    def process_multimodal_input(self, text_input, image_path, audio_path, task_description):
        """Process text, image, and audio inputs together"""
        
        # Process audio to text description
        audio_description = self.audio_processor.audio_to_text_description(audio_path)
        
        # Combine all text inputs
        combined_text = f"""
        Task: {task_description}
        
        Text Input: {text_input}
        
        Audio Analysis: {audio_description}
        
        Please analyze the provided image in the context of this text and audio information.
        Consider how all three modalities work together to provide a complete understanding.
        """
        
        # Process image with combined text context
        result = self.text_image_processor.analyze_image_with_text(
            image_path,
            combined_text
        )
        
        return {
            'text_input': text_input,
            'audio_analysis': audio_description,
            'multimodal_result': result
        }
    
    def analyze_content_consistency(self, text_input, image_path, audio_path):
        """Check if text, image, and audio are consistent with each other"""
        
        # Get individual analyses
        result = self.process_multimodal_input(
            text_input,
            image_path,
            audio_path,
            "Analyze the consistency between text description, visual content, and audio characteristics"
        )
        
        # Ask for consistency check
        consistency_check = self.text_image_processor.analyze_image_with_text(
            image_path,
            f"""
            Text: {text_input}
            Audio: {result['audio_analysis']}
            
            Rate the consistency between these three inputs (text, image, audio) on a scale of 1-10.
            Explain any discrepancies you notice.
            """
        )
        
        result['consistency_analysis'] = consistency_check
        return result

# Example usage
fusion = MultiModalFusion()

# Analyze a video frame with transcript and audio
result = fusion.analyze_content_consistency(
    text_input="The presenter is explaining machine learning concepts to a classroom of students",
    image_path="classroom_scene.jpg",
    audio_path="presentation_audio.wav"
)

print("Multi-modal analysis result:")
print(result['multimodal_result'])
print("\nConsistency check:")
print(result['consistency_analysis'])

class ContentModerationSystem:
    def __init__(self):
        self.fusion = MultiModalFusion()
    
    def moderate_content(self, post_text, image_path, audio_path=None):
        """Moderate content across multiple modalities"""
        
        moderation_prompt = """
        Analyze this content for potential policy violations:
        - Hate speech or harassment
        - Misinformation or false claims
        - Inappropriate or harmful content
        - Spam or promotional content
        
        Provide a safety score (1-10, where 10 is completely safe) and explain your reasoning.
        """
        
        if audio_path:
            result = self.fusion.process_multimodal_input(
                post_text,
                image_path,
                audio_path,
                moderation_prompt
            )
        else:
            result = self.fusion.text_image_processor.analyze_image_with_text(
                image_path,
                f"Text: {post_text}\n\n{moderation_prompt}"
            )
        
        return result

# Example usage
moderator = ContentModerationSystem()
moderation_result = moderator.moderate_content(
    "Check out this amazing product!",
    "product_ad.jpg",
    "product_audio.wav"
)

class EducationalContentAnalyzer:
    def __init__(self):
        self.fusion = MultiModalFusion()
    
    def analyze_learning_material(self, lesson_text, diagram_path, lecture_audio):
        """Analyze educational content effectiveness"""
        
        analysis_prompt = """
        Evaluate this educational content:
        - Clarity of explanation
        - Visual aid effectiveness
        - Audio quality and delivery
        - Overall learning value
        
        Suggest improvements for better learning outcomes.
        """
        
        result = self.fusion.process_multimodal_input(
            lesson_text,
            diagram_path,
            lecture_audio,
            analysis_prompt
        )
        
        return result

# Example usage
analyzer = EducationalContentAnalyzer()
education_result = analyzer.analyze_learning_material(
    "Introduction to photosynthesis in plants",
    "photosynthesis_diagram.png",
    "teacher_explanation.wav"
)

class OptimizedMultiModalProcessor:
    def __init__(self):
        self.fusion = MultiModalFusion()
        self.cache = {}
    
    def process_with_caching(self, cache_key, text_input, image_path, audio_path, task):
        """Process with result caching for repeated queries"""
        
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        result = self.fusion.process_multimodal_input(
            text_input, image_path, audio_path, task
        )
        
        self.cache[cache_key] = result
        return result
    
    def batch_process(self, inputs_list):
        """Process multiple multimodal inputs efficiently"""
        results = []
        
        for i, inputs in enumerate(inputs_list):
            print(f"Processing batch {i+1}/{len(inputs_list)}")
            
            result = self.fusion.process_multimodal_input(
                inputs['text'],
                inputs['image'],
                inputs['audio'],
                inputs['task']
            )
            
            results.append(result)
        
        return results

# Example batch processing
optimizer = OptimizedMultiModalProcessor()

batch_inputs = [
    {
        'text': 'Product description 1',
        'image': 'product1.jpg',
        'audio': 'review1.wav',
        'task': 'Product analysis'
    },
    {
        'text': 'Product description 2',
        'image': 'product2.jpg',
        'audio': 'review2.wav',
        'task': 'Product analysis'
    }
]

batch_results = optimizer.batch_process(batch_inputs)

class RobustMultiModalProcessor:
    def __init__(self):
        self.fusion = MultiModalFusion()
    
    def validate_inputs(self, text_input, image_path, audio_path):
        """Validate all inputs before processing"""
        errors = []
        
        # Validate text
        if not text_input or len(text_input.strip()) < 10:
            errors.append("Text input too short or empty")
        
        # Validate image
        try:
            with Image.open(image_path) as img:
                if img.width < 224 or img.height < 224:
                    errors.append("Image resolution too low")
        except Exception as e:
            errors.append(f"Image validation failed: {str(e)}")
        
        # Validate audio
        try:
            y, sr = librosa.load(audio_path, duration=1.0)  # Load first second
            if len(y) == 0:
                errors.append("Audio file is empty")
        except Exception as e:
            errors.append(f"Audio validation failed: {str(e)}")
        
        return errors
    
    def safe_process(self, text_input, image_path, audio_path, task):
        """Process with comprehensive error handling"""
        
        # Validate inputs
        errors = self.validate_inputs(text_input, image_path, audio_path)
        if errors:
            return {'error': 'Validation failed', 'details': errors}
        
        try:
            result = self.fusion.process_multimodal_input(
                text_input, image_path, audio_path, task
            )
            return {'success': True, 'result': result}
        
        except Exception as e:
            return {'error': 'Processing failed', 'details': str(e)}

# Example usage with error handling
robust_processor = RobustMultiModalProcessor()
safe_result = robust_processor.safe_process(
    "Sample text",
    "sample_image.jpg",
    "sample_audio.wav",
    "Analysis task"
)

if 'error' in safe_result:
    print(f"Error: {safe_result['error']}")
    print(f"Details: {safe_result['details']}")
else:
    print("Processing successful!")
    print(safe_result['result'])

def troubleshoot_ollama_setup():
    """Diagnose common Ollama setup issues"""
    
    try:
        # Check if Ollama is running
        response = ollama.list()
        print("✓ Ollama is running")
        
        # Check available models
        models = response.get('models', [])
        if not models:
            print("⚠ No models installed")
            print("Run: ollama pull llava:13b")
            return False
        
        # Check for vision models
        vision_models = [m for m in models if 'llava' in m['name'] or 'vision' in m['name']]
        if not vision_models:
            print("⚠ No vision models found")
            print("Run: ollama pull llava:13b")
            return False
        
        print(f"✓ Found {len(vision_models)} vision models")
        return True
        
    except Exception as e:
        print(f"✗ Ollama connection failed: {str(e)}")
        print("Make sure Ollama is installed and running")
        return False

# Run diagnostics
if troubleshoot_ollama_setup():
    print("Setup looks good!")
else:
    print("Please fix the issues above before proceeding")

class MemoryEfficientProcessor:
    def __init__(self):
        self.fusion = MultiModalFusion()
        self.max_image_size = (1024, 1024)
    
    def resize_image_if_needed(self, image_path):
        """Resize large images to prevent memory issues"""
        with Image.open(image_path) as img:
            if img.width > self.max_image_size[0] or img.height > self.max_image_size[1]:
                img.thumbnail(self.max_image_size, Image.Resampling.LANCZOS)
                
                # Save resized image temporarily
                temp_path = f"temp_resized_{image_path}"
                img.save(temp_path)
                return temp_path
        
        return image_path
    
    def process_efficiently(self, text_input, image_path, audio_path, task):
        """Process with memory optimization"""
        
        # Resize image if needed
        processed_image_path = self.resize_image_if_needed(image_path)
        
        try:
            result = self.fusion.process_multimodal_input(
                text_input, processed_image_path, audio_path, task
            )
            return result
        
        finally:
            # Clean up temporary files
            if processed_image_path != image_path:
                import os
                os.remove(processed_image_path)

class CrossModalAnalyzer:
    def __init__(self):
        self.fusion = MultiModalFusion()
    
    def analyze_cross_modal_relationships(self, text_input, image_path, audio_path):
        """Analyze how different modalities relate to each other"""
        
        # Analyze text-image relationships
        text_image_analysis = self.fusion.text_image_processor.analyze_image_with_text(
            image_path,
            f"""
            Text: {text_input}
            
            How does this image relate to the text? What elements in the image correspond to concepts mentioned in the text?
            """
        )
        
        # Analyze audio characteristics
        audio_features = self.fusion.audio_processor.extract_audio_features(audio_path)
        
        # Combine all analyses
        cross_modal_prompt = f"""
        Text: {text_input}
        
        Image Analysis: {text_image_analysis}
        
        Audio Features: Duration {audio_features['duration']:.1f}s, 
        Tempo {audio_features['tempo']:.1f} BPM
        
        Analyze the relationships between these three modalities:
        1. How do they complement each other?
        2. Are there any contradictions?
        3. What information is unique to each modality?
        4. How would removing one modality affect understanding?
        """
        
        cross_modal_result = self.fusion.text_image_processor.analyze_image_with_text(
            image_path,
            cross_modal_prompt
        )
        
        return {
            'text_image_relationship': text_image_analysis,
            'audio_features': audio_features,
            'cross_modal_analysis': cross_modal_result
        }

# Example usage
cross_modal = CrossModalAnalyzer()
relationship_analysis = cross_modal.analyze_cross_modal_relationships(
    "A musician performing an energetic rock song",
    "concert_photo.jpg",
    "rock_performance.wav"
)

import logging
from datetime import datetime
import json

class ProductionMultiModalSystem:
    def __init__(self, config_path="config.json"):
        self.setup_logging()
        self.load_config(config_path)
        self.fusion = MultiModalFusion()
        self.metrics = {
            'processed_items': 0,
            'errors': 0,
            'average_processing_time': 0
        }
    
    def setup_logging(self):
        """Configure logging for production use"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('multimodal_system.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def load_config(self, config_path):
        """Load configuration from JSON file"""
        try:
            with open(config_path, 'r') as f:
                self.config = json.load(f)
        except FileNotFoundError:
            self.logger.warning(f"Config file {config_path} not found, using defaults")
            self.config = {
                'max_image_size': [1024, 1024],
                'max_audio_duration': 300,  # 5 minutes
                'timeout_seconds': 60
            }
    
    def process_request(self, request_data):
        """Process a complete multimodal request"""
        start_time = datetime.now()
        
        try:
            self.logger.info(f"Processing request: {request_data.get('id', 'unknown')}")
            
            # Validate request
            if not self.validate_request(request_data):
                raise ValueError("Invalid request format")
            
            # Process the multimodal input
            result = self.fusion.process_multimodal_input(
                request_data['text'],
                request_data['image_path'],
                request_data['audio_path'],
                request_data['task']
            )
            
            # Update metrics
            processing_time = (datetime.now() - start_time).total_seconds()
            self.update_metrics(processing_time, success=True)
            
            self.logger.info(f"Successfully processed request in {processing_time:.2f}s")
            
            return {
                'success': True,
                'result': result,
                'processing_time': processing_time,
                'timestamp': datetime.now().isoformat()
            }
            
        except Exception as e:
            self.logger.error(f"Processing failed: {str(e)}")
            self.update_metrics(0, success=False)
            
            return {
                'success': False,
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            }
    
    def validate_request(self, request_data):
        """Validate incoming request format"""
        required_fields = ['text', 'image_path', 'audio_path', 'task']
        return all(field in request_data for field in required_fields)
    
    def update_metrics(self, processing_time, success=True):
        """Update system metrics"""
        self.metrics['processed_items'] += 1
        
        if not success:
            self.metrics['errors'] += 1
        else:
            # Update average processing time
            current_avg = self.metrics['average_processing_time']
            total_items = self.metrics['processed_items']
            
            self.metrics['average_processing_time'] = (
                (current_avg * (total_items - 1) + processing_time) / total_items
            )
    
    def get_health_status(self):
        """Return system health metrics"""
        error_rate = self.metrics['errors'] / max(self.metrics['processed_items'], 1)
        
        return {
            'status': 'healthy' if error_rate < 0.1 else 'degraded',
            'metrics': self.metrics,
            'error_rate': error_rate,
            'timestamp': datetime.now().isoformat()
        }

# Example production usage
production_system = ProductionMultiModalSystem()

# Process a request
request = {
    'id': 'req_001',
    'text': 'Product review analysis',
    'image_path': 'product_image.jpg',
    'audio_path': 'customer_review.wav',
    'task': 'Analyze customer sentiment across text, image, and audio'
}

result = production_system.process_request(request)
health = production_system.get_health_status()

print(f"Processing result: {result['success']}")
print(f"System health: {health['status']}")