Agentic RSS Reader：用本地小模型实现智能信息过滤

每天早上一打开 RSS 阅读器，几十上百条未读扑面而来，真正值得点开的可能只有三五条；等你看完两条，剩下的已经成了永远清不掉的数字。这是几乎所有 RSS 重度用户都遇到过的尴尬：RSS 解决了“订阅”的问题，却没有解决“理解”的问题。

RSSMonster 这个项目的定位很有意思。从项目名和关键词来看，它重新定义了 RSS 阅读器：不再是一个“按时间排序的抓取工具”，而是一个 agentic 的信息代理——用本地 embeddings 做语义层，用小模型做推理层，再由 agent 逻辑决定哪些内容值得读、哪些内容可以忽略、哪些内容需要主动推送给你。它同时踩中了两个正在快速升温的技术方向：Agentic 应用落地，以及本地小模型推理。

这篇文章不打算只复述项目介绍，而是从 RSS 信息过载这个真实痛点出发，系统拆解 agentic RSS reader 的概念、架构、部署方式和核心代码。读完你能回答三个问题：Agentic 在 RSS 场景里到底解决什么；为什么 embedding 和小模型是这套系统的关键；以及如何用 Python 从零实现一个可运行的最小版本。

1. RSS 阅读器的困境：能聚合，不代表能理解

RSS 的核心价值至今没有被替代：去中心化、用户掌握订阅主权、没有平台算法的强行干预。但传统的 RSS 阅读器在功能上长期停留在“抓取—排序—标记已读”这条流水线上。订阅源越来越多之后，你会发现三个非常现实的痛点：

第一，信息过载。订阅 30 个技术博客，每天产生的未读数轻松超过 100。阅读器只会告诉你“有 100 条未读”，不会告诉你其中哪 5 条真正值得读。最后阅读变成了数字义务，而不是信息获取。

第二，语义缺失。你可以在阅读器里建文件夹、设关键词规则，但关键词过滤本质上是字符串匹配。它分不清“这篇文章讨论 embedding 的应用”和“这篇文章提到 embedding 这个词但是广告”，也分不清两篇内容几乎一样的文章来自不同转载源。

第三，重复内容。同一个技术事件，经常是官方博客、社区翻译、个人解读各发一遍。传统阅读器会忠实地把三条全部摆到你面前，由你人工去重。

与此同时，第三方在线 RSS 服务也存在不少问题：有些服务生成的订阅链接有效期不稳定，有开发者反馈 24 小时后就会失效；内容要经过别人的服务器，阅读数据完全不在自己手里；免费额度有限，体验起伏很大。

所以，RSSMonster 真正值得关注的地方，不是它把抓取做得更快，而是它把处理链路升级了：本地 embedding 负责语义理解，本地小模型负责内容推理，agent 逻辑负责决策和执行。它让 RSS 阅读器第一次有了“判断力”。

2. 核心概念拆解：Agentic、Embeddings、Small Models 到底是什么

在往下看架构之前，先把三个关键概念说清楚。这三个词单独看都不新，但组合在一个 RSS 阅读器里，会产生完全不同的工作方式。

2.1 Agentic：从定时拉取到主动判断

Agentic 不是一个框架，而是一种系统组织方式。传统程序是“你输入什么，它执行什么”；Agentic 系统则包含感知、记忆、规划、执行四个环节，能根据上下文决定下一步动作。

放在 RSS 场景里，agentic 意味着：

• 感知：读取新文章，生成摘要和标签；
• 记忆：把历史文章向量化，形成个人语义记忆库；
• 规划：判断这篇文章是否重要、与已读内容是否重复、该推送给用户还是归档；
• 执行：推送通知、更新摘要、维护订阅源状态。

换句话说，传统 RSS 是“把所有内容推给你”，Agentic RSS 是“先替你想一遍，再把真正重要的内容给你”。

2.2 Local Embeddings：本地语义化

Embedding 是把文本转换成一串浮点数向量，语义相近的文本在向量空间里距离也近。这个技术本身不新鲜，但“本地”两个字是关键。

