1 项目背景业务场景「云帆科技」的系统平稳运行两个月后运维小李被问到三个灵魂问题竟然一个都答不上来CTO 问“一份文档上传后它的原始文件存在哪元数据存在哪向量存在哪”安全审计问“用户删了一个数据集所有关联的数据真的都清干净了吗MinIO 里的文件还在不在”性能调优问“为什么知识库问答有时候快0.5 秒、有时候慢8 秒同一个问题隔 10 分钟再问结果还不一样”小李发现她对 RAGFlow 内部四种存储MySQL、Redis、MinIO、文档引擎的分工并不清楚。她不知道一份文档的完整生命周期在这四种存储中是如何流转的也就无法回答任何关于数据一致性、性能优化、安全审计的问题。痛点不理解多存储协同机制的后果数据残留风险以为删了数据集就万事大吉其实 MinIO 里的原始文件和文档引擎里的向量都还在——合规审计时被发现已删除数据仍可访问。性能瓶颈不知问答慢不知道是 MySQL 查询慢、Redis 响应慢、MinIO 下载慢还是文档引擎检索慢——缺乏分段的性能视图。缓存不一致MySQL 中文档状态已经是success但 Redis 缓存里还是parsing导致控制台显示的状态延迟。跨存储事务缺失文档解析到一半MySQL 写入成功了但向量写入失败了——数据处于半成功的不一致状态。一份文档在四种存储中的完整足迹 用户上传 employee_handbook.pdf │ ├── MySQL: INSERT INTO document (id, name, statusuploading, ...) │ └── 元数据文件名、状态、大小、上传者、解析配置 │ ├── MinIO: PUT /tenant_01/ds_abc/doc_xyz/employee_handbook.pdf │ └── 原始文件 解析中间产物OCR缓存、缩略图等 │ ├── Redis: XADD ragflow_tasks {doc_id: xyz, action: parse} │ └── 任务队列 临时状态缓存 Session 数据 │ └── Infinity/ES: PUT /ds_abc/_doc/chunk_001 └── 文本内容 向量(1024维) 全文索引 元数据2 项目设计小胖举着手机上的四个图标“大师RAGFlow 用了 MySQL、Redis、MinIO 还有一个什么 Infinity——这四个东西各管什么我老搞混。就好比以前我们一个 MySQL 就搞定所有事。”大师“这就是典型的多存储分层架构。给你打个比方——你家里管东西也不会所有东西都塞一个柜子吧重要的文件放保险柜衣服放衣柜零食放冰箱零钱放手边抽屉。每样东西按它的用途和访问特性放在最合适的地方。”RAGFlow 四层存储 你家里的四类收纳 MySQL 家庭账本记录每笔收支、每件物品在哪 存什么元数据——用户、租户、文档名、状态、时间戳 特点精确查询、事务保证、数据稳定 Redis 手边的即时贴临时记录、用完就扔或很快过时 存什么任务队列、Session、临时缓存 特点极快、内存存储、可丢失可重建 MinIO 储物间放大件物品不经常翻 存什么原始文件、解析中间产物 特点大容量、对象存储、按路径存取 Infinity/ES 智能书柜可以按内容搜索而不是按书名 存什么Chunk 文本、向量、全文索引 特点语义搜索、全文检索、高维向量技术映射MySQL 账本精确记账Redis 便利贴快速临时MinIO 储物间大件物品Infinity 搜索引擎按内容找。小胖“那具体一份文档在各存储中是怎么流转的你能不能从头到尾串一遍”大师“来我们跟踪 employee_handbook.pdf 从上传到可检索的全过程”阶段1上传API Server 1. 用户点击上传 → 文件通过 HTTP multipart 到达 API Server 2. API Server 将文件写入 MinIO 路径: /tenant_01/dataset_abc/documents/doc_xyz/employee_handbook.pdf 3. API Server 在 MySQL 创建 Document 记录 INSERT INTO document (id, name, tenant_id, dataset_id, statusuploading, file_size2.5MB, created_atNOW()) 4. API Server 向 Redis 投递解析任务 XADD ragflow_tasks * doc_id doc_xyz action parse 阶段2解析Task Executor 5. Worker 从 Redis XREADGROUP 拉取任务 6. 更新 MySQL: status parsing 7. 从 MinIO 下载原始文件 8. DeepDoc 解析 切片 → 产物写回 MinIO (可选缓存OCR结果) 9. Embedding 将每个 Chunk 转为向量 10. 向量文本写入 Infinity/ES PUT /dataset_abc/_doc/chunk_001 {...} 11. 更新 MySQL: status success, chunk_count 47 阶段3检索问答API Server 12. 用户提问 → API Server 将问题向量化 13. 在 Infinity/ES 中做向量关键词混合检索 14. 可选从 Redis 读取缓存的检索结果热数据加速 15. LLM 生成答案 → 返回给用户 16. 聊天记录写入 MySQL小白眼睛从笔记本上移开“跨这么多存储怎么保证一致性比如第 10 步向量写入成功了但第 11 步 MySQL 更新失败了——这时候文档状态是什么”大师“这是分布式系统中的经典问题。RAGFlow 目前的策略是’最终一致性’”状态以 MySQL 为准控制台显示的文档状态始终从 MySQL 读取。向量以文档引擎为准如果向量写入成功但 MySQL 没更新成功文档状态仍是 ‘parsing’Task Executor 重启后会触发幂等重试——先清旧向量再重新写入。文件以 MinIO 为准删除文档时先删 MinIO 文件 → 删 MySQL 记录 → 异步删文档引擎索引。这是’先删源再删引用’的策略保证源没了引用不会悬空。“但不完善之处在于如果 MySQL 删成功了但 MinIO 删除失败网络抖动MinIO 中会产生孤儿文件。建议定期运行一致性检查脚本。”技术映射多存储一致性 跨店购物——你在一家店下单但要在三家店取货要确保账单(MySQL)、现货(MinIO)、配送单(Infinity)三者的信息一致。小胖“那性能呢为什么同一个问题有时 0.5 秒有时 8 秒”大师“四个存储中任何一个都可能成为瓶颈取决于当时的状态”慢的可能环节症状定位方法MySQL 慢查询API 列表页加载中转很久SHOW PROCESSLIST看慢查询Redis 延迟任务入队/出队慢redis-cli --latencyMinIO 下载慢首次检索大文档慢第二次快mc admin trace看下载耗时Infinity/ES 慢检索请求本身慢向量搜索耗时查看检索日志的took字段LLM API 超时生成答案阶段慢API Server 中 LLM 调用耗时“其中最常见的性能差异来源是’冷热数据’——第一次检索某个数据集时索引数据没有在文档引擎的内存中OS Page Cache需要从磁盘加载慢。第二次检索时数据已在内存中快。这就是为什么同一个问题问两次第二次明显快。”技术映射冷热数据 冰箱里的菜——第一次拿要从冰格深处翻慢拿出来放在台面上热数据第二次拿随手就够到了快。小白“那怎么监控这四个存储的健康状态”大师“关键监控指标”四存储健康监控看板 MySQL: ├── 连接数 (Threads_connected): 100 ├── 慢查询 ( 1s): 5/分钟 ├── 复制延迟 (如果是主从): 3 秒 └── 磁盘使用率: 80% Redis: ├── 内存使用率: 70% ├── Stream 队列长度: 50 ├── PEL 积压: 10 ├── 连接数: 200 └── 延迟: 2ms MinIO: ├── 磁盘使用率: 80% ├── 上传/下载吞吐: 正常范围 └── 健康检查: /minio/health/live 200 Infinity/ES: ├── 集群状态: green (非 yellow/red) ├── 索引文档数: 按预期增长 ├── 查询延迟 P95: 500ms └── JVM 堆使用率: 75% (仅 ES)3 项目实战环境准备目标跟踪一份文档在四种存储中的完整足迹理解数据流转。前提RAGFlow 已部署工具已安装mysql、redis-cli、mcMinIO Client、curl。分步实现步骤1上传前——记录初始状态目标创建数据集记录四种存储的初始快照。# 初始快照脚本DS_NAME数据流转测试-$(date%s)# 1. MySQL 初始状态echo MySQL 初始状态 dockerexecragflow-mysql mysql-uroot -pxxxragflow-e\SELECT COUNT(*) AS doc_count FROM document WHERE name LIKE %测试%;# 2. Redis 初始状态echo Redis 初始状态 dockerexecragflow-redis redis-cli XLEN ragflow_tasks# 3. MinIO 初始状态echo MinIO 初始状态 # 需要先配置 mc 客户端# mc alias set ragflow-minio http://localhost:9000 user passwordmclsragflow-minio/ragflow/2/dev/null# 4. Infinity 初始状态echo Infinity 初始状态 curl-shttp://localhost:23820/admin/node/status|jq.data步骤2上传——观察四种存储的写入目标上传文档后立即在各存储中查找足迹。# step2_trace_upload.py - 追踪上传importtimefromragflowimportRAGFlow ragRAGFlow(api_keyxxx,base_urlhttp://localhost/api/v1)dsrag.create_dataset(name数据流转测试)# 上传前时间戳upload_starttime.time()docds.upload_document(test_docs/考勤管理办法.pdf)print(f文档ID:{doc.id})# 步骤1立即查 MySQL应该能看到记录状态uploadingimportsubprocess resultsubprocess.run([docker,exec,ragflow-mysql,mysql,-u,root,f-p{DB_PASS},ragflow,-N,-e,fSELECT id, name, status FROM document WHERE id{doc.id}],capture_outputTrue,textTrue)print(fMySQL:{result.stdout.strip()})# 步骤2查 MinIO应该已有文件print(MinIO 检查: 在 tenant_*/dataset_*/documents/{doc.id}/ 目录下应有原文件)# 步骤3查 Redis应该有任务消息resultsubprocess.run([docker,exec,ragflow-redis,redis-cli,XRANGE,ragflow_tasks,-,],capture_outputTrue,textTrue)# 在 XRANGE 输出中搜索 doc.idtask_countresult.stdout.count(doc.id)print(fRedis 队列中包含该文档的任务数:{task_count})# 步骤4等待解析完成后查 Infinitytime.sleep(60)# 等解析完成docds.get_document(doc.id)print(f解析状态:{doc.status}, 切片数:{doc.chunk_count})# Infinity 中应有对应的索引数据# curl -X GET http://localhost:23820/db/ragflow/table/{dataset_id}/data?limit5步骤3删除——验证数据清理完整性目标删除数据集后检查四存储是否都有清理。# step3_verify_deletion.shDS_IDyour-dataset-idecho 删除前快照 echoMySQL 文档数:dockerexecragflow-mysql mysql-uroot -pxxxragflow-N-e\SELECT COUNT(*) FROM document WHERE dataset_id$DS_ID;echoRedis 队列长度:dockerexecragflow-redis redis-cli XLEN ragflow_tasksecho执行删除...curl-XDELETEhttp://localhost/api/v1/datasets/$DS_ID\-HAuthorization: Bearer$TOKENsleep10# 等异步清理完成echoecho 删除后检查 echoMySQL 文档数 (应为 0):dockerexecragflow-mysql mysql-uroot -pxxxragflow-N-e\SELECT COUNT(*) FROM document WHERE dataset_id$DS_ID;echoMinIO 残留检查:# 检查该 dataset 的 MinIO 目录是否被清理# mc ls ragflow-minio/ragflow/tenant_*/$DS_ID/ 应该报错 No such fileechoInfinity 索引检查:# 检查 Infinity 中该数据集的表是否被删除# curl http://localhost:23820/db/ragflow/table/$DS_ID 应该返回 404坑点删除操作不是瞬时的——MySQL 先标记删除后台异步任务逐步清理 MinIO 文件大文件可能延时和 Infinity 索引。10 秒的等待可能不够。步骤4一致性巡检脚本目标定期检查四存储间的数据一致性。# step4_consistency_check.py - 一致性巡检importsubprocessimportjsondefcheck_consistency():issues[]# 1. MySQL 中有记录但 MinIO 无文件# 查询所有 statussuccess 的文档IDdocsquery_mysql(SELECT id, name FROM document WHERE statussuccess LIMIT 100)fordoc_id,doc_nameindocs:minio_pathftenant_*/dataset_*/documents/{doc_id}/{doc_name}ifnotminio_file_exists(minio_path):issues.append(f[孤儿记录] MySQL有但MinIO无:{doc_id}/{doc_name})# 2. MySQL 有记录但在 Infinitity 中无索引# 查询成功且 chunk_count 0 的文档docsquery_mysql(SELECT d.id, d.dataset_id FROM document d WHERE d.statussuccess AND d.chunk_count 0)fordoc_id,ds_idindocs:chunk_countget_infinity_chunk_count(ds_id,doc_id)mysql_chunk_countquery_mysql(fSELECT chunk_count FROM document WHERE id{doc_id})ifchunk_count!mysql_chunk_count:issues.append(f[向量丢失] doc{doc_id}, MySQL chunks{mysql_chunk_count}, fInfinity chunks{chunk_count})# 3. MinIO 有文件但 MySQL 无记录孤儿文件minio_fileslist_all_minio_files()forpathinminio_files:doc_idextract_doc_id_from_path(path)exists_in_mysqlquery_mysql_exists(fSELECT 1 FROM document WHERE id{doc_id})ifnotexists_in_mysql:issues.append(f[孤儿文件] MinIO有但MySQL无:{path})returnissuesdefgenerate_consistency_report(issues):生成一致性报告ifnotissues:print(✅ 数据一致性检查通过)returnprint(f⚠ 发现{len(issues)}个一致性问题:)fori,issueinenumerate(issues,1):print(f{i}.{issue})# 自动化修复建议orphan_records[iforiinissuesif孤儿记录ini]orphan_files[iforiinissuesif孤儿文件ini]vector_loss[iforiinissuesif向量丢失ini]print(f\n修复建议:)iforphan_records:print(f -{len(orphan_records)}个孤儿记录可安全删除 MySQL 记录)iforphan_files:print(f -{len(orphan_files)}个孤儿文件可安全删除 MinIO 文件 f(或触发重新索引))ifvector_loss:print(f -{len(vector_loss)}个向量丢失触发文档重新解析)# 运行巡检issuescheck_consistency()generate_consistency_report(issues)步骤5跨存储性能分析目标定位一次问答请求中各存储环节的耗时。# step5_perf_breakdown.py - 跨存储耗时分析importtimeimportrequestsfromcontextlibimportcontextmanagercontextmanagerdeftimer(label):starttime.time()yieldelapsed(time.time()-start)*1000print(f [{label}]{elapsed:.0f}ms)defprofile_chat_request(question,dataset_id):剖面分析一次问答请求的各环节耗时print(f\n 剖面分析:{question[:30]}... )total_starttime.time()# 阶段1问题 Embedding调用 LLM APIwithtimer(Embedding):# 这步在 API Server 内部完成pass# 阶段2向量检索Infinity/ESwithtimer(向量检索 (Infinity)):# 模拟curl Infinity 的 /search 端点pass# 阶段3关键词检索Infinity/ESwithtimer(关键词检索 (Infinity)):pass# 阶段4Rerank 重排序withtimer(Rerank):pass# 阶段5LLM 生成答案withtimer(LLM 生成):pass# 阶段6结果组装与返回withtimer(结果组装):passtotal(time.time()-total_start)*1000print(f 总耗时:{total:.0f}ms)# curl 方式获取检索延迟# curl -s http://localhost:23820/db/ragflow/table/ds_id/search \# -H Content-Type: application/json \# -d {query: {match: {content: 年假}}} | jq .took测试验证# test_storage_integration.pyimportpytestclassTestStorageConsistency:deftest_upload_creates_all_records(self):验证上传后在所有存储中都有对应记录docds.upload_document(test.pdf)time.sleep(30)# 等待流转# MySQLmysql_recordquery_mysql(fSELECT * FROM document WHERE id{doc.id})assertmysql_recordisnotNone,MySQL 中无记录# MinIO (通过 SDK 间接验证如果文件不存在获取文档时会报错)retrievedds.get_document(doc.id)assertretrievedisnotNone,无法获取文档MinIO 可能异常# Redis 任务应已被消费队列中无残留redis_hasredis_has_task(doc.id)assertnotredis_has,fRedis 队列中仍有该文档任务:{doc.id}deftest_delete_cleans_all_stores(self):验证删除在四种存储中都被清理# ... 删除前后的对比pass完整代码清单路径说明api/db/db_models.pyMySQL/Peewee 模型定义Document, File 等rag/utils/ob_conn.pyInfinity 连接器向量关键词检索rag/utils/storage_factory.py存储工厂ES/Infinity 二选一api/db/services/file_service.py文件服务MinIO 操作封装rag/svr/task_executor.py任务执行串联四种存储4 项目总结优点 缺点维度四层存储分体单体 PostgreSQLFirebase 全家桶AWS 托管服务各组件最优匹配★★★ 专用即最优★★☆ 统一但平庸★★☆ 集成方便★★★ 托管优化运维复杂度★★☆ 4 个组件★★★ 1 个组件★★★ 无运维★★☆ 托管减负数据一致性★★☆ 最终一致★★★ 强一致★★★ 强一致★★★ 强一致扩展性★★★ 独立扩缩★★☆ 垂直扩展★★☆ 平台限制★★★ 弹性成本★★★ 开源免费★★★ 免费★★☆ 免费额度有限★★☆ 按量付费适用场景需要专用存储的中型项目向量存文档引擎、文件存对象存储、元数据存关系型——各取所长。数据量大、访问模式多样化文件是顺序大 IOMinIO元数据是随机小 IOMySQL向量是相似度计算Infinity。需要独立扩展特定组件检索请求量大 → 扩 Infinity 节点文件多 → 扩 MinIO 磁盘。合规审计要求需要追踪文档在各存储中的完整生命周期能回答删干净了吗。成本敏感的开源方案全部用开源组件搭建无云厂商锁定。不适用场景极简部署需求如果只需要管理几十份文档四层存储是过度设计。需要跨存储事务MySQL 和 MinIO 之间的操作不支持分布式事务如写入文件数据库记录原子性。注意事项删除是异步的删除数据集后MinIO 文件和 Infinity 索引的清理是后台任务可能有分钟级延迟。缓存一致性陷阱Redis 中的缓存数据如文档状态与 MySQL 可能存在短暂不一致——控制台显示的解析中可能已经实际完成了。Infinity 与 ES 的切换成本从 ES 切换到 Infinity 后已有数据不互通必须全量重建。MinIO 桶策略生产环境 MinIO 必须关闭匿名访问API Server 作为唯一入口做鉴权。数据库连接池Peewee ORM 默认的连接池较小高并发时 MySQL 可能报 “Too many connections”。常见踩坑经验故障现象根因解决方法控制台显示解析成功但检索为空MySQL 状态更新了但 Infinity 写入失败查 Task Executor 日志中的索引写入 ERROR文档删除后 MinIO 文件仍在异步清理任务失败或延迟手动清理或等待下一次 GCInfinity 查询突然变慢 10 倍索引从未优化磁盘碎片累积触发 Infinity 的 optimize 操作MySQL 连接数耗尽连接池未释放或 API Server 副本数过多增加max_connections优化连接池回收策略Redis 缓存数据与 MySQL 不一致缓存更新策略不是先写 MySQL 再失效 Redis确保代码中遵循 Cache-Aside 模式思考题RAGFlow 的删除操作涉及四个存储的联动清理MySQL、Redis、MinIO、Infinity。如果这四个清理步骤中的任意一个失败了如网络抖动就会出现幽灵数据。请设计一个完整的分布式删除 Saga方案包含正向操作、补偿操作和最终一致性检查。公司文档量从 1000 份增长到 10 万份后Infinity 的检索延迟从 200ms 增长到了 2 秒。除了垂直扩容加内存之外你如何利用 MySQL 中的元数据为 Infinity 做分区路由——比如只搜索最近 30 天新增的文档答案提示见第20章末尾或附录 D。延伸阅读与资源10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析
第19章：RAGFlow四种存储MySQL/PostgreSQL、Redis、MinIO 与文档引擎协作
1 项目背景业务场景「云帆科技」的系统平稳运行两个月后运维小李被问到三个灵魂问题竟然一个都答不上来CTO 问“一份文档上传后它的原始文件存在哪元数据存在哪向量存在哪”安全审计问“用户删了一个数据集所有关联的数据真的都清干净了吗MinIO 里的文件还在不在”性能调优问“为什么知识库问答有时候快0.5 秒、有时候慢8 秒同一个问题隔 10 分钟再问结果还不一样”小李发现她对 RAGFlow 内部四种存储MySQL、Redis、MinIO、文档引擎的分工并不清楚。她不知道一份文档的完整生命周期在这四种存储中是如何流转的也就无法回答任何关于数据一致性、性能优化、安全审计的问题。痛点不理解多存储协同机制的后果数据残留风险以为删了数据集就万事大吉其实 MinIO 里的原始文件和文档引擎里的向量都还在——合规审计时被发现已删除数据仍可访问。性能瓶颈不知问答慢不知道是 MySQL 查询慢、Redis 响应慢、MinIO 下载慢还是文档引擎检索慢——缺乏分段的性能视图。缓存不一致MySQL 中文档状态已经是success但 Redis 缓存里还是parsing导致控制台显示的状态延迟。跨存储事务缺失文档解析到一半MySQL 写入成功了但向量写入失败了——数据处于半成功的不一致状态。一份文档在四种存储中的完整足迹 用户上传 employee_handbook.pdf │ ├── MySQL: INSERT INTO document (id, name, statusuploading, ...) │ └── 元数据文件名、状态、大小、上传者、解析配置 │ ├── MinIO: PUT /tenant_01/ds_abc/doc_xyz/employee_handbook.pdf │ └── 原始文件 解析中间产物OCR缓存、缩略图等 │ ├── Redis: XADD ragflow_tasks {doc_id: xyz, action: parse} │ └── 任务队列 临时状态缓存 Session 数据 │ └── Infinity/ES: PUT /ds_abc/_doc/chunk_001 └── 文本内容 向量(1024维) 全文索引 元数据2 项目设计小胖举着手机上的四个图标“大师RAGFlow 用了 MySQL、Redis、MinIO 还有一个什么 Infinity——这四个东西各管什么我老搞混。就好比以前我们一个 MySQL 就搞定所有事。”大师“这就是典型的多存储分层架构。给你打个比方——你家里管东西也不会所有东西都塞一个柜子吧重要的文件放保险柜衣服放衣柜零食放冰箱零钱放手边抽屉。每样东西按它的用途和访问特性放在最合适的地方。”RAGFlow 四层存储 你家里的四类收纳 MySQL 家庭账本记录每笔收支、每件物品在哪 存什么元数据——用户、租户、文档名、状态、时间戳 特点精确查询、事务保证、数据稳定 Redis 手边的即时贴临时记录、用完就扔或很快过时 存什么任务队列、Session、临时缓存 特点极快、内存存储、可丢失可重建 MinIO 储物间放大件物品不经常翻 存什么原始文件、解析中间产物 特点大容量、对象存储、按路径存取 Infinity/ES 智能书柜可以按内容搜索而不是按书名 存什么Chunk 文本、向量、全文索引 特点语义搜索、全文检索、高维向量技术映射MySQL 账本精确记账Redis 便利贴快速临时MinIO 储物间大件物品Infinity 搜索引擎按内容找。小胖“那具体一份文档在各存储中是怎么流转的你能不能从头到尾串一遍”大师“来我们跟踪 employee_handbook.pdf 从上传到可检索的全过程”阶段1上传API Server 1. 用户点击上传 → 文件通过 HTTP multipart 到达 API Server 2. API Server 将文件写入 MinIO 路径: /tenant_01/dataset_abc/documents/doc_xyz/employee_handbook.pdf 3. API Server 在 MySQL 创建 Document 记录 INSERT INTO document (id, name, tenant_id, dataset_id, statusuploading, file_size2.5MB, created_atNOW()) 4. API Server 向 Redis 投递解析任务 XADD ragflow_tasks * doc_id doc_xyz action parse 阶段2解析Task Executor 5. Worker 从 Redis XREADGROUP 拉取任务 6. 更新 MySQL: status parsing 7. 从 MinIO 下载原始文件 8. DeepDoc 解析 切片 → 产物写回 MinIO (可选缓存OCR结果) 9. Embedding 将每个 Chunk 转为向量 10. 向量文本写入 Infinity/ES PUT /dataset_abc/_doc/chunk_001 {...} 11. 更新 MySQL: status success, chunk_count 47 阶段3检索问答API Server 12. 用户提问 → API Server 将问题向量化 13. 在 Infinity/ES 中做向量关键词混合检索 14. 可选从 Redis 读取缓存的检索结果热数据加速 15. LLM 生成答案 → 返回给用户 16. 聊天记录写入 MySQL小白眼睛从笔记本上移开“跨这么多存储怎么保证一致性比如第 10 步向量写入成功了但第 11 步 MySQL 更新失败了——这时候文档状态是什么”大师“这是分布式系统中的经典问题。RAGFlow 目前的策略是’最终一致性’”状态以 MySQL 为准控制台显示的文档状态始终从 MySQL 读取。向量以文档引擎为准如果向量写入成功但 MySQL 没更新成功文档状态仍是 ‘parsing’Task Executor 重启后会触发幂等重试——先清旧向量再重新写入。文件以 MinIO 为准删除文档时先删 MinIO 文件 → 删 MySQL 记录 → 异步删文档引擎索引。这是’先删源再删引用’的策略保证源没了引用不会悬空。“但不完善之处在于如果 MySQL 删成功了但 MinIO 删除失败网络抖动MinIO 中会产生孤儿文件。建议定期运行一致性检查脚本。”技术映射多存储一致性 跨店购物——你在一家店下单但要在三家店取货要确保账单(MySQL)、现货(MinIO)、配送单(Infinity)三者的信息一致。小胖“那性能呢为什么同一个问题有时 0.5 秒有时 8 秒”大师“四个存储中任何一个都可能成为瓶颈取决于当时的状态”慢的可能环节症状定位方法MySQL 慢查询API 列表页加载中转很久SHOW PROCESSLIST看慢查询Redis 延迟任务入队/出队慢redis-cli --latencyMinIO 下载慢首次检索大文档慢第二次快mc admin trace看下载耗时Infinity/ES 慢检索请求本身慢向量搜索耗时查看检索日志的took字段LLM API 超时生成答案阶段慢API Server 中 LLM 调用耗时“其中最常见的性能差异来源是’冷热数据’——第一次检索某个数据集时索引数据没有在文档引擎的内存中OS Page Cache需要从磁盘加载慢。第二次检索时数据已在内存中快。这就是为什么同一个问题问两次第二次明显快。”技术映射冷热数据 冰箱里的菜——第一次拿要从冰格深处翻慢拿出来放在台面上热数据第二次拿随手就够到了快。小白“那怎么监控这四个存储的健康状态”大师“关键监控指标”四存储健康监控看板 MySQL: ├── 连接数 (Threads_connected): 100 ├── 慢查询 ( 1s): 5/分钟 ├── 复制延迟 (如果是主从): 3 秒 └── 磁盘使用率: 80% Redis: ├── 内存使用率: 70% ├── Stream 队列长度: 50 ├── PEL 积压: 10 ├── 连接数: 200 └── 延迟: 2ms MinIO: ├── 磁盘使用率: 80% ├── 上传/下载吞吐: 正常范围 └── 健康检查: /minio/health/live 200 Infinity/ES: ├── 集群状态: green (非 yellow/red) ├── 索引文档数: 按预期增长 ├── 查询延迟 P95: 500ms └── JVM 堆使用率: 75% (仅 ES)3 项目实战环境准备目标跟踪一份文档在四种存储中的完整足迹理解数据流转。前提RAGFlow 已部署工具已安装mysql、redis-cli、mcMinIO Client、curl。分步实现步骤1上传前——记录初始状态目标创建数据集记录四种存储的初始快照。# 初始快照脚本DS_NAME数据流转测试-$(date%s)# 1. MySQL 初始状态echo MySQL 初始状态 dockerexecragflow-mysql mysql-uroot -pxxxragflow-e\SELECT COUNT(*) AS doc_count FROM document WHERE name LIKE %测试%;# 2. Redis 初始状态echo Redis 初始状态 dockerexecragflow-redis redis-cli XLEN ragflow_tasks# 3. MinIO 初始状态echo MinIO 初始状态 # 需要先配置 mc 客户端# mc alias set ragflow-minio http://localhost:9000 user passwordmclsragflow-minio/ragflow/2/dev/null# 4. Infinity 初始状态echo Infinity 初始状态 curl-shttp://localhost:23820/admin/node/status|jq.data步骤2上传——观察四种存储的写入目标上传文档后立即在各存储中查找足迹。# step2_trace_upload.py - 追踪上传importtimefromragflowimportRAGFlow ragRAGFlow(api_keyxxx,base_urlhttp://localhost/api/v1)dsrag.create_dataset(name数据流转测试)# 上传前时间戳upload_starttime.time()docds.upload_document(test_docs/考勤管理办法.pdf)print(f文档ID:{doc.id})# 步骤1立即查 MySQL应该能看到记录状态uploadingimportsubprocess resultsubprocess.run([docker,exec,ragflow-mysql,mysql,-u,root,f-p{DB_PASS},ragflow,-N,-e,fSELECT id, name, status FROM document WHERE id{doc.id}],capture_outputTrue,textTrue)print(fMySQL:{result.stdout.strip()})# 步骤2查 MinIO应该已有文件print(MinIO 检查: 在 tenant_*/dataset_*/documents/{doc.id}/ 目录下应有原文件)# 步骤3查 Redis应该有任务消息resultsubprocess.run([docker,exec,ragflow-redis,redis-cli,XRANGE,ragflow_tasks,-,],capture_outputTrue,textTrue)# 在 XRANGE 输出中搜索 doc.idtask_countresult.stdout.count(doc.id)print(fRedis 队列中包含该文档的任务数:{task_count})# 步骤4等待解析完成后查 Infinitytime.sleep(60)# 等解析完成docds.get_document(doc.id)print(f解析状态:{doc.status}, 切片数:{doc.chunk_count})# Infinity 中应有对应的索引数据# curl -X GET http://localhost:23820/db/ragflow/table/{dataset_id}/data?limit5步骤3删除——验证数据清理完整性目标删除数据集后检查四存储是否都有清理。# step3_verify_deletion.shDS_IDyour-dataset-idecho 删除前快照 echoMySQL 文档数:dockerexecragflow-mysql mysql-uroot -pxxxragflow-N-e\SELECT COUNT(*) FROM document WHERE dataset_id$DS_ID;echoRedis 队列长度:dockerexecragflow-redis redis-cli XLEN ragflow_tasksecho执行删除...curl-XDELETEhttp://localhost/api/v1/datasets/$DS_ID\-HAuthorization: Bearer$TOKENsleep10# 等异步清理完成echoecho 删除后检查 echoMySQL 文档数 (应为 0):dockerexecragflow-mysql mysql-uroot -pxxxragflow-N-e\SELECT COUNT(*) FROM document WHERE dataset_id$DS_ID;echoMinIO 残留检查:# 检查该 dataset 的 MinIO 目录是否被清理# mc ls ragflow-minio/ragflow/tenant_*/$DS_ID/ 应该报错 No such fileechoInfinity 索引检查:# 检查 Infinity 中该数据集的表是否被删除# curl http://localhost:23820/db/ragflow/table/$DS_ID 应该返回 404坑点删除操作不是瞬时的——MySQL 先标记删除后台异步任务逐步清理 MinIO 文件大文件可能延时和 Infinity 索引。10 秒的等待可能不够。步骤4一致性巡检脚本目标定期检查四存储间的数据一致性。# step4_consistency_check.py - 一致性巡检importsubprocessimportjsondefcheck_consistency():issues[]# 1. MySQL 中有记录但 MinIO 无文件# 查询所有 statussuccess 的文档IDdocsquery_mysql(SELECT id, name FROM document WHERE statussuccess LIMIT 100)fordoc_id,doc_nameindocs:minio_pathftenant_*/dataset_*/documents/{doc_id}/{doc_name}ifnotminio_file_exists(minio_path):issues.append(f[孤儿记录] MySQL有但MinIO无:{doc_id}/{doc_name})# 2. MySQL 有记录但在 Infinitity 中无索引# 查询成功且 chunk_count 0 的文档docsquery_mysql(SELECT d.id, d.dataset_id FROM document d WHERE d.statussuccess AND d.chunk_count 0)fordoc_id,ds_idindocs:chunk_countget_infinity_chunk_count(ds_id,doc_id)mysql_chunk_countquery_mysql(fSELECT chunk_count FROM document WHERE id{doc_id})ifchunk_count!mysql_chunk_count:issues.append(f[向量丢失] doc{doc_id}, MySQL chunks{mysql_chunk_count}, fInfinity chunks{chunk_count})# 3. MinIO 有文件但 MySQL 无记录孤儿文件minio_fileslist_all_minio_files()forpathinminio_files:doc_idextract_doc_id_from_path(path)exists_in_mysqlquery_mysql_exists(fSELECT 1 FROM document WHERE id{doc_id})ifnotexists_in_mysql:issues.append(f[孤儿文件] MinIO有但MySQL无:{path})returnissuesdefgenerate_consistency_report(issues):生成一致性报告ifnotissues:print(✅ 数据一致性检查通过)returnprint(f⚠ 发现{len(issues)}个一致性问题:)fori,issueinenumerate(issues,1):print(f{i}.{issue})# 自动化修复建议orphan_records[iforiinissuesif孤儿记录ini]orphan_files[iforiinissuesif孤儿文件ini]vector_loss[iforiinissuesif向量丢失ini]print(f\n修复建议:)iforphan_records:print(f -{len(orphan_records)}个孤儿记录可安全删除 MySQL 记录)iforphan_files:print(f -{len(orphan_files)}个孤儿文件可安全删除 MinIO 文件 f(或触发重新索引))ifvector_loss:print(f -{len(vector_loss)}个向量丢失触发文档重新解析)# 运行巡检issuescheck_consistency()generate_consistency_report(issues)步骤5跨存储性能分析目标定位一次问答请求中各存储环节的耗时。# step5_perf_breakdown.py - 跨存储耗时分析importtimeimportrequestsfromcontextlibimportcontextmanagercontextmanagerdeftimer(label):starttime.time()yieldelapsed(time.time()-start)*1000print(f [{label}]{elapsed:.0f}ms)defprofile_chat_request(question,dataset_id):剖面分析一次问答请求的各环节耗时print(f\n 剖面分析:{question[:30]}... )total_starttime.time()# 阶段1问题 Embedding调用 LLM APIwithtimer(Embedding):# 这步在 API Server 内部完成pass# 阶段2向量检索Infinity/ESwithtimer(向量检索 (Infinity)):# 模拟curl Infinity 的 /search 端点pass# 阶段3关键词检索Infinity/ESwithtimer(关键词检索 (Infinity)):pass# 阶段4Rerank 重排序withtimer(Rerank):pass# 阶段5LLM 生成答案withtimer(LLM 生成):pass# 阶段6结果组装与返回withtimer(结果组装):passtotal(time.time()-total_start)*1000print(f 总耗时:{total:.0f}ms)# curl 方式获取检索延迟# curl -s http://localhost:23820/db/ragflow/table/ds_id/search \# -H Content-Type: application/json \# -d {query: {match: {content: 年假}}} | jq .took测试验证# test_storage_integration.pyimportpytestclassTestStorageConsistency:deftest_upload_creates_all_records(self):验证上传后在所有存储中都有对应记录docds.upload_document(test.pdf)time.sleep(30)# 等待流转# MySQLmysql_recordquery_mysql(fSELECT * FROM document WHERE id{doc.id})assertmysql_recordisnotNone,MySQL 中无记录# MinIO (通过 SDK 间接验证如果文件不存在获取文档时会报错)retrievedds.get_document(doc.id)assertretrievedisnotNone,无法获取文档MinIO 可能异常# Redis 任务应已被消费队列中无残留redis_hasredis_has_task(doc.id)assertnotredis_has,fRedis 队列中仍有该文档任务:{doc.id}deftest_delete_cleans_all_stores(self):验证删除在四种存储中都被清理# ... 删除前后的对比pass完整代码清单路径说明api/db/db_models.pyMySQL/Peewee 模型定义Document, File 等rag/utils/ob_conn.pyInfinity 连接器向量关键词检索rag/utils/storage_factory.py存储工厂ES/Infinity 二选一api/db/services/file_service.py文件服务MinIO 操作封装rag/svr/task_executor.py任务执行串联四种存储4 项目总结优点 缺点维度四层存储分体单体 PostgreSQLFirebase 全家桶AWS 托管服务各组件最优匹配★★★ 专用即最优★★☆ 统一但平庸★★☆ 集成方便★★★ 托管优化运维复杂度★★☆ 4 个组件★★★ 1 个组件★★★ 无运维★★☆ 托管减负数据一致性★★☆ 最终一致★★★ 强一致★★★ 强一致★★★ 强一致扩展性★★★ 独立扩缩★★☆ 垂直扩展★★☆ 平台限制★★★ 弹性成本★★★ 开源免费★★★ 免费★★☆ 免费额度有限★★☆ 按量付费适用场景需要专用存储的中型项目向量存文档引擎、文件存对象存储、元数据存关系型——各取所长。数据量大、访问模式多样化文件是顺序大 IOMinIO元数据是随机小 IOMySQL向量是相似度计算Infinity。需要独立扩展特定组件检索请求量大 → 扩 Infinity 节点文件多 → 扩 MinIO 磁盘。合规审计要求需要追踪文档在各存储中的完整生命周期能回答删干净了吗。成本敏感的开源方案全部用开源组件搭建无云厂商锁定。不适用场景极简部署需求如果只需要管理几十份文档四层存储是过度设计。需要跨存储事务MySQL 和 MinIO 之间的操作不支持分布式事务如写入文件数据库记录原子性。注意事项删除是异步的删除数据集后MinIO 文件和 Infinity 索引的清理是后台任务可能有分钟级延迟。缓存一致性陷阱Redis 中的缓存数据如文档状态与 MySQL 可能存在短暂不一致——控制台显示的解析中可能已经实际完成了。Infinity 与 ES 的切换成本从 ES 切换到 Infinity 后已有数据不互通必须全量重建。MinIO 桶策略生产环境 MinIO 必须关闭匿名访问API Server 作为唯一入口做鉴权。数据库连接池Peewee ORM 默认的连接池较小高并发时 MySQL 可能报 “Too many connections”。常见踩坑经验故障现象根因解决方法控制台显示解析成功但检索为空MySQL 状态更新了但 Infinity 写入失败查 Task Executor 日志中的索引写入 ERROR文档删除后 MinIO 文件仍在异步清理任务失败或延迟手动清理或等待下一次 GCInfinity 查询突然变慢 10 倍索引从未优化磁盘碎片累积触发 Infinity 的 optimize 操作MySQL 连接数耗尽连接池未释放或 API Server 副本数过多增加max_connections优化连接池回收策略Redis 缓存数据与 MySQL 不一致缓存更新策略不是先写 MySQL 再失效 Redis确保代码中遵循 Cache-Aside 模式思考题RAGFlow 的删除操作涉及四个存储的联动清理MySQL、Redis、MinIO、Infinity。如果这四个清理步骤中的任意一个失败了如网络抖动就会出现幽灵数据。请设计一个完整的分布式删除 Saga方案包含正向操作、补偿操作和最终一致性检查。公司文档量从 1000 份增长到 10 万份后Infinity 的检索延迟从 200ms 增长到了 2 秒。除了垂直扩容加内存之外你如何利用 MySQL 中的元数据为 Infinity 做分区路由——比如只搜索最近 30 天新增的文档答案提示见第20章末尾或附录 D。延伸阅读与资源10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析
