DEV Community

shashank ms
shashank ms

Posted on

Monitoring Multimodal Reasoning Performance: Metrics, Benchmarks, and Tools

I built a lightweight monitor that evaluates vision-language reasoning on chart images and tracks accuracy and latency over time. It runs against Oxlo.ai's API, where request-based pricing keeps batch evaluation costs flat even when prompts include long document context. If you are shipping a multimodal agent, this gives you a reproducible baseline you can run in CI.

What you'll need

  • Python 3.10 or newer
  • An Oxlo.ai API key from https://portal.oxlo.ai
  • The OpenAI SDK: pip install openai
  • Two sample chart images saved as PNG files in a ./charts directory

Step 1: Initialize the Oxlo.ai client

First, I import the dependencies and configure the OpenAI SDK to point at Oxlo.ai. I also set up a local SQLite database and a directory for the chart images.

import os
import json
import time
import sqlite3
import base64
from pathlib import Path
from openai import OpenAI

client = OpenAI(
    base_url="https://api.oxlo.ai/v1",
    api_key=os.environ.get("OXLO_API_KEY", "YOUR_OXLO_API_KEY")
)

DB_PATH = "reasoning_metrics.db"
CHARTS_DIR = Path("charts")
CHARTS_DIR.mkdir(exist_ok=True)

Step 2: Build a local benchmark dataset

Next, I define a small benchmark dataset as JSON. Each entry links to a chart image, asks a concrete question, and provides the ground-truth answer I will score against.

BENCHMARK = [
    {
        "image": "charts/revenue_q1.png",
        "question": "What was the total revenue in Q1?",
        "answer": "$12.4M"
    },
    {
        "image": "charts/users_growth.png",
        "question": "Which month had the highest user growth percentage?",
        "answer": "March"
    }
]

with open("benchmark.json", "w") as f:
    json.dump(BENCHMARK, f, indent=2)

Step 3: Lock in the system prompt

The system prompt forces structured JSON output so I can parse reasoning and final answers reliably without brittle regex.

SYSTEM_PROMPT = """You are a multimodal reasoning evaluator. You will receive a chart image and a question. Think step by step, then provide your final answer as a single JSON object containing exactly two keys: reasoning and answer. Output only valid JSON with no markdown formatting."""

Step 4: Run vision inference

This helper encodes a local image to base64 and sends it to kimi-k2.6 on Oxlo.ai, which handles both vision and reasoning in one call.

def encode_image(image_path):
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

def run_inference(image_path, question):
    b64 = encode_image(image_path)
    data_url = f"data:image/png;base64,{b64}"

    response = client.chat.completions.create(
        model="kimi-k2.6",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": question},
                    {"type": "image_url", "image_url": {"url": data_url}}
                ]
            },
        ],
        max_tokens=1024,
    )

    raw = response.choices[0].message.content.strip()
    raw = raw.removeprefix("

```json").removeprefix("```

").removesuffix("

```

").strip()
    return json.loads(raw)

Step 5: Build the evaluator and metrics logger

I store every run in SQLite. For scoring, I use exact match when possible, and fall back to an LLM judge via llama-3.3-70b on Oxlo.ai when the wording differs.

def init_db():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute("""
        CREATE TABLE IF NOT EXISTS evaluations (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            image TEXT,
            question TEXT,
            expected TEXT,
            prediction TEXT,
            score REAL,
            latency_ms INTEGER,
            model TEXT,
            timestamp REAL
        )
    """)
    conn.commit()
    conn.close()

def judge_correctness(expected, predicted):
    prompt = f"Ground truth answer: {expected}\nModel answer: {predicted}\nRate semantic correctness from 0.0 to 1.0. Reply with only the number."
    resp = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=10,
    )
    try:
        return float(resp.choices[0].message.content.strip())
    except ValueError:
        return 0.0

def evaluate_sample(image_path, question, expected):
    start = time.time()
    prediction_obj = run_inference(image_path, question)
    latency = int((time.time() - start) * 1000)

    predicted = str(prediction_obj.get("answer", ""))
    if predicted.lower().strip() == expected.lower().strip():
        score = 1.0
    else:
        score = judge_correctness(expected, predicted)

    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute(
        "INSERT INTO evaluations (image, question, expected, prediction, score, latency_ms, model, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        (str(image_path), question, expected, predicted, score, latency, "kimi-k2.6", time.time())
    )
    conn.commit()
    conn.close()
    return prediction_obj, score, latency

Step 6: Batch process the benchmark

The benchmark loop loads the dataset, skips missing images, and records latency and score for each sample.

def run_benchmark():
    init_db()
    with open("benchmark.json") as f:
        dataset = json.load(f)

    for item in dataset:
        img = Path(item["image"])
        if not img.exists():
            print(f"Skipping missing image: {img}")
            continue
        pred, score, latency = evaluate_sample(img, item["question"], item["answer"])
        print(f"Q: {item['question']}")
        print(f"A: {pred.get('answer')} | Score: {score} | Latency: {latency}ms")
        print("-" * 40)

Step 7: Generate the performance report

Finally, I query the database to print aggregate stats and a short trace of recent evaluations.

def print_report():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute("""
        SELECT
            AVG(score) as avg_score,
            AVG(latency_ms) as avg_latency,
            MIN(score) as min_score,
            COUNT(*) as total
        FROM evaluations
    """)
    row = c.fetchone()

    print("\n=== Multimodal Reasoning Monitor Report ===")
    print(f"Total evaluations: {row[3]}")
    print(f"Average score: {row[0]:.2f}")
    print(f"Min score: {row[2]:.2f}")
    print(f"Average latency: {row[1]:.0f}ms")

    c.execute("""
        SELECT question, prediction, score, latency_ms
        FROM evaluations
        ORDER BY timestamp DESC
        LIMIT 5
    """)
    print("\nRecent runs:")
    for q, p, s, l in c.fetchall():
        print(f"  Score {s} | {l}ms | Q: {q[:50]}...")
    conn.close()

if __name__ == "__main__":
    run_benchmark()
    print_report()

Run it

Save everything into a single file named monitor.py, place your PNG charts in ./charts, and run the script.

export OXLO_API_KEY="sk-oxlo.ai-..."
python monitor.py

Example output after two evaluations:

Q: What was the total revenue in Q1?
A: $12.4M | Score: 1.0 | Latency: 340ms
----------------------------------------
Q: Which month had the highest user growth percentage?
A: March | Score: 1.0 | Latency: 298ms
----------------------------------------

=== Multimodal Reasoning Monitor Report ===
Total evaluations: 2
Average score: 1.00
Min score: 1.00
Average latency: 319ms

Recent runs:
  Score 1.0 | 298ms | Q: Which month had the highest user growth percentag...
  Score 1.0 | 340ms | Q: What was the total revenue in Q1?...

Wrap-up and next steps

From here, you can wire this script into a nightly GitHub Action and fail the build if the average score drops below 0.95. You can also expand the benchmark to long-context document understanding, which stays economical on Oxlo.ai because the flat per-request pricing does not scale with input length. See https://oxlo.ai/pricing for plan details.

Top comments (0)