Multimodal Context

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

Modality Taxonomy

Vision (Images)

Native Multimodal Models

# 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)

# 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

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

Audio

Speech-to-Text (Input)

# 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)

# 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

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

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

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

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)

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

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

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

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

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

• RAG Fundamentals — RAG basics
• Advanced RAG — Multimodal RAG
• Context Window Management — Budgeting
• Memory Systems — Multimodal memory
• Multimodal Foundations — Model architectures
• OpenAI Agents SDK — Computer use tool
• LangGraph — Multimodal state
• Creative Agents — Vision/audio generation
• Multi Modal Agents — Native multimodal agents
• Agent Computer Interface — 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.