1. 为什么我要自己搭一个 AI 资讯聚合平台
信息过载这件事,做技术的人体会最深。我每天要跟踪的大模型动态、Agent 框架更新、开源项目 release、行业落地案例,散落在几十个渠道里:技术社区、公众号、RSS、邮件列表、社交平台、论文预印本站点。早期我靠手动收藏和浏览器书签硬扛,结果是收藏夹里堆了几百条“稍后再看”,真正读过的不到十分之一。后来试过几个现成的资讯类产品,要么推送的内容和我关注的方向偏差太大,要么广告和标题党混在一起,要么干脆停更了。折腾到最后我发现,与其等别人喂饭,不如自己搭一个——AI 资讯聚合平台,把信息源、过滤规则、摘要生成、推送渠道全部握在自己手里。
这个平台要解决的核心问题其实就三个:信息从哪来、怎么筛、怎么送到我面前。听起来简单,但每一环都有坑。信息源要处理不同格式(RSS、HTML 抓取、API 返回的 JSON),筛选要能按关键词、来源权重、时间新鲜度综合打分,推送要能适配我不同场景下的阅读习惯——早上通勤看摘要卡片,午休看深度长文,晚上整理归档。我前后迭代了三版,从最初一个跑在本地的小脚本,到现在一套相对稳定的自动化流水线,中间踩过的坑足够写一篇长文。
这篇文章适合谁看?如果你是有一定编程基础、想给自己或团队搭一套信息管道的开发者,可以直接抄作业;如果你是产品经理或内容运营,想理解资讯聚合背后的技术逻辑,也能从架构设计和取舍思路里拿到参考;哪怕你只是想给自己搭一个“每日 AI 简报”的小工具,文中的最小可行方案也能让你两小时内跑起来。我会把选型理由、参数计算、实操步骤、排查经验全部摊开讲,不藏私。
2. 平台整体架构与核心思路拆解
2.1 从需求倒推架构:四个模块的职责边界
搭任何系统之前,我习惯先把需求拆成“输入—处理—输出”三段。AI 资讯聚合平台的需求可以拆成四个模块:采集层、处理层、存储层、分发层。采集层负责从各种信息源把原始内容拉回来;处理层做清洗、去重、分类、摘要;存储层把处理好的内容落库,方便检索和回溯;分发层按我的阅读习惯把内容推到不同终端。
为什么是这四个而不是更多?我试过把“分类”和“摘要”拆成两个独立服务,结果发现它们共享同一份文本预处理逻辑,拆开后反而要重复做分词和清洗,维护成本翻倍。后来合并成处理层,用一条流水线串起来,代码量少了三分之一,调试也简单。这个取舍的逻辑是:模块划分应该跟着数据流走,而不是跟着功能名词走。数据从采集到分发是一条直线,中间任何需要共享上下文的环节都应该放在同一个模块里。
另一个关键决策是同步还是异步。早期我用同步方式,采集完立刻处理、处理完立刻推送,结果一个源响应慢就卡住整条链路。后来改成异步队列,采集层只管往队列里丢原始数据,处理层按自己的节奏消费,分发层再订阅处理结果。这样即使某个源挂了,也不影响其他源的正常流转。队列我用的是 Redis 的 List 结构,简单够用,没必要上 Kafka——我的数据量每天也就几千条,Kafka 的运维成本反而划不来。
2.2 技术选型:为什么是 Python + SQLite + 定时任务
技术栈这块我纠结过一阵。候选方案有三个:Python 脚本 + SQLite、Node.js + MongoDB、Go + PostgreSQL。最后选了第一个,理由很实在:AI 相关的文本处理库,Python 生态最全。无论是调大模型 API 做摘要,还是用 jieba 做中文分词,或者用 sentence-transformers 做语义去重,Python 都有现成的轮子。Node.js 在这块要差一截,Go 就更不用说了,很多库要么没有要么不成熟。
数据库选 SQLite 而不是 PostgreSQL,是因为我的部署环境是一台 2 核 4G 的小服务器,SQLite 零配置、单文件、备份就是复制一个文件,运维成本几乎为零。有人会担心并发写入问题,但我的场景是单进程写入、多进程读取,SQLite 的 WAL 模式完全扛得住。实测下来,每天几千条的写入量,查询响应都在毫秒级。如果哪天数据量涨到百万级,再迁移到 PostgreSQL 也不迟,SQL 语法基本兼容,迁移成本可控。
定时任务我用的是系统的 cron,而不是 Airflow 或 Celery Beat。原因很简单:我的任务调度需求就是“每 30 分钟跑一次采集”“每天早上 8 点跑一次推送”,cron 一行配置搞定,没必要引入重型调度框架。Airflow 适合有复杂依赖关系的 DAG,我这就一条直线,用它是杀鸡用牛刀。这里有个经验:技术选型要匹配当前规模,不要为想象中的未来过度设计。我见过太多项目一上来就上微服务、上消息队列集群,结果日活不过百,维护成本倒是先把自己拖垮了。
2.3 信息源管理:RSS 优先,抓取兜底,API 补充
信息源的接入方式直接决定了采集层的复杂度。我把信息源分成三类:标准 RSS 源、无 RSS 的网页、提供 API 的服务。RSS 源最好处理,用 feedparser 库几行代码就能解析出标题、链接、发布时间、正文。无 RSS 的网页就得写抓取规则,用 requests + BeautifulSoup 提取内容,这块最麻烦,因为网页结构一变规则就失效。提供 API 的服务最省心,但通常有频率限制,得做限流和缓存。
我的策略是:能 RSS 就 RSS,不能 RSS 就看有没有 API,都没有才写抓取。目前我的源列表里有 40 多个源,其中 RSS 占七成,API 占两成,抓取占一成。这个比例是有意控制的,因为抓取规则的维护成本太高。我给自己定了个规矩:一个抓取源如果一个月内失效超过两次,就考虑放弃或者找替代源。毕竟资讯聚合的目的是减负,不是给自己找活干。
源的质量比数量重要。早期我贪多,塞了上百个源进去,结果每天推送上千条,根本看不过来。后来我做了一轮精简,按“信息密度”和“与我关注方向的相关度”两个维度打分,砍到 40 个左右。信息密度指的是这个源里有多少条是我真正会点开看的,相关度指的是内容和我关注的领域(大模型、Agent、AI 工程实践)的匹配程度。砍完之后,每天推送量降到 100 条以内,阅读完成率反而上去了。
3. 核心细节解析与实操要点
3.1 采集层:如何处理不同格式的信息源
采集层的核心任务是把异构的原始数据统一成标准结构。我定义了一个统一的 Article 数据结构,包含title、url、source、published_at、content、raw_html六个字段。不管源是什么格式,采集器最终都要输出这个结构。这样做的好处是下游处理层不用关心数据从哪来,只管消费标准结构就行。
RSS 源的采集用 feedparser,代码大概是这样:
import feedparser from datetime import datetime def fetch_rss(url, source_name): feed = feedparser.parse(url) articles = [] for entry in feed.entries: articles.append({ 'title': entry.get('title', ''), 'url': entry.get('link', ''), 'source': source_name, 'published_at': entry.get('published_parsed', None), 'content': entry.get('summary', ''), 'raw_html': entry.get('content', [{}])[0].get('value', '') }) return articles这里有个坑:published_parsed返回的是时间元组,需要转成 datetime 对象才能入库。另外不同源的summary字段质量参差不齐,有的给全文,有的只给一句话,所以正文的完整获取还得靠后续的正文提取环节。
网页抓取我用的是 requests + BeautifulSoup 的组合,但更推荐 readability-lxml 这个库,它能自动识别网页正文区域,比手写 CSS 选择器鲁棒得多。用法是:
from readability import Document import requests def fetch_webpage(url, source_name): resp = requests.get(url, timeout=10, headers={'User-Agent': 'Mozilla/5.0'}) doc = Document(resp.text) return { 'title': doc.title(), 'url': url, 'source': source_name, 'content': doc.summary(), 'raw_html': resp.text }注意:抓取网页一定要设置 User-Agent,否则很多站点会直接返回 403。另外 timeout 必须设,不然一个慢站点能把整个采集任务拖死。我一般设 10 秒,超过就放弃这个源,下一轮再试。
API 源的采集要看具体服务的文档,但通用做法是加一层缓存,避免重复请求。我用的是 requests-cache 库,把响应缓存到本地 SQLite,设置 30 分钟过期。这样即使采集任务频繁跑,也不会把 API 配额耗光。
3.2 处理层:去重、分类、摘要的三道工序
处理层是整个平台最核心的部分,我把它拆成三道工序:去重、分类、摘要。顺序不能乱,因为去重能减少后续工序的计算量,分类结果能指导摘要的详略程度。
去重分两个层次:精确去重和语义去重。精确去重就是比对 URL 和标题的哈希值,相同就丢弃。语义去重是比对内容的相似度,因为同一件事可能被多个源报道,标题不同但内容高度重合。语义去重我用的是 sentence-transformers 的paraphrase-multilingual-MiniLM-L12-v2模型,把内容转成向量,计算余弦相似度,超过 0.85 就认为是重复内容,保留发布时间最早的那条。这个阈值是我调了几次定下来的,太低会误杀相关但不重复的内容,太高又起不到去重效果。
分类我用的是关键词匹配 + 规则引擎,没有上机器学习模型。原因是我关注的领域比较固定,关键词表维护起来很直观,而且规则引擎的决策过程可解释,出问题好排查。我的分类体系是:大模型动态、Agent 框架、AI 工程实践、行业落地、论文速递、工具推荐六个类别。每个类别对应一组关键词,比如“Agent 框架”对应agent、langchain、autogen、crewai等。分类结果会写回 Article 结构,供后续筛选和推送使用。
摘要生成我调的是大模型 API,用的是“抽取式 + 生成式”混合策略。先抽取正文的前三段和包含关键词的句子,拼成一个精简版文本,再让模型基于这个精简版生成 100 字以内的摘要。这样做的好处是既控制了 token 消耗,又保证了摘要的信息密度。prompt 我改了好几版,最终定下来的是:
请用中文为以下 AI 资讯生成一段不超过 100 字的摘要,要求: 1. 保留核心事实和数据 2. 说明这件事对 AI 从业者的意义 3. 不要用“本文”“这篇文章”等指代词 4. 直接输出摘要内容,不要加任何前缀 正文:{content}提示:摘要生成一定要做失败重试和降级处理。API 偶尔会超时或返回空结果,这时候就降级用抽取式摘要,保证流水线不中断。我在代码里用 try-except 包住 API 调用,失败就 fallback 到
content[:200]。
3.3 存储层:表结构设计与索引优化
存储层用 SQLite,表结构设计得比较简单,但索引这块我花了不少心思。主表articles的字段包括id、title、url、source、category、published_at、content、summary、created_at。索引建了三个:url唯一索引(用于精确去重)、published_at普通索引(用于按时间排序)、category普通索引(用于按类别筛选)。
为什么不在title上建索引?因为标题的查询需求很少,而且标题长度不一,索引效果不好。真正高频的查询是“按时间倒序取最近 N 条”和“按类别取最近 N 条”,这两个查询用published_at和category索引就能覆盖。实测下来,10 万条数据量下,这两个查询都在 10 毫秒以内。
还有一个细节:content字段存的是纯文本,raw_html我单独存了一个文件,没有入库。原因是 HTML 体积大,入库会让数据库文件膨胀得很快,而且我很少需要回溯原始 HTML。如果确实需要,按id去文件目录里找就行。这个取舍的逻辑是:数据库存高频访问的结构化数据,非结构化的大文本放文件系统。
3.4 分发层:多终端推送的适配策略
分发层要解决的是“怎么把内容送到我面前”。我的阅读场景有三个:手机通勤、电脑办公、平板深度阅读。对应的推送渠道是:微信服务号模板消息、邮件简报、RSS 输出。
微信推送我用的是服务号的模板消息接口,每天早中晚各推一次,每次推 5 条精选。精选的逻辑是:按来源权重和分类权重综合打分,取 top 5。来源权重是我手动配的,比如官方博客权重 1.0,技术社区权重 0.8,聚合类源权重 0.5。分类权重按我当前关注的重点动态调整,比如最近在搞 Agent 开发,Agent 框架类的权重就调高。
邮件简报用的是 SMTP 直接发,格式是 HTML,包含标题、摘要、链接三要素。邮件的好处是可以存档,方便回溯。我设了一个规则:每天推送的内容自动抄送一份到一个专门的邮箱文件夹,月底导出成 Markdown 归档。
RSS 输出是给阅读器用的,我用的是 Flask 起了一个简单的服务,把最新内容渲染成 RSS 格式。这样我可以在任何支持 RSS 的阅读器里订阅自己的聚合结果,灵活性最高。这个服务的代码不到 50 行,核心就是用feedgen库生成 XML。
4. 实操过程与核心环节实现
4.1 环境准备与依赖安装
先说环境。我用的是 Ubuntu 22.04,Python 3.10。为什么不追新用 3.12?因为有些文本处理库对 3.12 的支持还不完善,3.10 是当前最稳的版本。依赖管理用 pip + requirements.txt,没有上 poetry 或 conda,因为我的依赖不多,pip 够用。
核心依赖清单:
feedparser==6.0.10 requests==2.31.0 beautifulsoup4==4.12.2 readability-lxml==0.8.1 sentence-transformers==2.2.2 jieba==0.42.1 flask==3.0.0 feedgen==1.0.0 requests-cache==1.1.0安装的时候有个坑:sentence-transformers 会连带装 torch,体积很大,下载慢。如果你的服务器在国内,建议先配好 pip 镜像源,能省不少时间。另外 torch 装 CPU 版就行,不需要 GPU,因为我的语义去重是离线批量跑的,CPU 足够。
数据库初始化用一段 SQL:
CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, url TEXT UNIQUE NOT NULL, source TEXT NOT NULL, category TEXT, published_at DATETIME, content TEXT, summary TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_published ON articles(published_at); CREATE INDEX IF NOT EXISTS idx_category ON articles(category);注意:
url字段一定要加 UNIQUE 约束,这是精确去重的第一道防线。我在代码里用INSERT OR IGNORE语句,重复的 URL 会自动跳过,不用手动查重。
4.2 采集任务的定时调度配置
采集任务我用 cron 调度,每 30 分钟跑一次。crontab 配置如下:
*/30 * * * * cd /home/user/ai_aggregator && /usr/bin/python3 collect.py >> logs/collect.log 2>&1 0 8,12,20 * * * cd /home/user/ai_aggregator && /usr/bin/python3 push.py >> logs/push.log 2>&1 0 3 * * * cd /home/user/ai_aggregator && /usr/bin/python3 cleanup.py >> logs/cleanup.log 2>&1三条任务分别是:采集、推送、清理。清理任务是每天凌晨 3 点跑,把 30 天前的原始 HTML 文件删掉,数据库里的记录保留但清空content字段,只留标题和摘要。这样数据库不会无限膨胀。
为什么采集间隔是 30 分钟而不是更短?因为我的源里更新频率最高的也就每小时几条,30 分钟足够覆盖。设太短反而浪费资源,还可能触发源站的限流。这个间隔要根据自己的源列表来定,没有标准答案。我的经验是:先统计一周内各源的更新频率,取中位数作为采集间隔的参考。
4.3 语义去重的参数计算与调优
语义去重这块值得展开讲,因为参数调优直接影响效果。核心参数有两个:相似度阈值和向量模型的选择。
阈值我最终定在 0.85,这个数字是这么算出来的:我手动标注了 200 对内容,其中 100 对是真正重复的(同一事件的不同报道),100 对是相关但不重复的(同一主题的不同事件)。然后用模型算相似度,画 ROC 曲线,找最佳切分点。0.85 对应的准确率是 92%,召回率是 88%,综合效果最好。如果阈值降到 0.80,召回率升到 95% 但准确率降到 85%,会误杀一些有价值的内容;升到 0.90,准确率到 96% 但召回率降到 78%,会漏掉一些重复内容。
模型选择上,我对比过三个:paraphrase-multilingual-MiniLM-L12-v2、distiluse-base-multilingual-cased-v2、text2vec-base-chinese。第一个速度最快,效果也够用,最终选了它。第二个效果略好但慢 30%,第三个对中文优化更好但英文内容处理差。我的源里中英文各半,所以选了平衡性最好的第一个。
计算过程是这样的:先把每篇文章的正文截取前 500 字(太长的部分对相似度判断贡献不大),用模型转成 384 维向量,然后两两计算余弦相似度。为了控制计算量,我只在“同一分类内”做两两比对,跨分类的不比对。这样计算量从 O(n²) 降到 O(n²/k),k 是分类数。实测下来,每天 1000 条新内容,去重计算耗时约 2 分钟,可以接受。
4.4 摘要生成的 prompt 调优与成本控制
摘要生成是唯一需要调外部 API 的环节,所以成本控制很重要。我算过一笔账:每篇文章的 prompt 加正文约 800 token,输出 100 token,按当前主流模型的价格,每篇成本约 0.002 元。每天 1000 篇就是 2 元,一个月 60 元。这个成本可以接受,但如果源列表扩大十倍,成本就上去了。
控制成本的手段有三个:只对精选内容做摘要、批量请求、缓存结果。第一个手段最有效:我先用规则引擎给内容打分,只有分数超过阈值的才送去做摘要,大概占全部内容的 30%。第二个手段是把多条内容拼成一个请求,让模型一次生成多条摘要,能省 20% 左右的 token。第三个手段是缓存,同一篇文章的摘要只生成一次,后续直接读缓存。
prompt 调优我踩过一个坑:早期我让模型“总结这篇文章”,结果模型经常输出“这篇文章讲述了……”这种废话开头。后来改成“直接输出摘要内容,不要加任何前缀”,效果好多了。另一个坑是模型有时候会编造数据,比如原文说“提升了 20%”,模型写成“提升了 30%”。解决办法是在 prompt 里加一句“所有数据必须来自原文,不得编造”,能大幅降低幻觉率。
5. 常见问题与排查技巧实录
5.1 采集失败:源站改版、限流、编码问题
采集失败是最常见的问题,我整理了一个排查表:
| 现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 返回 403 | 缺少 User-Agent 或被识别为爬虫 | 检查请求头 | 加 User-Agent,降低频率 |
| 返回 429 | 触发限流 | 看响应头 Retry-After | 加缓存,延长采集间隔 |
| 内容为空 | 网页结构改版 | 对比 raw_html 和提取结果 | 更新抓取规则或换 readability |
| 乱码 | 编码识别错误 | 检查响应头 Content-Type | 手动指定 encoding |
| 超时 | 源站响应慢 | 看日志耗时 | 设 timeout,失败跳过 |
编码问题我遇到过好几次,最典型的是某站点返回的 Content-Type 是text/html但实际编码是 GBK,requests 默认按 UTF-8 解码就乱码了。解决办法是用chardet库自动检测编码,或者手动指定resp.encoding = 'gbk'。这个坑不常遇到,但遇到一次能折腾半天。
提示:采集日志一定要记详细,包括源名称、URL、耗时、状态码、错误信息。我用的日志格式是
[时间] [源名] [状态] [耗时] [错误],出问题时 grep 一下就能定位。
5.2 去重误杀:相关内容和重复内容的边界
去重最大的风险是误杀。我遇到过两次:一次是把两篇“同一模型的不同评测”判为重复,另一次是把“同一公司的两个产品发布”判为重复。这两次都是因为相似度刚好卡在阈值附近。
解决办法有两个:一是引入时间窗口,只有发布时间相差 24 小时以内的才做语义去重,超过 24 小时的不比对。这样能避免把不同时间的事件误判为重复。二是保留人工复核入口,被去重的内容不直接删除,而是标记为duplicate状态,我可以在后台看到并手动恢复。这个机制救过我好几次,强烈建议加上。
另一个经验是:去重结果要记录相似度分数,方便后续调优。我在数据库里加了一个similarity字段,记录每条内容与哪条重复、相似度多少。这样当我觉得去重效果不对时,可以查这个字段,看看是不是阈值设得有问题。
5.3 推送打扰:如何平衡及时性和阅读体验
推送太频繁会变成打扰,太稀疏又会漏掉重要信息。我调了好几轮才找到平衡点。目前的策略是:早中晚各推一次,每次 5 条,重要内容即时推。
“重要内容”的判定标准是:来源权重 1.0 且分类权重前二,或者标题里包含我预设的紧急关键词(比如“发布”“开源”“重大更新”)。这类内容不走定时推送,而是即时推。即时推的通道和定时推分开,用的是另一个模板消息,避免混在一起。
阅读体验这块,我做了两件事:一是摘要控制在 100 字以内,手机上两屏能看完;二是每条内容带一个“稍后读”按钮,点一下就把链接存到待读列表,不会打断当前阅读。这个按钮的实现很简单,就是一个带参数的 URL,点开后往数据库写一条记录。
注意:推送时间要避开休息时段。我最初设的是早 7 点推,结果发现那个点我还没起床,推送就被淹没了。后来改成早 8 点,正好是通勤时间,打开率明显提升。这个要根据自己的作息调,没有通用答案。
5.4 数据库膨胀:清理策略与归档方案
SQLite 单文件数据库用久了会膨胀,我的数据库从最初的几 MB 涨到了 2GB。主要占用是content字段,每篇文章的正文平均 5KB,10 万篇就是 500MB。加上索引和 WAL 文件,2GB 是合理的。
清理策略我分三层:热数据、温数据、冷数据。30 天以内的是热数据,保留完整内容;30 天到 180 天的是温数据,清空content只留标题和摘要;180 天以上的是冷数据,导出成 JSON 文件后从数据库删除。这个策略跑下来,数据库稳定在 500MB 左右,查询性能没有明显下降。
归档方案我用的是按月导出,每个月 1 号把上个月的数据导成 JSON Lines 格式,压缩后存到对象存储。这样既保留了历史数据,又不占用数据库空间。导出脚本很简单,就是SELECT * FROM articles WHERE created_at BETWEEN ...然后逐行写 JSON。
6. 我踩过的坑和几条实在经验
第一版平台我犯的最大错误是没有做失败隔离。一个源挂了,整个采集任务就中断,后面的源也不跑了。后来改成每个源独立 try-except,一个源失败不影响其他源,采集完成率从 70% 提升到 99%。这个改动只花了半小时,但效果立竿见影。所以我的第一条经验是:任何涉及外部依赖的环节,都要做失败隔离和降级处理。
第二个坑是过度依赖单一信息源。早期我特别依赖某个技术社区,结果它改版后抓取规则失效,我的平台断更了一周。后来我把源列表扩充到 40 个,并且做了源健康度监控,连续失败三次就发告警邮件。现在即使某个源挂了,整体内容量也不会明显下降。
第三个坑是摘要质量不稳定。同一个 prompt,模型有时候输出很好,有时候输出废话。我试过换模型、调 temperature、加 few-shot 示例,最后发现最有效的办法是加一个后处理环节:检查摘要长度是否在 50 到 150 字之间,是否包含“本文”“这篇文章”等禁用词,是否和原文有足够的关键词重叠。不满足就重新生成或降级。这个后处理环节把摘要可用率从 80% 提升到 95%。
最后一个经验是关于迭代节奏的。我见过很多人搭这类平台,一上来就想做全功能,结果三个月还没上线。我的做法是:第一版只做采集和存储,能看就行;第二版加去重和分类;第三版加摘要和推送。每版都能用,每版都有反馈,迭代起来心里有底。先跑通最小闭环,再逐步优化,这个原则适用于所有个人项目。
平台跑到现在快一年了,每天稳定处理 800 到 1200 条内容,推送 15 条精选,我自己的信息获取效率至少提升了三倍。后续我打算加两个功能:一是基于阅读行为的个性化排序,点开过的类别权重自动调高;二是多设备同步已读状态,手机上读过的电脑上不再显示。这两个功能都不复杂,等有空了慢慢加。