Atlas Stream Processing 및 MongoDB Vector Search를 사용한 실시간 미디어 추천

데이터 소스를 설정합니다.

이 솔루션은 동일한 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 페이지로 Go 합니다.

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

   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 cluster 에 있어야 합니다.

1. 스트림 처리 작업 공간의 창에서 Manage를 클릭하세요.

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

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

   • 연결 유형: Atlas Database

   • 연결 이름: userevents

   • Atlas Cluster: 클릭스트림 및 아티클 데이터가 있는 클러스터 의 이름( 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:user_events news 연결된 클릭스트림 클러스터 (userevents 연결)의 데이터베이스 에 있는 컬렉션 에서 원시 클릭스트림 이벤트를 읽습니다.

   • $lookup articles article_id: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을 호출하여 데이터를 보강할 활성화 있도록 LLM 제공자 ( 예시: Azure OpenAI)에 외부 HTTPS 연결을 추가합니다.

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 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 출력에서 요약을 추출하고 새 컬렉션 에 쓰기 (write) .

• $match: 데이터베이스 에 불완전한 데이터가 기록되지 않도록 LLM 호출이 실패하고 오류를 반환한 세션을 모두 필터링합니다.

• $addFields:summary LLM 출력에서 필드 추출하여 문서 의 최상위 수준에 추가합니다.

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

스트림 프로세서를 시작합니다.

클릭스트림 데이터 요약을 시작할 준비가 되면 스트림 처리 작업 공간의 스트림 프로세서 목록에서 userIntentSummarizer 프로세서의 Start 아이콘을 클릭합니다.

이 파이프라인 사용자의 관심사를 포착하는 세션 요약이 포함된 문서를 user_intent 컬렉션 에 쓰기 (write) 해야 합니다. 예시 들면 다음과 같습니다.

{
 "_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 컬렉션 description 의 필드 articles 유형으로 autoEmbed 인덱싱하는 이라는 이름의 MongoDB Vector Search 인덱스 만듭니다. 이는 MongoDB Vector Search가 자동 임베딩을 사용하여 컬렉션 description 에 문서가 삽입되거나 업데이트될 때마다 필드 에 대한 벡터 임베딩을 자동으로 생성하도록 지시합니다.

이 JSON 정의를 사용하여 이러한 필드에 벡터 인덱스 생성합니다.

• description 필드 autoEmbed 유형으로 지정하여 컬렉션 에 문서가 삽입되거나 업데이트될 때마다 voyage-4-large 임베딩 모델을 사용하여 description 필드 에 대한 벡터 임베딩을 자동으로 생성하도록 MongoDB Vector Search에 지시합니다.

• 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
      }
    }
}]