从 PDF 到向量检索:一个最简单的本地 RAG 入库实例

最近看了不少企业级 RAG 平台的源码,发现一个通病:一上来就是 Milvus + Neo4j + MinIO 好几个中间件一起怼,想学习"文档怎么变成向量"这个核心链路,反而被一堆基础设施绕晕了。

这篇文章把这条链路拆到最简:解析 → 分块 → 向量化 → 入库,四步走完,全程本地跑,不需要任何服务器、不需要 Docker,一个 Python 脚本搞定。

技术选型

四个环节全是"进程内"跑的库,没有一个是要单独部署的服务。

环境准备

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 save_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()# 插入完成后构建索引,加速检索save_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]save_to_zvec(chunks, embeddings)print(f"处理完成:{pdf_path} -> {len(chunks)} 个 chunk 已入库")build_kb_from_pdf("sample.pdf")

小结

这套流程的核心思路就四步:

1. 解析:把 PDF 转成保留结构的 Markdown,而不是拍扁的纯文本
2. 分块:按标题+段落切,避免语义被硬切断
3. 向量化:用中文效果好的开源 embedding 模型
4. 入库:选一个不需要起服务的嵌入式向量库

整个脚本没有 Docker、没有独立服务、没有云依赖,python build_kb.py 直接跑完,数据全部落在本地目录里。如果以后需求变复杂了(海量文档、多租户、知识图谱),再往上叠 MinerU、Milvus 这些重组件也不迟——但对于个人项目或者快速验证一个想法,这四步已经够用。