Tách nền ảnh hàng loạt bằng Python: rembg hay API

Bạn có một thư mục vài trăm ảnh sản phẩm, ảnh chân dung hoặc ảnh đăng bán, và ảnh nào cũng cần tách nền. Làm thủ công trong trình chỉnh sửa ảnh là mất cả ngày bấm chuột. Trong Python, việc xóa nền ảnh hàng loạt chỉ là một vòng lặp, và quyết định thật sự là thứ gì chạy bên trong vòng lặp: một mô hình mã nguồn mở trên máy của bạn, hay một API HTTP.

Hướng dẫn này xây dựng cả hai cách, kèm script hoàn chỉnh bạn có thể chạy trên một thư mục ngay hôm nay, và đi qua những phần giúp một tác vụ hàng loạt chạy ổn định: chọn mô hình, CPU hay GPU, thử lại, giới hạn tốc độ, idempotency và chạy tiếp sau khi bị lỗi giữa chừng.

Cách A hay cách B?

Nhiều nhóm dùng cả hai: mô hình cục bộ cho thử nghiệm và công cụ nội bộ, API cho production, nơi chất lượng ổn định và không phải quản lý máy chủ GPU quan trọng hơn.

Cách A: rembg, con đường mã nguồn mở

rembg là một thư viện Python và CLI theo giấy phép MIT, gói nhiều mô hình tách nền và chạy chúng bằng ONNX Runtime. Tại thời điểm viết bài, phiên bản hiện tại là 2.0.86, yêu cầu Python 3.11 trở lên và thấp hơn 3.14.

Cài đặt

Chọn backend phù hợp với phần cứng của bạn. Bản CPU chạy được ở mọi nơi:

python -m venv .venv && source .venv/bin/activate
pip install "rembg[cpu]"        # library only
pip install "rembg[cpu,cli]"    # library plus the rembg command

Với GPU NVIDIA, hãy cài rembg[gpu] thay thế. Dự án lưu ý rằng bản này cần onnxruntime-gpu cùng phiên bản CUDA và cuDNN tương thích, và gợi ý quay về bản CPU nếu bạn không cài được chúng. GPU AMD trên Linux dùng rembg[rocm] với onnxruntime-rocm.

rembg tự động chọn execution provider CUDA (hoặc ROCm) khi ONNX Runtime báo là có sẵn, và cảnh báo nếu phải quay về CPU. Để kiểm tra ONNX Runtime nhìn thấy gì:

import onnxruntime
print(onnxruntime.get_available_providers())  # look for CUDAExecutionProvider

Chọn mô hình

Mô hình được tải về ở lần dùng đầu tiên vào ~/.rembg/models/ (đặt REMBG_HOME để đổi thư mục). Một số mô hình hữu ích trong danh sách của dự án:

Có hai điểm quan trọng hơn vẻ ngoài của chúng. Thứ nhất, mô hình mặc định không miễn phí cho mục đích thương mại: trọng số mô hình có giấy phép riêng, tách biệt với giấy phép MIT của rembg, nên hãy kiểm tra nguồn của mô hình bạn dùng trong sản phẩm. Thứ hai, các mô hình mới hơn (bria-rmbg, birefnet-*, isnet-general-use) cho viền mềm, đúng hình dáng, còn họ u2net cũ thường cho viền cứng và thô hơn.

Tùy chọn chất lượng viền

rembg có ba cách xử lý các điểm ảnh viền mềm như tóc và lông:

• Mặc định: ổn với chủ thể có viền rõ như sản phẩm, ô tô và logo.
• decontaminate=True (-dc trên CLI): loại bỏ quầng màu mà nền cũ để lại ở các điểm ảnh viền. Tùy chọn này chỉ đổi màu, không bao giờ đổi hình dạng mask, tốn rất ít tài nguyên và có thể bật an toàn cho cả lô ảnh.
• alpha_matting=True (-a): ước lượng lại chính đường viền khi hình dạng mask bị sai, ví dụ sợi tóc bị cắt mất. Cách này chậm hơn nhiều và có thể không hội tụ, khi đó rembg quay về ảnh cắt đã khử quầng màu.

Với các lô ảnh chạy không cần giám sát, dự án khuyên dùng -dc thay vì -a vì nhanh hơn và không thể thất bại. Hướng dẫn về kênh alpha giải thích vì sao quầng màu xuất hiện.

Cách nhanh: dùng CLI

rembg p -m birefnet-general -dc photos/ cutouts/

rembg p xử lý cả thư mục; thêm -w để theo dõi thư mục và xử lý file mới ngay khi chúng xuất hiện.

Script Python

Phiên bản này tải mô hình một lần, bỏ qua các file đã xong để bạn có thể dừng rồi chạy lại, và vẫn tiếp tục khi một file bị lỗi:

import sys
from pathlib import Path

from rembg import new_session, remove

EXTS = {".jpg", ".jpeg", ".png", ".webp"}


def main(in_dir: str, out_dir: str, model: str = "birefnet-general") -> None:
    src_dir, dst_dir = Path(in_dir), Path(out_dir)
    dst_dir.mkdir(parents=True, exist_ok=True)
    session = new_session(model)  # load the model once, reuse it for every image

    files = sorted(p for p in src_dir.iterdir() if p.suffix.lower() in EXTS)
    for i, src in enumerate(files, 1):
        dst = dst_dir / f"{src.stem}.png"
        if dst.exists():
            continue  # resume where a previous run stopped
        try:
            out = remove(src.read_bytes(), session=session, decontaminate=True)
            dst.write_bytes(out)
            print(f"[{i}/{len(files)}] ok      {src.name}")
        except Exception as e:  # one bad file shouldn't stop the batch
            print(f"[{i}/{len(files)}] failed  {src.name}: {e}")


if __name__ == "__main__":
    main(*sys.argv[1:])

Chạy bằng lệnh python batch_rembg.py photos cutouts birefnet-general. Tạo session một lần là điều quan trọng: tải mô hình mất nhiều thời gian hơn hẳn so với chạy nó, và README của rembg khuyên dùng lại session để xử lý hàng loạt nhanh hơn. Kết quả luôn là PNG vì ảnh có kênh alpha.

Những điểm yếu của rembg

• Lần chạy đầu chậm vì phải tải mô hình, và mô hình lớn cần nhiều RAM hoặc VRAM.
• Tốc độ trên CPU khá khiêm tốn với các mô hình chất lượng cao. Hãy thử với 20 ảnh tiêu biểu trước khi xử lý cả nghìn ảnh.
• Giấy phép và driver là việc của bạn. Cố định phiên bản rembg và mô hình để lần nâng cấp sau không âm thầm làm thay đổi kết quả.

Cách B: API HTTP

Với API, phần việc nặng chạy trên GPU của bên khác, và script của bạn chủ yếu lo xử lý mạng cho tốt. Ví dụ dùng API tách nền của removebackground.dev vì hợp đồng API được công bố dưới dạng OpenAPI 3.1 tại /openapi.json; các mẫu tương tự áp dụng cho mọi dịch vụ giống vậy.

Hợp đồng API

• POST https://removebackground.dev/v1/remove với key trong header x-api-key.
• Gửi ảnh dưới dạng trường multipart image_file (JPG, PNG hoặc WebP, tối đa 25 MB), hoặc dưới dạng JSON với image_url.
• Các trường tùy chọn gồm format (png, jpg hoặc webp), bg_color, size (preview, medium, hd, full) và crop.
• 200 nghĩa là body chính là ảnh đã xong. 202 nghĩa là ảnh vẫn đang xử lý sau khoảng 55 giây: body là JSON chứa id của job, và bạn poll GET /v1/jobs/{id}/result, endpoint này trả về 409 not_ready kèm Retry-After cho đến khi ảnh sẵn sàng. Đừng bao giờ lưu body của phản hồi 202 thành ảnh.
• Tối đa 60 yêu cầu xử lý mỗi phút cho mỗi tài khoản; vượt quá sẽ nhận 429 rate_limited kèm Retry-After.
• Header Idempotency-Key giúp thử lại an toàn: cùng key với cùng yêu cầu sẽ trả về job ban đầu thay vì tính phí lần nữa. Cùng key với yêu cầu khác sẽ nhận 409 idempotency_key_reused.
• Kết quả bị xóa sau 48 giờ, nên hãy lưu ngay khi nhận được.

Quy tắc thử lại để không bị tính phí hai lần

Mẹo là làm cho idempotency key có tính xác định: một mã hash từ byte của ảnh cộng với các tùy chọn. Nếu script bị dừng giữa chừng và bạn chạy lại, mỗi lần gửi lại đều mang cùng key, nên không ảnh nào bị xử lý hay tính phí hai lần.

Script Python

import hashlib
import mimetypes
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import requests

API = os.environ.get("REMOVEBG_API", "https://removebackground.dev")
KEY = os.environ["REMOVEBG_API_KEY"]  # keep the key out of your code
OPTIONS = {"format": "png", "size": "full"}  # or {"format": "jpg", "bg_color": "ffffff"}
WORKERS = 4  # the API allows 60 processing requests a minute
EXTS = {".jpg", ".jpeg", ".png", ".webp"}
RETRYABLE = {429, 500, 503}


def idempotency_key(data: bytes) -> str:
    # Same image + same options = same key, so a rerun never pays twice.
    h = hashlib.sha256(data)
    h.update(repr(sorted(OPTIONS.items())).encode())
    return h.hexdigest()


