ARM版飞牛RK3588满血激活NPU， Immich（rk3588/3566/rk3568/rk3576 NPU 加速版）

andychao

飞牛 NAS Docker Compose 部署 Immich（arm RK3588 NPU 加速版）

一、前置说明

1. 适用于飞牛 arm OS 1.1.3105 版本
2. 基于 RK3588 芯片，NPU 硬件加速推理
3. 已配置国内镜像，纯净部署
4. immich 机器学习 NPU 识别补丁
5. rk3566/rk3568/rk3576/rk3588 模型文件分享

---

二、部署文件

1. docker-compose.yml

yaml

name: immich

services:
  immich-server:
    container_name: immich_server
    image: ghcr.nju.edu.cn/immich-app/immich-server:${IMMICH_VERSION:-release}
    volumes:
      - ${UPLOAD_LOCATION}:/data
      - /etc/localtime:/etc/localtime:ro
      - /vol2/1000/超的照片:/photo # 外部相册（可自行修改路径）
    env_file:
      - .env
    ports:
      - '2283:2283'
    depends_on:
      - redis
      - database
    restart: always
    healthcheck:
      disable: false

  immich-machine-learning:
    container_name: immich_machine_learning
    image: ghcr.nju.edu.cn/immich-app/immich-machine-learning:${IMMICH_VERSION:-release}-rknn
    command: ["python", "/appdata/rknn_patcher.py"]
    privileged: true
    volumes:
      - ${CACHE_LOCATION}:/cache
      - ./rknn_patcher.py:/appdata/rknn_patcher.py:ro
      - /sys/class/drm/card0/device/uevent:/usr/src/immich_ml/device_uevent:ro
      - /sys/firmware/devicetree/base:/usr/src/immich_ml/device_tree:ro
      - /dev:/dev:rw
      - /sys:/sys:rw
    env_file:
      - .env
    environment:
      - MACHINE_LEARNING__RKNN=true
      - DEVICE=npu
    ipc: host
    restart: always
    healthcheck:
      disable: false

  redis:
    container_name: immich_redis
    image: docker.io/valkey/valkey:9@sha256:3b55fbaa0cd93cf0d9d961f405e4dfcc70efe325e2d84da207a0a8e6d8fde4f9
    healthcheck:
      test: redis-cli ping || exit 1
    restart: always

  database:
    container_name: immich_postgres
    image: ghcr.nju.edu.cn/immich-app/postgres:14-vectorchord0.4.3-pgvectors0.2.0@sha256:bcf63357191b76a916ae5eb93464d65c07511da41e3bf7a8416db519b40b1c23
    environment:
      POSTGRES_PASSWORD: ${DB_PASSWORD}
      POSTGRES_USER: ${DB_USERNAME}
      POSTGRES_DB: ${DB_DATABASE_NAME}
      POSTGRES_INITDB_ARGS: '--data-checksums'
    volumes:
      - ${DB_DATA_LOCATION}:/var/lib/postgresql/data
    shm_size: 128mb
    restart: always
    healthcheck:
      disable: false

volumes:
  model-cache:

2. .env 环境变量

env

# 时区
TZ=Asia/Shanghai

# Immich版本
IMMICH_VERSION=v2

# 照片上传目录
UPLOAD_LOCATION=./library
# AI模型缓存目录
CACHE_LOCATION=./cache
# 数据库存储目录
DB_DATA_LOCATION=./postgres

# RK3588 NPU 线程数（建议2~3）
MACHINE_LEARNING_RKNN_THREADS=2

# 数据库密码
DB_PASSWORD=postgres

# 固定配置（无需改动）
DB_USERNAME=postgres
DB_DATABASE_NAME=immich

---

# 三、rknn_patcher.py（ immich 机器学习 NPU 专用补丁）

#!/usr/bin/env python3
import os, shutil, stat, ast, re

def _setup_dev_nodes():
    render_128 = "/dev/dri/renderD128"
    render_129 = "/dev/dri/renderD129"

    try:
        if os.path.isdir(render_129):
            os.rmdir(render_129)

        if not os.path.exists(render_129) and os.path.exists(render_128):
            if not os.geteuid() == 0:
                print("[PATCH] Skip mknod: no root permission")
                return

            dev = os.stat(render_128).st_rdev
            os.mknod(render_129, stat.S_IFCHR | 0o660, dev)
            shutil.chown(render_129, "root", "video")
            print(f"[PATCH] Created {render_129} safely")
    except Exception as e:
        print(f"[PATCH] Skip device setup: {e}")

    try:
        dt_source = "/usr/src/immich_ml/device_tree"
        dt_target = "/proc/device-tree"
        if os.path.isdir(dt_source) and not os.path.islink(dt_target):
            if os.path.islink(dt_target):
                os.unlink(dt_target)
            os.symlink(dt_source, dt_target)
            print("[PATCH] Created device-tree symlink safely")
    except Exception as e:
        print(f"[PATCH] Device-tree link skipped: {e}")

