リアルタイム メディアの推奨

データソースを設定してください。

このソリューションでは、同じMongoDBクラスターの newsデータベースにある次の 2 つのコレクションを使用します。

{
  "_id": {
    "$oid": "696493bfbc1084032ac0adfe"
  },
  "title": "Ukraine updates, Day 6: ‘We are sacrificing our lives for freedom,’ Zelenskyy gets standing ovation after speech to European parliament",
  "link": "https://nationalpost.com/news/world/ukraine-updates-day-6-russia-kyiv",
  "keywords": null,
  "creator": [
    "National Post Wire Services"
  ],
  "video_url": null,
  "description": "Russia escalated shelling overnight of key cities in Ukraine as its troops on the ground move slowly in a large convoy toward the capital, Kyiv",
  "content": "8:20 a.m. EST — Ukraine's Zelenskyy tells EU: 'Prove that you are with us\" Read More",
  "pubDate": "2022-03-01 13:45:04",
  "expire_at": "Wed, 07 Sep 2022 13:45:04 GMT",
  "image_url": null,
  "source_id": "nationalpost",
  "country": [
    "canada"
  ],
  "category": [
    "top"
  ],
  "language": "english"
}

{
  "_id": {
    "$oid": "696a1ecd66a51be18fffb8fa"
  },
  "user_id": "user-2",
  "session_id": "sess-6a0d8837-a5f9-4ef5-8b00-78e9bf7a825c",
  "timestamp": {
    "$date": "2026-01-16T16:49:41.208Z"
  },
  "event_type": "read",
  "article_id": {
    "$oid": "696493e6bc1084032ac116ed"
  },
  "device": "desktop",
  "metadata": {
    "time_on_page": 54,
    "referral": "https://guzman.com/main/search/listmain.jsp"
  }
}

Stream Processing ワークスペースを作成します。

1. Atlas で、プロジェクトの [Stream Processing] ページに移動します。

   1. まだ表示されていない場合は、プロジェクトを含む組織をナビゲーション バーの [ Organizations ] メニューで選択します。

   2. まだ表示されていない場合は、ナビゲーション バーの Projects メニューからプロジェクトを選択します。

   3. サイドバーで、 Streaming Data見出しの下のStream Processingをクリックします。

      Atlas Stream Processingページが表示されます。

2. [Create a workspace] をクリックします。

3. Create a stream processing workspaceページで、ワークスペースを次のように設定します。

   • Tier: SP30

   • Provider: AWS

   • Region: us-east-1

   • Workspace Name: article-personalization

クリックストリームと記事データの接続を追加します。

ストリーム プロセッサは、エンリッチメントを実行するために、クリックストリーム データと記事メタデータの両方に接続する必要があります。両方のデータソースは同じ Atlas クラスターにある必要があります。

1. Stream Processing ワークスペースのペインで、Manage をクリックします。

