在标题切割之后做 refine_chunks 精细切割，并梳理 item_name 识别节点的动机、bge-m3 与 Milvus 的准备工作以及节点的实现步骤。

本篇要点

  • refine_chunks 精细切割：长块短切（带 overlap）、短块合并、parent_title / part 补全
  • 为什么必须识别 item_name：用主体名缩小向量检索范围
  • 向量回顾：稠密 dense（注意维度，推荐 1024）与稀疏 sparse
  • 准备工作：bge-m3 的下载与两种加载方案、Milvus 部署与 MilvusGateway 封装
  • item_name 识别节点的作用、参数与六步实现流程

内容回顾

1、Day04：node_document_split 节点

  • 先定下切块产物结构：content（标题 + 内容）、title、file_title、parent_title、part，向量相关字段留到后面补
  • 四大切割原则：语义优先、长度可控、边界清晰且有合理重叠、可追溯
  • 项目切割参数：CHUNK_MAX_SIZE = 1000、CHUNK_SIZE = 600、CHUNK_OVERLAP = 50、CHUNK_MIN = 400
  • 节点实现走到两步：load_markdown_content 负责参数获取、非空校验与 \r\n 数据清洗，split_document 用正则按 # 标题做语义切割，产出 list[dict{title, content, file_title}]

今天内容

1、refine_chunks 精细切割

1.1 为什么还要精细切割

上一节的 split_document 只解决了一件事：按标题切，保证语义不跨域。但它是”跟着标题走”的，切出来的块长度完全不可控——有的标题下面挂了几百行，块远远超过 CHUNK_SIZE；有的标题只有一两行，切出来就是短碎片。

所以要在标题切割之后再补一步精细切割，目标是三件事：

  1. 长 → 短：过长的块按 CHUNK_SIZE 再切一刀，并保留合理的 overlap，切割过程中不能破坏语义。
  2. 短 → 合并：同一标题下过短的碎片合并起来，让边界清晰、长度可用。
  3. 属性补全：把 parent_title、part 补齐，保证切块可追溯。

调用方式很简单，入参就是上一步的 chunks：

1
refine_chunks = refine_chunks(chunks)

篇幅上沿用 Day04 定下的四个常量：CHUNK_SIZE = 600、CHUNK_MAX_SIZE = 1000、CHUNK_MIN = 400、CHUNK_OVERLAP = 50。

1.2 入参与出参

入参 chunks：[{title, content, file_title}]。其中 content 可能过长也可能过短，且此时还没有 parent_title、part。

出参 refine_chunks：精细切割后的新列表，字段含义如下。

字段 含义
title 切片标题，被长切后会变成 title_子片段_1、title_子片段_2 这样的形式
content 精细切割后的正文，切割时不能破坏语义
file_title 文件名，原样携带
parent_title 原来的 title，用于标识这些子片段属于同一个父标题
part 二次切割产生的序号 1、2、3…，由”过长向短切割”生成

也就是说，输出的语义是：title = (title_子片段_1 or 2)，content = 精细切割，parent_title = 原来的 title，part = 1 / 2 / ...。

1.3 整体步骤

  1. 定义一个存储最终结果的 list：refine_list = [dict]。
  2. 循环处理 chunks，拿到每个 chunk 的 content，先判断长度。
  3. content > 600 的块触发长切，交给 _split_long_chunk(chunk)，返回 list[dict]（带上 parent_title、part），用 extend 合并进结果集。
  4. content <= 600 的块直接原样 append 进结果集。
  5. 把同一个标题下过短的块合并（< 400，合并后不超过 1000），抽成方法 _merge_short_chunks_same_parent_title(refine_list)。
  6. 补全属性：没有 parent_title 的用 title 兜底，没有 part 的补 1。
  7. return refine_list。

伪代码结构如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 1. 最终结果容器
refine_list = []

# 2. 循环处理 chunks，判断长短
for chunk in chunks:
    if len(chunk["content"]) > CHUNK_SIZE:          # content > 600，触发长切
        # 3. 过长向短切割，返回 list[dict]，携带 parent_title / part
        sub_chunks = _split_long_chunk(chunk)
        refine_list.extend(sub_chunks)
    else:
        refine_list.append(chunk)

# 4. 同一个标题下过短的进行合并（< 400，且合并后 > 1000 也要收住）
refine_list = _merge_short_chunks_same_parent_title(refine_list)

# 5. 属性完善和补充
for chunk in refine_list:
    if "parent_title" not in chunk:
        ...
    if "part" not in chunk:
        ...

return refine_list

1.4 长块短切：_split_long_chunk

  • 判定条件：content > CHUNK_SIZE（即 600），说明这个块太长了，需要”过长向短切割”。
  • 切割方式：用递归切割器，按 CHUNK_SIZE = 600 为基准长度切，同时保留 CHUNK_OVERLAP = 50 的重叠——重叠比例大约落在 8% ~ 12%，保证相邻块之间上下文连贯、语义不被切断。
  • 切割产物：每个子片段都要写回 parent_title（原来的 title）和 part（1、2、3…），title 变成 title_子片段_x。

