/ /

`$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`	string	必須	接続レジストリ内の Atlas 接続の名前。Atlas Stream Processing は、パース時にその他の接続タイプを拒否します。
`from.db`	literal string	必須	ターゲットデータベースの名前。
`from.coll`	literal string	必須	MongoDB Vector Search インデックスを含むターゲットコレクションの名前。
`as`	string	必須	検索結果配列のストリーミングドキュメントの出力フィールド。MongoDB ベクトル検索が一致するドキュメントを返さない場合、Atlas Stream Processing はこのフィールドを空の配列に設定します。警告: このフィールドが空の場合、またはストリームメタデータフィールド名と衝突する場合、Atlas Stream Processing は開始時に `StreamProcessorInvalidOptions` エラーを報告します。
`queryVector`	配列 \| フィールドパス式	条件付き	一致するドキュメントを検索するための明示的なベクトル埋め込み。
`query`	ドキュメント	条件付き	自動埋め込み用のテキストクエリを指定するドキュメント。
`query.text`	string \| フィールドパス式	条件付き	自動埋め込みに使用するクエリテキスト。
`model`	string	任意	MongoDB ベクトル検索が `query.text` をベクトルに変換するために使用する埋め込みモデル。`query` と一緒にのみ有効です。 `model`を省略すると、MongoDB ベクトル検索は、移行先 MongoDB ベクトル検索インデックスの定義で構成されたモデルを使用します。
`index`	string	必須	クエリする MongoDB ベクトル検索インデックスの名前。
`path`	string	必須	検索対象となるベクトル埋め込みを含むターゲットコレクションのフィールド。
`numCandidates`	整数	条件付き	検索中にMongoDB ベクトル検索が考慮する候補の数。`exact` が`true` でない限り、必須です。値を選択する方法については、「 numKeys の選択」を参照してください。
`limit`	整数	必須	返される結果の最大数。`numCandidates` を超えてはなりません。警告: `limit` が `numCandidates` を超えると、すべてのドキュメントはランタイム時に `PlanExecutor` エラーで失敗します。
`filter`	ドキュメント	任意	ベクトル検索の前に適用するプレフィルター式。サポートされている演算子については、「フィルター」を参照してください。警告: `filter` でサポートされていない演算子を使用すると、すべてのドキュメントでランタイム時に `PlanExecutor` エラーが発生します。
`exact`	ブール値	任意	Ann 検索または ENN 検索のどちらを実行するかを指定するフラグ。`numCandidates` を省略する場合は、このフィールドを指定します。値は次のいずれかになります。 `false` は、Ann 検索を実行するために `true` : ENN 検索を実行デフォルトは `false` です。
`searchNodePreference`	ドキュメント	任意	Atlas Stream Processing がクエリを特定の検索ノードにルーティングするために使用する `key` フィールドを含むドキュメント。`key` 値は空ではない string である必要があります。
`pipeline`	配列	任意	Atlas Stream Processing が検索結果に対して実行する追加の集計ステージ。これらは入力ドキュメントに接続される前に実行されます。
`let`	ドキュメント	任意	Atlas Stream Processing が各入力ドキュメントに対して評価する変数。各値は、リテラル、フィールドパス、または式ことができます。`pipeline` ステージは解決された値を`$$<variable>` として参照できます。

動作

クエリモード

$vectorSearch queryVectorは、2 query.textつの相互に排他的なクエリモードをサポートします。ベクトル埋め込みでは、自動埋め込みではです。モードのどちらでも、リテラル値またはフィールドパス式を指定できるため、単一のストリームプロセッサが入力ドキュメントごとに異なるクエリを発行できます。

Error Handling

対象の MongoDB ベクトル検索インデックスが存在しない場合、$vectorSearch が初めて実行されると、ストリームプロセッサーはエラー状態になります。Atlas Stream Processing は MongoServerError エラーを報告し、ドキュメントのプロセシングを停止します。

