切片向量化节点

本文档详细介绍知识库导入流程中的切片向量化节点（BgeEmbeddingNode），该节点负责为每个文档切片生成稠密向量和稀疏向量，为后续的混合检索提供向量基础。

学习理念：切片向量化是 RAG 系统的"翻译"环节——将文本切片"翻译"为计算机能理解的向量语言。BgeEmbeddingNode 的核心是用 BGE-M3 同时生成稠密向量（语义匹配）和稀疏向量（关键词匹配），两者互补实现混合检索。向量质量直接决定后续检索的精度。
海外对标：BgeEmbeddingNode 的"稠密+稀疏双向量"方案对标 LlamaIndex 的 HybridRetriever 和 Milvus 官方推荐的混合检索架构。BGE-M3 模型（BAAI / 北京智源）是 2025-2026 年开源社区最广泛使用的多语言混合嵌入模型，在 MTEB 中文榜单中排名前五。CSR 稀疏矩阵提取 + L2 归一化的流程与 Pinecone 的 SPARSE_INVERTED_INDEX 标准一致。

本节 AI 替代率：~85% | 人工干预率：~15%

角色	能力范围
🤖 AI 擅长	BGE-M3 模型调用代码、CSR 矩阵提取、L2 归一化实现、批量处理骨架、测试代码
👤 人类需理解	稠密 vs 稀疏向量的互补性、L2 归一化的数学含义、item_name 拼接的设计意图（让同商品切片向量更接近）

阅读指引

颜色	章节	AI 替代率	人工干预	说明
🟡	§1 任务目标	~95%	~5%	学习目标明确
🟢	§2 核心概念扫盲	~95%	~5%	全部是图片说明，无需阅读代码
🟡	§3 整体流程	~90%	~10%	理解数据流转即可
🟠	§4 分步实现	~80%	~20%	Step 5~7(向量提取)是核心逻辑
🔴	§4.4 代码实现	~75%	~25%	BgeEmbeddingNode 整体 + 稀疏向量归一化
🟢	§5 单元测试	~95%	~5%	看预期输出即可
🟡	§6 总结	~90%	~10%	要点回顾

技术栈健康度标签体系

技术	健康度	建议
BGE-M3	🔥 巅峰	多语言混合嵌入模型的事实标准。同时输出稠密+稀疏向量，Milvus 混合检索的标配。需 GPU 推理（CUDA 12+）。
CSR Sparse Matrix	🟢 稳定	BGE-M3 输出稀疏向量的标准格式。`indices` + `data` 提取为 `{token_id: weight}` 是固定套路。
L2 Normalization	🟢 稳定	向量归一化的标准方法，使 IP 度量等价于余弦相似度。原理简单但影响重大。
numpy	🔥 巅峰	数值计算标准库，稀疏向量归一化的核心依赖。
批处理 (Batch Processing)	🟢 稳定	大规模向量化的标准策略。`embedding_batch_size`（默认5）控制 GPU 显存占用。

体系说明：🟢🟡🟠🔴 标识学习优先级 / AI 替代率；🔥🟢⏳⚠️💀 标识技术栈健康度。

中英文对照表

English	中文	本质
Dense Vector	稠密向量	每个维度都有值的浮点数数组，捕捉语义相似性
Sparse Vector	稀疏向量	只有少数维度有值的字典，捕捉关键词精确匹配
Hybrid Embedding	混合嵌入	同时生成稠密和稀疏两种向量的嵌入方式
CSR (Compressed Sparse Row)	压缩稀疏行	稀疏矩阵的高效存储格式（indptr + indices + data）
L2 Normalization	L2 归一化	将向量缩放到单位长度，使内积等价于余弦相似度
Batch Processing	批处理	将大数据集切分为小块依次处理，控制内存/显存
Embedding Dimension	嵌入维度	稠密向量的长度（BGE-M3 为 1024）
IP (Inner Product)	内积	Milvus 中使用的向量相似度度量，等价于归一化后的余弦相似度

