Sistema de Recomendação do Twitter

O sistema de recomendação complexo do Twitter, onde múltiplos componentes trabalham em conjunto para selecionar e classificar os tweets que são exibidos para os usuários na aba "Para você". Vamos descrever cada componente e fase do processo, ilustrando com diagramas e exemplos de código onde aplicável.

Fases do Sistema de Recomendação

Candidate Sourcing

Esta fase é responsável por selecionar os tweets candidatos que podem ser recomendados ao usuário. Os candidatos são coletados de diferentes fontes e algoritmos, como RealGraph, TweepCred, Trust & Safety, GraphJets, etc.

Light Ranker (Earlybird)

Depois de obter os candidatos, um modelo de machine learning leve (light ranker) faz uma primeira classificação desses tweets.

Heavy Ranker

Os tweets classificados pelo light ranker são então processados por um modelo mais pesado e complexo (heavy ranker) para uma classificação mais precisa.

Heurísticas e Filtros

Após o ranqueamento dos tweets, eles passam pelo projeto Home Mixer, onde são aplicadas heurísticas e filtros de visibilidade para assegurar que os tweets exibidos ao usuário estejam de acordo com as políticas do Twitter.

Diagrama do Processo

Vamos representar isso graficamente para entender melhor o fluxo:

               +---------------------+
               |  Candidate Sourcing |
               +----------+----------+
                          |
                          v
                +---------+----------+
                |    Light Ranker    |
                |     (Earlybird)    |
                +---------+----------+
                          |
                          v
                +---------+----------+
                |   Heavy Ranker     |
                |     (PyTorch)      |
                +---------+----------+
                          |
                          v
                +---------+----------+
                | Heuristics & Filters|
                +---------------------+
                          |
                          v
                +---------+----------+
                |    Final Tweets    |
                +---------------------+

Descrição dos Componentes e Exemplos
RealGraph
RealGraph é uma parte do sistema que provavelmente constrói um grafo de interações em tempo real entre usuários e conteúdo. Embora o código completo do RealGraph não esteja disponível, uma implementação simples de um grafo pode ser feita usando networkx.
import networkx as nx

class RealGraph:
    def __init__(self):
        self.graph = nx.DiGraph()

    def add_interaction(self, user_id, tweet_id):
        self.graph.add_edge(user_id, tweet_id)

    def get_user_interactions(self, user_id):
        return list(self.graph.successors(user_id))

# Exemplo de uso
rg = RealGraph()
rg.add_interaction('user1', 'tweet1')
rg.add_interaction('user1', 'tweet2')
rg.add_interaction('user2', 'tweet1')

print(rg.get_user_interactions('user1'))
TweepCred
TweepCred é um componente que avalia a credibilidade dos usuários e conteúdo. Uma implementação simplificada pode envolver a atribuição de uma pontuação de credibilidade baseada em interações passadas.
class TweepCred:
    def __init__(self):
        self.credibility_scores = {}

    def set_credibility(self, user_id, score):
        self.credibility_scores[user_id] = score

    def get_credibility(self, user_id):
        return self.credibility_scores.get(user_id, 0)

# Exemplo de uso
tc = TweepCred()
tc.set_credibility('user1', 75)
print(tc.get_credibility('user1'))
Trust & Safety (T&S)
Trust & Safety usa modelos de aprendizado de máquina para detectar e filtrar conteúdo impróprio. Aqui está um exemplo básico de um filtro usando TensorFlow.
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense
import numpy as np

# Modelo de exemplo para filtrar conteúdo tóxico
model = Sequential([
    Dense(128, activation='relu', input_shape=(100,)),
    Dense(64, activation='relu'),
    Dense(1, activation='sigmoid')
])

model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])

# Exemplo de dados (vetores de características fictícios)
data = np.random.rand(1000, 100)
labels = np.random.randint(2, size=(1000, 1))

model.fit(data, labels, epochs=10, batch_size=32)

# Função para verificar se um tweet é tóxico
def is_toxic(tweet_features):
    return model.predict(np.array([tweet_features])) > 0.5

