RAG系统大文件并发处理优化：流式传输、并发控制与内存优化实战

1. 大文件并发场景下RAG系统的真实瓶颈在哪

做过RAG知识库的人大概率都经历过这样一个阶段：小规模文档跑得挺顺，几百个PDF丢进去，检索效果也还行，但一旦文档量级上来、单个文件动辄几百MB甚至上GB，整个系统就开始不对劲了。上传接口超时、内存飙升、检索延迟从几百毫秒变成好几秒，严重的时候服务直接OOM挂掉。这不是模型的问题，也不是向量数据库的问题，而是大文件并发处理这条链路上多个环节同时出了问题。

我最初搭建RAG知识库的时候，用的是最朴素的方案：用户上传文件，后端接收完整文件后写入磁盘，然后同步做文本提取、分块、向量化、入库。这套流程在单用户、小文件场景下完全没问题，但一旦面对大文件并发上传，问题就集中爆发了。最典型的表现是：三个用户同时上传200MB以上的PDF，服务的内存占用直接从2GB飙到8GB以上，响应时间从秒级变成分钟级，最后Nginx直接返回504。

这里面的核心矛盾在于：RAG的文档处理链路是计算密集型和IO密集型混合的，而大文件并发会把这两个维度的压力同时放大。文本提取阶段需要把整个文件加载到内存中解析，分块阶段需要维护完整的分块列表，向量化阶段需要批量调用Embedding模型，每个环节都在吃内存和CPU。如果再加上并发，内存占用就是线性叠加的。

所以这篇文章要聊的不是RAG的检索策略优化，也不是Embedding模型选型，而是如何让RAG系统在大文件并发上传和处理的场景下保持稳定。涉及的核心技术点包括流式传输、并发控制、内存优化、分阶段处理、背压机制等。适合已经搭建过基础RAG系统、正在面对性能瓶颈的开发者，也适合正在设计RAG知识库架构、希望提前规避这些问题的同学。

下面我会从实际踩坑经历出发，把整条链路上每个环节的问题和解决方案拆开来讲，包括具体的参数配置、代码实现思路和实测数据。

2. 从上传到入库：大文件在RAG链路里到底经历了什么

2.1 一个500MB PDF的完整处理旅程

先把这个过程拆解清楚，后面讨论优化才有依据。假设用户上传了一个500MB的PDF文件，在典型的RAG系统中，它会经历以下阶段：

第一阶段：文件接收。后端服务通过HTTP接口接收文件。如果用的是传统的multipart/form-data方式，框架通常会把整个文件缓冲到内存或临时磁盘再交给业务代码。500MB的文件意味着至少500MB的内存或磁盘IO开销。

第二阶段：文本提取。用PyPDF2、pdfplumber或unstructured等工具解析PDF。这个阶段通常需要把文件完整加载到内存中，解析过程中还会产生大量的中间对象。实测下来，一个500MB的PDF在解析时峰值内存可能达到文件大小的3到5倍，也就是1.5GB到2.5GB。

第三阶段：文本分块。把提取出来的长文本按照固定大小或语义边界切分成chunk。这个阶段本身内存开销不大，但如果一次性把所有chunk都放在内存里等待向量化，chunk数量多的时候也会占用可观的内存。

第四阶段：向量化。调用Embedding模型把每个chunk转成向量。如果用的是本地模型（比如通过Ollama部署的模型），这个过程是CPU或GPU密集型的；如果用的是远程API，则受限于网络延迟和速率限制。

第五阶段：入库。把向量和对应的文本、元数据写入向量数据库。这个阶段主要是IO操作，但批量写入时如果批次过大，也会造成内存压力。

把这五个阶段串起来看，一个500MB的PDF从上传到入库，峰值内存占用可能超过3GB，处理时间可能超过5分钟。如果三个用户同时上传，内存需求就是9GB以上，处理时间还会因为CPU竞争而进一步拉长。

2.2 并发放大效应：为什么问题不是线性增长的

