`$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`	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` でない限り必須です。値の選択方法については、「numCandidates の選択」を参照してください。
`limit`	整数	必須	返される結果の最大数。`numCandidates` を超えてはなりません。警告: `limit` が `numCandidates` を超えると、すべてのドキュメントはランタイム時に `PlanExecutor` エラーで失敗します。
`filter`	ドキュメント	任意	ベクトル検索の前に適用する事前フィルター式。サポートされている演算子については、フィルターを参照してください。警告: `filter` でサポートされていない演算子を使用すると、すべてのドキュメントでランタイム時に `PlanExecutor` エラーが発生します。
`exact`	ブール値	任意	近似最近傍探索検索または厳密最近傍探索検索のどちらを実行するかを指定するフラグ。`numCandidates` を省略する場合は、このフィールドを提供します。値は次のいずれかになります。 `false` 近似最近傍探索を実行するには `true` 厳密最近傍探索検索を実行するには、デフォルトは `false` です。
`searchNodePreference`	ドキュメント	任意	Atlas Stream Processing がクエリを特定の検索ノードにルーティングするために使用する `key` フィールドを含むドキュメント。`key` 値は空ではない string である必要があります。
`pipeline`	配列	任意	Atlas Stream Processing が検索結果に対して実行する追加の集計ステージ。これらは入力ドキュメントに接続される前に実行されます。
`let`	ドキュメント	任意	Atlas Stream Processing が各入力ドキュメントに対して評価する変数。各値は、リテラル、フィールドパス、または式であることができます。`pipeline` ステージは、解決された値を `$$<variable>` として参照できます。

動作

クエリモード

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

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 値の埋め込みを生成します。この集計には、次の 3 つのステージがあります。

$sourceステージでは、Atlas データベースとの接続を確立し、特定の app データベースの searches コレクションを対象とします。fullDocumentOnly オプションでは、ソースコレクションでプレイメージとポストイメージを有効にする必要があります。
$vectorSearch ステージは、各ドキュメントの topic フィールドのテキストを使用して、sample_mflix.embedded_movies コレクションの plot_vector_index インデックスをクエリし、最大 5 件の結果を 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 ステージはこれらの候補を再オーダーし、最も関連性の高いパッセージのみがダウンストリームに進みます。この集計は 4 段階で行われます。

$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 最も関連性の高い 8 つのパッセージを保持します。
- $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.' }    ...  ]}

戻る

$iceberg

セキュリティ

$vectorSearch 集計ステージ

定義

配置

構文

動作

クエリ モード

Error Handling

結果の処理

例

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

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

`$vectorSearch` 集計ステージ

クエリモード