实时AI新闻聚合器构建指南：多源采集、去重与摘要推送

每天打开十几个标签页翻 AI 新闻，应该是不少技术从业者现在的常态。模型发布、开源项目更新、论文上线、框架迭代，信息密度高到一个人根本刷不过来。也正是因为这个痛点，出现了很多面向 AI 领域的新闻聚合工具，其中一类很典型的项目，就是 Hacker News 上以 "Show HN" 形式发布的开源聚合器：实时抓取多个新闻源，用大模型自动生成摘要，再通过实时推送和每日 Digest 邮件触达用户。

这里需要先给一个明确判断：这种聚合器的技术难点不在"抓新闻"，也不在"发邮件"，而在三个看起来简单、实际决定成败的环节。第一是去重。同一件事，Hacker News、Reddit、Twitter/X、多个技术媒体都会报道，标题不同、角度不同，但实质是同一件事，如果不去重，用户看到的 Digest 里会有三到四条相似新闻，体验会非常差。第二是摘要质量。直接截取 RSS 的 description 片段是远远不够的，摘要要能提炼出"这件事是什么、为什么重要、对开发者意味着什么"。第三是时效性。实时不是真的做到毫秒级推送，而是要在新闻发布后的几分钟内完成采集、去重、摘要入库，这个 pipeline 的延迟决定了产品的价值。

如果你正在考虑做类似的信息聚合工具，不管是面向 AI 新闻、开源项目动态，还是某个垂直领域的技术资讯，这篇文章的思路都可以直接迁移。文章会从架构设计讲起，逐步拆解数据采集、AI 摘要、智能去重、实时推送和每日 Digest 的完整实现，最后给出常见问题排查和工程建议。读完你可以自己跑通一个最小可用的实时 AI 新闻聚合器。

1. 为什么需要实时 AI 新闻聚合器

先说一个经常被误解的问题：实时聚合的价值到底是什么？很多人以为实时就是快，越快越好。但站在用户视角看，真正重要的不是"新闻出现后 1 秒内推送给你"，而是"当你打开信息流时，里面没有重复、没有噪音、没有错过关键动态"。新闻聚合产品的核心竞争力，是过滤和整理，而不是单纯的速度。

AI 领域的信息有一个明显特征："事件密度高，单一信源不完整"。一个新模型的发布，可能在 Hacker News 上是一条讨论帖，在 TechCrunch 上是一篇报道，在 arXiv 上是一篇技术论文，在某个开发者的博客上是一篇使用心得。用户不可能自己去盯这么多渠道，所以聚合器要做的，是把同一事件的多视角内容合并成一条结构化信息：标题是什么、谁发布的、核心亮点是什么、社区反应如何、大家普遍在讨论什么。

这个需求在五年前没有这么强烈，因为 AI 新闻的发布节奏还相对平缓。但现在几乎每周都有重量级发布，靠人工维护信息源列表、手动写摘要的方式已经完全不可行。这也是为什么"实时 AI 新闻聚合器"这类项目在社区里越来越多，而且几乎无一例外地引入了大模型来做摘要。大模型在这里的作用不是炫技，而是把"摘要"这个环节的边际成本降到了可以忽略不计，让聚合器可以规模化处理上百个信息源。

对开发者来说，自己动手做一个聚合器的意义，不只是得到一个私人新闻工具，更是一次完整的信息系统实战：你会接触到定时任务、多源数据归一化、缓存与去重、大模型 API 调用、WebSocket 推送、邮件模板渲染，这套组合在业务开发中非常常见。所以说，这个项目是一个很好的"练手级"真实系统，规模不大，但五脏俱全。

2. 核心概念与架构方案

在写代码之前，先把几个关键概念理清楚。第一个概念是"近实时"（near real-time）。对于新闻聚合来说，真正的"实时"通常是不必要的，因为新闻源本身有发布延迟，RSS 和 API 的更新也不是立刻可见。比较务实的做法是采用轮询策略，每隔 1 到 5 分钟拉取一次源数据。这个时间窗口下，用户基本感受不到延迟，但系统的压力和复杂度会低很多。