很多人会直觉地认为，三个并发就是单个处理的三倍资源消耗。但实际情况比这糟糕得多。原因在于：

内存碎片化。多个大文件同时解析时，Python的垃圾回收机制在高内存压力下效率会下降，内存碎片增加，实际占用可能比理论值高出30%到50%。

IO竞争。多个文件同时读写磁盘，磁盘IOPS被打满，每个文件的处理速度都会下降，导致文件在内存中停留的时间更长，进一步加剧内存压力。

CPU上下文切换。文本提取和向量化都是CPU密集型操作，并发执行时CPU频繁在不同线程间切换，实际有效计算时间占比下降。

连接池耗尽。如果向量数据库或Embedding服务的连接池大小有限，并发请求会排队等待，导致请求超时。

我在实测中观察到：单个500MB PDF处理耗时约4分钟，峰值内存2.8GB；两个并发时，每个耗时约7分钟，峰值内存合计6.5GB；三个并发时，每个耗时超过12分钟，峰值内存合计突破11GB，服务开始出现OOM。这就是并发放大效应。

3. 流式传输：把文件接收阶段的内存开销降到最低

3.1 传统上传方式的问题

大部分Web框架默认的文件上传处理方式是把整个请求体缓冲下来再交给业务代码。以FastAPI为例，如果你用UploadFile，它底层会用SpooledTemporaryFile，超过一定大小（默认1MB）就会写到磁盘。这看起来好像没问题，但实际上：

• 文件先写到磁盘临时目录，然后业务代码再读出来处理，多了一次完整的磁盘IO。
• 临时文件的生命周期管理容易出问题，高并发时临时目录可能被写满。
• 如果业务代码需要把文件转发到其他服务（比如独立的文档处理服务），又要重新读一遍。

3.2 流式接收的实现思路

流式传输的核心思想是：边接收边处理，不在内存或磁盘上保留完整文件。具体实现上有几种方案：

方案一：分片上传 + 流式合并。前端把大文件切成固定大小的分片（比如5MB），逐个上传，后端每收到一个分片就追加写入目标文件。这样单次内存占用不超过一个分片的大小。缺点是前端实现复杂一些，需要处理分片顺序、断点续传等逻辑。

方案二：流式请求体读取。后端直接从请求流中按块读取数据，每读一块就写入目标文件或送入处理管道。以FastAPI为例，可以通过request.stream()来逐块读取：

from fastapi import Request import aiofiles async def upload_stream(request: Request, file_id: str): file_path = f"/data/uploads/{file_id}" async with aiofiles.open(file_path, "wb") as f: async for chunk in request.stream(): await f.write(chunk) return {"status": "ok", "path": file_path}

这种方式下，内存中同时存在的数据不超过一个chunk的大小（通常几十KB），内存开销几乎可以忽略。

方案三：直接流式送入处理管道。更进一步，如果文本提取工具支持流式输入，可以边接收边提取，完全省去落盘步骤。但实际中大部分PDF解析工具都需要完整的文件或可随机访问的文件对象，所以这个方案适用场景有限。比较务实的做法还是先流式落盘，再从磁盘流式读取进行解析。

3.3 流式传输的注意事项

流式接收虽然能大幅降低内存开销，但有几个坑需要注意：

注意：流式接收时如果客户端中断连接，已经写入的部分文件需要清理，否则会产生大量垃圾文件。建议在异常处理中加上清理逻辑，或者用定时任务扫描孤儿文件。

注意：流式接收无法在接收阶段做完整的文件校验（比如MD5），需要在接收完成后单独校验。如果对文件完整性要求高，建议前端在上传前计算文件哈希，上传完成后后端再算一次做比对。

另外，流式接收的吞吐量受限于网络带宽和磁盘写入速度。如果磁盘写入是瓶颈，可以考虑先写入内存缓冲区再批量落盘，但这样又会增加内存开销，需要根据实际情况权衡。