def wait(res, default=2.0):
    try:
        return float(res.headers.get("Retry-After", default))
    except ValueError:
        return default


def save(res, dst: Path) -> Path:
    if res.status_code != 200 or not res.headers.get("Content-Type", "").startswith("image/"):
        raise RuntimeError(f"HTTP {res.status_code}: {res.text[:300]}")
    dst.write_bytes(res.content)
    return dst


def process(src: Path, dst: Path) -> Path:
    data = src.read_bytes()
    mime = mimetypes.guess_type(src.name)[0] or "application/octet-stream"
    headers = {"x-api-key": KEY, "Idempotency-Key": idempotency_key(data)}

    for attempt in range(6):
        try:
            res = requests.post(
                f"{API}/v1/remove",
                files={"image_file": (src.name, data, mime)},
                data=OPTIONS,
                headers=headers,
                timeout=(10, 120),
            )
        except (requests.ConnectionError, requests.Timeout):
            time.sleep(2 ** attempt)  # same key: safe to resend
            continue
        if res.status_code in RETRYABLE:
            time.sleep(wait(res, 2 ** attempt))
            continue
        break
    else:
        raise RuntimeError("gave up after retries")

    if res.status_code != 202:
        return save(res, dst)  # 200 is the image; 4xx and 502 raise

    # 202: still processing. Poll the same job; never upload the image again.
    job_id = res.json()["id"]
    deadline = time.monotonic() + 600
    while time.monotonic() < deadline:
        time.sleep(wait(res))
        try:
            res = requests.get(f"{API}/v1/jobs/{job_id}/result", headers=headers, timeout=30)
        except (requests.ConnectionError, requests.Timeout):
            continue  # reading a job is free and safe to repeat
        if res.status_code == 409 and res.json()["error"]["code"] == "not_ready":
            continue
        if res.status_code >= 500:
            continue
        return save(res, dst)
    raise TimeoutError(f"job {job_id} still running; fetch it later")


def main(in_dir: str, out_dir: str) -> None:
    src_dir, dst_dir = Path(in_dir), Path(out_dir)
    dst_dir.mkdir(parents=True, exist_ok=True)
    ext = ".jpg" if OPTIONS.get("format") == "jpg" else "." + OPTIONS.get("format", "png")
    todo = [
        (p, dst_dir / (p.stem + ext))
        for p in sorted(src_dir.iterdir())
        if p.suffix.lower() in EXTS and not (dst_dir / (p.stem + ext)).exists()
    ]
    print(f"{len(todo)} images to process")

    failed = []
    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        jobs = {pool.submit(process, s, d): s for s, d in todo}
        for fut in as_completed(jobs):
            src = jobs[fut]
            try:
                print("ok    ", src.name, "->", fut.result().name)
            except Exception as e:
                failed.append(src.name)
                print("failed", src.name, e)
    print(f"done: {len(todo) - len(failed)} ok, {len(failed)} failed")


if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2])

Chạy bằng lệnh REMOVEBG_API_KEY=... python batch_api.py photos cutouts. Bốn worker là mức khởi đầu vừa phải; nếu phản hồi nhanh khiến bạn vượt giới hạn mỗi phút, phần xử lý 429 sẽ chờ rồi gửi lại với cùng key. Để gửi URL thay vì file, hãy post JSON:

res = requests.post(
    f"{API}/v1/remove",
    json={"image_url": "https://example.com/shoe.jpg", "format": "png"},
    headers={"x-api-key": KEY, "Idempotency-Key": "shoe-front-v1"},
    timeout=(10, 120),
)

Với danh sách URL dài, cùng API này có POST /v1/batch, xếp hàng tối đa 100 URL ảnh thành một job để bạn poll.

Làm cho mọi tác vụ hàng loạt chạy ổn định

• Bỏ qua các file đã xử lý xong để lần chạy lại tiếp tục thay vì bắt đầu từ đầu. Cả hai script đều làm điều này.
• Cẩn thận trùng tên. shoe.jpg và shoe.png trong cùng thư mục đều thành shoe.png; hãy đổi tên file đầu vào hoặc giữ phần mở rộng gốc trong tên file đầu ra.
• Ghi lỗi vào một file và chỉ chạy lại những file đó.
• Kiểm tra ngẫu nhiên một vài ảnh ở mức zoom 100% trên nền tối và nền sáng trước khi xử lý toàn bộ.
• Giữ nguyên ảnh gốc. Ghi kết quả vào một thư mục riêng.
• Chọn định dạng đầu ra theo nơi dùng. PNG hoặc WebP cho nền trong suốt, JPG nền trắng cho các sàn thương mại điện tử. Xem PNG, JPEG hay WebP.