_setup_dev_nodes()

def _patch_npu_address():
    so_file = "/opt/venv/lib/python3.11/site-packages/rknnlite/api/rknn_runtime.cpython-311-aarch64-linux-gnu.so"
    backup_file = so_file + ".orig"

    try:
        if not os.path.exists(so_file):
            return

        if not os.path.exists(backup_file):
            shutil.copy2(so_file, backup_file)
            print(f"[PATCH] Backup original .so to {backup_file}")

        with open(so_file, 'rb') as f:
            data = f.read()

        old = b'ffbc0000.npu'
        new = b'fdab0000.npu'

        if old not in data:
            print(f"[PATCH] NPU address already patched or not found")
            return

        patched = data.replace(old, new, 1)
        tmp = so_file + ".tmp"
        with open(tmp, 'wb') as f:
            f.write(patched)
        os.rename(tmp, so_file)
        print(f"[PATCH] Patched NPU address safely")
    except Exception as e:
        print(f"[PATCH] NPU patch skipped: {e}")

_patch_npu_address()

rknnpool_target = "/usr/src/immich_ml/sessions/rknn/rknnpool.py"
rknnpool_backup = rknnpool_target + ".orig"
rknnpool_patched = '''# This code is from leafqycc/rknn-multi-threaded
# Following Apache License 2.0

import logging
from concurrent.futures import Future, ThreadPoolExecutor
from pathlib import Path
from queue import Queue
from typing import Callable

import numpy as np
from numpy.typing import NDArray

from immich_ml.config import log
from immich_ml.models.constants import RKNN_COREMASK_SUPPORTED_SOCS, RKNN_SUPPORTED_SOCS


def get_soc(device_tree_path: Path | str) -> str | None:
    try:
        with Path(device_tree_path).open() as f:
            device_compatible_str = f.read()
            for soc in RKNN_SUPPORTED_SOCS:
                if soc in device_compatible_str:
                    return soc
            log.warning("Device is not supported for RKNN")
    except OSError as e:
        log.warning(f"Could not read {device_tree_path}. Reason: %s", e)
    return None


soc_name = None
is_available = False

def _detect_soc() -> str | None:
    soc = None
    for path in [
        "/sys/class/drm/card0/device/uevent",
        "/usr/src/immich_ml/device_uevent",
    ]:
        try:
            with open(path) as f:
                content = f.read()
            for candidate in ["rk3588", "rk3576", "rk3568", "rk3566"]:
                if candidate in content:
                    soc = candidate
                    break
        except OSError:
            pass
        if soc:
            break
    if not soc:
        soc = get_soc("/proc/device-tree/compatible")
    return soc

try:
    from rknnlite.api import RKNNLite
    soc_name = _detect_soc()
    is_available = soc_name is not None
    if is_available:
        log.info(f"RKNN detected: SOC={soc_name}")
    else:
        log.info("RKNN SDK installed but SOC not detected, will use CPU fallback")
except ImportError:
    log.debug("RKNN is not available")


def init_rknn(model_path: str) -> "RKNNLite":
    if not is_available:
        raise RuntimeError("rknn is not available!")
    rknn_lite = RKNNLite()
    rknn_lite.rknn_log.logger.setLevel(logging.ERROR)
    ret = rknn_lite.load_rknn(model_path)
    if ret != 0:
        raise RuntimeError("Failed to load RKNN model")

    if soc_name in RKNN_COREMASK_SUPPORTED_SOCS:
        ret = rknn_lite.init_runtime(core_mask=RKNNLite.NPU_CORE_AUTO)
    else:
        ret = rknn_lite.init_runtime()

    if ret != 0:
        raise RuntimeError("Failed to initialize RKNN runtime environment")

    return rknn_lite


class RknnPoolExecutor:
    def __init__(
        self,
        model_path: str,
        tpes: int,
        func: Callable[["RKNNLite", list[NDArray[np.float32]]], list[NDArray[np.float32]]],
    ) -> None:
        self.tpes = tpes
        self.queue: Queue[Future[list[NDArray[np.float32]]]] = Queue()
        self.rknn_pool = [init_rknn(model_path) for _ in range(tpes)]
        self.pool = ThreadPoolExecutor(max_workers=tpes)
        self.func = func
        self.num = 0

    def put(self, inputs: list[NDArray[np.float32]]) -> None:
        self.queue.put(self.pool.submit(self.func, self.rknn_pool[self.num % self.tpes], inputs))
        self.num += 1

    def get(self) -> list[NDArray[np.float32]] | None:
        if self.queue.empty():
            return None
        future = self.queue.get()
        return future.result()

    def shutdown(self) -> None:
        self.pool.shutdown(wait=True)
'''

