近两年 Show HN 上反复出现一类工具,骨架都差不多:把 CLIP 这类图文对比模型搬到本地,给整个照片库和视频库建一个语义索引,然后用一句自然语言去找"去年夏天那条狗跳进湖里"的画面。它们的共同点是——模型在你自己机器上跑,素材不出硬盘。这篇教程就把这套骨架在 macOS 上完整搭一遍,图片和视频逐帧都能搜。
适用场景
适合照片和视频主要存在本地、又不想把家庭影像上传到云端的人。典型需求是:素材量在几万到几十万张(段)之间,想用自然语言而不是文件名、日期去找东西,比如"红色皮划艇""白板上的架构图""雨天的公交站"。视频部分尤其有价值——你不需要记得它出现在第几分钟,索引会按固定间隔抽帧,把每一帧都变成可检索的条目。
环境与前置条件
- 操作系统:macOS 12 及以上。Apple Silicon(M 系列芯片)可以用 MPS 加速,Intel 机器同样能跑,只是慢一些。
- 运行时:Python 3.10 及以上,具体小版本以官方文档当前版本为准;Homebrew 用于装系统级依赖。
- 其他工具:
ffmpeg/ffprobe,负责视频抽帧和读取时长。 - 内存:16GB 起步比较从容。向量检索那一步会把索引读进内存,十万条 512 维 float32 向量大约 200MB,模型本身也要占一部分。
- 磁盘:抽出来的帧是临时文件,跑完就删;真正长期占空间的是索引库,十万条量级在几百 MB。建议留出 10GB 以上余量,因为视频抽帧过程中会短暂堆积 JPEG。
- 网络:首次运行需要联网下载模型权重(几百 MB 量级),之后可以完全离线。
分步骤部署
第 1 步:装 Homebrew 与 ffmpeg
如果还没装 Homebrew,按官网首页给出的命令安装,以官方页面为准。然后:
```bash
brew install ffmpeg
ffmpeg -version
ffprobe -version
```
能看到版本信息输出即成功。ffprobe 是随 ffmpeg 一起装的。
第 2 步:建项目目录和虚拟环境
```bash
mkdir -p ~/clipsearch && cd ~/clipsearch
python3 -m venv .venv
source .venv/bin/activate
python -V
```
提示符前面出现 (.venv) 就对了。后面所有命令都在这个虚拟环境里执行。
第 3 步:安装 Python 依赖
```bash
pip install --upgrade pip
pip install torch open_clip_torch pillow pillow-heif numpy tqdm
```
torch:PyTorch,会自动带上 macOS 的 MPS 支持。open_clip_torch:CLIP 模型和预训练权重的加载器。pillow-heif:让 Pillow 能读 iPhone 拍的 HEIC 照片,这一步很容易被忽略,不加的话照片库里一大半文件会被跳过。
装完后验证一下 MPS 是否可用:
```bash
python -c "import torch; print(torch.backends.mps.is_available())"
```
输出 True 表示会走 GPU 加速。
第 4 步:给终端开完全磁盘访问权限
如果要直接索引 macOS「照片」App 的图库,原图在:
```
~/Pictures/Photos Library.photoslibrary/originals
```
先试着列一下目录:
```bash
ls ~/Pictures/Photos\ Library.photoslibrary/originals | head
```
如果报 Operation not permitted,去「系统设置 → 隐私与安全性 → 完全磁盘访问权限」,把你用的终端(Terminal 或 iTerm)加进去并打开开关,然后完全退出终端再重开。这一步不做,脚本会安静地读到零个文件。
不想动系统权限的话,另一个做法是先从照片 App 里导出到一个普通文件夹,只索引那个文件夹。
第 5 步:写索引脚本
新建 index.py:
```python
#!/usr/bin/env python3
"""扫描照片和视频,抽帧并写入 CLIP 向量索引。"""
import argparse, os, sqlite3, subprocess, sys, tempfile
from pathlib import Path
import numpy as np
import torch
import open_clip
from PIL import Image
from pillow_heif import register_heif_opener
from tqdm import tqdm
register_heif_opener()
IMAGE_EXT = {".jpg", ".jpeg", ".png", ".heic", ".webp", ".tif", ".tiff", ".bmp"}
VIDEO_EXT = {".mp4", ".mov", ".m4v", ".avi", ".mkv"}
具体模型名与权重名以 open_clip 官方文档为准
MODEL_NAME = os.environ.get("CLIP_MODEL", "ViT-B-32")
PRETRAINED = os.environ.get("CLIP_PRETRAINED", "openai")
def pick_device():
return "mps" if torch.backends.mps.is_available() else "cpu"
def load_model():
model, _, preprocess = open_clip.create_model_and_transforms(
MODEL_NAME, pretrained=PRETRAINED
)
model.eval().to(pick_device())
tokenizer = open_clip.get_tokenizer(MODEL_NAME)
return model, preprocess, tokenizer
def init_db(conn):
conn.execute("""
CREATE TABLE IF NOT EXISTS items(
id INTEGER PRIMARY KEY,
path TEXT NOT NULL,
kind TEXT NOT NULL,
ts REAL,
taken_at TEXT,
embedding BLOB NOT NULL)""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_path ON items(path)")
conn.commit()
def already_indexed(conn, path):
cur = conn.execute("SELECT 1 FROM items WHERE path=? LIMIT 1", (path,))
return cur.fetchone() is not None
@torch.no_grad()
def embed_images(model, preprocess, paths, batch_size=16):
chunks = []
for i in range(0, len(paths), batch_size):
tensors = []
for p in paths[i:i + batch_size]:
try:
img = Image.open(p).convert("RGB")
tensors.append(preprocess(img))
except Exception as e:
print(f"[skip] {p}: {e}", file=sys.stderr)
if not tensors:
continue
x = torch.stack(tensors).to(pick_device())
f = model.encode_image(x)
f = f / f.norm(dim=-1, keepdim=True)
chunks.append(f.cpu().numpy().astype("float32"))
return np.vstack(chunks) if chunks else None
def probe_duration(path):
out = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=nw=1:nk=1", str(path)],
capture_output=True, text=True)
try:
return float(out.stdout.strip())
except ValueError:
return None
def extract_frames(video, outdir, interval):
pattern = str(Path(outdir) / "%06d.jpg")
scale=512:-2 把长边压到 512,既省磁盘也够 CLIP 用
cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", "-i", str(video),
"-vf", f"fps=1/{interval},scale=512:-2", "-q:v", "4", pattern]
subprocess.run(cmd, check=True)
return sorted(Path(outdir).glob("*.jpg"))
def iter_files(roots):
wanted = IMAGE_EXT | VIDEO_EXT
for root in roots:
for p in Path(root).rglob("*"):
if p.is_file() and p.suffix.lower() in wanted:
yield p
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--db", default="index.db")
ap.add_argument("--interval", type=float, default=2.0,
help="视频抽帧间隔(秒),默认每 2 秒一帧")
ap.add_argument("--roots", nargs="+", required=True)
args = ap.parse_args()
model, preprocess, _ = load_model()
conn = sqlite3.connect(args.db)
init_db(conn)
files = list(iter_files(args.roots))
print(f"发现 {len(files)} 个候选文件")
for p in tqdm(files, desc="indexing"):
sp = str(p.resolve())
if already_indexed(conn, sp):
continue
if p.suffix.lower() in IMAGE_EXT:
vecs = embed_images(model, preprocess, [sp], 1)
if vecs is None:
continue
conn.execute(
"INSERT INTO items(path,kind,ts,taken_at,embedding) VALUES(?,?,?,?,?)",
(sp, "image", None, None, vecs[0].tobytes()))
else:
with tempfile.TemporaryDirectory() as td:
frames = extract_frames(sp, td, args.interval)
vecs = embed_images(model, preprocess, frames, 16)
if vecs is None:
continue
for i, v in enumerate(vecs, start=1):
t = (i - 1) * args.interval
conn.execute(
"INSERT INTO items(path,kind,ts,taken_at,embedding) "
"VALUES(?,?,?,?,?)",
(sp, "video_frame", t, None, v.tobytes()))
conn.commit()
conn.commit()
conn.close()
print("索引完成")
if __name__ == "__main__":
main()
```
几个关键设计说明:
- 视频逐帧靠 ffmpeg 的 fps 滤镜。
fps=1/2表示每 2 秒取一帧,输出文件从000001.jpg开始编号,所以第 i 帧对应的视频时间戳是(i-1) * interval。秒级定位足够你跳到那一瞬间。 - 帧先抽到临时目录,算完向量立刻删掉,磁盘不会被撑爆。
- embedding 以 float32 二进制存进 SQLite,检索时再读出来还原成矩阵。
- 重复运行是增量的,已经索引过的文件路径会被跳过。
第 6 步:跑索引
先拿一个小文件夹试水:
```bash
python index.py --roots ~/Pictures/test --interval 2
```
进度条走完,同目录下出现 index.db。确认没问题后,再指向整个照片库:
```bash
nohup python index.py \
--roots "$HOME/Pictures/Photos Library.photoslibrary/originals" \
--interval 2 > index.log 2>&1 &
tail -f index.log
```
大库会跑很久,具体时长取决于芯片和素材量,用你自己的机器实测为准。长视频嫌帧太多,把 --interval 调到 5 或 10,精度换速度。
第 7 步:写查询脚本
新建 search.py:
```python
#!/usr/bin/env python3
import argparse, sqlite3
import numpy as np
import torch
from index import load_model, pick_device
def main():
ap = argparse.ArgumentParser()
ap.add_argument("query")
ap.add_argument("--db", default="index.db")
ap.add_argument("--top", type=int, default=10)
args = ap.parse_args()
model, _, tokenizer = load_model()
with torch.no_grad():
tok = tokenizer([args.query]).to(pick_device())
q = model.encode_text(tok)
q = q / q.norm(dim=-1, keepdim=True)
q = q.cpu().numpy().astype("float32")[0]
conn = sqlite3.connect(args.db)
rows = conn.execute("SELECT path, kind, ts, embedding FROM items").fetchall()
if not rows:
print("索引为空,先运行 index.py")
return
mat = np.vstack([np.frombuffer(r[3], dtype="float32") for r in rows])
mat = mat / (np.linalg.norm(mat, axis=1, keepdims=True) + 1e-8)
scores = mat @ q
for i in np.argsort(-scores)[:args.top]:
path, kind, ts, _ = rows[i]
loc = f" 跳到 {ts:.1f} 秒" if ts is not None else ""
print(f"{scores[i]:.3f} {path}{loc}")
if __name__ == "__main__":
main()
```
用法:
```bash
python search.py "海边日落"
python search.py "白板上的流程图" --top 5
python search.py "有人在弹吉他"
```
输出每行是「相似度分数 + 文件路径」,视频帧还会带上时间戳。中英文都可以试,用哪个模型、对中文效果如何,以模型官方说明为准;如果中文检索明显不准,换一个中文语料训练的 CLIP 变体重新建索引即可。
想更直观的话,可以在结果里加上 open -R <path> 直接在访达里定位文件。
验证部署是否成功
三步验证,从粗到细:
1. 查条目数
```bash
sqlite3 index.db "SELECT kind, COUNT(*) FROM items GROUP BY kind;"
```
预期输出两行,image 和 video_frame 各有一个数字。image 的数量应该和你的图片总数接近;video_frame 大致等于所有视频总时长除以抽帧间隔。
2. 用已知答案的词查
挑一张你明确记得内容的照片,比如有一张黑猫趴在键盘上的照片,查:
```bash
python search.py "猫趴在键盘上" --top 5
```
预期那张照片出现在前几条里,相似度分数明显高于后面的条目。
3. 断网验证隐私
```bash
HF_HUB_OFFLINE=1 python search.py "红色皮划艇"
```
把 Wi-Fi 关掉再跑一次同样可以。这条命令强制 Hugging Face 库走离线模式,如果还能正常出结果,说明推理链路完全不依赖网络。
常见报错与解决
报错:ModuleNotFoundError: No module named 'torch' 或 ffmpeg: command not found
→ 原因:虚拟环境没激活,或者系统里没装 ffmpeg。
→ 解决:
```bash
source ~/clipsearch/.venv/bin/activate
brew install ffmpeg
```
报错:OSError: [Errno 1] Operation not permitted: '.../Photos Library.photoslibrary/...'
→ 原因:macOS 的隐私保护拦住了终端对照片图库的读取。
→ 解决:系统设置 → 隐私与安全性 → 完全磁盘访问权限,勾选你的终端,然后彻底退出终端重新打开。用 ls 确认能列出文件后,删掉 index.db 重跑索引。
报错:RuntimeError: Placeholder storage has not been allocated on MPS device 或某个算子在 MPS 上未实现
→ 原因:PyTorch 的 MPS 后端还没覆盖全部算子。
→ 解决:加上回退开关,让不支持的算子自动落到 CPU:
```bash
PYTORCH_ENABLE_MPS_FALLBACK=1 python index.py --roots ~/Pictures/test
```
嫌麻烦就直接指定 CPU 跑:python -c "import torch" 前先设 export PYTORCH_ENABLE_MPS_FALLBACK=1,写进 ~/.zshrc 一劳永逸(会略微变慢)。
报错:UnidentifiedImageError: cannot identify image file,或者统计出来图片数量远少于实际
→ 原因:iPhone 的 HEIC/HEIF 格式 Pillow 默认读不了。
→ 解决:
```bash
pip install pillow-heif
python -c "from pillow_heif import register_heif_opener; register_heif_opener(); print('ok')"
```
脚本里已经有 register_heif_opener(),装完重跑即可。
报错:进程被系统杀掉,或日志里出现 Killed
→ 原因:批量太大或图片分辨率太高,统一内存被打满。
→ 解决:把 embed_images 的 batch_size 从 16 降到 4,并把抽帧的 scale=512:-2 改成 scale=384:-2;同时把抽帧间隔调大,减少条目数。
报错:sqlite3.OperationalError: database is locked
→ 原因:后台索引进程还没结束,你又启动了查询或第二个索引进程。
→ 解决:tail -f index.log 确认索引跑完再查询;需要中断时用 kill <PID>,然后删除 index.db-journal 之类的残留文件再重试。
后续维护
增量与重建。日常新增素材,直接重跑 index.py,已有路径会跳过。注意两点:一是移动或重命名过的文件会被当成新文件重复索引,可以定期用 sqlite3 index.db "DELETE FROM items WHERE path NOT IN (...)" 之类的方式清理,或者干脆重建;二是换了模型或权重就必须重建整个索引,因为旧向量和新模型的向量空间不可比,混在一起检索结果会毫无意义。重建前先把旧库改名备份。
备份。要备份的就两样:index.db 和脚本。索引库是单个文件,直接复制即可,但要先停掉写入进程,否则拷到的是不一致的快照。停止后再 cp index.db index.db.bak,或者用 sqlite3 index.db ".backup backup.db",这个命令支持在运行中安全备份。
升级。pip install -U torch open_clip_torch 升级依赖后,先拿小文件夹跑一遍冒烟测试;ffmpeg 用 brew upgrade ffmpeg 升级。升级后如果感觉抽帧时间戳对不上,检查 ffmpeg 的 fps 滤镜行为是否有变化,必要时把时间戳改成用 ffprobe 逐帧读取。
日志与监控。长任务用 nohup ... > index.log 2>&1 & 跑,用 tail -f index.log 盯进度,grep -c "^\[skip\]" index.log 能统计跳过了多少读不出来的文件——这个数字突然变大通常意味着某类格式没被支持。磁盘方面,df -h 看剩余空间,索引库和临时帧目录是两个主要的增长点,记得给 TMPDIR 留足空间。
隐私。整条链路是本地推理,模型权重从 Hugging Face 下载一次之后可以完全离线,前面已经用 HF_HUB_OFFLINE=1 验证过。有一点要清楚:index.db 里存的是你所有素材的完整路径和画面特征向量,它本身就是一个敏感文件,别随手丢进网盘同步目录或者云备份里。想再稳一点,把项目目录放在 FileVault 加密的卷上,并把 HF_HOME 指向本地目录避免缓存散落各处。
