使用Atlas Stream Processing和MongoDB Vector Search 进行实时媒体推荐

设置数据源。

此解决方案使用同一MongoDB 集群的 news数据库中的以下两个集合：

{
  "_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中， Go项目的 Stream Processing 页面。

   1. 如果尚未显示，请从导航栏上的 Organizations 菜单中选择包含项目的组织。

   2. 如果尚未显示，请从导航栏的 Projects 菜单中选择您的项目。

   3. 在侧边栏中，单击 Streaming Data 标题下的 Stream Processing。

      此时将显示 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 Cluster：点击流和文章数据所在集群的名称（示例 ClickstreamCluster）

   • 执行方式：Read and write to any database

4. 单击 Save changes 以创建连接。

创建持久流处理器。

创建一个名为 userIntentSummarizer 的流处理器，其中包含从 user_events集合中读取原始点击流事件并使用文章元数据丰富这些事件的阶段。

1. 在Atlas项目的 Stream Processing 页面上，单击流处理工作区窗格中的Manage 。

2. 在JSON editor 中，将以下 JSON定义复制并粘贴到JSON编辑器文本框中，以定义具有以下阶段的名为 userIntentSummarizer的流处理器：

   • $sourceuser_eventsnews：从已连接的点击流集群（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提供商连接到流处理工作区。

将外部 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 阶段汇总用户会话。

添加 阶段以直接从流处理管道调用 LLM提供商。此解决方案调用Azure OpenAI$https 生成每个会话的自然语言摘要，根据会话中的文章标题描述用户的兴趣。

将此阶段添加到管道的$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 LLM 输出中提取 字段并将其添加到文档的顶层。

• $project：从文档中删除原始 LLM 输出，以减少噪音和存储成本。

• $mergeuser_intentnews：将生成的文档写入点击流集群（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"
        }
    }
}

启动流处理器。

当您准备好开始汇总点击流数据时，请在流处理工作区的流处理器列表中单击 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"
 ]
}

为语义检索准备文章数据。

在执行语义搜索之前，您需要为文章数据生成向量嵌入。为此，创建一个名为 的MongoDB Vector Search索引，将vector_index description集合的articles 字段索引为autoEmbed 类型。每当在集合中插入或更新文档时，这会指示MongoDB Vector Search 使用自动嵌入自动为description 字段生成向量嵌入。

使用此JSON定义在这些字段上创建向量索引：

• description字段作为 autoEmbed 类型，用于指示MongoDB Vector Search 每当在集合中插入或更新文档时，使用 voyage-4-large 嵌入模型自动为 description字段生成向量嵌入。

• title字段为 filter 类型，用于使用该字段中的字符串值对数据进行预筛选，以进行语义搜索。这允许您从搜索结果中排除用户已阅读的文章。

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