用 AI 搭一条自动化热点监控流水线：RSS 聚合、智能筛选、定时推送一次跑通

每天上班第一件事，是不是先刷一遍各种网站看有什么新东西？怎么说呢，我认识一个做内容的朋友。每天早上要开十来个页面：几个资讯站、两个社区、一个项目趋势榜。刷完花半小时看完能用的没几条，大部分是噪音。而且刷的过程很影响上午的状态，一打开手机就停不下来。

从使用者痛点一来看。信息噪音与低效浏览

• 多源手动打开，耗时 30+ 分钟，却只有寥寥可用信息。
• 重复内容导致浪费时间。
• 手动筛选受情绪波动影响，容易错过关键热点。说起来，

后来我把这件事整个自动化了：一条流水线。早上七点自动跑完，把筛选好的热点直接推到我手机里。今天这篇文章就把这条流水线的完整搭建过程写出来每一段都有能直接跑的代码。你照着搭，也能拥有一套自己的 AI 热点雷达。

至于先说架构，三段式流水线

整条流水线分成三段，各干各的活：

1. 采集段从各信息源拉取最新内容
2. 筛选段用大模型理解内容。按你的标准打分、分类、去重
3. 推送段把筛选结果推到目标渠道

三段之间用队列串起来任何一段挂了不影响其他段。这个设计最大的好处是：换数据源、换模型、换推送渠道，都是改一段的事不用动整条链。

说到第一段。RSS 采集

RSS 到现在依然是最稳定的信息获取协议，没有反爬烦恼，格式统一。Python 环境里有现成的库，几行代码就能拉取。

import feedparser
SOURCES =
def fetch_all:
items =
for url in SOURCES:
feed = feedparser.parse
for entry in feed.entries:
items.append({
"title": entry.get。"link": entry.get,"summary": or ""),"published": entry.get,})
return items

几个注意点的观点是，

• feedparser 对绝大多数 RSS/Atom 源都兼容，不用关心格式差异。
• 每个源取前 N 条就够，后面做增量去重。
• summary 字段有的源是 HTML。先截断再做清洗，别把标签带进筛选。

再看第二段。AI 筛选

采集是体力活，筛选才是这条流水线的灵魂。直接全量推送给你会被信息淹没，必须让 AI 先替你读一遍、排个优先级。

筛选的主要原因是：给大模型一个评分标准，让它对每条内容打分并给出理由。标准可以非常具体，比如“与 AI 工具、自动化、效率相关的内容加 +1；标题含具体数字加 +1，纯广告不加分”。下面示例展示如何调用 OpenAI 完成打分：

import json
from openai import OpenAI
client = OpenAI
def score_item:
prompt = f"""你是热点筛选助手。根据以下标准给这条内容打分并说明原因：
- 与 AI 工具/自动化/效率相关：+1
- 标题包含具体数字或结果：+1
- 有实操案例或代码：+1
- 发布日期在最近7天内：+1
- 明显广告或标题党：-2
内容{item}
从内容摘要来看，{item}
只输出 JSON 格式，例如 {{"score": 5,"reason": "理由"}}"""
resp = client.chat.completions.create(
model="gpt-4o-mini"。messages=,response_format={"type": "json_object"},)
result = json.loads
item = result
item = result
return item

User Pain Point 二： 手动阅读难以保持统一标准 → 使用 AI 打分实现“同质化评价”，避免因个人情绪导致的信息遗漏或误判。

调用大模型有成本。所以要做两层调整：

1. 只在标题和摘要上打分，不全文分析，token 消耗极低。
2. 加入关键词预筛。只把命中关键词的条目送进模型，其余直接丢弃。

KEYWORDS =
def cheap_prefilter:
text = .lower
return any in text for k in KEYWORDS)

再看第三段，推送

筛选完之后把高分内容推送到你常用渠道。我用的是公司微信机器人，因为手机上提醒最直接；你换成邮件、Telegram 都行，接口大同小异。

import requests
WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
def push_digest:
if not top_items:
return
text = "📮 今日热点速递
"
for i,it in enumerate:
text += f"{i}. {it}
{it}
再看理由。{it}
"
requests.post

推送格式按「标题 + 链接 + 筛选理由」组织，这样你在手机上扫一眼就能决定点不点开，不用再进原文页。老实说，

说到调度。cron 定时跑

流水线写好了剩下就是让它每天自动跑。 最简单的方式是程序自带的 cron：

# 每天早上7点跑一次
0 7 * * * cd /path/to/project && /usr/bin/python3 pipeline.py>> logs/run.log 2>&1

If you need finer control。use APScheduler inside Python:

from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler
scheduler.add_job
scheduler.add_job
scheduler.start

数据源管理——别硬编码在代码里

刚开始我图省事，把数据源直接写死在列表里。后来想加一个源，就得改代码再部署，非常麻烦。改成配置文件后只需要编辑 YAML 就可以完成新增或权重调整。

sources:
- name: "黑客新闻"
至于url，"https://news.ycombinator.com/rss"
category: "技术"
再看weight，1
- name: "GitHub 趋势"
url的观点是，"https://github.com/trending.atom"
category: "开源"
至于weight，2
- name: "掘金热门"
再看url，"https://juejin.cn/rss"
category: "开发者社区"
至于weight，1

import yaml
with open as f:
SOURCES = yaml.safe_load
def fetch_all:
items =
for src in SOURCES:
至于try。feed = feedparser.parse
for entry in feed.entries:
items.append({
"title": entry.get,"link": entry.get,"summary": or ""),"source": src,"category": src,"weight": src,})
except Exception as e:
print
continue
return items

增量与去重——SQLite 做记忆库