第二个概念是 AI 摘要 pipeline。一个典型的摘要流程包括：采集原始内容、清洗 HTML、判断内容是否值得摘要、调用大模型生成摘要、再对摘要做长度和格式控制。这个流程的每一步都可能成为瓶颈。比如有些新闻网页正文提取不干净，会把导航栏、广告文案一并交给大模型，既浪费 token 又影响摘要质量。

第三个概念是每日 Digest。它和实时推送是互补关系：实时推送解决"立刻知道"的问题，每日 Digest 解决"早上花十分钟看完过去一天最重要的 AI 动态"的问题。Digest 不是简单地把所有新闻列出来，而是应该按重要程度、话题领域或关注度分组，让读者先看重点再看列表。

架构上可以拆成五个模块：

这样拆分的好处是每个模块可以独立开发和替换。比如一开始用 SQLite 存储，后续数据量大了再换 PostgreSQL，上层逻辑不用大幅改动；摘要服务今天用一家大模型 API，明天换另一家，也只需要改调用层。

从数据流的角度看，整个系统是一条单向 pipeline：采集器把数据写入待处理队列，处理器完成清洗和去重，摘要服务生成内容摘要，最终落入存储层，触达层再从存储层读取数据完成推送和发信。理解这条数据流，比理解任何一个具体模块都重要，因为几乎所有疑难问题的排查，都要回到"数据现在流到了哪一环"这个问题上。

3. 环境准备与依赖清单

整体方案选择 Python 技术栈，原因是数据处理生态成熟，RSS 解析、HTTP 请求、定时任务、AI 调用都有现成的库，适合快速搭建原型。如果你更熟悉 Node.js 或 Go，也可以用同样的思路迁移，本文以 Python 为例。

你需要在环境中准备以下内容：

• Python 3.10 或更高版本（具体版本以你本机环境为准，建议使用虚拟环境）。
• 一个可以访问的 LLM API，用于生成摘要。无论你使用哪一家服务商，建议确认它提供 OpenAI 兼容的 Chat Completions 接口，这样可以统一调用方式。
• 可选的 SMTP 邮箱配置，用于发送每日 Digest。开发阶段可以先使用日志输出代替真实邮件，避免反复发信。
• 一个操作系统终端和任意代码编辑器。

创建项目目录并初始化虚拟环境：

mkdir ai-news-aggregator cd ai-news-aggregator python3 -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate

然后是依赖清单，本文用一个精简版本：

pip install fastapi uvicorn[standard] httpx feedparser apscheduler sqlalchemy jinja2

各个库在整个项目中的角色如下：

• fastapi 和 uvicorn：提供 REST API 和 WebSocket 服务，作为系统对外接口。
• httpx：发送 HTTP 请求，既用于抓取新闻源，也用于调用 LLM API。
• feedparser：解析 RSS/Atom 订阅源，这是新闻采集最常用的库。
• apscheduler：管理定时任务，负责周期采集和每日 Digest 的触发。
• sqlalchemy：数据库 ORM，统一操作存储层。
• jinja2：渲染 Digest 邮件的 HTML 模板。

依赖版本不需要刻意追求最新，选择稳定的即可。如果安装时遇到网络问题，可以先配置国内 PyPI 镜像，再把安装命令重新执行一遍。安装完成后，可以用pip list确认所有包都已正常写入当前虚拟环境。

4. 数据采集层：多源接入与实时化

采集层的目标是尽量以低成本拿到丰富的新闻数据。常用的公开源包括 Hacker News 的 Algolia API、各类技术媒体的 RSS 订阅、arXiv 的论文更新接口等。对于聚合器来说，源的数量不是越多越好，因为每增加一个源，就多一份需要维护的解析逻辑和反作弊策略。建议从 3 到 5 个高质量源起步，验证整个 pipeline 后再扩大。

以 Hacker News 为例，它提供的官方 API 非常简单，可以直接按关键词搜索最近的帖子：

curl "https://hn.algolia.com/api/v1/search_by_date?query=AI&tags=story&hitsPerPage=20"

返回的 JSON 中包含标题、URL、创建时间、评论数、分数等字段。对于聚合器来说，这些字段非常好用，因为评分和评论数是判断新闻热度的直接依据。

