🎯 课程主题
项目第三~第五阶段：嵌入模型初始化 → 文档加载与自定义分隔符的递归切分 → 向量化 → 构建数据格式并 Upsert 写入 Milvus，完成知识库数据入库的完整流程。
📝 核心知识点
嵌入模型初始化：使用
OpenAIEmbeddings(model=EMBED_MODEL_NAME)创建嵌入模型实例，后续统一调用其embed_documents()批量向量化文本，embed_query()向量化单条查询。文档加载：使用 LangChain 的
TextLoader加载本地知识库文件，需指定file_path和encoding="utf-8"。可通过 IDE 的"另存为"功能查看文件的实际编码格式。文档切分（RecursiveCharacterTextSplitter）：
chunk_size=200：每个切片的最大字符数chunk_overlap=80：相邻切片之间的重叠字符数separators：自定义分隔符优先级列表，递归地按顺序尝试切分- 例：
["## ", "\n\n", "\n", "。", " ", ""]，先按##（二级标题标记）切，再按段落、换行、句号、空格、最终按字符切
- 例：
- 使用
splitter.split_documents(documents)对Document列表切分（不是split_text）
向量化：由于
embed_documents()接受List[str]，需从chunks中提取doc.page_content组装为字符串列表，再传入生成List[List[float]]。构建 data 并 Upsert：
- 每条记录包含
id（主键，从 0 开始）、vector（对应索引的向量）、text（切片内容page_content）、source（原始文件名）、chunk_id（切片序号） - 调用
client.upsert(collection_name, data)写入 - 调用
client.flush(collection_name)落盘
- 每条记录包含
Upsert 幂等性原理：
id为主键不可重复。再次运行 Upsert 时，相同id的记录会被覆盖而非新增，因此数据总量不变。upsert_count为本次操作涉及条数，row_count（通过get_collection_statistics查看）才反映真实记录数。
🏗️ 架构与工作流
[阶段3] 嵌入模型初始化: OpenAIEmbeddings(model=...)
↓
[阶段4] 读取文档并切分:
TextLoader(file_path, encoding) → loader.load() → List[Document] →
RecursiveCharacterTextSplitter(chunk_size, chunk_overlap, separators) →
splitter.split_documents(documents) → List[Document] (chunks)
↓
[阶段5] 向量化并写入 Milvus:
[chunk.page_content for chunk in chunks] →
embed_model.embed_documents(texts) → vectors →
构建 data [{id, vector, text, source, chunk_id}] →
client.upsert(collection_name, data) → client.flush() →
client.get_collection_statistics() 验证
💻 代码实战
# ============ 阶段3: 初始化嵌入模型 ============
from langchain_openai import OpenAIEmbeddings
embed_model = OpenAIEmbeddings(model=EMBED_MODEL_NAME) # 维度默认1024
# ============ 阶段4: 读取文档并切分 ============
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 4.1 加载文档
loader = TextLoader(file_path=DATA_FILE, encoding="utf-8")
documents = loader.load() # List[Document]
# 4.2 配置切分器（自定义分隔符策略）
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200,
chunk_overlap=80,
separators=["## ", "\n\n", "\n", "。", " ", ""]
)
# 4.3 执行切分
chunks = text_splitter.split_documents(documents)
print(f"文档共切分为 {len(chunks)} 个 chunk")
# 查看切片内容示例
for i, chunk in enumerate(chunks):
print(f"chunk{i}: {chunk.page_content}")
# 实际调试时可限制打印前几个
# ============ 阶段5: 生成向量并写入 Milvus ============
# 5.1 提取文本列表 → 向量化
texts = [chunk.page_content for chunk in chunks]
vectors = embed_model.embed_documents(texts) # List[List[float]]，与 chunks 等长
# 5.2 构建待插入数据
data = []
for i in range(len(chunks)):
data.append({
"id": i,
"vector": vectors[i],
"text": chunks[i].page_content,
"source": DATA_FILE,
"chunk_id": i
})
# 5.3 Upsert 写入
insert_result = client.upsert(
collection_name=COLLECTION_NAME,
data=data
)
print("insert result:", insert_result)
# 5.4 刷新落盘
client.flush(collection_name=COLLECTION_NAME)
# 5.5 验证数据量
stats = client.get_collection_statistics(collection_name=COLLECTION_NAME)
print("collection statistics:", stats)
# 5.6 额外验证：通过 query 查询实际记录数
results = client.query(
collection_name=COLLECTION_NAME,
filter="id >= 0",
output_fields=["id", "chunk_id"]
)
print(f"实际记录数: {len(results)}")
⚠️ 常见问题与避坑指南
split_textvssplit_documents：输入是List[Document]时必须用split_documents()，输入是List[str]时才用split_text()。用错会导致类型错误。embed_documents的入参：它接受List[str]，不能直接传List[Document]。需先用列表推导式提取page_content。row_countvsupsert_count：Upsert 返回的upsert_count不等于 Collection 中的实际记录数。重复运行时，Upsert 覆盖已有主键记录而非新增，upsert_count只反映"本次涉及条数"，真实记录数需通过get_collection_statistics或query无过滤条件查询获得。编码问题：加载文本文件时务必指定正确的
encoding，否则中文内容可能乱码。不确定编码时，可用 IDE 的"另存为"功能查看或使用chardet库探测。自定义分隔符顺序：分隔符列表的顺序决定切分优先级，粗粒度在前（如
##标题标记），细粒度在后（如空格、字符）。中文场景建议加入中文标点（。、，、！等）作为分隔符。
💡 个人总结与延伸
本节将 RAG 的前半段链路（加载 → 切分 → 向量化 → 存储）串联为完整可运行的代码。自定义分隔符是提升检索质量的关键技巧——根据文档结构（标题层级、段落、标点特征）设计合理的分隔符优先级，能生成语义更完整的 chunk，直接影响后续检索的命中精度。
chunk_size=200 和 chunk_overlap=80 是示例配置，实际项目中需根据知识库内容特征调优：代码文档可能适合更大的 size，问答类知识可能适合更小的粒度。overlap 越大，chunk 间信息冗余越高但语义断裂越少，需在召回率和存储成本间权衡。