2. [ Connection Registryタブで、右上の [ + Add Connection ] をクリックします。

3. Edit Connectionページで、接続を次のように構成します。

   • 接続タイプ：Atlas Database

   • 接続名: userevents

   • Atlas クラスター: クリックストリームと記事データが存在するクラスターの名前（例: ClickstreamCluster）

   • 次として実行：Read and write to any database

4. Save changes をクリックして接続を作成します。

永続的なストリーム プロセッサを作成します。

userIntentSummarizer　というストリームプロセッサと、未加工のクリックストリームイベントを user_events コレクションから読み取り、記事メタデータでイベントを強化するステージを作成します。

1. Atlasプロジェクトの Stream Processing ページで、Stream Processing ワークスペースのペインにあるManage をクリックします。

2. JSON editor で、次のJSON定義をコピーしてJSONエディターのテキストボックスに貼り付け、これらのステージを持つ userIntentSummarizer という名前のストリーム プロセッサを定義します。

   • $source: 接続されたクリックストリームクラスター（userevents 接続）の news データベースにある user_events コレクションから未加工のクリックストリームイベントを読み取ります。

   • $lookup:article_id フィールドに基づいてarticles コレクションと生のクリックストリームイベントを結合し、description 、keywords 、title フィールドから関連する記事のメタデータを取り込みます。

   • $addFields:article_details フィールドから description 、keywords 、title フィールドをイベントストリームのトップレベルに投影し、下流ステージから簡単にアクセスできるようにします。

   • $project: 下流プロセシングの関連フィールドを投影します。

   {
     "$source": {
       "connectionName": "userevents",
       "db": "news",
       "collection": "user_events"
     }
   },
   {
     "$lookup": {
       "from": {
         "connectionName": "userevents",
         "db": "news",
         "coll": "articles"
       },
       "localField": "article_id",
       "foreignField": "_id",
       "as": "article_details",
       "pipeline": [
         {
           "$project": {
             "_id": 0,
             "description": 1,
             "keywords": 1,
             "title": 1
           }
         }
       ]
     }
   },
   {
     "$addFields": {
       "description": { "$arrayElemAt": [ "$article_details.description", 0 ] },
       "keywords":    { "$arrayElemAt": [ "$article_details.keywords", 0 ] },
       "title":       { "$arrayElemAt": [ "$article_details.title", 0 ] }
     }
   },
   {
     "$project": {
       "userId": 1,
       "article_id": 1,
       "eventTime": 1,
       "event_type": 1,
       "device": 1,
       "session_id": 1,
       "description": 1,
       "keywords": 1,
       "title": 1
     }
   },

3. 変更を保存するには、[Update stream processor] をクリックします。

LLM プロバイダーをStream Processingワークスペースに接続します。

外部 HTTPS 接続を LLM プロバイダー（例: Azure OpenAI）に追加して、ストリームプロセッサを有効にし、パイプラインから LLM を直接呼び出してデータを強化します。

1. Stream Processing ワークスペースのペインで、Manage をクリックします。

2. [ Connection Registryタブで、右上の [ + Add Connection ] をクリックします。

3. 次のように接続を構成します。

   • 接続タイプ：HTTPS

   • 接続名: azureopenai

   • URL: Azure OpenAIインスタンスのエンドポイントURL

   • ヘッダー: これらのキーと値のペアをヘッダーに追加します。

     • キー: Content-Type、値: application/json

     • キー: api-key、値: Azure OpenAI APIキー

$sessionWindowステージを使用して、クリックストリーム データをセッション化します。

$sessionWindow ステージをストリームプロセッサ パイプラインに追加して、特定のセッションギャップに基づいて関連するイベントをセッションにグループ化します。このソリューションでは、セッションを60秒を超える非アクティブギャップのない同じ session_id からのイベントのシーケンスとして定義します。

このステージを、エンリッチメントステージの後に userIntentSummarizer パイプラインに追加します。

{
  "$sessionWindow": {
    "partitionBy": "$session_id",
    "gap": { "unit": "second", "size": 60 },
    "pipeline": [{
      "$group": {
        "_id": "$session_id", "titles": {
          "$push": "$title"
        }
      }
    }]
  }
}

$https ステージを使用して、ユーザーセッションを要約します。

$https ステージを追加して、Stream Processing パイプラインから LLM プロバイダーを直接呼び出します。このソリューションは Azure OpenAI を呼び出して、セッションの記事のタイトルに基づいてユーザーの関心を説明する各セッションの自然言語要約を生成します。

$sessionWindowステージの後に、このステージをパイプラインに追加します。

{
  "$https": {
    "connectionName": "azureopenai",
    "method": "POST",
    "as": "apiResults",
    "config": { "parseJsonStrings": true },
    "payload": [
      {
        "$project": {
          "_id": 0,
          "model": "gpt-4o-mini",
          "response_format": { "type": "json_object" },
          "messages": [
            {
                "role": "system",
                "content": "You are an analytical assistant that specializes in behavioral summarization. You analyze short-term reading activity and infer user interests without making personal or sensitive assumptions the create a special field called summary. Summary must be a special field in the response. Respond only in JSON format.Return a JSON object with the following keys in this order: \n reasoning: (Internal scratchpad, briefly explain your analysis) \n user_interests: (The list of inferred interests) \n summary: (A concise summary based on the interests above)"
            },
            { 
              "role": "user", 
              "content": { "$toString": "$titles" } 
            }
          ]
        }
      }
    ]
  }
}

セッションサマリーを新しいコレクションに書き込みます。

次のステージをパイプラインに追加して、LLM 出力からサマリーを抽出し、それを新しいコレクションに書き込みます。

• $match: LLM 呼び出しが失敗しエラーが返されたセッションを除外し、データベースへの不完全なデータの書き込みを回避します。

• $addFields: LLMの出力からsummaryフィールドを抽出し、ドキュメントの最上位に追加します。

• $project: ドキュメントから未加工の LLM 出力を削除し、ノイズとストレージのコストを削減します。

• $merge：結果のドキュメントをクリックストリーム クラスター（userevents 接続）の news データベースにある user_intent と言う新しいコレクションに書き込みます。このコレクションの各ドキュメントはユーザーセッションを表し、ユーザーの興味の要約が含まれています。

{
    "$match": {
        "titles": {
            "$exists": true
        },
        "apiResults": {
            "$exists": true
        }
    }
},
{
    "$addFields": {
        "summary": {
            "$arrayElemAt": [
                "$apiResults.choices.message.content.summary",
                0
            ]
        }
    }
},
{
    "$project": {
        "apiResults": 0
    }
},
{
    "$merge": {
        "into": {
            "coll": "user_intent",
            "connectionName": "userevents",
            "db": "news"
        }
    }
}

ストリーム プロセッサを起動します。

クリックストリームデータを要約する準備ができたら、Stream Processing ワークスペースのストリームプロセッサのリストにある userIntentSummarizer プロセッサの Start アイコンをクリックします。

このパイプラインは、ユーザーの関心を捉えたセッション要約を含むドキュメントを user_intent コレクションに書き込みます。たとえば、

{
 "_id": "sess-6a0d8837-a5f9-4ef5-8b00-78e9bf7a825c",
 "summary": "The user seems interested in geopolitical developments, especially in the Middle East, US political strategies involving Trump, and legal aspects of government operations.",
 "titles": [
   "Israel and Hamas agree to part of Trump's Gaza peace plan, will free hostages and prisoners",
   "Top officials from US and Qatar join talks aimed at brokering peace in Gaza",
   "How Trump secured a Gaza breakthrough",
   "Ontario's anti-tariff ad is clever, effective and legally sound, experts say",
   "Shutdown? Trump's been dismantling the government all year",
   "AP News Summary at 7:58 p.m. EDT"
 ]
}

セマンティック検索用に記事データを準備します。

セマンティック検索を実行する前に、記事データのベクトル埋め込みを生成する必要があります。これを行うには、articles コレクションの description フィールドを autoEmbed タイプとしてインデックス化する vector_index という MongoDB Vector Search インデックスを作成します。これは、MongoDB Vector Search に自動埋め込みを使用して、コレクションにドキュメントが挿入または更新されるたびに description フィールドのベクトル埋め込みを自動的に生成するように指示します。

このJSON定義を使用して、これらのフィールドにベクトルインデックスを作成します。

• autoEmbed タイプとしての description フィールドは、ドキュメントがコレクションに挿入された場合や更新された場合に、MongoDB Vector Search にvoyage-4-large 埋め込みモデルを使用して、description フィールドのベクトル埋め込みを自動生成するよう指示します。

• title フィールドを filter タイプとして使用し、フィールドの string 値を使用してセマンティック検索のデータを前処理します。これにより、ユーザーがすでに読んだ記事を除外できます。

{
  "fields": [
    {
      "type": "autoEmbed",
      "modalitytype": "text",
      "path": "description",
      "model": "voyage-4-large"
    },
    {
      "type": "filter",
      "path": "title"
    }
  ]
}

セマンティック検索クエリを実行して、パーソナライズされた推薦を提供します。

ユーザーがサイトにアクセスすると、現在のセッションの概要を取得し、クエリとして使用して記事のカタログにたいしベクトル検索を実行します。インデックスで自動埋め込みを有効にしたため、MongoDB Vector Search はクエリ時に概要の埋め込みを自動的に生成し、それを有効なクエリベクトルとして使用します。

この例は、セッション summary をクエリベクトルとして使用し、titles フィールドに基づいてユーザーがすでに読んだ記事を除外する簡素化されたベクトル検索を示しています。

[{
    "$vectorSearch": {
      "index": "vector_index", // Vector index with autoEmbed on article descriptions
      "path": "description",
      "query": {
        "text": "<session-summary>" // Session summary from user_intent document
      },
      "numCandidates": 100,
      "filter": { 
        "title": { "$nin": [<read-titles>] } // Exclude articles the array of titles in the user_intent document
      }
    }
}]