실시간 미디어 권장 사항

데이터 소스를 설정하세요.

이 솔루션은 동일한 MongoDB cluster 의 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 Workspace을 생성합니다.

1. Atlas에서 프로젝트의 Stream Processing 페이지로 이동합니다.

   1. 아직 표시되지 않은 경우 탐색 표시줄의 조직 메뉴에서 프로젝트 포함된 조직 선택합니다.

   2. 아직 표시되지 않은 경우 내비게이션 바의 Projects 메뉴에서 프로젝트를 선택합니다.

   3. 사이드바에서 Streaming Data 제목 아래의 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. 스트림 처리 작업 공간의 창에서 Manage를 클릭하세요.

2. Connection Registry 탭에서 오른쪽 상단의 + Add Connection 을 클릭합니다.

3. Edit Connection 페이지에서 연결을 다음과 같이 설정합니다.

   • 연결 유형: Atlas Database

   • 연결 이름: userevents

   • Atlas 클러스터: 클릭스트림과 기사 데이터가 위치한 클러스터의 이름(예시: ClickstreamCluster)

   • 다음으로 실행: Read and write to any database

4. Save changes를 클릭하여 연결을 만듭니다.

영구 스트림 프로세서를 생성합니다.

user_events 컬렉션에서 원시 클릭스트림 이벤트를 읽고 기사 메타데이터로 이벤트를 보강하는 단계가 포함된 userIntentSummarizer라는 스트림 프로세서를 만듭니다.

1. Atlas 프로젝트의 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 작업 공간에 연결합니다.

LLM 제공자(예: Azure OpenAI)에 외부 HTTPS 연결을 추가하여 스트림 프로세서가 파이프라인에서 직접 LLM을 호출하여 데이터를 보강할 수 있도록 합니다.

1. 스트림 처리 작업 공간의 창에서 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 단계를 사용하여 사용자 세션을 요약합니다.

Stream Processing 파이프라인에서 직접 LLM 제공자를 호출할 수 있는 $https 단계를 추가합니다. 이 솔루션은 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 데이터베이스에 있는 user_intent라는 새 컬렉션에 기록합니다(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"
 ]
}

의미론적 검색을 위해 기사 데이터를 준비하세요.

시맨틱 검색을 수행하려면 먼저 기사 데이터의 벡터 임베딩을 생성해야 합니다. 이렇게 하려면 articles 컬렉션의 description 필드를 autoEmbed 유형으로 인덱싱하는 vector_index라는 이름의 MongoDB Vector Search 인덱스를 생성합니다. 이는 MongoDB Vector Search가 컬렉션에 문서가 삽입되거나 업데이트될 때마다 Automated Embedding을 사용하여 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
      }
    }
}]