RSS 源的解析同样简单。下面是一个最小示例，用 feedparser 读取某个 RSS 源并提取标题和链接：

# src/collector.py import feedparser def fetch_rss(url: str): feed = feedparser.parse(url) entries = [] for item in feed.entries[:20]: entries.append({ "title": item.get("title", ""), "link": item.get("link", ""), "published": item.get("published", ""), "summary": item.get("summary", "")[:500], }) return entries

在真实项目中，采集器需要同时管理多个源，并且每个源有不同的限流策略。比较好的做法是用一个统一的 source 配置表，把每个源的 URL、类型、拉取间隔、权重都放在一起，而不是在代码里硬编码。下面是一个典型的配置结构：

# src/sources.py SOURCES = [ { "name": "hackernews_ai", "type": "hn", "query": "AI", "interval_minutes": 3, "weight": 3, }, { "name": "techcrunch_ai", "type": "rss", "url": "https://techcrunch.com/category/artificial-intelligence/feed/", "interval_minutes": 10, "weight": 2, }, { "name": "arxiv_ai", "type": "rss", "url": "https://export.arxiv.org/rss/cs.AI", "interval_minutes": 30, "weight": 1, }, ]

这里有一个容易踩坑的地方：不同源的字段结构不一样，Hacker News 接口返回的是 JSON，RSS 源返回的是 XML 解析后的 dict，arXiv 的条目字段又略有不同。统一的做法是在采集器内部把不同源都"归一化"成统一的 NewsItem 结构，后续的清洗、去重、摘要都只依赖这个统一结构，而不是关心来源差异。

归一化之后，采集器需要把新闻写入数据库。写入前要先做一层"是否存在"的检查，一般用链接的哈希值作为唯一键。如果链接太长，可以对链接取 MD5 或 SHA1 的前 16 位作为 ID，这样既能避免重复写入，也为后续去重打下基础。这里再提醒一点：思路是每次采集都增量处理，而不是把整个源的历史数据全量灌入，否则数据量会快速增长，去重成本也随之上升。

5. AI 摘要与智能去重

采集完成只是第一步，真正拉开差距的是摘要和去重。这两个环节直接决定用户最终看到的信息质量，值得花最多时间打磨。

去重分为两个层面。第一层是"完全相似"，同一条链接被多个源转载，或者同一来源重复抓取，这种情况用内容哈希就能解决。遇到内容完全相同的文章，直接丢弃或更新时间字段即可。第二层是"语义相似"，这条难度高很多。比如"OpenAI 发布新模型"这条新闻，可能出现在 Hacker News 的讨论帖、TechCrunch 的标题、某个开发者的博客里，标题完全不同，但讲的确实是同一件事。

对这种语义相似新闻，最简单的方案是先用关键词归一化，再结合向量相似度判断。一个务实的手法是：对每篇新闻的"标题 + 摘要前 100 字"生成文本向量，然后与新入库的新闻向量计算余弦相似度，超过阈值的就标记为重复。开发阶段可以直接把向量存在内存里，或者用 SQLite 的 JSON 字段保存，等数据量真正上来后再接入向量数据库。

下面是使用 LLM 做摘要的关键代码。这里使用 OpenAI 兼容的 Chat Completions 接口，通过环境变量管理 API Key，避免把敏感信息写进代码仓库：

# src/summarizer.py import os import httpx OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1") OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "") MODEL_NAME = os.getenv("SUMMARY_MODEL", "gpt-4o-mini") SUMMARY_PROMPT = """你是一名专业的 AI 技术编辑。请根据下面的新闻内容，生成一段中文摘要。 要求： 1. 用 3 到 5 句话概括新闻的核心信息。 2. 说明这件事为什么重要，对开发者或技术从业者有什么影响。 3. 不要加入原文没有的信息，不要使用"据悉""消息人士称"这类模糊表达。 新闻标题：{title} 新闻正文摘要：{content} 中文摘要：""" def summarize_with_llm(title: str, content: str) -> str: prompt = SUMMARY_PROMPT.format(title=title, content=content[:2000]) payload = { "model": MODEL_NAME, "messages": [ {"role": "system", "content": "你是一个严谨的中文技术编辑。"}, {"role": "user", "content": prompt}, ], "temperature": 0.3, } headers = { "Authorization": f"Bearer {OPENAI_API_KEY}", "Content-Type": "application/json", } with httpx.Client(timeout=30) as client: resp = client.post(f"{OPENAI_BASE_URL}/chat/completions", json=payload, headers=headers) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"].strip()

