/ /

`$vectorSearch` 聚合阶段

定义

$vectorSearch 阶段使用 MongoDB 向量搜索查询的结果丰富每个流媒体文档。对于每个输入文档，该阶段都会针对集合运行 $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`	literal string	必需	目标数据库的名称。
`from.coll`	literal string	必需	包含 MongoDB 向量搜索索引的目标集合的名称。
`as`	字符串	必需	流式文档上用于搜索结果数组的输出字段。如果 MongoDB 向量搜索未返回匹配文档，Atlas Stream Processing 会将此字段设置为空数组。警告：如果此字段为空或与流元数据字段名称冲突，Atlas Stream Processing 将在启动时报告 `StreamProcessorInvalidOptions` 错误。
`queryVector`	数组 \| 字段路径（Field Path）表达式	可选的	用于搜索匹配文档的显式向量嵌入。
`query`	文档	可选的	指定自动嵌入的文本查询的文档。
`query.text`	string \| 字段路径（Field Path）表达式	可选的	用于自动嵌入的查询文本。
`model`	字符串	Optional	嵌入模型 MongoDB 向量搜索用于将 `query.text` 转换为向量。仅在 `query` 旁边有效。当您省略 `model` 时，MongoDB 向量搜索会使用目标 MongoDB 向量搜索索引定义中配置的模型。
`index`	字符串	必需	要查询的 MongoDB 向量搜索索引的名称。
`path`	字符串	必需	目标集合中的字段，其中包含用于搜索的向量嵌入。
`numCandidates`	int	可选的	MongoDB 向量搜索在搜索期间考虑的候选数量。除非 `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 向量搜索索引不存在，则当 $vectorSearch 首次运行时，流处理器将进入错误状态。Atlas Stream Processing 报告 MongoServerError 错误，并停止处理文档。

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

处理结果

在 Atlas Stream Processing 将搜索结果附加到输入文档之前，使用 pipeline 进一步处理搜索结果。Atlas Stream Processing 将 pipeline 原样转发到 MongoDB 向量搜索。要引用 pipeline 内的输入文档中的值，请通过 let 传递并将其引用为 $$<variable>。

示例

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

带有自动嵌入的语义搜索增强

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

$source 阶段与 Atlas 数据库建立连接，具体针对 app 数据库中的 searches 集合。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 阶段检索设立候选段落以最大限度地提高召回率，pipeline 中的 $rerank 阶段对这些候选段落重新排序，以便只有最相关的段落继续下游。此聚合有四个阶段：

$source 阶段与 Atlas 数据库建立连接，具体针对 llm_requests 数据库中的 events 集合。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 聚合阶段

定义