RAG 父子块关联设计：同一张表 + 两步入库，让子块准确找到父块

摘要

在 RAG 系统中，文档切分直接决定检索质量。父子块方案用短子块保证检索命中率、长父块保证回答完整性，两者之间必须建立可靠的关联。本文不讲整个上传流程，只聚焦一个核心问题：父子块在数据库里如何设计和关联，以及设计过程中做了哪些考量。

一、核心问题：父子块如何关联

父子块方案中，一个父块会被切成多个子块。入库时面临两个问题：

关联发生在入库那一刻。切分阶段生成的是临时草稿列表，子块用临时索引 parent_index 标记父块；入库阶段父块先写入获得真实数据库 ID，子块再通过 parent_index 映射表找到这个 ID 并写入自己的 parent_id 字段。下文逐个拆解。

二、存哪里：同一张表还是分两张表？

建表时有两个选择：

• 方案A：父块一张表，子块一张表，通过外键关联

• 方案B：父子块同一张表，通过 is_parent 区分类型，通过 parent_id 自关联

本项目采用 方案B，原因有三：

第一，查询更简单。 如果分两张表，要查某个文档的所有分块需要 UNION 两张表；而同一张表只需要一个 SELECT 条件 document_id = ?。

第二，子块找父块更直接。 子块拿到自己的 parent_id，直接在同表里查 id = parent_id 就能拿到父块。如果分两张表，需要跨表 JOIN。

第三，扩展性更好。 未来如果引入更多层级（比如爷爷块、孙块），同一张表加 parent_id 就能无限延伸，分两张表则无法支持。

分块表结构

sql

CREATE TABLE document_chunks (
id INT PRIMARY KEY AUTO_INCREMENT,
document_id INT NOT NULL, — 所属文档
parent_id INT DEFAULT NULL, — 父块ID（子块用）
content TEXT NOT NULL, — 分块文本
is_parent BOOLEAN DEFAULT FALSE, — True=父块, False=子块
FOREIGN KEY (parent_id) REFERENCES document_chunks(id)
);

两个字段决定一切：

父块没有父，所以 parent_id = NULL。子块必须有父，所以 parent_id 指向某个父块的 id。

三、关联的难点：临时索引如何转为数据库ID

切分阶段生成的草稿列表是临时的，子块只知道自己在列表里属于哪个父块，用 parent_index 记录父块在列表中的位置：

text

列表索引: 0 1 2 3 4 5
内容: [父块A] [子块A1] [子块A2] [父块B] [子块B1] [子块B2]
↑
parent_index=0
↑
parent_index=3

入库时，父块A 写入数据库获得真实 ID（比如 101），子块A1 和 A2 必须把 parent_id 填成 101。

难点在于：子块入库时，父块A 刚刚写入数据库，程序需要一种方式把“列表位置0”映射成“数据库ID 101”。

四、解决方案：两步入库 + 映射表

4.1 第一步：父块先入库，建立映射

先遍历一次草稿列表，只处理 is_parent=True 的记录。

每插入一个父块，数据库生成自增 ID。建立一个 Python 字典，把“列表索引”映射到“数据库ID”。

python

parent_map = {}

for idx, draft in enumerate(drafts):
if draft.is_parent:
chunk = DocumentChunk(
document_id=doc_id,
content=draft.content,
is_parent=True,
parent_id=None
)
session.add(chunk)
session.flush() # 立即获得数据库生成的ID
parent_map[idx] = chunk.id # 列表位置0 → 数据库ID 101

执行完后，parent_map = {0: 101, 3: 102}，表示列表位置0的父块对应数据库ID 101，位置3对应102。

4.2 第二步：子块入库，查映射填parent_id

再遍历一次，只处理 is_parent=False 的记录。

子块记录的 parent_index 正好是它在列表里看到父块的位置，用这个去 parent_map 里查，拿到真正的数据库ID填入 parent_id。

python

for idx, draft in enumerate(drafts):
if not draft.is_parent:
chunk = DocumentChunk(
document_id=doc_id,
content=draft.content,
is_parent=False,
parent_id=parent_map.get(draft.parent_index) # 0 → 101
)
session.add(chunk)

session.commit()

五、为什么必须分两步

读者可能会问：为什么不一次性插完？

如果一次性把所有分块（父子混合）插入数据库，子块插入时父块可能还没生成，parent_id 拿不到真实ID。数据库外键约束会直接报错。

那能不能先生成父块的数据库ID再一次性插入？ 也不行。SQLAlchemy 的 flush() 会立即生成ID，但此时父块还未提交，子块如果同时插入，需要在同一个事务中拿到父块的ID——这又回到了“父块必须先插入”的问题。

分两步就是为了让父块先拿到数据库ID，子块才能正确引用。

六、完整代码实现（含注释）

6.1 分块草稿结构（chunking.py）

python

@dataclass
class ChunkDraft:
content: str # 分块文本
section: str | None # 所属章节
is_parent: bool # True=父块, False=子块
parent_index: int | None = None # 子块记录父块在列表中的位置

6.2 父子分块主函数（chunking.py）

先按标题分段，再依次切成父块和子块。切分过程中记录每个子块的 parent_index。

python

def split_policy_text(text: str) -> list[ChunkDraft]:
sections = split_by_headings(text) # 按章节标题分段
drafts = []

for section_title, section_body in sections:
# 章节内容切成父块（1200字符）
parent_chunks = window_split(section_body, 1200, 100)

for parent_chunk in parent_chunks:
parent_index = len(drafts) # 父块在列表中的位置

# 添加父块
drafts.append(ChunkDraft(
content=parent_chunk,
section=section_title,
is_parent=True
))

# 父块切成子块（420字符，重叠80）
child_chunks = window_split(parent_chunk, 420, 80)

for child_chunk in child_chunks:
drafts.append(ChunkDraft(
content=child_chunk,
section=section_title,
is_parent=False,
parent_index=parent_index # 关键：记录父块位置
))

return drafts

6.3 入库服务（保存关联）

这个函数实现前文说的两步入库逻辑，代码很短但很关键。

python

def save_chunks_to_db(session, document_id: int, drafts: list[ChunkDraft]) -> int:
# 第一步：父块入库，建立索引→数据库ID映射
parent_map = {}
for idx, draft in enumerate(drafts):
if draft.is_parent:
chunk = DocumentChunk(
document_id=document_id,
content=draft.content,
is_parent=True,
parent_id=None
)
session.add(chunk)
session.flush()
parent_map[idx] = chunk.id

# 第二步：子块入库，通过映射表填充parent_id
for idx, draft in enumerate(drafts):
if not draft.is_parent:
chunk = DocumentChunk(
document_id=document_id,
content=draft.content,
is_parent=False,
parent_id=parent_map.get(draft.parent_index)
)
session.add(chunk)

session.commit()
return sum(1 for d in drafts if not d.is_parent)

七、查询示例：从子块找回父块

关联建好后，查询某个子块对应的父块非常简单：

python

# 假设子块ID是 11
sub_chunk = session.query(DocumentChunk).filter_by(id=11).first()
parent_chunk = session.query(DocumentChunk).filter_by(id=sub_chunk.parent_id).first()

# parent_chunk.content 就是完整的父块全文

同一张表、一个自关联字段，查询成本极低。

八、总结

父子块关联设计的核心可以概括为：

整个设计的核心思想是：在临时列表里用索引建立关联，入库时通过两步走把索引转换成真实数据库ID。 既保证了外键约束不报错，又让每一条子块都能准确找到自己的父块。