记录 node_bge_embedding 的向量生成与批量策略、node_import_milvus 的集合创建与先删后插，以及 FastAPI 路径、参数、响应与异步任务回顾。

本篇要点

  • node_bge_embedding：把 item_name + content 拼起来生成向量，批量提交
  • 批量数量的约束来源：bge-m3 的 8192 token 上下文窗口、CPU 缓存与 GPU 显存，最终 6 个一批
  • node_import_milvus：创建 schema / index、先删后插、导入完成日志
  • FastAPI 回顾：路径、参数、响应（文件 / json / 流式）与异步 BackgroundTasks

内容回顾

1、Day05：精细切割与 item_name 识别节点

  • refine_chunks 精细切割：过长的块（> 600）交给 _split_long_chunk 短切并保留 overlap，过短的块（< 400）用 _merge_short_chunks_same_parent_title 合并，合并上限 1000，最后补全 parent_title / part
  • 识别 item_name 的动机：用主体名缩小检索范围；一个文件对应一个 item_name，一个 item_name 对应多个文件，所以入库前必须清空旧数据、用 file_title 做标识
  • 准备工作：bge-m3 从魔搭下载（shared/tool/download_bgem3.py），加载有官方的 FlagEmbedding 与 Milvus 的 BGEM3EmbeddingFunction 两套方案
  • 基础设施：Milvus standalone 部署、shared/clients/milvus_utils.py 的 get_milvus_client()，以及 /infra/vectorstore/milvus_gateway.py 里的 MilvusGateway

今天内容

1、node_bge_embedding 节点实现

1.1 作用与参数

作用： 给 chunks 生成向量，是插入向量库之前的最后一道准备工作。

参数：

  • 使用：chunks / md_path / item_name
  • 响应：embeddings_content（向量）

两个细节（重点）：

  1. 生成向量时用的是 item_name + content，不是单纯拿 content 去生成。
  2. 生成过程要 批量生成，不能一条一条来。

1.2 为什么是 item_name + content

单独把 content 丢给嵌入式模型，向量里只有这一块的语义；而同一份文档里，很多块的正文长得非常像（例如一堆同结构的说明段落），只凭 content 很难区分它们属于哪个主体。

把 item_name 拼在 content 前面，等于给每个切块打上主体标签：向量里同时包含了”这是哪份资料的哪一段”，检索时更容易命中正确的 item_name 范围。入模拼接之后，一段文本的长度大致会变成这一段的样子：

1
1 chunk = item_name + content  =>  600 / 1000 / 400 / 600 / 1000 ...

也就是说，原本按 600、1000、400 控制好的切块，拼接上 item_name 之后还会更长一些，这直接影响后面的批量大小。

1.3 步骤

  1. 获取并校验参数（state）-> chunks / item_name。
  2. 批量进行向量生成：(chunks, item_name, batch_number: int = 6) -> chunks。
    • 遍历 chunks，每个 chunk 的 content = item_name + content；
    • 调用生成向量函数 generate_embeddings(texts: list[str])，嵌入式模型本身支持批量生成，所以把拼好的多段文本一次性传进去；
    • 需要考虑两个约束（见 1.4）。
  3. 更新 state：embeddings_content。
1
2
3
4
5
6
7
8
9
10
11
12
13
# 伪代码
def node_bge_embedding(state, batch_number: int = 6):
    chunks, item_name = load_and_check(state)     # 参数获取 + 校验

    for chunk in chunks:
        chunk["content"] = item_name + chunk["content"]   # item_name 拼进正文

    # 嵌入式模型支持批量生成，按 batch_number 分批
    texts = [chunk["content"] for chunk in chunks]
    embeddings = generate_embeddings(texts)       # result['dense'] / result['sparse']

    state["embeddings_content"] = embeddings
    return state

1.4 批量大小为什么是 6 个一批

批量生成必须同时看两个约束，任何一头超了都会出问题。

约束 1：嵌入式模型的上下文窗口 —— bge-m3 是 8192 token