4. 并发控制：不是限制用户，而是保护系统

4.1 为什么不能无限并发

很多人觉得并发越高吞吐量越大，但在RAG的大文件处理场景下，这个假设不成立。因为每个处理任务都是资源密集型的，当并发数超过系统承载能力时，所有任务的完成时间都会急剧拉长，甚至全部失败。这就是典型的拥塞崩溃。

我做过一组对比测试：在4核8GB的机器上，用不同的并发数处理10个200MB的PDF文件，结果如下：

可以看到，并发数为2时总耗时最短，超过这个数之后总耗时反而增加，失败率也上升。所以并发控制的目标不是限制用户，而是找到系统的最佳工作点。

4.2 信号量 + 队列的组合方案

最实用的并发控制方案是信号量控制并发数 + 队列缓冲待处理任务。具体来说：

• 用一个信号量限制同时处理的任务数量（比如设置为CPU核心数或根据内存计算出的安全值）。
• 超出的任务放入队列等待，而不是直接拒绝。
• 队列设置最大长度，超过时返回"系统繁忙"提示，避免无限堆积。

在Python中可以用asyncio.Semaphore来实现：

import asyncio MAX_CONCURRENT = 2 semaphore = asyncio.Semaphore(MAX_CONCURRENT) async def process_document(file_path: str): async with semaphore: # 文本提取、分块、向量化、入库 await extract_text(file_path) await chunk_and_embed(file_path)

这个方案的关键在于MAX_CONCURRENT的取值。我的经验是：

• 如果文本提取和向量化都在同一台机器上，取min(CPU核心数, 可用内存 / 单任务峰值内存)。
• 如果向量化走远程API，可以适当放宽，因为远程调用不占本地CPU。
• 建议留出20%的资源余量给系统本身和其他服务。

4.3 动态并发调整

固定并发数在负载波动大的场景下不够灵活。更好的做法是根据系统实时负载动态调整并发数。比如：

• 监控内存使用率，当超过80%时降低并发数。
• 监控任务队列长度，当队列积压时适当提高并发数（如果有资源余量）。
• 监控任务平均处理时间，当处理时间异常升高时降低并发数。

这个逻辑可以用一个简单的反馈控制循环来实现，不需要太复杂。我自己的做法是每30秒检查一次系统负载，根据预设的阈值调整信号量的许可数。

4.4 任务优先级与公平性

在多用户场景下，还需要考虑任务的优先级和公平性。比如：

• 小文件优先处理，避免被大文件阻塞。
• 同一用户的任务串行处理，避免单个用户占满所有并发槽位。
• 支持取消排队中的任务，用户等不及了可以主动取消。

这些策略可以根据实际业务需求来定，核心原则是不要让单个大文件或单个用户拖垮整个系统。

5. 内存优化：从"全量加载"到"分而治之"

5.1 文本提取阶段的内存优化

文本提取是大文件处理中内存开销最大的环节。以PDF为例，大部分解析库的工作方式是：加载整个文件 → 解析页面树 → 逐页提取文本 → 返回完整文本。这个过程中，文件本身、解析中间对象、提取结果同时存在于内存中。

优化思路有几种：

逐页处理。不要一次性提取整个PDF的文本，而是逐页提取，每提取一页就送入后续的分块和向量化流程。这样内存中同时只存在一页的文本和解析对象。PyPDF2和pdfplumber都支持按页操作：

import pdfplumber def extract_page_by_page(file_path: str): with pdfplumber.open(file_path) as pdf: for page in pdf.pages: text = page.extract_text() if text: yield text page.flush_cache() # 释放页面缓存

及时释放引用。提取完一页后，确保相关的对象引用被释放，让垃圾回收器可以回收内存。在Python中，可以用del显式删除不再需要的对象，或者用gc.collect()强制回收（但不要频繁调用，会影响性能）。

