/ /

`$vectorSearch` 集計ステージ

定義

$vectorSearchステージでは、各ストリーミングドキュメントがMongoDB ベクトル検索クエリの結果で豊富になります。入力ドキュメントごとに、ステージはコレクションに対して $vectorSearch クエリを実行し、指定されたフィールドに結果をアタッチします。

配置

$vectorSearch は、パイプラインの中間ステージである必要があります。パイプラインのまたは Sink$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 ベクトル検索インデックスを含むターゲットコレクションの名前。
`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` を超えてはなりません。警告: `limitnumCandidates`がを超えると、すべてのドキュメントは実行時に失敗し、`PlanExecutor` エラーが発生します。
`filter`	ドキュメント	任意	ベクトル検索の前に適用するプレフィルター式。サポートされている演算子については、「フィルター」を参照してください。警告: `filter`がサポートされていない演算子を使用する場合、すべてのドキュメントは実行時に失敗し、`PlanExecutor` エラーが発生します。
`exact`	ブール値	任意	Ann 検索または ENN 検索のどちらを実行するかを指定するフラグ。`numCandidates` を省略する場合は、このフィールドを指定します。値は次のいずれかになります。 `false` は、Ann 検索を実行するために `true` : ENN 検索を実行デフォルトは `false` です。
`searchNodePreference`	ドキュメント	任意	Atlas Stream Processing がクエリを特定の検索ノードにルーティングするために使用する `key`フィールドを持つドキュメント。 `key` の値は空でない文字列である必要があります。
`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 つのステージがあります。

ステージは $sourceAtlas データベースとの接続を確立し、具体的にはデータベース内のsearches コレクションを対象とします。appfullDocumentOnly オプションでは、ソースコレクションで変更前と変更後のイメージを有効にする必要があります。
$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.' }
  ]
}

$rerank による精度が必要な RG コンテキスト取得

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

ステージは $sourceAtlas データベースとの接続を確立し、具体的にはデータベース内のevents コレクションを対象とします。llm_requestsfullDocumentOnly オプションでは、ソースコレクションで変更前と変更後のイメージを有効にする必要があります。
$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ステージでは、下流で不要になったため、フィールドとフィールドが削除されます。
$it ステージは、エンタープライズなドキュメントを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 集計ステージ

定義

配置

構文

動作

クエリ モード