如何使用 MongoDB Atlas 和 Python 构建向量搜索应用

在不久前，计算机像高效却字面理解的图书管理员。如果您搜索“giant robot battles”，它只会找到包含这些精确词语的标题。像“Kaiju Mech Warfare”这样完美匹配语义的作品却会被忽略。计算机能识别词，却错过了其中的韵律——缺乏理解、细微差别和语境。这就是关键词搜索时代，多年来它一直是我们的最佳选择。

随后，革命到来。受人脑启发，人工智能发展出新的学习方式。AI 不再只是记忆词语，而是开始理解它们之间的关系。它知道“king”与“man”的关系，正如“queen”与“woman”的关系。它能识别“happy”“joyful”“ecstatic”属于同一情感领域，而“sombre”则完全不同。

AI 如何做到的？通过创建一张意义地图。

什么是向量搜索？意义地图

想象一个广袤的多维星系，每个概念——一个词、一句话、一张图片、一首歌——都是一颗星。在这个星系中，含义相近的星星会聚在一起，而无关的相隔光年。

一个向量嵌入就是这颗星在星系中的坐标。它是一个数字列表（例如[0.12, -0.45, 0.89, ...]），能在这张意义地图上精确定位一个概念。

向量搜索是导航这张地图的颠覆性能力。当您提问“一个关于孤儿男孩发现自己是巫师的电影”时，我们不会去搜这些精确的关键词。相反，我们会：

1. 在地图上找到您问题的坐标（向量）。
2. 定位最近的邻近星（我们的电影剧情）。

距离最近的星代表语义上最相似的结果。AI 正是通过这种方式，从字面管理员进化为理解您真正意思的智者。

为什么这如此重要？

按含义而非关键词搜索，为各行业的现代 AI 应用提供动力。这一能力使得：

• Netflix 能根据您的观看记录的相似“氛围”推荐您不熟悉的电影。
• Spotify 能为您策划感觉像您最爱歌曲的歌单，无论曲风或年代。
• 聊天机器人和大语言模型（LLM） 能把握您问题背后的意图，提供相关且具备上下文意识的回答。
• 电商网站 能根据您上传的图片展示外观相似的商品。

向量搜索连接了人类意图与数字数据，从根本上改变我们与信息互动的方式。这项技术让数字系统不再生硬，而更像直觉型的协作伙伴。

一起上手：用 MongoDB 构建您的第一个向量搜索应用

欢迎来到搜索的新前沿！MongoDB 在 Atlas 中引入了向量搜索。这让您可以直接在 MongoDB 中存储和查询向量嵌入（如来自文本、图像或音频的嵌入）——无需另行使用像 向量数据库（如 Pinecone 或 Weaviate）。

在本教程中，我们将使用 Python 和 MongoDB Atlas 构建一个简单而强大的鞋类推荐引擎。您将学习如何将文本转换为有意义的数值表示（向量），然后使用 Atlas 内置的向量搜索来查找剧情相似的电影。

您将学到什么

• 如何连接到 MongoDB Atlas 集群，并通过 Python 的 pymongo 库安全连接。
• 如何创建并插入文档，其中包含向量嵌入。
• 如何生成向量嵌入：使用 sentence-transformers 库从文本生成。
• 如何在 MongoDB Atlas UI 中构建向量搜索索引。
• 如何使用 语义搜索查询，并借助 $vectorSearch 聚合管道来实现。

先决条件

• 一个免费的 MongoDB Atlas 账户
• 本地已安装 Python 3.9+
• 安装依赖：

pip install pymongo sentence-transformers numpy技术支持

步骤 1：创建 MongoDB Atlas 集群

• 登录 Atlas。

• 创建免费集群。创建用户并复制用户名和密码。

• 创建用户后，点击选择连接方法。您会看到下方第一个界面。点击驱动程序（Drivers）并进入第二个界面。选择 Python 作为驱动以及相应版本。复制连接字符串。

• 前往数据库部署（Database Deployments） → 点击浏览集合（Browse Collections） → 创建一个新数据库：
• 数据库名：vectorDemo
• 集合名：items

• 启用向量搜索：
• 在 MongoDB Atlas UI 中，导航至Vector Search Indexes → 为items 创建新索引。
• 索引名称：“vector_index”
• 使用以下定义：

{
  "fields": [
    {
      "type": "vector",
      "path": "embedding",
      "numDimensions": 768,
      "similarity": "cosine"
    }
  ]
}
技术支持

注意：768 是我们将使用的 sentence-transformers 模型的嵌入维度。

理解索引定义：

