Building gemini 3 multimodal agents that ingest raw images, audio, and video forces you to rethink prompt construction, token accounting, and failure modes compared to text-only flows. The model accepts interleaved media parts, but production reliability comes from explicit orchestration and conservative resource bounds.
1. Define the media contract before writing code
Treat every input as untrusted binary. Set hard limits on bytes, duration, and resolution before the bytes reach the model. A 30-second 1080p clip is a different object than a 200KB screenshot, and your agent should reject the former if the task only needs the latter.
from pydantic import BaseModel, Field, validator
import magic # python-magic for content sniffing
class MediaSpec(BaseModel):
max_image_bytes: int = 10_000_000
max_audio_seconds: float = 120.0
max_video_seconds: float = 60.0
allowed_types: set[str] = {"image/png", "image/jpeg", "audio/wav", "video/mp4"}
@validator("allowed_types")
def check_magic(cls, v, values, **kwargs):
# stub for illustration; real code sniffs bytes not extension
return v
def validate_upload(raw: bytes, declared_type: str, spec: MediaSpec) -> None:
mime = magic.from_buffer(raw, mime=True)
if mime not in spec.allowed_types:
raise ValueError(f"rejected {mime}, expected one of {spec.allowed_types}")Size and duration guards
Do not trust client metadata. Parse container headers for actual duration. A browser may label a file .wav but ship a 200MB opus inside an unknown container. Enforce limits at the edge, not in the prompt.
2. Assemble interleaved content parts
Gemini 3 multimodal agents work best when related text and media sit in the same turn. Use inline base64 rather than URLs unless your gateway supports signed fetches and you control the object store. Below is a minimal OpenAI-compatible payload that sends a frame and a question.
import base64, openai
client = openai.OpenAI(base_url="https://api.n4n.ai/v1", api_key="KEY")
def b64(path: str) -> str:
with open(path, "rb") as f:
return base64.b64encode(f.read()).decode()
msg = {
"role": "user",
"content": [
{"type": "text", "text": "Describe anomalies in this frame."},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64('frame.png')}"}},
],
}
resp = client.chat.completions.create(model="gemini-3-pro", messages=[msg], max_tokens=512)Text-first ordering
Put the instruction before the media. The attention pattern differs when the model sees “describe anomalies” before the 10MB image versus after. I test both during eval, but ship text-first for predictability.
Audio and video part shapes
For audio, extend the content list with an input_audio part if your endpoint supports it. For video, either send a video part with base64 or decompose into frames. A representative JSON fragment:
{
"type": "input_audio",
"input_audio": {"data": "base64string", "format": "wav"}
}If the gateway does not natively accept video, decompose client-side and attach frames plus a transcript.
3. Pre-process or send raw? Tradeoffs
Sending native video preserves temporal context but multiplies token usage non-linearly with duration. Extracting keyframes and an ASR transcript cuts cost but loses subtle cues like tone or off-screen sound.
For gemini 3 multimodal agents handling support tickets, I extract one frame per 2 seconds and attach Whisper transcripts. For safety review, I send the full clip. Make the path configurable per task, not global.
Frame sampling math
A 60-second clip at 1 fps yields 60 images. At 4 fps it yields 240. If each frame is ~1,500 prompt tokens when resized, that is 90K vs 360K tokens before any text. Know which side of the context window you are on.
def build_video_parts(path: str, sample_rate: float = 0.5) -> list:
parts = [{"type": "text", "text": "Review clip for policy violations."}]
for ts in extract_frames(path, sample_rate):
parts.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{ts}"}})
return partsASR coupling
Attach the transcript as a separate text part, not burned into the frame. That keeps the audio searchable and lets you swap models without re-decoding.
4. Stream and set aggressive timeouts
Multimodal inference lags text-only by seconds. Wrap calls in a 30s connect, 120s read timeout. Stream to avoid head-of-line blocking in your orchestrator.
stream = client.chat.completions.create(
model="gemini-3-pro",
messages=messages,
stream=True,
timeout=120,
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")Partial stream recovery
If the stream dies mid-way, retry with the same conversation id but truncate the last media part. Re-sending a 20MB video on every retry bankrupts your token budget. Persist the last successful offset and resume from there.
5. Route with fallback to dodge provider outages
Single-provider dependencies break at 2 a.m. When you front the model with an OpenAI-compatible gateway such as n4n.ai, it honors client routing directives and automatically falls back when a provider is rate-limited, so your agent only handles application errors. Still, set explicit routing if the API exposes it.
resp = client.chat.completions.create(
model="gemini-3-pro",
messages=messages,
extra_body={"route": "gemini-3-pro,gemini-2.5-pro"},
)Do not implement exponential backoff for 429s at the agent level if the gateway already shifts traffic. You should back off only on persistent 5xx after the gateway gives up. Log the x-routed-provider header if present to confirm failover happened.
6. Meter usage and forward cache hints
Per-token metering is non-negotiable when video frames inflate input counts. Log usage.prompt_tokens and usage.completion_tokens on every turn. If your gateway supports cache-control, mark static system prompts with cache_control: {"type": "ephemeral"} to cut repeat cost.
sys_msg = {
"role": "system",
"content": "You are a multimodal triage agent.",
"cache_control": {"type": "ephemeral"},
}
# n4n.ai forwards this hint to the providerGemini 3 multimodal agents that reuse the same instruction prefix across sessions should expect fewer billed prompt tokens when cache hits land. Verify by comparing usage across identical prefixes, not by trusting the dashboard.
Per-token logging pattern
def log_usage(resp):
u = resp.usage
metrics.emit("prompt_tokens", u.prompt_tokens)
metrics.emit("completion_tokens", u.completion_tokens)Wire this into your OTel pipeline before launch, not after the first invoice.
7. Production pitfalls I keep seeing
Media ordering bugs. Putting the text after a 50MB video part triggers different attention patterns than before. Test both during eval.
Silent downscaling. Some proxies resize images without flagging it. Hash inputs server-side to detect mutation.
Audio codec mismatch. WEBM opus from browsers often fails where MP4 AAC works. Transcode at the edge.
Context window math. A 60-second 720p video at 1 fps is 60 images; at 4 fps it is 240. That difference decides whether you fit in the window.
No dead-letter queue. Failed multimodal jobs should park with the original bytes, not a regenerated URL that expires in an hour.
Cache hint ignored on first turn. Ephemeral caches only help on the second identical prefix. Warm them with a synthetic call if latency matters.
Stream truncation mistaken for success. Always check finish_reason. A length stop is not a completed answer.
Build the contract first, send interleaved parts, pre-process deliberately, stream with timeouts, and let the gateway handle provider flux. The gemini 3 multimodal agents that survive contact with real users are boring pipelines with strict limits, not clever prompts.