try:
    ast.parse(rknnpool_patched)
except SyntaxError:
    raise SystemExit("[PATCH] rknnpool patched code has syntax error, aborting")

if os.path.exists(rknnpool_target) and not os.path.exists(rknnpool_backup):
    try:
        shutil.copy2(rknnpool_target, rknnpool_backup)
        print(f"[PATCH] Backup {rknnpool_target}")
    except Exception:
        pass

with open(rknnpool_target, "w") as f:
    f.write(rknnpool_patched)
print(f"[PATCH] Updated {rknnpool_target} safely")

base_target = "/usr/src/immich_ml/models/base.py"

try:
    with open(base_target) as f:
        c = f.read()

    if "_CLIP_TEXTUAL_CPU_FALLBACK" not in c:
        if not os.path.exists(base_target + ".orig"):
            shutil.copy2(base_target, base_target + ".orig")

        old_format = (
            '    @property\n'
            '    def _model_format_default(self) -> ModelFormat:\n'
            '        if rknn.is_available:\n'
            '            return ModelFormat.RKNN\n'
            '        elif immich_ml.sessions.ann.loader.is_available and settings.ann:\n'
            '            return ModelFormat.ARMNN\n'
            '        else:\n'
            '            return ModelFormat.ONNX'
        )
        new_format = (
            '    @property  # _CLIP_TEXTUAL_CPU_FALLBACK\n'
            '    def _model_format_default(self) -> ModelFormat:\n'
            '        if rknn.is_available:\n'
            '            # CLIP textual: CumSum op not supported in librknnrt.so, use ONNX (CPU)\n'
            '            if self.identity[1] == ModelTask.SEARCH and self.identity[0] == ModelType.TEXTUAL:\n'
            '                return ModelFormat.ONNX\n'
            '            return ModelFormat.RKNN\n'
            '        elif immich_ml.sessions.ann.loader.is_available and settings.ann:\n'
            '            return ModelFormat.ARMNN\n'
            '        else:\n'
            '            return ModelFormat.ONNX'
        )
        patched = c.replace(old_format, new_format)

        if patched == c:
            print("[PATCH] CLIP textual fallback: pattern not found, skipping")
        else:
            ast.parse(patched)
            with open(base_target, "w") as f:
                f.write(patched)
            print("[PATCH] CLIP textual model will use ONNX (CPU) instead of RKNN")
    else:
        print("[PATCH] CLIP textual fallback already applied")
except Exception as e:
    print(f"[PATCH] CLIP textual fallback skipped: {e}")

try:
    with open(base_target) as f:
        c = f.read()

    if "_backup_rknpu_ext" not in c:
        if not os.path.exists(base_target + ".orig"):
            shutil.copy2(base_target, base_target + ".orig")

        old_pattern = r'    def clear_cache\(self\) -> None:\n    def _make_session'
        new_code = '''    def clear_cache(self) -> None:
        self._backup_rknpu_ext()
        if self.cache_dir.exists():
            for item in self.cache_dir.iterdir():
                if item.name != 'rknpu':
                    if item.is_dir():
                        shutil.rmtree(item)
                    else:
                        item.unlink()
        self._restore_rknpu_ext()

    def _make_session'''
        patched = re.sub(old_pattern, new_code, c)
        ast.parse(patched)
        with open(base_target, "w") as f:
            f.write(patched)
        print("Patched base.py safely")
    else:
        print("base.py already patched")
except Exception as e:
    print(f"[PATCH] base.py patch skipped: {e}")

os.execvp("tini", ["tini", "--", "python", "-m", "immich_ml"])

---

四、部署命令（飞牛 NAS 直接用）

# 启动
docker-compose up -d

# 查看日志
docker logs -f immich_machine_learning

---

五、访问地址

NAS IP:2283

---

直接复制三个文件到同一目录，一键启动即可满血 NPU 推理。