1. 环境准备搭建你的本地AI工作台想自己动手搞一个能“记住”大量文档并且能像人一样回答问题的AI助手吗听起来很酷但你是不是觉得第一步“环境配置”就头大别担心我当年也是这么过来的。今天我就带你用最接地气的方式在你自己电脑上从零开始搭一个完全本地的知识库系统。我们不用依赖任何复杂的云服务所有东西都跑在你自己的机器上数据也完全私有安全又可控。这个系统的核心是三个“明星组件”LangChain、Milvus Lite和GTE-Large-Zh。LangChain就像是一个聪明的“导演”负责指挥整个流程比如怎么切分你的文档怎么调用模型。Milvus Lite是一个轻量级的“记忆库”专门用来存储和快速查找文本转换成的向量你可以把向量理解成文本的“数学指纹”。而GTE-Large-Zh则是一位强大的“翻译官”负责把中文文本精准地转换成这些向量。把它们组合起来你就能得到一个专业的RAG检索增强生成原型这绝对是当前AI应用开发的热门方向。那么开始之前我们需要准备点啥首先你的电脑操作系统最好是Ubuntu或者macOS这两个系统对我们要用的工具支持最好。Windows用户也别急通过WSL2安装Ubuntu子系统体验几乎一样。Python版本我强烈建议用3.10或以上很多新库的兼容性更好。硬件方面有个独立GPU当然是最爽的处理模型推理会快很多。但如果没有用CPU也能跑我实测过就是需要多点耐心处理一篇长文档可能要多等一会儿。最后确保网络通畅因为我们需要从网上下载预训练好的模型文件。接下来我们创建一个干净的Python虚拟环境这是避免各种包版本冲突的黄金法则。打开你的终端跟着我一步步来# 创建并激活虚拟环境 python -m venv venv # 在 macOS/Linux 上 source venv/bin/activate # 在 Windows 上如果你用cmd # venv\Scripts\activate激活后你的命令行前面通常会显示(venv)这就说明你已经在这个独立的环境里了。然后我们安装核心的“三件套”以及它们的帮手# 先升级pip到最新 pip install -U pip # 安装核心框架和数据库客户端 pip install langchain pymilvus2.4.2 # 安装模型加载和计算库 pip install transformers torch sentencepiece # 安装LangChain社区对Milvus的集成支持不同版本包名可能不同多装几个确保兼容 pip install langchain-milvus langchain-community这里有个关键点pymilvus这个包从2.4.2版本开始就内置了Milvus Lite。这意味着你不需要单独安装或启动一个数据库服务。当你代码里指定一个本地文件路径比如./my_knowledge.db作为连接地址时它会自动在背后以轻量级模式运行所有数据都保存在这个文件里用完即关超级方便。这完美契合了我们“本地、轻量、开箱即用”的目标。2. 核心组件深度解析与选型思考环境搭好了我们得搞清楚手里的“兵器”到底强在哪为什么要选它们。这就像组装电脑你得知道CPU、显卡各自负责什么才能搭配出最佳性能。首先说说LangChain。你可以把它想象成一个功能强大的“乐高底座”或者“工作流编排器”。它本身不直接提供向量模型或者数据库但它定义了一套标准的接口和组件比如Document文档、TextSplitter文本分割器、Embeddings嵌入模型接口、VectorStore向量存储接口。我们的任务就是用代码把这些“乐高积木”按照正确的顺序拼装起来。它的最大价值在于“标准化”和“可替换性”。今天我们用Milvus Lite明天如果觉得性能不够想换成Chroma或者Qdrant只需要换掉VectorStore这块积木其他流程代码几乎不用动。这对于快速实验和迭代至关重要。然后是Milvus Lite它是明星向量数据库Milvus的“单机轻量版”。它的核心任务就两个存向量、查向量。为什么不用普通的SQL数据库存因为向量之间的“相似度”计算比如余弦相似度在传统数据库里效率极低。Milvus专门为此优化使用了像FLAT、IVF_FLAT、HNSW这样的索引算法能在百万甚至千万级别的向量中实现毫秒级的相似检索。Lite版本去掉了分布式、多副本这些企业级特性只保留了最核心的向量检索能力并以一个本地文件的形式存在特别适合我们这种个人开发、原型验证的场景。它默认使用的FLAT索引是最精确的暴力计算数据量小的时候没问题如果后面你的知识库文档暴涨到几十万条可以考虑升级到完整版Milvus并换用更快的近似索引。最后是重头戏嵌入模型GTE-Large-Zh。这是整个系统的“大脑”决定了你的AI助手理解中文的深度。简单来说它的工作就是把一段文字比如“北京是中国的首都”转换成一个由1024个数字组成的列表即1024维向量。这个转换过程非常精妙语义相近的句子它们的向量在数学空间里的距离也会很近。比如“北京是中国的首都”和“中国的首都是北京”这两个向量就会非常接近。市面上中文模型很多为什么选它第一它是专门为中文优化的在中文语义理解任务上表现名列前茅。第二它完全开源免费我们可以随意下载到本地使用。第三它的输出向量维度是1024这是一个比较均衡的维度既能保留丰富的语义信息又不会让后续的存储和计算负担过重。你需要记住它的一个关键限制最大输入长度是512个token。对于中文粗略估算1个token约等于1.5到2个汉字。所以我们后续切分文本时必须保证每一段文本的长度远低于这个限制否则超出的部分会被直接截断丢失信息。3. 实战第一步设计文本处理流水线现在我们正式进入编码实战环节。假设你手头有一份庞大的产品手册PDF或者一堆技术博客的Markdown文件你想让AI读懂它们。第一步不是急着喂给模型而是要做好“预处理”也就是把大块文本切成适合消化的小块。这一步至关重要切得好检索质量高切得不好AI的回答就会支离破碎。为什么一定要切分直接扔给模型不行吗不行。主要有两个原因一是刚才提到的模型有512 token的长度限制。二是从检索效果看过长的文本包含的信息太杂检索出来的段落可能只有一小部分相关会引入很多噪声。我们的策略是先粗后细重叠切割。我建议你先用空行、章节标题等明显分隔符把文档分成较大的“节”Section。然后对每一节再用更细的粒度进行滑动窗口式的切割。这里我强烈推荐使用LangChain提供的RecursiveCharacterTextSplitter它比简单的按字符分割更智能。它会优先尝试按段落\n\n分割如果段落还是太长就按句子.、!、?分再不行才按字符分。这样可以最大程度保持语义的完整性。同时一定要设置chunk_overlap重叠量。我一般设置50-100个字符的重叠。这就像你看书时翻页会回顾前一页的最后几行保证了上下文衔接避免一个完整的句子或概念被硬生生割裂在两个不同的块里导致检索时丢失关键信息。让我们看看具体的代码实现。下面的prepare_documents函数演示了一个完整的处理流程从读取多种格式的原始文件到智能分割最终生成LangChain标准的Document对象列表。每个Document包含文本内容page_content和元数据metadata比如来源文件名、页码、章节等这些元数据在后续检索和展示时非常有用。from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader from langchain_core.documents import Document import os def prepare_documents(file_path: str, chunk_size: int 400, chunk_overlap: int 80) - List[Document]: 读取文件并根据扩展名选择加载器然后进行智能文本分割。 # 根据文件类型选择加载器 ext os.path.splitext(file_path)[1].lower() if ext .pdf: loader PyPDFLoader(file_path) elif ext .md or ext .markdown: loader UnstructuredMarkdownLoader(file_path) else: # 默认识别为文本文件 loader TextLoader(file_path, encodingutf-8) # 加载原始文档 raw_docs loader.load() print(f从 {file_path} 加载了 {len(raw_docs)} 个原始页面/段落。) # 初始化智能文本分割器 # separators 参数定义了分割的优先级顺序 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) # 对每个原始文档进行分割 all_chunks [] for doc in raw_docs: chunks text_splitter.split_documents([doc]) all_chunks.extend(chunks) print(f分割后共得到 {len(all_chunks)} 个文本块。) return all_chunks # 使用示例 documents prepare_documents(./产品手册.pdf) for i, doc in enumerate(documents[:2]): # 打印前两个块看看效果 print(f--- 块 {i1} ---) print(f元数据: {doc.metadata}) print(f内容预览: {doc.page_content[:200]}...\n)运行这段代码你可以直观地看到一本厚厚的PDF是如何被转化成一个个规整的、带有上下文重叠的文本块的。这个过程是构建高质量知识库的基石值得你多花点时间根据自己文档的特点调整chunk_size和chunk_overlap参数。4. 让机器理解文本封装GTE-Large-Zh嵌入模型文本切好了接下来就要请出我们的“翻译官”GTE-Large-Zh模型把文字变成向量。我们需要自己写一个包装类让它符合LangChain的Embeddings接口标准。这样LangChain的“导演”才能指挥它工作。这个封装类的核心逻辑其实不复杂主要做三件事1. 加载模型和分词器2. 实现批量文本编码3. 将编码结果归一化。归一化是为了将所有向量缩放到单位长度这样后续计算余弦相似度会更高效和稳定。这里我采用模型官方推荐的做法使用模型输出的第一个token即[CLS]token的隐藏状态作为整个句子的向量表示然后进行L2归一化。我强烈建议你在这里加入批量处理的逻辑。如果你有上千个文本块一个一个地送给模型推理那速度简直是一场噩梦。通过批量送入可以极大利用GPU或CPU的并行计算能力。下面是我优化后的GTEEmbeddings类它支持自定义批次大小并且能自动判断使用GPU还是CPU。from typing import List, Optional import torch import torch.nn.functional as F from transformers import AutoTokenizer, AutoModel from langchain.embeddings.base import Embeddings class GTEEmbeddings(Embeddings): def __init__(self, model_name: str thenlper/gte-large-zh, device: Optional[str] None, batch_size: int 32, # 增大批次以提升效率 normalize_embeddings: bool True): self.model_name model_name # 自动选择设备优先GPU self.device device or (cuda if torch.cuda.is_available() else cpu) self.batch_size batch_size self.normalize normalize_embeddings print(f正在加载模型 {model_name} 到设备: {self.device}...) # 加载分词器和模型 self.tokenizer AutoTokenizer.from_pretrained(model_name) self.model AutoModel.from_pretrained(model_name).to(self.device) self.model.eval() # 设置为评估模式关闭dropout等训练层 def _embed_batch(self, texts: List[str]) - List[List[float]]: 核心的批量编码方法 all_embeddings [] with torch.no_grad(): # 关闭梯度计算节省内存 for i in range(0, len(texts), self.batch_size): batch_texts texts[i:i self.batch_size] # 对批次进行编码和填充 encoded_input self.tokenizer( batch_texts, paddingTrue, truncationTrue, max_length512, # 严格遵守模型限制 return_tensorspt # 返回PyTorch张量 ).to(self.device) model_output self.model(**encoded_input) # 取每个序列的 [CLS] token 对应的向量 batch_embeddings model_output.last_hidden_state[:, 0, :] if self.normalize: # L2 归一化 batch_embeddings F.normalize(batch_embeddings, p2, dim1) # 转移到CPU并转为Python列表 all_embeddings.extend(batch_embeddings.cpu().tolist()) return all_embeddings def embed_documents(self, texts: List[str]) - List[List[float]]: LangChain接口为文档列表生成嵌入 if not texts: return [] return self._embed_batch(texts) def embed_query(self, text: str) - List[float]: LangChain接口为单个查询生成嵌入 return self._embed_batch([text])[0] # 测试一下封装好的类 if __name__ __main__: embedder GTEEmbeddings() test_texts [机器学习是人工智能的一个分支。, 深度学习利用神经网络进行学习。] vectors embedder.embed_documents(test_texts) print(f生成的向量维度: {len(vectors[0])}) print(f第一个向量的前10个值: {vectors[0][:10]})第一次运行这段代码时它会从Hugging Face下载大约1.4GB的模型文件请保持网络畅通。下载完成后模型会缓存在本地下次就快了。你可以用几段测试文本看看输出向量的样子感受一下文字是如何变成一串神秘数字的。5. 构建本地记忆库集成Milvus Lite文本变成了向量我们需要一个地方把它们存起来并且要能快速查找。这就是Milvus Lite的舞台了。我们将通过LangChain的集成接口把前面准备好的文档块和生成的向量稳稳地存入本地的Milvus数据库文件中。这个过程的核心是Milvus.from_documents方法。你只需要把文档列表和我们刚写好的嵌入模型对象传给它它就会在背后自动完成一系列复杂操作计算所有文档的向量、在本地创建或连接一个Milvus Lite数据库文件、建立集合Collection相当于数据库的表、定义好向量字段的维度1024维、插入数据并创建索引。这一切都被封装成了一行简单的调用这就是使用框架的巨大便利。不过为了更精细的控制和更好的错误处理我建议你了解一些关键参数并做适当的封装。下面的create_vector_store函数展示了更健壮的做法from langchain_milvus import Milvus # 根据你的安装导入可能不同 import os def create_vector_store(documents, embeddings, persist_dir./milvus_data): 创建或连接Milvus Lite向量存储并插入文档。 # 确保持久化目录存在 os.makedirs(persist_dir, exist_okTrue) # 定义本地数据库文件路径 uri ffile://{os.path.abspath(persist_dir)}/knowledge_base.db connection_args { uri: uri, # 可以添加其他连接参数但Lite模式下大部分用默认即可 } print(f正在连接/创建Milvus Lite数据库位置: {uri}) try: # 关键的一步将文档和向量存入数据库 vector_store Milvus.from_documents( documentsdocuments, embeddingembeddings, # 传入我们自定义的GTEEmbeddings实例 collection_namemy_knowledge_collection, # 集合名称可自定义 connection_argsconnection_args, drop_oldTrue, # 如果同名集合已存在则删除重建。初次运行后可以改为False # 可选指定向量字段的索引参数Lite版仅支持FLAT索引 index_params{ metric_type: L2, # 距离度量方式L2或IP内积。GTE模型归一化后用IP更快。 index_type: FLAT, # Lite版固定为FLAT } ) print(向量存储创建成功并已完成数据插入) return vector_store except Exception as e: print(f创建向量存储时发生错误: {e}) # 可以在这里添加更详细的错误处理比如检查pymilvus版本等 raise # 整合前几步完成入库流水线 print( 开始知识库构建流水线 ) # 1. 准备文档 docs prepare_documents(./你的文档.txt) # 2. 初始化嵌入模型 embed_model GTEEmbeddings() # 3. 创建向量存储并入库 vector_db create_vector_store(docs, embed_model) print( 所有数据已成功存入本地知识库 )执行完这段代码后你会在指定的milvus_data目录下看到生成的数据文件。这个文件就是你的知识库本体你可以把它复制到任何其他电脑上使用。drop_oldTrue参数在第一次运行时是安全的它会清空旧数据。当你后续想往已有的知识库里添加新文档时一定要把这个参数改为False否则之前的数据就没了。这就是增量更新的方法。6. 让知识库“活”起来执行语义检索与问答数据入库只是完成了“记忆”的过程如何“回忆”才是体现价值的关键。现在我们的知识库已经准备就绪可以接受你的提问了。检索的核心是similarity_search方法你输入一个问题它会将问题转换成向量然后在库中寻找最相似的几个文本块向量距离最近。单纯的检索返回原始文本块已经很有用了。但我们可以更进一步结合一个大语言模型LLM比如ChatGLM、Qwen或者通过API调用GPT构建一个完整的问答系统。这就是RAG的完整形态检索Retrieval 增强Augmentation 生成Generation。系统先检索出相关的知识片段然后把它们和问题一起拼成“提示词”Prompt送给LLM让它生成一个准确、连贯且基于你知识库的答案。下面这个ask_question函数演示了从检索到生成答案的完整流程。我这里以调用本地部署的ChatGLM3为例你也可以替换成任何兼容OpenAI API格式的模型服务。from typing import List # 假设你有一个本地LLM服务这里以OpenAI API格式为例 from openai import OpenAI def ask_question(query: str, vector_store, llm_client, k: int 4): 1. 从向量库检索相关文档。 2. 将检索结果和问题组合成Prompt。 3. 调用LLM生成答案。 print(f用户提问: {query}) print(正在语义检索相关文档...) # 第一步相似性检索 relevant_docs: List[Document] vector_store.similarity_search(query, kk) print(f检索到 {len(relevant_docs)} 条相关段落。) # 构建上下文 context \n\n---\n\n.join([doc.page_content for doc in relevant_docs]) # 第二步构建Prompt模板 prompt_template f请基于以下提供的上下文信息回答用户的问题。如果上下文信息不足以回答问题请直接说“根据已知信息无法回答该问题”不要编造信息。 上下文信息 {context} 用户问题{query} 请给出准确、详细的回答 print(正在调用大语言模型生成答案...) # 第三步调用LLM try: # 这里需要根据你实际使用的LLM API进行调整 response llm_client.chat.completions.create( modelchatglm3-6b, # 或你的模型名称 messages[ {role: system, content: 你是一个专业的助手严格根据提供的上下文回答问题。}, {role: user, content: prompt_template} ], temperature0.1, # 低温度使输出更确定更依赖上下文 max_tokens500 ) answer response.choices[0].message.content except Exception as e: print(f调用LLM时出错: {e}) answer 抱歉生成答案时出现错误。 # 第四步输出结果 print(\n *50) print(【最终答案】) print(answer) print(*50) print(\n【参考来源】) for i, doc in enumerate(relevant_docs): print(f来源 {i1} (来自 {doc.metadata.get(source, 未知)}): {doc.page_content[:150]}...) return answer # 使用示例 if __name__ __main__: # 初始化LLM客户端示例为本地OpenAI兼容API client OpenAI(base_urlhttp://localhost:8000/v1, api_keynot-needed) # 加载已有的向量存储假设之前已经创建好 embed_model GTEEmbeddings() existing_vector_store Milvus( embedding_functionembed_model, collection_namemy_knowledge_collection, connection_args{uri: file://./milvus_data/knowledge_base.db} ) # 开始问答 while True: user_q input(\n请输入你的问题 (输入quit退出): ) if user_q.lower() quit: break ask_question(user_q, existing_vector_store, client)运行这个交互式问答脚本你就可以像使用ChatGPT一样向你的专属知识库提问了。你会发现它的回答严格基于你喂给它的文档内容不会胡编乱造。这对于构建企业知识库、技术文档助手、个人学习笔记系统等场景实用性极强。7. 避坑指南与性能优化实战走通了全流程你可能已经成功运行起来了。但想让它跑得更快、更稳、处理更大数据量还需要注意一些细节。这些都是我趟过坑后总结的经验。第一个坑模型下载慢或失败。Hugging Face模型服务器在国外国内下载大文件可能不稳定。解决办法有两个一是使用国内镜像站比如在代码中设置环境变量HF_ENDPOINThttps://hf-mirror.com。二是提前用git lfs下载模型文件到本地然后在代码中指定本地路径model_name/path/to/your/local/gte-large-zh。第二个坑内存溢出OOM。处理超长文档或批量很大时尤其是用GPU容易内存不足。对策是1. 减小batch_size比如从32降到16或8。2. 在_embed_batch函数中及时将计算完的向量从GPU转移到CPU.cpu()并释放缓存torch.cuda.empty_cache()。3. 对于超长文档一定要确保分割后的块远小于512 token避免单个批次因填充padding导致内存暴涨。第三个坑Milvus Lite连接或插入错误。最常见的是端口冲突如果以前运行过其他Milvus服务或文件权限问题。确保你用于存储数据库文件的目录有读写权限。如果遇到端口错误可以尝试在连接参数中指定不同的临时端口虽然Lite版通常自动管理。另一个常见错误是集合已存在但结构Schema不匹配比如向量维度不对。这时需要先手动删除旧的集合文件或者确保drop_oldTrue。性能优化方面当你的知识库文档超过一万条时就需要考虑以下策略了索引策略Milvus Lite只支持FLAT索引这是精确检索但速度随数据量线性下降。当数据量变大例如超过10万你必须升级到Milvus完整版并使用IVF_FLAT、HNSW等近似索引能在精度损失极小的情况下换来几十上百倍的检索速度提升。批量插入优化from_documents内部是分批的但如果你的文档量极大百万级可以考虑直接用pymilvus的bulk_writer接口或者先将所有向量计算好再一次性批量插入减少事务开销。嵌入缓存对于不变的历史文档计算好的向量可以保存到本地文件如NumPy数组.npy或简单的键值数据库如Redis。下次构建时先计算文本的哈希值如MD5如果已有缓存直接加载向量跳过模型推理能节省大量时间。检索后处理Rerank有时相似性检索返回的前k个结果可能包含一些相关性不高但某些关键词匹配的片段。可以在LLM生成前加入一个“重排序”步骤使用一个更小更快的交叉编码器Cross-Encoder模型对检索结果进行精排只把最相关的几个片段送给LLM这能显著提升最终答案的质量。构建本地知识库不是一个一蹴而就的项目而是一个可以持续迭代的系统。你可以从处理单个文档开始逐步扩展到扫描整个文件夹下的所有PDF、Word、Markdown文件。你可以尝试不同的文本分割策略观察对问答质量的影响。你还可以将这套系统封装成一个简单的Web应用用Gradio或Streamlit分享给同事或朋友使用。最重要的是你拥有了一个完全受自己控制、数据私有的智能知识库这为后续探索更复杂的AI应用打下了坚实的基础。