这里真正值得注意的是 Prompt 的设计。很多聚合器摘要质量差，不是模型不行，而是 Prompt 太随意。好的摘要 Prompt 至少包含三个要素：角色设定、格式要求、信息边界。角色设定让模型以专业编辑的口吻输出，格式要求控制返回长度，信息边界避免模型编造内容。此外，把 temperature 设置为 0.3 左右，可以减少"自由发挥"的空间，保证摘要的稳定性。

调用 LLM 时的成本控制也很重要。并不是所有新闻都值得生成摘要。有些低价值新闻，比如只有标题没有任何正文的短消息，直接跳过即可。合理的做法是在进入摘要服务前加一道过滤：标题少于 10 个字、正文少于 100 个字、或者热度分低于阈值的一律跳过。这样可以节省大量 token 费用，也能让摘要服务的响应速度更快。

6. 实时推送与每日 Digest 生成

新闻完成摘要入库后，需要解决两个触达问题：实时推送和每日 Digest。这两个通道面向的使用场景完全不同，技术实现也有差异。

实时推送在架构上最直接的做法是 WebSocket。前端或移动端建立长连接后，服务端在新闻入库时主动把数据推给客户端。FastAPI 对 WebSocket 的支持很好，下面是一个最小实现：

# src/realtime.py from fastapi import FastAPI, WebSocket, WebSocketDisconnect app = FastAPI() class ConnectionManager: def __init__(self): self.active_connections: list[WebSocket] = [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: dict): for connection in self.active_connections: await connection.send_json(message) manager = ConnectionManager() @app.websocket("/ws/news") async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: await websocket.receive_text() except WebSocketDisconnect: manager.disconnect(websocket)

需要注意，生产环境中的 WebSocket 连接应配合消息队列使用。采集器进程和摘要进程把新闻写入队列，推送服务监听队列，再通过 WebSocket 广播给客户端。如果所有逻辑都放在一个进程里，开发简单但后期扩展受限，尤其是推送量大时容易出现瓶颈。

每日 Digest 的生成则可以设计成一个每日定时任务，通过 APScheduler 触发。它的任务包括：从数据库读取最近 24 小时内入库的新闻、按热度分排序、按主题分组、渲染邮件模板、发送邮件。下面的代码展示了使用 APScheduler 注册定时任务的基本方式：

# src/digest.py from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime def generate_daily_digest(): # 实际项目中，这里应该从数据库读取最近 24 小时的数据 print(f"[{datetime.now()}] 开始生成每日 Digest") items = load_today_items() if not items: print("今日没有新闻，跳过") return html = render_digest_html(items) send_digest_email(html) print(f"[{datetime.now()}] Digest 发送完成，共 {len(items)} 条") def start_scheduler(): scheduler = BlockingScheduler() # 每天早上 8 点执行 scheduler.add_job(generate_daily_digest, "cron", hour=8, minute=0) scheduler.start()

Digest 的邮件模板可以直接用 Jinja2 渲染。模板里至少要包含三个区块：今日最重要的 5 条新闻、按主题分类的新闻列表、原始链接。邮件正文不宜过长，因为用户的注意力有限。Digest 的价值在于"帮助用户快速决定要不要点进原文"，而不是把原文全文搬进邮件。

邮件发送在开发阶段可以用控制台输出代替，或者接入 SMTP 服务。这里给出一个简单的邮件发送函数，配置项通过环境变量注入：