# Exemplo de uso
print(is_toxic(np.random.rand(100)))

Processo de Sourcing
Candidate Sourcing
O sistema coleta tweets candidatos usando várias fontes e algoritmos (RealGraph, TweepCred, etc.).
Light Ranker (Earlybird)
Este modelo de machine learning inicial faz uma classificação rápida dos tweets candidatos. Abaixo está um exemplo simplificado usando um classificador básico.
from sklearn.linear_model import LogisticRegression

# Exemplo de dados de treinamento
X_train = np.random.rand(100, 10)  # Características dos tweets
y_train = np.random.randint(2, size=(100,))

# Treinando o light ranker
light_ranker = LogisticRegression()
light_ranker.fit(X_train, y_train)

# Função para classificar tweets
def classify_tweet(tweet_features):
    return light_ranker.predict_proba(np.array([tweet_features]))[0, 1]

# Exemplo de uso
print(classify_tweet(np.random.rand(10)))
Heavy Ranker
O heavy ranker faz uma classificação mais detalhada usando um modelo mais complexo, como uma rede neural profunda em PyTorch.
import torch
import torch.nn as nn
import torch.optim as optim

# Modelo de exemplo em PyTorch
class HeavyRanker(nn.Module):
    def __init__(self):
        super(HeavyRanker, self).__init__()
        self.fc1 = nn.Linear(10, 128)
        self.fc2 = nn.Linear(128, 64)
        self.fc3 = nn.Linear(64, 1)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = torch.sigmoid(self.fc3(x))
        return x

model = HeavyRanker()
criterion = nn.BCELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# Exemplo de dados de treinamento
X_train = torch.rand(100, 10)
y_train = torch.rand(100, 1)

# Treinando o heavy ranker
for epoch in range(100):
    optimizer.zero_grad()
    outputs = model(X_train)
    loss = criterion(outputs, y_train)
    loss.backward()
    optimizer.step()

# Função para classificar tweets
def classify_tweet_heavy(tweet_features):
    with torch.no_grad():
        return model(torch.tensor(tweet_features, dtype=torch.float32)).item()

# Exemplo de uso
print(classify_tweet_heavy(np.random.rand(10)))
Heurísticas e Filtros
Depois de classificados, os tweets passam por heurísticas adicionais e filtros para garantir conformidade com as políticas.
def apply_filters(tweets):
    # Filtro de exemplo: Remover tweets que não atendem a um certo limiar de score
    filtered_tweets = [tweet for tweet in tweets if tweet['score'] > 0.5]
    return filtered_tweets

# Exemplo de uso
tweets = [{'id': 'tweet1', 'score': 0.6}, {'id': 'tweet2', 'score': 0.4}]
print(apply_filters(tweets))
Conclusão
O sistema de recomendação do Twitter é altamente complexo, envolvendo múltiplos componentes que trabalham juntos para selecionar, classificar e filtrar os tweets a serem recomendados aos usuários. Cada componente desempenha um papel crucial no processamento, desde a coleta de candidatos até a aplicação de modelos de machine learning e filtros de política. Os exemplos fornecidos aqui são simplificações para ilustrar os conceitos básicos envolvidos. O sistema de recomendação do Twitter envolve diversas técnicas avançadas de machine learning e álgebra linear. A implementação de similaridade de cosseno e embeddings são cruciais para calcular a relevância e recomendar tweets aos usuários. O uso de frameworks como TensorFlow, PyTorch e ONNX demonstra a necessidade de interoperabilidade e eficiência no processamento de grandes volumes de dados em tempo real.
# Exemplo de dados de treinamento
X_train = np.random.rand(100, 10)  # Características dos tweets
y_train = np.random.randint(2, size=(100,))

# Treinando o light ranker
light_ranker = LogisticRegression()
light_ranker.fit(X_train, y_train)

# Função para classificar tweets
def classify_tweet(tweet_features):
    return light_ranker.predict_proba(np.array([tweet_features]))[0, 1]

# Exemplo de uso
print(classify_tweet(np.random.rand(10)))