拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

搭建AI日报自动化流水线:从采集到分发的全流程解析

搭建AI日报自动化流水线:从采集到分发的全流程解析

每天早上醒来第一件事不是睁眼,是刷AI资讯——公众号、微博、arXiv、Hacker News、Product Hunt、少数派、机器之心……我算过,零零散散加起来有二十多个源。刷到快迟到的时候,脑子里还是一团乱麻,昨天到底发生了什么大事?说不清楚。这不是自律问题,是信息整理效率问题。

为了把自己从这种低效循环里捞出来,我花了一个周末搭了一条自动化流水线:每天定时抓取全网的AI动态,清洗去重后用大模型精炼成一份结构化行业日报,早上8:30准点推到我的手机上。到现在跑了150多天,每天稳定产出,几乎没有断更过。这篇就把整条流水线的设计、代码逻辑和踩坑过程都拆开讲清楚,给也有同样需求的人做个参考。

1. 为什么我要折腾一条AI日报流水线

先说清楚动机,这决定了你做出来的东西长什么样。

我的痛点是典型的“AI从业者信息过载”。我不是只看新闻,还要看论文、开源项目、产品发版、融资动态、政策信号。这些信息分散在不同平台:论文在arXiv,开源在GitHub,产品在Product Hunt,评论在Hacker News,中文深度解读在公众号。想靠人肉每天把这么多源刷完,少说一个半小时,而且刷完就忘,根本沉淀不下来。

市面上其实有大把的AI日报订阅。我订过几个,也让人工智能自动生成过。但这类现成产品的毛病很统一:要么太杂,什么都能塞进去;要么太浅,每条新闻只给一句标题;要么太慢,等我看到的时候,圈里人已经讨论完两轮了。我真正想要的东西,现成产品给不了:它得覆盖论文、开源、产品、投融资这几个硬核维度,每天只要10到15条就够,每条有信息增量而不是复读原文,还必须在早上8点半之前出现。

另外一个点是,这份日报不是只给自己看的。我所在的团队每天早上开站会,日报可以直接成为站会的引子,谁刚好负责相关模块就顺手接下来说两句。后来我也把日报推送给几个朋友的小群,别人觉得“挺有用”,这让我意识到它的价值高于一个普通的RSS阅读器。

所以这条流水线的需求边界,其实从第一天就很清楚:

  • 覆盖范围:论文、开源项目、产品动态、投融资、技术评论,五大类
  • 产出格式:Markdown,10到15条,每条配来源链接和推荐理由
  • 时效要求:8:30前必须可见,失败率必须低于我手动整理的失败率
  • 维护成本:每小时花在它身上的时间不能超过10分钟
  • 运行成本:每天API花费控制在几块钱以内

想明白这些,后面所有技术选型都有了解题方向。说白了,我要的不是“全网所有AI信息”,而是“每天最有信息增量的那十几条”。

2. 流水线整体骨架:采集、清洗、加工、分发四段式

整个系统我用了最简单的管道模型。上一段的输出就是下一段的输入,中间不搞复杂的消息队列,不搞容器编排,就是一个Python脚本管到底,每段用不同函数实现。

这么设计的理由很直接。我只有一台小服务器,也不想为日报单独买一堆基础设施。管道模型最契合这种场景:每个阶段职责单一,坏了只修一段;想换大模型API只动加工段;想增加信息源只在采集段加一行配置。维护成本被锁死在最小。

以下是四段式的责任划分:

阶段职责输入输出主要技术
采集定时抓取各信息源原始数据信息源配置JSON格式原始条目requests, feedparser, RSSHub
清洗去噪、去广告、去重、归一化原始条目干净的结构化候选池BeautifulSoup, simhash
加工用大模型筛选、排序、写推荐理由候选池精炼日报内容大模型API
分发推送到手机/邮箱/IM精炼日报各端可见的最终产物飞书Webhook, SMTP

有人可能会问,为什么不干脆把“采集”这步也交给大模型,让AI自己上网搜?我试过,先说结论:现阶段千万别这么干。原因有三个。

第一是成本。让大模型直接浏览几十个网页并把内容归纳出来,token消耗是每天几百K,成本轻松翻二三十倍。

