Skip to content

Recipe: Multi-turn Image Sessions

Goal

Generate an image, refine it in a second step, save the whole conversation as JSON, restore it in another process and keep editing, then read what every step cost.

ImageSession is the multi-turn surface. For a single generation or edit, the one-shot generate_image / edit_image calls are simpler. Image Sessions explains the rules this recipe relies on: persistence modes, the envelope guarantees, recovery and bounds.

Prerequisites

  • OPENAI_API_KEY for gpt-image-2 (the organisation must be verified for GPT Image models), or GOOGLE_GEMINI_API_KEY for the Gemini image rows. Keys are read from .env at the repo root.
  • Only OpenAI and Google generate images. An Anthropic or Fireworks model raises MultimediaValidationError before anything is sent.

Generate, then edit the result (OpenAI)

from doc_processing_toolkit.image_session import ImageSession

session = ImageSession("gpt-image-2", quality="low", size="1024x1024")

first = session.start("A flat red circle centred on a plain white background")
second = session.step("Make the circle blue and add a thin black border")

print(first.status, second.status)         # completed completed
print(second.mime_type)                    # image/png

with open("circle.png", "wb") as f:
    f.write(session.latest_image())        # the provider's original bytes, never a re-encode

start() opens the conversation and step() continues it: each call is one step, and each returns an ImageTurn. The session keeps the history, so step() edits the previous result — you do not pass the image back in. ImageSession is also importable as multimedia.ImageSession.

Settings given to the constructor (size, quality, output_format, background, ...) apply to every step and are validated against the registry when the session is created, exactly as the one-shot image functions validate them.

Step from an async server

Inside an event loop — a FastAPI handler, a websocket server — use astep, which runs the render on a worker thread instead of blocking the loop:

from doc_processing_toolkit.image_session import ImageSession


async def refine(envelope: dict, instruction: str) -> dict:
    session = ImageSession.from_dict(envelope)
    turn = await session.astep(instruction)     # same ImageTurn step() returns
    if turn.status != "completed":
        raise RuntimeError(f"{turn.error_code}: {turn.error}")
    return session.to_dict()                    # persist it for the next request

Await one step before starting the next: a second astep() (or step()) on the same session while one is running raises MultimediaValidationError instead of queueing, because each step edits the previous result.

Save the session and restore it somewhere else

import json

envelope = session.to_dict()               # JSON-serialisable
text = json.dumps(envelope)                # store it anywhere: a file, a blob store, a queue message

# ...later, possibly in another process...
restored = ImageSession.from_dict(json.loads(text))
third = restored.step("Now put it on a dark grey background")

print(restored.persistence)                # 'replay'
print(len(restored.turns))                 # 3: the restored turns and the new one

The envelope carries the whole conversation, so from_dict() needs nothing but the mapping. It checks every image's SHA-256 on the way back in and refuses an envelope written by a newer schema version.

Envelopes are large — keep them out of a database column

Images travel as base64. In the run that verified this recipe, two 1024x1024 low-quality turns made a 2.1 MB envelope. Put the envelope in object storage, or pass to_dict(externalize=True) with an image_store so the envelope keeps references and hashes instead of the bytes; restore it with from_dict(envelope, image_loader=...). See What the envelope guarantees.

The same flow on Google

session = ImageSession("gemini-3.1-flash-lite-image", size="1K", aspect_ratio="1:1")
session.start("A flat red circle centred on a plain white background")

restored = ImageSession.from_dict(json.loads(json.dumps(session.to_dict())))
turn = restored.step("Make the circle blue")

print(turn.status)                         # completed
print(turn.mime_type)                      # the type the model returned — image/jpeg in our run
print(turn.recovered_from)                 # None: the replayed history was accepted as-is
print(turn.signatures_lost)                # False: every thought signature was sent back unchanged

Two differences from OpenAI are worth knowing:

  • Sizes are resolution classes, not pixels. Gemini takes "1K", "2K" or "4K" plus an aspect_ratio; gemini-3.1-flash-lite-image accepts "1K" only. OpenAI-only options (quality, background, output_format, ...) are refused locally on Google.
  • The history carries thought signatures. Google rejects a replayed image part whose signature does not match the exact bytes it returned, which is why the session — not your code — owns the history, and why the envelope stores the provider's own bytes.

