Разработка AI-системы генерации новостных дайджестов
Персонализированный новостной дайджест из сотен источников — задача, которую человек не решит вручную в приемлемые сроки. AI-система мониторит источники, кластеризует публикации по темам, удаляет дубли и генерирует связный дайджест под конкретного пользователя или сегмент аудитории.
Пайплайн сбора и обработки
class NewsDigestPipeline:
def __init__(self, sources: list[NewsSource]):
self.crawler = NewsCrawler(sources)
self.deduplicator = SemanticDeduplicator(threshold=0.85)
self.clusterer = NewsClusterer()
self.summarizer = NewsSummarizer()
self.ranker = PersonalizedRanker()
async def generate_digest(
self,
user_profile: UserProfile,
period_hours: int = 24
) -> Digest:
# 1. Сбор новостей за период
articles = await self.crawler.fetch_since(
datetime.utcnow() - timedelta(hours=period_hours)
)
# 2. Удаление дублей (одна новость из 20 источников → 1 запись)
unique_articles = self.deduplicator.deduplicate(articles)
# 3. Кластеризация по событиям
clusters = self.clusterer.cluster(unique_articles)
# 4. Персонализированный ранжинг кластеров
ranked_clusters = self.ranker.rank(clusters, user_profile)
# 5. Генерация резюме по кластеру (multi-document summarization)
summaries = [
self.summarizer.summarize_cluster(cluster)
for cluster in ranked_clusters[:user_profile.digest_size]
]
return Digest(items=summaries, generated_at=datetime.utcnow())
Дедупликация новостей
Одно событие освещается десятками изданий. Near-duplicate detection:
class SemanticDeduplicator:
def __init__(self, threshold: float = 0.85):
self.encoder = SentenceTransformer("paraphrase-multilingual-mpnet-base-v2")
self.threshold = threshold
def deduplicate(self, articles: list[Article]) -> list[Article]:
# Энкодирование заголовков + лида
texts = [f"{a.title}. {a.lead}" for a in articles]
embeddings = self.encoder.encode(texts, batch_size=256)
# MinHash LSH для эффективного поиска похожих
lsh = MinHashLSH(threshold=self.threshold, num_perm=128)
groups = lsh.find_groups(embeddings)
# Из каждой группы берём первоисточник (по времени публикации)
result = []
for group in groups:
primary = min(group, key=lambda a: a.published_at)
primary.alternative_sources = [a.url for a in group if a != primary]
result.append(primary)
return result
Multi-document summarization для кластера
Задача: из 5–20 статей об одном событии составить краткое резюме без потери ключевых деталей. Стратегия map-reduce:
def summarize_cluster(articles: list[Article]) -> ClusterSummary:
# Ранжирование статей по авторитетности источника и полноте
ranked = rank_articles_by_quality(articles)
if len(articles) <= 3:
# Небольшой кластер — прямая суммаризация
combined = "\n\n".join(a.full_text for a in ranked[:3])
summary = llm.generate(f"Кратко изложи ключевые факты:\n{combined}", max_tokens=200)
else:
# Большой кластер — map-reduce
individual_summaries = [
llm.generate(f"Выдели ключевые факты (2-3 предложения):\n{a.full_text}", max_tokens=100)
for a in ranked[:10]
]
# Объединяем уникальные факты
summary = llm.generate(
f"Составь связный абзац из этих фактов (без повторов):\n" +
"\n".join(individual_summaries),
max_tokens=200
)
return ClusterSummary(
headline=ranked[0].title,
summary=summary,
key_sources=[a.url for a in ranked[:3]],
article_count=len(articles),
topic_tags=extract_tags(articles)
)
Персонализация
Три уровня персонализации:
Тематические интересы: явные (пользователь выбрал рубрики) + неявные (клики, время чтения). Collaborative filtering для новых пользователей.
Глубина материала: одни предпочитают краткий абзац, другие — развёрнутый анализ. Определяется по поведению.
Формат доставки: email-дайджест, Telegram-бот, push-уведомления в приложении, RSS-лента. Частота: утром, вечером, раз в неделю — по выбору пользователя.
Метрики качества дайджестов
- CTR по статьям: какой % материалов пользователь открывает — цель 15%+
- Read-through rate: дочитываемость — цель 60%+
- Diversity score: разнообразие тематик — не все статьи об одном
- Freshness: среднее время от события до дайджеста — цель < 4 часа для важных новостей