`$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` 运行精确最近邻 (ENN) 搜索默认值为 `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 文档模式，请参阅死信队列（DLQ）。

处理结果

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

示例

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

以下示例使用电影推荐丰富入站搜索事件。每个输入文档都有一个 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 阶段将增强的文档写入 app 数据库中的 enriched_searches 集合。

{  "$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.' }  ]}

以下示例检索用于 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 聚合阶段

定义

展示位置

语法

行为

查询模式

Error Handling

处理结果

示例

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

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

`$vectorSearch` 聚合阶段