原文把它比作”车的空间”：一次塞进去的文本不能超过模型窗口。

  • token 和字符的换算关系：中文大致是 字符 ≈ token / 1.3，英文大致是 字符 ≈ token / 0.8。
  • 按这个关系，叠加上 item_name 之后的切块，6 个一批刚好把窗口用满又不溢出。

约束 2：CPU 缓存空间和 GPU 显存 —— 超不过的

  • 批量越大，单次生成占用的内存 / 显存越多；
  • 真实占用还要乘上并发：单次占用 * 并发，这才是峰值；
  • 所以批量不能只看 token 够不够，还要看机器扛不扛得住。

结论： 上下文窗口算出来是 6 个一批，显存 / 内存的实际限制也是 6 个一批，两个约束正好一致，所以 batch_number = 6。

1.5 与本节点相关的前置约定

  • item_name 由上一个节点（item_name 识别节点）产出，本节点直接复用，不再重新识别。
  • 本节点只负责”生成向量”，不负责写库；真正的落库在下一个节点。
  • embeddings_content 里同时带着稠密向量和稀疏向量（result['dense'] / result['sparse']），入库时两个字段都要写。

2、node_import_milvus 节点实现

2.1 作用与参数

作用： 把 embeddings_content 存储到向量数据库。

参数：

  • 使用：embeddings_content
  • 响应：state（最终由图的 .invoke 统一跑完整条链路）

2.2 步骤一：获取并校验

  • 获取并校验 embeddings_content，没有数据就直接报错；
  • 同时需要 file_title，因为后面的删除和插入都要用它做标识。

2.3 步骤二：准备和创建 Milvus 集合（chunks）

流程是：获取 Milvus 客户端 → 检查集合是否存在 → 存在就打印日志直接返回 → 不存在就打印日志并创建。

1
2
3
4
5
6
7
8
9
1. 获取 milvus 客户端
2. 检查是否存在：has_collection()
   - 存在：日志 + return
   - 不存在：日志 + 创建
3. 创建 schema
   schema 指定列信息：
   chunk_id, title, file_title, parent_title, item_name, part, content, dense_vector, sparse_vector
4. 创建 index：dense_vector, sparse_vector
5. 创建集合

注意： schema 里的列就是我们一路攒下来的字段——切割阶段产出的 chunk_id、title、file_title、parent_title、part、content，识别阶段补的 item_name，以及本阶段生成的 dense_vector、sparse_vector。稠密向量和稀疏向量都要建索引，否则混合检索会退化。

2.4 步骤三：删除和插入数据（先删后插）

顺序非常关键：

1
2
3
4
5
# 先删除
delete 集合 file_title == 'file_title'

# 再插入
insert(data = embeddings_content)

为什么要先删？ 因为一个 item_name 会对应多个文件，同一个文件重新导入时如果直接插入，旧向量会残留，检索结果里就会出现重复甚至过期的内容。用 file_title == 'file_title' 精准删掉这个文件上一次导入的数据，再插入新数据，就能保证库里一个文件只有一份最新向量。

2.5 步骤四：日志与返回

  • 打印日志 info：项目导入模块全部完成 + 时间戳，方便在前端轮询状态时看到进度。
  • 最后 return state，把更新后的 state 交回给图继续流转。

3、FastAPI 使用回顾

3.1 FastAPI 的作用

FastAPI 对外暴露我们的功能：提供一个 http 协议的网络地址，供客户端访问。

1
客户端 -> http -> fastapi 函数 -> 真正的业务逻辑

一个接口函数基本固定四步：

  1. 设置地址：@fastapi对象.get / post / put / delete(/地址) 修饰函数。
  2. 接收参数（params 的 ?key=value、paths 的 /xx/1、json、文件二进制）。
  3. 调用业务逻辑，调用方法。
  4. 响应前端数据（文件 / json / streaming）。

3.2 如何设置路径

