Personalized RSS Feed

Streamlit + Pinecone + FirstBatch = 🚀

Here we build an RSS feed with personalization using Streamlit + Pinecone + FirstBatch.

Go to streamlit app.

Algorithm outlined in detail.

import streamlit as st
from firstbatch import FirstBatch, AlgorithmLabel, Pinecone, Config, UserAction, Signal
import pinecone

FIRST_BATCH_API_KEY = st.secrets["api"]["firstbatch_api_key"]
PINECONE_API_KEY = st.secrets["api"]["pinecone_api_key"]
PINECONE_ENV = st.secrets["api"]["pinecone_env"]
CUSTOM_ALGO_ID = st.secrets["custom_algo_id"]

# Display items as a feed
def display_feed_item():
    for i, b in enumerate(st.session_state.batches):
        image_url = b.data["img_link"]
        link_url = b.data["link"]
        title = b.data["title"]
        text = b.data["text"]

        with st.container():
            st.markdown("""
                <style>
                    .box {
                        border: 1px solid white;
                        padding: 10px;
                        margin: 5px 0;
                    }
                    .box-title {
                        font-size: 24px;  # Adjust the title size as desired
                        margin-bottom: 10px;
                    }
                </style>
            """, unsafe_allow_html=True)

            # Wrap content in a div with the class "box"
            st.markdown(f"""
                <div class="box">
                    <div class="box-title">
                        <a href="{link_url}">{title}</a>
                    </div>
                    <img src="{image_url}" width="450">  <!-- Adjust the image width as desired -->
                    <p>{text}</p>
                </div>
            """, unsafe_allow_html=True)

        # Check if this id is already in the liked items
        if st.button(f'Like', i):
            signal(st.session_state.ids[i])
            st.session_state.likes.append(b)

# Display sidebar 
def display_sidebar():
    st.sidebar.title("RSS Feed Algorithm")
    lurl = "https://firstbatch.gitbook.io/firstbatch-sdk/get-started/introduction"
    st.sidebar.markdown(f"""
    The recommendation algorithm uses embeddings for personalization. Here's a breakdown:
    Level 1: Initial state of the algorithm. Content is fully randomized.

    Level 2: Tightly anchored to signals, introducing minimal variations and randomness. It pulls content closely aligned with current signals.

    Level 3: Leverages signals while integrating elements of randomness and exploration. It retrieves content that resonates with current signals and content that is contextually linked to these signals. 

    Level 4: Strongly add elements of randomness and exploration. It retrieves content that resembles with current signals but nudged into new directions.

    For all nodes,  Only the last three signals influence the calculation of embeddings, ensuring that personalization is influenced predominantly by recent interactions.

    Algorithm in [detail](https://firstbatch.gitbook.io/rss-feed-algorithm/)

    For SDK Reference: [FirstBatch SDK Docs]({lurl})

    """)

    # Display signalled (liked) items
    st.sidebar.subheader("Liked Items")
    for item in st.session_state.likes:
        title = item.data["title"]
        link_url = item.data["link"]
        st.sidebar.markdown(f"[{title}]({link_url})")

# Send selected RSS post as a signal to firstbatch
def signal(cid):
    st.session_state.personalized.add_signal(st.session_state.session, UserAction(Signal.LIKE), cid)


def main():
    st.title("Personalized RSS Feed")

    # Initialize Pinecone and other services

    if 'personalized' not in st.session_state:
        # Initialize FirstBatch
        config = Config(batch_size=10, verbose=True, enable_history=True)
        personalized = FirstBatch(api_key=FIRST_BATCH_API_KEY, config=config)
        # Attach VectorDB
        pinecone.init(api_key=PINECONE_API_KEY, environment=PINECONE_ENV)
        index = pinecone.Index("rss-2")
        personalized.add_vdb("rss_db", Pinecone(index, embedding_size=384))
        st.session_state.personalized = personalized
        # Create session with Custom Algorithm
        st.session_state.session = st.session_state.personalized.session(AlgorithmLabel.CUSTOM,
                                                                         vdbid="rss_db",
                                                                         custom_id=CUSTOM_ALGO_ID)
        # State variables to prevent reload of content
        st.session_state.batches = []
        st.session_state.ids = []
        st.session_state.likes = []
        ids, batch = st.session_state.personalized.batch(st.session_state.session)
        st.session_state.batches += batch
        st.session_state.ids += ids

    display_feed_item()
    display_sidebar()

    # If "Load more" button is clicked, increase the batch count and rerun the app to fetch more data
    if st.button("Load more"):
        ids, batch = st.session_state.personalized.batch(st.session_state.session)
        st.session_state.batches += batch
        st.session_state.ids += ids
        st.experimental_rerun()


if __name__ == '__main__':
    main()