注意： 切长块时不能”死按字符数切”，必须在同一个语义边界内切；边界清晰、不跨语义，这是切割原则里排在第一位的。

1.5 短块合并：_merge_short_chunks_same_parent_title

  • 判定条件：同一个 parent_title 下、长度小于 CHUNK_MIN = 400 的块，判定为短碎片，需要尝试合并。
  • 合并上限：合并之后不能超过 CHUNK_MAX_SIZE = 1000，避免为了消除碎片反而造出过大块、导致向量失真。
  • 为什么要合并：碎块比长块更危险——检索召回时碎块语义不完整，会拉低召回率；把碎块并回 600 左右，才能兼顾”召回率”和”准确率”。

1.6 属性补全与结果备份

长切过的块已经有了 parent_title 和 part；没被长切、也没参与合并的块还是”裸的”，所以要统一补一遍：

  • 没有 parent_title → 用原来的 title 兜底，保证每个块都能追溯到父标题。
  • 没有 part → 补 1。

这样一个 chunks 里的块，无论是长切出来的还是原样保留的，结构都统一了，后续节点不用再判断字段是否存在。

今天的收尾里还提到要做数据备份：精细切割的结果会落成本地文件（后续节点读取的 文件名_new.json），既方便排查切割效果，也让下游节点可以直接从文件恢复 chunks。

1.7 两个待优化点

今日内容里明确记下了 refine_chunks 的两处遗留问题，属于”后面要回来看”的坑：

优化 1：part 断档的问题。 长切 + 短合并组合之后，part 序号会出现 1、3 这种断档，编号不再连续。

优化 2：短块合并的策略问题。 例如四段 300 300 300 300 应该合并成 900 这种更大的块，而 300 应该优先和相邻块凑成 600、600，合并粒度需要再调。

2、item_name 识别节点

2.1 为什么要识别 item_name

核心目的只有一句话：通过 item_name 缩小文档检索范围。

检索链路是：item_name -> 文档 -> 向量相似度搜索，先按主体名把候选范围框住，再做向量相似度搜索，结果才更加精准。

这里有两个必须记住的关系：

  • 一个文件 → 一个 item_name。
  • 一个 item_name → 对应多个文件。

正因为一个 item_name 下会挂多个文件，所以：

上传向量数据的时候，要清空旧数据，并且要用 file_title 做标识。 否则同一个文件重新导入时会留下重复的旧向量。

2.2 向量基础回顾：稠密与稀疏

  • 向量：把文字转换成计算机可识别的高维数组。
  • 通过向量可以计算相似度，这也是向量检索的基础。
  • 向量分两类：
    • 稠密向量 dense：表达语义，注意维度，推荐 1024。
    • 稀疏向量 sparse：本质是 dict（词 → 权重）。

2.3 准备工作一：下载 bge-m3

  • 下载地址（魔搭直接下载）：https://www.modelscope.cn/models/BAAI/bge-m3/files
  • 也可以使用项目里的工具脚本下载：shared / tool / download_bgem3.py（一定要明确下载位置，后面 .env 要按这个路径配）。

2.4 准备工作二：加载 bge-m3 的两种方案

加载的思路是：用客户端（py 文件和函数）加载权重文件，在代码里使用并生成向量。项目里对比了两种方案。

方案 1：官方提供的客户端（FlagEmbedding）

1
pip install -U FlagEmbedding
1
2
3
4
from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel('BAAI/bge-m3',
                       use_fp16=True)

方案 2：Milvus 提供的 bge-m3 加载工具

文档地址：https://milvus.io/docs/zh/embed-with-bgm-m3.md（2024 年两边互相道谢的那波联动）。

BGE-M3 因其在多语言、多功能和多粒度方面的能力而得名。BGE-M3 能够支持 100 多种语言，为多语言和跨语言检索任务树立了新的标杆。它在单一框架内执行密集检索、多向量检索和稀疏检索的独特能力，使其成为广泛的信息检索（IR）应用的理想选择。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from pymilvus.model.hybrid import BGEM3EmbeddingFunction

bge_m3_ef = BGEM3EmbeddingFunction(
    model_name='BAAI/bge-m3',  # Specify the model name
    device='cpu',              # Specify the device to use, e.g., 'cpu' or 'cuda:0'
    use_fp16=False            # Specify whether to use fp16. Set to `False` if `device` is `cpu`.
)

docs = [
    "Artificial intelligence was founded as an academic discipline in 1956.",
    "Alan Turing was the first person to conduct substantial research in AI.",
    "Born in Maida Vale, London, Turing was raised in southern England.",
]

docs_embeddings = bge_m3_ef.encode_documents(docs)

项目最终走的是 Milvus 的这套 BGEM3EmbeddingFunction，因为后面入库、检索都在 Milvus 里，向量生成标准和向量库保持一致最省事。

2.5 封装到 shared / embedding_utils 与 infra

（1）位置：shared / embedding_utils

（2）配置：

