1. 引言
在上一篇实战记录中，我们已经完成了上传文档的解析工作，将 PDF、Word、TXT 等格式的文档内容提取为纯文本。然而，仅仅解析出文本还不够——要让 AI 智能体能够"读懂"并检索这些内容，还需要将文本转换为向量并存入向量数据库。
本篇是 RAG（检索增强生成）项目实战的第七篇，核心任务是：将解析后的文档内容进行切割、向量化，并写入 Milvus 向量数据库，为后续的语义检索和问答环节打好数据基础。
2. 整体流程概览
文档从上传到入库，整体流程如下：
本篇聚焦 C、D、E 三个环节：文本切割、向量化、入库。
3. 环境配置
3.1 .env 文件新增参数
在项目根目录的.env文件中，新增以下配置：
# 文本切割长度配置 CHUNK_SIZE=500 CHUNK_OVERLAP=50 # 本地模型驱动（cpu:cpu，gpu:cuda，互联网:remote） DEVICE="cpu" # 向量模型地址（可以从前面第4篇提到的国内模型官网上面下载一个向量模型放到指定的文件夹下） EMBEDDING_MODEL_PATH="D:/AI_Model/modelscope/AI-ModelScope/bge-base-zh-v1.5" # 向量数据库配置 MILVUS_URL="http://192.168.4.77:19530" MILVUS_USER="root" MILVUS_PASSWORD="******" # 向量库集合索引 MILVUS_COLLECTION_WEBSITE="ai_demo_website" MILVUS_COLLECTION_DOC="ai_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_splitter=RecursiveCharacterTextSplitter(chunk_size=setting.chunk_size,chunk_overlap=setting.chunk_overlap)splits=text_splitter.split_text(doc_str)4.1 切割参数说明
- chunk_size（500）：每个文本块的最大字符数。过大会导致向量化精度下降，过小则会丢失上下文语义。
- chunk_overlap（50）：相邻文本块之间的重叠字符数。重叠部分可以保留跨块的上下文信息，避免语义断裂。
4.2 为什么使用 RecursiveCharacterTextSplitter
RecursiveCharacterTextSplitter是 LangChain 提供的递归式文本切割器，它会优先按照段落、句子、标点等自然边界进行切割，尽量保持语义完整性，而不是简单粗暴地按固定长度硬切。
5. 文档切割后入库（这个操作实际在上一篇的内容已有）
文本切割完成后，将每个文本块向量化并写入 Milvus：
# 确保文档集合存在且 schema 正确（doc_str 为文本字段、vector 为向量字段）milvus_service.ensure_doc_collection(setting.milvus_collection_doc)# 写入向量库milvus_service.batch_save_update(collection_name=setting.milvus_collection_doc,document_list=[{'id':str(uuid4()),'name':file.filename,'doc_str':split}forsplitinsplits],embedding_column_name='doc_str',save=True)5.1 入库逻辑说明
- ensure_doc_collection：确保目标集合存在，且 schema 正确（
doc_str为文本字段、vector为向量字段）。 - batch_save_update：批量写入向量库。每个文本块生成一个 UUID 作为主键，并携带文件名和文本内容。
- embedding_column_name=‘doc_str’：指定对
doc_str字段进行向量化。 - save=True：表示执行插入操作；若为
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_client=MilvusClient(uri=setting.milvus_url,user=setting.milvus_user,password=setting.milvus_password)logger.info('Milvus连接成功')defbatch_save_update(collection_name:str,document_list:List[Dict],embedding_column_name:str='text',save:bool=True):text_list=[doc[embedding_column_name]fordocindocument_list]embedding_list=embedding_service.embedding_text(text_list)foriinrange(len(text_list)):document_list[i]['vector']=embedding_list[i]ifsaveisTrue:milvus_client.insert(collection_name=collection_name,data=document_list)else:milvus_client.upsert(collection_name=collection_name,data=document_list)defdelete_by_filter(collection_name:str,ids:Union[list,str,int]=None,filter:str="")->Dict:returnmilvus_client.delete(collection_name=collection_name,filter=filter,ids=ids)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(f"Unsupported filter value type:{type(value)}")filter_parts=["("+part+")"forpartinfilter_parts]return" and ".join(filter_parts)defquery(collection_name:str,filter:str=None,limit=20,output_fields:List[str]=["text"])->List[Dict]:returnmilvus_client.query(collection_name=collection_name,filter=filter,limit=limit,output_fields=output_fields)defsearch(collection_name:str,embedding:List[float],limit:int,filter:str=None,output_fields:List[str]=["vector"])->List[Dict]:res=milvus_client.search(collection_name=collection_name,data=[embedding],limit=limit,filter=filter,output_fields=output_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:str="id",id_type:str="string",vector_field_name:str="vector",properties:Dict=None):""" 创建集合 :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 智能体真正"读懂"文档内容。
