Multimodal Skill

Role

You are an elite multimodal AI engineer. You extract intelligence from images,

audio, video, and documents using the world's best vision and audio models.

You know which model to use for each task, how to batch process at scale,

and how to integrate multimodal outputs into downstream AI pipelines.

Model Selection Guide

Part 1: Image Analysis with Claude Vision

import anthropic
import base64
from pathlib import Path

claude = anthropic.Anthropic()


def encode_image_base64(image_path: str) -> tuple[str, str]:
    """Encode image to base64 and detect media type."""
    path = Path(image_path)
    suffix = path.suffix.lower()
    media_types = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".gif": "image/gif",
        ".webp": "image/webp"
    }
    media_type = media_types.get(suffix, "image/jpeg")

    with open(image_path, "rb") as f:
        data = base64.standard_b64encode(f.read()).decode("utf-8")

    return data, media_type


def analyse_image_claude(
    image_path: str,
    prompt: str,
    model: str = "claude-haiku-4-5-20251001",
    system: str = None
) -> str:
    """Analyse an image using Claude's vision capability."""
    data, media_type = encode_image_base64(image_path)

    messages = [{
        "role": "user",
        "content": [
            {
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": media_type,
                    "data": data
                }
            },
            {
                "type": "text",
                "text": prompt
            }
        ]
    }]

    kwargs = {"model": model, "max_tokens": 2048, "messages": messages}
    if system:
        kwargs["system"] = system

    response = claude.messages.create(**kwargs)
    return response.content[0].text


def analyse_image_from_url(url: str, prompt: str,
                             model: str = "claude-haiku-4-5-20251001") -> str:
    """Analyse an image from a URL (no download needed)."""
    response = claude.messages.create(
        model=model,
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "url", "url": url}},
                {"type": "text", "text": prompt}
            ]
        }]
    )
    return response.content[0].text

Part 2: Chart & Data Extraction from Images

from pydantic import BaseModel, Field
from typing import Optional
import instructor

claude_structured = instructor.from_anthropic(anthropic.Anthropic())


class ChartData(BaseModel):
    """Structured data extracted from a chart image."""
    chart_type: str = Field(..., description="bar, line, pie, scatter, table, etc.")
    title: Optional[str] = None
    x_axis_label: Optional[str] = None
    y_axis_label: Optional[str] = None
    data_series: list[dict] = Field(default_factory=list,
                                     description="List of {label, values} dicts")
    key_insight: str = Field(..., description="One-sentence summary of what the chart shows")
    numeric_values: list[dict] = Field(default_factory=list,
                                        description="All numeric values found")


def extract_chart_data(image_path: str) -> ChartData:
    """
    Extract structured data from a chart or graph image.
    Used for ingesting client-supplied charts into RAG pipelines.
    """
    data, media_type = encode_image_base64(image_path)

    return claude_structured.messages.create(
        model="claude-sonnet-4-5-20250929",  # Use Sonnet for complex charts
        max_tokens=2048,
        response_model=ChartData,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {"type": "base64", "media_type": media_type, "data": data}
                },
                {
                    "type": "text",
                    "text": "Extract all data and information from this chart."
                }
            ]
        }]
    )


def batch_analyse_images(image_paths: list[str],
                           prompt: str,
                           model: str = "claude-haiku-4-5-20251001") -> list[dict]:
    """Analyse multiple images in parallel."""
    import concurrent.futures

    def _analyse_one(path: str) -> dict:
        try:
            result = analyse_image_claude(path, prompt, model)
            return {"path": path, "analysis": result, "success": True}
        except Exception as e:
            return {"path": path, "error": str(e), "success": False}

    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(_analyse_one, image_paths))

    return results

Part 3: Audio Transcription

Cloud Whisper (via OpenAI API)

from openai import OpenAI
from pathlib import Path

oai = OpenAI()

def transcribe_audio(audio_path: str,
                      language: str = "en",
                      response_format: str = "text",
                      prompt: str = None) -> str:
    """
    Transcribe audio using OpenAI Whisper API.
    Supports: mp3, mp4, mpeg, mpga, m4a, wav, webm
    response_format: text, json, srt, vtt, verbose_json
    """
    with open(audio_path, "rb") as audio_file:
        kwargs = {
            "model": "whisper-1",
            "file": audio_file,
            "language": language,
            "response_format": response_format
        }
        if prompt:
            kwargs["prompt"] = prompt  # Context prompt improves accuracy

        transcript = oai.audio.transcriptions.create(**kwargs)

    return transcript if response_format == "text" else transcript.text


def transcribe_with_timestamps(audio_path: str) -> list[dict]:
    """Transcribe with word-level timestamps."""
    with open(audio_path, "rb") as audio_file:
        response = oai.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            response_format="verbose_json",
            timestamp_granularities=["word", "segment"]
        )

    return [
        {
            "text": seg["text"],
            "start": seg["start"],
            "end": seg["end"],
            "confidence": seg.get("avg_logprob", 0)
        }
        for seg in response.segments
    ]

Local Whisper (faster-whisper, fully private)

# pip install faster-whisper
from faster_whisper import WhisperModel
import logging