💡 程序员比喻
BGE-M3 混合嵌入 就像 Elasticsearch 的 multi_match query + term query——dense = match（语义匹配），sparse = term（精确匹配），两者组合 = should 子句。
CSR 矩阵提取 就像解析 .git/objects——indptr = commit tree，indices = blob hash，data = blob content。
L2 归一化 就像 git diff 的规范化——把不同长度的 diff 统一成相同的"尺度"，才能公平比较。
item_name 拼接 就像 Docker 的 LABEL——给每个容器（切片）打上同组标签，让同组容器更容易被一起找到。
批处理 就像 kubectl rollout 的 maxSurge——不一次性全部处理，分批 rollout，失败也只影响当前批次。

1. 任务目标

1.1 本章目标

通过本章学习，你将掌握：

BGE-M3 模型原理：理解稠密向量与稀疏向量的区别与互补性
CSR 稀疏矩阵：学会从 CSR 格式中提取稀疏向量
L2 归一化：理解向量归一化的数学原理及其在检索中的作用
批量处理策略：掌握大规模向量化的分批处理技术
错误容忍设计：学会设计单批失败不影响全局的健壮架构

1.2 涉及文件

knowledge/
├── processor/import_process/nodes/
│   └── bge_embedding.py           # 切片向量化节点（本章重点）
│
└── tools/
    ├── embedding_utils.py          # BGE-M3 模型封装
    └── normalize_sparse_vector.py  # 稀疏向量 L2 归一化

1.3 节点在流程中的位置

2. 核心概念扫盲

2.1 为什么切片需要向量化？

向量化是将文本转换为数学表示的过程，是实现语义检索的基础：

2.2 稠密向量 vs 稀疏向量

BGE-M3 模型同时输出两种向量，各有优势：

2.3 CSR 稀疏矩阵格式

BGE-M3 返回的稀疏向量采用 CSR（Compressed Sparse Row） 格式存储：

2.4 L2 归一化

L2 归一化将向量"拉伸"或"压缩"到单位长度，使得余弦相似度计算更准确：

2.5 批量处理策略

大规模向量化需要分批处理，以平衡内存占用和处理效率：

3. 切片向量化业务处理流程（总）

3.1 整体流程概述

3.2 数据流转图

4. 切片向量化业务处理流程（分）

4.1 目标

为每个文档切片生成高质量的稠密向量和稀疏向量，使其能够支持后续的混合检索（语义检索 + 关键词检索）。

4.2 需求分析

需求项	说明	解决方案
向量质量	需要同时支持语义检索和关键词精确匹配	使用 BGE-M3 生成稠密+稀疏双向量
商品关联	同一商品的切片向量应有较高相似度	拼接 item_name 到输入文本
大规模处理	可能有数百甚至上千个切片	分批处理，避免 OOM
错误容忍	单个切片失败不应影响全局	批次级错误捕获，失败时返回原数据
向量归一化	Milvus 使用 IP 度量，需要归一化	对稀疏向量做 L2 归一化

4.3 实现流程

4.3.1 实现流程图

4.3.2 具体实现步骤

Step 1：获取并验证切片数据

从状态中获取切片列表，并验证其有效性。

输入：

state: 图状态字典，包含前序节点处理后的数据

处理逻辑：

从 state 中获取 "chunks" 字段
检查 chunks 是否为非空列表
如果无效，抛出 EmbeddingError 异常

输出：

验证通过的切片列表 chunks

代码片段：

🟡 【P1 看注释就行】 Step 1 代码模式固定——从 state 获取 chunks → 判空 → 抛异常。

python

def process(self, state: ImportGraphState) -> ImportGraphState:
    config = get_config()

    # 获取切片
    chunks = state.get("chunks", [])
    if not isinstance(chunks, list) or not chunks:
        raise EmbeddingError("chunks 为空或无效", node_name=self.name)

    self.log_step("step_1", f"开始为 {len(chunks)} 个切片生成向量")

Step 2：初始化 BGE-M3 模型

获取或初始化 BGE-M3 嵌入模型实例。

处理逻辑：

调用 get_bge_m3_model() 获取模型单例
模型根据环境变量配置加载路径、设备和精度
首次调用时加载模型到 GPU，后续调用复用

