实时AI新闻聚合器:用大模型实现智能采集与每日摘要

发布时间:2026/9/6 5:47:38
实时AI新闻聚合器:用大模型实现智能采集与每日摘要 如果你每天都要追AI领域的新动态大概率会有一种感觉信息不是太少而是太多。arXiv上每天新增的论文、GitHub Trending上的新项目、Hacker News上面的讨论帖、各大厂的模型发布公告、技术公众号的解读文章这些信息源叠加在一起根本不是一个人能看完的量。更麻烦的是这些信息源之间高度重叠。同一个模型发布会同时出现在官方博客、技术媒体、社区讨论和公众号转载里。你花了一个小时刷信息流真正有增量价值的可能只有两三条。这种状态持续久了获取信息的效率会越来越低甚至让人产生一种“不刷就落后”的焦虑。所以最近在Hacker News上出现的一类工具值得关注实时AI新闻聚合器Real-time AI news aggregator代表项目就包括“Show HN: Real-time AI news aggregator with daily digest”这类作品。它并不是又一个RSS阅读器而是一个带有AI理解和判断能力的新闻过滤系统。这篇文章会从这类项目的定位出发拆解它的核心功能、技术架构和设计边界然后用一个最小可运行示例带你实现“实时采集 每日摘要”这条核心链路。最后我会重点聊聊工程化落地时最容易踩的坑。这类工具真正解决的问题不是“如何获取更多新闻”而是“如何用更少的时间从大量噪声里提取出真正值得看的几条”。它把流程从“人去找新闻”变成了“系统帮你判断新闻的重要性”这也是我认为它值得写一篇完整文章的原因。1. 为什么实时AI新闻聚合器值得关注做技术的人应该都有过这样的经历为了跟上一个热门方向你关注了几十个RSS源、加入了无数个微信群和讨论组、每天打开各种App刷推荐流。但信息越多反而越不知道今天到底发生了什么重要的事。传统的信息获取方式有几个明显的缺陷第一没有优先级。RSS阅读器把所有文章按时间排列一篇模型技术报告和一篇营销软文权重相同你需要自己判断先看哪篇。第二没有跨源去重。同一个消息在官方博客、科技媒体、社区讨论里各出现一次你浪费了大量时间在不同的标题里看同一件事。第三没有迟滞控制。突发消息出现后你可能要等公众号写完解读、等推荐算法把它推到你面前才能看到。实时AI新闻聚合器要解决的问题就是这三个。它通过程序化采集把新闻的延迟从“小时级”压缩到“分钟级”通过大模型对文章内容做理解和重要性排序把“需要你自己判断”的部分交给模型完成再把分散在不同信息源的同类报道合并去重最后生成一份结构化的每日摘要。你不一定需要在新闻出现的第一秒就刷到它但至少不应该在一个重要发布过去半天之后还要靠朋友转发才知道。从使用场景来看这类工具特别适合三类人AI工程师需要第一时间感知模型和框架的变化技术管理者需要快速了解行业动态但没时间逐条刷还有做AI产品和技术调研的人需要持续追踪特定方向而不是泛泛地看新闻。当然它也有自己的适用边界。它更适合帮助你“知道发生了什么”它的摘要能力做得再好也无法替代你自己去读原始技术文档、跑一遍Demo或者做深入的代码研究。所以对它的正确预期是它能当好一个高质量的信息助理而不是一个替你写代码的顾问。2. 核心概念与功能边界很多人第一次看到“实时AI新闻聚合器”这个名词会把它和RSS阅读器、Google News、社交网络信息流混为一谈。实际上它们的差异还挺大。为了后面讲架构不跑偏我们先统一几个核心概念。2.1 什么是实时AI新闻聚合器实时AI新闻聚合器是一个自动化的信息处理系统它持续从多个信息源采集新闻利用AI模型对内容做理解、去重和重要性排序然后以实时推送或周期性摘要的形式输出给用户。关键词是“理解”和“排序”。传统的RSS阅读器只是把XML内容解析后排列出来它不关心这篇文章讲什么。而实时AI新闻聚合器会把文章的标题、正文、来源、发布时间等信息一起交给大模型分析判断“这条新闻对目标读者是否有价值”再决定要不要推荐、推荐权重有多高。我不建议在一篇文章里堆砌几十个卖点。真正有区分度的能力其实只有四个实时采集、跨源去重、重要性排序、摘要生成。这四个能力是否可靠决定了一个聚合器能不能用。2.2 Daily Digest 为什么有用Daily Digest也就是“每日摘要”是这类产品的标配功能。它的工作方式是这样的系统在指定时间点把当天采集到的所有新闻经过滤和排序后挑出最重要的3到5条生成一份带有结论性质的摘要推送给用户。这里有个设计细节值得注意每日摘要把“实时”和“总结”结合在了一起。实时采集负责不错过突发内容每日摘要负责让用户有一个固定的时间节点去回顾。实时性是应对偶发事件摘要是应对信息规整两者并不冲突而是互补。很多工具做不好摘要原因是把摘要理解成了“把标题拼在一起”。真正的摘要需要把多条相关新闻合并、提炼核心信息、指明它为什么重要。这一层理解能力只有大模型能承担。这也是为什么近一两年这类工具才集中出现——不是爬虫技术多难而是理解和生成能力终于到了可用的阶段。2.3 实时不等于“毫秒级推送”“实时”这个词在工程上有很大的误解空间。对新闻聚合器来说真正的实时推送给开发者带来的价值往往不如业务稳定性重要。从实践看绝大多数场景把延迟控制在“分钟级”已经足够。配置一台定时任务每5分钟到30分钟轮询一次RSS源和公开API再对新增内容做增量处理就能覆盖大部分AI新闻的时效需求。毕竟模型发布、开源项目更新、技术博客发文都不是秒级事件。官方在发公告之前往往已经有预告、Paper页面或者GitHub仓库预提交。真正你无法绕过的信息源反而是那些即时讨论社区但这些源也伴随着巨大的噪声。所以在架构设计上我更倾向于把这类系统定义为“准实时系统”采集延迟控制在分钟级摘要产出周期固定为天或小时。这个定位能显著降低系统复杂度同时完全满足使用需求。3. 技术架构拆解一个聚合器的典型分层虽然每个产品的具体实现方式很不同但一个可用的实时AI新闻聚合器在架构上大体都能拆成四层数据采集层、处理理解层、存储层、分发层。下面把这四层展开说说方便你对整体系统有一个全局认识。3.1 数据采集层RSS、API、爬虫数据采集层负责从各个信息源获取原始内容。最常用的渠道有三种RSS订阅源、平台公开API、网页爬虫。RSS是最轻量的方式。Hacker News、arXiv、各大技术博客都提供RSS用feedparser这类库解析即可不需要处理复杂的页面结构。公开API则适合那些不提供RSS但开放接口的平台比如GitHub的events API、某些科技媒体的API端点。网页爬虫是兜底方案适合采集RSS和API都没有覆盖的站点但需要处理页面结构变化、反爬、robots协议和站点服务条款维护成本是三种方式里最高的。这里有一个非常重要的工程原则优先使用RSS和官方API把网页爬虫作为最后手段。输入材料足够多的时候一个聚合器一天可以采集几千条新闻但真正有效的只有几十条。采集层的重点不是“能不能抓到”而是“稳不稳定”、以及“有没有过度抓取”。过度抓取会给对方服务器造成压力也可能把自己整个IP段封掉。采集层的输出应该是一份统一的、结构化的原始条目包含标题、链接、发布时间、正文摘要、来源、分类等字段。不同来源的字段命名可能不同在这一层做一次标准化后面处理层会省掉大量麻烦。3.2 处理理解层去重、排序、摘要处理理解层是整个系统的大脑也是和传统RSS工具拉出差距的地方。去重是一个容易被低估的模块。同一个AI新闻可能在厂商官方博客、Google News、Hacker News、Twitter讨论、公众号转载里出现五次。如果不去重摘要里就会出现五条内容高度雷同的新闻。轻量级的去重可以用标题或链接的哈希值判断更可靠的方案是用SimHash或MinHash计算语义相似度对相似度超过阈值的文章做合并。排序和摘要建议直接用大模型完成。具体做法是把标准化后的新闻条目批量发送给模型让模型用JSON结构返回排序结果和摘要。相比传统的关键词打分方案大模型能理解上下文能判断“这篇不是AI新闻的重要发布而只是测评博客的借题发挥”。摘要生成则要设计合适的prompt让模型输出“标题、来源、一句话提炼、对开发者的意义”这样的结构化内容而不是一段自然语言流水账。处理层的另一个细节是增量更新。全量重新处理所有新闻在数据量变大后不现实更合理的做法是把已处理条目的ID存入缓存或数据库每次只对新抓到的内容做去重和摘要。3.3 存储与分发层存储层不一定非得用重型数据库。早期一个SQLite文件完全够用表结构只需要原始文章表、摘要结果表、已处理ID表。等到数据量大了再迁移到PostgreSQL或者MySQL。很多开发者在这个阶段容易过度设计一上来就上Kafka、上向量数据库、上灰度发布平台。但从实际体感看一个日处理量几千条的新闻聚合器单机加SQLite就可以跑得很舒服。过早引入分布式组件只会增加运维成本并不会提升用户体验。分发层负责把摘要送到用户面前。常见形式有三种Web页面、邮件摘要、即时通讯Webhook。Web页面适合查看历史记录和全文邮件摘要适合每天定时发送Webhook比如企业微信、钉钉、Telegram Bot则适合做实时推送。从工程角度看Webhook分发最简单只要在关键节点调用Webhook URL发送消息即可。综合这四层来看一个最小可用系统的复杂度并不高。最难的部分永远在处理理解层的质量上同样是大模型不同的去重策略和prompt设计产出的摘要质量可以差出一个量级。4. 最小可运行示例实时采集 每日摘要前面讲了那么多架构层的概念下面进入可以直接上手的部分。我用Python实现一个最小可运行的新闻聚合器演示“实时采集 每日摘要”的核心链路。这个示例不复杂但它包含了完整的数据流你可以在此基础上扩展成自己的项目。4.1 环境准备运行环境建议满足以下条件Python 3.10 或更高版本主要是为了类型注解和更好的异步支持一个可以访问的LLM API服务支持OpenAI兼容格式能正常访问目标RSS源需要安装的依赖库如下pip install feedparser requests openai apscheduler pyyaml这里简单说明一下这几个库的用途feedparser解析RSS/Atom订阅源是采集层的主力。requests发送HTTP请求获取RSS内容。openaiOpenAI官方Python SDK也兼容大量第三方模型网关。apscheduler定时任务调度库用来实现周期性采集和每日摘要。pyyaml解析配置文件。4.2 项目结构与配置我习惯把项目拆成四个文件职责很清楚ai-news-digest/ ├── config.yaml # 配置文件 ├── requirements.txt # 依赖清单 └── src/ ├── __init__.py ├── fetcher.py # 采集与去重 ├── summarizer.py # 摘要生成 └── scheduler.py # 调度入口配置文件这样写feeds: - name: Hacker News url: https://news.ycombinator.com/rss category: community - name: arXiv AI url: https://export.arxiv.org/rss/cs.AI category: paper - name: Google AI Blog url: https://blog.google/technology/ai/rss/ category: official llm: api_key: sk-your-key model: gpt-4o-mini base_url: schedule: digest_hour: 9 timezone: Asia/Shanghai interval_minutes: 30配置里需要注意几个地方。feeds列表可以根据自己的兴趣增减但不要一上来加太多源30分钟轮询一次源太多容易被限流。llm.model推荐选择便宜、速度快的模型因为聚合器的核心调用量并不低每日摘要加上重要性排序每天可能要调用几十次到上百次LLM接口。base_url留着空值就使用OpenAI官方地址如果你用的是其他平台的兼容接口可以在这里填平台地址。依赖清单如下feedparser requests openai apscheduler pyyaml不写死版本号是因为这些库的接口在近期都比较稳定安装最新版即可。如果你的项目对可复现性要求高再用pip freeze requirements.lock固定版本。4.3 采集模块抓取RSS并做增量去重采集模块的代码放在src/fetcher.py里核心逻辑包括加载配置、请求RSS、标准化学段、增量去重。# 文件路径src/fetcher.py import hashlib import time import feedparser import requests import yaml from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def load_config(pathconfig.yaml): 加载YAML配置文件。 with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def fetch_rss(url, timeout10): 抓取RSS内容带基础重试。 headers { User-Agent: ai-news-digest/0.1 (educational demo) } session requests.Session() retry Retry(total2, backoff_factor1, status_forcelist[500, 502, 503, 504]) session.mount(https://, HTTPAdapter(max_retriesretry)) resp session.get(url, headersheaders, timeouttimeout) resp.raise_for_status() return feedparser.parse(resp.content) def normalize_entry(entry, source_name, category): 把RSS条目转换成统一结构。 title entry.get(title, ).strip() link entry.get(link, ).strip() or entry.get(id, ).strip() published entry.get(published, entry.get(updated, )) summary entry.get(summary, entry.get(description, )) item_id hashlib.sha1( f{source_name}:{link}.encode(utf-8) ).hexdigest() return { item_id: item_id, title: title, link: link, published: published, summary: summary, source: source_name, category: category, fetched_at: int(time.time()), } def deduplicate(items, seen_ids): 基于item_id做增量去重。 new_items [] for item in items: if item[item_id] in seen_ids: continue seen_ids.add(item[item_id]) new_items.append(item) return new_items简单解释一下关键点。fetch_rss里设置了User-Agent和重试策略这是采集程序的基本素养。任何对外抓取请求都应该带上明确的身份标识和适度的重试机制一方面方便对方站点识别和联系你另一方面也能避免网络抖动导致整个任务失败。不要用默认的python-requests标识去高频抓取很容易被风险策略拦截也不要设置无限重试会给自己和对方都带来压力。normalize_entry把不同RSS源的字段统一成相同结构。不同RSS源在发布时间、summary字段的命名上都不太一样这里的get默认值顺序就是为了兼容常见差异。item_id是去重的核心它使用来源名加链接做SHA1哈希。因为链接在绝大多数情况下是唯一的这个方案足够稳定只有当你需要面对“同一新闻在不同站点出现”这样的语义级去重时才需要换成SimHash方案。deduplicate需要配合一个外部维护的seen_ids集合使用在调度器里我们会在内存里维护这个集合并在每次采集结束后继续保留。4.4 摘要生成模块LLM调用与结构化输出摘要模块放在src/summarizer.py它做的事情是把当天采集到的新闻列表发送给LLM要求模型挑选最重要的几条并生成结构化摘要。# 文件路径src/summarizer.py import json from openai import OpenAI SYSTEM_PROMPT 你是一名资深AI技术编辑。请从给定的新闻列表中挑选出最重要、最值得关注的3到5条新闻 并生成一份结构化的每日摘要。要求如下 1. 优先选择与AI模型发布、AI工程实践、开源项目、产品发布相关的新闻。 2. 忽略明显的营销软文和与AI无关的内容。 3. 多条新闻如果指向同一事件合并为一条并在来源中列出多个来源。 4. 返回严格的JSON格式字段为 date: 摘要日期 items: 数组每个元素包含title, source, summary, why_it_matters 5. summary控制在60字以内why_it_matters控制在60字以内使用中文。 def build_daily_digest(client, news_items, modelgpt-4o-mini): 调用LLM生成每日摘要。 news_text json.dumps(news_items, ensure_asciiFalse, indent2) resp client.chat.completions.create( modelmodel, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: f今日待处理的新闻列表\n{news_text}}, ], temperature0.2, response_format{type: json_object}, ) content resp.choices[0].message.content return json.loads(content)这个模块看起来短但Prompt里面浓缩了很多质量工程的经验。SYSTEM_PROMPT里明确写了“忽略营销软文”“重复内容合并为一条”这是摘要质量的关键。如果你不告诉模型这些规则它倾向于把所有新闻都罗列出来摘要就退化成“标题列表”。“对开发者的意义”这个字段也很重要它对应模型对信息价值的二次判断。只输出“某公司发布了某个模型”是不够的还要回答“它和之前方案比有什么变化、对做工程的人意味着什么”。结构化输出用response_format{type: json_object}这是OpenAI Chat Completions接口支持的参数能极大降低JSON解析失败的概率。如果你用的其他模型网关不兼容这个参数可以把这行去掉并在Prompt里强调“只输出JSON不要输出其他提示语”。不过在实践里去掉之后偶尔会出现输出被markdown代码块包裹、或者额外输出说明文字的情况解析时要做异常处理。调用方的OpenAI客户端没有在这个文件里初始化而是通过参数传入这是为了把client初始化放在调度入口统一管理。后面你如果想替换成其他SDK比如接入国内模型的OpenAI兼容接口只需要在入口处修改初始化逻辑。4.5 调度入口定时采集与每日摘要调度模块放在src/scheduler.py负责把采集和摘要两个模块串起来并注册定时任务。# 文件路径src/scheduler.py import json import time from apscheduler.schedulers.background import BackgroundScheduler from openai import OpenAI from fetcher import deduplicate, fetch_rss, load_config, normalize_entry from summarizer import build_daily_digest def collect_latest(cfg): 遍历所有RSS源做增量采集和去重。 seen_ids set() all_items [] for source in cfg[feeds]: try: parsed fetch_rss(source[url]) entries [ normalize_entry( entry, source[name], source.get(category, general) ) for entry in parsed.entries ] fresh_items deduplicate(entries, seen_ids) all_items.extend(fresh_items) print( f[采集] {source[name]} 新增 {len(fresh_items)} 条 f累计 {len(all_items)} 条 ) except Exception as exc: print(f[采集] 处理源 {source[name]} 失败: {exc}) return all_items def run_digest(cfg): 执行一次完整的摘要生产流程。 client OpenAI( api_keycfg[llm][api_key], base_urlcfg[llm].get(base_url) or None, ) items collect_latest(cfg) if not items: print([摘要] 暂无新内容跳过本次摘要) return # 为了避免单次请求内容过长最多取前50条交给LLM处理 digest build_daily_digest( client, items[:50], cfg[llm][model] ) print([摘要] 生成结果) print(json.dumps(digest, ensure_asciiFalse, indent2)) if __name__ __main__: cfg load_config() schedule_cfg cfg.get(schedule, {}) scheduler BackgroundScheduler( timezoneschedule_cfg.get(timezone, Asia/Shanghai) ) # 每30分钟做一次增量采集 scheduler.add_job( run_digest, interval, minutesschedule_cfg.get(interval_minutes, 30), args[cfg], idinterval_digest, ) # 每天早上9点生成一份完整每日摘要 scheduler.add_job( run_digest, cron, hourschedule_cfg.get(digest_hour, 9), args[cfg], iddaily_digest, ) scheduler.start() print([调度] 聚合器已启动按 CtrlC 退出) try: while True: time.sleep(1) except KeyboardInterrupt: print([调度] 已停止)这段代码有三个地方值得展开解释。第一两个任务都调用了run_digest它们之间的差异只在触发频率上。间隔任务负责“准实时”地处理新增内容每日任务负责在固定时间点生成完整回顾。如果你的实际需求是“只在每天固定时间推送”可以去掉interval任务。反过来如果你希望推送频率更高可以把interval_minutes调小但不要低于10RSS源更新没那么频繁太短的间隔只会徒增请求压力。第二seen_ids这个去重集合是在collect_latest内部定义的这意味着它只生效于“一次采集过程”。对于示例来说这个够用了因为每次采集都会把当前RSS源里的最新内容抓回来与上次重叠的内容会被去重。不过如果你在两次采集之间发生了进程重启这个集合就会丢失上次已经处理过的新闻会被再次抓取。解决方向有两个把seen_ids持久化到SQLite或者以链接为条件查数据库。生产环境里我更推荐后者。第三items[:50]这个上限是出于成本和窗口长度的考虑。50条新闻序列化之后系统提示词加上内容已经比较长再往上加会让摘要质量和响应速度都下降。如果你的信息源特别多建议先对原始列表按来源权重和关键词做一个初筛而不是无脑取前50条。5. 运行结果与效果验证代码写完下面进入验证环节。把你自己的RSS源配到config.yaml里然后把api_key填成你自己的有效密钥在项目根目录执行cd ai-news-digest pip install feedparser requests openai apscheduler pyyaml python src/scheduler.py如果你只是想先跑一次看看效果不想等定时任务可以临时在scheduler.py末尾加一行run_digest(cfg)或者写一个一次性的调用脚本# 文件路径run_once.py from src.fetcher import load_config from src.scheduler import run_digest if __name__ __main__: cfg load_config() run_digest(cfg)运行后采集成功的标志是控制台出现类似这样的日志[采集] Hacker News 新增 18 条累计 18 条 [采集] arXiv AI 新增 42 条累计 60 条 [摘要] 生成结果摘要输出是一段结构化的JSON大致长这样{ date: 2025-01-15, items: [ { title: 某研究团队发布新一代多模态模型, source: arXiv AI, Hacker News, summary: 该模型在视觉语言任务上达到了新的SOTA水平并开源了权重。, why_it_matters: 开源权重意味着中小团队可以直接基于它做微调工程落地门槛明显降低。 } ] }如何判断是否成功有四个标准采集日志里每个源都有“新增N条”的输出而不是异常堆栈。RSS源数量少时可以故意选一个大源比如Hacker News确保短时间内能抓到数据。摘要JSON能被正确解析不是普通文本更不是答辩式的解释文字。items里的内容与当天新闻事实相符没有明显张冠李戴。如果运行失败第一步永远先看错误信息出现在哪一层。网络层失败会抛requests异常解析层失败会出现在feedparser.parse调用附近LLM调用失败则大概率是API Key、网络或模型名称配置错误。逐层向上排查不要一上来就怀疑是代码逻辑不对。6. 常见问题与排查思路实际使用和开发过程中有几个问题出现的频率非常高。我把它们整理成表格方便你按图索骥。问题现象可能原因排查方式解决方案某个RSS源总是采集失败目标站点限流、RSS地址失效curl手动请求该URL查看状态码和响应内容更新RSS地址或对该源增加更长的重试间隔LLM返回内容解析失败模型输出被markdown代码块包裹或接口不支持JSON模式打印原始content内容确认格式在Prompt里强调只输出JSON解析前做strip处理过滤代码块标记摘要总是重复推荐同一类内容没有跨源语义去重同一事件被多条原始新闻命中检查原始新闻列表里是否存在相似标题引入SimHash或LLM层面的“合并同一事件”指令每天定时摘要不触发时区设置错误或进程被外部杀掉查看服务日志确认调度器是否还在运行统一使用Asia/Shanghai时区用systemd或Supervisor守护进程采集频率过高导致IP被暂时封禁轮询间隔太短没有合理的退避策略查看目标站点的响应码是否出现403把间隔调大到30分钟配置指数退避重试API成本超出预算每次调用把全量新闻发给LLMPrompt过长查看每日调用量和token用量增加新闻条数上限先做规则初筛再交给LLM这六个问题里前两个是新手最容易遇到的。RSS源失效特别好理解。很多博客迁移或改版后RSS地址会静默变化旧地址可能返回404也可能返回一个404页面但HTTP状态码是200。后一种情况更隐蔽需要用feedparser.parse后的返回结构判断是否有entries而不仅仅是看状态码。LLM输出解析失败常见原因是某些模型网关不支持response_format参数或者即使返回了JSON也会包裹在json代码块里。示例代码在build_daily_digest里用了response_format如果切到不兼容的网关建议在解析时加上一层容错# 文件路径src/summarizer.py容错版本片段 import re def safe_parse_json(content: str) - dict: content content.strip() # 去掉可能的markdown代码块包裹 if content.startswith(): content re.sub(r^(?:json)?|$, , content, flagsre.MULTILINE) return json.loads(content)用这个函数替换原json.loads能解决大部分解析异常。7. 最佳实践与工程建议例子的部分讲完了接下来是这些经验放在真实项目里需要注意的地方。7.1 数据获取必须注意合规做新闻聚合器最容易吃暗亏的就是数据合规问题。RSS和官方API是相对安全的获取渠道使用前也应该看一下目标站点是否有明确的抓取限制。网页爬虫要格外小心先确认robots.txt再确认网站服务条款不要用暴力请求去碰对方不想开放的页面。更不要为了抓取数据去绕过登录、验证码或其他访问控制这既违反站点服务条款也可能带来法律风险。在项目文档里也建议写清楚每个数据源的抓取频率和抓取范围做到可追溯。这不仅是给外部查看者看也是给自己做约束。没有约束的采集程序跑着跑着就会变成一个对别人不友好的压力测试工具。7.2 配置管理要留出安全边界config.yaml里直接写了api_key这只是演示方便不要在生产环境照搬。真实项目里应该使用环境变量或者专门的密钥管理服务来保存API密钥。更稳妥的做法是让配置文件里只保留一个占位符SDK初始化时从环境变量读取密钥export OPENAI_API_KEYsk-your-key代码里对应改成import os client OpenAI( api_keyos.environ.get(OPENAI_API_KEY), base_urlcfg[llm].get(base_url) or None, )如果你的项目要提交到Git仓库记得把config.yaml敏感字段变量化或者使用config.example.yaml作为模板提交真正的配置加入.gitignore。7.3 摘要质量的关键在于Prompt和成本控制摘要模块的效果上限很大程度取决于Prompt设计。建议你花时间做的不是堆砌更多规则而是明确“模型在什么情况下不做推荐”。让模型学会舍弃比让模型学会推荐更重要。我在前面Prompt里写的“忽略营销软文”“重复内容合并”就是在训练模型做舍弃。成本控制则是另一个容易被忽略的点。大模型API按token计费如果你的RSS源很多每天把所有新闻全量发给模型很快就会产生可观的费用。三个可行的控制路径是采集后先用关键词和来源权重做一次初筛把明显无关的条目过滤掉把LLM调用窗口限制在最近10到50条对连续相同来源的相似内容做规则去重后再进模型。7.4 生产环境下要关注可观测性示例代码里用了print输出日志这只能满足本地调试。真实部署时建议换成标准日志库logging把日志写入文件或收集到集中日志平台。采集任务有没有跑、今天摘要有没有生成、LLM调用是否报错、API调用花费是多少这些都需要能随时回溯。进程守护也很重要。scheduler.py在前台运行一旦服务器重启或者进程退出定时任务就停了。推荐用systemd、Supervisor或者容器平台都行。关键判断标准是进程崩溃后能否自动拉起日志不会丢失账号密钥不落盘明文。如果把这个系统部署到多台机器上还要额外考虑分布式锁避免多个实例同时执行采集任务重复抓取RSS并重复调用LLM。单机阶段不需要考虑这个问题但架构上尽量给调度器预留一个“是否允许分布式运行”的开关。7.5 先小后大迭代式开发最后一条建议也许是最实在的不要一开始就追求大而全。先用三个RSS源跑通整条链路确认摘要质量满意之后再逐步增加信息源、增加去重算法、增加Web界面。新闻聚合器的价值是慢慢积累起来的它不是那种“一上来就必须完美”的系统。很多类似项目最后做不下去不是因为技术不行而是因为信息源维护和摘要质量打磨需要长期投入。你可以把这套系统当成一个能持续改进的个人基础设施而不是一次性交付的Demo。8. 总结与后续学习方向回到文章开头的问题实时AI新闻聚合器到底解决了什么它把“人逐条刷新闻”的低效行为变成了“机器先过滤、人只看结果”的高效流程。它的实时性负责追上新消息它的每日摘要负责把信息整理成可读的结论。不要期待它能替代你阅读全文一个高质量摘要能让你快速判断“这条新闻值不值得深入了解”已经完成了它的主要价值。如果你要动手实践我建议按这个顺序推进先把我给出的示例跑通用真实的RSS源和LLM API生成第一份每日摘要感受一下整条链路的数据流。跑通之后再根据自己的使用习惯调整信息源和Prompt。当你觉得“每天手动点开结果太麻烦”时再去做Web页面或者Webhook推送这一步会带来明显的体验提升。更进一步的方向有三个把内存去重改成数据库持久化解决跨重启的重复采集问题在去重模块里引入SimHash或向量相似度解决跨来源的同事件新闻合并问题把固定频率的定时采集改成基于Webhook的推送接收让系统从“轮询”升级成“事件驱动”进一步降低延迟。这些方向每一个都足够单独写一篇技术文章但起点都是今天这套最小链路。把示例代码保存好改成你自己的信息源跑两天看看摘要质量。你会发现技术细节并不复杂难的是持续维护信息源和让大模型稳定输出高质量结果。而这恰恰是这类产品真正有门槛的地方。