`$vectorSearch` 聚合阶段

定义

$vectorSearch阶段使用MongoDB Vector Search查询的结果丰富每个流媒体文档。对于每个输入文档，此阶段都会对集合运行 $vectorSearch查询，并将结果附加到您指定的字段。

展示位置

$vectorSearch 必须是管道的中间阶段。请勿将其用作管道的$source 或接收器。

语法

$vectorSearch 管道阶段采用以下原型形式：

{
  "$vectorSearch": {
    "from": {
      "connectionName": "<registered-atlas-connection>",
      "db": "<database>",
      "coll": "<collection>"
    },
    "as": "<output-field>",
    "queryVector": [<number>, ...] | <expression>,
    "query": {
      "text": "<string>" | <expression>
    },
    "model": "<embedding-model>",
    "index": "<vector-index-name>",
    "path": "<field-containing-vectors>",
    "numCandidates": <int>,
    "limit": <int>,
    "filter": { ... },
    "exact": <bool>,
    "searchNodePreference": {
      "key": "<preference>"
    },
    "pipeline": [ { $<aggregation-stage>: { ... } }, ... ],
    "let": {
      "<variable>": <expression>,
      . . .
    }
  }
}

$vectorSearch 阶段采用包含以下字段的文档：

字段	类型	必要性	说明
`from`	文档	必需	用于指定 `$vectorSearch` 查询每个输入文档的目标Atlas集合的文档。
`from.connectionName`	字符串	必需	连接注册表中Atlas连接的名称。 Atlas Stream Processing在解析时拒绝任何其他连接类型。
`from.db`	字面字符串	必需	目标数据库的名称。
`from.coll`	字面字符串	必需	包含MongoDB Vector Search索引的目标集合的名称。
`as`	字符串	必需	搜索结果大量的流媒体文档上的输出字段。如果MongoDB Vector Search 没有返回匹配的文档， Atlas Stream Processing会将该字段设置为空大量。警告：如果此字段为空或与流元数据字段名称冲突， Atlas Stream Processing将在启动时报告`StreamProcessorInvalidOptions` 错误。
`queryVector`	大量\|字段路径（Field Path）表达式	可选的	用于搜索匹配文档的显式向量嵌入。
`query`	文档	可选的	指定自动嵌入的文本查询的文档。
`query.text`	字符串 \|字段路径（Field Path）表达式	可选的	用于自动嵌入的查询文本。
`model`	字符串	Optional	MongoDB Vector Search 用于将 `query.text` 转换为向量的嵌入模型。仅与 `query` 一起使用时有效。当您省略 `model` 时， MongoDB Vector Search 将使用目标MongoDB Vector Search 索引定义中配置的模型。
`index`	字符串	必需	要查询的MongoDB Vector Search索引的名称。
`path`	字符串	必需	目标集合中包含要搜索的向量嵌入的字段。
`numCandidates`	int	可选的	MongoDB Vector Search 在搜索过程中考虑的候选对象数量。除非`exact` 为`true` ，否则为必填项。要学习；了解如何选择值，请参阅 numCandidates 选择。
`limit`	int	必需	要返回的最大结果数。不得超过 `numCandidates`。警告：如果 `limit`超过`numCandidates` ，每个文档都会在运行时失败并出现`PlanExecutor` 错误。
`filter`	文档	Optional	在向量搜索之前应用的预筛选表达式。有关支持的操作符，请参阅筛选器。警告：如果 `filter`使用了不受支持的操作符，则每个文档都会在运行时失败并出现`PlanExecutor` 错误。
`exact`	布尔	Optional	指定运行ANN 或 ENN搜索的标志。如果省略 `numCandidates`，请提供此字段。值可以是以下值之一： `false` 运行ANN搜索 `true` 运行新奥搜索默认值为 `false`。
`searchNodePreference`	文档	Optional	具有 `key`字段的文档， Atlas Stream Processing使用该文档将查询路由到特定搜索节点。 `key` 值必须是非空字符串。
`pipeline`	阵列	Optional	Atlas Stream Processing在将搜索结果附加到输入文档之前对搜索结果运行的其他聚合阶段。
`let`	文档	Optional	Atlas Stream Processing针对每个输入文档进行评估的变量。每个值都可以是文字、字段路径（Field Path）或表达式。`pipeline` 阶段可以将解析值引用为`$$<variable>` 。

行为

查询模式

$vectorSearch 支持两种互斥的查询模式：queryVector 表示向量嵌入，query.text 表示自动嵌入。在任一模式下，您都可以指定字面值或字段路径（Field Path）表达式，以便单个流处理器可以为每个输入文档发出不同的查询。

Error Handling

如果目标MongoDB Vector Search索引不存在，则当 $vectorSearch 首次运行时，流处理器会进入错误状态。 Atlas Stream Processing报告 MongoServerError 错误并停止处理文档。

如果queryVector 或query.text 在运行时计算出的类型错误， Atlas Stream Processing仅将有问题的文档路由到死信队列（DLQ）（如果已配置），否则会将其删除。流处理器会继续进程后续文档。有关 DLQ文档模式，请参阅死信队列。

处理结果

在Atlas Stream Processing将搜索结果附加到输入文档之前，使用 pipeline 进一步进程搜索结果。 Atlas Stream Processing将 pipeline 逐字转发到MongoDB Vector Search。要引用 pipeline 内输入文档的值，请将其传递给 let 并将其作为 $$<variable> 引用。

示例

以下示例查询sample_mflix 数据集，其中包含有关电影和电影院的数据。要运行它们，请将示例数据加载到Atlas 集群，并在集合上创建MongoDB Vectorembedded_movies Search索引。

通过自动嵌入丰富语义搜索