环境变量配置：

BGE_M3_PATH: 模型路径，默认 "BAAI/bge-m3"
BGE_DEVICE: 推理设备，默认 "cuda:0"
BGE_FP16: 是否启用 FP16，默认启用

代码片段：

🟡 【P1 看注释就行】 Step 2 初始化代码固定——get_bge_m3_model() 获取单例，首次加载到 GPU。

python

# 初始化 BGE-M3
try:
    bge_m3_ef = get_bge_m3_model()
except Exception as e:
    raise EmbeddingError(f"初始化 BGE-M3 失败: {e}", node_name=self.name, cause=e)

Step 3：分批处理切片

按照配置的批次大小，分批调用模型生成向量。

输入：

chunks: 切片列表
batch_size: 每批处理的切片数量（来自配置）

处理逻辑：

按 batch_size 将切片列表切分为多个批次
对每个批次调用 _process_batch() 方法
将各批次结果合并到 output_data 列表

代码片段：

🟡 【P1 看注释就行】 Step 3 分批处理——range(0, len(chunks), batch_size) 的标准分片模式。

python

# 批量处理
output_data = []
batch_size = config.embedding_batch_size

for i in range(0, len(chunks), batch_size):
    batch = chunks[i:i + batch_size]
    batch_output = self._process_batch(bge_m3_ef, batch, i, len(chunks))
    output_data.extend(batch_output)

Step 4：构造输入文本

为每个切片构造模型输入文本。

处理逻辑：

拼接 item_name 和 content 字段
使用换行符分隔
处理空值情况（转为空字符串）

设计原因：

将商品名称编码到向量中，使同一商品的切片具有更高相似度
便于后续按商品维度进行向量过滤和聚合

代码片段：

🟡 【P1 看注释就行】 Step 4 输入构造——拼接 item_name + content，使同商品切片向量更接近。

python

# 构造输入文本：item_name + content
texts = [
    (doc.get("item_name", "") or "") + "\n" + (doc.get("content", "") or "")
    for doc in batch
]

Step 5：调用模型生成向量

批量调用 BGE-M3 模型生成嵌入。

输入：

texts: 拼接后的文本列表

输出：

embeddings: 包含 "dense" 和 "sparse" 两个键的字典
- dense: numpy.ndarray，形状为 (batch_size, 1024)
- sparse: scipy.sparse.csr_matrix，CSR 格式稀疏矩阵

代码片段：

🟡 【P1 看注释就行】 Step 5 模型调用——encode_documents(texts) 批量生成。注意 not embeddings 的空值检查。

python

# 批量生成向量
embeddings = bge_m3_ef.encode_documents(texts)

if not embeddings:
    self.logger.warning(f"批次 {start_idx + 1}-{start_idx + len(batch)} 未能生成向量")
    return batch  # 返回原始数据

Step 6：提取稠密向量

从模型输出中提取稠密向量。

处理逻辑：

从 embeddings["dense"] 中按索引获取对应行
调用 .tolist() 转换为 Python 列表

代码片段：

🟢 【P2 后面可以查】 Step 6 稠密向量提取——embeddings["dense"][j].tolist() 是最简单的部分。

python

# 提取稠密向量
dense_vector = embeddings["dense"][j].tolist()

Step 7：提取并归一化稀疏向量

从 CSR 矩阵中提取稀疏向量并进行 L2 归一化。

处理逻辑：

使用 indptr 数组确定当前行的起止索引
从 indices 数组提取 Token ID
从 data 数组提取对应权重
组装为字典格式
调用 normalize_sparse_vector() 进行 L2 归一化

代码片段：

🔥 【P0 必须要学】 Step 7 稀疏向量提取是混合检索的核心。理解 CSR 三数组：indptr[j] 到 indptr[j+1] 确定当前行的非零元素范围，indices = Token ID，data = 权重。最后 normalize_sparse_vector() 做 L2 归一化。

python

# 提取稀疏向量
start = embeddings["sparse"].indptr[j]
end = embeddings["sparse"].indptr[j + 1]
token_ids = embeddings["sparse"].indices[start:end].tolist()
weights = embeddings["sparse"].data[start:end].tolist()
sparse_dict = dict(zip(token_ids, weights))

