`$vectorSearch` etapa de agregación

Definición

La etapa $vectorSearch enriquece cada documento de transmisión con los resultados de una query de búsqueda vectorial de MongoDB. Para cada documento de entrada, la etapa ejecuta una query $vectorSearch en una colección y adjunta los resultados a un campo que especifique.

Ubicación

$vectorSearch debe ser una etapa intermedia de tu pipeline. No lo utilices como $source o receptor de tu pipeline.

Sintaxis

La etapa de pipeline $vectorSearch tiene la siguiente forma de prototipo:

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

La etapa $vectorSearch procesa un documento con los siguientes campos:

Campo	Tipo	Necesidad	Descripción
`from`	Documento	Requerido	Documento que especifica la colección de Atlas de destino que `$vectorSearch` query para cada documento de entrada.
`from.connectionName`	string	Requerido	Nombre de una conexión de Atlas en su Registro de Conexiones. Atlas Stream Processing rechaza cualquier otro tipo de conexión en el momento del análisis.
`from.db`	string literal	Requerido	Nombre de la base de datos de destino.
`from.coll`	string literal	Requerido	Nombre de la colección de destino que contiene el índice de búsqueda vectorial de MongoDB.
`as`	string	Requerido	Campo de salida en el documento de transmisión para el arreglo de resultados de la búsqueda. Si la búsqueda vectorial de MongoDB no devuelve ningún documento coincidente, Atlas Stream Processing establece este campo en un arreglo vacío. ADVERTENCIA: si este campo está vacío o choca con el nombre del campo de metadatos de la transmisión, Atlas Stream Processing informa de un error `StreamProcessorInvalidOptions` al inicio.
`queryVector`	arreglo \| expresión de ruta de campo	Condicional	Incrustación vectorial explícita para buscar documentos coincidentes.
`query`	Documento	Condicional	Documento que especifica la query de texto para la incrustación automática.
`query.text`	string \| expresión de ruta de campo	Condicional	Query de texto para usar para la incrustación automática.
`model`	string	Opcional	Modelo de incrustación que utiliza la búsqueda vectorial de MongoDB para convertir `query.text` en un vector. Válido solo junto con `query`. Cuando omite `model`, la búsqueda vectorial de MongoDB utiliza el modelo configurado en la definición del índice de búsqueda vectorial de MongoDB de destino.
`index`	string	Requerido	Nombre del índice de MongoDB búsqueda vectorial para la query.
`path`	string	Requerido	Campo en la colección de destino que contiene las incrustaciones vectoriales para buscar.
`numCandidates`	Int	Condicional	Número de candidatos que la búsqueda vectorial de MongoDB considera durante la búsqueda. Obligatorio a menos que `exact` sea `true`. Para aprender cómo elegir un valor, consulte Selección de numCandidates.
`limit`	Int	Requerido	Número máximo de resultados a devolver. No debe exceder `numCandidates`. ADVERTENCIA: Si `limit` supera `numCandidates`, cada documento falla en tiempo de ejecución con un error `PlanExecutor`.
`filter`	Documento	Opcional	Expresión de prefiltro que se aplicará antes de la búsqueda vectorial. Para conocer los operadores compatibles, consulte Filtro. ADVERTENCIA: Si `filter` utiliza un operador no compatible, todos los documentos fallarán en tiempo de ejecución con un error `PlanExecutor`.
`exact`	booleano	Opcional	Indicador que especifica si se debe ejecutar una búsqueda de ANN o ENN. Proporcione este campo si omite `numCandidates`. El valor puede ser uno de los siguientes: `false` para ejecutar la búsqueda de ANN `true` para ejecutar ENN search Se establece por defecto en `false`.
`searchNodePreference`	Documento	Opcional	Documento con un campo `key` que Atlas Stream Processing utiliza para enrutar la query a un nodo de búsqueda específico. El valor `key` debe ser una string no vacía.
`pipeline`	arreglo	Opcional	Etapas de agregación adicionales que Atlas Stream Processing ejecuta en los resultados de la búsqueda, antes de adjuntarlos al documento de entrada.
`let`	Documento	Opcional	Variables que Atlas Stream Processing evalúa con cada documento de entrada. Cada valor puede ser un literal, una ruta de campo o una expresión. Las etapas `pipeline` pueden hacer referencia a los valores resueltos como `$$<variable>`.

Comportamiento

Modos de query

