用用 Python + Embedding + Ollama 这个经典组合搭建本地知识库构建企业的智能推荐系统，核心是构建一个 RAG（检索增强生成）系统。它的工作流程可以概括为：将你的私有文档向量化后存入本地向量化数据，当用户提问时，系统先从本地向量化数据中检索相关片段，再让大模型根据这些片段推荐精准的回答。
一个典型的企业级本地知识库，其技术架构大致如下：
🛠️技术选型：搭建系统的“工具箱”根据不同的需求和侧重点，你可以灵活选择以下组件：
核心框架 (RAG&应用)
LangChain/LlamaIndex: 最主流的RAG开发框架，提供了从文档加载、分块到与向量库、大模型交互的完整工具箱。
轻量级框架: 如 Embedchain，可用几行代码快速构建RAG应用；或 Cognee，它结合了向量和图检索。
低代码/开箱即用平台: 如 Dify.AI 和 AnythingLLM，提供了图形化界面，能大幅降低开发门槛。
向量数据库 (存储知识)
ChromaDB: 轻量级，纯Python实现，嵌入式使用，是入门和中小型项目的首选。
FAISS: 由Facebook开源的高效向量检索库，擅长处理大规模数据。
Milvus/LanceDB: 分别适用于大规模、高并发的企业级分布式场景，和希望基于文件存储、支持混合搜索的场景。
Embedding模型 (将文本转为向量)
BAAI/bge-large-zh-v1.5: 中文语义向量化效果领先，是中文场景的优选。
nomic-embed-text: 通过Ollama即可本地部署的通用Embedding模型。
DeepSeek/Qwen: 这些大模型本身也提供Embedding能力。
大语言模型 (LLM,负责生成答案)
Ollama: 最流行的本地大模型管理工具，能让你一键部署并运行Llama、DeepSeek、Qwen等开源模型。
DeepSeek/Qwen: 中文能力强，是构建企业知识库的性价比之选。
服务与交互
FastAPI: 用于将你的知识库封装成RESTful API服务，方便与内部系统集成。
Streamlit: 可快速搭建一个用于演示和调试的Web界面。
📝核心实现步骤：从文档到智能问答下面以 Python+Embedding+Ollama 这一经典组合为例，展示核心实现步骤：
bashpip install langchain chromadb pypdf sentence-transformers streamlit接着，参考官方文档安装并启动 Ollama，然后拉取你需要的模型，例如一个Embedding模型和一个对话模型：
bashollama pull bge-m3ollama pull deepseek-r1:1.5b
- 向量化与存储 (构建知识库) 将文本块转换为向量,以使用多语言强模型bge-m3为例
- 首先确保已在本地部署好ollama和需要的大模型：
import pandas as pdimport requestsimport numpy as npimport pickleimport time# ========== 配置 ==========CSV_PATH =r"leaf_paths.csv"# 你的类目路径文件EMBED_MODEL ="bge-m3"# 使用多语言强模型OLLAMA_URL ="http://localhost:11434/api/embeddings"CACHE_FILE ="path_vectors_bge.pkl"# 新的缓存文件名，避免覆盖旧的# ========== 加载数据 ==========df = pd.read_csv(CSV_PATH)# 重要：你的列名是 nodePathpaths = df['nodePath'].tolist()print(f"📂 已加载 {len(paths)} 条类目路径")# ========== 向量生成函数 ==========def get_embedding(text):try: resp = requests.post( OLLAMA_URL, json={"model": EMBED_MODEL,"prompt": text}, timeout=60)if resp.status_code ==200:return np.array(resp.json()['embedding'])else: print(f"⚠️ 请求失败，状态码: {resp.status_code}")returnNoneexcept Exception as e: print(f"❌ 异常: {e}")returnNone# ========== 主流程 ==========print(f"⏳ 正在生成向量（使用 {EMBED_MODEL}）...")vectors =[]total = len(paths)for i, p in enumerate(paths): vec = get_embedding(p)if vec isnotNone: vectors.append(vec)# 每 50 条打印进度if(i +1)%50==0or(i +1)== total: print(f" 进度: {i +1}/{total} ({((i +1)/ total)*100:.1f}%)")# 轻微延迟，避免 Ollama 过载 time.sleep(0.01)# ========== 保存缓存 ==========if vectors:with open(CACHE_FILE,"wb")as f: pickle.dump({"paths": paths,"vectors": np.array(vectors)}, f) print(f"✅ 缓存已保存至 {CACHE_FILE}，共 {len(vectors)} 条向量")else: print("❌ 未生成任何向量，请检查网络或 Ollama 服务")
4. 实现智能推荐 (RAG核心)当用户输入问题时，从向量库检索相关文档，并结合大模型推荐最合适答案# category_classifier.pyimport numpy as npimport pickleimport requestsfrom sklearn.metrics.pairwise import cosine_similarityimport osimport time# ===== 添加 BASE_DIR：当前脚本所在目录 =====BASE_DIR = os.path.dirname(os.path.abspath(__file__))BLACKLIST_CATE =["其他",""]class CategoryClassifier:def __init__(self, cache_path: str, ollama_base: str ="http://localhost:11434", embed_model: str ="bge-m3",# llm_model: str = "qwen2.5:3b", # 可用# llm_model: str = "qwen3.5:2b", # 可用 不能精排 llm_model: str ="deepseek-r1:1.5b",# 可用# llm_model: str = "deepseek-r1:7b", # 可用 不能精排 blacklist: list =None):""" cache_path: path_vectors_bge.pkl 的完整路径 blacklist: 要过滤掉的关键词列表，例如 ["其他", "Other"] """ self.ollama_base = ollama_base self.embed_model = embed_model self.llm_model = llm_model self.cache_path = cache_path self.paths =None self.vectors =None# 黑名单默认值（若未传入则使用内置默认）if blacklist isNone: blacklist = BLACKLIST_CATE self.blacklist = blacklist self._load_cache()def _load_cache(self):ifnot os.path.exists(self.cache_path):raise FileNotFoundError(f"缓存文件 {self.cache_path} 未找到")with open(self.cache_path,"rb")as f: data = pickle.load(f) self.paths = data["paths"] self.vectors = data["vectors"] print(f"[分类器] ✅ 已加载 {len(self.paths)} 条类目向量（维度 {self.vectors.shape[1]}）")def set_blacklist(self, blacklist: list):"""动态更新黑名单""" self.blacklist = blacklist print(f"[分类器] 黑名单已更新: {self.blacklist}")def _is_blacklisted(self, path: str) -> bool:"""检查路径是否包含黑名单关键词"""for keyword in self.blacklist:if keyword in path:returnTruereturnFalsedef _get_embedding(self, text: str): url =f"{self.ollama_base}/api/embeddings"try: resp = requests.post(url, json={"model": self.embed_model,"prompt": text}, timeout=60) resp.raise_for_status()return np.array(resp.json()['embedding'])except Exception as e: print(f"[分类器] ❌ 向量获取失败: {e}")returnNonedef _llm_rerank(self, query: str, candidates: list):""" 使用 LLM 从候选路径中精排 candidates: list of (path, score) """ifnot candidates:returnNone candidate_text ="\n".join([f"{i+1}. {p}"for i,(p, _)in enumerate(candidates)]) prompt =f"""你是一个类目映射专家。用户输入的商品标题（可能是英语、法语、德语等）：{query}候选类目路径（中文）：{candidate_text}请从上述候选中选择最匹配的一条，只输出完整的路径，不要输出任何解释。如果都不匹配，输出"NULL"。""" url =f"{self.ollama_base}/api/generate"try: resp = requests.post( url, json={"model": self.llm_model,"prompt": prompt,"stream":False,"options":{"temperature":0}}, timeout=60) resp.raise_for_status() answer = resp.json()['response'].strip()# 校验答案是否在候选列表中for p, _ in candidates:if p == answer:return p# 模糊匹配for p, _ in candidates:if p in answer or answer in p:return p# 兜底返回第一个return candidates[0][0]if candidates elseNoneexcept Exception as e: print(f"[分类器] ⚠️ LLM 精排失败: {e}")return candidates[0][0]if candidates elseNonedef classify(self, title: str) -> str:""" 输入商品标题，返回匹配的中文类目路径（已过滤黑名单），若失败返回 None """ifnot title:returnNone# 1. 获取标题向量 query_vec = self._get_embedding(title)if query_vec isNone:returnNone# 2. 计算余弦相似度，先取 Top 20 以便过滤后仍有足够候选 similarities = cosine_similarity([query_vec], self.vectors)[0] top_indices = np.argsort(similarities)[-20:][::-1]# 取20个 raw_candidates =[(self.paths[i], similarities[i])for i in top_indices]# 3. 过滤黑名单 filtered_candidates =[(p, s)for p, s in raw_candidates ifnot self._is_blacklisted(p)]# 如果过滤后为空，则放宽限制，取未被过滤的第一个（但这种情况极少）ifnot filtered_candidates: print("[分类器] ⚠️ 所有候选均被黑名单过滤，使用未过滤的第一个") filtered_candidates = raw_candidates[:1]# 取前5个作为最终候选 final_candidates = filtered_candidates[:5]# 4. 如果最高分高于阈值，直接采纳 THRESHOLD =0.72if final_candidates[0][1]>= THRESHOLD:return final_candidates[0][0]# 5. 否则使用 LLM 精排（同样会基于过滤后的候选）return self._llm_rerank(title, final_candidates)if __name__ =="__main__":import sys# 使用默认路径，用户可根据需要修改 cache_path = os.path.join(BASE_DIR,"data","path_vectors_bge.pkl") blacklist =["其他",""] classifier = CategoryClassifier(cache_path, blacklist=blacklist)# 如果命令行有参数，则作为标题，否则使用预设列表if len(sys.argv)>1: titles =[" ".join(sys.argv[1:])]else: titles =["Shower Tray Base System 72x72 For Bathroom Renovation Wet Room Shower Remodeling Central Drain ABS Flange Shower Curb Membrane Roll Slope Joists EPS Foam Core Construction","Artificial plant wall 24 pieces 10x10 inches privacy hedge artificial green artificial boxwood panel PE privacy screen artificial plants 25x25 cm privacy hedge wall decoration plant wall"]for title in titles: print("\n"+"="*80) print(f"标题: {title}") print("-"*80)# 手动执行分类过程以显示候选 query_vec = classifier._get_embedding(title)if query_vec isNone: print("向量获取失败")continue similarities = cosine_similarity([query_vec], classifier.vectors)[0] top_indices = np.argsort(similarities)[-20:][::-1] raw_candidates =[(classifier.paths[i], similarities[i])for i in top_indices] print("原始 Top 10 候选（过滤前）:")for i,(p, s)in enumerate(raw_candidates[:10]): print(f"{i+1}. {p[:80]}... ({s:.3f})")# 应用黑名单过滤 filtered_candidates =[(p, s)for p, s in raw_candidates ifnot classifier._is_blacklisted(p)] print(f"过滤后剩余 {len(filtered_candidates)} 个候选")if filtered_candidates: print("过滤后的 Top 5:")for i,(p, s)in enumerate(filtered_candidates[:5]): print(f"{i+1}. {p[:80]}... ({s:.3f})")else: print("所有候选均被过滤，将使用未过滤的第一个") filtered_candidates = raw_candidates[:1] final_candidates = filtered_candidates[:5]# 判断是否直接采纳if final_candidates[0][1]>=0.72: final_path = final_candidates[0][0] method ="直接采纳（高置信度）"else: final_path = classifier._llm_rerank(title, final_candidates) method ="LLM 精排" print(f"最终结果 ( {method} ) : {final_path}")
💎 总结
总的来说，用Python搭建本地企业知识库是一个成熟且有清晰路径的任务。可以从 Python + Embedding + Ollama 这个经典组合入手，快速构建一个原型。在验证效果后，再根据企业的数据规模、并发需求和安全标准，逐步演进到更复杂的架构（如引入Milvus、ES等）。
快速上手指南：
安装：pip install langchain chromadb pypdf ollama。
启动Ollama：ollama pull bge-m3 和 ollama pull deepseek-r1:1.5b。
运行代码：将上面的核心代码整合到一个Python脚本中，放入你的文档即可运行。