logging.getLogger("faster_whisper").setLevel(logging.WARNING)

class LocalTranscriber:
    """
    Local audio transcription using faster-whisper.
    4x faster than openai-whisper, supports GPU/CPU.
    Fully private — no API calls.

    Models: tiny, base, small, medium, large-v3
    Device: cpu, cuda, auto
    """
    _model = None

    def __init__(self, model_size: str = "base",
                  device: str = "cpu",
                  compute_type: str = "int8"):
        if not self.__class__._model:
            self.__class__._model = WhisperModel(
                model_size, device=device, compute_type=compute_type
            )
        self.model = self.__class__._model

    def transcribe(self, audio_path: str,
                    language: str = "en",
                    task: str = "transcribe") -> dict:
        """
        Transcribe audio locally.
        task: "transcribe" or "translate" (to English)
        """
        segments, info = self.model.transcribe(
            audio_path,
            language=language,
            task=task,
            beam_size=5,
            word_timestamps=True
        )

        text_parts = []
        segment_data = []

        for seg in segments:
            text_parts.append(seg.text.strip())
            segment_data.append({
                "start": seg.start,
                "end": seg.end,
                "text": seg.text.strip(),
                "words": [{"word": w.word, "start": w.start, "end": w.end}
                          for w in (seg.words or [])]
            })

        return {
            "text": " ".join(text_parts),
            "segments": segment_data,
            "language": info.language,
            "duration": info.duration
        }

Part 4: Video Processing

# pip install opencv-python
import cv2
import os
from pathlib import Path

def extract_video_frames(video_path: str,
                          output_dir: str,
                          fps: int = 1,
                          max_frames: int = 100) -> list[str]:
    """
    Extract frames from video at specified FPS.
    Used for indexing video content in RAG pipelines.
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    cap = cv2.VideoCapture(video_path)
    video_fps = cap.get(cv2.CAP_PROP_FPS)
    frame_interval = int(video_fps / fps)

    frame_paths = []
    frame_count = 0
    extracted = 0

    while cap.isOpened() and extracted < max_frames:
        ret, frame = cap.read()
        if not ret:
            break

        if frame_count % frame_interval == 0:
            timestamp = frame_count / video_fps
            filename = output_path / f"frame_{extracted:04d}_t{timestamp:.1f}s.jpg"
            cv2.imwrite(str(filename), frame)
            frame_paths.append(str(filename))
            extracted += 1

        frame_count += 1

    cap.release()
    return frame_paths


def analyse_video(video_path: str, question: str,
                   llm_fn: callable,
                   frames_per_second: int = 1,
                   max_frames: int = 30) -> dict:
    """
    Answer a question about a video by analysing extracted frames.
    """
    import tempfile

    with tempfile.TemporaryDirectory() as tmp_dir:
        frames = extract_video_frames(video_path, tmp_dir,
                                       fps=frames_per_second,
                                       max_frames=max_frames)

        # Analyse frames in batches
        frame_analyses = batch_analyse_images(
            frames,
            prompt=f"Describe what you see in this frame. Focus on: {question}",
            model="claude-haiku-4-5-20251001"
        )

        successful = [f for f in frame_analyses if f["success"]]

        # Synthesise
        timeline = "\n".join([
            f"Frame {i+1}: {f['analysis'][:200]}"
            for i, f in enumerate(successful)
        ])

        summary = llm_fn(f"""You've been given descriptions of {len(successful)} frames
from a video. Answer this question about the video:

Question: {question}

Frame descriptions (in chronological order):
{timeline}

Answer:""")

    return {
        "answer": summary,
        "frames_analysed": len(successful),
        "video_path": video_path
    }

Part 5: IMI-Specific Multimodal Applications

def extract_data_from_imi_report_image(image_path: str) -> dict:
    """Extract structured data from an IMI report chart or infographic."""
    chart_data = extract_chart_data(image_path)
    return {
        "chart_type": chart_data.chart_type,
        "title": chart_data.title,
        "key_insight": chart_data.key_insight,
        "data": chart_data.data_series,
        "values": chart_data.numeric_values
    }


def transcribe_focus_group(audio_path: str,
                             brand: str = None) -> dict:
    """
    Transcribe a focus group recording for qualitative analysis.
    Uses domain prompt to improve sports/marketing terminology accuracy.
    """
    sports_prompt = (
        "This is a focus group about sports fan engagement and brand loyalty. "
        "Terms include: fan index, tribal fans, sponsor activation, brand equity."
    )
    if brand:
        sports_prompt += f" The brand being discussed is {brand}."

    transcript = transcribe_audio(
        audio_path,
        prompt=sports_prompt,
        response_format="text"
    )

    return {
        "transcript": transcript,
        "word_count": len(transcript.split()),
        "brand": brand
    }

Output Standards

• Always use claude-haiku for simple image tasks (cost efficiency), claude-sonnet for complex chart/data extraction
• Use local faster-whisper for any audio containing client-confidential discussions
• Extract frame timestamps — essential for video content citation
• Batch image analysis with max 5 concurrent API calls
• For IMI: always include sports domain prompt in Whisper calls for better terminology

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)