リアルタイムメディアのおすすめ

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

このソリューションでは、同じ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 ページで、ストリーム処理ワークスペースのManage ペインにある [] をクリックします。

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

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

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

   • $addFields:description keywordstitlearticle_detailsフィールドから 、 、 フィールドをイベントストリームの最上位にプロジェクションし、下流のステージから簡単にアクセスできるようにします。

   • $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指定されたセッション ギャップに基づいて関連するイベントをセッションにグループ化するには、ストリーム プロセッサパイプラインに ステージを追加します。このソリューションでは、セッションは、非アクティブなギャップがsession_id 60秒より長い同じ からのイベントのシーケンスとしてセッションを定義します。

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

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

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

$httpsストリーム処理パイプラインから LM プロバイダーを直接呼び出すには、 ステージを追加します。このソリューションでは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:summary LM 出力から フィールドを抽出し、それをドキュメントの最上位に追加します。

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

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

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

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

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