用本地 embedding 模型，你可以做到：

• 语义去重：两篇文章措辞完全不同，但讲的是同一件事，向量相似度会很高，可以直接过滤；
• 内容聚类：把一周的文章按主题聚类，生成每日/每周综述；
• 相关性过滤：结合你自己设定的偏好向量，筛选“与我关注领域相关”的内容。

相比关键词，embedding 能识别同义表达和上下文，这是 RSS 内容治理能力提升的关键一步。

2.3 Small Models：本地推理

Small Models 指的是参数量在 1B 到 8B 之间、可以跑在本地 CPU 或消费级 GPU 上的模型，比如 Qwen 的小尺寸版本、Phi、Gemma、Llama 3.2 小参数版等。它们的能力不如云端旗舰大模型，但做分类、摘要、标签抽取这类“高频低难度”任务已经足够。

本地小模型有几个直接收益：不产生 API 费用、阅读数据不出本机、离线也能跑、延迟可控。对 RSS 这种每天要处理几十上百篇文章的场景，如果每篇都调云端大模型，成本和时间都不可接受；本地小模型把边际成本降到了接近零。

2.4 和 Agentic RAG 的关系

最近 Agentic RAG 很火，它和 RSSMonster 的思路其实是同构的。传统 RAG 是一次向量检索然后把结果丢给模型；Agentic RAG 则把检索变成智能体工作流中的一环，系统可以反复检索、判断、再行动。

RSSMonster 本质上是一个垂直领域的 agentic 系统：历史订阅内容就是你的“知识库”，embedding 负责检索和去重，小模型负责理解，agent 负责决策。理解这一点，你就能把它的设计思路迁移到其他信息处理任务上。

3. 本地优先：这个项目真正的技术判断

RSSMonster 值得深挖的第一个判断，是把“本地”作为默认前提。这不是为了卖硬件，而是三个真实收益叠加的结果。

数据隐私。阅读习惯是很私密的数据。你订阅了什么、每天在看什么、对哪些话题感兴趣，这些信息一旦经过云端服务，就脱离了你自己的控制。本地 embeddings 和本地模型意味着整条链路的数据都不出本机，这在企业或个人隐私敏感场景里是硬需求。

成本可控。假设你订阅了 50 个源，每天新增 200 篇文章。如果全部调云端大模型做摘要，按 token 计费，一个月下来是一笔持续开支。本地小模型每个月的成本只有电费。对于个人开发者，这是一道很简单的算术题。

可用性。在线 RSS 服务可能有有效期、限流、关停风险，云端 API 也可能有配额限制。本地部署的 RSSMonster 没有“服务过期”的概念，数据都在你磁盘上，随时可以离线阅读、离线整理。

把三者放在一起，本地优先不是情怀选择，而是工程选择。它牺牲了云端大模型的极致能力，换来了隐私、成本和可用性的确定性。对于“每天处理大量低难度文本任务”的 RSS 场景，这笔交换非常划算。

4. RSSMonster 的整体架构与工作流拆解

从项目定位和命名方式看，RSSMonster 的架构可以拆成五个层次。虽然我没有拿到完整源码，但按照“本地 embeddings + small models + agentic”这个组合，工作流大概率是这样一条链路：

摄取层：定时拉取订阅源，解析 RSS/Atom，做增量更新，只处理新增条目。这一步解决的是“数据从哪来”。

语义层：对文章标题和正文生成 embedding，写入本地向量存储，计算与历史内容的相似度，完成去重和聚类。这一步解决的是“哪些内容见过”。

推理层：用小模型对通过去重的文章生成摘要、抽取标签、判断主题相关度。这一步解决的是“文章讲了什么”。

决策层：Agent 根据规则和模型输出做决策：重要内容立即推送，一般内容进入每日摘要，低相关内容只入库不打扰。这一步解决的是“要不要打扰用户”。

执行层：通过 Webhook、钉钉机器人、邮件或本地通知把结果送达用户，同时更新阅读状态。