使用轻量级解析库。不同的PDF解析库内存开销差异很大。实测下来，pdfplumber功能强但内存开销大，PyPDF2相对轻量但功能有限，pymupdf（fitz）在内存和速度上比较均衡。选型时需要根据实际文档特点来权衡。

5.2 分块与向量化的流水线设计

传统做法是：提取完所有文本 → 分块 → 所有chunk一起向量化 → 一起入库。这个模式的问题在于，所有chunk需要同时存在于内存中。

改成流水线模式后，每个阶段可以独立进行，chunk像流水线上的零件一样逐个通过：

async def pipeline(file_path: str): chunk_buffer = [] async for page_text in extract_page_by_page(file_path): chunks = split_text(page_text) for chunk in chunks: chunk_buffer.append(chunk) if len(chunk_buffer) >= BATCH_SIZE: await embed_and_store(chunk_buffer) chunk_buffer.clear() if chunk_buffer: await embed_and_store(chunk_buffer)

这样内存中同时存在的chunk数量不超过BATCH_SIZE，可以控制在很小的范围内。BATCH_SIZE的取值需要权衡：太小会导致Embedding调用次数过多，太大则内存开销增加。我的经验值是16到64之间，具体取决于chunk大小和Embedding模型的批处理能力。

5.3 向量化阶段的批处理与限流

向量化阶段有两个内存相关的点需要注意：

批处理大小。调用Embedding模型时，一次传入多个文本比逐个传入效率高，但批处理太大会增加内存开销。如果用的是本地模型，批处理大小还受限于GPU显存。建议从较小的批次开始测试，逐步增大直到找到性能和内存的平衡点。

结果缓冲。向量化结果（浮点数数组）本身也占内存。一个768维的float32向量占3KB左右，一万个chunk就是30MB，看起来不多，但如果同时处理多个大文件，累积起来也可观。建议向量化完一批就立即入库，不要在内存中积压。

5.4 向量数据库写入的批量控制

写入向量数据库时，批量大小同样需要控制。大部分向量数据库（如Milvus、Qdrant、Weaviate）都支持批量写入，但批量太大会导致客户端和服务端的内存压力都增加。建议：

• 批量大小控制在100到500之间。
• 写入时使用异步接口，避免阻塞主处理流程。
• 监控写入延迟，如果延迟升高说明批量太大或数据库压力过大。

另外，写入时要注意幂等性。如果同一个文件被重复处理（比如用户重复上传或任务重试），需要避免产生重复的向量记录。通常的做法是用文件哈希作为去重键，写入前先检查是否已存在。

6. 背压机制：让系统在过载时优雅降级

6.1 什么是背压，为什么RAG系统需要它

背压（Backpressure）是指当系统下游处理能力不足时，向上游传递压力，让上游降低生产速度或暂停生产。在RAG系统中，背压机制可以防止任务无限堆积导致系统崩溃。

举个实际场景：用户批量上传了50个大文件，如果系统不加限制地接收所有任务，队列会迅速膨胀，内存被队列中的任务元数据占满，最终OOM。有了背压机制后，当队列达到上限时，系统会拒绝新任务并返回明确的提示，让用户知道当前系统繁忙，稍后重试。

6.2 队列长度与拒绝策略

队列长度的设置需要根据系统处理能力和可接受的最大延迟来定。假设单任务平均处理时间是3分钟，并发数是2，那么系统每小时最多处理40个任务。如果队列长度设为20，意味着最坏情况下用户需要等待30分钟。这个等待时间是否可接受，取决于业务场景。

拒绝策略有几种：

• 直接拒绝：返回429状态码，提示"系统繁忙，请稍后重试"。
• 降级接收：接收任务但降低处理优先级，比如放到低优先级队列，等系统空闲时再处理。
• 部分接收：对于批量上传，只接收前N个文件，其余返回"超出当前处理能力"。

我自己的做法是结合使用：正常负载下接收所有任务，队列达到80%时开始降级接收（提示用户可能延迟），队列满时直接拒绝。

6.3 超时与重试的边界

