不用LangChain的本地RAG实战：Ollama+Chroma+Python轻量级实现

1. 为什么非要绕开 LangChain一个本地 RAG 的“裸机”执念我第一次在公司内部技术分享会上讲完“不用 LangChain 写 Ollama RAG”这个选题台下有位刚转行半年的同事举手问“老师LangChain 不就是为这种事生的吗自己造轮子是不是有点……费力不讨好”我当时没急着回答而是反问“你上一次看 LangChain 的源码是在调load_qa_chain还是ConversationalRetrievalChain的时候有没有发现光是搞懂它怎么把Document塞进PromptTemplate再喂给LLMChain就已经要翻三遍文档、查五次 GitHub issue”这不是抬杠。这是过去两年我带过 7 个 RAG 实战项目后的真实体感LangChain 是一套功能完备的“RAG 操作系统”但当你只需要跑通一个本地知识库问答它就像用波音 787 去送外卖——引擎轰鸣油耗惊人而你真正需要的可能只是一辆改装过的电动自行车。关键词里反复出现的 “Ollama”、“本地”、“Python”、“RAG”已经勾勒出最典型的轻量级场景一台 32GB 内存的 MacBook Pro一个存了 200 页 PDF 的产品手册目录一个想随时问“客户投诉率最高的三个模块是什么”的产品经理。没有 Kubernetes没有 Redis 缓存集群没有跨模型路由调度——只有ollama run qwen2:7b启动时那一声清脆的终端提示音。所以“不用 LangChain”不是叛逆而是回归本质。RAG 的核心逻辑其实就三步切把你的 PDF/Markdown/Word 文档切成语义连贯的小段chunk嵌用一个嵌入模型embedding model把每一段变成一串数字向量vector搜答用户提问 → 把问题也变成向量 → 在向量库里找最相似的几段 → 把这几段和问题一起塞给大模型让它生成答案。LangChain 把这三步封装成DocumentLoaderTextSplitterEmbeddingsVectorStoreRetrievalQA的流水线。而我们要做的就是亲手拧紧每一颗螺丝——不是为了炫技而是为了在pip install langchain失败三次、ollama pull nomic-embed-text卡在 98%、或者某天发现ConversationalRetrievalChain默认把 chunk 长度设成 1000 字导致关键表格被硬生生劈成两半时能立刻定位到哪一行代码在作祟而不是在 12 万行源码里盲人摸象。更现实的驱动力来自部署端。上周我帮一家做工业设备维保的客户部署知识库他们产线边缘服务器只有 Ubuntu 20.04 Python 3.8pip install langchain直接报错pydantic2.0.0 is required而他们另一个系统又强依赖pydantic 2.6。最后我们删掉 LangChain用langchain-community里拆出来的HuggingFaceEmbeddings和Chroma的原生 API三天内上线。客户说“原来 RAG 不是黑盒子是能拧开盖子换电池的。”这就是本文的全部出发点给你一套可审计、可调试、可嵌入任意现有 Python 工程的 RAG 最小可行实现MVP所有依赖明明白白列在requirements.txt里所有向量计算逻辑写在embedder.py里所有检索策略藏在retriever.py的一个similarity_search_with_score方法里。它不追求“企业级”但保证“今天下午三点写完四点就能让老板对着自己写的 PDF 问问题”。2. 从零开始的四大支柱文件解析、分块、嵌入、检索RAG 的骨架必须结实否则再大的模型也撑不起一句靠谱的回答。我们不碰 LangChain 的UnstructuredPDFLoader或RecursiveCharacterTextSplitter而是用最基础、最可控的 Python 生态组合pypdf解析 PDF、markdown-it-py渲染 Markdown、jieba做中文分句、sentence-transformers提供嵌入能力、chromadb存储向量。这四个模块就是我们 RAG 的四大支柱每一个都值得掰开揉碎讲透。2.1 文件解析拒绝“黑盒加载”只信自己读出来的字节流LangChain 的DirectoryLoader看似方便但它内部会自动跳过.git、.DS_Store还会对.md文件做额外的 frontmatter 清洗。而我们的客户手册里恰恰有个README.md里用 YAML frontmatter 标注了每个章节的更新日期——这个信息正是后续做“按时间过滤检索结果”的关键依据。所以我们自己写file_loader.pyimport os import re from pathlib import Path from typing import List, Dict, Any def load_files_from_dir(directory: str, extensions: List[str] None) - List[Dict[str, Any]]: 手动遍历目录按扩展名加载文件内容保留原始路径与元数据 extensions: 如 [.pdf, .md, .txt]默认加载全部文本类文件 if extensions is None: extensions [.pdf, .md, .txt, .docx] documents [] for root, _, files in os.walk(directory): for file in files: file_path Path(root) / file # 跳过隐藏文件和常见二进制文件 if file.startswith(.) or file_path.suffix.lower() in [.png, .jpg, .exe]: continue # 只处理指定扩展名 if file_path.suffix.lower() not in extensions: continue try: # PDF 处理用 pypdf不依赖 OCR纯文本提取 if file_path.suffix.lower() .pdf: from pypdf import PdfReader reader PdfReader(file_path) content \n.join([page.extract_text() or for page in reader.pages]) # 提取页码信息用于后续溯源 metadata { source: str(file_path), file_type: pdf, page_count: len(reader.pages) } # Markdown 处理用 markdown-it-py 渲染避免正则误伤 elif file_path.suffix.lower() .md: import markdown_it md markdown_it.MarkdownIt() with open(file_path, r, encodingutf-8) as f: raw_content f.read() # 提取 frontmatterYAML 头部 fm_match re.match(r^---\s*\n(.*?)\n---\s*\n, raw_content, re.DOTALL) frontmatter {} if fm_match: import yaml try: frontmatter yaml.safe_load(fm_match.group(1)) or {} except yaml.YAMLError: pass content raw_content[fm_match.end():] else: content raw_content metadata { source: str(file_path), file_type: markdown, frontmatter: frontmatter } # 纯文本直接读取 else: with open(file_path, r, encodingutf-8) as f: content f.read() metadata {source: str(file_path), file_type: text} # 强制清理不可见字符Windows 换行符、零宽空格等 content re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f], , content) content re.sub(r\s, , content).strip() if content: # 过滤空内容 documents.append({ page_content: content, metadata: metadata }) except Exception as e: print(f⚠️ 加载失败 {file_path}: {str(e)}) continue return documents这段代码的价值远不止于“能读文件”。它让你彻底掌控三个关键点元数据完整性frontmatter里的last_updated: 2024-05-20会被完整保留后续可以作为filter参数传给 Chroma错误容忍度PDF 里某一页扫描件没文字extract_text()返回None我们用空字符串兜底不中断整个流程编码鲁棒性强制utf-8失败时捕获异常并跳过而不是让整个程序崩在UnicodeDecodeError上。提示很多团队踩的第一个坑就是用open(file).read()加载 PDF——它读出来的是二进制乱码。记住PDF 是结构化文档不是纯文本必须用pypdf或fitzPyMuPDF这类专用库。2.2 文本分块中文不是英文分句逻辑必须重写LangChain 默认的RecursiveCharacterTextSplitter以\n\n、\n、 空格为分隔符这对英文文档尚可但对中文简直是灾难。比如一段产品参数表| 模块 | 功耗 | 工作温度 | |------|------|----------| | 主控板 | 12W | -20℃ ~ 70℃ | | 传感器A | 3.5W | -40℃ ~ 85℃ |用\n切会把表头和第一行数据劈开用。切又会把“-20℃ ~ 70℃”里的。当成分句点。我们必须为中文定制分块逻辑。核心思路是先按语义单元粗分标题、段落、列表项再在单元内按句子精修。我们用jieba做分词辅助但主逻辑靠正则import re import jieba def chinese_text_splitter(text: str, chunk_size: int 300, chunk_overlap: int 50) - List[str]: 专为中文优化的分块器优先按标题、列表、段落分割再按句子微调 # 步骤1识别并保留标题#、##、### 开头或中文“第一章”、“1.”等 title_pattern r(#{1,6}\s.|\s*[第零一二三四五六七八九十][章篇]\s*.|\s*\d\.\s*.) sections re.split(title_pattern, text) chunks [] current_chunk for section in sections: if not section.strip(): continue # 如果是标题尝试单独成块如果太短或并入前一块 if re.match(title_pattern, section.strip()): if len(current_chunk) chunk_size * 0.7: chunks.append(current_chunk) current_chunk section.strip() else: current_chunk \n section.strip() continue # 步骤2对非标题内容按中文句号、问号、感叹号、分号、冒号分割 # 但排除小数点如 3.5W、省略号……、温度符号℃后的点 sentence_pattern r(?!\d)\.(?!\d)|(?!\.)\?(?!\.)|(?!\.)\!(?!\.)|| sentences re.split(sentence_pattern, section) for sent in sentences: sent sent.strip() if not sent: continue # 步骤3合并短句少于15字的句子大概率是不完整片段 if len(sent) 15 and current_chunk: current_chunk sent continue # 步骤4检查是否超过 chunk_size超了就切 if len(current_chunk) len(sent) chunk_size: if current_chunk: chunks.append(current_chunk) current_chunk sent else: current_chunk sent # 处理最后一块 if current_chunk: chunks.append(current_chunk) # 步骤5应用 overlap重叠 final_chunks [] for i, chunk in enumerate(chunks): if i 0: final_chunks.append(chunk) else: # 取前一块的后 overlap 个字符 当前块 prev_end chunks[i-1][-chunk_overlap:] if len(chunks[i-1]) chunk_overlap else chunks[i-1] merged prev_end chunk final_chunks.append(merged) return final_chunks # 实测效果对比 test_text 主控板功耗为12W。工作温度范围是-20℃ ~ 70℃。传感器A功耗3.5W。 print(chinese_text_splitter(test_text, chunk_size20)) # 输出[主控板功耗为12W。工作温度范围是-20℃ ~ 70℃。, 工作温度范围是-20℃ ~ 70℃。传感器A功耗3.5W。] # 看到了吗第二块开头重复了第一块结尾这就是 overlap 的价值——避免关键信息被切在边界上。这个分块器的精髓在于它把“中文句子”的定义权还给了业务它知道12W后面的.不是句号它知道℃后面的.也不是句号它知道第3.5节里的.是序号的一部分不是句子结束。而 LangChain 的分块器永远在猜。2.3 嵌入模型为什么选nomic-embed-text一场速度与精度的平衡术Ollama 社区最火的嵌入模型是nomic-embed-text但它在ollama list里显示为nomic-embed-text:latest很多人直接ollama run nomic-embed-text结果发现它根本不能像qwen2:7b那样交互式聊天——因为它是纯嵌入模型没有generate接口。我们必须理解嵌入模型Embedding Model和大语言模型LLM是两类完全不同的东西。前者是“翻译官”把文字翻译成向量后者是“作家”根据向量和指令生成文字。nomic-embed-text就是前者它的任务只有一个POST /api/embeddings输入文本输出向量数组。所以我们不通过 Ollama 的/api/chat调用它而是用ollama.embeddings()这个专用方法import ollama def get_embedding(text: str, model: str nomic-embed-text) - List[float]: 调用 Ollama 嵌入模型返回向量 注意model 必须是已 pull 的嵌入模型如 nomic-embed-text, mxbai-embed-large try: response ollama.embeddings(modelmodel, prompttext) return response[embedding] except Exception as e: print(f❌ 嵌入失败 {text[:20]}...: {e}) return [0.0] * 768 # 返回零向量避免中断流程 # 测试 vec1 get_embedding(主控板功耗为12W) vec2 get_embedding(主板耗电量是12瓦) print(f余弦相似度: {cosine_similarity([vec1], [vec2])[0][0]:.4f}) # 实测输出0.8231 —— 中文同义词映射很准为什么是nomic-embed-text而不是mxbai-embed-large我们做了实测对比模型向量维度单次嵌入耗时MacBook M2中文同义词相似度测试集内存占用是否支持中文长文本nomic-embed-text768120ms0.821.2GB✅ 支持 8K tokensmxbai-embed-large1024210ms0.791.8GB❌ 超过 512 tokens 易 OOMbge-m3(需手动加载)1024350ms0.852.4GB✅ 但需额外 pip install结论很清晰nomic-embed-text是本地部署的“甜点区间”——它比mxbai快近一倍内存占用低 33%而精度只差 0.03。对于一个 200 页的产品手册约 10 万 token用mxbai嵌入要 3 分钟用nomic只要 90 秒。而那 0.03 的精度差距在“客户投诉率最高的模块”这种事实型查询里几乎不影响 top-3 结果。注意nomic-embed-text的国内镜像源是https://registry.cn-hangzhou.aliyuncs.com/ollama/nomic-embed-text拉取命令OLLAMA_BASE_URLhttps://registry.cn-hangzhou.aliyuncs.com/ollama ollama pull nomic-embed-text这能解决“ollama pull 太慢”的热搜痛点。2.4 向量数据库Chroma 的极简主义哲学LangChain 绑定Chroma时会自动创建PersistentClient并管理 collection。但我们直接用chromadb原生 API只为一个目的把向量存储的控制权牢牢握在自己手里。import chromadb from chromadb.config import Settings def init_chroma_db(persist_dir: str ./chroma_db) - chromadb.Client: 初始化 Chroma 客户端使用磁盘持久化 client chromadb.PersistentClient( pathpersist_dir, settingsSettings(anonymized_telemetryFalse) # 关闭遥测 ) return client def create_or_get_collection(client: chromadb.Client, name: str) - chromadb.Collection: 创建或获取 collection设置自定义 embedding function可选 try: collection client.get_collection(namename) print(f✅ 已存在 collection: {name}当前文档数: {collection.count()}) except ValueError: # collection 不存在创建新的 collection client.create_collection( namename, metadata{hnsw:space: cosine} # 使用余弦相似度 ) print(f 新建 collection: {name}) return collection def add_documents_to_collection(collection: chromadb.Collection, documents: List[Dict], embeddings: List[List[float]], ids: List[str]): 批量添加文档、向量、元数据到 collection collection.add( documents[doc[page_content] for doc in documents], embeddingsembeddings, metadatas[doc[metadata] for doc in documents], idsids ) print(f 已添加 {len(documents)} 个文档到 collection) # 使用示例 client init_chroma_db(./my_rag_db) coll create_or_get_collection(client, product_manual) # 假设 docs 是 load_files_from_dir 的输出embeds 是批量调用 get_embedding 的结果 add_documents_to_collection(coll, docs, embeds, [fdoc_{i} for i in range(len(docs))])Chroma 的极简主义体现在三个地方无服务端它就是一个 Python 库数据存在本地./chroma_db文件夹里ls ./chroma_db就能看到index/、metadata/等文件夹你可以随时rm -rf ./chroma_db彻底重置无配置地狱不需要docker-compose.yml不需要chroma-server进程pip install chromadb后直接import就能用无抽象泄漏collection.add()的参数就是你要存的东西——文档、向量、元数据、ID没有Document类、没有BaseRetriever接口所见即所得。这才是本地 RAG 应该有的样子没有中间商赚差价没有抽象层挡视线一切都在你眼皮底下运行。3. 检索增强生成RAG的核心引擎如何让大模型“看见”你的知识库RAG 的灵魂不在“R”检索也不在“G”生成而在“AG”——如何把检索到的结果以最有效的方式“增强”进大模型的上下文让它生成的答案既准确又自然。LangChain 的RetrievalQA默认把检索结果拼成一段话塞给 LLM这在简单问答中可行但在真实业务中它会犯两个致命错误上下文污染把无关的元数据如{source: /docs/manual_v2.pdf, page: 42}原样塞进去LLM 可能会照抄这些字段答出“根据 manual_v2.pdf 第42页……”这种不专业的答案信息稀释把 5 个 chunk 拼成 2000 字的“背景材料”LLM 的注意力会平均分配关键数据如“投诉率 23.7%”反而被淹没。我们必须亲手设计这个“增强”过程。3.1 检索阶段不只是找相似还要懂业务规则chromadb.Collection.query()默认返回n_results5个最相似的 chunk但这只是起点。真正的业务需求往往更复杂时间过滤客户问“最新版手册里API 认证方式是什么”我们必须只检索frontmatter.last_updated最新的几个 chunk来源过滤问“传感器模块的功耗参数”应该只查source包含sensor的 PDF相关性阈值相似度低于 0.5 的结果宁可不返回也不能让 LLM 看到噪声。所以我们封装一个智能检索器smart_retriever.pyimport chromadb from typing import List, Dict, Any, Optional import numpy as np def smart_retrieve( collection: chromadb.Collection, query_text: str, model: str nomic-embed-text, n_results: int 5, score_threshold: float 0.4, filters: Optional[Dict[str, Any]] None ) - List[Dict[str, Any]]: 智能检索支持相似度阈值、元数据过滤、结果重排序 filters: 如 {source: {$contains: sensor}, frontmatter.version: v3.2} # 步骤1获取查询向量 query_embedding get_embedding(query_text, model) # 步骤2执行带过滤的查询 results collection.query( query_embeddings[query_embedding], n_resultsn_results, wherefilters or {} # Chroma 原生支持 SQL-like 过滤 ) # 步骤3应用相似度阈值过滤 filtered_results [] for i, doc in enumerate(results[documents][0]): score results[distances][0][i] # Chroma 返回的是距离distance越小越好我们转为相似度similarity similarity 1.0 - score if similarity score_threshold: filtered_results.append({ content: doc, metadata: results[metadatas][0][i], similarity: similarity, id: results[ids][0][i] }) # 步骤4按相似度降序排列 filtered_results.sort(keylambda x: x[similarity], reverseTrue) return filtered_results # 使用示例只查传感器相关的最新内容 filters { source: {$contains: sensor}, frontmatter.last_updated: {$gt: 2024-01-01} } results smart_retrieve(coll, 传感器A的最大工作电流是多少, filtersfilters)这个smart_retrieve的威力在于它把 Chroma 的where过滤语法直接暴露给你{source: {$contains: sensor}}比 LangChain 的MetadataFilter更直观它把distance转成similarity让你一眼看懂“0.82 是高还是低”它允许你动态调整score_threshold在“召回率”和“准确率”之间做业务权衡。3.2 生成阶段Prompt 工程不是玄学是结构化填空LangChain 的PromptTemplate用{context}和{question}占位看似灵活但实际中你会发现{context}里混着元数据LLM 会胡说{question}是原始用户输入可能带错别字或口语化表达需要预处理。我们放弃模板改用结构化 Prompt 构建器def build_rag_prompt(retrieved_docs: List[Dict], user_question: str, model_name: str qwen2:7b) - str: 构建 RAG Prompt严格分离 context、question、instruction # 步骤1清洗检索结果只留 page_content丢弃所有 metadata clean_contexts [doc[content] for doc in retrieved_docs] # 步骤2对用户问题做轻量预处理去口语、补全缩写 processed_question user_question.strip() # 示例把 功耗多少 补全为 该模块的功耗参数是多少 if 功耗 in processed_question and 多少 in processed_question: processed_question processed_question.replace(多少, 参数是多少) # 步骤3构建三段式 Prompt instruction ( 你是一个专业的产品技术支持助手。请严格基于以下提供的【知识库内容】回答问题。\n 要求\n - 只回答问题不要复述问题不要加解释性语句\n - 如果知识库中没有明确答案回答根据现有资料无法确定\n - 数值答案必须带单位如 W、℃、ms\n - 不要提及根据知识库、根据文档等字眼。 ) context_section 【知识库内容】\n \n---\n.join(clean_contexts) question_section f【用户问题】\n{processed_question} full_prompt f{instruction}\n\n{context_section}\n\n{question_section} return full_prompt # 测试 prompt build_rag_prompt(results, 传感器A的最大工作电流是多少) print(prompt[:200] ...) # 输出你是一个专业的产品技术支持助手。请严格基于以下提供的【知识库内容】回答问题。 # 要求 # - 只回答问题不要复述问题不要加解释性语句 # ... # 【知识库内容】 # 传感器A电气参数 # - 工作电压5V ± 5% # - 最大工作电流120mA # - 待机电流2μA # ... # 【用户问题】 # 传感器A的最大工作电流是多少这个 Prompt 的设计哲学是把 LLM 当成一个严格的“填空机器人”而不是一个自由发挥的作家。instruction是操作手册告诉它“只能做什么”context_section是干净的数据源不含任何干扰信息question_section是明确的任务指令。实测中用这个 Prompt 调qwen2:7b95% 的数值型问题如功耗、温度、尺寸都能给出精准答案且格式统一120mA不会出现大约 120 毫安或120 毫安左右这种模糊表述。3.3 完整 RAG 流水线把四步串成一个函数现在把前面所有模块串起来形成一个可调用的rag_query()函数def rag_query( user_question: str, collection: chromadb.Collection, ollama_model: str qwen2:7b, embedding_model: str nomic-embed-text, n_retrieve: int 3, score_threshold: float 0.5, filters: Optional[Dict] None ) - Dict[str, Any]: 端到端 RAG 查询函数 返回: {answer: str, retrieved_docs: List[Dict], prompt: str, cost_ms: int} import time start_time time.time() # 1. 检索 retrieved smart_retrieve( collectioncollection, query_textuser_question, modelembedding_model, n_resultsn_retrieve, score_thresholdscore_threshold, filtersfilters ) # 2. 构建 Prompt prompt build_rag_prompt(retrieved, user_question, ollama_model) # 3. 调用 Ollama 生成 try: response ollama.chat( modelollama_model, messages[{role: user, content: prompt}], options{temperature: 0.1, num_predict: 256} # 低温防幻觉 ) answer response[message][content].strip() except Exception as e: answer f❌ 生成失败: {str(e)} end_time time.time() return { answer: answer, retrieved_docs: retrieved, prompt: prompt, cost_ms: int((end_time - start_time) * 1000) } # 最终使用 result rag_query( user_question主控板的工作温度范围是多少, collectioncoll, filters{source: {$contains: mainboard}} ) print( 答案:, result[answer]) print(⏱️ 耗时:, result[cost_ms], ms) # 输出 答案: -20℃ ~ 70℃ # ⏱️ 耗时: 1842 ms这个函数就是我们 RAG 的“心脏”。它不依赖任何外部框架所有参数n_retrieve、score_threshold、filters都暴露给你你可以根据业务场景随时调整查政策文件把score_threshold提到0.7宁缺毋滥查客服话术把n_retrieve设为5多给些参考查实时日志加filters{timestamp: {$gt: 2024-05-20}}。注意ollama.chat()的options参数是关键。temperature0.1让 LLM 尽量保守避免编造num_predict256限制最大输出长度防止它滔滔不绝跑题。这些细节LangChain 的LLMChain默认值往往不适合 RAG 场景。4. 实战排障那些让 RAG “看起来能跑实际总不准”的隐形陷阱写完代码python rag_demo.py一跑终端输出 答案: -20℃ ~ 70℃你可能会欢呼雀跃。但别急接下来的三天你会被各种“看起来合理实际离谱”的问题折磨得怀疑人生。我把这些坑按发生频率排序告诉你怎么快速定位、怎么永久修复。4.1 问题检索结果明明有答案LLM 就是不答——根源在 Prompt 的“信任危机”现象smart_retrieve()返回的retrieved_docs里第一条就是“主控板工作温度-20℃ ~ 70℃”但rag_query()的最终答案却是“根据现有资料无法确定”。排查链路先print(result[prompt])看 Prompt 长什么样发现【知识库内容】部分里那条关键句子被裹在一大段无关的 PCB 布局描述里占了 800 字再看ollama.chat()的num_predict256意味着 LLM 最多看 256 个 token 的上下文而那 800 字的 chunk 已经超限关键信息被截断。根因Chunk 太大导致关键信息在 Token 截断时丢失。LangChain 默认chunk_size1000而qwen2:7b的上下文窗口是 32K token但 Ollama 默认只喂给它 2K token由num_ctx参数控制Ollama 默认是 2048。解决方案在分块时把chunk_size从 1000 降到 300我们前面chinese_text_splitter的默认值确保每个 chunk 都能完整进入 LLM 视野在ollama.chat()时显式指定更大的上下文如果模型支持response ollama.chat( modelqwen2:7b, messages[{role: user, content: prompt}], options