1. 为什么我要把 ultralytics.data.loaders 翻个底朝天如果你用过 YOLOv8 或者 YOLO11大概率每天都会和ultralytics这个库打交道。训练、验证、导出、推理一条龙下来非常丝滑。但大多数人的使用习惯是“调包侠”模式——写个三五行代码把YOLO(yolov8n.pt)一实例化剩下的事情全部交给框架处理。我之前也一直是这么干的直到有一次需要自定义数据加载流程需求非常刁钻训练数据不是标准的图片文件夹而是一堆分布在不同 NAS 路径下的视频帧而且每个样本都有额外的标签元信息需要和 YOLO 格式的 txt 标签做对齐。我试了 Ultralytics 官方文档里提到的各种数据组织方式都不顺手。没办法只能打开源码硬啃。啃来啃去发现整个数据链路里最核心、最绕不开的就是ultralytics/data/loaders.py这个文件。ultralytics.data.loaders这个模块承载了从“数据源”到“batch 张量”之间的所有转换逻辑是 YOLO 系列训练和推理的数据入口。这篇文章就是我对loaders.py子模块的一次完整代码详读记录。我会把这个文件里涉及的几个核心类、加载流程、底层逻辑、常见坑全部拆开来讲并且结合我自己在改数据流时遇到的实际问题给出可复现的经验。适合的人群有两类一类是想彻底搞懂 Ultralytics 数据流水线的开发者你不想只做调包侠想弄明白“图片到底是怎么从磁盘变成模型输入的三维张量”的。另一类是和我一样需要在 YOLO 框架上做数据层深度定制的工程师你需要替换、扩展或魔改 loader这篇文章可以帮你少走很多弯路。先说明我阅读的版本是当前较新的 ultralytics 8.x 系列源码整体结构在 8.0 到 8.3 之间变动不大核心设计稳定。后续代码行号可能随版本略有偏移但类名、方法名和关键逻辑是通用的。2. loaders.py 在整个 ultralytics 数据流中的位置2.1 数据模块的层级关系在给代码逐行分析之前我觉得非常有必要先把loaders.py在整体项目中的“生态位”讲清楚。Ultralytics 的数据相关代码主要集中在ultralytics/data/目录下这个目录里的文件各司其职dataset.py定义了YOLODataset和ClassificationDataset等核心数据集类负责从标注文件label构建样本索引并在__getitem__时返回“图像 标签”的完整样本。build.py负责把数据集配置yaml 文件解析成实际的 dataset 实例相当于一个工厂函数集合。build_dataloader和build_yolo_dataset都在这里。dataloader.py定义了DataLoader的子类主要是在 PyTorch 原版 DataLoader 基础上增加了一些分布式采样、worker 进程管理的细节。loaders.py这个文件是今天的主角。它定义了若干Source相关的类比如LoadImages、LoadScreenshots、LoadStreams、LoadPilAndNumpy、LoadTensor等等。用一句话概括loaders.py的作用它把一个“数据源描述”转变成一个可迭代的、产出单帧图像张量的流。这里的数据源描述可以是一张图片路径、一个文件夹路径、一个视频文件路径、一个摄像头设备号、一组 PIL Image / ndarray 对象、甚至是一个已经存在于内存中的 tensor。只要是你能想到的图像输入方式这个模块基本都覆盖了。2.2 一个直观的数据流比喻你可以把整个 YOLO 数据链路想象成一条“自来水管道”数据源是“水源”可能是水库文件夹、河流视频流、水龙头摄像头等等。loaders.py是“取水泵站”负责把不同来源的水图像帧抽出来统一变成标准的“水”BGR 格式、特定尺寸、特定 dtype 的 numpy 数组。dataset.py是“水处理厂”在取水泵站的基础上再叠加 resize、归一化、增强、标签编码等处理。dataloader.py是“供水调度中心”控制多线程、多进程并行抽水和处理。模型训练/推理是“用户端水龙头”拿到的已经是干净可用的“饮用水”了。所以如果你想修改从“水源”到“标准图像”之间的逻辑就应该去改loaders.py如果你想改“标准图像到最终训练样本”的逻辑那重点在dataset.py。很多人分不清这个边界导致想自定义数据增强结果却去改了 loader当然事倍功半。2.3 为什么说 loaders.py 是数据链路的“前置必修课”我个人的切身体会如果你跳过loaders.py直接去看dataset.py你会觉得dataset.py里到处都是魔法。self.source是什么im 是从哪里来的为什么有时候im是一个 numpy 数组有时候又变成了PIL.Image这些问题的答案全部要回溯到loaders.py。换句话说loaders.py定义了整个数据流的第一公里。第一公里跑不顺后面所有逻辑都会出问题。而且在实际工程中很多人遇到的“训练时图片加载慢”“视频流推理卡顿”“内存暴涨”“多进程死锁”等问题根因都在loaders.py这一层。所以理解它不只是为了“看懂源码”更是为了排查和优化真实问题。3. LoadImages最常用的图像与视频加载器LoadImages是你在训练、验证、推理中最常见到的类。当你在命令行里写yolo predict source./images或yolo train dataxxx.yaml时如果数据源是文件夹或视频底层基本都会走这个类。3.1 核心属性与初始化逻辑看代码先看__init__。LoadImages的初始化方法核心作用是把path解析成具体的文件列表并判断类型。class LoadImages: def __init__(self, path, batch1, vid_stride1): files [] for p in sorted(path) if isinstance(path, (list, tuple)) else [path]: p str(Path(p).resolve()) if p.isdir(): # 遍历目录筛选出图片和视频文件 files.extend(glob.glob(str(p / ** / *.*), recursiveTrue)) elif p.endswith(.txt): # txt 文件里每行是一个文件路径 files.extend(Path(p).read_text().splitlines()) else: files.append(p) # 过滤出受支持的图片和视频格式 images [x for x in files if x.split(.)[-1].lower() in IMG_FORMATS] videos [x for x in files if x.split(.)[-1].lower() in VID_FORMATS] self.ni len(images) self.nv len(videos) ...这里有几个关键设计意图支持传入 list/tuple 路径这意味着你可以一次性传入多个图片路径列表或者多个文件夹路径列表loader 内部会将它们合并。实际项目里我经常用它来同时加载多个不同相机拍的数据集。.txt文件路径列表支持你可以准备一个 txt 文件里面每行写一个图片的绝对路径loader 会按行读取。这在处理大规模分布式存储数据集时非常实用不需要把所有文件拷到同一个目录。区分图片和视频图片和视频分别计数self.ni表示图片数量self.nv表示视频数量self.files属性里会同时存图片和视频路径。遍历时优先输出图片帧图片全部输出完后再输出视频帧。3.2__iter__与__next__的协作机制LoadImages实现了迭代器协议。在 PyTorch 的 DataLoader 中当num_workers 0时每个 worker 进程会通过迭代器不断拉取数据。其中__iter__方法做了大量初始化工作def __iter__(self): self.count 0 self.video_flag [False] * self.ni self.video_cap [None] * self.ni self.mode image for i in range(self.ni): self.video_flag[i] self.files[i].split(.)[-1].lower() in VID_FORMATS for i in range(self.ni, self.ni self.nv): self.video_flag.append(True) ... return self这段代码的巧妙之处在于video_flag的前ni个元素对应self.files前ni个图片文件值全为False后nv个元素对应视频文件值全为True。这样在__next__中就可以通过self.video_flag[self.count]快速判断当前应该走“读图片”分支还是“读视频帧”分支。__next__的核心逻辑是def __next__(self): if self.count self.nf: # 所有文件都遍历完关闭视频并抛出 StopIteration raise StopIteration path self.files[self.count] if self.video_flag[self.count]: # 视频分支 self.mode video ret_val, im0 self.video_cap[self.count].read() if not ret_val: # 当前视频读完了切到下一个文件 self.count 1 return self.__next__() ... else: # 图片分支 im0 cv2.imread(path) if im0 is None: LOGGER.warning(fWarning: 图片读取失败跳过: {path}) self.count 1 return self.__next__() ... return self.mode, im0, self.video_cap[self.count], path, ...注意这个递归调用return self.__next__()它的存在是为了自动跳过损坏的图片或已经播放结束的视频让迭代器永远“向前看”直到找到有效的下一帧。这种设计非常实用但初学者容易忽略如果你在自定义 loader 中不处理“读取失败”的情况很容易让整个数据流卡死。3.3vid_stride参数的底层逻辑很多人在使用predict或val时见过vid_stride这个参数但不一定清楚它的底层实现。它的含义是“视频抽帧步长”例如vid_stride2表示每 2 帧取 1 帧。在LoadImages中这个参数体现在视频分支的处理上# 在 __init__ 中记录 self.vid_stride vid_stride ... # 在 __next__ 视频分支中 if self.video_cap[self.count].get(cv2.CAP_PROP_POS_FRAMES) % self.vid_stride: # 不是目标帧则跳过 self.count 1 # 或者是直接读取下一帧视版本而定更近的版本中为了让视频读取更高效会结合cv2.CAP_PROP_POS_FRAMES判断当前读取的帧位置如果当前位置 % vid_stride ! 0就继续读下一帧而不返回。从使用者的角度理解vid_stride越大视频处理速度越快但会损失时间维度的连续性。在跟踪任务里如果vid_stride设置过大目标会出现明显的跳变导致跟踪 ID 频繁切换。我实测下来跟踪任务建议vid_stride不超过 5如果单纯做目标检测且视频较长vid_stride2或3是性价比很高的选择。3.4 实操心得如何给 LoadImages 增加“视频循环播放”能力默认情况下LoadImages遍历完最后一个视频帧后就会停止。但我在做视频流模型稳定性测试时经常希望同一个视频能循环播放模拟长时间运行的摄像头。这个需求不需要改框架只需要在调用时包一层“循环迭代器”def cycle_video_loader(source, vid_stride2): while True: loader LoadImages(source, vid_stridevid_stride) for *_, im, *_ in loader: yield im这里用while True不断创建新的LoadImages实例每次迭代完一个完整视频后重新开始。实测在长时间稳定性测试中这种循环方式非常稳定而且不会产生内存泄漏。注意每次创建新的 loader 会重新打开视频文件如果视频很大建议在循环外部做一次cv2.VideoCapture的复用但那样会牺牲代码简洁性。根据场景取舍。4. LoadStreams视频流与摄像头的并发读取方案LoadStreams比LoadImages复杂不少因为它的数据源是“流”——既可以是 RTSP/RTMP 网络流也可以是本地摄像头设备号。这类数据源的最大特点是帧率不稳定、网络延迟不可控、读取时间不均匀。如果直接在__next__里同步读帧遇到网络卡顿时会拖垮整个推理管线。4.1 多线程缓存的核心设计LoadStreams的初始化逻辑中有几个关键点class LoadStreams: def __init__(self, sourcesfile.streams, batch1, vid_stride1, bufferFalse): ... self.buffer buffer self.mode stream self.sources [x.strip() for x in sources.split(,)] # 支持逗号分隔多个流 self.vid_stride vid_stride self.caps [cv2.VideoCapture(x) for x in self.sources] ... # 为每个流创建独立的读取线程 self.threads [Thread(targetself.update, args(i,), daemonTrue) for i in range(n)] for t in self.threads: t.start()这里最核心的设计是每个视频流独立一个线程。每个线程调用update(i)方法反复从对应的VideoCapture中读取新帧并存入一个长度为 1 的列表self.imgs[i]中。def update(self, i): while True: if self.caps[i].isOpened(): success, im self.caps[i].read() if success: self.imgs[i] im self.frames[i] 1 else: # 读取失败时尝试重连 self.caps[i].open(self.sources[i]) time.sleep(0.01)为什么用列表而不是用一个普通变量因为 Python 的 GIL 不会保证多线程下普通变量的可见性但列表元素的赋值是原子操作读线程可以安全地读取最新帧。这是非常典型的“多线程生产者-消费者”模式生产者是update线程消费者是__next__中的主线程。4.2 为什么缓冲区不是越大越好很多人在看源码时会问为什么self.imgs[i]只保存一帧如果把缓冲区设置得大一些比如保存最近 10 帧是不是推理更平滑答案是在实时流处理场景下缓冲区越大数据延迟越大。假设摄像头帧率为 30 FPS缓冲区为 10 帧那你获取到的画面就有 10/30 ≈ 0.33 秒的延迟。对于实时监控或自动驾驶场景这个延迟是不可接受的。因此 Ultralytics 默认只缓存最新的一帧保证拿到的是“尽可能实时”的数据。我自己的经验是如果视频源帧率较低比如 5 FPS但推理速度很快可以在update循环里加一个time.sleep控制读取频率避免无意义地占用 CPU。例如def update(self, i): while True: ... if success: self.imgs[i] im time.sleep(1 / self.target_fps) # 按目标帧率读取4.3 多路视频流的内存与报错处理LoadStreams默认支持多个视频流同时输入比如一个 4 路摄像头的监控系统。但这里有个隐藏问题self.imgs是列表每个元素是一帧图像如果摄像头分辨率很高如 4K3840x2160x3一路视频流一帧就有约 25 MB 内存。4 路就是 100 MB。如果每路都缓存多帧内存压力会非常大。因此官方注释里专门提到要让视频流 buffer 较大时请显式传入 bufferTrue。在默认bufferFalse时只缓存最新一帧这是内存占用最小的方案。我在实际部署时还遇到过一个问题某些 RTSP 流会在凌晨自动断开因为摄像头端做 NTP 校时或网络波动。LoadStreams的update里已经做了重连逻辑if not success: self.caps[i].open(self.sources[i])但重连频率没有限制。如果网络一直不通这里会形成“读失败 - open - 再读失败 - 再 open”的高频死循环CPU 占用直接飙升。我踩过这个坑后会在外部包装一层重试计数class RetryLoadStreams(LoadStreams): def update(self, i): reconnections 0 while True: ... if not success: reconnections 1 if reconnections 10: LOGGER.warning(f流 {i} 多次重连失败等待 30 秒) time.sleep(30) self.caps[i].open(self.sources[i]) else: reconnections 0这样的重连机制在长稳测试中表现良好推荐参考。5. LoadPilAndNumpy 与 LoadTensor打通内存数据输入的最后一公里这两个类通常被忽略但在实际工程中非常有用尤其是在你已经在内存里有了图像数据、不想通过磁盘 IO 绕一圈的时候。5.1 LoadPilAndNumpy 的使用场景LoadPilAndNumpy支持传入一个包含PIL.Image或numpy.ndarray的迭代对象。它内部会逐个检查每个元素的类型def __init__(self, imgs): if not isinstance(imgs, (list, tuple)): imgs [imgs] self.imgs imgs self.mode image self.ni len(imgs) def __next__(self): im self.imgs[self.count] if isinstance(im, PIL.Image.Image): im np.asarray(im)[..., ::-1] # PIL 是 RGB转成 BGR elif isinstance(im, np.ndarray): im im[..., ::-1] # 假设输入是 RGB转成 BGR ...这里有个非常重要的细节Ultralytics 内部约定图像通道顺序是 BGR因为底层大量使用 OpenCV。而 PIL.Image 读取的图像是 RGB 顺序因此需要通过np.asarray(im)[..., ::-1]做通道反转。如果你自己构造 PIL 图像时没有注意这一点输入到模型后颜色会偏蓝偏红一团糟而且检查时还很难定位问题。5.2 LoadTensor 与 GPU 内存输入LoadTensor则更进一步它接受一个形状为(N, C, H, W)的 PyTorch Tensor并且可以自动处理设备迁移def __init__(self, imgs): self.imgs imgs self.mode tensor self.ni len(imgs) if isinstance(imgs, (list, tuple)) else 1 def __next__(self): im self.imgs[self.count] if isinstance(im, torch.Tensor): im im.cpu().numpy().transpose(1, 2, 0) # CHW - HWC ...我一般会在两个场景下使用LoadTensor推理服务里已有一批预处理后的图像 tensor需要直接送入 YOLO 的 predict 流程希望省掉重新解码的过程。做数据增强对比实验时希望把同一批 tensor 输入多个不同的 pipeline。需要注意的是如果 tensor 在 GPU 上LoadTensor默认会调用.cpu()这会带来一次 GPU-CPU 的拷贝。如果数据非常大且频繁调用性能瓶颈可能从“读取图像”变成“拷贝数据”。更优的做法是直接构造一个不拷贝的 loader或者直接调用模型.predict方法并传入已预处理好的 tensor。这一点在官方文档里没有明说是我实测对比后发现的。5.3 我为什么通常建议直接用 LoadTensor 做推理优化在实时视频分析服务中性能优化往往见缝插针。我做过一版“视频抽帧 预处理 推理”的服务最初每次都把帧写到磁盘再读吞吐量在 20 FPS 左右后来改成直接用LoadTensor从内存传入吞吐量提升到了 35 FPS。优化效果主要来自减少了磁盘 IO 和图像解码的重复操作。但这不代表LoadTensor是万能的。它只负责“输入格式转换”不包含缩放、归一化等预处理真正的预处理还是要在别处做。所以我的建议是如果你只是想快速测试用LoadImages最省心如果你想做服务化部署且要求低延迟熟悉一下LoadTensor和LoadPilAndNumpy会很有帮助。6. 源码中的底层细节图像预处理与迭代器协议6.1 BGR、letterbox 与数据格式约定在 loader 类中图像从数据源读到以后普遍都会经过一个转换步骤保证输出是 HWC 格式的 BGR 图像。以LoadImages为例im0 cv2.imread(path) # 已经是 BGRHWC而LoadPilAndNumpy中则要做通道反转因为输入可能是 RGB。所有 loader 最终产出的图像在进入后续流程时都统一是BGR、HWC、uint8、0-255范围内的numpy.ndarray。这个约定是整个 Ultralytics 数据流的基石。紧接着的letterbox操作一般发生在dataset.py的load_image或__getitem__中但 loader 层有时也会触发会把图像缩放并填充到固定的正方形尺寸例如 640x640。选择 letterbox 而不是直接 resize 的原因非常直观直接拉伸会改变目标的长宽比导致检测框坐标失真。letterbox 用灰色边框填充多余区域保持图像内容不失真对检测精度影响最小。我在做自定义 loader 时会刻意保持这个约定不变只改“图像来源”部分。一旦破坏了 BGR/HWC 的约定后面很多内置增强都会出问题。6.2__len__与迭代结束条件LoadImages实现了__len__方法def __len__(self): return self.nf # 图片数量 视频总帧数对于视频它的帧数计算方式是self.nf sum(int(cv2.VideoCapture(f).get(cv2.CAP_PROP_FRAME_COUNT)) for f in videos) ni注意这里每个视频都会重新打开一次VideoCapture来计算总帧数。当视频文件很多比如上千个短视频时初始化阶段会非常慢因为每个视频都要打开、读取元信息、关闭。我处理过一个包含 2000 个短视频的数据集仅初始化就花了将近 20 秒。优化办法是提前用一个全局字典缓存视频路径对应的帧数避免重复打开。6.3 迭代器协议中的StopIteration所有 loader 在迭代完所有数据后都会通过raise StopIteration结束。但在 PyTorch DataLoader 的多进程环境下如果StopIteration抛出时机不对会导致 worker 进程异常。Ultralytics 内部通过build.py中的_InfiniteDataLoader等类处理了这个问题——它会在迭代结束时自动重新初始化实现无限循环采样。我自己写自定义 loader 时也遇到过类似问题如果在__iter__里不重置self.count多轮 epoch 训练时数据会从上次结束的位置继续导致每个 epoch 数据分布不一致。这一点务必注意__iter__必须重置所有状态变量。7. 大规模数据加载时的性能优化与踩坑总结7.1 多进程 worker 与 loaders 的交互在训练 YOLO 模型时数据加载通常开了多个 worker 进程。PyTorch 的 DataLoader 会将 dataset 复制到每个 worker 中每个 worker 独立调用 dataset 的__getitem__。而loaders.py通常被 dataset 内部引用。这里有一个经典陷阱当 num_workers 0 时每个 worker 会各自持有一份 loader 的副本。例如使用LoadImages加载视频时每个 worker 都会打开同一个视频文件并各自从第 0 帧开始读取。如果你的数据并行策略需要所有 worker 看到不同片段就需要额外的分片逻辑。Ultralytics 在build.py的build_dataloader里已经处理了这个问题但如果你自定义 loader 套进自己的训练脚本很容易踩到“数据重复”的坑。我的建议是自定义 loader 时尽量让 loader 本身是无状态的或者只在主进程初始化一次通过worker_init_fn传入子进程。这样能避免很多多进程下的诡异行为。7.2 视频流加载的内存泄漏排查我在部署实时多路视频流检测服务时遇到过内存持续上涨的情况。排查过程非常痛苦最后定位到问题在LoadStreams中如果摄像头断流后self.caps[i].open(self.sources[i])返回失败但并没有释放旧资源cv2.VideoCapture对象会持有旧的底层缓冲反复重连就会导致内存缓慢增加。解决方法是在重连前显式调用self.caps[i].release()并重新创建VideoCapture对象。我在自己的分支里做了如下修改if not success: self.caps[i].release() self.caps[i] cv2.VideoCapture(self.sources[i])经过这个修改后内存曲线变得平稳。如果你在生产环境中也遇到类似问题可以优先检查这一块。7.3 图像读取失败时的静默跳过机制LoadImages在处理图片时如果cv2.imread返回None会打印一个警告并跳过。这个机制很实用但也藏着一个隐患如果一张图片因为文件名后缀被识别为图片但实际内容已损坏cv2.imread返回None后代码会递归调用__next__跳到下一张。整个过程对调用方来说是无感的但如果坏图片占比较高你会发现自己“数据量”比预期少了很多。建议在使用前先对数据集做一次完整性检查import cv2 from pathlib import Path def check_images_ok(folder): bad [] for p in Path(folder).rglob(*.*): if p.suffix.lower() in {.jpg, .jpeg, .png, .bmp}: im cv2.imread(str(p)) if im is None: bad.append(str(p)) return bad坏图比例超过 0.1% 时需要认真对待建议直接剔除而不是依赖 loader 的跳过机制。7.4 从 loader 到数据增强的无缝衔接很多读者会问既然 loader 已经输出了图像为什么不直接在loaders.py里做数据增强原因是职责分离。loaders.py的定位是“将数据源变成标准图像”不应该包含随机翻转、马赛克增强等训练相关操作。这些操作放在dataset.py中因为它们在每个 epoch 都是变化的而 loader 的产出应该是稳定可复现的。我在源码中对比后确认Ultralytics 对这一设计保持了高度一致。所以你如果想扩展一个新的“数据源”只需要继承或模仿loaders.py中的类输出标准图像即可后续的dataset.py和增强流程都能无缝衔接不用改任何其他模块。8. 如何快速验证你自定义的 loader 是否合格8.1 三步验证法我自己在写完自定义 loader 后通常会用下面三步验证它的正确性第一步检查输出格式。写一个简单的循环打印每次迭代返回的mode、图像shape、dtype、路径等字段。确保图像是HWC、BGR、uint8。loader MyCustomLoader(...) for mode, im, cap, path, *extra in loader: print(mode, im.shape, im.dtype, path) break第二步检查迭代完整性。统计len(loader)与实际迭代次数必须完全一致。如果不一致说明你的 loader 在某个分支里提前终止或死循环。第三步接入 YOLO 预测流程。把 loader 传给model.predict(sourceloader)跑一个最小测试观察预测结果是否符合预期。这一步能暴露格式约定上的隐性错误。8.2 常见失败模式对照现象可能原因排查方法推理结果颜色异常通道顺序不是 BGR检查图像转换逻辑确认有没有做::-1反转训练到一半卡死多进程下 loader 竞争同一个文件句柄减少 worker 数量或在 loader 中加入进程锁视频流推理延迟越来越高缓冲区不断堆积确认self.imgs[i]只用最新帧不要累积历史初始化很久大量视频被重复打开计算帧数使用帧数缓存机制内存持续上涨重连时未释放旧的 VideoCapture在重新 open 前显式 release部分图像缺失cv2.imread返回 None 后被跳过检测文件完整性或调整路径8.3 保留一个最小可复现模板为了方便后续开发我长期维护一个自定义 loader 的最小模板from ultralytics.data.loaders import LoadImages class MyCustomLoader(LoadImages): def __init__(self, path, batch1, vid_stride1): super().__init__(path, batch, vid_stride) # 这里做你的额外初始化 def __next__(self): # 调用父类拿到原始结果 result super().__next__() mode, im, cap, path, *extra result # 对 im 做自定义处理 # 自定义逻辑... return mode, im, cap, path, *extra保持“继承 重写__next__”的方式可以复用大部分逻辑改动最小也最容易 review。9. 结合 ultralytics 最新版本的变化随着 ultralytics 版本更新loaders.py也在不断优化。我看过 8.0.x 到 8.3.x 的演进主要变化有几点早期版本里LoadImages和LoadStreams都会在__init__中大量使用os.path处理路径新版本逐渐迁移到pathlib.Path对 Windows 路径的兼容性更好。视频帧读取逐渐引入vid_stride更高效的控制逻辑避免逐帧读但逐帧丢弃。LoadStreams的线程更新函数在较新版本中加入了self.auto自动判断是否需要重连逻辑更稳健。每个 loader 类都增加了更清晰的关键字参数说明方便 IDE 提示。这些变化本质上是工程细节的完善核心架构并没有大改。所以你现在读这份源码分析就算一段时间后再升级版本依然能顺畅迁移。在使用 ultralytics 做推理时如果传递的source是整数会被判定为摄像头设备号进去LoadStreams分支如果传递字符串路径但带.txt后缀会走文件列表分支。这些规则全部集中在build.py的load_inference_source函数中。如果你之后想深入可以按这个顺序继续读源码ultralytics/data/build.py中的load_inference_source看它如何把source分发给不同的 loader。ultralytics/data/dataset.py中的YOLODataset看 loader 的输出如何被进一步处理成训练样本。ultralytics/data/augment.py看数据增强如何基于标准图像做变换。读完这三个文件整个数据链路基本就完全打通了。我自己的体会是源码阅读最容易犯的错误是“只看类名不看调用关系”。loaders.py里的类再多最终都要被build.py或dataset.py调用。先理清调用顺序再深入每个类的内部逻辑效率会高很多。希望这篇文章能让你少走我踩过的弯路。 YOLO数据流水线核心：ultralytics.data.loaders源码深度解析
1. 为什么我要把 ultralytics.data.loaders 翻个底朝天如果你用过 YOLOv8 或者 YOLO11大概率每天都会和ultralytics这个库打交道。训练、验证、导出、推理一条龙下来非常丝滑。但大多数人的使用习惯是“调包侠”模式——写个三五行代码把YOLO(yolov8n.pt)一实例化剩下的事情全部交给框架处理。我之前也一直是这么干的直到有一次需要自定义数据加载流程需求非常刁钻训练数据不是标准的图片文件夹而是一堆分布在不同 NAS 路径下的视频帧而且每个样本都有额外的标签元信息需要和 YOLO 格式的 txt 标签做对齐。我试了 Ultralytics 官方文档里提到的各种数据组织方式都不顺手。没办法只能打开源码硬啃。啃来啃去发现整个数据链路里最核心、最绕不开的就是ultralytics/data/loaders.py这个文件。ultralytics.data.loaders这个模块承载了从“数据源”到“batch 张量”之间的所有转换逻辑是 YOLO 系列训练和推理的数据入口。这篇文章就是我对loaders.py子模块的一次完整代码详读记录。我会把这个文件里涉及的几个核心类、加载流程、底层逻辑、常见坑全部拆开来讲并且结合我自己在改数据流时遇到的实际问题给出可复现的经验。适合的人群有两类一类是想彻底搞懂 Ultralytics 数据流水线的开发者你不想只做调包侠想弄明白“图片到底是怎么从磁盘变成模型输入的三维张量”的。另一类是和我一样需要在 YOLO 框架上做数据层深度定制的工程师你需要替换、扩展或魔改 loader这篇文章可以帮你少走很多弯路。先说明我阅读的版本是当前较新的 ultralytics 8.x 系列源码整体结构在 8.0 到 8.3 之间变动不大核心设计稳定。后续代码行号可能随版本略有偏移但类名、方法名和关键逻辑是通用的。2. loaders.py 在整个 ultralytics 数据流中的位置2.1 数据模块的层级关系在给代码逐行分析之前我觉得非常有必要先把loaders.py在整体项目中的“生态位”讲清楚。Ultralytics 的数据相关代码主要集中在ultralytics/data/目录下这个目录里的文件各司其职dataset.py定义了YOLODataset和ClassificationDataset等核心数据集类负责从标注文件label构建样本索引并在__getitem__时返回“图像 标签”的完整样本。build.py负责把数据集配置yaml 文件解析成实际的 dataset 实例相当于一个工厂函数集合。build_dataloader和build_yolo_dataset都在这里。dataloader.py定义了DataLoader的子类主要是在 PyTorch 原版 DataLoader 基础上增加了一些分布式采样、worker 进程管理的细节。loaders.py这个文件是今天的主角。它定义了若干Source相关的类比如LoadImages、LoadScreenshots、LoadStreams、LoadPilAndNumpy、LoadTensor等等。用一句话概括loaders.py的作用它把一个“数据源描述”转变成一个可迭代的、产出单帧图像张量的流。这里的数据源描述可以是一张图片路径、一个文件夹路径、一个视频文件路径、一个摄像头设备号、一组 PIL Image / ndarray 对象、甚至是一个已经存在于内存中的 tensor。只要是你能想到的图像输入方式这个模块基本都覆盖了。2.2 一个直观的数据流比喻你可以把整个 YOLO 数据链路想象成一条“自来水管道”数据源是“水源”可能是水库文件夹、河流视频流、水龙头摄像头等等。loaders.py是“取水泵站”负责把不同来源的水图像帧抽出来统一变成标准的“水”BGR 格式、特定尺寸、特定 dtype 的 numpy 数组。dataset.py是“水处理厂”在取水泵站的基础上再叠加 resize、归一化、增强、标签编码等处理。dataloader.py是“供水调度中心”控制多线程、多进程并行抽水和处理。模型训练/推理是“用户端水龙头”拿到的已经是干净可用的“饮用水”了。所以如果你想修改从“水源”到“标准图像”之间的逻辑就应该去改loaders.py如果你想改“标准图像到最终训练样本”的逻辑那重点在dataset.py。很多人分不清这个边界导致想自定义数据增强结果却去改了 loader当然事倍功半。2.3 为什么说 loaders.py 是数据链路的“前置必修课”我个人的切身体会如果你跳过loaders.py直接去看dataset.py你会觉得dataset.py里到处都是魔法。self.source是什么im 是从哪里来的为什么有时候im是一个 numpy 数组有时候又变成了PIL.Image这些问题的答案全部要回溯到loaders.py。换句话说loaders.py定义了整个数据流的第一公里。第一公里跑不顺后面所有逻辑都会出问题。而且在实际工程中很多人遇到的“训练时图片加载慢”“视频流推理卡顿”“内存暴涨”“多进程死锁”等问题根因都在loaders.py这一层。所以理解它不只是为了“看懂源码”更是为了排查和优化真实问题。3. LoadImages最常用的图像与视频加载器LoadImages是你在训练、验证、推理中最常见到的类。当你在命令行里写yolo predict source./images或yolo train dataxxx.yaml时如果数据源是文件夹或视频底层基本都会走这个类。3.1 核心属性与初始化逻辑看代码先看__init__。LoadImages的初始化方法核心作用是把path解析成具体的文件列表并判断类型。class LoadImages: def __init__(self, path, batch1, vid_stride1): files [] for p in sorted(path) if isinstance(path, (list, tuple)) else [path]: p str(Path(p).resolve()) if p.isdir(): # 遍历目录筛选出图片和视频文件 files.extend(glob.glob(str(p / ** / *.*), recursiveTrue)) elif p.endswith(.txt): # txt 文件里每行是一个文件路径 files.extend(Path(p).read_text().splitlines()) else: files.append(p) # 过滤出受支持的图片和视频格式 images [x for x in files if x.split(.)[-1].lower() in IMG_FORMATS] videos [x for x in files if x.split(.)[-1].lower() in VID_FORMATS] self.ni len(images) self.nv len(videos) ...这里有几个关键设计意图支持传入 list/tuple 路径这意味着你可以一次性传入多个图片路径列表或者多个文件夹路径列表loader 内部会将它们合并。实际项目里我经常用它来同时加载多个不同相机拍的数据集。.txt文件路径列表支持你可以准备一个 txt 文件里面每行写一个图片的绝对路径loader 会按行读取。这在处理大规模分布式存储数据集时非常实用不需要把所有文件拷到同一个目录。区分图片和视频图片和视频分别计数self.ni表示图片数量self.nv表示视频数量self.files属性里会同时存图片和视频路径。遍历时优先输出图片帧图片全部输出完后再输出视频帧。3.2__iter__与__next__的协作机制LoadImages实现了迭代器协议。在 PyTorch 的 DataLoader 中当num_workers 0时每个 worker 进程会通过迭代器不断拉取数据。其中__iter__方法做了大量初始化工作def __iter__(self): self.count 0 self.video_flag [False] * self.ni self.video_cap [None] * self.ni self.mode image for i in range(self.ni): self.video_flag[i] self.files[i].split(.)[-1].lower() in VID_FORMATS for i in range(self.ni, self.ni self.nv): self.video_flag.append(True) ... return self这段代码的巧妙之处在于video_flag的前ni个元素对应self.files前ni个图片文件值全为False后nv个元素对应视频文件值全为True。这样在__next__中就可以通过self.video_flag[self.count]快速判断当前应该走“读图片”分支还是“读视频帧”分支。__next__的核心逻辑是def __next__(self): if self.count self.nf: # 所有文件都遍历完关闭视频并抛出 StopIteration raise StopIteration path self.files[self.count] if self.video_flag[self.count]: # 视频分支 self.mode video ret_val, im0 self.video_cap[self.count].read() if not ret_val: # 当前视频读完了切到下一个文件 self.count 1 return self.__next__() ... else: # 图片分支 im0 cv2.imread(path) if im0 is None: LOGGER.warning(fWarning: 图片读取失败跳过: {path}) self.count 1 return self.__next__() ... return self.mode, im0, self.video_cap[self.count], path, ...注意这个递归调用return self.__next__()它的存在是为了自动跳过损坏的图片或已经播放结束的视频让迭代器永远“向前看”直到找到有效的下一帧。这种设计非常实用但初学者容易忽略如果你在自定义 loader 中不处理“读取失败”的情况很容易让整个数据流卡死。3.3vid_stride参数的底层逻辑很多人在使用predict或val时见过vid_stride这个参数但不一定清楚它的底层实现。它的含义是“视频抽帧步长”例如vid_stride2表示每 2 帧取 1 帧。在LoadImages中这个参数体现在视频分支的处理上# 在 __init__ 中记录 self.vid_stride vid_stride ... # 在 __next__ 视频分支中 if self.video_cap[self.count].get(cv2.CAP_PROP_POS_FRAMES) % self.vid_stride: # 不是目标帧则跳过 self.count 1 # 或者是直接读取下一帧视版本而定更近的版本中为了让视频读取更高效会结合cv2.CAP_PROP_POS_FRAMES判断当前读取的帧位置如果当前位置 % vid_stride ! 0就继续读下一帧而不返回。从使用者的角度理解vid_stride越大视频处理速度越快但会损失时间维度的连续性。在跟踪任务里如果vid_stride设置过大目标会出现明显的跳变导致跟踪 ID 频繁切换。我实测下来跟踪任务建议vid_stride不超过 5如果单纯做目标检测且视频较长vid_stride2或3是性价比很高的选择。3.4 实操心得如何给 LoadImages 增加“视频循环播放”能力默认情况下LoadImages遍历完最后一个视频帧后就会停止。但我在做视频流模型稳定性测试时经常希望同一个视频能循环播放模拟长时间运行的摄像头。这个需求不需要改框架只需要在调用时包一层“循环迭代器”def cycle_video_loader(source, vid_stride2): while True: loader LoadImages(source, vid_stridevid_stride) for *_, im, *_ in loader: yield im这里用while True不断创建新的LoadImages实例每次迭代完一个完整视频后重新开始。实测在长时间稳定性测试中这种循环方式非常稳定而且不会产生内存泄漏。注意每次创建新的 loader 会重新打开视频文件如果视频很大建议在循环外部做一次cv2.VideoCapture的复用但那样会牺牲代码简洁性。根据场景取舍。4. LoadStreams视频流与摄像头的并发读取方案LoadStreams比LoadImages复杂不少因为它的数据源是“流”——既可以是 RTSP/RTMP 网络流也可以是本地摄像头设备号。这类数据源的最大特点是帧率不稳定、网络延迟不可控、读取时间不均匀。如果直接在__next__里同步读帧遇到网络卡顿时会拖垮整个推理管线。4.1 多线程缓存的核心设计LoadStreams的初始化逻辑中有几个关键点class LoadStreams: def __init__(self, sourcesfile.streams, batch1, vid_stride1, bufferFalse): ... self.buffer buffer self.mode stream self.sources [x.strip() for x in sources.split(,)] # 支持逗号分隔多个流 self.vid_stride vid_stride self.caps [cv2.VideoCapture(x) for x in self.sources] ... # 为每个流创建独立的读取线程 self.threads [Thread(targetself.update, args(i,), daemonTrue) for i in range(n)] for t in self.threads: t.start()这里最核心的设计是每个视频流独立一个线程。每个线程调用update(i)方法反复从对应的VideoCapture中读取新帧并存入一个长度为 1 的列表self.imgs[i]中。def update(self, i): while True: if self.caps[i].isOpened(): success, im self.caps[i].read() if success: self.imgs[i] im self.frames[i] 1 else: # 读取失败时尝试重连 self.caps[i].open(self.sources[i]) time.sleep(0.01)为什么用列表而不是用一个普通变量因为 Python 的 GIL 不会保证多线程下普通变量的可见性但列表元素的赋值是原子操作读线程可以安全地读取最新帧。这是非常典型的“多线程生产者-消费者”模式生产者是update线程消费者是__next__中的主线程。4.2 为什么缓冲区不是越大越好很多人在看源码时会问为什么self.imgs[i]只保存一帧如果把缓冲区设置得大一些比如保存最近 10 帧是不是推理更平滑答案是在实时流处理场景下缓冲区越大数据延迟越大。假设摄像头帧率为 30 FPS缓冲区为 10 帧那你获取到的画面就有 10/30 ≈ 0.33 秒的延迟。对于实时监控或自动驾驶场景这个延迟是不可接受的。因此 Ultralytics 默认只缓存最新的一帧保证拿到的是“尽可能实时”的数据。我自己的经验是如果视频源帧率较低比如 5 FPS但推理速度很快可以在update循环里加一个time.sleep控制读取频率避免无意义地占用 CPU。例如def update(self, i): while True: ... if success: self.imgs[i] im time.sleep(1 / self.target_fps) # 按目标帧率读取4.3 多路视频流的内存与报错处理LoadStreams默认支持多个视频流同时输入比如一个 4 路摄像头的监控系统。但这里有个隐藏问题self.imgs是列表每个元素是一帧图像如果摄像头分辨率很高如 4K3840x2160x3一路视频流一帧就有约 25 MB 内存。4 路就是 100 MB。如果每路都缓存多帧内存压力会非常大。因此官方注释里专门提到要让视频流 buffer 较大时请显式传入 bufferTrue。在默认bufferFalse时只缓存最新一帧这是内存占用最小的方案。我在实际部署时还遇到过一个问题某些 RTSP 流会在凌晨自动断开因为摄像头端做 NTP 校时或网络波动。LoadStreams的update里已经做了重连逻辑if not success: self.caps[i].open(self.sources[i])但重连频率没有限制。如果网络一直不通这里会形成“读失败 - open - 再读失败 - 再 open”的高频死循环CPU 占用直接飙升。我踩过这个坑后会在外部包装一层重试计数class RetryLoadStreams(LoadStreams): def update(self, i): reconnections 0 while True: ... if not success: reconnections 1 if reconnections 10: LOGGER.warning(f流 {i} 多次重连失败等待 30 秒) time.sleep(30) self.caps[i].open(self.sources[i]) else: reconnections 0这样的重连机制在长稳测试中表现良好推荐参考。5. LoadPilAndNumpy 与 LoadTensor打通内存数据输入的最后一公里这两个类通常被忽略但在实际工程中非常有用尤其是在你已经在内存里有了图像数据、不想通过磁盘 IO 绕一圈的时候。5.1 LoadPilAndNumpy 的使用场景LoadPilAndNumpy支持传入一个包含PIL.Image或numpy.ndarray的迭代对象。它内部会逐个检查每个元素的类型def __init__(self, imgs): if not isinstance(imgs, (list, tuple)): imgs [imgs] self.imgs imgs self.mode image self.ni len(imgs) def __next__(self): im self.imgs[self.count] if isinstance(im, PIL.Image.Image): im np.asarray(im)[..., ::-1] # PIL 是 RGB转成 BGR elif isinstance(im, np.ndarray): im im[..., ::-1] # 假设输入是 RGB转成 BGR ...这里有个非常重要的细节Ultralytics 内部约定图像通道顺序是 BGR因为底层大量使用 OpenCV。而 PIL.Image 读取的图像是 RGB 顺序因此需要通过np.asarray(im)[..., ::-1]做通道反转。如果你自己构造 PIL 图像时没有注意这一点输入到模型后颜色会偏蓝偏红一团糟而且检查时还很难定位问题。5.2 LoadTensor 与 GPU 内存输入LoadTensor则更进一步它接受一个形状为(N, C, H, W)的 PyTorch Tensor并且可以自动处理设备迁移def __init__(self, imgs): self.imgs imgs self.mode tensor self.ni len(imgs) if isinstance(imgs, (list, tuple)) else 1 def __next__(self): im self.imgs[self.count] if isinstance(im, torch.Tensor): im im.cpu().numpy().transpose(1, 2, 0) # CHW - HWC ...我一般会在两个场景下使用LoadTensor推理服务里已有一批预处理后的图像 tensor需要直接送入 YOLO 的 predict 流程希望省掉重新解码的过程。做数据增强对比实验时希望把同一批 tensor 输入多个不同的 pipeline。需要注意的是如果 tensor 在 GPU 上LoadTensor默认会调用.cpu()这会带来一次 GPU-CPU 的拷贝。如果数据非常大且频繁调用性能瓶颈可能从“读取图像”变成“拷贝数据”。更优的做法是直接构造一个不拷贝的 loader或者直接调用模型.predict方法并传入已预处理好的 tensor。这一点在官方文档里没有明说是我实测对比后发现的。5.3 我为什么通常建议直接用 LoadTensor 做推理优化在实时视频分析服务中性能优化往往见缝插针。我做过一版“视频抽帧 预处理 推理”的服务最初每次都把帧写到磁盘再读吞吐量在 20 FPS 左右后来改成直接用LoadTensor从内存传入吞吐量提升到了 35 FPS。优化效果主要来自减少了磁盘 IO 和图像解码的重复操作。但这不代表LoadTensor是万能的。它只负责“输入格式转换”不包含缩放、归一化等预处理真正的预处理还是要在别处做。所以我的建议是如果你只是想快速测试用LoadImages最省心如果你想做服务化部署且要求低延迟熟悉一下LoadTensor和LoadPilAndNumpy会很有帮助。6. 源码中的底层细节图像预处理与迭代器协议6.1 BGR、letterbox 与数据格式约定在 loader 类中图像从数据源读到以后普遍都会经过一个转换步骤保证输出是 HWC 格式的 BGR 图像。以LoadImages为例im0 cv2.imread(path) # 已经是 BGRHWC而LoadPilAndNumpy中则要做通道反转因为输入可能是 RGB。所有 loader 最终产出的图像在进入后续流程时都统一是BGR、HWC、uint8、0-255范围内的numpy.ndarray。这个约定是整个 Ultralytics 数据流的基石。紧接着的letterbox操作一般发生在dataset.py的load_image或__getitem__中但 loader 层有时也会触发会把图像缩放并填充到固定的正方形尺寸例如 640x640。选择 letterbox 而不是直接 resize 的原因非常直观直接拉伸会改变目标的长宽比导致检测框坐标失真。letterbox 用灰色边框填充多余区域保持图像内容不失真对检测精度影响最小。我在做自定义 loader 时会刻意保持这个约定不变只改“图像来源”部分。一旦破坏了 BGR/HWC 的约定后面很多内置增强都会出问题。6.2__len__与迭代结束条件LoadImages实现了__len__方法def __len__(self): return self.nf # 图片数量 视频总帧数对于视频它的帧数计算方式是self.nf sum(int(cv2.VideoCapture(f).get(cv2.CAP_PROP_FRAME_COUNT)) for f in videos) ni注意这里每个视频都会重新打开一次VideoCapture来计算总帧数。当视频文件很多比如上千个短视频时初始化阶段会非常慢因为每个视频都要打开、读取元信息、关闭。我处理过一个包含 2000 个短视频的数据集仅初始化就花了将近 20 秒。优化办法是提前用一个全局字典缓存视频路径对应的帧数避免重复打开。6.3 迭代器协议中的StopIteration所有 loader 在迭代完所有数据后都会通过raise StopIteration结束。但在 PyTorch DataLoader 的多进程环境下如果StopIteration抛出时机不对会导致 worker 进程异常。Ultralytics 内部通过build.py中的_InfiniteDataLoader等类处理了这个问题——它会在迭代结束时自动重新初始化实现无限循环采样。我自己写自定义 loader 时也遇到过类似问题如果在__iter__里不重置self.count多轮 epoch 训练时数据会从上次结束的位置继续导致每个 epoch 数据分布不一致。这一点务必注意__iter__必须重置所有状态变量。7. 大规模数据加载时的性能优化与踩坑总结7.1 多进程 worker 与 loaders 的交互在训练 YOLO 模型时数据加载通常开了多个 worker 进程。PyTorch 的 DataLoader 会将 dataset 复制到每个 worker 中每个 worker 独立调用 dataset 的__getitem__。而loaders.py通常被 dataset 内部引用。这里有一个经典陷阱当 num_workers 0 时每个 worker 会各自持有一份 loader 的副本。例如使用LoadImages加载视频时每个 worker 都会打开同一个视频文件并各自从第 0 帧开始读取。如果你的数据并行策略需要所有 worker 看到不同片段就需要额外的分片逻辑。Ultralytics 在build.py的build_dataloader里已经处理了这个问题但如果你自定义 loader 套进自己的训练脚本很容易踩到“数据重复”的坑。我的建议是自定义 loader 时尽量让 loader 本身是无状态的或者只在主进程初始化一次通过worker_init_fn传入子进程。这样能避免很多多进程下的诡异行为。7.2 视频流加载的内存泄漏排查我在部署实时多路视频流检测服务时遇到过内存持续上涨的情况。排查过程非常痛苦最后定位到问题在LoadStreams中如果摄像头断流后self.caps[i].open(self.sources[i])返回失败但并没有释放旧资源cv2.VideoCapture对象会持有旧的底层缓冲反复重连就会导致内存缓慢增加。解决方法是在重连前显式调用self.caps[i].release()并重新创建VideoCapture对象。我在自己的分支里做了如下修改if not success: self.caps[i].release() self.caps[i] cv2.VideoCapture(self.sources[i])经过这个修改后内存曲线变得平稳。如果你在生产环境中也遇到类似问题可以优先检查这一块。7.3 图像读取失败时的静默跳过机制LoadImages在处理图片时如果cv2.imread返回None会打印一个警告并跳过。这个机制很实用但也藏着一个隐患如果一张图片因为文件名后缀被识别为图片但实际内容已损坏cv2.imread返回None后代码会递归调用__next__跳到下一张。整个过程对调用方来说是无感的但如果坏图片占比较高你会发现自己“数据量”比预期少了很多。建议在使用前先对数据集做一次完整性检查import cv2 from pathlib import Path def check_images_ok(folder): bad [] for p in Path(folder).rglob(*.*): if p.suffix.lower() in {.jpg, .jpeg, .png, .bmp}: im cv2.imread(str(p)) if im is None: bad.append(str(p)) return bad坏图比例超过 0.1% 时需要认真对待建议直接剔除而不是依赖 loader 的跳过机制。7.4 从 loader 到数据增强的无缝衔接很多读者会问既然 loader 已经输出了图像为什么不直接在loaders.py里做数据增强原因是职责分离。loaders.py的定位是“将数据源变成标准图像”不应该包含随机翻转、马赛克增强等训练相关操作。这些操作放在dataset.py中因为它们在每个 epoch 都是变化的而 loader 的产出应该是稳定可复现的。我在源码中对比后确认Ultralytics 对这一设计保持了高度一致。所以你如果想扩展一个新的“数据源”只需要继承或模仿loaders.py中的类输出标准图像即可后续的dataset.py和增强流程都能无缝衔接不用改任何其他模块。8. 如何快速验证你自定义的 loader 是否合格8.1 三步验证法我自己在写完自定义 loader 后通常会用下面三步验证它的正确性第一步检查输出格式。写一个简单的循环打印每次迭代返回的mode、图像shape、dtype、路径等字段。确保图像是HWC、BGR、uint8。loader MyCustomLoader(...) for mode, im, cap, path, *extra in loader: print(mode, im.shape, im.dtype, path) break第二步检查迭代完整性。统计len(loader)与实际迭代次数必须完全一致。如果不一致说明你的 loader 在某个分支里提前终止或死循环。第三步接入 YOLO 预测流程。把 loader 传给model.predict(sourceloader)跑一个最小测试观察预测结果是否符合预期。这一步能暴露格式约定上的隐性错误。8.2 常见失败模式对照现象可能原因排查方法推理结果颜色异常通道顺序不是 BGR检查图像转换逻辑确认有没有做::-1反转训练到一半卡死多进程下 loader 竞争同一个文件句柄减少 worker 数量或在 loader 中加入进程锁视频流推理延迟越来越高缓冲区不断堆积确认self.imgs[i]只用最新帧不要累积历史初始化很久大量视频被重复打开计算帧数使用帧数缓存机制内存持续上涨重连时未释放旧的 VideoCapture在重新 open 前显式 release部分图像缺失cv2.imread返回 None 后被跳过检测文件完整性或调整路径8.3 保留一个最小可复现模板为了方便后续开发我长期维护一个自定义 loader 的最小模板from ultralytics.data.loaders import LoadImages class MyCustomLoader(LoadImages): def __init__(self, path, batch1, vid_stride1): super().__init__(path, batch, vid_stride) # 这里做你的额外初始化 def __next__(self): # 调用父类拿到原始结果 result super().__next__() mode, im, cap, path, *extra result # 对 im 做自定义处理 # 自定义逻辑... return mode, im, cap, path, *extra保持“继承 重写__next__”的方式可以复用大部分逻辑改动最小也最容易 review。9. 结合 ultralytics 最新版本的变化随着 ultralytics 版本更新loaders.py也在不断优化。我看过 8.0.x 到 8.3.x 的演进主要变化有几点早期版本里LoadImages和LoadStreams都会在__init__中大量使用os.path处理路径新版本逐渐迁移到pathlib.Path对 Windows 路径的兼容性更好。视频帧读取逐渐引入vid_stride更高效的控制逻辑避免逐帧读但逐帧丢弃。LoadStreams的线程更新函数在较新版本中加入了self.auto自动判断是否需要重连逻辑更稳健。每个 loader 类都增加了更清晰的关键字参数说明方便 IDE 提示。这些变化本质上是工程细节的完善核心架构并没有大改。所以你现在读这份源码分析就算一段时间后再升级版本依然能顺畅迁移。在使用 ultralytics 做推理时如果传递的source是整数会被判定为摄像头设备号进去LoadStreams分支如果传递字符串路径但带.txt后缀会走文件列表分支。这些规则全部集中在build.py的load_inference_source函数中。如果你之后想深入可以按这个顺序继续读源码ultralytics/data/build.py中的load_inference_source看它如何把source分发给不同的 loader。ultralytics/data/dataset.py中的YOLODataset看 loader 的输出如何被进一步处理成训练样本。ultralytics/data/augment.py看数据增强如何基于标准图像做变换。读完这三个文件整个数据链路基本就完全打通了。我自己的体会是源码阅读最容易犯的错误是“只看类名不看调用关系”。loaders.py里的类再多最终都要被build.py或dataset.py调用。先理清调用顺序再深入每个类的内部逻辑效率会高很多。希望这篇文章能让你少走我踩过的弯路。