# L2 归一化
sparse_vector = normalize_sparse_vector(sparse_dict)

Step 8：组装输出数据

将原始切片信息与生成的向量合并为输出字典。

输出字段：

content: 切片文本内容
title: 切片标题
parent_title: 父级标题
part: 切片序号
file_title: 源文件标题
item_name: 商品名称
dense_vector: 稠密向量（新增）
sparse_vector: 稀疏向量（新增）

代码片段：

🟡 【P1 看注释就行】 Step 8 组装输出——将原始字段 + 向量合并为一个字典。注意保留 content / title 等字段供后续节点使用。

python

# 构建输出
item = {
    "content": doc.get("content"),
    "title": doc.get("title"),
    "parent_title": doc.get("parent_title", ""),
    "part": doc.get("part", 0),
    "file_title": doc.get("file_title"),
    "item_name": doc.get("item_name"),
    "dense_vector": dense_vector,
    "sparse_vector": sparse_vector
}
output.append(item)

Step 9：更新状态并返回

将处理结果写回状态并返回。

代码片段：

🟢 【P2 后面可以查】 Step 9 返回——更新 state["chunks"]。

python

self.log_step("step_2", f"向量化完成，共 {len(output_data)} 个切片")
state["chunks"] = output_data

return state

4.4 代码实现

4.4.1 完整节点代码

🔥 【P0 必须要学】 BgeEmbeddingNode 是导入流程中"从文本到向量"的转换节点。重点理解：_process_batch 的完整流程——构造文本 → encode → 提取稠密 → 提取稀疏(CSR) → 归一化 → 组装。try/except 确保单批失败返回原始数据（降级）。

python

"""
BGE-M3 向量化节点

为文档切片生成稠密和稀疏向量
"""
import json
import os
from typing import List

from knowledge.processor.import_process.base import BaseNode, setup_logging
from knowledge.processor.import_process.state import ImportGraphState
from knowledge.processor.import_process.config import get_config
from knowledge.processor.import_process.exceptions import EmbeddingError
from knowledge.tools.embedding_utils import get_bge_m3_model
from knowledge.tools.normalize_sparse_vector import normalize_sparse_vector


class BgeEmbeddingNode(BaseNode):
    """
    BGE-M3 向量化节点

    为每个切片生成稠密向量和稀疏向量，
    用于后续的向量检索。
    """

    name = "bge_embedding"

    def process(self, state: ImportGraphState) -> ImportGraphState:
        """
        执行向量化

        Args:
            state: 图状态

        Returns:
            更新后的状态（chunks 中包含向量）
        """
        config = get_config()

        # 获取切片
        chunks = state.get("chunks", [])
        if not isinstance(chunks, list) or not chunks:
            raise EmbeddingError("chunks 为空或无效", node_name=self.name)

        self.log_step("step_1", f"开始为 {len(chunks)} 个切片生成向量")

        # 初始化 BGE-M3
        try:
            bge_m3_ef = get_bge_m3_model()
        except Exception as e:
            raise EmbeddingError(f"初始化 BGE-M3 失败: {e}", node_name=self.name, cause=e)

        # 批量处理
        output_data = []
        batch_size = config.embedding_batch_size

        for i in range(0, len(chunks), batch_size):
            batch = chunks[i:i + batch_size]
            batch_output = self._process_batch(bge_m3_ef, batch, i, len(chunks))
            output_data.extend(batch_output)

        self.log_step("step_2", f"向量化完成，共 {len(output_data)} 个切片")
        state["chunks"] = output_data

        return state

    def _process_batch(
            self,
            bge_m3_ef,
            batch: List[dict],
            start_idx: int,
            total: int
    ) -> List[dict]:
        """处理一个批次的切片"""
        try:
            # 构造输入文本：item_name + content
            texts = [
                (doc.get("item_name", "") or "") + "\n" + (doc.get("content", "") or "")
                for doc in batch
            ]

            # 批量生成向量
            embeddings = bge_m3_ef.encode_documents(texts)

            if not embeddings:
                self.logger.warning(f"批次 {start_idx + 1}-{start_idx + len(batch)} 未能生成向量")
                return batch

            # 处理结果
            output = []
            for j, doc in enumerate(batch):
                # 提取稠密向量
                dense_vector = embeddings["dense"][j].tolist()

                # 提取稀疏向量
                start = embeddings["sparse"].indptr[j]
                end = embeddings["sparse"].indptr[j + 1]
                token_ids = embeddings["sparse"].indices[start:end].tolist()
                weights = embeddings["sparse"].data[start:end].tolist()
                sparse_dict = dict(zip(token_ids, weights))
                sparse_vector = normalize_sparse_vector(sparse_dict)

                # 构建输出
                item = {
                    "content": doc.get("content"),
                    "title": doc.get("title"),
                    "parent_title": doc.get("parent_title", ""),
                    "part": doc.get("part", 0),
                    "file_title": doc.get("file_title"),
                    "item_name": doc.get("item_name"),
                    "dense_vector": dense_vector,
                    "sparse_vector": sparse_vector
                }
                output.append(item)

            self.logger.info(
                f"成功处理批次 {start_idx + 1}-{min(start_idx + len(batch), total)}/{total}"
            )
            return output

        except Exception as e:
            self.logger.error(
                f"批次 {start_idx + 1}-{start_idx + len(batch)} 处理失败: {e}"
            )
            # 返回原始数据，不含向量
            return batch