A PDF or a clip as the starting point

On gemini-nano-banana-2.1 the first turn can start from a document or a video clip, and the session can be told how hard to think:

session = ImageSession("gemini-nano-banana-2.1", size="1K", thinking_level="minimal")
poster = session.start("Make a poster from this notice, keeping every word of its text", image="notice.pdf")
poster = session.step("Make the title red and add a small rowing boat in the corner")
print(poster.output_tokens)                # includes the thinking tokens, billed at the text rate

The PDF stays in the replayed history, so the second turn re-sends it and pays for it again; see References and thinking. A clip over 20 MB is uploaded once through the Files API instead and kept for the session, which re-uploads it from the kept bytes if it expires — nothing to do in the calling code.

signatures_lost=True on a turn means the session had to rebuild from the last image because the signed history was rejected, and the rebuild produced an image that replaced it (a rebuild that comes back blocked leaves the signed history, and the flag, untouched); the accumulated prompts are restated so earlier edits survive, and recovered_from names the recovery that ran. A recovered step can cost more than a clean one, because the rebuild re-sends the image as input. See Recovery is bounded and visible.

Persistence modes

persistence defaults to "auto", which resolves to "replay" on both providers: the session holds the conversation itself, so the envelope is self-contained and every step is costed exactly.

persistence="server" is an OpenAI-only opt-in that lets the Responses API hold the chain. It needs a response_model — the mainline model that hosts the image tool — and both models are billed:

session = ImageSession(
    "gpt-image-2",
    persistence="server",
    response_model="gpt-5.5",
    quality="low",
)

response_model is checked only as "an OpenAI model the registry knows"; one outside the short list both OpenAI pages document as tool hosts (image_session.OPENAI_IMAGE_TOOL_HOSTS_DOCUMENTED — gpt-5, gpt-5-nano, gpt-5.2, gpt-5.5) is sent anyway with a warning. It is not cheaper (OpenAI re-bills the chained history as input either way), stored responses expire after 30 days, and each step bills the host model as well as the image model. Stay on "replay" unless you have a reason not to. On Google, persistence="server" raises.

Read what each step cost

for turn in restored.turns:
    print(turn.index, turn.status, f"${turn.cost:.6f}", turn.cost_estimated, len(turn.cost_records))

print(f"${restored.total_cost:.6f}")       # sum of every turn's cost
print(restored.cost_estimated)             # True if any component anywhere was derived
print(restored.cost_breakdown)
# {'cost_input_text_usd': ..., 'cost_input_image_usd': ..., 'cost_text_usd': ...,
#  'cost_image_usd': ..., 'cost_cache_discount_usd': ...}

From the verifying run (gpt-image-2, quality="low", 1024x1024):

Turn Input tokens Output tokens Cost
0 — generate 16 196 $0.00596
1 — edit 1,040 196 $0.01415
2 — edit after restore 1,038 196 $0.01414

The edits cost more than the first generation because each one sends the previous image back as input, billed at the row's input_image rate. gemini-3.1-flash-lite-image came to about $0.034 per 1K step.

Each turn carries cost_records, one per API request, and turn.cost is their sum; the session total is the sum over turns. A replay step makes one request. An OpenAI persistence="server" step makes two records — the mainline model and the image model — each priced from the usage the API reports for that model. The image record is flagged cost_estimated=True only when a response carries no image usage and the published per-quality table is used instead.

Failed and bounded steps

A step that fails still returns a turn — status="failed", with error, error_code and retryable — and its cost is still recorded, because the provider billed it. A failed turn does not advance the conversation: the next step() continues from the last good image. Check turn.status before using turn.image_bytes, which is None on a failed turn; a moderation_blocked turn is never retryable.

max_turns (default 16) and max_history_bytes (default 32 MiB) raise MultimediaValidationError rather than silently dropping earlier turns. Pass None to either to disable it.

What to validate

  • session.latest_image() returns bytes after start(), and step() edits that image rather than starting over
  • json.dumps(session.to_dict()) succeeds, and ImageSession.from_dict() continues with the same number of turns
  • restored.total_cost equals sum(t.cost for t in restored.turns)
  • on Google, a restored step completes with recovered_from is None and signatures_lost is False

Test sources

  • tests/unit/test_image_session.py
  • tests/integration/test_image_session_live.py