# src/notifier.py import os import smtplib from email.mime.text import MIMEText SMTP_HOST = os.getenv("SMTP_HOST", "smtp.example.com") SMTP_PORT = int(os.getenv("SMTP_PORT", "465")) SMTP_USER = os.getenv("SMTP_USER", "") SMTP_PASS = os.getenv("SMTP_PASS", "") MAIL_TO = os.getenv("MAIL_TO", "") def send_digest_email(html: str): if not SMTP_USER or not MAIL_TO: print("未配置 SMTP，跳过邮件发送，直接在日志输出 HTML") print(html[:500]) return msg = MIMEText(html, "html", "utf-8") msg["Subject"] = "AI 新闻每日 Digest" msg["From"] = SMTP_USER msg["To"] = MAIL_TO with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT) as server: server.login(SMTP_USER, SMTP_PASS) server.sendmail(SMTP_USER, [MAIL_TO], msg.as_string())

这里要强调一个细节：邮件发送是外部依赖最重的环节，SMTP 服务不稳定、账号认证失败、被判定为垃圾邮件都会导致 Digest 发送失败。所以在设计上，发送失败不应该影响新闻本身入库，更不应该阻塞整个 pipeline。异步发送、失败重试、发送结果日志，这三个机制建议从一开始就加上。

7. 完整运行与效果验证

把前面几节的模块组合起来，就可以跑通一个完整的实时 AI 新闻聚合器。项目的最小入口可以设计成两个进程：一个是 FastAPI 服务，提供 API 和 WebSocket；另一个是后台任务进程，负责采集、摘要和定时 Digest。整体采用"一个 API 进程 + 一个 Worker 进程"的模式，职责清晰，也方便后续拆分部署。

先创建数据库和表结构。这里用 SQLAlchemy 定义一个最简模型：

# src/models.py from sqlalchemy import Column, String, Integer, DateTime, Text, create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime Base = declarative_base() class NewsItem(Base): __tablename__ = "news_items" id = Column(String(32), primary_key=True) title = Column(String(500), nullable=False) url = Column(String(1000), nullable=False) source = Column(String(100), nullable=False) summary = Column(Text, default="") score = Column(Integer, default=0) published_at = Column(DateTime, default=datetime.utcnow) created_at = Column(DateTime, default=datetime.utcnow) engine = create_engine("sqlite:///news.db") Base.metadata.create_all(engine) SessionLocal = sessionmaker(bind=engine)

接着是 Worker 进程的入口，负责周期采集任务和每日 Digest 任务：

# src/worker.py from apscheduler.schedulers.blocking import BlockingScheduler from src.collector import fetch_rss from src.sources import SOURCES from src.digest import generate_daily_digest def run_collection(): for source in SOURCES: # 这里只演示 RSS 源，实际需要根据 source["type"] 分发 if source.get("type") != "rss": continue items = fetch_rss(source["url"]) print(f"[collect] {source['name']} 返回 {len(items)} 条") for item in items[:5]: print(f" - {item['title'][:60]}") def main(): scheduler = BlockingScheduler(timezone="Asia/Shanghai") scheduler.add_job(run_collection, "interval", minutes=5) scheduler.add_job(generate_daily_digest, "cron", hour=8, minute=0) scheduler.start() if __name__ == "__main__": main()

启动服务的命令可以这样写：

# 终端 1：启动 API 和 WebSocket 服务 uvicorn src.main:app --host 0.0.0.0 --port 8000 # 终端 2：启动采集与 Digest 后台任务 python -m src.worker

验证时需要关注几个关键指标：

• 采集是否正常：观察日志中是否出现"返回 N 条"信息。如果没有新数据，检查源配置和网络连通性。
• 摘要是否生成：查看数据库中 NewsItem 的 summary 字段是否已经有内容。
• 实时推送是否生效：用一个 WebSocket 客户端连接 ws://localhost:8000/ws/news，当新新闻入库时，应该能收到 JSON 消息。
• Digest 是否发送：在配置了 SMTP 的情况下查看邮件，如果未配置，日志中会输出 HTML 片段。

预期输出大致如下：

[2025-01-15 08:00:01] 开始生成每日 Digest [2025-01-15 08:00:01] 今日新闻共 34 条 [2025-01-15 08:00:03] 未配置 SMTP，跳过邮件发送，直接在日志输出 HTML [2025-01-15 08:00:03] Digest 生成完成