以下示例使用电影推荐来丰富传入的搜索事件。每个输入文档都有一个 topic字段。此管道使用 query.text， MongoDB Vector Search 使用 voyage-4-lite 模型为每个文档的 topic 值生成嵌入。此聚合分为三个阶段：

$source阶段与Atlas数据库建立连接，特别针对searches 数据库中的app 集合。fullDocumentOnly 选项要求在源集合上启用前像和后像。
$vectorSearch 阶段使用每个文档的 topic字段中的文本查询 sample_mflix.embedded_movies集合上的 plot_vector_index索引，并将最多五个结果附加到 recommendations字段。 pipeline字段将每个结果限制为其 title 和 plot 字段。
$merge阶段将扩充后的文档写入enriched_searches 数据库中的app 集合。

{
  "$source": {
    "connectionName": "srcCluster",
    "db": "app",
    "coll": "searches",
    "config": { "fullDocument": "required", "fullDocumentOnly": true }
  },
  "$vectorSearch": {
    "from": {
      "connectionName": "atlasCluster",
      "db": "sample_mflix",
      "coll": "embedded_movies"
    },
    "as": "recommendations",
    "query": {
      "text": "$topic"
    },
    "model": "voyage-4-lite",
    "index": "plot_vector_index",
    "path": "plot",
    "numCandidates": 50,
    "limit": 5,
    "pipeline": [
      { "$project": { "_id": 0, "title": 1, "plot": 1 } }
    ]
  },
  "$merge": {
    "into": {
      "connectionName": "atlasCluster",
      "db": "app",
      "coll": "enriched_searches"
    }
  }
}

写入输出集合的丰富文档如下所示：

{
  _id: ObjectId('6a4beebd75f00b6ddf98a8c0'),
  topic: 'gangster crime drama',
  recommendations: [
    { title: 'Scarface', plot: 'An ambitious and near insanely violent gangster climbs the ladder of success in the mob, but his weaknesses prove to be his downfall.' },
    { title: 'That Demon Within', plot: 'A dutiful cop, guilt-ridden over saving the life of a gang leader, becomes obsessed with bringing down the crime syndicate of the man he saved.' },
    { title: 'Billa 2', plot: 'An ordinary man from the slums enters an underworld gang and becomes the most feared underworld don.' },
    { title: 'A Better Tomorrow', plot: 'A reforming ex-gangster tries to reconcile with his estranged policeman brother, but the ties to his former gang are difficult to break.' },
    { title: 'Singham', plot: 'A humiliated gangster uses his influence and goon power to terrorize a newly transferred police officer.' }
  ]
}

使用 $rerank 进行精度关键的 RAG 上下文检索

以下示例为 RAG管道检索高精度电影情节段落。由于每个传入事件已包含预先计算的 queryEmbedding字段，因此管道使用。queryVector $vectorSearch阶段检索大量候选设立以最大限度地提高召回率，$rerank 中的pipeline 阶段对这些候选段落重新排序，以便只有最相关的段落继续下游。此聚合有四个阶段：

$source阶段与Atlas数据库建立连接，特别针对events 数据库中的llm_requests 集合。fullDocumentOnly 选项要求在源集合上启用前像和后像。
$vectorSearch 阶段使用每个文档的 queryEmbedding字段作为 queryVector 来查询 sample_mflix.embedded_movies集合上的 plot_embedding_index索引。它将 numCandidates 设为高电平以最大限度地提高召回率，并将最多 50 个匹配项传递到 pipeline 中。在 pipeline 内部：
- $rerank使用rerank-2.5-lite 模型按与查询文本的相关性对候选进行重新排序。
- $limit 保留八个最相关的段落。
- $project 仅返回构建提示所需的字段。
query.text 值引用 let 中定义的 searchText 变量，该变量解析为每个文档的 queryText字段。 Atlas Stream Processing将结果存储在 context字段中。
$unset阶段删除了queryEmbedding 和queryText 字段，因为下游不再需要它们。
$emit 阶段将扩展后的文档发送到qa_enriched 主题。

重要

在运行使用$rerank 的管道之前，请确认Atlas项目管理员已启用原生重排名。

{
  "$source": {
    "connectionName": "appCluster",
    "db": "llm_requests",
    "coll": "events",
    "config": { "fullDocument": "required", "fullDocumentOnly": true }
  },
  "$vectorSearch": {
    "from": {
      "connectionName": "atlasCluster",
      "db": "sample_mflix",
      "coll": "embedded_movies"
    },
    "as": "context",
    "queryVector": "$queryEmbedding",
    "index": "plot_embedding_index",
    "path": "plot_embedding",
    "numCandidates": 200,
    "limit": 50,
    "let": { "searchText": "$queryText" },
    "pipeline": [
      {
        "$rerank": {
          "query": { "text": "$$searchText" },
          "path": "plot",
          "model": "rerank-2.5-lite",
          "numDocsToRerank": 50
        }
      },
      { "$limit": 8 },
      { "$project": { "_id": 0, "title": 1, "plot": 1 } }
    ]
  },
  "$unset": ["queryEmbedding", "queryText"],
  "$emit": {
    "connectionName": "appKafka",
    "topic": "qa_enriched"
  }
}

发送到该主题的文档如下所示：

{
  _id: ObjectId('6a4c0cbc37d690084f3115da'),
  context: [
    { title: 'A Better Tomorrow', plot: 'A reforming ex-gangster tries to reconcile with his estranged policeman brother, but the ties to his former gang are difficult to break.' },
    { title: 'Yuma', plot: 'A story about the rise and fall of Zyga - a Polish kid in his early twenties.' },
    { title: 'Ghost Dog: The Way of the Samurai', plot: 'An African American mafia hit man who models himself after the samurai of old finds himself targeted for death by the mob.' }
    ...
  ]
}

$vectorSearch 聚合阶段

定义