4.4.2 稀疏向量归一化工具

🟡 【P1 看注释就行】 normalize_sparse_vector() 工具函数——np.linalg.norm(values) 计算 L2 范数，values / l2_norm 归一化。注意 1e-9 的零范数保护。

python

import numpy as np

def normalize_sparse_vector(sparse_vec):
    """
    对稀疏向量做 L2 归一化（仅处理非零维度，不影响零维度）

    :param sparse_vec: 原始稀疏向量（dict 格式：{维度: 数值}）
    :return: 归一化后的稀疏向量
    """
    if not sparse_vec:  # 空向量直接返回
        return sparse_vec

    # 提取非零维度的数值
    values = np.array(list(sparse_vec.values()), dtype=np.float64)

    # 计算 L2 范数（避免除以 0）
    l2_norm = np.linalg.norm(values)
    if l2_norm < 1e-9:  # 范数接近 0 时，直接返回原向量
        return sparse_vec

    # 归一化：每个数值除以 L2 范数
    normalized_values = values / l2_norm

    # 重建稀疏向量 dict
    return dict(zip(sparse_vec.keys(), normalized_values))

5. 单元测试

在 bge_embedding.py 文件末尾添加独立测试代码：

🟢 【P2 后面可以查】 测试代码量大但模式固定——读取 JSON → process → 验证向量维度。看预期输出即可。

python

# ================================================================== #
#                        兼容 & 测试                                   #
# ================================================================== #

# 兼容原有调用方式
node_bge_embedding = BgeEmbeddingNode()

