Multimodal AI Systems in Production: Building with GPT-5, Vision, and Audio in 2026

from openai import OpenAI
import base64
from pathlib import Path

client = OpenAI()

class SequentialMultimodalPipeline:
    """
    Sequential pipeline for processing multiple modalities.
    Best for: Batch processing, non-real-time applications
    """

    def __init__(self):
        self.client = client

    def process_image_with_context(self, image_path, text_question, audio_context=None):
        """
        Process image with textual question and optional audio context.

        Args:
            image_path: Path to image file
            text_question: Question about the image
            audio_context: Optional audio file for additional context

        Returns:
            Comprehensive response combining all modalities
        """

        # Step 1: Encode image
        with open(image_path, "rb") as image_file:
            base64_image = base64.b64encode(image_file.read()).decode('utf-8')

        # Step 2: If audio provided, transcribe it first
        audio_transcript = ""
        if audio_context:
            with open(audio_context, "rb") as audio_file:
                transcript_response = self.client.audio.transcriptions.create(
                    model="whisper-1",
                    file=audio_file
                )
                audio_transcript = transcript_response.text

        # Step 3: Combine all modalities in GPT-5 vision request
        messages = [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": f"""Analyze this image and answer the question.

                        Question: {text_question}

                        {f"Additional audio context: {audio_transcript}" if audio_transcript else ""}

                        Provide a detailed analysis considering all provided information.
                        """
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}",
                            "detail": "high"  # Use "high" for detailed analysis
                        }
                    }
                ]
            }
        ]

        response = self.client.chat.completions.create(
            model="gpt-4o",  # Use "gpt-5" when available for best results
            messages=messages,
            max_tokens=1000,
            temperature=0.7
        )

        return response.choices[0].message.content

# Example usage
if __name__ == "__main__":
    pipeline = SequentialMultimodalPipeline()

    result = pipeline.process_image_with_context(
        image_path="medical_chart.jpg",
        text_question="What trends do you see in this patient's vital signs?",
        audio_context="doctor_notes.mp3"  # Optional
    )

    print(f"Analysis:\\n{result}")

class UnifiedMultimodalSystem:
    """
    Unified system processing all modalities in a single request.
    Best for: Real-time applications, interactive experiences
    """

    def __init__(self):
        self.client = OpenAI()

    def analyze_multimodal_content(
        self,
        images: list[str],
        text: str,
        audio_path: str = None,
        generate_audio_response: bool = False
    ):
        """
        Unified multimodal analysis with optional audio output.

        Args:
            images: List of image paths
            text: Text query or context
            audio_path: Optional audio file path
            generate_audio_response: Whether to generate audio response

        Returns:
            Dict with text response and optional audio
        """

        # Prepare multimodal message
        content_parts = [{"type": "text", "text": text}]

        # Add images
        for img_path in images:
            with open(img_path, "rb") as img_file:
                base64_img = base64.b64encode(img_file.read()).decode('utf-8')
                content_parts.append({
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/jpeg;base64,{base64_img}",
                        "detail": "high"
                    }
                })

        # Transcribe audio if provided
        if audio_path:
            with open(audio_path, "rb") as audio:
                transcript = self.client.audio.transcriptions.create(
                    model="whisper-1",
                    file=audio,
                    language="en"
                )
                content_parts[0]["text"] += f"\\n\\nAudio context: {transcript.text}"

        # Get GPT-5 response
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": content_parts}],
            max_tokens=1500
        )

        text_response = response.choices[0].message.content

        result = {"text": text_response}

        # Generate audio response if requested
        if generate_audio_response:
            speech_response = self.client.audio.speech.create(
                model="tts-1-hd",
                voice="nova",
                input=text_response,
                speed=1.0
            )

            audio_file = Path("response_audio.mp3")
            speech_response.stream_to_file(audio_file)
            result["audio_file"] = str(audio_file)

        return result

# Example: Interactive customer support
if __name__ == "__main__":
    system = UnifiedMultimodalSystem()

    result = system.analyze_multimodal_content(
        images=["product_issue_photo1.jpg", "product_issue_photo2.jpg"],
        text="I'm having trouble with my device. Can you help diagnose the issue?",
        audio_path="customer_description.mp3",
        generate_audio_response=True
    )

    print(f"Response: {result['text']}")
    print(f"Audio saved: {result.get('audio_file')}")

class StreamingMultimodalSystem:
    """
    Streaming system for real-time multimodal AI applications.
    Best for: Voice assistants, live customer support, interactive tutoring
    """

    def __init__(self):
        self.client = OpenAI()

    def stream_multimodal_analysis(self, image_path, initial_text):
        """
        Stream responses for real-time interaction.

        Args:
            image_path: Image to analyze
            initial_text: Initial query

        Yields:
            Response chunks as they're generated
        """

        # Encode image
        with open(image_path, "rb") as img:
            base64_img = base64.b64encode(img.read()).decode('utf-8')

        # Create streaming request
        stream = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": initial_text},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_img}"
                        }
                    }
                ]
            }],
            max_tokens=1000,
            stream=True  # Enable streaming
        )

        full_response = ""

        for chunk in stream:
            if chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                full_response += content
                yield content

        return full_response

# Example: Live presentation analysis
if __name__ == "__main__":
    system = StreamingMultimodalSystem()

    print("Analyzing presentation slide...")

    for chunk in system.stream_multimodal_analysis(
        image_path="presentation_slide.jpg",
        initial_text="Analyze this slide and suggest improvements for clarity."
    ):
        print(chunk, end="", flush=True)

    print("\\n\\nAnalysis complete!")

class MedicalDiagnosticAssistant:
    """
    Multimodal AI for medical diagnostics.
    Combines patient images, voice notes, and electronic health records.
    """

    def __init__(self):
        self.client = OpenAI()
        self.conversation_history = []

    def analyze_patient_case(
        self,
        medical_images: list[str],
        doctor_notes_audio: str,
        patient_history: dict,
        specific_question: str = None
    ):
        """
        Comprehensive patient case analysis.

        Note: This is for educational purposes. Real medical applications
        require regulatory approval and proper medical oversight.
        """

        # Transcribe doctor's audio notes
        with open(doctor_notes_audio, "rb") as audio:
            notes = self.client.audio.transcriptions.create(
                model="whisper-1",
                file=audio
            )

        # Prepare context
        context = f"""Medical Case Analysis

        Patient History:
        - Age: {patient_history.get('age')}
        - Conditions: {', '.join(patient_history.get('conditions', []))}
        - Medications: {', '.join(patient_history.get('medications', []))}

        Doctor's Notes: {notes.text}

        {f"Specific Question: {specific_question}" if specific_question else ""}

        Please analyze the provided medical images in the context of this
        patient history and provide insights.
        """

        # Prepare multimodal content
        content = [{"type": "text", "text": context}]

        for img_path in medical_images:
            with open(img_path, "rb") as img:
                b64_img = base64.b64encode(img.read()).decode('utf-8')
                content.append({
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{b64_img}"}
                })

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": "You are a medical AI assistant. Provide detailed analysis but always recommend professional medical consultation."
                },
                {"role": "user", "content": content}
            ],
            temperature=0.3  # Lower temperature for medical accuracy
        )

        return response.choices[0].message.content

# Example usage
if __name__ == "__main__":
    assistant = MedicalDiagnosticAssistant()

    analysis = assistant.analyze_patient_case(
        medical_images=["xray_chest.jpg", "lab_results.jpg"],
        doctor_notes_audio="examination_notes.mp3",
        patient_history={
            "age": 45,
            "conditions": ["hypertension", "type 2 diabetes"],
            "medications": ["metformin", "lisinopril"]
        },
        specific_question="Are there any concerning patterns in these results?"
    )

    print(f"Medical Analysis:\\n{analysis}")

class ContentCreationStudio:
    """
    Multimodal AI for automated content creation.
    Generates video scripts, voiceovers, and visual descriptions.
    """

    def __init__(self):
        self.client = OpenAI()

    def create_video_content(
        self,
        topic: str,
        reference_images: list[str],
        style: str = "professional",
        duration_seconds: int = 60
    ):
        """
        Generate complete video content package.

        Returns:
            Dict with script, voiceover, and visual suggestions
        """

        # Analyze reference images
        image_content = []
        for img_path in reference_images:
            with open(img_path, "rb") as img:
                b64 = base64.b64encode(img.read()).decode('utf-8')
                image_content.append({
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{b64}"}
                })

        # Generate script based on images and topic
        script_prompt = {
            "type": "text",
            "text": f"""Create a {duration_seconds}-second video script about: {topic}

            Style: {style}

            Based on the reference images provided, create:
            1. Engaging narrative script
            2. Scene descriptions
            3. Visual elements to include
            4. Timing breakdown

            Format as a production-ready script.
            """
        }

        script_response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [script_prompt] + image_content
            }],
            temperature=0.8
        )

        script = script_response.choices[0].message.content

        # Generate voiceover audio
        speech = self.client.audio.speech.create(
            model="tts-1-hd",
            voice="onyx",  # Professional male voice
            input=script,
            speed=0.95
        )

        voiceover_path = Path(f"voiceover_{topic.replace(' ', '_')}.mp3")
        speech.stream_to_file(voiceover_path)

        return {
            "script": script,
            "voiceover": str(voiceover_path),
            "duration": duration_seconds
        }

# Example: Marketing video creation
if __name__ == "__main__":
    studio = ContentCreationStudio()

    content = studio.create_video_content(
        topic="AI-powered productivity tools",
        reference_images=["product_screenshot1.jpg", "product_screenshot2.jpg"],
        style="engaging and professional",
        duration_seconds=90
    )

    print(f"Script:\\n{content['script']}\\n")
    print(f"Voiceover: {content['voiceover']}")

import hashlib
import json
from functools import lru_cache

class OptimizedMultimodalSystem:
    """
    Production-optimized multimodal system with caching.
    """

    def __init__(self):
        self.client = OpenAI()
        self.image_cache = {}

    def _get_image_hash(self, image_path):
        """Generate hash for image caching."""
        with open(image_path, "rb") as f:
            return hashlib.md5(f.read()).hexdigest()

    @lru_cache(maxsize=128)
    def _cached_vision_analysis(self, image_hash, prompt):
        """Cache vision analysis results."""
        # Actual API call happens here
        return self._analyze_image_internal(image_hash, prompt)

    def analyze_with_cache(self, image_path, prompt):
        """
        Analyze image with intelligent caching.

        For repeated analyses of same image, returns cached result.
        """
        img_hash = self._get_image_hash(image_path)

        # Check cache first
        cache_key = f"{img_hash}:{prompt}"
        if cache_key in self.image_cache:
            return self.image_cache[cache_key]

        # Process if not cached
        with open(image_path, "rb") as img:
            b64_img = base64.b64encode(img.read()).decode('utf-8')

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/jpeg;base64,{b64_img}"}
                    }
                ]
            }]
        )

        result = response.choices[0].message.content
        self.image_cache[cache_key] = result
        return result

class BatchMultimodalProcessor:
    """
    Batch processor for cost and performance optimization.
    """

    def __init__(self, batch_size=10):
        self.client = OpenAI()
        self.batch_size = batch_size

    def process_images_batch(self, image_prompts: list[dict]):
        """
        Process multiple images efficiently.

        Args:
            image_prompts: List of {"image": path, "prompt": text} dicts

        Returns:
            List of analysis results
        """
        results = []

        # Process in batches to avoid rate limits
        for i in range(0, len(image_prompts), self.batch_size):
            batch = image_prompts[i:i + self.batch_size]

            for item in batch:
                with open(item["image"], "rb") as img:
                    b64_img = base64.b64encode(img.read()).decode('utf-8')

                response = self.client.chat.completions.create(
                    model="gpt-4o",
                    messages=[{
                        "role": "user",
                        "content": [
                            {"type": "text", "text": item["prompt"]},
                            {
                                "type": "image_url",
                                "image_url": {
                                    "url": f"data:image/jpeg;base64,{b64_img}"
                                }
                            }
                        ]
                    }]
                )

                results.append({
                    "image": item["image"],
                    "analysis": response.choices[0].message.content
                })

            # Brief pause between batches to respect rate limits
            if i + self.batch_size < len(image_prompts):
                import time
                time.sleep(1)

        return results

import logging
from tenacity import retry, stop_after_attempt, wait_exponential

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ProductionMultimodalSystem:
    """
    Production-ready multimodal system with enterprise features.
    """

    def __init__(self):
        self.client = OpenAI()

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def analyze_with_retry(self, image_path, prompt):
        """
        Analyze with automatic retry logic.
        """
        try:
            logger.info(f"Analyzing image: {image_path}")

            with open(image_path, "rb") as img:
                b64_img = base64.b64encode(img.read()).decode('utf-8')

            response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": prompt},
                        {
                            "type": "image_url",
                            "image_url": {"url": f"data:image/jpeg;base64,{b64_img}"}
                        }
                    ]
                }],
                timeout=30  # 30 second timeout
            )

            logger.info("Analysis successful")
            return response.choices[0].message.content

        except Exception as e:
            logger.error(f"Analysis failed: {str(e)}")
            raise