Image generation

gutenberg_kg.image_gen

Image generation — local MLX or remote mflux-serve.

Two code paths: generate() — local Flux2Klein via mflux (Apple Silicon only) generate_via_server() — HTTP call to a running mflux-serve instance generate_auto() — server if GUTENKG_IMAGE_ENDPOINT / server_url set, else local fallback

Environment variables

GUTENKG_IMAGE_MODEL HuggingFace repo or mflux model name (default: mlx-community/flux2-klein-4b-4bit) GUTENKG_IMAGE_STEPS Inference steps (default: 4) GUTENKG_IMAGE_SIZE Default output size WIDTHxHEIGHT (default: 1536x1024) GUTENKG_IMAGE_ENDPOINT Base URL of a running mflux-serve instance (default: empty — use local generation) Example: http://localhost:8090 (mflux-server default)

discover_image_endpoint(candidates=DEFAULT_IMAGE_ENDPOINTS, timeout=1.5)

Find a local image server that is actually up.

Tries /health first and falls back to /v1/models. Both routes are served by gutenkg-image-server and gutenkg-sdxl-server, but /health is the newer one, and an already-running server — or a third-party mflux-serve — will only answer the model listing. Falling back costs one extra request against a port that is live either way.

A connection refused, a timeout, or any non-200 all mean "not this one".

Parameters:

Name	Type	Description	Default
candidates	tuple[str, ...]	Base URLs to try, in order.	DEFAULT_IMAGE_ENDPOINTS
timeout	float	Per-candidate timeout in seconds.	1.5

Returns:

Type	Description
str | None	The first responding base URL, or None if none answered.

Source code in src/gutenberg_kg/image_gen.py

def discover_image_endpoint(
    candidates: tuple[str, ...] = DEFAULT_IMAGE_ENDPOINTS,
    timeout: float = 1.5,
) -> str | None:
    """
    Find a local image server that is actually up.

    Tries ``/health`` first and falls back to ``/v1/models``.  Both routes are
    served by `gutenkg-image-server` and `gutenkg-sdxl-server`, but ``/health``
    is the newer one, and an already-running server — or a third-party
    mflux-serve — will only answer the model listing.  Falling back costs one
    extra request against a port that is live either way.

    A connection refused, a timeout, or any non-200 all mean "not this one".

    :param candidates: Base URLs to try, in order.
    :param timeout: Per-candidate timeout in seconds.
    :return: The first responding base URL, or None if none answered.
    """
    import httpx

    for base in candidates:
        root = base.rstrip("/")
        for route in ("/health", "/v1/models"):
            try:
                reply = httpx.get(f"{root}{route}", timeout=timeout)
            except Exception:  # noqa: BLE001 - any failure just means "try the next"
                break  # port is down; no point trying its other route
            if reply.status_code == 200:
                return base
    return None

generate(prompt, *, size=None, seed=None, output_path=None, model_name=None, steps=None)

Generate an image locally via Flux2Klein (Apple Silicon / mflux).

Parameters:

Name	Type	Description	Default
prompt	str	Text description of the image to generate.	required
size	str | None	Output size "WIDTHxHEIGHT" (default: GUTENKG_IMAGE_SIZE or 1536x1024).	None
seed	int | None	Random seed for reproducibility (random if omitted).	None
output_path	str | Path | None	If given, save the PNG here in addition to returning it.	None
model_name	str | None	Override the HF model repo (default: mlx-community/flux2-klein-4b-4bit).	None
steps	int | None	Override inference steps (default: 4).	None

Returns:

Type	Description
Image	PIL Image.

Source code in src/gutenberg_kg/image_gen.py