if __name__ == '__main__':
    """
    独立测试 BgeEmbeddingNode

    测试流程：
    1. 从上一个节点的输出文件读取状态
    2. 执行向量化处理
    3. 将结果保存到临时文件
    4. 验证输出数据结构
    """

    setup_logging()

    # ----------------------------------------------------------------
    # Step 1: 配置路径
    # ----------------------------------------------------------------
    temp_dir = r"D:\develop\develop\workspace\pycharm\usage\shopkeeper_brain_v260213\knowledge\processor\import_process\temp"

    # 输入：上一个商品识别节点处理后的状态
    input_path = os.path.join(temp_dir, "chunks_item_name.json")

    # 输出：向量化后的状态
    output_path = os.path.join(temp_dir, "chunks_item_name_vector.json")

    # ----------------------------------------------------------------
    # Step 2: 读取输入数据
    # ----------------------------------------------------------------
    print(f"正在读取输入文件: {input_path}")

    with open(input_path, "r", encoding="utf-8") as f:
        content = json.load(f)

    chunks = content.get('chunks', [])
    print(f"读取到 {len(chunks)} 个切片")

    # ----------------------------------------------------------------
    # Step 3: 构建状态并执行处理
    # ----------------------------------------------------------------
    state = {
        "chunks": chunks
    }

    print("开始执行向量化...")
    result = node_bge_embedding.process(state)

    # ----------------------------------------------------------------
    # Step 4: 验证输出数据
    # ----------------------------------------------------------------
    output_chunks = result.get("chunks", [])
    print(f"\n处理完成，共 {len(output_chunks)} 个切片")

    # 检查第一个切片的向量
    if output_chunks:
        first_chunk = output_chunks[0]

        print("\n第一个切片数据结构:")
        print(f"  - content: {first_chunk.get('content', '')[:50]}...")
        print(f"  - title: {first_chunk.get('title', '')}")
        print(f"  - item_name: {first_chunk.get('item_name', '')}")

        dense_vec = first_chunk.get('dense_vector', [])
        sparse_vec = first_chunk.get('sparse_vector', {})

        print(f"\n向量信息:")
        print(f"  - dense_vector 维度: {len(dense_vec)}")
        print(f"  - dense_vector 前5维: {dense_vec[:5] if dense_vec else '无'}")
        print(f"  - sparse_vector 非零元素数: {len(sparse_vec)}")
        print(f"  - sparse_vector 前3项: {dict(list(sparse_vec.items())[:3]) if sparse_vec else '无'}")

        # 验证稀疏向量是否已归一化
        if sparse_vec:
            import numpy as np
            values = np.array(list(sparse_vec.values()))
            l2_norm = np.linalg.norm(values)
            print(f"  - sparse_vector L2范数: {l2_norm:.6f} (归一化后应接近1.0)")

    # ----------------------------------------------------------------
    # Step 5: 保存输出文件
    # ----------------------------------------------------------------
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=4)

    print(f"\n已保存到: {output_path}")

测试执行

bash

# 进入项目目录
cd D:\develop\develop\workspace\pycharm\usage\shopkeeper_brain_v260213

# 执行测试
python -m knowledge.processor.import_process.nodes.bge_embedding

预期输出

正在读取输入文件: D:\...\temp\chunks_item_name.json
读取到 15 个切片
开始执行向量化...
2024-xx-xx [INFO] bge_embedding - [step_1] 开始为 15 个切片生成向量
2024-xx-xx [INFO] bge_embedding - 成功处理批次 1-15/15
2024-xx-xx [INFO] bge_embedding - [step_2] 向量化完成，共 15 个切片

处理完成，共 15 个切片

第一个切片数据结构:
  - content: 福禄克15B+数字万用表是一款专业级测量仪器...
  - title: 产品介绍
  - item_name: 福禄克15B+数字万用表

向量信息:
  - dense_vector 维度: 1024
  - dense_vector 前5维: [0.023456, -0.156789, 0.089012, ...]
  - sparse_vector 非零元素数: 47
  - sparse_vector 前3项: {12456: 0.234567, 7823: 0.198765, 45612: 0.176543}
  - sparse_vector L2范数: 1.000000 (归一化后应接近1.0)

已保存到: D:\...\temp\chunks_item_name_vector.json

6. 总结

本章详细介绍了切片向量化节点的实现，核心要点包括：

BGE-M3 双向量：同时生成稠密向量（语义理解）和稀疏向量（关键词匹配）
CSR 矩阵解析：使用 indptr、indices、data 三数组提取稀疏向量
L2 归一化：使稀疏向量适配 Milvus 的 IP 度量
批量处理：分批调用模型，控制内存并支持错误容忍
商品名拼接：将 item_name 编码到向量中，提升同商品切片的相似度

企业痛点映射

痛点	传统方案	AI Agent 向量化方案	效率提升
文本无法被语义检索	关键词搜索（搜"万用表"搜不到"数字多用表"）	稠密向量捕捉语义相似性	语义搜索召回率提升 ~60%（预估）
商品型号无法精确匹配	语义搜索混淆 "15B+" 和 "15B"	稀疏向量保留关键词精确匹配	型号匹配准确率 ~100%（稀疏向量精确命中）
大规模文档向量化 OOM	一次性全部加载到 GPU	分批处理（batch_size 可配置）	GPU 显存占用降低 ~80%
同商品切片分散难聚合	文本写入时手动分组	item_name 拼接进向量	同商品切片向量相似度自动提升

