`$vectorSearch` Estágio de agregação

Definição

O estágio $vectorSearch enriquece cada documento de streaming com resultados de uma query de pesquisa vetorial do MongoDB. Para cada documento de entrada, o estágio executa uma query $vectorSearch em uma coleção e anexa os resultados a um campo especificado.

Colocação

$vectorSearch deve ser um estágio intermediário do seu pipeline. Não o utilize como $source ou destino do seu pipeline.

Sintaxe

O estágio de pipeline do $vectorSearch tem a seguinte forma de protótipo:

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

O estágio $vectorSearch recebe um documento com os seguintes campos:

Campo	Tipo	necessidade	Descrição
`from`	documento	Obrigatório	Documento que especifica a coleção Atlas de destino que `$vectorSearch` query para cada documento de entrada.
`from.connectionName`	string	Obrigatório	Nome de uma conexão Atlas no seu registro de conexões. O Atlas Stream Processing rejeita qualquer outro tipo de conexão no momento da análise.
`from.db`	string literal	Obrigatório	Nome do banco de dados de destino.
`from.coll`	string literal	Obrigatório	Nome da coleção de destino que contém o índice de pesquisa vetorial do MongoDB.
`as`	string	Obrigatório	Campo de saída no documento de streaming para o array de resultados da pesquisa. Se a pesquisa vetorial do MongoDB não retornar nenhum documento correspondente, o Atlas Stream Processing definirá esse campo como um array vazio. AVISO: Se este campo estiver vazio ou colidir com o nome do campo de metadados do fluxo, o Atlas Stream Processing relatará um erro `StreamProcessorInvalidOptions` no início.
`queryVector`	array \| expressão de caminho do campo	Condicional	embedding de vetor explícito para pesquisar documentos correspondentes.
`query`	documento	Condicional	Documento que especifica a query de texto para embedding automático.
`query.text`	string \| expressão de caminho do campo	Condicional	Query de texto a ser usada para auto-embedding.
`model`	string	Opcional	Modelo de embedding que a pesquisa vetorial do MongoDB usa para converter `query.text` em um vetor. Válido apenas junto com `query`. Quando você omite `model`, o MongoDB Vector Search usa o modelo configurado na definição do índice de pesquisa vetorial do MongoDB de destino.
`index`	string	Obrigatório	Nome do índice de pesquisa vetorial do MongoDB para query.
`path`	string	Obrigatório	Campo na coleção de destino que contém os embeddings de vetor a serem pesquisados.
`numCandidates`	int	Condicional	Número de candidatos que a pesquisa vetorial do MongoDB considera durante a pesquisa. Obrigatório, a menos que `exact` seja `true`. Para saber como escolher um valor, consulte Seleção numCandidates.
`limit`	int	Obrigatório	Número máximo de resultados a serem retornados. Não deve exceder `numCandidates`. AVISO: Se `limit` exceder `numCandidates`, todos os documentos falharão em tempo de execução com um erro `PlanExecutor`.
`filter`	documento	Opcional	Expressão de pré-filtro a ser aplicada antes da pesquisa vetorial. Para operadores compatíveis, consulte filtro. AVISO: Se `filter` usar um operador não compatível, todos os documentos falharão em tempo de execução com um erro `PlanExecutor`.
`exact`	booleano	Opcional	Sinalizador que especifica se a pesquisa ANN ou ENN deve ser executada. Forneça este campo se você omitir `numCandidates`. O valor pode ser um dos seguintes: `false` para executar a pesquisa ANN `true` para executar a pesquisa ENN Padrão é `false`.
`searchNodePreference`	documento	Opcional	Documento com um campo `key` que o Atlas Stream Processing usa para rotear a query para um nó de pesquisa específico. O valor `key` deve ser uma string não vazia.
`pipeline`	array	Opcional	Estágios de agregação adicionais que o Atlas Stream Processing executa em relação aos resultados da pesquisa, antes de anexá-los ao documento de entrada.
`let`	documento	Opcional	Variáveis que o Atlas Stream Processing avalia em relação a cada documento de entrada. Cada valor pode ser um literal, um caminho do campo ou uma expressão. `pipeline` estágios podem referenciar os valores resolvidos como `$$<variable>`.

Comportamento

Modos de query