1
2
3
4
5
6
7
8
BGE_M3_PATH=D:\ai_models\modelscope_cache\models\BAAI\bge-m3
BGE_M3=BAAI/bge-m3
# gpu
# BGE_DEVICE=cuda:0
# cpu
BGE_DEVICE=cpu
# cpu 需要关闭 0  GPU 上生成向量通常默认开启了半精度浮点数（FP16）加速，而 CPU 上通常使用单精度（FP32）
BGE_FP16=0

注意： CPU 上要写 BGE_FP16=0，因为 FP16 半精度加速是 GPU 的默认玩法，CPU 上用的是 FP32 单精度。

如果要使用 GPU 怎么办？

1
2
BGE_DEVICE=cuda:0
BGE_FP16=1

注意： 改成 GPU 之后，相关的依赖也要换成 GPU 版本（详见教案）。

（3）对外方法：

1
2
3
4
def get_bge_m3_ef()                       # 获取嵌入式模型（不支持直接使用）
def generate_embeddings(texts: list[str]) -> dict[str, Any]
    # result['dense'][0 1 2 3 ...]
    # result['sparse'][0 1 2 3 ...]

（4）再封装到 infra / llm / providers：

1
2
3
4
5
6
7
# 获取嵌入式模型
def bge_m3_embedding(self):
    return get_bge_m3_ef()

# 生成向量
def generate_embeddings(self, texts: list[str]):
    return generate_embeddings(texts)

2.6 准备工作三：部署 Milvus

部署用的是官方 standalone 的 docker compose 文件：

1
2
3
4
5
6
wget https://github.com/milvus-io/milvus/releases/download/v2.4.11/milvus-standalone-docker-compose.yml -O docker-compose.yml

vim docker-compose.yml   # minio 9000 -> 03 02:9000（把内置 minio 的 9000 端口让开，避免和本地已有服务冲突）

docker compose up -d     # 启动
docker compose down      # 停止
  • 可视化工具：attu。
  • 连接地址（笔记里记的是）：39.105.7.90:19530。

2.7 Milvus 客户端与 MilvusGateway 封装

（1）Milvus 加载工具

  • 工具位置：shared / clients / milvus_utils.py
  • 对外函数：get_milvus_client() -> milvus_client
  • 配置：.env（url 交给 get_milvus_client() 去读，另一个集合的名字写在代码里）

（2）统一封装到 infra

封装位置：/infra/vectorstore/milvus_gateway.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from app.infra.config.providers import infra_config
from app.shared.clients import get_milvus_client


class MilvusGateway:
    # 提供两个属性：集合名
    @property
    def chunk_collection_name(self):
        return infra_config.milvus_config.chunks_collection

    # 提供一个获取客户端
    @property
    def item_name_collection_name(self):
        return infra_config.milvus_config.item_name_collection

    @property
    def milvus_client(self):
        return get_milvus_client()


milvus_gateway = MilvusGateway()

这样节点里只要用 milvus_gateway.chunk_collection_name / milvus_gateway.item_name_collection_name / milvus_gateway.milvus_client，不用再关心配置从哪来。

整条准备链路串起来就是：item_name -> 向量 -> 嵌入式模型 bge3 -> 加载方式（官方 / milvus） -> milvus 向量生成、归一化 -> 准备嵌入式模型 -> milvus 准备。

2.8 节点实现分析

作用： 识别 item_name，即 chunks -> item_name，以及 item_name -> milvus。

参数：

  • 使用：chunks / file_title / md_path（文件名_new.json）
  • 生成：item_name，以及带 item_name 字段的 chunks

步骤：

  1. 获取并校验参数（state）-> chunks / file_title
    1.1 获取参数 file_title、chunks、md_path；
    1.2 非空校验：file_title 为空走 md_path 兜底，chunks 为空则从 md_path / json 读取；
    1.3 返回结果。
  2. 根据 chunks 内容识别 item_name（chunks, file_title 兜底）-> item_name
    2.1 获取语言模型对象 chat()；
    2.2 加载提示词字符串（.prompt）；
    2.3 把提示词封装成 Message；
    2.4 封装 LangChain 调用链（chains）；
    2.5 执行调用链，拿到 item_name；
    2.6 非空校验：识别结果为空就用 file_title 兜底；
    2.7 返回 item_name。
  3. 前置准备工作（item_name, chunks）：把 item_name 补进 chunks。
  4. 如果没有 collection，就创建 item_name 对应的集合（prepared_milvus_item_name_collection(None) -> None）：
    4.1 判断有没有对应的 item_name 集合；
    4.2 有的话打印日志，直接 return 跳出方法；
    4.3 没有则创建集合。
  5. 删除或者保存数据（item_name, file_title）（delete_and_insert_item_name(item_name, file_title)）：
    5.1 给 item_name 生成稠密和稀疏向量，补全插入数据需要的属性（含 PK）；
    5.2 插入之前，先删除该文件关联的旧数据（file_title == file_title）；
    5.3 完成本次数据插入。
  6. 更新 state：chunks | item_name。

今日内容小结： refine_chunks 精细切割（递归切割器、短合并问题）、数据备份、嵌入式模型准备与 Milvus 准备工作、参数校验和 item_name 识别。