利用 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的流处理器：

   • $source：从所连接的点击流集群 (userevents 连接) 的 news 数据库中的 user_events 集合读取原始点击流事件。

   • $lookup: 基于 article_id 字段将原始点击流事件与 articles 集合进行关联，从 description、keywords 和 title 字段引入相关的文章元数据。

   • $addFields：将 description、keywords 和 title 字段从 article_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 提供商到流处理工作区。

添加与您的 LLM 提供商（例如 Azure OpenAI）之间的外部 HTTPS 连接，以便流处理器能够直接从管道调用 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 阶段，直接从流处理管道调用您的 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：将生成的文档写入到点击流集群 news 数据库（userevents 连接）中名为 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"
        }
    }
}

启动流处理器。

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

准备用于语义检索的文章数据。

在执行语义搜索之前，您需要为文章数据生成向量嵌入。为此，创建一个名为 vector_index 的 MongoDB 向量搜索索引，将 articles 集合的 description 字段索引为 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
      }
    }
}]