ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

基于LLM与记忆系统的行业资讯Agent:从RSS聚合到每日简报的自动化实践

基于LLM与记忆系统的行业资讯Agent:从RSS聚合到每日简报的自动化实践 做这个Agent之前我先交代一个背景这不是我第一次碰自动整理行业资讯这类需求。在此之前我先后用纯RSS订阅、规则脚本两代方案扛了将近一年每次都因为维护成本过高而放弃。直到我把LLM、工具调用、记忆系统组合到一起重写成了第四个版本它才真正变成了每天早上雷打不动给我产出日报的东西。这个Agent 四项目解决的核心问题很简单把行业内散落在几十个信息源里的有效信息自动采集、去重、摘要、分类、排序最后生成一份像真人编辑整理过的每日简报并且随着使用越来越贴合我的关注偏好。这篇博文适合两类人看。一类是想做资讯聚合但受限于不知道从哪下手的人另一类是正在研究Agent架构与编排、想知道一个能长期稳定运行的Agent到底该怎么设计的人。我把实现的思路、代码、踩坑过程都摊开讲你可以直接照搬也可以只拿走里面的部分环节。1. 在动手写代码之前先说说前两代工具是怎么把我逼疯的1.1 第一代纯RSS订阅的问题不是技术是信息过载最早我开着Feedly订阅了大概30个行业网站和技术博客的RSS。听起来很省事实际上坚持不到一个月就废了。每天打开阅读器未读数稳定在三位数我根本没有耐心逐条看完。更麻烦的是RSS源里混杂着大量重复转载、活动通知、版本发布公告真正值得看的行业分析往往被淹没在噪音里。时间一长我形成了严重的未读焦虑最后索性不再打开它。这一轮给我的教训是信息获取工具的核心价值不是拿到所有信息而是在正确的时间给用户提供极少量高价值信息。阅读器天然做不到这一点它只是把问题从我去找变成了我等推送筛选负担一样落在我头上。1.2 第二代规则脚本死在规则爆炸上第二代我开始写脚本用关键词白名单加正则去过滤RSS条目。刚开始很爽命中关键词的文章会进我的待读清单误报率也还能接受。但两周之后我开始频繁调整规则这个关键词太泛了要删那个来源的标题格式特殊要加正则某个话题需要用排除词兜底……规则文件越写越长从几行膨胀到上百行。真正压垮我的是一次关键词打架事件。我同时设置了AI Agent和智能体两个规则结果所有同时含两个词的文章都被重复处理我在去重逻辑上又打了一堆补丁。最后规则库的复杂度已经接近一个小型业务系统维护它每天要花掉我半小时于是我把它整个删了。第二代方案让我明白了另一件事规则的维护成本会随时间线性甚至指数增长而好的筛选应该是理解语义后过滤不是靠关键词死磕。1.3 第三代的尝试半自动就够用了并不够第三代我偷懒做了个半自动版本脚本负责收集和初筛每天把候选清单发到我的聊天工具里我手动挑出有价值的存进笔记。这个方案确实减轻了找信息的压力但每天处理清单本身变成了一种隐性任务。有一次连续出差三天没看清单累积了七十多条我彻底摆烂再没打开过。到了这一步我对第四代的需求已经非常清晰它必须全自动完成从采集、筛选、摘要到输出整份日报的闭环我需要投入的精力下限应该降到只读成品最多偶尔点个赞点个收藏来反馈偏好。在需求已经如此明确的背景下我开始设计这个真正的Agent而不是又一个带了LLM的脚本。2. Agent四的整体架构一条从原始信息到成品日报的流水线2.1 数据流设计六道工序环环相扣整个系统的数据流可以用一条链路说清楚触发调度、采集抓取、清洗去重、LLM理解与加工、偏好排序、生成日报。其中每一步都依赖上一步的输出所以我在设计时定了一个原则每个环节必须保证输出是结构化、可校验的数据绝对不把LLM生成的文本直接传给下一个环节。具体来说采集层输出的是标准化后的Article对象包含标题、来源、链接、发布时间、正文摘要、作者等字段清洗去重层输出的是确认没有见过的候选文章列表LLM加工层输出的是一份带有分类、重要度评分、一句话摘要、标签的StructuredRecord。每条数据都有严格的字段约束和校验逻辑任何一个环节出问题都会被及时发现而不是让错误一路传递到最后变成一份看起来正常实则缺头少尾的日报。2.2 框架选型为什么我最终没有用现成Agent框架在技术选型时我认认真真评估过市面上主流的Agent框架。LangGraph的多节点编排、CrewAI的多角色协作者、AutoGPT的全自动任务分解都各有特色但逐一试用后我放弃了原因有三点。第一这个任务的链路是相对固定的并不需要模型在运行时动态决定下一步做什么。固定流程用框架的状态图来表达反而增加复杂度。第二框架层引入的抽象在排障时要多绕一层对于只有我一个人维护的项目简单直接就是最高的工程美德。第三我需要自定义的部分比如布隆去重、向量召回、调度状态机框架都不会主动给我我依然要写大量胶水代码。所以最终我选了轻量自研加少量辅助库的方案也就是业内常说的裸Agent加Harness模式。我自己定义Agent的核心循环收集输入、调用工具、组织上下文、交LLM生成输出、校验结果。Harness负责调度、状态管理、重试、日志、和外部系统的对接。这套组合既有Agent的智能又有传统脚本的稳定性。2.3 项目目录结构清晰到三个月不看还能秒上手的程度我没有用很复杂的工程结构但目录划分花了心思核心目标是一个新成员甚至三个月后的我自己能快速定位问题。agent4/ ├── scheduler.py # 调度入口负责触发整个工作流 ├── collector/ │ ├── sources.py # 信息源清单管理 │ ├── fetcher.py # 抓取RSS/网页 │ └── cleaner.py # 字段清洗、HTML标签剥离 ├── dedup/ │ ├── bloom.py # 布隆过滤器去重 │ └── index.py # 已见文章索引表 ├── processor/ │ ├── summarize.py # LLM摘要与分类 │ ├── schema.py # 结构化输出模型定义 │ └── prompt.py # Prompt模板 ├── memory/ │ ├── store.py # 向量库操作 │ └── feedback.py # 用户反馈记录 ├── reporter/ │ ├── render.py # Markdown/HTML日报渲染 │ └── deliver.py # 推送渠道适配 ├── state/ │ ├── db.py # SQLite状态库 │ └── retry.py # 重试逻辑 └── main.py # 工作流编排主文件这个结构搭配对应的状态数据库和日志体系整体跑下来让我觉得维护成本低了很多。接下来我按环节拆开讲每个模块的实现和注意点。3. 采集层拿不到干净的数据后面的LLM再强也白搭3.1 信息源管理与RSS解析的细节信息源管理我一开始直接写在列表里后来发现不行。有些站点改版会导致RSS地址失效有些站点一个月才更新一篇文章统一按固定频率抓取很浪费。我用一个JSON文件管理所有源给每个源配置抓取频率和优先级。{ sources: [ { name: example_tech, url: https://example.com/feed.xml, type: rss, interval_minutes: 60, priority: 1, categories: [AI, Agent] } ] }RSS解析我用的是Python的feedparser库它处理各种RSS变体和Atom格式都很稳能解析出标题、链接、发布时间、作者、标签和正文描述。解析后我会做一个字段清洗把HTML标签去掉把相对链接转成绝对链接把各种时间格式统一成ISO8601。这些看似不起眼的步骤决定了后面LLM拿到的数据质量。需要注意的一个坑是部分站点的RSS提供的是摘要而不是全文摘要内容经常被截断到一半。我做了个策略优先抓原文正文用httpx带浏览器UA去请求文章页面再用Readability算法抽取正文。抓正文这件事会显著增加请求量所以不是所有源都开只对白名单里的高价值源开启。3.2 去重布隆过滤器加字段级碰撞双保险资讯类Agent最烦的就是同一篇文章被多个源转发或者同一个事被不同媒体各报一遍。我一个人盯不过来的事必须由系统在源头解决。我做了两层去重。第一层是布隆过滤器对每篇文章的URL和标题的归一化结果计算哈希判断是否曾经见过。布隆过滤器存在误判率但它占用内存极小、查询极快可以挡掉绝大多数的重复采集。第二层是字段级碰撞检测。布隆只能判断见过完全一样的东西但很多转载会改标题。我对清洗后的正文前200个字做SimHash再和近三天的文章索引做海明距离比对相似度超过阈值就认定重复。这样即使标题和URL完全不同只要正文大段一致一样能识别出来。代价是多了一次近线计算但日均信息量在两三百篇的规模下这个开销完全可以忽略。3.3 请求频率、超时与失败降级抓取外部源必须讲礼貌。我统一设置了请求头把单源抓取间隔控制在合理范围默认超时10秒重试一次再失败就跳过本轮。因为跑的是个人项目不能因为一个源挂了就把整条链路卡死。还有一个容易忽略的点部分站点对无头浏览器或者高频率请求会返回验证页面。我维护了一个疑似被拦截的检测列表把返回内容里的特征词和状态码组合判断如果某个源连续三次被拦截就自动暂停抓取并在日志里告警。涉及到真实用户场景时爬虫需要尊重网站的robots协议和服务条款我所以有合规风险的信息源一律不接入绝不使用绕反爬手段。这个底线保证项目可以长期安全运行。4. LLM加工环节让模型干编辑的活而不是让它自由发挥4.1 Prompt设计思路稳定输出比文采更重要当采集清洗之后剩下的任务是把每篇候选文章变成一条标准化的记录分类、重要度、摘要、关键词、涉及实体。这部分我交给LLM来做但Prompt花了很多轮迭代。我的Prompt模板大体分成三块角色设定、输入数据、输出约束。角色设定是你是一位深耕行业多年的资深编辑输入数据给定文章的标题和正文输出约束是最关键的部分要求模型严格按照JSON格式返回不要输出任何解释、前缀或Markdown代码块标记。我还在Prompt里放了一条如果你不确定分类返回uncategorized不要猜避免模型强行归类导致后续排序错乱。实际操作中我发现了一个反直觉的点在系统提示里要求简练不如给一个摘要不超过80字的字数上限因为前者模型理解很模糊后者能直接约束token数量。同样要求找出重要信息不如明确说根据行业关注度、时效性、影响范围三个维度打分分值1到5可操作性强很多。4.2 结构化输出JSON Schema与容错解析LLM调用我走的是OpenAI兼容接口在请求参数里带了response_format为json_object。现在的模型只要接口支持基本能保证返回合法JSON但依然有概率在特殊场景下翻车所以我第一道防线就是JSON解析容错。import json from pydantic import BaseModel, ValidationError class ProcessedRecord(BaseModel): category: str importance: int summary: str tags: list[str] entities: list[str] def parse_llm_result(raw: str) - ProcessedRecord | None: try: data json.loads(raw) return ProcessedRecord(**data) except (json.JSONDecodeError, ValidationError): return None如果解析失败我会重试一次把上一次的原始输出追加进Prompt让它检查JSON格式后重新输出。第二次还失败就直接丢弃这条记录并写进失败日志。这种解析失败宁可丢弃也不硬修的策略是为了保证整条数据链路的质量不会因为一条垃圾数据被污染。4.3 分类打标与聚合逻辑每篇记录被LLM打上分类和标签后日报生成的逻辑就清晰了。我维护了一套固定的分类体系先按分类聚合分类之下再按重要度排序。同时系统会把同分类里被多个来源报道过的内容识别为热点在日报里用今日热点小节突出展示。聚合逻辑的代码并不复杂但它依赖前面每一步都干净。实践证明只要分类标签打准了聚合渲染端就是纯模板活不怎么需要改。真正要求反复调整的是Prompt和分类体系的边界比如行业动态和产品发布两个分类的边界如果模糊模型就会随机归类这时候应该回到Prompt里做例子补充而不是在聚合层写一堆特殊规则。5. Agent记忆把偏好沉淀下来让它越用越懂你5.1 短期记忆与长期记忆的边界Agent能不能越用越懂你关键看记忆设计。我参考了认知科学里短期和长期的划分在系统里做了两层。短期记忆是当前这一轮日报生成过程中产生的上下文比如本批所有候选文章的摘要和评分它只存在于单个执行周期内跑完就清空。长期记忆是跨周期的用户偏好和反馈历史存在向量数据库里每一轮都会被检索出来参与排序。这两层记忆如果不分开很快会导致整个上下文爆炸。尤其经过几十天的运行积累的反馈记录会有数千条全部塞给模型既不经济也容易让模型注意力分散。5.2 向量召回实现用偏好样本校准重要度长期记忆的实际使用方式是每当用户对某条日报里的文章表达反馈比如标记为收藏、点开阅读、或者忽略系统会把这篇文章的标题、摘要、标签向量化以后存入向量库并附上反馈类型作为元数据。到了下一轮日报生成时我会对当天的候选文章做一次向量检索找出与用户历史偏好最相近的文章把这些匹配度得分作为最终重要度的加权因子。比如一个用户长期标注收藏了关于AI推理成本的深度分析文章那么系统之后遇到相关话题时会自动把它的重要度上调而如果用户反复忽略某类产品发布类内容同类文章的权重就会逐步下降。这种机制的本质是把用户反馈变成可计算的偏好向量而不是靠人去维护规则。5.3 反馈回路一个不打扰用户的设计让用户每天额外去给文章打好评差评是很不现实的。我的做法是让反馈行为尽量无感日报推送到聊天工具里以后点击链接打开文章算正反馈在日报里点收藏算强正反馈连续多天出现在日报里但从未被点开的链接会被打折处理。这些反馈信号会被日志系统自动记录下来不需要用户做任何主动操作久而久之长期记忆库就会沉淀下一份相当准确的兴趣画像。这个设计帮我解决了前面几代工具最大的问题就是工具不懂我因为现在它真的在观察我的行为。6. 定时调度与异常恢复连续跑一个月不崩的工程细节6.1 调度方案为什么我放弃了crontab刚开始我用系统自带的crontab做定时触发后来发现一个问题调度进程本身没有被托管只要某次运行时网络断开或者依赖服务重启那次任务就会静默失败而crontab不会自动补跑。所以我换成了Python进程常驻加APScheduler的方式。APScheduler支持CronTrigger可以精确控制每天哪个时段执行。进程常驻的好处是错误日志、状态检查、手动触发都方便还有一个优势是我可以在同一个进程里注册多个触发节点比如早中晚各跑一次增量采集晚间执行完整日报生成。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler() def daily_info_job(): orchestrator DailyInfoOrchestrator() orchestrator.run() scheduler.add_job( daily_info_job, triggerCronTrigger(hour7, minute30, timezoneAsia/Shanghai), iddaily_info, replace_existingTrue, )6.2 幂等设计与失败重试宁可重跑不可脏跑定时任务最怕的不是失败而是重复执行产生脏数据。每次运行开始时会生成一个run_id作为本次执行的唯一标识所有采集、加工、推送记录都会带上这个run_id。如果某个环节失败系统会在状态库里标记当前run_id的状态为partial然后根据失败类型决定是原地重试还是全部重来。我的重试策略分三层。第一层是单次网络请求的短暂重试间隔5秒到30秒第二层是整个采集环节的批次重试比如某个源连续失败就跳过该源进入下一轮第三层是日报生成失败后的整体重试每天最多重试三次第三次还失败就把收集到的原始候选文章打包发给我由我自己手动处理。这个兜底通道看起来很简单但它保证了我从来没有出现过某一天没有任何输出的情况。6.3 状态机与可观测性像监控线上服务一样监控个人Agent整个执行流程被我抽象成一个简单的状态机节点包括pending、collecting、deduping、processing、rendering、delivered和failed。每个节点切换都会写进SQLite状态库配上时间戳和运行指标。可观测性做得很重因为我想知道这个Agent每天都在干什么。每轮执行结束后会生成一份元数据报告包含本轮抓到的文章数、去重数、LLM耗时、调用token数、各环节延迟等。这些数据不仅帮我定位问题也为后续优化提供了依据。有一次我发现某几个源的采集延迟特别高一查是对方的CDN节点慢于是我把这些源调整到了空闲时段抓取整个链路就顺畅了。7. 连续运行30天的数据复盘与下一步进化方向7.1 数据统计信息过载被压到了一个我可以接受的量级系统已经连续运行了30天我拿到了一些比较有意思的数字指标数值接入信息源数量27个RSS加6个网页源每天原始抓取文章数平均约260篇去重后候选文章数平均约118篇LLM加工后进入日报文章数平均约24篇标为高重要度的文章数平均约5篇我实际点开阅读的比例约61%这个数字对比最直接说明问题之前上千条未读数带来的焦虑消失了每天花在找信息上的时间从一小时以上降到了不到十分钟。我每天早上打开日报重点看高重要度部分其他分类扫一眼标题有感兴趣的再点开。决策质量也变高了因为聚合和评分替代了我大部分低效的浏览动作。7.2 发现的问题分类边界、时效性冲突、过度个性化数据复盘中我也发现了几个还没解决的顽疾。分类边界问题最明显模型经常把公司财报分进行业动态因为很多财报解读同时涉及公司和行业。我试过补充示例和调整分类体系有所改善但边界永远存在。时效性冲突也很典型一条三天的旧闻今天被重新热炒它能拿到很高的重要度分但它其实已经不算新闻了。目前的方案是在发布时间上做指数衰减惩罚但衰减系数还需要继续调。另外过度个性化也有风险记忆系统会让我越来越只看到偏好相关的信息反而错过了一些应该知道的全局动态。我计划在排序中加入一个探索因子每隔一段时间主动引入少量非偏好推荐。7.3 后续可以考虑的几个方向下一步我可能会尝试多Agent协作一个采集Agent专职盯源一个分析Agent专职做深度摘要一个编辑Agent负责最终编排。其实现在这套流水线已经具备雏形只是各环节之间还是串行调用没有真正并行跑起来。引入多Agent之后每个子Agent可以独立重试、独立扩缩容整体吞吐和稳定性还能再上一个台阶。目前这个版本我个人用起来已经足够舒服尤其是记忆模块带来的个性化体验是前两代工具完全做不到的。我也发现这个Agent的四字经验和前几代的区别其实不在模型多聪明而在工程化的细节到位数据流清晰、状态可观测、容错有兜底、记忆可持续积累。如果你也想搭一个自己的资讯Agent优先把这四点考虑进去大概率能少走不少弯路。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进