第二是噪声。你让模型去网上搜AI新闻,它会搜出大量低质量内容,尤其是中文互联网,标题党浓度极高。指望模型在每个网页上自动判断可信度,现阶段还是太理想。

第三是幻觉。模型在处理海量外部信息的时候,特别容易把A媒体的内容和B媒体的细节拼在一起,或者编出原文根本没有的结论。这是模型在长文本上下文里的固有问题,不是换一家API就能解决的。

所以我的原则是:大模型只按我的规则加工,不负责发现信息。发现信息用爬虫和RSS完成,这个分工后来被证明极其省钱省心。规则负责过滤垃圾,模型负责提炼价值,两条腿走路,谁也别替代谁。

3. 信息采集:找到稳定的“源头活水”才算开始

采集是整个流水线的源头,这一环废了,后面再强也白搭。我的信息源分成了五类,每一类的获取方式差异很大。

第一类是RSS源,这是最省力的。机器之心、少数派、arXiv的论文通告、Hacker News的Top Stories、Product Hunt每日榜单,这些要么自带RSS输出,要么可以通过RSSHub生成订阅地址。解析RSS我直接用feedparser,一个函数通吃所有xml:

import requests import feedparser def fetch_rss(url, limit=30): resp = requests.get(url, timeout=10) resp.raise_for_status() feed = feedparser.parse(resp.content) entries = [] for entry in feed["entries"][:limit]: entries.append({ "title": entry.get("title", "").strip(), "link": entry.get("link", ""), "published": entry.get("published", ""), "source": entry.get("source", {}).get("title", url), "summary": entry.get("summary", "")[:500] }) return entries

第二类是开放API。arXiv的官方API可以直接按分类取最新论文,我用它拉cs.AI、cs.CL、cs.LG这几个大类的当日新增。

ARXIV_API = "http://export.arxiv.org/api/query" def fetch_arxiv(date_str, max_results=60): query = "cat:cs.AI OR cat:cs.LG OR cat:cs.CL" params = { "search_query": query, "start": 0, "max_results": max_results, "sortBy": "submittedDate", "sortOrder": "descending" } resp = requests.get(ARXIV_API, params=params, timeout=15) return parse_atom_xml(resp.content) # feedparser也能解析Atom

第三类是GitHub Trending。大家在推特上讨论的很多新项目,源头就在这。我习惯把每天Trending前30的仓库摘要拉进来,再让后续环节筛选。Trending本身没有官方API,目前稳定做法是通过RSSHub的/github/trending路由生成订阅地址。

第四类是微信公众号。这其实是中文AI圈信息密度最高的地方,但也是技术上最麻烦的地方。微信没有公开的RSS,我个人的折中方案是用RSSHub将几个重点公众号转成RSS,再加手工维护的白名单关键词,只保留和AI强相关的内容。这块如果不想折腾,可以直接放弃,换成订阅大量科技媒体的公开RSS,信息覆盖上差不了太多。

第五类是行业垂直源,比如机器之心、量子位、爱范儿、The Verge的AI频道。这些源的消息时效性高,但我把它们的权重调低,因为它们的内容经常是重复的,同一个事件十几个媒体各写一遍。权重低不代表不抓,而是代表在加工阶段劣后处理。

采集频率上,我没有做成每15分钟一次的高频轮询,而是控制在一小时一次增量抓取,然后把当天所有增量汇总成原始候选池。因为日报是每天早上出一次,中间任何时刻抓到的数据都不会实时推送,所以不需要高频率。一天下来,我的候选池大概有400到600条原始条目,但这600条里大部分是不能用的垃圾,所以清洗段的压力就开始变大了。

还有一个绕不开的问题:反爬。很多人一上来就上无头浏览器,模拟点击,结果把服务器IP搞封了,连基础网页都访问不了。我的经验是“先君子后小人”:优先用官方API和RSS,实在没有再考虑对单个页面做针对性解析。对用户来说,保持礼貌的抓取频率,设置合理的User-Agent,基本能处理掉90%的问题。无头浏览器是最后手段,不要在第一版就上。

4. 内容加工:大模型在这里不是“生成”,而是“乱中取序”

清洗之后,候选池里大概还有100到150条看起来像是AI相关的条目。但这个量还是太大,直接塞给大模型也不划算,所以我设计了两层筛选。