1
2
3
@fastapi对象.get("/地址")      # 也支持 post / put / delete
def 函数():
    ...

3.3 如何接收参数（四种）

（1）路径参数

  • 设置动态路径：/{key}/{key}
  • 形参列表：与 key 同名的参数，写成 key: 类型 = None，是否给默认值决定了是否必须传递。

（2）param / query 参数

  • 形式：?key=value&key=value
  • 特点：不影响路径
  • 形参列表：请求参数同名变量 + : 类型 = None

（3）json 参数

  • 定义接收的类型（JSON 就是对象）：
1
2
class 类(BaseModel):
    key: 类型 = None   # 必须传递的就不给默认值
  • 形参列表里直接写 变量: 类。

（4）文件参数

1
2
3
4
5
def 接口(upload: FileUpload = File(...)):
    upload.file          # 获取上传文件的引用
    upload.file.filename
    upload.file.content_type
    upload.file.read()

3.4 如何响应数据

（1）响应文件数据

1
2
3
4
5
6
7
8
9
10
@erdaye.get("/html/import")
def import_html():
    import_html_path_obj: Path = PROJECT_ROOT / "app" / "resources" / "html" / "import.html"

    return FileResponse(
        filename=import_html_path_obj.name,
        # content_disposition_type="inline",
        path=str(import_html_path_obj),
        media_type=guess_type(import_html_path_obj.name)[0],
    )

（2）响应 json 数据

方案 1：直接返回 dict。

1
2
3
4
5
return {
    "code": 200,
    "message": "Files uploaded successfully, total: 2",
    "task_ids": ["uuid-1", "uuid-2"],
}

方案 2：接收用 BaseModel，返回也用 BaseModel。

1
2
3
4
class xxxXx(BaseModel):
    code: int = 200
    message: str | None = None
    task_ids: list[str]

（3）响应流式数据

1
2
3
4
5
6
7
8
9
10
11
async def generate_stream():
    # 模拟流式输出（逐字返回）
    words = ["你", "好", "，", "这", "是", "流", "式", "响", "应"]
    for word in words:
        await asyncio.sleep(0.5)
        yield word.encode("utf-8")  # 流式输出需返回字节流


@erdaye.get("/stream")
async def stream_response():
    return StreamingResponse(generate_stream(), media_type="text/event-stream")

注意： 流式输出的函数必须是 async，并且 yield 出来的应该是字节流。

3.5 异步方法调用

为什么需要异步？ 一次请求里，前端不想死等，也不想一直看转圈圈，所以只能把耗时的业务动作开成异步。

1
2
3
前端 -> fastapi 1. 接收
                 2. 开启异步执行业务
                 3. 返回结果给前端

前端后续也能拿到结果，有三种方式：

  1. 笨办法：前端轮询。 每 1s 调一次接口，用 task_id 查当前状态——导入模块走的就是这条路。
  2. 后端主动推送：WebSocket（双通道）。 后端有变化就推给前端，前端也能响应回去——市场罗盘用的是它。
  3. 后端主动推送：SSE（单通道）。 只能后端 → 前端推送——查询模块用的是它。

接口里如何执行异步任务，两步：

  1. 定义一个任务函数（真正要被执行的业务逻辑）。
  2. 在接口中使用 backgroundtasks 添加这个异步函数。
1
2
3
4
5
6
7
@erdaye.post("/import")
async def import_file(background_tasks: BackgroundTasks, ...):
    # 1. 接收、校验、生成 task_id
    # 2. 把耗时任务挂到后台执行
    background_tasks.add_task(任务函数, 参数...)
    # 3. 立刻返回 task_id 给前端
    return {"code": 200, "message": "...", "task_ids": [...]}

3.6 今日与明日安排

  1. 第一遍导入模块：今天全力做通。
  2. FastAPI 练习：设置路径、接收参数、响应数据、异步任务四个点都过一遍；遗留一个 bug —— html 下载打不开。
  3. 第二遍开搞：导入模块跑通后开始第二轮。