queryVectorquery.textまたはが実行時に間違ったタイプと評価された場合、Atlas Stream Processing は、競合ドキュメントのみをデッドレターキュー(DLQ)にルーティングするか、そうでない場合は削除します。ストリームプロセッサは、後のドキュメントの処理を続行します。 DLQドキュメントスキーマについては、「デッド文字キュー」を参照してください。

結果の処理

Atlas Stream Processing が入力ドキュメントに接続する前に、pipeline を使用して検索結果をさらに処理します。Atlas Stream Processing は pipeline を MongoDB ベクトル検索にそのまま転送します。pipeline 内の入力ドキュメントから値を参照するには、let を介して渡し、$$<variable> として参照します。

例

次の例では、映画と映画館に関するデータが含まれる sample_mflix データセットをクエリします。これらを実行するには、サンプルデータを Atlas クラスターにロードし、コレクションにMongoDBembedded_movies ベクトル検索インデックスを作成します。

次の例では、受信した検索イベントを映画のおすすめで強化します。各入力ドキュメントには topic フィールドがあります。このパイプラインでは query.text を使用し、MongoDB ベクトル検索は voyage-4-lite モデルを使用して、各ドキュメントの topic 値の埋め込みを生成します。この集計には、次の 3 つのステージがあります。

$sourceステージでは、Atlas データベースとの接続を確立し、特定の app データベースの searches コレクションを対象とします。fullDocumentOnly オプションでは、ソースコレクションでプレイメージとポストイメージを有効にする必要があります。
$vectorSearch ステージは、各ドキュメントの topic フィールドのテキストを使用して、sample_mflix.embedded_movies コレクションの plot_vector_index インデックスをクエリし、最大 5 件の結果を 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.' }
  ]
}

次の例では、 RAGパイプラインの高精度の映画プロットパスを検索します。各受信イベントには事前計算されたqueryEmbedding フィールドがすでに含まれているため、パイプラインはを使用します。queryVector $vectorSearchステージでは、再現率を最大化するために候補パスのセットが検索され、$rerank のpipeline ステージでは、それらの候補の順序が再配置され、最も関連性の高いパスのみが下流に残ります。この集計には 4 つのステージがあります。

$sourceステージでは、Atlas データベースとの接続を確立し、特定の llm_requests データベースの events コレクションを対象とします。fullDocumentOnly オプションでは、ソースコレクションでプレイメージとポストイメージを有効にする必要があります。
$vectorSearchステージは、各ドキュメントの queryEmbedding フィールドを queryVector として使用して、sample_mflix.embedded_moviesコレクションの plot_embedding_indexインデックスをクエリします。これにより、numCandidates が高く設定されているため、再現率が最大化され、最大 50 件の一致が pipeline に渡されます。pipeline内:
- $rerankrerank-2.5-liteは、モデルを使用して、クエリテキストに関連することで候補を並べ替えます。
- $limit 最も関連性の高い 8 つのパッセージを保持します。
- $project プロンプトをビルドするのに必要なフィールドのみを返します。
query.text 値は、let で定義された searchText 変数を参照します。これは各ドキュメントの queryText フィールドに解決されます。Atlas Stream Processing は結果を context フィールドに保存します。
$unsetqueryEmbeddingqueryTextステージでは、下流で不要になったため、フィールドとフィールドが削除されます。
$emit ステージは、強化されたドキュメントを qa_enriched トピックに送信します。

重要

を使用するパイプラインを実行する前に、Atlasプロジェクト管理者がネイティブ再ランク付けを有効にしていることを確認してください。$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"
  }
}

トピックに出力されるドキュメントは次のとおりです。

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

戻る

$iceberg

セキュリティ

$vectorSearch 集計ステージ

定義

配置

構文

動作

クエリ モード

Error Handling

結果の処理

例

自動埋め込みによるセマンティック検索の強化

精度が重要な RAG コンテキスト検索と $rerank

`$vectorSearch` 集計ステージ

クエリモード