第一层是规则筛选。我维护了一个低质量关键词黑名单,像“震惊”“突发”“彻底颠覆”“重磅”“99%的人都不知道”这类词,直接过滤。命中标题或摘要的条目,直接进垃圾桶。还有一套来源评级:权威源(arXiv官方、重要研究机构博客)打分高,综合媒体打分中等,自媒体标题党打分低。规则筛选最大的价值是在进大模型之前把明显没营养的内容干掉,我实测能砍掉40%以上的垃圾,也等于省了40%的API费用。

第二层才是大模型筛选。这一步我的做法是:把候选条目的“标题+摘要+链接+来源”拼成一个结构化的文本块,然后交给我写死的提示词去处理。我用的是通用的大模型API,跑了几款主流模型之后,效果虽然有些细微差别,但流程都一样:

你是AI行业日报主编。以下是今天收集到的候选资讯条目,每条包含:编号、来源、标题、摘要、链接。 请执行以下任务: 1. 筛选:挑出今日最有价值的10至15条。剔除与AI主题无关的内容、明显重复报道、标题党、软文。 2. 排序:按高中低三档重要性排序,同一事件的多篇报道只保留信息增量最大的一篇。 3. 改写:每条用一句话提炼核心信息点,50到80字,必须有信息增量,不要复述标题。 4. 标注:每条输出格式为“【第X条】标题|来源|一句话点评|原始链接”。 5. 规则:所有链接必须来自上文给出的候选条目,绝对禁止自行编造链接或来源。 候选资讯: ...

这里特别要强调“链接必须来自候选条目”这句。不加这个限定的话,模型会给你编造一堆看起来真实但根本打不开的链接,这是我在测试阶段踩得最深的一个坑。加了限定之后,模型输出里的伪链接现象基本绝迹。

还有一个核心原则:让模型重组信息,而不是创作信息。模型从候选集里挑内容、排序、写推荐语,这属于“加工”;让模型“写一段今天AI领域最值得关注的新闻”,这属于“创作”,后果就是把几天前的旧闻当成新闻重新包装,甚至直接编新闻。日报最怕的就是这个。

和单纯用LLM输出不同,头条质量直接影响读者完读率。为了让前三条一定是最重磅的,我在提示词里加了额外的约束:“前三条必须是今天新发布的重大动态,包括但不限于重要论文、大厂产品发布、知名团队新项目。”这相当于人工给模型划了个重点区,效果立竿见影。

所有模型输出,我还会做一道格式校验,确保它确实输出了10到15条、每条都带了链接、链接都在候选池里。校验不过就重试一次,重试再不行就降级成只发前一天的日报加“今日人工补录”标注,绝不裸奔。

5. 早上8:30,调度与发布是怎么串联起来的

加工阶段生成的还是文本,要让它在早上8:30准确出现在我的手机上,得靠调度和发布这两根隐形链条。

调度我用的是cron。脚本本身是纯Python,所以cron只要调起一个入口即可:

30 8 * * * cd /opt/ai-daily && /usr/bin/python3 pipeline.py >> /var/log/ai-daily/run.log 2>&1

这个cron看着简单,里面有一个极其经典的时区坑:大部分云服务器的默认时区是UTC,你写30 8 * * *,得到的是UTC早上8点半,换算成北京时间是下午4点半。我第一版就犯了这个错,第一天8:30没等来日报,下午4点多倒是收到了一份“早报”。解决方法是显式指定时区等级,确保服务器是Asia/Shanghai,或者在cron里加CRON_TZ=Asia/Shanghai,一劳永逸。

发布链路我是分层设计的。优先级最高的是飞书群机器人Webhook,只需要一个POST请求,不用处理各种平台审核,也不依赖手机通知权限。我的发布函数大概是这样的:

import requests FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/xxx" def publish_feishu(content): payload = {"msg_type": "text", "content": {"text": content}} requests.post(FEISHU_WEBHOOK, json=payload, timeout=10)

如果当天什么平台都挂了,最后兜底的是邮件。邮件用SMTP就能发,我特意选了一个极简封装,几十行代码搞定,稳定性比IM webhook高很多。