def generate(
    prompt: str,
    *,
    size: str | None = None,
    seed: int | None = None,
    output_path: str | Path | None = None,
    model_name: str | None = None,
    steps: int | None = None,
) -> PILImage:
    """Generate an image locally via Flux2Klein (Apple Silicon / mflux).

    :param prompt: Text description of the image to generate.
    :param size: Output size ``"WIDTHxHEIGHT"`` (default: GUTENKG_IMAGE_SIZE or 1536x1024).
    :param seed: Random seed for reproducibility (random if omitted).
    :param output_path: If given, save the PNG here in addition to returning it.
    :param model_name: Override the HF model repo (default: mlx-community/flux2-klein-4b-4bit).
    :param steps: Override inference steps (default: 4).
    :returns: PIL Image.
    """
    model_name = model_name or os.environ.get("GUTENKG_IMAGE_MODEL", _DEFAULT_MODEL)
    steps = steps or int(os.environ.get("GUTENKG_IMAGE_STEPS", _DEFAULT_STEPS))
    seed = seed if seed is not None else random.randint(0, 2**31 - 1)
    size = size or os.environ.get("GUTENKG_IMAGE_SIZE", _DEFAULT_SIZE)
    width, height = _parse_size(size) or _DEFAULT_DIMS

    model = _load_model(model_name)
    result = model.generate_image(
        seed=seed,
        prompt=prompt,
        width=width,
        height=height,
        guidance=1.0,
        num_inference_steps=steps,
        scheduler="flow_match_euler_discrete",
    )

    pil_image: PILImage = result.image

    if output_path is not None:
        pil_image.save(str(output_path))

    return pil_image

generate_auto(prompt, *, server_url=None, size=None, seed=None, steps=None, model_name=None)

Generate an image, preferring a remote server and falling back to local mflux.

Resolution order: 1. server_url argument (non-empty string) 2. GUTENKG_IMAGE_ENDPOINT environment variable 3. Local generate() — requires mflux on Apple Silicon

Parameters:

Name	Type	Description	Default
prompt	str	Text description of the image to generate.	required
server_url	str | None	Override server URL; pass None to use env var or local.	None
size	str | None	Output size "WIDTHxHEIGHT" (default: GUTENKG_IMAGE_SIZE or 1536x1024).	None
seed	int | None	Optional integer seed for reproducibility.	None
steps	int | None	Override inference steps.	None
model_name	str | None	Local model override (ignored when using server).	None

Returns:

Type	Description
Image	PIL Image.

Source code in src/gutenberg_kg/image_gen.py

def generate_auto(
    prompt: str,
    *,
    server_url: str | None = None,
    size: str | None = None,
    seed: int | None = None,
    steps: int | None = None,
    model_name: str | None = None,
) -> PILImage:
    """Generate an image, preferring a remote server and falling back to local mflux.

    Resolution order:
      1. *server_url* argument (non-empty string)
      2. GUTENKG_IMAGE_ENDPOINT environment variable
      3. Local generate() — requires mflux on Apple Silicon

    :param prompt: Text description of the image to generate.
    :param server_url: Override server URL; pass None to use env var or local.
    :param size: Output size ``"WIDTHxHEIGHT"`` (default: GUTENKG_IMAGE_SIZE or 1536x1024).
    :param seed: Optional integer seed for reproducibility.
    :param steps: Override inference steps.
    :param model_name: Local model override (ignored when using server).
    :returns: PIL Image.
    """
    url = server_url or os.environ.get("GUTENKG_IMAGE_ENDPOINT", "")
    if url:
        return generate_via_server(prompt, server_url=url, size=size, seed=seed, steps=steps)
    return generate(prompt, size=size, seed=seed, steps=steps, model_name=model_name)

generate_via_server(prompt, *, server_url, size=None, seed=None, steps=None)

Generate an image by calling a running mflux-serve HTTP server.

Requires only httpx + pillow — safe to call from Linux containers or any environment without mflux installed.

Parameters:

Name	Type	Description	Default
prompt	str	Text description of the image to generate.	required
server_url	str	Base URL of the mflux-serve instance, e.g. http://localhost:8088.	required
size	str | None	Output size "WIDTHxHEIGHT" (default: GUTENKG_IMAGE_SIZE or 1536x1024).	None
seed	int | None	Optional integer seed for reproducibility.	None
steps	int | None	Override inference steps (default: GUTENKG_IMAGE_STEPS or 4).	None

Returns:

Type	Description
Image	PIL Image decoded from the server response.

Source code in src/gutenberg_kg/image_gen.py