整个系统的判断力来自两层：embedding 提供“语义相似度”这个客观度量，小模型提供“内容价值”这个主观判断，agent 再把两者组合成动作。这也是 agentic 设计最值得学习的地方——不是用一个大模型包办一切，而是让不同模型各司其职。

5. 环境准备与前置条件

动手之前，先准备环境。下面的示例以 Python 为主，适合在 Linux 或 macOS 上运行，Windows 也基本兼容，但部分模型加载逻辑建议在 Linux 服务器上验证。

建议配置：

• Python 3.10 及以上版本；
• 至少 8GB 内存，16GB 更稳妥；
• 有 NVIDIA GPU 会更流畅，但没有 GPU 也能跑，只是小模型推理慢一些；
• 磁盘预留 10GB 以上，用于存放模型权重和向量数据。

先创建虚拟环境并安装依赖：

python3 -m venv rssmonster-env source rssmonster-env/bin/activate pip install --upgrade pip

依赖清单如下，建议放在requirements.txt：

feedparser>=6.0 requests>=2.31 sentence-transformers>=2.7 transformers>=4.40 torch>=2.1

安装命令：

pip install -r requirements.txt

说明一下：sentence-transformers 负责本地 embedding，transformers 负责加载小模型，feedparser 负责解析 RSS 源，requests 负责推送 Webhook。版本以你实际安装时 pip 解析到的为准，这里只给最低参考版本。

6. 完整示例：从零实现一个最小版 Agentic RSS 管道

下面用一个最小但完整的实现，把整个链路跑通。这个示例包含四个文件：抓取、语义、推理、决策推送。

6.1 抓取与解析 RSS 源

新建fetch_feeds.py，实现订阅源抓取和条目解析：

# fetch_feeds.py import feedparser # 把这里替换成你真实订阅的 RSS/Atom 地址 FEEDS = [ "https://example.com/feed.xml", "https://example.org/rss.xml", ] def fetch_entries(feed_url): feed = feedparser.parse(feed_url) entries = [] for e in feed.entries: entries.append({ "feed": feed.feed.get("title", feed_url), "title": e.get("title", "").strip(), "link": e.get("link", ""), "summary": e.get("summary", "")[:2000], "published": e.get("published", ""), }) return entries def fetch_all(): all_entries = [] for url in FEEDS: try: all_entries.extend(fetch_entries(url)) except Exception as exc: print(f"[warn] fetch {url} failed: {exc}") return all_entries

这段代码做了三件事：解析订阅源、提取标题和摘要、把异常源隔离，避免单个源挂掉影响整体流程。实际项目中还可以用 ETag 和 Last-Modified 做增量请求，减少重复抓取。

6.2 生成本地 Embedding 并做语义去重

新建semantic_store.py，用本地 embedding 模型实现语义去重：

# semantic_store.py import numpy as np from sentence_transformers import SentenceTransformer model = SentenceTransformer("BAAI/bge-small-zh-v1.5") def embed_text(text): return model.encode(text, normalize_embeddings=True) class SemanticStore: def __init__(self, threshold=0.86): self.threshold = threshold self.items = [] # (embedding, meta) def _max_similarity(self, emb): if not self.items: return 0.0 sims = [float(np.dot(emb, item[0])) for item in self.items] return max(sims) def is_duplicate(self, emb): return self._max_similarity(emb) >= self.threshold def add(self, emb, meta): self.items.append((emb, meta))

关键逻辑在于normalize_embeddings=True。向量归一化之后，点积就是余弦相似度，省去手动计算模长的步骤。threshold是去重阈值，建议在 0.85 到 0.90 之间调试：太高会漏掉重复内容，太低会误杀相似但不同的文章。

这个示例为了可读性把向量放在内存里，生产环境建议换成 sqlite-vec、Chroma 或 LanceDB 持久化存储。

6.3 用本地小模型生成摘要

新建summarize.py，用 Qwen2.5 小尺寸版生成中文摘要。第一次运行会下载模型权重，需要保持网络畅通：

