1. 大文件并发场景下 RAG 的真实挑战做过 RAG 项目的人大概都有个共识Demo 跑通只要一个下午但真把它丢到生产环境里让几十上百号人同时上传几十兆甚至上百兆的文档系统还能稳得住那就是另一回事了。我前后经手过三四个 RAG 知识库项目从最初用 LangChain 拼一个能问答的原型到后来要支撑整个团队日常文档检索中间踩的坑基本都集中在同一个地方——大文件 并发这两个条件叠加的时候系统会以一种非常难看的方式崩掉。这篇文章想聊的就是这件事当你的 RAG 知识库需要同时处理大文件上传、解析、切分、向量化、入库这一整条链路并且还要扛住多人并发的时候到底该怎么设计。核心关键词就两个——RAG和大文件并发。我会把整条链路的拆解思路、关键参数怎么定、并发怎么控、瓶颈卡在哪、怎么排查都摊开讲一遍。适合已经跑通过基础 RAG 流程、准备往生产环境推进的开发者也适合正在被上传一个大 PDF 就把服务卡死折磨的同学。先说清楚一个前提RAG 本身不复杂检索增强生成这套逻辑无非是文档切块、向量化、存库、检索、拼上下文喂给大模型。真正难的是工程化。大文件并发之所以是个坎是因为它同时压垮了三个环节——IO、CPU、内存而且这三个环节的瓶颈会互相传导。你以为是向量库慢其实是 PDF 解析把内存吃满了你以为是模型推理慢其实是切分逻辑在单线程里排队。不把这条链路拆清楚调优就是瞎猜。下面我按整体设计思路 → 核心环节拆解 → 完整实操流程 → 问题排查这个顺序来讲每一部分都会给出我实际用过的方案和参数能直接抄的我就写清楚需要根据场景调整的我会说明判断依据。2. 整体架构设计与并发思路拆解2.1 为什么不能上传即处理新手最容易犯的错是把整个 RAG 流程写成同步的用户上传文件 → 后端接收 → 解析 → 切分 → 向量化 → 入库 → 返回成功。小文件没问题几百 KB 的 Markdown 秒级完成。但一个 80MB 的 PDF光解析可能就要几十秒向量化几千个 chunk 又要几分钟。这期间 HTTP 连接一直挂着用户以为卡死了服务器线程也被占着。来五个这样的请求服务基本就废了。所以第一个设计决策就是上传和处理必须解耦。上传接口只负责把文件落到对象存储或本地磁盘然后往消息队列里丢一个任务立刻返回一个 task_id。真正的解析、切分、向量化交给后台的 worker 异步做。前端拿着 task_id 轮询进度或者用 SSE/WebSocket 推状态。这个决策背后的逻辑很直白HTTP 请求的生命周期应该是短的而文档处理的生命周期是长的两者强行绑在一起就是把长任务的成本转嫁给了连接层。解耦之后上传接口的 QPS 可以很高处理能力则通过 worker 数量独立伸缩。2.2 并发模型的选择进程、线程还是协程Python 生态里做 RAG绕不开 GIL。PDF 解析比如 PyMuPDF、文本切分、embedding 推理这些都是 CPU 密集型操作多线程在 GIL 下根本跑不满多核。我试过用 ThreadPoolExecutor 并发解析实测下来 CPU 利用率卡在 120% 左右假设 8 核基本等于单核在跑。所以并发模型我推荐多进程 队列的组合。用 Celery 或者更轻量的 RQ 做任务队列worker 以进程方式启动每个进程独立处理一个文档任务。进程数一般设为 CPU 核数或者核数 1。这样每个进程能真正吃满一个核8 核机器就能并行处理 8 个文档。但这里有个坑embedding 模型如果加载在每个 worker 进程里8 个进程就是 8 份模型内存直接爆炸。一个 bge-large 模型大概 1.3GB8 份就是 10GB 起步。解决办法有两个一是用独立的 embedding 服务比如用 FastAPI 单独起一个服务worker 通过 HTTP 调用二是用共享内存或者模型量化。我一般选前者虽然多了一次网络调用但内存可控而且 embedding 服务本身可以做批处理吞吐反而更高。2.3 大文件的分块处理策略大文件不能一次性读进内存。一个 200MB 的 PDFPyMuPDF 打开后如果一次性 extract 所有文本内存峰值可能到 1GB 以上。并发几个这样的请求内存就爆了。我的做法是按页流式处理。PyMuPDF 支持逐页读取处理完一页就释放一页的内存。具体来说遍历每一页提取文本立刻做切分切分后的 chunk 攒够一批比如 64 个就送去 embeddingembedding 完立刻入库然后清空这批 chunk。这样内存占用是恒定的跟文件大小无关只跟批大小有关。这个策略的关键在于不要让整个文档的 chunk 同时存在于内存里。很多人习惯先把所有 chunk 切好存成一个 list再统一 embedding这在处理大文件时就是内存杀手。流式处理虽然代码复杂一点但内存曲线是平的非常稳。2.4 向量库的写入并发向量库这块Milvus、Qdrant、Weaviate 我都用过。并发写入时最容易出问题的是批量大小和并发数的平衡。批量太小网络往返次数多吞吐上不去批量太大单次请求内存高而且向量库服务端可能超时。我的经验值是单次批量 100 到 256 个向量并发写入数控制在 4 到 8 之间。这个范围是实测出来的再往上加Qdrant 的写入延迟会明显上升而且容易出现部分失败。另外写入一定要做幂等用文档 ID chunk 序号做唯一键重复写入时覆盖而不是追加这样任务重试不会产生脏数据。3. 核心环节拆解与关键参数实操3.1 文档解析不同格式的处理要点RAG 知识库常见的文档格式有 PDF、Word、Markdown、HTML、纯文本。每种格式的解析坑都不一样。PDF 是最麻烦的。扫描版 PDF 需要 OCR这个成本很高一般我会在解析前先判断如果提取出的文本长度小于某个阈值比如每页少于 50 个字符就判定为扫描版走 OCR 流程或者直接标记为需人工处理。文本版 PDF 用 PyMuPDF 就够了速度快内存可控。注意 PyMuPDF 的page.get_text()默认会保留一些排版信息如果不需要用get_text(text)更干净。Word 文档用 python-docx但要注意表格和图片。表格里的文本需要单独提取否则会丢失。图片如果知识库需要支持图片检索热词里有人问rag知识库能存储图片嘛那就要走多模态 embedding这是另一个话题这里先不展开。Markdown 和 HTML 相对简单但要注意代码块。代码块里的内容如果被切分逻辑打散检索时会出问题。我的做法是在切分前先识别代码块把整个代码块作为一个不可分割的单元。3.2 文本切分chunk 大小和重叠的取舍切分是 RAG 里最容易被忽视但影响最大的环节。chunk 太大检索精度下降因为一个 chunk 里混了太多主题chunk 太小上下文不完整模型回答时缺信息。我的默认参数是chunk size 512 tokensoverlap 64 tokens。这个组合在大多数场景下表现均衡。但要注意这是 token 数不是字符数。中文一个字符大约 1.5 到 2 个 token所以 512 tokens 大概是 300 到 350 个中文字。如果你的文档是技术文档句子长、术语多可以适当放大到 768 tokens。overlap 的作用是防止句子被切断导致语义丢失。64 tokens 的重叠大概能覆盖一到两句话足够衔接上下文。overlap 太大会导致重复内容多检索时返回一堆相似 chunk浪费上下文窗口。切分工具我用的是 LangChain 的 RecursiveCharacterTextSplitter它支持按段落、句子、字符逐级降级切分比单纯按字符数切要合理得多。配置的时候把 separators 设成[\n\n, \n, 。, , , ., , ]中英文都能兼顾。3.3 Embedding 批处理与并发控制Embedding 是整条链路里最耗时的环节。一个 100 页的 PDF切出来大概 500 到 800 个 chunk用 bge-large 在 CPU 上跑每个 chunk 大概 50 到 100ms总共要 40 到 80 秒。如果用 GPU能快 10 倍以上。批处理是关键。单条 embedding 的效率极低因为模型前向传播的固定开销占大头。把 batch size 设到 32 或 64吞吐能提升 5 到 8 倍。但 batch 太大会吃内存GPU 上尤其明显。我的经验是GPU 显存 8GB 的话bge-large 用 batch 64 没问题CPU 的话batch 32 比较稳。并发控制这块我建议用信号量限制同时进行的 embedding 请求数。如果是独立 embedding 服务服务端自己会排队客户端并发数设成 4 到 8 就行。如果是进程内调用模型那并发数基本等于 worker 进程数不需要额外控制但要确保每个进程的 batch 不要太大否则内存扛不住。3.4 向量入库的批量与重试入库这块前面说了批量 100 到 256。这里补充一个细节入库要和 embedding 流水线化。不要等所有 chunk 都 embedding 完再统一入库而是 embedding 完一批就入库一批。这样即使中途失败已经入库的部分不用重做重试时从断点继续。重试策略用指数退避初始延迟 1 秒最大延迟 30 秒重试 3 次。如果 3 次还失败把任务标记为失败记录失败原因人工介入。不要无限重试否则一个坏任务会一直占着 worker。4. 完整实操流程与关键代码4.1 环境准备与依赖清单先列一下我常用的技术栈这套组合在多个项目里验证过比较稳任务队列Celery RedisRedis 同时做 broker 和 result backend文档解析PyMuPDFPDF、python-docxWord、markdownMarkdown切分LangChain 的 RecursiveCharacterTextSplitterEmbeddingbge-large-zh-v1.5用 FastAPI 单独起服务向量库Qdrant轻量单机部署方便Web 框架FastAPI安装依赖pip install fastapi uvicorn celery redis pymupdf python-docx markdown langchain qdrant-client sentence-transformersQdrant 用 Docker 起一个单机版docker run -d -p 6333:6333 -v $(pwd)/qdrant_storage:/qdrant/storage qdrant/qdrant4.2 上传接口与任务分发上传接口只做三件事存文件、生成 task_id、丢任务。from fastapi import FastAPI, UploadFile from celery import Celery import uuid, os app FastAPI() celery_app Celery(rag, brokerredis://localhost:6379/0, backendredis://localhost:6379/1) UPLOAD_DIR ./uploads os.makedirs(UPLOAD_DIR, exist_okTrue) app.post(/upload) async def upload(file: UploadFile): task_id str(uuid.uuid4()) file_path os.path.join(UPLOAD_DIR, f{task_id}_{file.filename}) # 流式写盘避免大文件占内存 with open(file_path, wb) as f: while chunk : await file.read(1024 * 1024): f.write(chunk) celery_app.send_task(tasks.process_document, args[task_id, file_path]) return {task_id: task_id, status: queued}注意这里用了await file.read(1024 * 1024)分块读每次 1MB避免把整个文件读进内存。这是处理大文件的基本功。4.3 后台任务的流式处理实现Celery 任务里做解析、切分、embedding、入库。核心是流式不要攒。from celery import shared_task import fitz from langchain.text_splitter import RecursiveCharacterTextSplitter import requests from qdrant_client import QdrantClient from qdrant_client.models import PointStruct shared_task(bindTrue, max_retries3) def process_document(self, task_id, file_path): try: splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap64, separators[\n\n, \n, 。, , , ., , ] ) qdrant QdrantClient(hostlocalhost, port6333) doc fitz.open(file_path) batch_chunks, batch_ids [], [] chunk_idx 0 for page in doc: text page.get_text(text) if len(text.strip()) 50: continue for chunk in splitter.split_text(text): batch_chunks.append(chunk) batch_ids.append(f{task_id}_{chunk_idx}) chunk_idx 1 if len(batch_chunks) 64: flush_batch(qdrant, batch_chunks, batch_ids, task_id) batch_chunks, batch_ids [], [] if batch_chunks: flush_batch(qdrant, batch_chunks, batch_ids, task_id) doc.close() return {task_id: task_id, chunks: chunk_idx} except Exception as exc: raise self.retry(excexc, countdown2 ** self.request.retries) def flush_batch(qdrant, chunks, ids, task_id): resp requests.post(http://localhost:8001/embed, json{texts: chunks}) vectors resp.json()[vectors] points [ PointStruct(idabs(hash(i)) % (10**12), vectorv, payload{text: c, doc_id: task_id}) for i, c, v in zip(ids, chunks, vectors) ] qdrant.upsert(collection_namedocs, pointspoints)这段代码的关键点每攒够 64 个 chunk 就 flush 一次embedding 和入库都在 flush 里完成然后清空列表。内存占用恒定在 64 个 chunk 的量级跟文件大小无关。4.4 Embedding 服务的批处理实现独立的 embedding 服务用 FastAPI 起加载模型一次常驻内存。from fastapi import FastAPI from sentence_transformers import SentenceTransformer from pydantic import BaseModel app FastAPI() model SentenceTransformer(BAAI/bge-large-zh-v1.5) class EmbedRequest(BaseModel): texts: list[str] app.post(/embed) def embed(req: EmbedRequest): vectors model.encode(req.texts, batch_size32, normalize_embeddingsTrue) return {vectors: vectors.tolist()}normalize_embeddingsTrue很重要归一化后的向量用余弦相似度检索时等价于内积Qdrant 里可以用更快的距离计算方式。4.5 并发 worker 的启动配置Celery worker 用进程模式启动并发数设为 CPU 核数celery -A tasks worker --loglevelinfo --concurrency8 --poolprefork--poolprefork是默认的多进程模式每个 worker 进程独立处理任务。8 核机器设 8 个进程每个进程处理一个文档互不干扰。但要注意如果 embedding 是进程内调用不是独立服务那 8 个进程就是 8 份模型内存扛不住。所以强烈建议用独立 embedding 服务worker 只做解析、切分和 HTTP 调用内存占用小很多。5. 常见问题与排查技巧实录5.1 内存暴涨的排查路径内存暴涨是最常见的问题。排查顺序先看是不是文件一次性读进了内存再看是不是 chunk 列表攒太多最后看是不是 embedding 模型重复加载。用memory_profiler或者简单的psutil打点在处理过程中每隔几秒记录一次 RSS。如果内存曲线是持续上升的基本就是攒数据了如果是阶梯式上升然后不降可能是模型加载或者向量库客户端缓存。我遇到过一次内存一直涨最后发现是 Qdrant 客户端的连接池没释放每个任务创建一个新客户端旧的不回收。改成全局单例客户端就好了。5.2 任务卡死与超时处理任务卡死通常是某个环节阻塞了。最常见的是 embedding 服务响应慢或者向量库写入超时。给每个 HTTP 调用加超时embedding 设 30 秒向量库写入设 10 秒。超时后抛异常让 Celery 重试。另外Celery 要设task_time_limit和task_soft_time_limit硬超时设 10 分钟软超时设 8 分钟。软超时抛异常可以捕获做清理硬超时直接杀进程。没有超时限制的话一个卡死的任务会永远占着 worker。5.3 检索命中率低的调优热词里有人提到 rag hit rate这确实是 RAG 的核心指标。命中率低通常是三个原因chunk 切得不好、embedding 模型不适合领域、检索策略太单一。chunk 问题前面说了调整 size 和 overlap。embedding 模型如果领域特殊比如医疗、法律通用模型效果会差需要微调或者换领域模型。检索策略上单纯向量检索容易漏掉关键词匹配的情况可以加一路 BM25 做混合检索两路结果用 RRFReciprocal Rank Fusion融合命中率能提升 10 到 20 个百分点。5.4 常见问题速查表问题现象可能原因排查方法解决方向内存持续上涨chunk 列表攒太多打点记录 RSS改流式处理及时清空任务卡死不结束无超时限制看 worker 日志加 task_time_limit入库部分失败批量太大或并发太高看向量库日志降批量到 128并发降到 4检索结果重复overlap 太大检查 chunk 内容overlap 降到 32 到 64embedding 慢单条调用或 batch 太小看服务端 QPSbatch 提到 32 到 64扫描版 PDF 无内容未做 OCR检查提取文本长度加 OCR 流程或标记人工5.5 几个踩过的坑第一个坑Celery 的prefork模式下如果在模块顶层加载模型每个子进程都会加载一份内存直接翻倍。解决办法是把模型加载放到函数内部或者用worker_process_init信号在进程启动时加载一次。第二个坑Qdrant 的upsert如果 point id 用随机数重复写入会产生重复数据。一定要用确定性的 id比如文档 ID 加 chunk 序号的哈希。第三个坑大文件上传时如果前端用FormData一次性提交浏览器内存也会爆。前端也要做分片上传后端合并。这个虽然不在 RAG 核心链路里但实际项目中一定会遇到。第四个坑Redis 作为 broker 时如果任务消息太大比如把文件内容塞进消息里Redis 内存会涨得很快。任务消息里只放文件路径和 task_id不要放文件内容。6. 性能压测与容量估算6.1 单文档处理耗时拆解以一个 50MB、200 页的文本版 PDF 为例实测各环节耗时解析PyMuPDF 逐页约 8 秒切分约 2 秒Embeddingbge-largeGPUbatch 64约 15 秒入库Qdrantbatch 128约 5 秒总计约 30 秒如果是 CPU 跑 embedding这部分会变成 120 到 180 秒总耗时 2 到 3 分钟。所以 GPU 对 RAG 的吞吐提升是决定性的。6.2 并发容量估算假设 8 核 CPU、32GB 内存、单张 8GB 显存的 GPU。worker 进程 8 个每个进程处理一个文档内存占用约 500MB不含模型8 个进程 4GB。embedding 服务占 2GB 显存和 3GB 内存。Qdrant 占 4GB 内存。总共约 11GB 内存还有余量。吞吐上GPU embedding 是瓶颈。bge-large 在 8GB 显存上batch 64 的吞吐大概是每秒 200 到 300 个 chunk。一个 200 页 PDF 约 600 个 chunk需要 2 到 3 秒的 GPU 时间。8 个 worker 并发时GPU 会排队实际吞吐约每分钟 2 到 3 个文档。如果要更高吞吐需要加 GPU 或者用更小的模型。6.3 瓶颈定位方法压测时用py-spy抓 worker 的调用栈看时间花在哪。如果大量时间在requests.post说明 embedding 服务是瓶颈如果在qdrant.upsert说明向量库是瓶颈如果在page.get_text说明解析是瓶颈。定位到瓶颈后针对性优化embedding 瓶颈就加 GPU 或换小模型向量库瓶颈就加索引或者分片解析瓶颈就换更快的解析库或者预处理。7. 一些延伸思考这套方案跑下来支撑团队内部几十人日常使用是没问题的。但如果要往更大规模走还有几个方向可以扩展。一是增量更新。现在每次上传都是全量处理如果文档只是小改全量重跑很浪费。可以做文档指纹比如内容哈希只处理变化的 chunk。这需要维护文档到 chunk 的映射关系复杂度会上升但能省很多算力。二是多模态支持。热词里有人问 RAG 知识库能不能存图片。可以但要用多模态 embedding 模型比如 CLIP把图片和文本映射到同一向量空间。这样检索时可以用文本搜图片也可以用图片搜文本。实现上比纯文本复杂主要是图片预处理和存储的成本。三是GraphRAG 和本体 RAG。这两个是最近比较热的方向核心思路是在向量检索之外引入实体和关系图谱解决知识割裂的问题。比如一个问题的答案分散在多个文档里纯向量检索可能只召回其中一部分而图谱能沿着关系把相关片段都找出来。这块我还在摸索等有成熟经验了再单独写一篇。四是检索质量评估。RAG 的 hit rate 不能靠感觉要建评估集。人工标注一批问题和对应的正确 chunk然后跑检索看召回率。有了评估集调参才有方向不然就是盲调。这套东西说到底核心就一句话把长任务拆成短任务把大内存拆成小内存把串行拆成并行。大文件并发之所以难是因为它同时挑战了这三个维度任何一个没处理好系统就会在压力下暴露问题。把链路拆清楚每个环节的瓶颈定位准参数调到位剩下的就是工程耐心了。