Remote & Agent 应用场景价值

Remote 场景价值：BGE-M3 模型存放在远端 GPU 服务器上（通过 BGE_M3_PATH / BGE_DEVICE 配置），团队成员无需本地 GPU 即可调用。批处理机制适合在 CI/CD Pipeline 中自动化运行。
Agent 落地场景：BgeEmbeddingNode 可封装为"文档向量化 Agent"——Agent 接收 chunks → 调用 BGE-M3 生成双向量 → 输出带向量的数据。参数 batch_size / device 可作为 Agent 的动态参数，根据显存容量自适应调整。

Git Commit 对应

本节切片向量化节点对应的提交记录（参考值，以实际版本为准）：

<待补充 — 建议在项目仓库中搜索 "bge_embedding.py" / "normalize_sparse_vector.py" 相关提交>

bash

cd shopkeeper_brain
# 查看向量化相关代码
git log --oneline --all -- knowledge/processor/import_process/nodes/bge_embedding.py

切片向量化节点 ​

阅读指引 ​

技术栈健康度标签体系 ​

中英文对照表 ​

1. 任务目标 ​

1.1 本章目标 ​

1.2 涉及文件 ​

1.3 节点在流程中的位置 ​

2. 核心概念扫盲 ​

2.1 为什么切片需要向量化？ ​

2.2 稠密向量 vs 稀疏向量 ​

2.3 CSR 稀疏矩阵格式 ​

2.4 L2 归一化 ​

2.5 批量处理策略 ​

3. 切片向量化业务处理流程（总） ​

3.1 整体流程概述 ​

3.2 数据流转图 ​

4. 切片向量化业务处理流程（分） ​

4.1 目标 ​

4.2 需求分析 ​

4.3 实现流程 ​

4.3.1 实现流程图 ​

4.3.2 具体实现步骤 ​

Step 1：获取并验证切片数据 ​

Step 2：初始化 BGE-M3 模型 ​

Step 3：分批处理切片 ​

Step 4：构造输入文本 ​

Step 5：调用模型生成向量 ​

Step 6：提取稠密向量 ​

Step 7：提取并归一化稀疏向量 ​

Step 8：组装输出数据 ​

Step 9：更新状态并返回 ​

4.4 代码实现 ​

4.4.1 完整节点代码 ​

4.4.2 稀疏向量归一化工具 ​

5. 单元测试 ​

测试执行 ​

预期输出 ​

6. 总结 ​

企业痛点映射 ​

Remote & Agent 应用场景价值 ​

Git Commit 对应 ​

切片向量化节点

阅读指引

技术栈健康度标签体系

中英文对照表

1. 任务目标

1.1 本章目标

1.2 涉及文件

1.3 节点在流程中的位置

2. 核心概念扫盲

2.1 为什么切片需要向量化？

2.2 稠密向量 vs 稀疏向量

2.3 CSR 稀疏矩阵格式

2.4 L2 归一化

2.5 批量处理策略

3. 切片向量化业务处理流程（总）

3.1 整体流程概述

3.2 数据流转图

4. 切片向量化业务处理流程（分）

4.1 目标

4.2 需求分析

4.3 实现流程

4.3.1 实现流程图

4.3.2 具体实现步骤

Step 1：获取并验证切片数据

Step 2：初始化 BGE-M3 模型

Step 3：分批处理切片

Step 4：构造输入文本

Step 5：调用模型生成向量

Step 6：提取稠密向量

Step 7：提取并归一化稀疏向量

Step 8：组装输出数据

Step 9：更新状态并返回

4.4 代码实现

4.4.1 完整节点代码

4.4.2 稀疏向量归一化工具

5. 单元测试

测试执行

预期输出

6. 总结

企业痛点映射

Remote & Agent 应用场景价值

Git Commit 对应