图_1数据存储模块 · 详细实现
对应 [[1_项目设计]]。保留全部文件、类与逐函数分步内容。
读图方式
- 蓝底是文件，紫底是类，灰底是函数；框宽按文字自适应，标题与分步文本分开。
- 每个函数 / 类有唯一编号，例如 F01 / C01；步骤末尾
〔→ F07〕表示调用或依赖对应编号。P 编号表示文件；P02:3精确定位到 P02 文件的第 3 步，F04:2定位函数 F04 的第 2 步。 - 只有相邻、能在文件内短距离连接且不穿越文字 / 其他函数框的关系画曲线。其余保留编号引用。重复使用的公共助手不再扇出多条长线。
- 编号仅用于查找，不表示全局执行顺序；每个框里的 1、2、3 才是该框的执行步骤。总图另有问答与入库主链。
文件与函数索引
P01 · backend/env.py
统一加载项目根目录环境变量。
- F01
load_env()（源码 10 行）：只加载一次 .env。
P02 · backend/infra/database.py
配置 PostgreSQL 引擎、会话工厂与 ORM 基类。
- F02
_clean_nul_chars()（源码 25 行）：递归清洗字符串和容器参数。 - F03
before_cursor_execute()（源码 60 行）：在底层 SQL 执行前净化绑定参数。 - F04
init_db()（源码 83 行）：注册模型并创建尚不存在的数据表。
P03 · backend/infra/cache.py
统一 Redis JSON 缓存与错误降级。
- F05
RedisCache.__init__()（源码 9 行）：保存连接配置与默认缓存参数。 - F06
RedisCache._get_client()（源码 15 行）：按需创建并复用 Redis 客户端。 - F07
RedisCache._key()（源码 20 行）：生成带项目命名空间的缓存键。 - F08
RedisCache.get_json()（源码 23 行）：读取并反序列化缓存。 - F09
RedisCache.set_json()（源码 32 行）：序列化对象并带过期时间写入。 - F10
RedisCache.delete()（源码 39 行）：删除一个缓存键。 - F11
RedisCache.delete_pattern()（源码 45 行）：按模式批量删除缓存键。 - C01
RedisCache：保存连接配置并提供缓存接口。
P04 · backend/db/models.py
向同一个 Base.metadata 注册四张 ORM 表。
- C02
User：声明账号、角色和用户会话关系。 - C03
ChatSession：声明用户会话与会话元数据。 - C04
ChatMessage：声明消息与可选 RAG Trace。 - C05
ParentChunk：声明 L1 / L2 父块全文及关联标识。
P05 · backend/indexing/milvus_client.py
无状态 Store 配合短生命周期 Milvus 连接。
- F12
MilvusSettings.from_env()（源码 24 行）：从环境变量生成不可变 Milvus 配置。 - F13
milvus_client_session()（源码 39 行）：在一次上下文中管理客户端连接。 - F14
_normalize_filter()（源码 49 行）：避免将空表达式交给普通查询。 - F15
MilvusStore.__init__()（源码 56 行）：保存配置，不持有长期连接。 - F16
MilvusStore.collection_name()（源码 60 行）：提供只读 Collection 名称。 - F17
MilvusStore._run()（源码 63 行）：让一个 RPC 回调在短连接内运行。 - F18
MilvusStore.session()（源码 68 行）：向一段业务流程提供同一条连接。 - F19
MilvusStore.ensure_collection()（源码 74 行）：首次创建 Schema、BM25 Function 与双索引。 - F20
MilvusStore.init_collection()（源码 127 行）：解析向量维度并提交建集合回调。 - F21
MilvusStore.init_collection._init()（源码 131 行）：在实际连接中确保集合存在。 - F22
MilvusStore.insert()（源码 136 行）：通过短连接插入一批记录。 - F23
MilvusStore.query()（源码 139 行）：准备单页普通查询并执行回调。 - F24
MilvusStore.query._query()（源码 149 行）：在客户端上执行单页查询。 - F25
MilvusStore.query_all()（源码 160 行）：准备分页拉取并复用同一次连接。 - F26
MilvusStore.query_all._query_all()（源码 165 行）：按 offset 连续拉取所有结果。 - F27
MilvusStore.get_chunks_by_ids()（源码 186 行）：把 Chunk ID 列表转换为条件查询。 - F28
MilvusStore.hybrid_retrieve()（源码 207 行）：组合 Dense 与 BM25 召回并做 RRF 融合。 - F29
MilvusStore.hybrid_retrieve._search()（源码 242 行）：提交混合搜索 RPC。 - F30
MilvusStore.dense_retrieve()（源码 270 行）：执行 Dense 降级召回并统一返回形状。 - F31
MilvusStore.dense_retrieve._search()（源码 276 行）：提交 Dense 搜索 RPC。 - F32
MilvusStore.delete()（源码 316 行）：按表达式删除集合记录。 - F33
MilvusStore.has_collection()（源码 321 行）：检查配置中的集合是否存在。 - F34
MilvusStore.drop_collection()（源码 324 行）：提交安全检查后删集合的回调。 - F35
MilvusStore.drop_collection._drop()（源码 325 行）：仅在集合存在时删除。 - F36
get_milvus_store()（源码 335 行）：懒创建并返回共享 Store 包装对象。 - C06
MilvusSettings：不可变的连接与集合配置。 - C07
MilvusStore：集合读写、检索与连接回调的统一入口。
P06 · backend/indexing/parent_chunk_store.py
组合 PostgreSQL 与 Redis 存取父块。
- F37
ParentChunkStore._to_dict()（源码 14 行）：把 ParentChunk ORM 记录转为业务字典。 - F38
ParentChunkStore._cache_key()（源码 29 行）：生成单个父块的业务缓存键。 - F39
ParentChunkStore.upsert_documents()（源码 32 行）：更新或插入父块并刷新缓存。 - F40
ParentChunkStore.get_documents_by_ids()（源码 85 行）：先缓存后数据库，按输入顺序返回父块。 - F41
ParentChunkStore.delete_by_filename()（源码 114 行）：按来源删除父块并清理对应缓存。 - C08
ParentChunkStore：持久化 L1 / L2，供 Auto-Merging 取全文。
调用与对象传递对照
| 出发位置 | 目标 | 关系 |
|---|---|---|
| P01 · 第 2 步 | F01 load_env |
入口先加载环境变量 |
| P02 · 第 1 步 | P02:3 backend/infra/database.py |
engine → SessionLocal.bind |
| P02 · 第 1 步 | F03 before_cursor_execute |
engine 事件注册 |
| P02 · 第 1 步 | F04:2 init_db |
engine → create_all(bind=engine) |
| F03 · 第 3 步 | F02 _clean_nul_chars |
SQL 参数清洗 |
| P02 · 第 4 步 | P04:1 backend/db/models.py |
跨文件导入同一个 Base |
| P04 · 第 1 步 | C02 User |
继承 Base |
| P04 · 第 1 步 | C03 ChatSession |
继承 Base |
| P04 · 第 1 步 | C04 ChatMessage |
继承 Base |
| P04 · 第 1 步 | C05 ParentChunk |
继承 Base |
| P02 · 第 4 步 | F04:2 init_db |
Base.metadata 建表 |
| F04 · 第 1 步 | P04 backend/db/models.py |
延迟导入注册模型 |
| P03 · 第 2 步 | F05 RedisCache.__init__ |
创建共享包装器 |
| F08 · 第 1 步 | F06 RedisCache._get_client |
复用客户端 |
| F09 · 第 1 步 | F06 RedisCache._get_client |
复用客户端 |
| F10 · 第 1 步 | F06 RedisCache._get_client |
复用客户端 |
| F11 · 第 1 步 | F06 RedisCache._get_client |
复用客户端 |
| F08 · 第 1 步 | F07 RedisCache._key |
缓存键 |
| P05 · 第 2 步 | F36 get_milvus_store |
延迟获取共享 Store |
| F36 · 第 1 步 | F15 MilvusStore.__init__ |
首次构造 |
| F15 · 第 1 步 | F12 MilvusSettings.from_env |
缺省配置 |
| F13 · 第 1 步 | F12 MilvusSettings.from_env |
缺省读取配置 |
| F17 · 第 1 步 | F13 milvus_client_session |
短连接上下文 |
| F18 · 第 1 步 | F13 milvus_client_session |
业务会话上下文 |
| F20 · 第 2 步 | F21 MilvusStore.init_collection._init |
传入回调 |
| F21 · 第 1 步 | F19 MilvusStore.ensure_collection |
创建 Schema 与索引 |
| F23 · 第 1 步 | F14 _normalize_filter |
调用辅助函数 |
| F25 · 第 1 步 | F14 _normalize_filter |
调用辅助函数 |
| F23 · 第 2 步 | F24 MilvusStore.query._query |
RPC 回调 |
| F25 · 第 2 步 | F26 MilvusStore.query_all._query_all |
RPC 回调 |
| F28 · 第 3 步 | F29 MilvusStore.hybrid_retrieve._search |
RPC 回调 |
| F30 · 第 1 步 | F31 MilvusStore.dense_retrieve._search |
RPC 回调 |
| F34 · 第 1 步 | F35 MilvusStore.drop_collection._drop |
RPC 回调 |
| F22 · 第 2 步 | F17 MilvusStore._run |
执行传入的 operation |
| F23 · 第 2 步 | F17 MilvusStore._run |
执行传入的 operation |
| F25 · 第 2 步 | F17 MilvusStore._run |
执行传入的 operation |
| F28 · 第 3 步 | F17 MilvusStore._run |
执行传入的 operation |
| F30 · 第 1 步 | F17 MilvusStore._run |
执行传入的 operation |
| F32 · 第 2 步 | F17 MilvusStore._run |
执行传入的 operation |
| F33 · 第 1 步 | F17 MilvusStore._run |
执行传入的 operation |
| F34 · 第 2 步 | F17 MilvusStore._run |
执行传入的 operation |
| F27 · 第 3 步 | F23 MilvusStore.query |
ID 列表转条件查询 |
| F39 · 第 4 步 | F38 ParentChunkStore._cache_key |
父块缓存键 |
| F40 · 第 3 步 | F37 ParentChunkStore._to_dict |
ORM 转字典 |
| F40 · 第 1 步 | F38 ParentChunkStore._cache_key |
父块缓存键 |
| P02 · 第 3 步 | F39:1 ParentChunkStore.upsert_documents |
SessionLocal() 创建数据库会话 |
| P02 · 第 3 步 | F40:2 ParentChunkStore.get_documents_by_ids |
SessionLocal() 创建数据库会话 |
| P02 · 第 3 步 | F41:1 ParentChunkStore.delete_by_filename |
SessionLocal() 创建数据库会话 |
| F40 · 第 1 步 | F08 RedisCache.get_json |
先读 Redis |
| F39 · 第 4 步 | F09 RedisCache.set_json |
刷新父块缓存 |
[[source/_posts/LLM/2_RAG/project/img/详细实现/项目实现|查看项目实现总图]]。概要图仍保留在“实现”文件夹。
Excalidraw Data
Text Elements
backend/：数据存储及共享基础设施 ^99c328ea
[[1_项目设计#1）env.py|P01 env.py：统一加载项目根目录环境变量]] ^81fab1d0
-
获取项目根目录；导入本文件本身不会读取 .env。 ^8eac0db2
-
应用入口或独立脚本先调用 load_env()，再导入依赖环境变量的模 块。 ^b924599e
load_env()：只加载一次 .env ^b504d0f6
-
若 _LOADED 已为 True，说明导入过，直接返回。 ^9c8cf6d5
-
调用 load_dotenv(PROJECT_ROOT / ".env") 读取配置。 ^8d415431
-
将 _LOADED 置 True，后续调用不重复加载。 ^37d324d2
infra/：连接关系数据库与 Redis 缓存 ^dbddd208
[[1_项目设计#1）database.py|P02 database.py：配置 PostgreSQL 引擎]] ^3c4a8cc8
-
engine：链接数据库创建的引擎对象。 〔→ P02:3 / F03 / F04:2〕 ^cc4e2615
-
编译字符清洗正则；在 engine 上注册 before_cursor_execute 监听器。 ^77c6810c
-
SessionLocal：创建session工厂，统一数据库会话，禁用自动 flush / 自动提交，提交后对象不过期。 〔→ F39:1 / F40:2 / F41:1〕 ^216422db
-
Base：生成 ORM 声明基类，用于后续继承创建表。 〔→ P04:1 / F04:2〕 ^f44bb230
before_cursor_execute()：在 SQL 执行前净化SQL参数 ^532cfa73
-
清洗参数 ^38db5b9c
-
返回清洗结果 ^d57e598e
F04 init_db()：数据库初始化并创建尚不存在的表 ^59b0c819
- 通过数据库引擎创建基本表/初始化数据库 ^dbe3cfc0
_clean_nul_chars()：递归清洗字符串 ^9241f587
- 字符串先做 Unicode NFC 标准化。
- 删除零宽字符、BOM 和不可见方向控制字符。
- 删除正则覆盖的非排版控制字符与 PUA 私有区字符。
- 用 UTF-8 ignore 编解码去除残损代理项；异常时逐字符排除代理项。 ^29bc3370
[[1_项目设计#2）cache.py|P03 cache.py：Redis 缓存读取、写入、删除]] ^292b96b2
- 创建共享包装对象；Redis 客户端在首次使用时才建立。 〔→ F05〕 ^24ed0c7d
C01 RedisCache：提供Redis缓存接口 ^ba922cc0
F05 init()：初始化redis配置 ^8aad528a
F06 _get_client()：获取 Redis 客户端 ^3632b9d3
- 有客户端则复用，无则根据redis url新建 ^59a51d22
F07 _key()：给所有key加上前缀 ^6823d81a
- 给所有key加上前缀 ^2206a52c
F08 get_json()：读取数据 ^27d87212
- 通过key读取数据〔→ F06 / F07〕
- 用json.loads() 解码
- 返回python对象 ^d859fdcd
F09 set_json()：写入数据 ^e20ebd74
- 用 json.dumps读取python对象 〔→ F06〕
- 设置过期时间 ^9d658d70
F10 delete()：删除key ^b45ad9da
- 删除key。 〔→ F06〕 ^aae7fcd1
F11 delete_pattern()：按模式批量删除key ^48a636b9
- pattern加前缀，批量删除匹配的key。 〔→ F06〕 ^2ca4997c
db/：定义关系数据库表 ^203e8dc3
[[1_项目设计#1）models.py|P04 models.py：定义四张 ORM 表]] ^91f8f659
-
从 database.py 导入同一个 Base实例 〔→ C02 / C03 / C04 / C05〕 ^ba8677b4
-
定义四个继承 Base 的类 ^8352ff69
C02 User：用户表 ^65676902
- 属性：id, 用户名，密码哈希，角色，生日
- 约束：用户删除时级联删除其会话。 ^118539f7
C04 ChatMessage：消息表 ^59c097c1
C03 ChatSession：会话表 ^e25f33b3
C05 ParentChunk：父块表（L1/L2） ^3a3975c3
indexing/：RAG操作封装 ^166abee4
[[1_项目设计#1）milvus_client.py|P05 milvus_client.py：无状态 Store 配合短生命周期 Milvus 连接]] ^0b8cfd72
定义向量数据库配置 ^ce4d662e
F13 milvus_client_session()：在一次上下文中管理客户端连接 ^6fb2a8c9
- 根据配置创建并返回数据库连接对象 〔→ F12〕 ^a792709e
F14 _normalize_filter()：避免查询为空 ^c8cdae8a
- strip() 清理过滤表达式
- 空值替换为 id >= 0 ^cb1e0c5b
F36 get_milvus_store()：懒创建并返回共享 Milus 实例 ^57978759
- 全局变量：创建 MilvusStore()。 〔→ F15〕 ^6d660703
C06 MilvusSettings：不可变的连接与集合配置 ^f8290c72
F12 from_env()：从环境变量生成 Milvus 配置 ^b57d5087
- 从环境变量生成 Milvus 配置 ^a61565f5
C07 MilvusStore：向量数据库读写、检索与连接回调的统一入口 ^2c48ee80
- 只持有 _settings，@property 暴露 collection_name。
- 一个类对应一个表 ^37027557
F15 init()：初始化配置 ^deeed04d
1.初始化向量数据库配置 ^dfd79e36
F16 collection_name：获取表名 ^115ca20c
- 从配置获取表名 ^454726a8
F17 _run()：统一封装的执行函数 ^b1185bfb
-
获取当前Milvus客户端 〔→ F13〕 ^9e6c2d8f
-
执行操作 ^bea2fe0b
F18 session()：复用同一条连接 ^2cd4f1eb
-
with 进入 milvus_client 〔→ F13〕 ^ff66b310
-
yield client 给调用方复用，退出上下文后关闭连接。 ^c3b70905
F20 init_collection()：数据库初始化，无表则建表 ^0a4e2c23
-
获取向量维度，默认 1024 ^bb1f701e
-
执行初始化 ^44bc3238
F21 _init()：调用建表函数 ^fa08c703
- 调用建表函数 〔→ F19〕 ^237724b1
F19 ensure_collection()：建表、初始化双索引 ^085825dd
F22 insert()：批量插入 ^7c3c4624
- 批量插入回调 〔→ F17〕 ^0d471ee9
F23 query()：单次查询 ^fa57c7c1
-
规范表达式 〔→ F14〕 ^5e4b2460
-
根据表达式和搜索字段回调查询 〔→ F24 / F17〕 ^4a64201d
F24 _query()：在客户端上执行查询 ^bbff5065
- 通过客户端执行查询（有最大查询数量） ^5fc5facc
F25 query_all()：分页查询（复用同一次连接） ^15b4f26a
-
规范表达式 〔→ F14〕 ^e15c918f
-
根据表达式和搜索字段回调查询 〔→ F26 / F17〕 ^0089506a
F26 _query_all()：按 offset 连续拉取所有结果 ^7413664b
- 循环加偏移量分批查询（有单次最大查询数量）
- 适合查询有大量数据，比如某个文档的所有chunk ^692e4f80
F27 get_chunks_by_ids()：根据Chunk ID 查询 ^1f588cfc
- 根据Chunk ID 查询chunk ^e1a673a4
F28 hybrid_retrieve()：混合检索 ^30fa9b55
F30 dense_retrieve()：向量检索 ^d6a38023
-
回调检索函数 〔→ F31 / F17〕 ^f766c79a
-
返回检索结果 ^1f2e7e6f
F31 _search()：执行向量检索 ^83d0a0d1
F32 delete()：删除记录 ^98c36ed1
- 删除回调 〔→ F17〕 ^fd19f087
F33 has_collection()：表是否存在 ^9106c669
- 回调判断表是否存在 〔→ F17〕 ^e109a70c
F34 drop_collection()：删除表 ^b9faaaba
- 回调删除 〔→ F17 ，F35〕 ^57c79f19
F35 _drop()：删除表 ^39890b4a
- 表存在则删除 ^9b73bc39
[[1_项目设计#2）parent_chunk_store.py|P06 parent_chunk_store.py：组合 PostgreSQL 与 Redis 存取父块]] ^cdbe0231
- 导入共享 redis cache、数据库会话、父块表模型。 ^0f635cbe
C08 ParentChunkStore：基于 PostgreSQL + Redis 的父级分块存储 ^069151c1
- 优先redis读取，写入同步更新，提供如下功能： 1. 父块写入/更新 2. 父块查询 3. 批量删除父块 ^1cd5cee1
F39 upsert_documents()：批量往数据库、redis更新或插入父块 ^35fcaeaf
-
创建 SessionLocal() 数据库会话。[P02:3] ^95da2920
-
逐条读取输入的chunk： ^50bf7777
-
组装准备写入字段
-
数据库已有则更新，无则插入 ^9a978061
-
写入redis 〔→ F38 / F09〕 ^4c11a6c7
F40 get_documents_by_ids()：根据 id查询块，先redis后数据库 ^50913a31
-
先查redis 记录命中结果和缺失 ID 〔→ F38 / F08〕 ^7ecf289b
-
若有缺失 ID，创建会话，使用数据库一次查询 ^52038fdb
-
缺失者写入Redis 〔→ F37〕 ^bfde11bf
F41 delete_by_filename()：按文件名批量删除数据库记录、redis ^7227ee48
-
查询该文件的父块及 ID 列表 ^24ee2fc3
-
批量删除并提交 PG，然后逐 ID 删除 Redis ^e9d2b693
F37 _to_dict()：把数据库记录转为python字典 ^a67c2c75
- 把数据库记录转为python字典 ^057b06e8
F38 _cache_key()：生成块的redis key ^d42fc084
-
把 chunk_id 拼为 parent_chunk:chunk_id
-
此时未加前缀 ^53bd3dec
-
redis url
-
所有key的前缀
-
缓存过期时间（TTL）
-
redis 客户端 ^Pxdw8wql
-
属性：id，用户id，会话id，元数据json，更新时间，创建时间
-
约束： （用户id，会话id）联合唯一， 外键：用户id 删除会话级联删除消息。 ^tTlISYoL
-
属性：id，会话id，类型，内容，时间戳，rag轨迹
-
约束： 外键：会话id ^Z1K9zWi9
-
属性：id，内容，来源文件名，文件类型，文件路径， 页码，父id，根id，块级别，块内序号，更新时间 ^R2wDIPlL
-
向量数据库封装： ^gyK6zhAR
连接向量数据库 ^zmzf3q93
条件过滤处理 ^l1NF1eUb
向量数据库操作 ^LL7ST8QW
-
定义属性： 主机，端口，表名，url，超时时间 ^XnB6OjQo
-
定义初始化配置函数 ^1geiZc32
-
定义表格字段： id，密集向量，稀疏向量，L3文本，文件名， 文件类型，文件路径，页码，chunk序号， chunkid，父chunkid，根chunkid，块级别
-
定义稀疏向量计算方式：BM25
-
建立密集向量索引： HNSW，内积相似度， 节点最大连接数，候选队列大小
-
建立稀疏向量索引
-
创建表 ^rtGuo3GP
F23 query()：单次查询 ^hgMZzly4
-
规范表达式 〔→ F14〕 ^FWTfSyVn
-
根据表达式和搜索字段回调查询 〔→ F24 / F17〕 ^AXBUGrCZ
F24 _query()：在客户端上执行查询 ^vKliZr2G
-
通过客户端执行查询（有最大查询数量） ^Oh1rchRK
-
执行检索，检索参数： 距离度量，候选队列长度，top_k，过滤条件 ^lnhRWhba
-
回调检索函数 〔→ F29 / F17〕 ^hhY1n0Wz
-
返回检索结果 ^Ji5V6H8q
F31 _search()：执行混合检索 ^sdEooej8
-
执行检索，检索参数： 向量检索请求， 稀疏检索请求，top_k，重排序 ^selEYMsV
-
定义向量检索请求（有参数）
-
定义稀疏检索请求（有参数）
-
定义重排方式 ^5q54DlpB
Element Links
81fab1d0: [[1_项目设计#1）env.py]]
3c4a8cc8: [[1_项目设计#1）database.py]]
292b96b2: [[1_项目设计#2）cache.py]]
91f8f659: [[1_项目设计#1）models.py]]
0b8cfd72: [[1_项目设计#1）milvus_client.py]]
cdbe0231: [[1_项目设计#2）parent_chunk_store.py]]
%%
Drawing
1 | N4KAkARALgngDgUwgLgAQQQDwMYEMA2AlgCYBOuA7hADTgQBuCpAzoQPYB2KqATLZMzYBXUtiRoIACyhQ4zZAHoFAc0JRJQgEYA6bGwC2CgF7N6hbEcK4OCtptbErHALRY8RMpWdx8Q1TdIEfARcZgRmBShcZQUebR4ATm0ARho6IIR9BA4oZm4AbXAwUDBSiBJuCE1sZIA2AFYagDM00shYRErA7CiOZWDWssxuZ2SABgAObQmJhJ5a5Jnk+vGA |
%%