$vectorSearch suporta dois modos de query mutuamente exclusivos: queryVector para embedding de vetor e query.text para embedding automático. Em qualquer modo, você pode especificar um valor literal ou uma expressão de caminho do campo, para que um único processador de fluxo possa emitir uma query diferente para cada documento de entrada.

Error Handling

Se o índice de pesquisa vetorial do MongoDB de destino não existir, o processador de fluxo entrará em um estado de erro quando $vectorSearch for executado pela primeira vez. O Atlas Stream Processing relata um erro MongoServerError e interrompe o processamento de documentos.

Se queryVector ou query.text for avaliado como o tipo errado em tempo de execução, o Atlas Stream Processing encaminha apenas o documento infrator para a fila de mensagens não entregues (DLQ), se configurado, ou o descarta caso contrário. O processador de fluxo continua a processar documentos posteriores. Para o esquema de documento DLQ, consulte fila de mensagens não entregues (DLQ).

Processar resultados

Use pipeline para processar ainda mais os resultados da pesquisa antes que o Atlas Stream Processing os anexe ao documento de entrada. O Atlas Stream Processing encaminha pipeline para o MongoDB pesquisa vetorial literalmente. Para referenciar um valor do documento de entrada dentro de pipeline, passe-o por let e referencie-o como $$<variable>.

Exemplos

Os exemplos a seguir consultam o conjunto de dados sample_mflix, que contém dados sobre filmes e salas de cinema. Para executá-los, carregue os dados de amostra em seu cluster Atlas e crie um índice de pesquisa vetorial MongoDB na coleção embedded_movies.

O exemplo a seguir enriquece os eventos de pesquisa de entrada com recomendações de filmes. Cada documento de entrada tem um campo topic. Este pipeline usa query.text, e a pesquisa vetorial do MongoDB gera o embedding para o valor topic de cada documento usando o modelo voyage-4-lite. Esta agregação tem três estágios:

O estágio $source estabelece uma conexão com o banco de dados Atlas, visando especificamente a coleção searches no banco de dados app. A opção fullDocumentOnly requer que as imagens pré e pós-imagem sejam habilitadas na coleção de origem.
O estágio $vectorSearch query o índice plot_vector_index na coleção sample_mflix.embedded_movies usando o texto no campo topic de cada documento e anexa até cinco resultados ao campo recommendations. O campo pipeline limita cada resultado aos seus campos title e plot.
O estágio $merge grava o documento enriquecido na coleção enriched_searches no banco de dados 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"    }  }}

Os documentos enriquecidos gravados na coleção de saída são semelhantes aos seguintes:

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

O exemplo a seguir recupera passagens de enredo de filme de alta precisão para um pipeline RAG. Como cada evento de entrada já inclui um campo queryEmbedding pré-calculado, o pipeline usa queryVector. O estágio $vectorSearch recupera um amplo conjunto de passagens candidatas para maximizar o recall, e o estágio $rerank em pipeline reordena esses candidatos para que apenas as passagens mais relevantes continuem downstream. Esta agregação tem quatro etapas:

O estágio $source estabelece uma conexão com o banco de dados Atlas, visando especificamente a coleção events no banco de dados llm_requests. A opção fullDocumentOnly requer que as imagens pré e pós-imagem sejam habilitadas na coleção de origem.
O estágio $vectorSearch query o índice plot_embedding_index na coleção sample_mflix.embedded_movies usando o campo queryEmbedding de cada documento como o queryVector. Ele define numCandidates alto para maximizar a recuperação e passa até 50 correspondências para pipeline. Dentro de pipeline:
- $rerank reordena os candidatos por relevância ao texto da query usando o modelo rerank-2.5-lite.
- $limit mantém as oito passagens mais relevantes.
- $project retorna apenas os campos necessários para criar o prompt.
O valor query.text referencia a variável searchText definida em let, que se resolve para o campo queryText de cada documento. O Atlas Stream Processing armazena os resultados no campo context.
O estágio $unset remove os campos queryEmbedding e queryText, pois não são mais necessários a jusante.
O estágio $emit envia o documento enriquecido para o tópico qa_enriched.

Importante

Confirme se o administrador do projeto Atlas habilitou o Native Reranking antes de executar um pipeline que usa $rerank.

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

Os documentos emitidos para o tópico são semelhantes aos seguintes:

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

Voltar

$iceberg

Segurança

$vectorSearch Estágio de agregação