大文件处理任务很容易超时。需要设置合理的超时时间，并且区分不同阶段的超时：

• 上传阶段超时：通常设置较短，比如60秒，因为流式上传不应该花太长时间。
• 处理阶段超时：根据文件大小动态设置，比如基础300秒加上每MB增加1秒。
• 入库阶段超时：设置较短，比如30秒，因为批量写入通常很快。

超时后的重试需要谨慎。如果是因为系统过载导致的超时，重试只会加剧过载。建议只对明确的临时性错误（如网络抖动）进行重试，并且使用指数退避策略。

6.4 监控与告警：提前发现问题

背压机制要发挥作用，前提是能及时发现系统过载。需要监控的指标包括：

这些指标可以通过Prometheus + Grafana来采集和展示，告警可以通过Webhook推送到团队沟通工具。

7. 实测数据与调优经验

7.1 优化前后的对比

在同一台4核8GB的机器上，用优化前后的方案分别处理10个200MB的PDF文件（并发上传），对比数据如下：

优化后的方案核心改动就是三点：流式接收文件、信号量控制并发数为2、逐页提取+流水线分块向量化。

7.2 不同文件类型的处理差异

不同类型的文档在解析阶段的表现差异很大：

• 纯文本PDF：解析快，内存开销小，主要瓶颈在向量化。
• 扫描版PDF：需要OCR，CPU开销极大，建议单独走OCR队列，并且并发数设为1。
• Word文档：解析相对轻量，但格式复杂时（大量表格、图片）内存开销会增加。
• HTML/ Markdown：解析最快，内存开销最小，可以适当提高并发数。

所以并发控制策略应该按文件类型区分，而不是一刀切。

7.3 几个容易忽略的细节

临时文件清理。流式接收和处理过程中会产生临时文件，如果清理不及时，磁盘会被写满。建议用定时任务每天清理超过24小时的临时文件。

连接池配置。向量数据库和Embedding服务的连接池大小要与并发数匹配。如果并发数是2但连接池只有1，第二个任务会等待连接，实际并发退化为1。

日志量控制。大文件处理过程中如果每个chunk都打日志，日志量会非常大，影响性能。建议只在关键节点打日志，比如文件接收完成、文本提取完成、向量化完成、入库完成。

优雅关闭。服务重启时需要等待正在处理的任务完成，或者把未完成的任务重新入队。直接kill进程会导致任务丢失和临时文件残留。

7.4 扩展思路：分布式处理

当单机处理能力达到上限时，可以考虑把文档处理拆分成独立的服务，部署到多台机器上。架构变成：上传服务负责接收文件并写入共享存储，消息队列传递任务，多个处理worker从队列消费任务。这样可以通过增加worker数量来水平扩展处理能力。

但分布式也带来了新的问题：任务状态管理、失败重试、结果一致性等。建议在单机方案优化到极限之后再考虑分布式，不要过早引入复杂度。

8. 写在最后

大文件并发处理是RAG系统从demo走向生产必须跨过的一道坎。核心思路其实不复杂：流式化减少内存占用，并发控制保护系统稳定，流水线化提高处理效率，背压机制实现优雅降级。但每个环节都有很多细节需要根据实际场景去调整。

我在实际项目中最深的体会是：不要等到系统崩溃了才去优化，而是在设计阶段就把这些机制考虑进去。因为RAG系统的文档处理链路很长，任何一个环节出问题都会影响整体稳定性，事后补救的成本远高于提前设计。

另外，监控和告警一定要尽早搭建。很多时候系统已经在过载边缘了，但没有监控就发现不了，直到用户反馈或者服务崩溃才知道。有了监控数据，调优也有依据，不用凭感觉猜。

最后分享一个实用技巧：在开发阶段可以用小文件模拟大文件的行为，比如用一个10MB的文件但把并发数调到很高，这样能快速复现大文件并发时的资源竞争问题，比每次都用真实大文件测试效率高得多。