每天打开十几个标签页翻 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 不是简单地把所有新闻列出来而是应该按重要程度、话题领域或关注度分组让读者先看重点再看列表。架构上可以拆成五个模块模块职责关键组件采集器定时抓取多源新闻数据RSS 解析器、Hacker News API、定时任务处理器清洗、去重、分类、打分文本哈希、向量相似度、规则引擎摘要服务调用大模型生成内容摘要LLM API、Prompt 模板、重试机制存储层保存新闻、摘要、用户订阅SQLite / PostgreSQL触达层实时推送和每日邮件WebSocket、邮件服务、定时触发器这样拆分的好处是每个模块可以独立开发和替换。比如一开始用 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?queryAItagsstoryhitsPerPage20返回的 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 接口返回的是 JSONRSS 源返回的是 XML 解析后的 dictarXiv 的条目字段又略有不同。统一的做法是在采集器内部把不同源都归一化成统一的 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(titletitle, contentcontent[:2000]) payload { model: MODEL_NAME, messages: [ {role: system, content: 你是一个严谨的中文技术编辑。}, {role: user, content: prompt}, ], temperature: 0.3, } headers { Authorization: fBearer {OPENAI_API_KEY}, Content-Type: application/json, } with httpx.Client(timeout30) as client: resp client.post(f{OPENAI_BASE_URL}/chat/completions, jsonpayload, headersheaders) 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, hour8, minute0) 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_keyTrue) title Column(String(500), nullableFalse) url Column(String(1000), nullableFalse) source Column(String(100), nullableFalse) summary Column(Text, default) score Column(Integer, default0) published_at Column(DateTime, defaultdatetime.utcnow) created_at Column(DateTime, defaultdatetime.utcnow) engine create_engine(sqlite:///news.db) Base.metadata.create_all(engine) SessionLocal sessionmaker(bindengine)接着是 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(timezoneAsia/Shanghai) scheduler.add_job(run_collection, interval, minutes5) scheduler.add_job(generate_daily_digest, cron, hour8, minute0) 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. 常见问题与排查思路在实际开发和运行过程中比较高频的问题集中在采集失败、摘要超时、重复新闻、邮件发送失败这几个方向。下面整理成排查表方便直接对照解决。问题现象可能原因排查方式解决方案采集不到任何新闻网络不可达或源地址过期在终端 curl 该源的 URL检查返回状态码更新源地址添加超时和重试机制摘要结果为空字符串LLM API Key 无效或触发限流查看调用日志单独测试一次摘要函数检查环境变量增加指数退避重试同一新闻重复出现在 Digest语义去重阈值过低打印相似度分数观察重复样本的分布调整阈值或增强关键词提取逻辑邮件发送失败SMTP 配置错误或端口被限制查看 SMTP 返回错误码检查账号授权码确认端口和加密方式WebSocket 收不到实时推送保存和广播不在同一个事务里在广播函数前后打日志确保新闻入库后主动调用广播函数定时任务不触发调度器时区配置错误检查调度器日志确认当前时间设置 timezone 参数写明确时区数据库越来越大缺少历史数据清理任务查看数据库表行数和存储占用增加定期清理任务保留最近 90 天即可这里重点说三个容易被忽视的问题。第一个是时区。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 的每个环节都控制在可观察、可维护、可替换的状态。