目录1. 节点作用与实现思路2. 步骤分解3. 准备Embedding模型和工具3.1 什么是 “生成词向量”3.2 “稀疏向量 稠密向量”3.3 安装Python依赖库3.4 Embedding下载模型3.5 工具代码导入1. 节点作用与实现思路作为文档结构化解析与差异化分类的核心关键节点依托大语言模型深度语义理解能力精准萃取文档核心主体、业务实体与专属概念快速判定文档所属品类与内容属性。通过全局主体标识绑定实现多源文档精准区分、内容归类、数据去重与精细化管控搭建实体与文本切片的强关联映射体系为后续实体级检索、语义对齐、定向过滤、结构化问答筑牢底层支撑大幅提升知识库检索精准度与数据治理能力。实现思路关键上下文精准裁剪优先截取文档高价值头部切片涵盖标题、概述等核心摘要信息精简输入上下文在控制推理成本的同时保障大模型主体识别的准确率。大模型语义萃取识别结合定制化业务提示词依托 LLM 深层语义解析能力智能提取文档核心主体、专属名词与业务标识完成文档类型自动判别与内容切面划分。全链路容错兜底设计针对大模型输出不稳定、识别异常、返回空值等场景配置异常捕获与默认兜底策略保障导入流程稳定运行避免单点故障中断全链路任务。实体向量化预处理将识别后的标准主体实体统一完成向量编码对接向量库实现跨表述语义匹配打通别名关联、语义联动能力实现模糊检索与精准召回。2. 步骤分解导入与配置: 引入必要的库LangChain, Milvus, etc.及配置参数。核心辅助函数: 包含字符串安全转义等辅助逻辑。主流程定义: LangGraph 节点的入口函数串联各个步骤。步骤 1: 获取输入: 校验 State 中的file_title和chunks。步骤 2: 构建上下文: 截取前 K 个切片作为 LLM 的识别素材。步骤 3: 调用 LLM: 使用大模型识别商品名称包含错误重试与兜底。步骤 4: 回填数据: 将识别结果更新回 State 和 Chunks 元数据。步骤 5: 生成向量: 调用 Embedding 模型生成 Dense/Sparse 向量。步骤 6: 保存结果: 将数据写入 Milvus 向量库并处理幂等性。单元测试: 独立运行的测试代码验证核心流程。3. 准备Embedding模型和工具3.1 什么是 “生成词向量”词向量Word Vector/Embedding就是把文字比如 “苏泊尔 5000W 大功率电磁炉”转换成计算机能理解的数字列表向量的过程。打个比方人类理解文字“苹果手机” 品牌苹果 品类手机计算机理解文字没法直接懂 “苹果手机”但能懂[0.23, -0.56, 1.89, ...]这样的数字列表词向量的作用把文字的语义信息含义、特征、关联度编码成数字让计算机能 “计算文字相似度”“分类文字”“检索相似内容”。举个简单例子文字对应的词向量简化版实际是几百 / 几千维苹果手机[0.23, -0.56, 1.89, 0.78]华为手机[0.21, -0.58, 1.91, 0.76]苹果笔记本[0.22, -0.55, 0.87, 0.79]计算机通过对比这些数字列表的相似度就能判断“苹果手机” 和 “华为手机” 更像数字差异小“苹果手机” 和 “苹果笔记本” 相似度低数字差异大。3.2 “稀疏向量 稠密向量”代码是基于BGE-M3模型生成两种词向量这是当前主流的多模态嵌入方案拆解类型特点用途稠密向量Dense Vector长度固定比如 768 维 / 1024 维每个位置都是连续数值如 0.23、-0.56捕捉文字的语义信息比如 “苹果手机” 的核心含义适合相似度计算稀疏向量Sparse Vector长度极长比如几十万维但只有少数位置有非 0 值其余都是 0捕捉文字的关键词 / 字面特征比如 “苹果”“5000W”“电磁炉”适合精准检索BGE-M3 模型同时输出这两种向量结合使用能兼顾 “语义理解” 和 “精准匹配”。3.3 安装Python依赖库在使用模型之前需要安装相关的 Python 依赖库。# 环境安装命令适配BGE-M3MilvusGPU/CPU版区分 # 【GPU版】安装CUDA 12.4版PyTorch含torchvision/torchaudioNVIDIA显卡GPU加速必备 # 适配有NVIDIA独显且驱动≥551.61后续BGE-M3可开启FP16半精度推理 uv pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu124 # 【备用-CPU版】无NVIDIA显卡AMD/Intel集显请用此命令直接安装CPU版PyTorch # 注释掉上方GPU版命令取消注释下方即可 uv add torch torchvision torchaudio # 安装Milvus和BGE-M3核心依赖所有环境必装无GPU/CPU区分 # pymilvus[model]Milvus Python客户端带模型相关依赖适配向量入库/检索 # FlagEmbeddingBGE-M3向量生成模型的核心依赖不可替代 # transformersFlagEmbedding底层依赖Hugging Face模型运行库 uv add pymilvus[model] FlagEmbedding transformers #⚠️安装FlagEmbedding的时候会自动安装一个cpu版本的torch替换掉之前的gpu版本的torch # 要解决这个问题需要做以下几个步骤 # 步骤1 先删除已经安装的FlagEmbedding先在pyproject.toml中确定一下自己安装的版本 uv remove FlagEmbeddin # 步骤2 将以下内容配置在pyproject.toml中 dependencies [ 其他之前安装过的配置, flagembeddingv1.3.5, torch2.10.0, torchvision0.25.0, torchaudio2.10.0, ] [tool.uv.sources] # 强制从 NVIDIA 源安装 torch { index pytorch-cuda } torchvision { index pytorch-cuda } torchaudio { index pytorch-cuda } [[tool.uv.index]] name pytorch-cuda url https://download.pytorch.org/whl/cu128 explicit true # 步骤3 删除锁文件并重新锁定 rm uv.lock uv lock # 步骤4重新同步环境 uv sync --reinstall # 步骤5验证 uv run python -c import torch; print(GPU:, torch.cuda.is_available())CUDA 每个版本都有最低算力要求CUDA 12.4 要求显卡的CUDA 算力≥3.5几乎 2016 年之后的 NVIDIA 独显都满足老款如 GTX 750 Ti 也达标主流显卡RTX30/40 系、GTX16/20 系全兼容几乎不用担心里程碑。直接打开 NVIDIA 官方算力表搜索自己的显卡型号看对应的Compute Capability算力数值NVIDIA 显卡 CUDA 算力官方查询地址桌面显卡看GeForce栏笔记本显卡看GeForce Notebook栏示例RTX 3060 算力 8.6、GTX 1650 算力 7.5、RTX 4090 算力 8.9都远大于 3.5完美适配 CUDA 12.4。3.4 Embedding下载模型如果访问 HuggingFace 较慢可以使用阿里云(阿里巴巴通义实验室原达摩院)的 ModelScope 社区下载。https://www.modelscope.cn/models/BAAI/bge-m31.安装 modelscope 库uv add modelscope2.运行 Python 脚本下载 创建一个临时的 Python 脚本例如 download_bge.py并运行frommodelscope.hub.snapshot_downloadimportsnapshot_download# 下载模型到当前目录下的 models/bge-m3 文件夹model_dirsnapshot_download(BAAI/bge-m3,cache_dirD:/ai_models/modelscope_cache/models)print(f模型已下载到:{model_dir})3. .env配置#embedding配置 # BGE-M3模型本地缓存/部署路径本地加载模型时使用指向ModelScope下载的模型目录 BGE_M3_PATHD:\ai_models\modelscope_cache\models\BAAI\bge-m3 # BGE-M3模型官方标识ModelScope/HuggingFace通用拉取模型时使用 BGE_M3BAAI/bge-m3 # BGE-M3运行设备cuda:0表示使用第1块GPUcpu表示使用CPUcuda:N表示第N1块GPU BGE_DEVICEcuda:0 # BGE-M3是否开启FP16半精度推理1开启GPU加速更高效0关闭兼容低版本GPU/CPU BGE_FP1614. 配置参数读取文件app.config.embedding_config.py# 导入核心依赖数据类、环境变量读取、路径处理fromdataclassesimportdataclassimportosfromdotenvimportload_dotenv# 提前加载.env配置文件保持和原代码一致只需执行一次load_dotenv()# 定义Embedding配置适配BGE-M3的所有配置类名embedding_configdataclassclassEmbeddingConfig:bge_m3_path:str# 本地模型路径bge_m3:str# 模型仓库标识bge_device:str# 运行设备(cuda:0/cpu)bge_fp16:bool# 是否开启半精度1True/0False# 实例化配置对象和原代码lm_config风格保持一致embedding_configEmbeddingConfig(bge_m3_pathos.getenv(BGE_M3_PATH),bge_m3os.getenv(BGE_M3),bge_deviceos.getenv(BGE_DEVICE),# 特殊处理将.env中的1/0转为布尔值兼容常见的数字/字符串格式bge_fp16os.getenv(BGE_FP16)in(1,True,true,1))3.5 工具代码导入文件app.lm.embedding_utils.pyfrompymilvus.model.hybridimportBGEM3EmbeddingFunctionfromapp.core.loggerimportloggerfromapp.conf.embedding_configimportembedding_config# 模型单例对象避免重复初始化_bge_m3_efNonedefget_bge_m3_ef(): 获取BGE-M3模型单例对象自动加载环境变量配置 :return: 初始化完成的BGEM3EmbeddingFunction实例 global_bge_m3_ef# 单例模式已初始化则直接返回避免重复加载模型if_bge_m3_efisnotNone:logger.debug(BGE-M3模型单例已存在直接返回实例)return_bge_m3_ef# 从环境变量加载配置无配置则使用默认值# 本地有可以使用本地地址 没有使用 BAAI/bge-m3 会自动下载 如果云端部署也可以使用url地址model_nameembedding_config.bge_m3_pathorBAAI/bge-m3deviceembedding_config.bge_deviceorcpuuse_fp16embedding_config.bge_fp16orFalse# 打印模型初始化配置便于问题排查logger.info(开始初始化BGE-M3模型,extra{model_name:model_name,device:device,use_fp16:use_fp16,normalize_embeddings:True})try:# 初始化BGE-M3模型开启原生L2归一化适配Milvus IP内积检索# pymilvus.model.hybrid.BGEM3EmbeddingFunction 在工程上最大的好处是# 和 Milvus 检索链路天然对齐 上线更稳更省事。_bge_m3_efBGEM3EmbeddingFunction(model_namemodel_name,devicedevice,use_fp16use_fp16,normalize_embeddingsTrue# 模型原生对稠密稀疏向量做L2归一化)logger.success(BGE-M3模型初始化成功已开启原生L2归一化)# “它把所有向量拉伸到统一长度模长为1让我们能在数据库中放心使用最快的内积IP检索既提速又不丢精度。”return_bge_m3_efexceptExceptionase:logger.error(fBGE-M3模型初始化失败{str(e)},exc_infoTrue)raise# 向上抛出异常由调用方处理defgenerate_embeddings(texts): 为文本列表生成稠密稀疏混合向量嵌入模型原生L2归一化 :param texts: 要生成嵌入的文本列表单文本也需封装为列表 :return: 字典格式的向量结果key为dense/sparse对应嵌套列表/字典列表 :raise: 向量生成过程中的异常由调用方捕获处理 # 入参合法性校验ifnotisinstance(texts,list)orlen(texts)0:logger.warning(生成向量入参不合法texts必须为非空列表)raiseValueError(参数texts必须是包含文本的非空列表)logger.info(f开始为{len(texts)}条文本生成混合向量嵌入)try:# 加载BGE-M3模型单例modelget_bge_m3_ef()# 模型编码生成向量返回dense稠密向量sparseCSR格式稀疏向量embeddingsmodel.encode_documents(texts)logger.debug(f模型编码完成开始解析稀疏向量格式共{len(texts)}条)# 初始化稀疏向量处理结果解析为字典格式适配序列化/存储processed_sparse[]# 把模型输出的 CSR 稀疏矩阵 按“每条文本一行”拆成 {特征索引: 权重} 字典# - indices 非零元素的“列号特征ID”# - data 对应列号的权重值# - indptr 每一行在 indices/data 里的起止位置指针# 数据示例:# indices [3, 8, 20, 1, 9]# data [0.7, 0.2, 0.1, 0.6, 0.4]# indptr [0, 3, 5]# 获取对应的数据# - 第0条文本用 0:3 indices[3,8,20] , data[0.7,0.2,0.1]# - 第1条文本用 3:5 indices[1,9] , data[0.6,0.4]foriinrange(len(texts)):# 提取第i个文本的稀疏向量索引np.int64 → Python int满足字典key可哈希要求sparse_indicesembeddings[sparse].indices[embeddings[sparse].indptr[i]:embeddings[sparse].indptr[i1]].tolist()# 提取第i个文本的稀疏向量权重np.float32 → Python float适配JSON序列化/接口返回sparse_dataembeddings[sparse].data[embeddings[sparse].indptr[i]:embeddings[sparse].indptr[i1]].tolist()# 构造{特征索引: 归一化权重}的稀疏向量字典sparse_dict{k:vfork,vinzip(sparse_indices,sparse_data)}processed_sparse.append(sparse_dict)# 构造最终返回结果稠密向量转列表解决numpy数组不可序列化问题result{dense:[emb.tolist()forembinembeddings[dense]],# 嵌套列表与输入文本一一对应sparse:processed_sparse# 字典列表模型已做L2归一化}logger.success(f{len(texts)}条文本向量生成完成格式已适配工业级使用)returnresultexceptExceptionase:logger.error(f文本向量生成失败{str(e)},exc_infoTrue)raise# 不吞异常向上传递让调用方做重试/降级处理 核心设计亮点适配说明 1. normalize_embeddingsTrue 的价值 - 检索更稳定 不同文本长短、词频差异不会把分数拉偏。 - IP 可近似 cosine 向量都归一化后 Inner Product 和余弦相似度等价Milvus 用 IP 检索就很合适。 - dense/sparse 都统一标尺 混合检索时两路分数更容易做融合不容易一边压死另一边。 - 减少异常高分 防止“模长大”的向量仅靠长度拿高分。 2. 彻底解决NumPy类型做key问题sparse_indices加.tolist()将np.int64转为Python原生int满足字典key的可哈希要求无报错风险 3. 稀疏值适配序列化sparse_data加.tolist()将np.float32转为Python原生float支持JSON写入/接口返回/Milvus入库等所有场景 4. 单例模式优化模型仅初始化一次避免重复加载耗时耗资源提升批量处理效率 5. 格式匹配业务调用返回dense嵌套列表、sparse字典列表与vector_result[dense][0]/sparse_vector[sparse][0]取值逻辑完美契合 6. 分级日志覆盖从模型初始化、向量生成到异常报错全流程日志记录便于生产环境问题排查 7. 入参合法性校验防止空列表/非列表入参导致的内部报错提升工具类健壮性。
