Recomendaciones de medios en tiempo real con Atlas Stream Processing y MongoDB Vector Search

Configura las fuentes de datos.

Esta solución utiliza las siguientes dos colecciones en la base de datos news del mismo clúster de MongoDB:

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

Cree un espacio de trabajo de Stream Processing.

1. En Atlas, se debe ir a la página Stream Processing del proyecto.

   1. Si aún no se muestra, seleccione la organización que contiene su proyecto en el menú Organizaciones de la barra de navegación.

   2. Si aún no se muestra, seleccione su proyecto en el menú Projects de la barra de navegación.

   3. En la barra lateral, haz clic en Stream Processing en la sección Streaming Data.

      Se muestra la página Stream Processing.

2. Haga clic en Create a workspace.

3. En la página Create a stream processing workspace, configura tu espacio de trabajo de la siguiente manera:

   • Tier: SP30

   • Provider: AWS

   • Region: us-east-1

   • Workspace Name: article-personalization

Agrega una conexión para tu datos de flujo de clics y de artículos.

El procesador de flujo necesita conexiones tanto a los datos de flujo de clics como a los metadatos de los artículos para realizar el enriquecimiento. Ambas fuentes de datos deben estar en el mismo clúster de Atlas.

1. En el panel de tu espacio de trabajo de procesamiento de flujos, haz clic en Manage.

2. En la pestaña Connection Registry, haz clic en + Add Connection en la parte superior derecha.

3. En la página Edit Connection, configura tu conexión de la siguiente manera:

   • Tipo de conexión: Atlas Database

   • Nombre de la conexión: userevents

   • Clúster de Atlas: El nombre del clúster en el que se encuentran los datos del artículo y del flujo de clics (por ejemplo, ClickstreamCluster)

   • Ejecutar como: Read and write to any database

4. Haz clic en Save changes para crear la conexión.

Crea un procesador de flujo persistente.

Cree un procesador de streaming llamado userIntentSummarizer con etapas que lean los eventos de flujo de clics sin procesar de la colección user_events y enriquezcan los eventos con los metadatos de los artículos.

1. En la página Stream Processing de tu proyecto de Atlas, haz clic en Manage en el panel del espacio de trabajo de Stream Processing.

2. En el JSON editor, copia y pega la siguiente definición JSON en el cuadro de texto del editor JSON para definir un procesador de flujo llamado userIntentSummarizer con estas etapas:

   • $source: Lee eventos de flujo de clics sin procesar de la colección user_events de la base de datos news de tu clúster de flujo de clics conectado (conexión userevents).

   • $lookup: Une los eventos de secuencia de clics sin procesar con la colección articles basándose en el campo article_id, incorporando metadatos relevantes del artículo de los campos description, keywords y title.

   • $addFields: Proyecta los campos description, keywords y title desde el campo article_details al nivel superior del flujo de eventos para que sean fácilmente accesibles para las etapas posteriores.

   • $project: Proyecta los campos pertinentes para el procesamiento posterior.

   {
     "$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. Haz clic en Update stream processor para guardar tus cambios.

Conecta tu proveedor de LLM a tu espacio de trabajo de Stream Processing.

Agrega una conexión HTTPS externa a tu proveedor de LLM (por ejemplo, Azure OpenAI) para permitir que el procesador de flujo llame al LLM directamente desde el pipeline para enriquecer tus datos:

1. En el panel de tu espacio de trabajo de procesamiento de flujos, haz clic en Manage.

2. En la pestaña Connection Registry, haz clic en + Add Connection en la parte superior derecha.

3. Configura la conexión de la siguiente manera:

   • Tipo de conexión: HTTPS

   • Nombre de la conexión: azureopenai

   • URL: La URL del endpoint para la instancia de Azure OpenAI.

   • Encabezados: Añade estos pares clave-valor a los encabezados:

     • Clave: Content-Type, Valor: application/json

     • Clave: api-key, Valor: Tu clave de API de Azure OpenAI

Separe los datos de flujo de clics utilizando la $sessionWindow etapa.

Agrega una etapa $sessionWindow al pipeline del procesador de flujos para agrupar eventos relacionados en sesiones en función de un intervalo de sesión especificado. Esta solución define una sesión como una secuencia de eventos del mismo session_id sin un intervalo de inactividad superior a 60 segundos.

Agrega esta etapa a tu pipeline userIntentSummarizer después de las etapas de enriquecimiento:

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

Resuma las sesiones de usuario utilizando la $https etapa.

Agrega una etapa $https para llamar directamente a tu proveedor de LLM desde tu pipeline de Stream Processing. Esta solución llama a Azure OpenAI para generar un resumen en lenguaje natural de cada sesión que describe los intereses del usuario basándose en los títulos de los artículos de la sesión.

Agrega esta etapa al pipeline después de la etapa $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" } 
            }
          ]
        }
      }
    ]
  }
}

Guarda los resúmenes de sesión en una nueva colección.

Agrega las siguientes etapas al pipeline para extraer el resumen de la salida del LLM y guardarlo en una nueva colección:

• $match: Filtra las sesiones en las que la llamada al LLM falló y devolvió un error para evitar escribir datos incompletos en la base de datos.

• $addFields: Extrae el campo summary de la salida del LLM y lo agrega al nivel superior del documento.

• $project: Remueve la salida LLM sin procesar del documento para reducir el ruido y los costos de almacenamiento.

• $merge: Guarda los documentos resultantes en una nueva colección llamada user_intent en la base de datos news del clúster de flujo de clics (conexión userevents). Cada documento de esta colección representa una sesión de usuario y contiene un resumen de los intereses del usuario.

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

Inicia el procesador de flujos.

Cuando estés listo para empezar a resumir los datos de flujo de clics, haz clic en el icono Start del procesador userIntentSummarizer en la lista de procesadores de streaming del espacio de trabajo de Stream Processing.

Este pipeline debe guardar documentos para la colección de user_intent que contengan resúmenes de sesión que capturen los intereses del usuario. Por ejemplo:

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

Prepara los datos del artículo para la recuperación semántica.

Antes de poder realizar búsquedas semánticas, debes generar incrustaciones vectoriales para los datos de artículos. Para ello, crea un índice de MongoDB Vector Search denominado vector_index que indexe el campo description de la colección articles como el tipo autoEmbed. Esto indica a MongoDB Vector Search que utilice incrustaciones automatizadas para generar automáticamente incrustaciones vectoriales para el campo description cada vez que se insertan o actualizan documentos en la colección.

Utiliza esta definición JSON para crear un índice vectorial en estos campos:

• El campo description con el tipo autoEmbed para instruir a MongoDB Vector Search que genere automáticamente incrustaciones vectoriales para el campo description utilizando el modelo de incrustación voyage-4-large cada vez que se inserten o actualicen documentos en la colección.

• El campo title como el tipo filter para prefiltrar los datos para la búsqueda semántica utilizando el valor de string en el campo. Esto te permite excluir artículos que el usuario ya ha leído de los resultados de búsqueda.

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

Ejecuta queries de búsqueda semántica para servir recomendaciones personalizadas.

Cuando un usuario visita tu sitio, toma el resumen de su sesión actual y lo usa como la query para una búsqueda vectorial contra tu catálogo de artículos. Como se habilitó la incrustación automatizada en el índice, MongoDB Vector Search genera automáticamente la incrustación para el resumen en el momento de la query y la usa como el vector de query efectivo.

Este ejemplo muestra una query de búsqueda vectorial simplificada que utiliza la sesión summary como vector de query y excluye los artículos que el usuario ya ha leído en función del campo 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
      }
    }
}]