如果摘要为空，第一步先看 LLM API 的返回是否正常，可以在单独的 Python 脚本中调用 summarize_with_llm 函数验证。如果 WebSocket 收不到消息，检查广播逻辑是否真的在新闻写入后被调用，很多问题其实出在"事件触发点"接错了。整体思路是逐步缩小范围：先验证网络层，再验证数据层，最后验证外部依赖。

8. 常见问题与排查思路

在实际开发和运行过程中，比较高频的问题集中在采集失败、摘要超时、重复新闻、邮件发送失败这几个方向。下面整理成排查表，方便直接对照解决。

这里重点说三个容易被忽视的问题。

第一个是时区。APScheduler 默认使用本地时区，但服务器可能配置为 UTC，导致原本设定早上 8 点发送的 Digest 变成了北京时间下午 4 点。解决办法是在创建调度器时显式指定 timezone，不要把时区依赖留给系统环境。

第二个是重试机制。LLM API 在高峰期经常返回 429 或 5xx，如果不对失败任务做重试，每天发送的 Digest 里就会缺内容。推荐采用退避重试策略，比如第一次失败后等 1 秒重试，第二次等 5 秒，第三次等 15 秒，最多重试 3 次。对于采集失败的任务，可以在下一次定时触发时自动补偿，因为新闻源是不断更新的，漏掉一轮通常不会有太大问题。

第三个是内容去重的时间窗口设置。去重不是只看当前批次，还要和最近一段时间的新闻做比较。比较合理的是保留 24 到 48 小时的时间窗口，窗口太短会出现跨批次重复，窗口太长又会把确实相关的后续报道误杀。

9. 最佳实践、总结与后续方向

到这里，一个实时 AI 新闻聚合器的最小闭环已经讲完了。从采集、去重、摘要到实时推送和每日 Digest，整个 pipeline 并不复杂，但要做好需要关注很多细节。最后总结几条工程建议，这些建议不限于这个具体项目，对类似的信息系统也有参考价值。

第一，把配置和代码分离。所有 API Key、SMTP 密码、源地址、模型名称都应该走环境变量或配置中心，不要硬编码。这个项目直接用环境变量读取，已经体现了这个思路，生产环境可以进一步引入配置管理工具，并把敏感配置放到专门的密钥服务中管理。

第二，监控每个环节的成功率。聚合器的核心是 pipeline，而 pipeline 最容易出现"中间某一步失败但整体进程没挂"的情况。建议给采集、摘要、发送三个环节分别打点，记录成功数和失败数。哪怕只是把指标输出到日志，也能在出问题时大幅缩短定位时间。

第三，控制摘要成本。LLM 摘要不是免费的，每天几十条、几百条新闻逐个调用 API，费用会随数据量线性增长。建议在进入摘要服务前增加一个分数门槛，只对热度高、信息量大的新闻生成摘要，其余的直接保留标题和链接。具体门槛值可以根据实际预算和内容质量反复调整。

第四，给用户配置权。好的聚合器不应该是一个固定内容的邮件日报，而是允许用户选择主题、调节推送频率、自定义关键词的系统。从架构上提前预留用户配置表，后续加功能时会轻松很多。即使第一版只服务自己，也要为模型扩展留好空间。

如果继续往深做，有三个方向值得探索。一是引入向量检索，让用户可以用自然语言搜索历史新闻；二是增加个人化推荐，根据用户点击行为调整新闻排序；三是支持多语言摘要，让同一篇新闻被不同语言的用户看到不同语言的版本。这三个方向都建立在本文这套基础架构之上，核心改动集中在存储层和触达层，采集和摘要层基本可以复用。

对于想快速上手实践的同学，建议先不要急着完善所有功能，而是先跑通"采集一张源表、摘要一个接口、发送一封 Digest 邮件"的最小链路，再逐步加实时推送、去重阈值调整和监控指标。这套思路不仅适用于 AI 新闻聚合，任何垂直领域的信息聚合工具都可以复用，关键是把数据 pipeline 的每个环节都控制在可观察、可维护、可替换的状态。