# Multimodal Context

Handling vision, audio, video, and structured data in agent context.

---

## Modality Taxonomy

| Modality | Input Format | Token Cost | Processing |
|---|---|---|---|
| **Text** | Tokens | 1x | Native |
| **Images** | Patches → tokens | ~256-1000/img | Vision encoder |
| **Audio** | Spectrogram → tokens | ~100-500/sec | Audio encoder |
| **Video** | Frames + audio | High | Frame sampling |
| **Structured** | JSON/CSV/Table → text | Variable | Serialization |
| **Code** | AST + tokens | 1x | Native |

---

## Vision (Images)

### Native Multimodal Models
```python
# GPT-4o, Claude 3.5, Gemini 1.5, Qwen-VL
from openai import AsyncOpenAI

client = AsyncOpenAI()

response = await client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": [
            {"type": "text", "text": "Analyze this architecture diagram"},
            {"type": "image_url", "image_url": {"url": "data:image/png;base64,..."}}
        ]}
    ]
)
```

### Vision Encoder + LLM (Legacy)
```python
# CLIP/SigLIP → Projector → LLM
from llama_index.multi_modal_llms import OpenAIMultiModal
from llama_index.core import MultiModalVectorStoreIndex

llm = OpenAIMultiModal(model="gpt-4o", max_new_tokens=4096)
index = MultiModalVectorStoreIndex.from_documents(documents)
```

### Image Processing Pipeline
```python
class ImageProcessor:
    def __init__(self):
        self.max_size = (1024, 1024)
        self.format = "PNG"
        self.quality = 85
    
    def prepare(self, image_path: str) -> dict:
        # 1. Load & resize
        img = Image.open(image_path)
        img.thumbnail(self.max_size)
        
        # 2. Optimize
        buffer = BytesIO()
        img.save(buffer, format=self.format, quality=self.quality)
        b64 = base64.b64encode(buffer.getvalue()).decode()
        
        # 3. Estimate tokens
        tokens = self.estimate_tokens(img.size)
        
        return {
            "type": "image_url",
            "image_url": {"url": f"data:image/png;base64,{b64}"},
            "tokens": tokens
        }
    
    def estimate_tokens(self, size: tuple) -> int:
        # GPT-4o: ~170 tokens base + 170 per 512x512 tile
        w, h = size
        tiles = max(1, (w * h) / (512 * 512))
        return int(170 + 170 * tiles)
```

### Vision Use Cases for Agents
| Use Case | Implementation |
|---|---|
| **Diagram analysis** | Architecture, flowcharts, ERDs |
| **UI inspection** | Screenshots → code, bugs |
| **Document parsing** | PDFs, scanned docs → structured data |
| **Code screenshots** | Image → code |
| **Monitoring dashboards** | Grafana → incident analysis |
| **Visual regression** | Before/after comparison |

---

## Audio

### Speech-to-Text (Input)
```python
# OpenAI Whisper, Deepgram, AssemblyAI
from openai import AsyncOpenAI

client = AsyncOpenAI()

transcript = await client.audio.transcriptions.create(
    model="whisper-1",
    file=audio_file,
    response_format="verbose_json",
    timestamp_granularities=["word", "segment"]
)
```

### Text-to-Speech (Output)
```python
# OpenAI TTS, ElevenLabs, PlayHT
speech = await client.audio.speech.create(
    model="tts-1-hd",
    voice="nova",
    input="Hello! I've completed the task.",
    response_format="mp3"
)
```

### Audio in Agent Context
```python
class AudioContext:
    def __init__(self, max_duration_seconds=60):
        self.max_duration = max_duration_seconds
    
    async def process_input(self, audio_file) -> AudioContextItem:
        # 1. Transcribe
        transcript = await self.transcribe(audio_file)
        
        # 2. Extract features
        features = await self.extract_features(audio_file)
        
        # 3. Create context item
        return AudioContextItem(
            transcript=transcript,
            duration=features.duration,
            language=features.language,
            speakers=features.speakers,
            tokens=estimate_audio_tokens(transcript)
        )
    
    async def generate_output(self, text: str) -> bytes:
        return await self.tts(text)
```

---

## Video

### Frame Sampling Strategies
```python
class VideoSampler:
    def __init__(self):
        self.strategies = {
            "uniform": self.uniform_sample,
            "keyframe": self.keyframe_sample,
            "scene": self.scene_change_sample,
            "adaptive": self.adaptive_sample
        }
    
    def uniform_sample(self, video_path, max_frames=16):
        # Evenly spaced frames
        cap = cv2.VideoCapture(video_path)
        total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        indices = np.linspace(0, total-1, max_frames, dtype=int)
        return [self.extract_frame(cap, i) for i in indices]
    
    def keyframe_sample(self, video_path, max_frames=16):
        # I-frames only (compression keyframes)
        cap = cv2.VideoCapture(video_path)
        keyframes = []
        while len(keyframes) < max_frames:
            ret, frame = cap.read()
            if not ret: break
            if self.is_keyframe(cap):
                keyframes.append(frame)
        return keyframes
    
    def adaptive_sample(self, video_path, max_frames=16):
        # Scene detection + uniform within scenes
        scenes = self.detect_scenes(video_path)
        frames = []
        for scene in scenes:
            frames.extend(self.uniform_sample(scene, max_frames // len(scenes)))
        return frames[:max_frames]
```

