We are building a multimodal incident diagnostician that consumes a Grafana screenshot and a raw log tail to produce a structured JSON root-cause analysis. It helps on-call engineers who are tired of context switching between dashboards and terminals at 3 a.m. The whole pipeline runs against Oxlo.ai's vision-capable models with flat per-request pricing, so adding a high-resolution image or extra log lines does not change the cost.
What you'll need
- An Oxlo.ai API key from https://portal.oxlo.ai
- Python 3.10 or newer
- The OpenAI SDK:
pip install openai
Step 1: Encode the screenshot
Vision models need images inlined as base64 data URLs. I wrote a small helper that detects the file extension and returns the proper RFC 2397 string. This keeps everything self-contained and avoids hosting images on a public CDN.
import base64
from pathlib import Path
def encode_image(path: str) -> str:
ext = Path(path).suffix.lstrip(".")
if ext == "jpg":
ext = "jpeg"
with open(path, "rb") as f:
b64 = base64.b64encode(f.read()).decode("utf-8")
return f"data:image/{ext};base64,{b64}"
if __name__ == "__main__":
image_b64 = encode_image("dashboard.png")
print(f"Encoded {len(image_b64)} characters")Step 2: Lock down the system prompt
Hallucination is the biggest risk when a model interprets a chart. I lock the model into a rigid inspection protocol so it states "unreadable" instead of guessing. This prompt is the entire contract for the agent.
SYSTEM_PROMPT = """You are an on-call site reliability engineer. You have two inputs:
1. A screenshot of a Grafana dashboard panel.
2. A short tail of raw application logs.
Follow this protocol exactly:
- First, list every visible metric name and its approximate value in the screenshot.
- Second, correlate any spikes or anomalies with timestamped ERROR or FATAL lines in the logs.
- Third, emit a single JSON object with keys: "anomaly_seen" (bool), "metric" (str), "log_signature" (str), "root_cause" (str), "remediation" (str).
If a metric is unreadable, state "unreadable" rather than guessing."""Step 3: Fire the multimodal request to Oxlo.ai
Kimi K2.6 handles vision, advanced reasoning, and a 131K context window, so we can stuff a large screenshot and verbose logs into one flat request. Because Oxlo.ai charges per request rather than per token, adding extra log lines or a high-resolution image does not inflate the cost. That predictability matters when you are iterating on prompts at 3 a.m. See https://oxlo.ai/pricing for current plan details.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
user_message = [
{"type": "image_url", "image_url": {"url": image_b64}},
{"type": "text", "text": "Logs:\n2024-05-21T03:14:22Z ERROR connection pool exhausted\n2024-05-21T03:14:23Z FATAL request timeout after 30s\n2024-05-21T03:14:25Z ERROR retry failed"}
]
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
response_format={"type": "json_object"},
temperature=0.2,
)
print(response.choices[0].message.content)Step 4: Parse and validate the structured output
Raw JSON from a vision model often arrives wrapped in markdown fences. I strip those guards and parse the payload with the standard library so downstream automation can act on the result without string matching.
import json
raw = response.choices[0].message.content.strip()
if raw.startswith("
```"):
raw = raw.split("\n", 1)[1].rsplit("```
", 1)[0].strip()
diag = json.loads(raw)
assert "anomaly_seen" in diag, "Missing anomaly_seen key"
print(f"Anomaly detected: {diag['anomaly_seen']}")
print(f"Root cause: {diag['root_cause']}")
print(f"Suggested fix: {diag['remediation']}")Step 5: Bridge vision to action with tool use
A screenshot is only one frame. I give the model a fetch_logs function so it can request additional log ranges before finalizing its diagnosis. This bridges vision and action, which is the hard part of engineering multimodal agents.
import json
def fetch_logs(start: str, end: str) -> str:
# Stub for your internal log store.
return f"Stub logs from {start} to {end}"
tools = [
{
"type": "function",
"function": {
"name": "fetch_logs",
"description": "Retrieve logs for a time range in ISO format.",
"parameters": {
"type": "object",
"properties": {
"start": {"type": "string"},
"end": {"type": "string"}
},
"required": ["start", "end"]
}
}
}
]
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
],
tools=tools,
temperature=0.2,
)
msg = response.choices[0].message
if msg.tool_calls:
tool_call = msg.tool_calls[0]
args = json.loads(tool_call.function.arguments)
extra_logs = fetch_logs(args["start"], args["end"])
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
{"role": "assistant", "content": msg.content or "", "tool_calls": [
{"id": tool_call.id, "type": tool_call.type, "function": {"name": tool_call.function.name, "arguments": tool_call.function.arguments}}
]},
{"role": "tool", "tool_call_id": tool_call.id, "content": extra_logs},
]
final = client.chat.completions.create(
model="kimi-k2.6",
messages=messages,
response_format={"type": "json_object"},
temperature=0.2,
)
print(final.choices[0].message.content)
else:
print(msg.content)Run it
I combine the stable pieces into a single diagnose() function. Pass it a local PNG and a log tail, and it returns a validated dict.
import base64
import json
from pathlib import Path
from openai import OpenAI
def encode_image(path: str) -> str:
ext = Path(path).suffix.lstrip(".")
if ext == "jpg":
ext = "jpeg"
with open(path, "rb") as f:
b64 = base64.b64encode(f.read()).decode("utf-8")
return f"data:image/{ext};base64,{b64}"
SYSTEM_PROMPT = """You are an on-call site reliability engineer. You have two inputs:
1. A screenshot of a Grafana dashboard panel.
2. A short tail of raw application logs.
Follow this protocol exactly:
- First, list every visible metric name and its approximate value in the screenshot.
- Second, correlate any spikes or anomalies with timestamped ERROR or FATAL lines in the logs.
- Third, emit a single JSON object with keys: "anomaly_seen" (bool), "metric" (str), "log_signature" (str), "root_cause" (str), "remediation" (str).
If a metric is unreadable, state "unreadable" rather than guessing."""
def diagnose(image_path: str, log_tail: str) -> dict:
image_b64 = encode_image(image_path)
user_content = [
{"type": "image_url", "image_url": {"url": image_b64}},
{"type": "text", "text": f"Logs:\n{log_tail}"}
]
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
r = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_content},
],
response_format={"type": "json_object"},
temperature=0.2,
)
raw = r.choices[0].message.content.strip()
if raw.startswith("
```"):
raw = raw.split("\n", 1)[1].rsplit("```
", 1)[0].strip()
return json.loads(raw)
if __name__ == "__main__":
logs = """2024-05-21T03:14:22Z ERROR connection pool exhausted
2024-05-21T03:14:23Z FATAL request timeout after 30s
2024-05-21T03:14:25Z ERROR retry failed"""
result = diagnose("dashboard.png", logs)
print(json.dumps(result, indent=2))Example output:
{
"anomaly_seen": true,
"metric": "DB connection pool usage",
"log_signature": "connection pool exhausted",
"root_cause": "The connection pool maxed out at 03:14 UTC, causing cascading timeouts.",
"remediation": "Increase pool size or add connection retry with exponential backoff."
}Wrap-up
Wire this function into a PagerDuty webhook so incoming pages automatically trigger the diagnostician and post the JSON summary back to Slack. If latency becomes critical, swap Kimi K2.6 for Gemma 3 27B on Oxlo.ai; the request-based pricing means you can A/B test the trade-off between speed and reasoning depth without watching token meters spin. Check https://oxlo.ai/pricing to see which plan fits your volume.
Top comments (0)