--- title: Generating an image and removing its background | Runware Docs url: https://runware.ai/docs/serverless/examples/chained-steps description: Run two models in one request so the image between them never leaves the worker, which no managed API can do. relatedDocuments: - https://runware.ai/docs/serverless/volumes - https://runware.ai/docs/serverless/writing-a-model --- ## Introduction A product shot generated and cut out of its background, in one call. Two different models run in one worker, and the image between them never leaves the process. A managed inference API cannot give you this, because **the only thing that crosses a request boundary is a finished image**. ## The app **model.py**: ```python """Two models in one request, so the image between them never crosses a boundary.""" from __future__ import annotations import base64 import io from pydantic import BaseModel, Field from runware_serverless import serve GENERATOR = "Qwen/Qwen-Image" """Apache 2.0 and ungated, so this deploys with no token.""" MATTE = "ZhengPeng7/BiRefNet" """MIT and ungated. Loading it runs code from its own repository, which is what trust_remote_code means, and that code has imports of its own: einops, kornia and timm are in requirements below because the matte needs them, not this file.""" CACHE = "/weights" """Mount a volume here. The generator alone is tens of gigabytes, so without one it is fetched again on every cold start and copied into every checkpoint.""" DEFAULT_STEPS = 30 MATTE_SIZE = 1024 class Request(BaseModel): prompt: str = Field(description="What to generate before the background is removed.") steps: int = Field(default=DEFAULT_STEPS, ge=1, le=100) class Generated(BaseModel): image: str = Field(description="The cut-out as a base64 PNG, with an alpha channel.") width: int height: int @serve class ChainedSteps: requirements = [ "torch==2.14.1", "torchvision==0.29.1", "diffusers==0.42.1", "transformers==5.21.0", "accelerate==1.15.0", "einops==0.8.2", "kornia==0.9.0", "timm==1.0.30", ] def load(self) -> None: if hasattr(self, "generator"): return import torch from diffusers import DiffusionPipeline from transformers import AutoModelForImageSegmentation self.generator = DiffusionPipeline.from_pretrained( GENERATOR, torch_dtype=torch.bfloat16, cache_dir=CACHE ).to("cuda") self.matte = AutoModelForImageSegmentation.from_pretrained( MATTE, trust_remote_code=True, cache_dir=CACHE ).to("cuda") self.matte.eval() self.generator(prompt="warmup", num_inference_steps=1) def predict(self, request: Request) -> Generated: import torch generated = self.generator( prompt=request.prompt, num_inference_steps=request.steps ).images[0] # The one thing a second API call cannot do. The image handed to the matte # is the object the generator produced: not encoded, not uploaded, not # queued again, and not waiting on a second worker to start. with torch.inference_mode(): alpha = self._alpha(generated) generated.putalpha(alpha) return Generated( image=_png(generated), width=generated.width, height=generated.height ) def _alpha(self, image: object) -> object: import torch from torchvision import transforms prepare = transforms.Compose( [ transforms.Resize((MATTE_SIZE, MATTE_SIZE)), transforms.ToTensor(), transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]), ] ) # Matched to the matte's own dtype, not assumed. Two models in one worker # are two dtypes, and the checkpoint decides each one: a float32 batch into # a half model fails at the first convolution, with a message about bias # types that says nothing about where it came from. weights = next(self.matte.parameters()) batch = prepare(image).unsqueeze(0).to(weights.device, weights.dtype) # pyright: ignore[reportArgumentType] mask = self.matte(batch)[-1].sigmoid().float().cpu()[0].squeeze() return transforms.ToPILImage()(mask).resize(image.size) # pyright: ignore[reportAttributeAccessIssue] def _png(image: object) -> str: buffer = io.BytesIO() image.save(buffer, format="PNG") # pyright: ignore[reportAttributeAccessIssue] return base64.b64encode(buffer.getvalue()).decode("ascii") ``` ## What stays in the process The image you hand to the matte is the object the generator returned, so you do not encode it to PNG, upload it, queue it again, or wait for a second worker to start. Two API calls pay all four of those for a step that takes a fraction of a second on a card that is already warm and already holding your image. ## Two models, one worker, one volume The generator is tens of gigabytes and the matte is small, so the second one costs you almost nothing: one cold start, one set of weights to fetch, one GPU to keep warm. Split them across two apps and you pay all three twice. Nothing in `model.py` names hardware, and your app has to fit the generator, so the matte rides along on whatever you chose. > [!WARNING] > `trust_remote_code=True` **runs code from the model's own repository**, and that code has imports of its own. BiRefNet needs `einops`, `kornia` and `timm`, which is why they are in `requirements` for a model your file never imports. Nothing warns you at build time. The build reads your endpoints without installing anything, so a missing import only shows up when a worker starts, as a `CrashLoopBackOff` and an `ImportError` in `apps logs`. **Read the remote code for its imports before you deploy it.** ## Licensing Qwen-Image is Apache 2.0 and BiRefNet is MIT, both ungated, so you deploy with no token and nothing here depends on who you are. ## Deploying it ```bash runware serverless deploy model.py \ --id chained-steps \ --gpu-type rtx-pro-6000 \ --volume /weights ``` The first deploy fetches both models into the volume. Later starts read them from it. ## Calling it The parameter is a model, so its fields sit under the parameter's own name. **TypeScript**: ```typescript import { createClient } from '@runware/sdk' const client = await createClient({ apiKey: process.env.RUNWARE_API_KEY }) const task = await client.invoke({ appId: 'chained-steps', endpointPath: 'predict', payload: { request: { prompt: 'a ceramic coffee cup on a plain surface, studio lighting' } } }) ``` **Python**: ```python import asyncio import os from runware import Runware async def main(): client = Runware(api_key=os.environ["RUNWARE_API_KEY"]) try: task = await client.invoke( "chained-steps", "predict", { "request": { "prompt": "a ceramic coffee cup on a plain surface, studio lighting" } }, ) finally: await client.close() asyncio.run(main()) ``` **cURL**: ```bash curl https://api.serverless.runware.ai/v1/apps/chained-steps/invoke-async/predict \ -H "Authorization: Bearer $RUNWARE_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7", "payload": { "request": { "prompt": "a ceramic coffee cup on a plain surface, studio lighting" } } }' ``` **CLI**: ```bash cat <<'EOF' | runware serverless apps invoke chained-steps predict -f - { "request": { "prompt": "a ceramic coffee cup on a plain surface, studio lighting" } } EOF ``` **JSON**: ```json { "taskId": "7c9e6679-7425-40de-944b-e07fc1f90ae7", "payload": { "request": { "prompt": "a ceramic coffee cup on a plain surface, studio lighting" } } } ``` The result is a base64 PNG **with an alpha channel** at `output.predictions[0].image`. > [!NOTE] > Call this one with `invoke-async`. A sync invoke has a deadline of roughly 105 seconds, and the first call spends longer than that fetching weights. > [!NOTE] > Measured on an RTX PRO 6000: **3m28s** for the first call with an empty volume, 44 seconds for a restart that finds the weights already there, and 14 seconds on a worker already serving. The volume does not keep a worker warm, it makes a **restart** cheap. The cut-out comes back in **RGBA**, with alpha running the full range from fully transparent to fully opaque.