# summarize.py import torch from transformers import AutoModelForCausalLM, AutoTokenizer model_id = "Qwen/Qwen2.5-1.5B-Instruct" device = "cuda" if torch.cuda.is_available() else "cpu" tokenizer = AutoTokenizer.from_pretrained(model_id) model = AutoModelForCausalLM.from_pretrained(model_id).to(device) def summarize(title, content, max_new_tokens=128): messages = [ {"role": "system", "content": "你是一个严谨的信息助理，只输出摘要本身，不要任何解释。"}, {"role": "user", "content": f"请用三句话概括这篇技术文章的核心内容。\n标题：{title}\n正文：{content[:1500]}"}, ] prompt = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) inputs = tokenizer(prompt, return_tensors="pt").to(device) outputs = model.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=False) result = tokenizer.decode( outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True, ) return result.strip()

这里值得注意两点：第一，1.5B 模型跑在 CPU 上，一篇 1500 字文章的摘要大约需要几十秒，属于可以接受的范围；第二，apply_chat_template会按模型官方格式组织对话，比自己拼 prompt 更可靠。

6.4 Agent 决策与 Webhook 推送

新建agent.py，实现决策和推送。推送以钉钉 Webhook 为例，换成企业微信或其他 Webhook 只是改 payload 格式的问题：

# agent.py import os import requests WEBHOOK_URL = os.getenv("DINGTALK_WEBHOOK_URL", "") def _should_push(title, summary): # 这里是最简规则，实际可以换成小模型打分 keywords = ["agent", "大模型", "RSS", "embedding", "本地模型"] text = title + " " + summary return any(k.lower() in text.lower() for k in keywords) def push_notice(entry, summary, tags): if not WEBHOOK_URL: print("[skip] WEBHOOK_URL 未配置") return headers = {"Content-Type": "application/json; charset=utf-8"} text = ( f"### [{entry['title']}]({entry['link']})\n\n" f"> {summary}\n\n" f"标签：{', '.join(tags)}\n\n" f"来源：{entry['feed']}" ) payload = { "msgtype": "markdown", "markdown": {"title": entry["title"][:30], "text": text}, } resp = requests.post(WEBHOOK_URL, json=payload, headers=headers, timeout=10) resp.raise_for_status() return resp.json() def agent_process(entry, summary): if _should_push(entry["title"], summary): return push_notice(entry, summary, ["重点关注"]) return None

建议把 Webhook 地址放在环境变量里，不要写死在代码或提交到 Git 仓库。钉钉机器人的安全设置如果配置了自定义关键词，推送内容必须包含这个关键词，否则会被拒绝。

最后用main.py把整条链路串起来：

# main.py from fetch_feeds import fetch_all from semantic_store import SemanticStore, embed_text from summarize import summarize from agent import agent_process store = SemanticStore(threshold=0.86) def main(): entries = fetch_all() print(f"抓取到 {len(entries)} 条新内容") for entry in entries: emb = embed_text(entry["title"] + " " + entry["summary"]) if store.is_duplicate(emb): print(f"[dedup] {entry['title']}") continue store.add(emb, entry) summary = summarize(entry["title"], entry["summary"]) agent_process(entry, summary) print(f"[processed] {entry['title']}") if __name__ == "__main__": main()

这个main.py就是一个极简的 agent 循环：感知（fetch）→ 记忆（embedding 去重）→ 推理（summarize）→ 决策（agent_process）。配置好订阅源之后，用 cron 或 systemd timer 定时执行即可。

7. 运行结果与效果验证

运行最小示例，验证整条链路：

export DINGTALK_WEBHOOK_URL="https://oapi.dingtalk.com/robot/send?access_token=你的TOKEN" python main.py

预期输出大致如下：

抓取到 12 条新内容 [processed] 如何用本地模型构建智能 RSS 阅读器 [dedup] 用本地模型构建 RSS 阅读器的最佳实践 [processed] Embedding 技术入门：从词向量到语义搜索 [skip] WEBHOOK_URL 未配置

验证重点有三个：