• "type": "vector"：告诉 MongoDB Atlas 这是一个向量索引。
• "path": "embedding"：关键之处在于，它指向文档中包含向量的字段。
• "numDimensions"：768：该值必须与模型生成的向量维度一致。
• "similarity"：“cosine”：用于衡量向量“接近度”的算法。它非常适合文本相似度。

步骤 2：将 Python 连接到 MongoDB Atlas

获取您的 MongoDB Atlas 连接字符串：

• 在 MongoDB Atlas 中，点击Connect → Drivers → Python。
• 在下面的代码段中替换<username>、<password> 和 <cluster-url>。

from pymongo import MongoClient

# Replace with your Atlas connection string
uri = "mongodb+srv://<username>:<password>@<cluster-url>/vectorDemo?retryWrites=true&w=majority"

client = MongoClient(uri)
db = client["vectorDemo"]
collection = db["items"]
print("Connected to MongoDB Atlas")

# Clear existing
collection.delete_many({})
print("Cleared collection.")技术支持

步骤 3：使用 Python 生成嵌入

我们将使用SentenceTransformers 将文本转换为嵌入。

from sentence_transformers import SentenceTransformer
# Choose model: higher-quality for search -> "sentence-transformers/all-mpnet-base-v2" (768-d)
# If you need faster/smaller: "all-MiniLM-L6-v2" (384-d)
MODEL_NAME = "sentence-transformers/all-mpnet-base-v2"
# MODEL_NAME = "all-MiniLM-L6-v2"

# Note: if you use all-mpnet-base-v2, your Atlas vector index must use numDimensions: 768
# If you use all-MiniLM-L6-v2, index numDimensions: 384
model = SentenceTransformer(MODEL_NAME)

def embed_text(text: str):
    return model.encode(text).tolist()技术支持

步骤 4：插入带嵌入的示例文档

让我们将一些产品描述及其嵌入存储到 MongoDB 中。

# Each product includes name, description, list of features, use_cases and tags.
products = [
    # Marathon / long-distance running (explicit!)
    {
        "name": "MarathonPro 3000",
        "description": "Designed specifically for marathon and ultra distance running: maximum cushioning, breathable knit upper, responsive midsole and engineered heel support for long-run comfort.",
        "features": ["marathon", "long-distance", "max cushioning", "breathable", "support"],
        "category": "shoes",
        "use_cases": ["marathon", "long-distance running", "road running"],
        "tags": ["running", "endurance", "comfort"]
    },
    {
        "name": "Enduro LongRun",
        "description": "Lightweight long-run trainer optimized for marathon pacing with added cushioning, stable platform, and enhanced forefoot energy return for hours of comfortable running.",
        "features": ["long-run", "lightweight", "energy return", "cushioning"],
        "category": "shoes",
        "use_cases": ["marathon", "road running"],
        "tags": ["running","marathon","comfort"]
    },

    # Road running / neutral
    {
        "name": "Red Road Runner",
        "description": "Neutral road running shoe for daily training: breathable mesh, moderate cushioning and responsive ride; suitable for tempo runs and long easy runs.",
        "features": ["road running", "breathable", "moderate cushioning"],
        "category": "shoes",
        "use_cases": ["training", "road running", "tempo runs"],
        "tags": ["running","training"]
    },

    # Trail running (explicit trail)
    {
        "name": "TrailMaster RidgeGrip",
        "description": "Trail-running shoe with aggressive outsole, rock plate for protection, waterproof upper, designed for technical trails and rugged terrain.",
        "features": ["trail", "high grip", "rock plate", "waterproof"],
        "category": "shoes",
        "use_cases": ["trail running", "hiking", "off-road"],
        "tags": ["trail","outdoors"]
    },

    # Tennis & court
    {
        "name": "CourtPro Tennis",
        "description": "Court shoe with lateral support, non-marking sole and reinforced toe; built for tennis footwork and short intense bursts, not for long-distance running.",
        "features": ["court", "lateral support", "non-marking sole"],
        "category": "shoes",
        "use_cases": ["tennis","court sports"],
        "tags": ["tennis","sport"]
    },

    # Formal
    {
        "name": "Oxford Leather Formal",
        "description": "Classic leather formal shoe for office and events: polished leather, sturdy sole, refined silhouette.",
        "features": ["formal", "leather", "office"],
        "category": "shoes",
        "use_cases": ["formal wear","office"],
        "tags": ["formal","leather"]
    },

    # Hiking
    {
        "name": "Alpine Trek Boots",
        "description": "High-ankle hiking boots with waterproof membrane, ankle support and durable lugged outsole for mountain treks.",
        "features": ["hiking", "waterproof", "ankle support", "lugged outsole"],
        "category": "shoes",
        "use_cases": ["hiking","trekking","outdoor"],
        "tags": ["outdoors","hiking"]
    },

    # Casual sneakers
    {
        "name": "Classic Canvas Sneaker",
        "description": "Casual everyday canvas sneaker with rubber sole, comfortable fit for daily wear and walking.",
        "features": ["casual","canvas", "everyday"],
        "category": "shoes",
        "use_cases": ["casual","walking"],
        "tags": ["casual"]
    },

    # Cross-trainer
    {
        "name": "CrossFit Trainer",
        "description": "Stable cross-trainer shoe engineered for gym workouts, short sprints, and lateral movements; flexible yet supportive.",
        "features": ["cross-train","stable","gym"],
        "category": "shoes",
        "use_cases": ["gym","cross-training"],
        "tags": ["training","gym"]
    },

    # Cleats
    {
        "name": "Pro Soccer Cleats",
        "description": "Low-profile football/soccer cleats with studded outsole; optimized for grip on grass and agility.",
        "features": ["cleats","studs","agility"],
        "category": "shoes",
        "use_cases": ["soccer","football"],
        "tags": ["sports","cleats"]
    }
]