### Video Understanding
```python
class VideoUnderstanding:
    def __init__(self, model="gpt-4o"):
        self.model = model
    
    async def analyze(self, video_path: str, query: str) -> str:
        frames = self.sampler.adaptive_sample(video_path, max_frames=16)
        
        # Prepare multimodal message
        content = [{"type": "text", "text": query}]
        for frame in frames:
            b64 = self.encode_frame(frame)
            content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}})
        
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": content}],
            max_tokens=2048
        )
        return response.choices[0].message.content
```

---

## Structured Data

### Serialization Strategies
```python
class StructuredDataSerializer:
    def serialize(self, data: Any, format: str = "json") -> str:
        if format == "json":
            return json.dumps(data, indent=2, default=str)
        elif format == "yaml":
            return yaml.dump(data, sort_keys=False)
        elif format == "csv":
            return self.to_csv(data)
        elif format == "markdown":
            return self.to_markdown(data)
        elif format == "sql":
            return self.to_sql_insert(data)
        elif format == "natural":
            return self.to_natural_language(data)
    
    def to_natural_language(self, data: dict) -> str:
        # LLM converts structured data to natural language
        prompt = f"Describe this data naturally:\n{json.dumps(data, indent=2)}"
        return llm.complete(prompt)
```

### Tabular Data (CSV, Excel, SQL)
```python
class TableProcessor:
    def process(self, table: pd.DataFrame, query: str) -> str:
        # 1. Schema summary
        schema = f"Columns: {list(table.columns)}\nDtypes: {table.dtypes.to_dict()}"
        
        # 2. Statistics
        stats = table.describe(include="all").to_string()
        
        # 3. Sample rows
        sample = table.head(5).to_markdown()
        
        # 4. Query-specific extraction
        relevant = self.extract_relevant(table, query)
        
        return f"""Table Context:
{schema}

Statistics:
{stats}

Sample:
{sample}

Relevant Data:
{relevant}"""
```

### Code as Structured Data
```python
class CodeProcessor:
    def process(self, code: str, language: str) -> CodeContext:
        tree = self.parse_ast(code, language)
        
        return CodeContext(
            language=language,
            imports=self.extract_imports(tree),
            functions=self.extract_functions(tree),
            classes=self.extract_classes(tree),
            complexity=self.cyclomatic_complexity(tree),
            dependencies=self.extract_dependencies(tree),
            tokens=estimate_tokens(code)
        )
```

---

## Multimodal Context Budgeting

### Token Estimation
```python
class MultimodalTokenEstimator:
    def estimate(self, item: ContextItem) -> int:
        if item.type == "text":
            return count_tokens(item.content)
        elif item.type == "image":
            return self.estimate_image_tokens(item.dimensions)
        elif item.type == "audio":
            return self.estimate_audio_tokens(item.duration)
        elif item.type == "video":
            return self.estimate_video_tokens(item.duration, item.sample_rate)
        elif item.type == "structured":
            return count_tokens(json.dumps(item.data))
        return 0
    
    def estimate_image_tokens(self, dims: tuple) -> int:
        w, h = dims
        tiles = max(1, (w * h) / (512 * 512))
        return 170 + int(170 * tiles)  # GPT-4o pricing
    
    def estimate_video_tokens(self, duration: float, fps: float) -> int:
        frames = duration * fps
        return int(frames * 170)  # Per frame
```

### Budget Allocation
```python
class MultimodalBudget:
    def __init__(self, max_tokens: int):
        self.max = max_tokens
        self.allocation = {
            "text": 0.40,
            "images": 0.30,
            "audio": 0.10,
            "video": 0.15,
            "structured": 0.05
        }
    
    def allocate(self, context: MultimodalContext) -> dict[str, int]:
        # Dynamic: prioritize based on task
        if context.task.requires_vision:
            self.allocation["images"] = 0.50
        if context.task.requires_audio:
            self.allocation["audio"] = 0.20
        
        return {k: int(self.max * v) for k, v in self.allocation.items()}
```

---

## Multimodal RAG

### Vision + Text Retrieval
```python
class MultimodalRAG:
    def __init__(self):
        self.text_index = VectorStoreIndex(...)
        self.image_index = MultiModalVectorStoreIndex(...)
    
    async def retrieve(self, query: str, image: Image = None) -> list:
        results = []
        
        # Text retrieval
        text_results = await self.text_index.query(query)
        results.extend(text_results)
        
        # Image retrieval (if query image provided)
        if image:
            image_results = await self.image_index.query(image)
            results.extend(image_results)
        
        # Cross-modal: text query → image results
        cross_results = await self.image_index.query(query)
        results.extend(cross_results)
        
        return self.rerank(results)
```

---

## Related Notes

- [[ContextEngineering/RAG-Fundamentals.md]] — RAG basics
- [[ContextEngineering/Advanced-RAG.md]] — Multimodal RAG
- [[ContextEngineering/Context-Window-Management.md]] — Budgeting
- [[ContextEngineering/Memory-Systems.md]] — Multimodal memory
- [[ModelArchitectures/Multimodal-Foundations.md]] — Model architectures
- [[Frameworks/OpenAI-Agents-SDK.md]] — Computer use tool
- [[Frameworks/LangGraph.md]] — Multimodal state
- [[Domains/Creative-Agents.md]] — Vision/audio generation
- [[Frontiers/Multi-Modal-Agents.md]] — Native multimodal agents
- [[Frontiers/Agent-Computer-Interface.md]] — Desktop vision

---

*Sources: GPT-4o/4V system cards, Gemini 1.5 technical report, Qwen-VL papers, LLaVA, MultiModalVectorStoreIndex, CLIP/SigLIP, Whisper, OpenAI TTS, video understanding papers (Video-LLaMA, Valley), multimodal RAG surveys.*