自动化链路里还有一个容易被忽略的点:失败恢复。cron只管按时启动,它不会管你的脚本跑到一半是不是崩了。所以我在管道入口加了一个整体重试逻辑,每一段单独try/except,一旦某一段抛出异常,整条管道不会中断,而是记日志后继续往下走。发布阶段则做了两层重试:第一次失败,等3分钟再试;再失败,等10分钟;第三次还失败,直接换备用的邮件渠道。

运行日志是必须的,不能省略。我的日志格式很简单:每个阶段的开始时间、结束时间、处理条数、异常堆栈。日常运维时只做一件事,就是每天花30秒扫一眼最终结果。如果当天日报正常推送,日志看都不用看;如果推送失败,日志里通常直接写着失败原因。

整个调度加发布设计完之后,我记得第一次看着手机在8:30整弹出日报通知的时候,那个感觉确实挺奇妙。一套自己搭的代码,取代了一个原本需要一小时的手工活。

6. 跑了150天,我踩过最多的坑和现在的调参心得

要说这套系统跑到现在,真不是一直岁月静好。复盘下来有几个高频问题,每个都是真金白银换来的教训。

坑一:重复报道。一个重磅技术一出来,第二天全球媒体像听到了指令一样,铺天盖地全是同一个消息。清洗环节虽然做了简单的标题字符串查重,但“OpenAI发布新模型”和“OpenAI推出GPT-5级新模型”这俩根本对不上,模型就会当成两条不同新闻输出。后来我加了两层去重:第一层,标题做归一化后算simhash相似度,超过阈值算重复;第二层,让大模型承担“语义去重”职责,在提示词里明确要求“同一事件只保留信息增量最大的一篇”。两层配合下来,日报里基本见不到一件事被说两遍了。

坑二:大模型幻觉链接。这个前面已经提过,是我心里最严重的坑,本质上是提示词设计问题。以前我写的提示词只说“输出原文链接”,模型就开始自由发挥,给你编出一个看起来很像arXiv风格的假网址。后来的解法就是强制限制模型只能从候选池里挑链接,并且在代码层面做了链接白名单校验,一旦发现模型输出了一个不在候选池里的链接,整篇日报重跑。这是底线,绝对不能让读者点开一个打不开的链接。

坑三:API限流和费用失控。一开始我图省事,把几百条候选文本一次性丢给模型,结果是token数爆炸,费用高不说,还经常触发限流,导致重试排队。优化方式是把候选池分成几批:第一批筛选,只给了标题和摘要;第二批点睛,只要第一批选出的优秀条目;第三批评分排序。这样每次API调用的输入都控制在几百token,费用降了三分之二。

坑四:低质内容混入。中文AI新闻领域有一个“标题党浓度特别高”的现象,我观察了好久才意识到,单纯靠黑名单很难彻底拦截那些“看似专业实为营销号”的内容。现在我给来源设置了A/B/C三级评分,A级来源直接进备选,B级来源需要过双重校验,C级来源即使命中关键词也只在当天缺稿时才用。这个配置看上去很原始,但比任何智能算法都稳。

坑五:大模型输出格式漂移。模型不是机器,它今天心情好就按Markdown列表输出,明天心情差就来一段散文。格式不稳定对后续自动发布影响很大。我后来在提示词里把输出格式写成了严格的模板,并且给模型一个输出示例,再在代码里做格式校验。校验失败就再跑一次,还失败就切备胎模型。这个“双保险”保障了日报格式的长期稳定。

最后的调参心得是,不要追求一次到位。这条流水线今天的长相,已经和第一版非常不一样了。第一版只有三个信息源,输出质量一般;现在有二十多个源,筛选逻辑复杂得多。我几乎每周会抽半小时,翻一翻日报的输出,哪类内容少了,哪个来源质量下滑了,就微调一下权重和黑名单。整套系统真正让我受益的,不只是一份日报本身,而是这种“极低成本维护、每日反馈迭代”的节奏。

到现在,我早上已经习惯先刷一眼日报,再决定要不要去深挖某个具体信息。它没有取代我的信息判读,但帮我把每天的信息起点从“一片混沌”变成了“一条清晰的线索”。如果你也面临类似的信息过载,建议先从自己的真实场景出发,搭一个最小可用版本——不要一开始就想做全网覆盖,先把一个类别的信息抓准,跑稳了,再往外扩。

返回列表