# ---------- Create 'canonical_text' to improve embedding quality for better scoring ----------
for p in products:
    # join features and others into one rich text blob
    features = ", ".join(p["features"])
    uses = ", ".join(p["use_cases"])
    tags = ", ".join(p["tags"])
    canonical = f"{p['name']}. {p['description']}. Features: {features}. Use cases: {uses}. Tags: {tags}."
    p["canonical_text"] = canonical

# ---------- Embed canonical_text----------
print("Embedding items...")
for p in products:
    p["embedding"] = embed_text(p["canonical_text"])

# ---------- Insert into MongoDB ----------
collection.insert_many(products)
技术支持

通过使用insert_many，我们对数据库执行一次高效的批量写入操作。

步骤 5：执行您的第一个向量搜索

到了最后也是最令人兴奋的一步！我们将使用MongoDB 的聚合框架和$vectorSearch——一个用于查找最佳鞋款查询的运算符。

query_text = "25k long running best shoes"
query_embedding = embed_text(query_text)

pipeline = [
    {
        "$vectorSearch": {
            "index": "vector_index",  # Use the vector search index you created
            "path": "embedding",
            "queryVector": query_embedding,
            "numCandidates": 100,
            "limit": 3
        }
    },
    {
        "$project": {
            "name": 1,
            "description":1,
            "score": {"$meta": "vectorSearchScore"}
        }
    }
]

results = collection.aggregate(pipeline)

print(f"\nSearch results for: '{query_text}'\n")
for r in results:
    print(f"- {r['name']} (score: {r['score']:.4f})")技术支持

剖析$vectorSearch 管道：

• $vectorSearch：核心运算符。它高效地将我们的queryVector 与所有文档中已建立索引的 plot_embedding 字段进行比较。
• $project：该阶段重塑输出。我们包含 name 和 description，排除 noisy _id 等字段，最重要的是，我们通过 $meta: "vectorSearchScore" 添加一个 score 字段。该分数范围为 0 到 1，表示结果与查询的相似度。

期望输出

如果一切设置正确，您将看到类似如下的结果：

Search results for: '25k long running best shoes'

- MarathonPro 3000 (score: 0.7980)
- Red Road Runner (score: 0.7479)
- Enduro LongRun (score: 0.7389)技术支持

请注意，“MarathonPro 3000”位居榜首——这正符合我们的预期。

恭喜您！您已从零构建了一个向量搜索应用。您学会了如何连接数据库、创建并验证数据、生成向量嵌入，并借助 MongoDB Atlas 的强大能力，以真正智能的方式查找信息。欢迎来到数据交互的未来。

总结

在本教程中，您成功学会了：

• 使用 Python 和 pymongo 连接到 MongoDB Atlas，为数据库驱动的应用打下基础。
• 创建嵌入（仅文本）来自一个真实的产品目录，使用如 all-mpnet-base-v2 等高质量模型，然后以适当的规范化方式存入 MongoDB。
• 构建丰富的结构化数据集，使用 canonical_text（名称、描述、特性、使用场景、标签），以提升嵌入的相关性。
• 执行向量相似度搜索，通过 MongoDB 的 $vectorSearch 聚合阶段，根据自然语言查询找到语义相似的产品。

后续步骤（自行尝试）

• 存储用户评价并构建一个语义搜索引擎。

尝试不同的相似度度量（dotProduct、euclidean）。