1. 引言在上一篇实战记录中我们已经完成了上传文档的解析工作将 PDF、Word、TXT 等格式的文档内容提取为纯文本。然而仅仅解析出文本还不够——要让 AI 智能体能够读懂并检索这些内容还需要将文本转换为向量并存入向量数据库。本篇是 RAG检索增强生成项目实战的第七篇核心任务是将解析后的文档内容进行切割、向量化并写入 Milvus 向量数据库为后续的语义检索和问答环节打好数据基础。2. 整体流程概览文档从上传到入库整体流程如下上传文档文档解析文本切割向量化 Embedding写入 Milvus可供语义检索本篇聚焦 C、D、E 三个环节文本切割、向量化、入库。3. 环境配置3.1 .env 文件新增参数在项目根目录的.env文件中新增以下配置# 文本切割长度配置 CHUNK_SIZE500 CHUNK_OVERLAP50 # 本地模型驱动cpu:cpugpu:cuda互联网:remote DEVICEcpu # 向量模型地址可以从前面第4篇提到的国内模型官网上面下载一个向量模型放到指定的文件夹下 EMBEDDING_MODEL_PATHD:/AI_Model/modelscope/AI-ModelScope/bge-base-zh-v1.5 # 向量数据库配置 MILVUS_URLhttp://192.168.4.77:19530 MILVUS_USERroot MILVUS_PASSWORD****** # 向量库集合索引 MILVUS_COLLECTION_WEBSITEai_demo_website MILVUS_COLLECTION_DOCai_demo_doc说明以上配置中数据库地址、用户名、密码等敏感信息已做脱敏处理实际部署时请替换为真实值。3.2 settings.py 新增参数在settings.py中将上述环境变量映射为配置字段# 文本切割长度配置chunk_size:intchunk_overlap:int# 模型驱动device:str# 向量模型embedding_model_path:str# 知识向量库milvus_url:strmilvus_user:strmilvus_password:str# 向量库索引milvus_collection_website:strmilvus_collection_doc:str4. 文档解析后切割这个操作实际在上一篇的内容已有文档解析得到完整文本后需要根据配置的chunk_size进行切割。切割的目的是将长文本拆分为适合向量化的片段避免单条文本过长导致向量化效果变差。# 小于文本字数时无需切割iflen(doc_str)setting.chunk_size:splits[doc_str]else:# 创建文本切割器text_splitterRecursiveCharacterTextSplitter(chunk_sizesetting.chunk_size,chunk_overlapsetting.chunk_overlap)splitstext_splitter.split_text(doc_str)4.1 切割参数说明chunk_size500每个文本块的最大字符数。过大会导致向量化精度下降过小则会丢失上下文语义。chunk_overlap50相邻文本块之间的重叠字符数。重叠部分可以保留跨块的上下文信息避免语义断裂。4.2 为什么使用 RecursiveCharacterTextSplitterRecursiveCharacterTextSplitter是 LangChain 提供的递归式文本切割器它会优先按照段落、句子、标点等自然边界进行切割尽量保持语义完整性而不是简单粗暴地按固定长度硬切。5. 文档切割后入库这个操作实际在上一篇的内容已有文本切割完成后将每个文本块向量化并写入 Milvus# 确保文档集合存在且 schema 正确doc_str 为文本字段、vector 为向量字段milvus_service.ensure_doc_collection(setting.milvus_collection_doc)# 写入向量库milvus_service.batch_save_update(collection_namesetting.milvus_collection_doc,document_list[{id:str(uuid4()),name:file.filename,doc_str:split}forsplitinsplits],embedding_column_namedoc_str,saveTrue)5.1 入库逻辑说明ensure_doc_collection确保目标集合存在且 schema 正确doc_str为文本字段、vector为向量字段。batch_save_update批量写入向量库。每个文本块生成一个 UUID 作为主键并携带文件名和文本内容。embedding_column_name‘doc_str’指定对doc_str字段进行向量化。saveTrue表示执行插入操作若为False则执行 upsert更新或插入。6. 用于操作 Milvus 的 Service下面是与 Milvus 交互的核心 Service 层代码封装了集合管理、数据写入、查询、删除等常用操作。#!/usr/bin/env python# -*- coding: utf-8 -*-# File : milvus_service.py# Author: hqg# Date : 2024/9/6# Desc : 用于操作milvus的servicefromtypingimportList,Dict,UnionfrompymilvusimportMilvusClient,CollectionSchema,FieldSchema,DataTypefromloguruimportloggerfromapp.configimportsettingfromapp.service.systemimportembedding_service logger.info(连接Milvus...)milvus_clientMilvusClient(urisetting.milvus_url,usersetting.milvus_user,passwordsetting.milvus_password)logger.info(Milvus连接成功)defbatch_save_update(collection_name:str,document_list:List[Dict],embedding_column_name:strtext,save:boolTrue):text_list[doc[embedding_column_name]fordocindocument_list]embedding_listembedding_service.embedding_text(text_list)foriinrange(len(text_list)):document_list[i][vector]embedding_list[i]ifsaveisTrue:milvus_client.insert(collection_namecollection_name,datadocument_list)else:milvus_client.upsert(collection_namecollection_name,datadocument_list)defdelete_by_filter(collection_name:str,ids:Union[list,str,int]None,filter:str)-Dict:returnmilvus_client.delete(collection_namecollection_name,filterfilter,idsids)defbuild_filter_by_dict(filter_dict:Dict)-str: 将字典条件转换为字符串条件 :param filter_dict: 字典条件 :return: 字符串条件 filter_parts[]forkey,valueinfilter_dict.items():ifvalueisNone:continue# if value is str, add double quotesifisinstance(value,str):filter_parts.append(f{key} {value})# if value is list, use inelifisinstance(value,list):in_str, .join([f{v}forvinvalue])filter_parts.append(f{key}in [{in_str}])else:raiseValueError(fUnsupported filter value type:{type(value)})filter_parts[(part)forpartinfilter_parts]return and .join(filter_parts)defquery(collection_name:str,filter:strNone,limit20,output_fields:List[str][text])-List[Dict]:returnmilvus_client.query(collection_namecollection_name,filterfilter,limitlimit,output_fieldsoutput_fields)defsearch(collection_name:str,embedding:List[float],limit:int,filter:strNone,output_fields:List[str][vector])-List[Dict]:resmilvus_client.search(collection_namecollection_name,data[embedding],limitlimit,filterfilter,output_fieldsoutput_fields)return[eforeinres[0]]defcheck_collection(collection_name:str): 检查集合是否存在 :param collection_name: 集合名称 :return: returnmilvus_client.has_collection(collection_name)defcreate_collection(collection_name:str,schema:CollectionSchema,primary_field_name:strid,id_type:strstring,vector_field_name:strvector,properties:DictNone): 创建集合 :param collection_name: 集合名称 :param schema: 集合结构 :param primary_field_name: 主键名字 :param id_type: id类型 :param vector_field_name: 向量字段名 :param properties: 集合属性 # 创建集合的具体实现pass6.1 核心方法说明方法功能说明batch_save_update批量向量化并写入/更新数据是入库的核心方法delete_by_filter按条件或 ID 删除数据build_filter_by_dict将字典条件转换为 Milvus 过滤表达式字符串query按条件查询数据search向量相似度检索返回最相似的 Top-K 结果check_collection检查集合是否存在create_collection创建集合入库后可以看到7. 小结本篇完成了 RAG 项目中文档向量化入库这一关键环节配置层在.env和settings.py中新增了文本切割、模型驱动、向量模型、Milvus 连接等配置。切割层使用RecursiveCharacterTextSplitter对解析后的长文本进行智能切割兼顾语义完整性与向量化效果。入库层通过milvus_service.batch_save_update将切割后的文本块批量向量化并写入 Milvus。Service 层封装了 Milvus 的增删查改与向量检索能力供上层业务调用。至此文档数据已经以向量的形式存储在 Milvus 中。下一篇实战记录我们将基于这些向量数据实现语义检索与 RAG 问答让 AI 智能体真正读懂文档内容。