指纹去重固然关键，但要真正做到增量更新，需要把「已经处理过的内容」持久化。我使用 SQLite 保存指纹：

import sqlite3
DB_PATH = "pipeline.db"
def init_db:
conn = sqlite3.connect
conn.execute("""
CREATE TABLE IF NOT EXISTS seen (
fingerprint TEXT PRIMARY KEY,title TEXT,processed_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
conn.close
def is_seen:
conn = sqlite3.connect
row = conn.execute).fetchone
conn.close
return row is not None
def mark_seen:
conn = sqlite3.connect
conn.execute VALUES ",)
conn.commit
conn.close

指纹生成示例

import hashlib。re
def fingerprint:
norm = re.sub)
return hashlib.md5).hexdigest

完整主循环——串起三段业务逻辑

def run_pipeline:
init_db
raw_items = fetch_all
fresh_items =
for item in raw_items:
fp = fingerprint
if is_seen:
continue
mark_seen
if cheap_prefilter:
fresh_items.append
scored_items =
scored_items.sort
# 阈值自行调节，例如>=5 为高质量
top_items = >= 5]
if top_items:
push_digest
print} 条，筛选 {len} 条，高分 {len} 条")
至于else，print
if __name__ == "__main__":
run_pipeline

踩过的坑

• Pitfall 1 – 去重不彻底：不同源会转载同一新闻且标题略有差异。方法是归一化标题后再哈希指纹，如上所示。
• Pitfall 2 – RSS 源抽风：网络抖动或返回空页导致整体停摆。每个 source 用 try/except 包裹，单源失败不阻断整体流程。
• Pitfall 3 – 推送频率失控：最初每条高分立即推送，一天冲击四十多条通知炸手机。改为「聚合批次」后恢复正常，只保留关键提醒。
• Pitfall 4 – 模型调用费用飙升：DAG 全量送入模型成本高。加关键词预筛后调用量削减约80%，效果基本不变。
• Pitfall 5 – 重试缺失导致间歇性挂掉：Apex 中对关键 API 加入指数退避重试机制，以防临时限流或网络异常。
.

关键调用重试装饰器示例

import time
def with_retry:
def wrapper:
for attempt in range:
说到try，return fn
except Exception as e:
if attempt == retries - 1:
raise
wait = base_delay *
print
time.sleep
# unreachable
return wrapper
# 示例使用：
# safe_score_item = with_retry
# safe_score_item

成本控制——一天到底花多少钱？话说回来，

AIGC 调用费用往往是大家犹豫不前的主要顾虑。实际运行中，我每天仅向模型发送数十条「标题+摘要」请求。用 gpt‑4o‑mini，每日成本不到 $0.01，一个月也不到 $0.30。如果某天热点激增，可通过以下两招继续压价：

• # 限制单轮最大打分数量。例如只保留前 N 条最高预估得分项进入模型；其余延后至下一轮处理，
• # 动态阈值调节——当当天调用次数逼近预算上限时提高评分阈值，以过滤更多低价值信息。

方向

• 多语言支持：Add English RSS sources and ask LLM to output Chinese summaries.
• 自定义评分规则：E‑commerce 可加入商品关键词权重；科研可加入技术栈关键词等。
• SQlite 持久化结果：Create a table “daily_digest” to accumulate monthly trends.L​LM 代理层：

推送 —邮件版

公司微信适合即时提醒。但如果想要更正式存档或分享给团队，可使用邮件方式。一样使用 Python 标准库 smtplib，无需额外依赖。

 import smtplib
from email.mime.text import MIMEText
SMTP_HOST =" smtp . qq . com "
SMTP_PORT =
SMTP_USER =" your @qq . com "
SMTP_PASS =" 授权码 "
def push_email:
body="
". join(
f"{i}. {it }
{ it }
理由 ：{ it }"
for i。it en umerate
)
msg=MIMEText
msg =" 今日热点速递 "
msg ="SMTP_USER "
msg ="your@ qq .com "
with smtplib.SMTP_SSLas s :
s.login
s.send_message

邮件版与机器人版可同时开启——机器人即时看，邮件长期存档查阅，两手抓更安全。

稳定性 — 重试与监控

长时间运行不可避免会遇到网络抖动、模型服务短暂不可用或数据源结构变化等异常。如果不加防护，一次故障可能导致整日信息缺失。说起来，我的三层防护措施如下 ：

• **单源失败不阻断** ：每个 RSS 源独立 try/except。失败仅记录日志继续执行其他源。
• **关键步骤指数退避重试** ：如前所示 `with_retry` 装饰器用于模型打分和网络请求。
• **运行日志留痕** ：每次任务结束写入统一日志文件，包括采集总数、过滤数、推送数还有错误摘要。便于快速定位故障根因，

最终 —— 可保存清单

1. 采集阶段使用 `feedparser` 拉取 RSS，多源容错；按理说，
2. 筛选阶段先关键词预筛。再交大模型打分，
3. 推送阶段按「标题 + 链接 + 理由」汇总发送；
4. Cron 或 APScheduler 实现每日两次定时运行；
5. 指纹去重 + 单源容错 + 批量汇总 推送 三大必避坑；
6. 成本调整通过预筛 & 摘要打分 控制 token 消耗；

搭建完毕后我每天早晨从「刷半小时网页」升级为「看一条推送」，省下来的时间全部用于深度阅读和实际产出。这套代码加配置不足三百行，可部署在最低配云主机上。每月费用几块钱以内，可谓性价比之王。话说回来，如果你还有想自动化却迟迟未行动的流程。可以留言交流，下篇可能就是你的需求实现！祝搭建顺利 🎉 .

---