$vectorSearch admite dos modos de query mutuamente excluyentes: queryVector para incrustación de vectores y query.text para incrustación automática. En cualquier modo, puede especificar un valor literal o una expresión de ruta de campo, de modo que un único procesador de transmisión pueda emitir una query diferente para cada documento de entrada.

Error Handling

Si el índice de búsqueda vectorial de MongoDB de destino no existe, el procesador de transmisión entra en un estado de error cuando $vectorSearch se ejecuta por primera vez. Atlas Stream Processing informa de un error MongoServerError y deja de procesar documentos.

Si queryVector o query.text evalúa el tipo incorrecto en tiempo de ejecución, Atlas Stream Processing solo enruta el documento infractor a la fila de letra muerta, si está configurado, o lo descarta de lo contrario. El procesador de transmisión continúa procesando documentos posteriores. Para el esquema de documentos DLQ, consulte fila de letra muerta.

Procesar resultados

Utilice pipeline para procesar aún más los resultados de la búsqueda antes de que Atlas Stream Processing los adjunte al documento de entrada. Atlas Stream Processing reenvía pipeline a MongoDB búsqueda vectorial textualmente. Para hacer referencia a un valor del documento de entrada dentro de pipeline, páselo a través de let y haga referencia a él como $$<variable>.

Ejemplos

Los siguientes ejemplos consultan el conjunto de datos sample_mflix, que contiene datos sobre películas y cines. Para ejecutarlos, cargue los datos de muestra en su clúster de Atlas y cree un índice de búsqueda vectorial de MongoDB en la colección embedded_movies.

Enriquecimiento de búsqueda semántica con incrustación automática

El siguiente ejemplo enriquece los eventos de búsqueda entrantes con recomendaciones de películas. Cada documento de entrada tiene un campo topic. Esta pipeline utiliza query.text, y la búsqueda vectorial de MongoDB genera la incrustación para el valor topic de cada documento utilizando el modelo voyage-4-lite. Esta agregación tiene tres etapas:

La etapa $source establece una conexión con la base de datos de Atlas, específicamente dirigida a la colección searches de la base de datos app. La opción fullDocumentOnly requiere que las imágenes previas y posteriores estén habilitadas en la colección de origen.
La etapa $vectorSearch consulta el índice plot_vector_index en la colección sample_mflix.embedded_movies utilizando el texto en el campo topic de cada documento y adjunta hasta cinco resultados al campo recommendations. El campo pipeline limita cada resultado a sus campos title y plot.
La etapa $merge guarda el documento enriquecido en la colección enriched_searches en la base de datos 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"
    }
  }
}

Los documentos enriquecidos escritos en la colección de salida se parecen a los siguientes:

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

Recuperación de contexto RAG de precisión crítica con $rerank

El siguiente ejemplo recupera pasajes de trama de películas de alta precisión para un pipeline RAG. Debido a que cada evento entrante ya incluye un campo queryEmbedding precalculado, el pipeline utiliza queryVector. La etapa $vectorSearch recupera un amplio conjunto de pasajes candidatos para maximizar la recuperación, y la etapa $rerank en pipeline reordena esos candidatos para que solo los pasajes más relevantes continúen en sentido descendente. Esta agregación tiene cuatro etapas:

La etapa $source establece una conexión con la base de datos de Atlas, específicamente dirigida a la colección events de la base de datos llm_requests. La opción fullDocumentOnly requiere que las imágenes previas y posteriores estén habilitadas en la colección de origen.
La etapa $vectorSearch query el índice plot_embedding_index en la colección sample_mflix.embedded_movies utilizando el campo queryEmbedding de cada documento como queryVector. Establece numCandidates alto para maximizar la recuperación y pasa hasta 50 coincidencias a pipeline. Dentro de pipeline:
- $rerank reordena los candidatos por relevancia para el texto de la query mediante el modelo rerank-2.5-lite.
- $limit mantiene los ocho pasajes más relevantes.
- $project devuelve solo los campos necesarios para compilar el prompt.
El valor query.text hace referencia a la variable searchText definida en let, que se resuelve en el campo queryText de cada documento. Atlas Stream Processing almacena los resultados en el campo context.
La etapa $unset remueve los campos queryEmbedding y queryText, ya que ya no son necesarios en el flujo descendente.
La etapa $emit envía el documento enriquecido al tema qa_enriched.

Importante

Confirme que el administrador de su proyecto de Atlas haya habilitado la reclasificación nativa antes de ejecutar un pipeline que utilice $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"
  }
}

Los documentos emitidos al tema se parecen a los siguientes:

{
  _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 etapa de agregación