def generate_via_server(
    prompt: str,
    *,
    server_url: str,
    size: str | None = None,
    seed: int | None = None,
    steps: int | None = None,
) -> PILImage:
    """Generate an image by calling a running mflux-serve HTTP server.

    Requires only httpx + pillow — safe to call from Linux containers or any
    environment without mflux installed.

    :param prompt: Text description of the image to generate.
    :param server_url: Base URL of the mflux-serve instance, e.g. http://localhost:8088.
    :param size: Output size ``"WIDTHxHEIGHT"`` (default: GUTENKG_IMAGE_SIZE or 1536x1024).
    :param seed: Optional integer seed for reproducibility.
    :param steps: Override inference steps (default: GUTENKG_IMAGE_STEPS or 4).
    :returns: PIL Image decoded from the server response.
    """
    import httpx
    from PIL import Image

    steps = steps or int(os.environ.get("GUTENKG_IMAGE_STEPS", _DEFAULT_STEPS))
    size = size or os.environ.get("GUTENKG_IMAGE_SIZE", _DEFAULT_SIZE)
    width, height = _parse_size(size) or _DEFAULT_DIMS

    payload: dict = {
        "prompt": prompt,
        "n": 1,
        "size": f"{width}x{height}",
        "num_inference_steps": steps,
        "response_format": "b64_json",
    }
    if seed is not None:
        payload["seed"] = seed

    resp = httpx.post(
        server_url.rstrip("/") + "/v1/images/generations",
        json=payload,
        timeout=httpx.Timeout(connect=5.0, read=300.0, write=10.0, pool=5.0),
    )
    resp.raise_for_status()
    b64 = resp.json()["data"][0]["b64_json"]
    return Image.open(BytesIO(base64.b64decode(b64)))

vlm_rewrite(corpus_text, *, base_url=None, model=None)

Rewrite corpus prose into a visual image-generation prompt via a local VLM.

Parameters:

Name	Type	Description	Default
corpus_text	str	Historical text to rewrite as a visual scene description.	required
base_url	str | None	OpenAI-compatible endpoint URL. Falls back to GUTENKG_VLM_ENDPOINT or http://localhost:8080/v1.	None
model	str | None	Model ID. Falls back to GUTENKG_VLM_MODEL or the Qwen3-30B default.	None

Returns:

Type	Description
tuple[str, str | None]	(prompt, error) — prompt is the rewritten text (or corpus_text on failure); error is None on success or a short message describing the failure.

Source code in src/gutenberg_kg/image_gen.py

def vlm_rewrite(
    corpus_text: str,
    *,
    base_url: str | None = None,
    model: str | None = None,
) -> tuple[str, str | None]:
    """Rewrite corpus prose into a visual image-generation prompt via a local VLM.

    :param corpus_text: Historical text to rewrite as a visual scene description.
    :param base_url: OpenAI-compatible endpoint URL. Falls back to GUTENKG_VLM_ENDPOINT
                     or http://localhost:8080/v1.
    :param model: Model ID. Falls back to GUTENKG_VLM_MODEL or the Qwen3-30B default.
    :returns: (prompt, error) — prompt is the rewritten text (or corpus_text on failure);
              error is None on success or a short message describing the failure.
    """
    import re

    try:
        from openai import OpenAI
    except ImportError:
        return corpus_text, "openai package not installed"

    url = base_url or os.environ.get("GUTENKG_VLM_ENDPOINT", _DEFAULT_VLM_BASE_URL)
    mdl = model or os.environ.get("GUTENKG_VLM_MODEL", _DEFAULT_VLM_MODEL)

    try:
        client = OpenAI(base_url=url, api_key="not-needed")
        response = client.chat.completions.create(
            model=mdl,
            messages=[
                {"role": "system", "content": _VLM_SYSTEM},
                {"role": "user", "content": corpus_text},
            ],
            max_tokens=300,
            temperature=0.7,
            extra_body={"think": False, "chat_template_kwargs": {"enable_thinking": False}},
        )
        raw = (response.choices[0].message.content or "").strip()
        # Strip any <think>…</think> blocks the model emitted despite being asked not to.
        cleaned = re.sub(r"<think>.*?</think>", "", raw, flags=re.DOTALL).strip()
        return (cleaned or corpus_text), None
    except Exception as exc:  # noqa: BLE001
        return corpus_text, str(exc)