最近深入研究了多个企业级RAG平台的源码,发现一个普遍问题:很多项目一开始就集成Milvus、Neo4j、MinIO等中间件,导致开发者想理解“如何将PDF文档转化为向量并实现检索”这一核心流程时,先被复杂的基础设施配置搞得晕头转向。

因此,本文彻底简化这条链路,仅保留四个关键步骤:解析 → 分块 → 向量化 → 入库。全程在本地运行,无需任何服务器、Docker容器,只需一个Python脚本即可完成。
技术选型
| 环节 | 用什么 | 为什么选它 |
|---|---|---|
| 解析 | pymupdf4llm | 轻量级纯Python库,几秒安装完成,能直接将PDF转换为保留标题结构的Markdown格式 |
| 分块 | 按标题/段落切分 | 无需额外依赖,仅需几行代码实现 |
| 向量化 | sentence-transformers + text2vec-base-chinese | 中文处理效果优异,且模型体积较小 |
| 入库 | Zvec | 阿里开源的嵌入式向量库,pip install 即可使用,无需启动服务 |
四个环节全部采用“进程内”运行的库,无需单独部署任何外部服务。
环境准备
pip install pymupdf4llm sentence-transformers zvec --break-system-packages
Step 1: 解析PDF —— 转换为Markdown格式
import pymupdf4llmdef parse_pdf(pdf_path: str) -> str:"""把 PDF 转成保留标题结构的 Markdown 文本"""md_text = pymupdf4llm.to_markdown(pdf_path)return md_textmarkdown_content = parse_pdf("sample.pdf")print(markdown_content[:500])
为什么不直接使用 PyPDF2 这类库进行纯文本提取?因为那样会导致标题、段落、表格等信息全部丢失,变成一长串无结构的纯文本。后续进行语义分块时无法按结构划分,容易将完整语句从中间截断。而pymupdf4llm 能够保留 Markdown 的标题标记(如#、##),基于标题层级进行分块,从而保证语义完整性。
Step 2: 分块 —— 按段落/标题切分
from typing import Listimport redef split_into_chunks(markdown_text: str, max_chars: int = 500) -> List[str]:"""按标题和段落切分,超长段落再按字数二次切分"""# 先按标题切成大块sections = re.split(r'n(?=#{1,3}s)', markdown_text)chunks = []for section in sections:section = section.strip()if not section:continue# 段落内部再按空行细分,避免单块过长paragraphs = [p.strip() for p in section.split("nn") if p.strip()]buffer = ""for p in paragraphs:if len(buffer) + len(p) > max_chars and buffer:chunks.append(buffer)buffer = pelse:buffer = f"{buffer}nn{p}" if buffer else pif buffer:chunks.append(buffer)return chunkschunks = split_into_chunks(markdown_content)print(f"共切分为 {len(chunks)} 个 chunk")for i, chunk in enumerate(chunks[:3]):print(f"[{i}] {chunk[:80]}...n")
Step 3: 向量化 —— 文本转Embedding
from sentence_transformers import SentenceTransformerembedding_model = SentenceTransformer("shibing624/text2vec-base-chinese")def embed_chunk(text: str) -> List[float]:embedding = embedding_model.encode(text, normalize_embeddings=True)return embedding.tolist()embeddings = [embed_chunk(chunk) for chunk in chunks]embedding_dim = len(embeddings[0])print(f"向量维度: {embedding_dim}")
Step 4: 入库 —— 写入Zvec
import zvec# 定义 schema:一个存原文的字段 + 一个存向量的字段collection_schema = zvec.CollectionSchema(name="pdf_kb",fields=[zvec.FieldSchema(name="text", data_type=zvec.DataType.STRING),],vectors=[zvec.VectorSchema(name="embedding",data_type=zvec.DataType.VECTOR_FP32,dimension=embedding_dim,index_param=zvec.HnswIndexParam(metric_type=zvec.MetricType.COSINE),),],)# 本地建库,数据落在 ./pdf_kb_data 目录下collection = zvec.create_and_open(path="./pdf_kb_data", schema=collection_schema)def sa ve_to_zvec(chunks: List[str], embeddings: List[List[float]]) -> None:for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):collection.insert(zvec.Doc(id=str(i),vectors={"embedding": embedding},fields={"text": chunk},))collection.optimize()# 插入完成后构建索引,加速检索sa ve_to_zvec(chunks, embeddings)print("入库完成")
验证一下:检索测试
def retrieve(query: str, top_k: int = 3) -> List[str]:query_embedding = embed_chunk(query)result = collection.query(queries=zvec.Query(field_name="embedding", vector=query_embedding),topk=top_k,)return [doc.fields["text"] for doc in result]results = retrieve("文档里提到的核心观点是什么?")for i, chunk in enumerate(results):print(f"[{i}] {chunk}n")
完整流程串起来
def build_kb_from_pdf(pdf_path: str):markdown_content = parse_pdf(pdf_path)chunks = split_into_chunks(markdown_content)embeddings = [embed_chunk(c) for c in chunks]sa ve_to_zvec(chunks, embeddings)print(f"处理完成:{pdf_path} -> {len(chunks)} 个 chunk 已入库")build_kb_from_pdf("sample.pdf")
小结
本流程的核心思路可以概括为以下四个步骤:
- 解析:将PDF转换为保留层级结构的Markdown格式,避免结构扁平化
- 分块:依据标题和段落进行切分,防止语义被强行割裂
- 向量化:采用中文表现优异的开源嵌入模型进行向量转换
- 入库:选用无需启动服务的嵌入式向量数据库完成存储
整个脚本不依赖Docker、独立服务或云资源,只需执行python build_kb.py即可完成从PDF到向量库的整个流程,所有数据直接保存在本地目录。未来若遇到海量文档、多租户或知识图谱等复杂场景,再考虑引入MinerU、Milvus等重型组件也为时不晚——但对于个人项目或快速原型验证,这套四步流程已经足够实用。
