用 MAX 文本嵌入模型构建语义知识库Embedding、KMeans 聚类与语义搜索实战【免费下载链接】mojoThe Modular Platform (includes MAX Mojo)项目地址: https://gitcode.com/GitHub_Trending/mo/mojo本指南基于 max/examples/embedding-knowledge-base/README.md 及其配套源码讲解如何用 MAX 部署的文本嵌入Embedding模型构建一个支持语义搜索与主题聚类的文档知识库系统。你将学会启动max serve嵌入端点、理解SmartKnowledgeBase的完整实现原理含重试机制、增量嵌入、KMeans 聚类、余弦相似度检索并直接运行仓库自带的示例程序。一、方案概览为什么用 Embedding 聚类管理文档传统的文档检索依赖关键词匹配无法理解语义。而文本嵌入Text Embedding把一段文字映射为高维数值向量语义相近的文本在向量空间中距离也更近从而支持语义搜索用自然语言提问返回含义最相近的文档而非仅靠字面匹配主题聚类把向量空间中的文档按相似度自动分组发现文档集合中的主题结构话题推荐针对一个查询判断它最可能属于哪个主题簇再返回该簇内全部文档。仓库中的embedding-knowledge-base示例正是这一思路的最小可运行实现MAX 负责把文本变成向量推理侧Python 侧的SmartKnowledgeBase类负责管理文档、增量生成向量、聚类与检索应用侧。这也与官方文档 docs/max/serve/embeddings.mdx 中介绍的 RAG、智能体上下文注入、个性化推荐与聚类分析等应用场景一脉相承。二、环境准备示例项目的依赖与结构示例项目位于 max/examples/embedding-knowledge-base/其目录结构如下embedding-knowledge-base/ ├── README.md # 使用说明本指南主体 ├── pyproject.toml # 项目元数据与依赖声明 └── src/ └── embeddings/ ├── __init__.py └── kb_system.py # 知识库核心实现从 pyproject.toml 可以看到运行该系统所需的 Python 依赖依赖版本约束作用numpy2.2.2,3向量数组存储与计算vstack、argsort 等scikit-learn1.6.1,2KMeans聚类与cosine_similarity余弦相似度requests2.32.3,3调用 MAX 的 HTTP 嵌入端点项目要求requires-python 3.11并使用 pixi 作为包管理工具pyproject.toml中声明了numpy、scikit-learn、requests三个依赖以及指向 conda 的 MAX 频道。安装好依赖后即可按下一步启动服务。三、启动嵌入模型服务max serve知识库系统的向量引擎由 MAX 提供。使用以下命令启动本地模型服务器max serve --model sentence-transformers/all-mpnet-base-v2该命令会加载all-mpnet-base-v2这一句子转换sentence-transformer嵌入模型并启动一个 OpenAI 兼容的 HTTP 服务。官方文档 docs/max/serve/embeddings.mdx 指出服务默认监听http://localhost:8000/v1/embeddings端点在终端打印出如下信息时即表示就绪Server ready on http://0.0.0.0:8000 (Press CTRLC to quit)关于模型与端点的几点补充说明源自 docs/max/serve/embeddings.mdxv1/embeddings端点与 OpenAI Embeddings API 完全兼容请求体为 JSON至少包含model模型 ID与input待嵌入文本两个字段一次请求可以传入多条文本服务返回的data数组中每个元素对应一条输入的向量MAX 同时支持 OpenAI Python SDK 与curl两种客户端方式访问详见后文与端点直接交互小节--model参数接受 Hugging Face 仓库 IDMAX 会根据仓库 ID 自动完成架构识别与加载参见 docs/max/develop/index.mdx 中关于 MAX 模型部署的描述。四、知识库核心实现SmartKnowledgeBase 源码拆解示例的知识库逻辑全部位于 max/examples/embedding-knowledge-base/src/embeddings/kb_system.py。下面按功能模块逐一拆解其实现原理。4.1 构造与状态管理class SmartKnowledgeBase: def __init__( self, endpoint: str http://localhost:8000/v1/embeddings ) - None: self.endpoint endpoint self.documents: list[str] [] self.doc_titles: list[str] [] self.embeddings: np.ndarray None self.clusters: dict[int, list[int]] {}默认端点与max serve的默认监听地址一致因此无需额外配置即可对接。类的四个核心状态分别是文档正文列表、文档标题列表、按行堆叠的嵌入向量矩阵embeddings[i]对应documents[i]、以及聚类结果clusters主题 ID → 文档下标列表。4.2 嵌入请求与重试机制def _get_embedding( self, texts: list[str], max_retries: int 3 ) - np.ndarray: Get embeddings with retry logic. for attempt in range(max_retries): try: response requests.post( self.endpoint, headers{Content-Type: application/json}, json{ input: texts, model: sentence-transformers/all-mpnet-base-v2, }, timeout5, ).json() return np.array( [item[embedding] for item in response[data]] ) except Exception as e: if attempt max_retries - 1: raise Exception( fFailed to get embeddings after {max_retries} attempts: {e} ) logger.warning(fAttempt {attempt 1} failed, retrying...)该方法直接以 HTTP POST 方式调用v1/embeddings端点批量输入texts是字符串列表一次请求可同时嵌入多条文本超时与重试单次请求超时 5 秒失败时最多重试 3 次max_retries3每次失败会打印警告日志最终失败抛出异常响应解析从 OpenAI 兼容的响应结构中提取data数组中每个元素的embedding字段组装为numpy二维数组。此外还提供了带缓存的单文本嵌入版本lru_cache(maxsize1000) # noqa: B019 def _get_embedding_cached(self, text: str) - np.ndarray: Cached version for single text embedding. return self._get_embedding([text])[0]lru_cache缓存最近 1000 条文本的嵌入结果在查询阶段避免对重复文本反复调用模型服务。4.3 添加文档增量嵌入def add_document(self, title: str, content: str) - None: Add a single document with title. self.doc_titles.append(title) self.documents.append(content) # Update embeddings if len(self.documents) 1: self.embeddings self._get_embedding([content]) else: self.embeddings np.vstack( [self.embeddings, self._get_embedding([content])] ) # Recluster if we have enough documents if len(self.documents) 3: self._cluster_documents()add_document采用增量策略每添加一篇文档就实时调用嵌入端点获取其向量通过np.vstack追加到现有向量矩阵的行尾保持向量行号 文档下标的对应关系。当文档数量达到 3 篇后每次新增都会触发一次重新聚类。4.4 聚类KMeans 自动分主题def _cluster_documents(self, n_clusters: int | None None) - None: Cluster documents into topics. if n_clusters is None: n_clusters max(2, len(self.documents) // 5) n_clusters min(n_clusters, len(self.documents)) kmeans KMeans(n_clustersn_clusters, random_state42).fit( self.embeddings ) self.clusters {} for i in range(n_clusters): self.clusters[i] np.where(kmeans.labels_ i)[0].tolist()聚类采用scikit-learn的 KMeans 算法簇数量自动推导默认取max(2, 文档总数 // 5)即每约 5 篇文档归为一个主题但至少 2 个簇安全上限簇数量不会超过文档总数否则 KMeans 无意义固定随机种子random_state42保证结果可复现结果组织按簇 ID 把成员文档的下标收集进self.clusters供后续按主题检索。4.5 语义搜索余弦相似度 Top-Kdef search( self, query: str, top_k: int 3 ) - list[tuple[str, str, float]]: Find documents most similar to the query. query_embedding self._get_embedding_cached(query) similarities cosine_similarity([query_embedding], self.embeddings)[0] top_indices np.argsort(similarities)[-top_k:][::-1] return [ (self.doc_titles[i], self.documents[i], similarities[i]) for i in top_indices ]搜索流程三步走先对查询文本调用缓存版嵌入方法得到查询向量用cosine_similarity计算查询向量与全部文档向量之间的余弦相似度用np.argsort取出相似度最高的top_k个文档下标默认 3 篇返回(标题, 正文, 相似度分数)三元组列表。4.6 主题查询与话题推荐def get_topic_documents(self, topic_id: int) - list[tuple[str, str]]: Get all documents in a topic cluster. return [ (self.doc_titles[i], self.documents[i]) for i in self.clusters.get(topic_id, []) ] def suggest_topics( self, query: str, top_k: int 2 ) - list[tuple[int, float]]: query_embedding self._get_embedding_cached(query) topic_similarities [] for topic_id, doc_indices in self.clusters.items(): topic_embeddings self.embeddings[doc_indices] similarity cosine_similarity( [query_embedding], topic_embeddings ).max() topic_similarities.append((topic_id, similarity)) # Remove [0] return sorted(topic_similarities, keylambda x: x[1], reverseTrue)[ :top_k ]get_topic_documents(topic_id)根据聚类结果取出某个主题簇内的全部文档suggest_topics(query)对每个主题簇计算查询向量与该簇内所有文档向量的余弦相似度并取最大值作为簇相关度排序后返回相关度最高的top_k个主题默认 2 个及其相关度分数。五、运行示例python -m embeddings.kb_system服务启动后在embedding-knowledge-base目录下执行python -m embeddings.kb_system源码if __name__ __main__分支会按顺序执行三类演示演示一语义搜索——向知识库提问How do I change my password?期望返回与密码修改语义最相近的Password Reset Guide文档及相似度分数演示二话题推荐——以Where can I update my credit card?为查询调用suggest_topics找出最相关的主题簇再列出该簇内全部文档演示三按主题取文档——直接调用get_topic_documents(0)打印 Topic 0 聚类中的所有文档。示例内置了 6 篇模拟文档覆盖密码重置 / 账户安全 / 账单概览 / 支付方式 / 安装指南 / 系统要求六类内容恰好构成 2 个以上主题簇便于观察聚类效果。运行时终端会打印INFO级别的嵌入请求日志与各演示结果输出形如Searching for password help: Title: Password Reset Guide Relevance: 0.86 Content: To reset your password: 1. Click Forgot Password ...相似度数值取决于模型实际输出不同输入会得到不同分数。注意运行前务必保证max serve已就绪即 8000 端口可访问否则_get_embedding在重试 3 次后会抛出Failed to get embeddings after 3 attempts异常。六、扩展直接与嵌入端点交互若不使用SmartKnowledgeBase也可以直接用 OpenAI SDK 或 curl 消费max serve提供的嵌入能力这一点在官方文档 docs/max/serve/embeddings.mdx 中有完整说明。使用 OpenAI Python SDKfrom openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1, api_keyEMPTY) response client.embeddings.create( modelsentence-transformers/all-mpnet-base-v2, inputRun an embedding model with MAX!, ) print(f{response.data[0].embedding[:5]})使用 curlcurl http://localhost:8000/v1/embeddings \ -H Content-Type: application/json \ -d { input: Run an embedding model with MAX!, model: sentence-transformers/all-mpnet-base-v2 }两种方式返回相同的 OpenAI 兼容 JSON 结构核心字段为data[].embedding数值向量。SmartKnowledgeBase._get_embedding正是直接解析这一结构来完成与 MAX 的对接。七、小结与扩展方向embedding-knowledge-base示例展示了MAX 推理 Python 应用的完整链路max serve提供 OpenAI 兼容的v1/embeddings端点SmartKnowledgeBase在其之上实现了文档管理、增量嵌入、KMeans 聚类与余弦相似度检索。基于这套骨架可以继续扩展对接真实文档源将add_document的输入替换为网页抓取、PDF 解析或数据库导出的真实文档引入 RAG把search返回的 Top-K 文档作为上下文注入大模型实现检索增强生成更换嵌入模型将max serve --model与请求体中的model字段替换为其他 Hugging Face 仓库 ID 的嵌入模型持久化向量将self.embeddings导出为文件或向量数据库避免每次启动重新计算。相关参考文件示例 README、核心实现 kb_system.py、项目依赖声明 pyproject.toml、MAX 嵌入端点官方文档 docs/max/serve/embeddings.mdx。【免费下载链接】mojoThe Modular Platform (includes MAX Mojo)项目地址: https://gitcode.com/GitHub_Trending/mo/mojo创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考