RAG 在新闻摘要中的应用多源信息聚合和时效性敏感的检索策略一、深度引言与场景痛点大家好我是赵咕咕。今年初我们给一个资讯平台做新闻摘要 RAG。需求是每天早上 8 点Agent 自动整合过去 24 小时的多源新闻生成一份 500 字的中文摘要简报。原型用标准 RAG 管道——每天凌晨跑一次批量检索把新闻做 embedding 存入向量库然后按 topic 聚类 LLM 摘要。结果第一版上线后用户投诉摘要里有一条新闻是三天前的。排查发现同一事件的多家媒体报道时间标签不一样Reuters 的报道是今天发的但新华网的相关报道是三天前发的。向量检索把三天前的文章也捞进来了——因为语义上高度相关。这就是新闻摘要 RAG 的核心挑战时间敏感度。普通知识库 RAG 不在意文档是三个月前还是三年前写的但新闻 RAG 里时效性是检索引擎的一等约束。二、底层机制与原理深度剖析2.1 新闻 RAG 与普通 RAG 的关键区别维度通用 RAG新闻 RAG时间敏感度低极高小时级信息来源单一内部知识库多源100 媒体重复与冗余低极高同一事件多源报道检索目标最相关最新 最全面 不重复更新频率周/月级小时/分钟级去重需求低核心功能2.2 时效性敏感的检索流水线2.3 核心创新点时间衰减 事件去重这个流水线有两个普通 RAG 不具备的能力时间衰减加权不是简单地把过去 24 小时的新闻筛出来而是对检索结果的向量相似度乘上一个时间衰减函数final_score similarity_score * exp(-λ * age_hours)这样1 小时前的新闻和 23 小时前的新闻在语义相同的条件下1 小时前的分数更高。λ的值取决于场景——即时新闻λ0.5深度分析λ0.05。事件去重同一个事件被多家媒体重复报道如美联储加息在检索结果中会占据大量槽位。需要通过文本相似度聚类同一个事件只保留一篇最有代表性的报道通常是发布时间最新 来源权重最高的那篇。三、生产级代码实现import asyncio import logging import math import hashlib from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from typing import Any logger logging.getLogger(__name__) dataclass class NewsArticle: 标准化新闻文章。 article_id: str title: str content: str source: str # Reuters / 新华网 / Bloomberg source_weight: float 1.0 # 来源权威权重 published_at: datetime field(default_factorylambda: datetime.now(timezone.utc)) event_id: str # 事件聚类 ID embedding: list[float] | None None language: str zh class NewsRAGAgent: 新闻摘要 RAG Agent。 核心特性 1. 时间衰减加权检索 2. 事件去重 3. 来源多样性保证 def __init__( self, llm_client: Any, vector_store: Any, decay_lambda: float 0.1, # 时间衰减系数 max_age_hours: int 24, top_k_per_event: int 1, # 每个事件最多保留几篇 min_source_diversity: int 3, # 最少来源多样性 ): self._llm llm_client self._vector_store vector_store self._decay_lambda decay_lambda self._max_age_hours max_age_hours self._top_k_per_event top_k_per_event self._min_source_diversity min_source_diversity async def generate_daily_brief( self, topic: str | None None, max_articles: int 10 ) - str: 生成每日新闻简报。 # 1) 检索候选文章 candidates await self._retrieve_news(topic) # 2) 事件去重 deduped await self._deduplicate_events(candidates) # 3) 来源多样性保证 diverse self._ensure_source_diversity(deduped) # 4) 时间排序最新优先 diverse.sort(keylambda a: a.published_at, reverseTrue) selected diverse[:max_articles] # 5) LLM 摘要生成 summary await self._generate_summary(selected, topic) return summary async def _retrieve_news( self, topic: str | None None ) - list[NewsArticle]: 时效性感知的新闻检索。 # 构建查询 query topic or 今日重要新闻 try: response await self._llm.embeddings.create( modeltext-embedding-3-small, inputquery, ) query_emb response.data[0].embedding except Exception as e: logger.error(Embedding 失败: %s, e) return [] # 检索取更多候选后续过滤 try: raw_results self._vector_store.search( query_emb, top_k100 ) except Exception as e: logger.error(向量检索失败: %s, e) return [] # 时间衰减 过滤 now datetime.now(timezone.utc) articles [] for r in raw_results: published r.get(published_at) if isinstance(published, str): published datetime.fromisoformat(published) # 时间过滤 age_hours (now - published).total_seconds() / 3600 if published else 0 if age_hours self._max_age_hours: continue # 时间衰减 time_decay math.exp(-self._decay_lambda * max(0, age_hours - 1)) adjusted_score r.get(score, 0) * time_decay # 来源权重 source_weight r.get(source_weight, 1.0) adjusted_score * source_weight articles.append(NewsArticle( article_idr.get(article_id, ), titler.get(title, ), contentr.get(content, ), sourcer.get(source, unknown), source_weightsource_weight, published_atpublished, event_idr.get(event_id, ), embeddingr.get(embedding), )) # 按调整后分数排序 articles.sort(keylambda a: a.source_weight, reverseTrue) return articles async def _deduplicate_events( self, articles: list[NewsArticle] ) - list[NewsArticle]: 事件去重同一事件只保留一篇最佳报道。 if not articles: return [] # 按发布时间排序 sorted_articles sorted( articles, keylambda a: a.published_at, reverseTrue ) # 方法一基于 event_id 的去重 seen_events: set[str] set() deduped: list[NewsArticle] [] for article in sorted_articles: event_key article.event_id or self._generate_event_key( article.title ) if event_key not in seen_events: deduped.append(article) seen_events.add(event_key) continue # 如果已存在但当前文章更新 → 替换 for i, existing in enumerate(deduped): existing_key existing.event_id or self._generate_event_key( existing.title ) if existing_key event_key: if article.published_at existing.published_at: deduped[i] article break # 方法二对没有 event_id 的文章做语义去重 final_deduped await self._semantic_dedup(deduped) return final_deduped staticmethod def _generate_event_key(title: str) - str: 基于标题生成事件聚类 key。 使用简化的 SimHash 思想提取关键词生成 hash。 # 去除标点、数字 import re clean re.sub(r[^\u4e00-\u9fff\w], , title) return hashlib.md5(clean.encode()).hexdigest()[:12] async def _semantic_dedup( self, articles: list[NewsArticle], threshold: float 0.85 ) - list[NewsArticle]: 基于语义相似度的二次去重。 if len(articles) 1: return articles keep: list[NewsArticle] [articles[0]] for article in articles[1:]: is_dup False for kept in keep: sim await self._compute_similarity(article, kept) if sim threshold: is_dup True # 保留来源权重更高的 if article.source_weight kept.source_weight: keep[keep.index(kept)] article break if not is_dup: keep.append(article) return keep async def _compute_similarity( self, a: NewsArticle, b: NewsArticle ) - float: 计算两篇文章的语义相似度。 优先使用已有的 embedding否则调用 API。 # 简化标题 Jaccard 相似度 a_words set(a.title) b_words set(b.title) if not a_words or not b_words: return 0.0 return len(a_words b_words) / len(a_words | b_words) def _ensure_source_diversity( self, articles: list[NewsArticle] ) - list[NewsArticle]: 确保来源多样性至少包含 N 个不同来源。 if len(articles) self._min_source_diversity: return articles result [] seen_sources: set[str] set() remaining: list[NewsArticle] [] # 第一轮每个来源取一篇 for article in articles: if article.source not in seen_sources: result.append(article) seen_sources.add(article.source) else: remaining.append(article) if len(seen_sources) self._min_source_diversity: break # 第二轮填充剩余槽位 result.extend(remaining) return result async def _generate_summary( self, articles: list[NewsArticle], topic: str | None None, ) - str: LLM 生成新闻摘要。 if not articles: return 今日无相关重要新闻。 # 构建文章列表 article_texts [] for i, article in enumerate(articles, 1): pub_time article.published_at.strftime(%H:%M) article_texts.append( f[{i}] {article.title}\n f 来源: {article.source} | 时间: {pub_time}\n f 摘要: {article.content[:150]}... ) articles_str \n\n.join(article_texts) topic_hint f主题聚焦: {topic}\n if topic else prompt f请根据以下新闻文章生成一份简洁的每日简报。 {topic_hint} 要求 1. 500 字以内 2. 按重要性排序最重要的新闻在前 3. 每条新闻包含来源和发布时间 4. 不重复报道同一事件 5. 仅使用提供的文章内容不要编造信息 ## 四、边界分析与架构权衡 {articles_str} 请生成简报 try: response await self._llm.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}], temperature0.3, max_tokens1000, ) return response.choices[0].message.content or 生成失败 except Exception as e: logger.error(摘要生成失败: %s, e) return f摘要生成失败: {e} async def main(): # Mock 测试 agent NewsRAGAgent( llm_clientNone, vector_storeNone, decay_lambda0.1, max_age_hours24, ) summary await agent.generate_daily_brief(topic科技) print(summary) if __name__ __main__: asyncio.run(main())代码关键设计时间衰减公式exp(-λ * age_hours)。对 1 小时内的新闻不衰减age_hours - 1保证新鲜内容不受影响1 小时后指数衰减。双重去重event_id 去重快速 语义相似度去重精确。对于没有 event_id 的文章通过标题 hash 和语义相似度双重保障。来源多样性先确保每个来源至少有一篇再填充剩余。避免摘要全部来自单一媒体。来源权重Reuters、Bloomberg 等权威媒体的文章分数更高在事件去重时优先保留。五、总结4.1 实时性 vs 完整性新闻 RAG 的最大挑战是如何在 24 小时内覆盖尽可能多的新闻同时不遗漏重要事件。策略分时检索不是一次性拉取 100 篇而是分 6 个时间窗口每 4 小时一段每个窗口取 Top-20最终合并去重。这样保证了时效分布均匀。增量更新突发新闻出现时如地震、重大事件触发增量检索和摘要更新。4.2 事件去重的粒度控制去重阈值0.85的选择太严0.95只有几乎相同的文章才被去重摘要中仍然存在大量重复。太松0.70不同但相关的文章被误判为重复摘要中丢失重要信息。新闻摘要场景建议从 0.85 开始根据实际效果调整。建议做 A/B 测试对比人工摘要来判断去重质量。4.3 来源可信度管理不同来源的权重设置来源类型权重说明Reuters / AP / AFP1.5国际通讯社事实核查最严格Bloomberg / WSJ1.3权威财经媒体新华网 / 人民日报1.2国内官方媒体新浪 / 腾讯新闻0.9商业媒体转载为主自媒体 / 博客0.5需额外核查权重影响检索排序和事件去重时的保留优先级。4.4 什么时候新闻 RAG 足够好场景是否适用通用新闻摘要无特定领域适用财经/政治新闻需专业分析需领域模型 RAG突发新闻报道分钟级需实时流处理 RAG深度调查报道不适用需要人工采访和分析五、总结新闻 RAG 和传统知识库 RAG 的根本区别在于时间是检索的核心维度而不是辅助过滤条件。设计方案时需要注意时间衰减是必修课新鲜度必须嵌入到检索评分中而不是事后过滤。事件去重比检索召回更重要新闻场景的最大问题是同一事件 50 篇报道而不是找不到相关新闻。来源多样性是质量保证单一来源的摘要缺乏视角用户能感知到偏颇。增量更新是生产必需早上的摘要到中午可能已经过时了需要支持增量更新。时间在 RAG 系统中有一种降维打击的能力——一条 1 小时前的高质量报道对用户的效用远超一条 23 小时前的更高语义相似度的报道。传统向量检索假设相似度 相关性但在新闻场景中这个假设需要被时间重新加权。下一篇预告Redis 连接管理优化连接池大小、空闲超时和健康检查的最佳配置。