第一，去重是否生效。观察是否有[dedup]输出。如果两篇标题不同但内容几乎相同的文章被识别为重复，说明 embedding 和阈值配置正确。如果该去重的没有去重，调低阈值；如果正常文章被误杀，调高阈值。

第二，摘要质量。打开推送或日志里的摘要，看是否准确概括了原文。摘要如果出现重复话、答非所问，优先检查 prompt 和正文截断长度。

第三，推送是否成功。钉钉机器人返回{"errcode":0}表示成功。如果返回 310000，通常是安全设置不匹配，比如配置了自定义关键词但推送内容不包含该关键词。

刚开始建议小规模跑：只订阅 3 个源，先跑一天，再逐步增加。观察模型推理耗时和内存占用，确认在当前机器上稳定运行之后，再放到生产环境。

8. 常见问题与排查方法

下面是最容易踩的几个坑，按影响程度从高到低排列：

其中模型加载和推理性能问题最常见。如果你只有 CPU，建议选择参数量更小的模型，或者直接使用 llama.cpp 的 GGUF 量化版本，推理速度会提升一个量级。

9. 最佳实践与工程建议

把最小示例跑通之后，如果想真正落地到日常使用，下面这些建议会很有价值。

订阅源管理。把订阅源清单抽成独立配置文件，用 YAML 或 JSON 维护，支持分组和启停。不要硬编码在 Python 文件里。这看起来是小事，但订阅源经常要增删，独立配置能避免频繁改代码。

增量抓取。每次全量拉取非常浪费。RSS 规范里的 ETag 和 Last-Modified 可以帮你做条件请求，服务器未更新时返回 304，直接跳过解析。这能大幅降低抓取耗时。

向量持久化。内存里的 SemanticStore 只适合演示。生产环境要把 embedding 落盘，推荐 sqlite-vec、Chroma 或 LanceDB，并建立索引，避免文章量上来之后相似度计算变成全量扫描。

模型分层策略。不要所有任务都用同一个模型。去重用 embedding 模型，摘要用 1.5B 小模型，复杂的长文归类或深度分析才考虑调用云端大模型。高频任务走本地，低频高难度任务走云端，这是成本和效果的最佳平衡点。

Agent 决策需要可观测性。Agent 不推送某篇文章时，最好记录原因：是重复、是低相关、还是不匹配关键词。所有决策都留日志。否则智能体变成一个黑盒，出了问题很难排查。

通知频率控制。Agentic 阅读器最容易犯的错是“太聪明”，每篇文章都想推给你。更好的做法是分三个级别：重要内容实时推送，普通内容进每日摘要，低相关内容只入库不通知。让用户每天只看一次汇总通知。

安全边界。Webhook 地址、模型路径、数据库位置都通过环境变量或配置文件注入，最小权限原则。涉及删除或重置向量库的操作，先备份数据，在测试环境验证后再执行。

10. 总结与后续学习方向

RSSMonster 给我们的启发，不在于“又一个 RSS 阅读器”，而在于它示范了 agentic 应用的一种落地方式：不追求大而全的通用智能体，而是把一个小场景做深——用 embedding 构建语义记忆，用小模型做本地推理，用 agent 逻辑完成决策闭环。这种“本地小模型 + Agentic 工作流”的模式，完全可以迁移到邮件过滤、论文阅读、舆情监控等类似的信息处理场景。

如果你准备动手实践，建议按这个顺序来：先把第 6 节的最小示例跑通，理解整条链路；然后接入自己的真实订阅源，运行一周，观察去重阈值和摘要质量；接着把向量存储换成持久化方案，加上定时任务；最后再考虑引入更多 agent 行为，比如每周综述、自动归档、跨源关联分析。

一个小小的提醒：RSSMonster 这类项目通常迭代很快，关注的模型版本和依赖也容易变化。实际部署时，以项目官方文档的最新说明为准，本文的架构思路和示例代码可以作为理解基础。信息过载不会消失，但有了本地语义层的 RSS 阅读器，至少你可以把“读完所有未读”这个不切实际的目标，换成更合理的“每天只看最重要的几篇”。