Genace
Sign inGet started

AI generation API LangGraph: make job observation a resumable graph step

LangGraph becomes useful after your application has a generation job identifier that must survive several decisions. A graph node can observe the job and produce an explicit application outcome without performing another paid submission. The example below is deliberately an observation graph: you provide a previously accepted ID, and each invocation makes one bounded status call.

1. Save the accepted ID first

Create a generation through POST /api/v1/images/generations and persist the returned job_id in your application operation record. Keep that side effect separate from nodes you expect to replay. The generation route does not advertise a request idempotency key, so replaying a submission after an uncertain network result can create another job.

Install langgraph and requests in a Python environment. The state below contains the remote identifier, the latest state, and an outcome intended for your application’s scheduler or user interface.

import os
import re
from typing import TypedDict
import requests
from langgraph.graph import StateGraph, START, END

class GenerationState(TypedDict, total=False):
    job_id: str
    state: str
    outputs: list | None
    error: dict | None
    outcome: str

def observe(state: GenerationState) -> dict:
    job_id = state["job_id"]
    if not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", job_id):
        raise ValueError("Invalid job identifier")
    response = requests.get(
        f"https://genace.ai/api/v1/jobs/{job_id}",
        params={"wait": 30},
        headers={"Authorization": f"Bearer {os.environ['GENACE_API_KEY']}"},
        timeout=45,
    )
    response.raise_for_status()
    job = response.json()
    return {"state": job["state"], "outputs": job.get("outputs"),
            "error": job.get("error")}

def classify(state: GenerationState) -> dict:
    current = state["state"]
    outcome = "ready" if current == "succeeded" else (
        "terminal_error" if current in {"failed", "canceled"} else "pending")
    return {"outcome": outcome}

builder = StateGraph(GenerationState)
builder.add_node("observe", observe)
builder.add_node("classify", classify)
builder.add_edge(START, "observe")
builder.add_edge("observe", "classify")
builder.add_edge("classify", END)
graph = builder.compile()

def inspect_saved_job(job_id: str):
    return graph.invoke({"job_id": job_id})

2. Schedule the pending branch outside this graph

The graph ends after classification, including when the generation is still pending. Your scheduler can persist the returned state and invoke observation again later. This keeps a long-running render from consuming an unbounded graph loop or hitting a recursion limit just because no result is ready yet.

The example does not include a durable LangGraph checkpointer. Calling compile() alone does not provide process-restart persistence. Use your application’s database for the operation record, and configure a durable checkpointer separately if you need graph-level recovery. An in-memory saver is appropriate for local experiments but cannot promise recovery after the host process exits.

3. Publish only validated outcomes

A ready outcome means the remote state is succeeded. Your asset consumer should still inspect the output list, select the media type it expects, and verify a nonempty URL. A terminal error should preserve the remote error object. Do not erase it by reducing every unsuccessful case to a generic timeout.

Keep the distinction between an observation exception and a terminal job failure. A temporary GET transport problem can leave the job running; it should not reroute the workflow into another submission. Likewise, the status response uses id, whereas the initial accepted response uses job_id.

Common mistakes and verification

Do not assume graph state makes external POST operations exactly once. Persistence and replay semantics need an explicit design around the side effect. Never place the server key in graph state that may be shown in traces or returned to a browser.

Verify an invocation makes exactly one GET, pending responses end cleanly, and failed or canceled responses classify as terminal errors. Restart your scheduler and confirm it loads the original job ID. Use null-output fixtures to ensure classification does not depend on an image already existing.

See LangGraph’s graph documentation, custom LangChain tools, and input validation for the surrounding contracts.

Continue with agents and automation

Models covered on this page

Where these facts come from