RSS 智能体怎么获取资讯？解析订阅条目并按稳定 ID 去重

RSS 智能体获取资讯时，先把订阅条目保存成可追溯记录，再交给后续摘要或筛选步骤。同一条资讯会在多个轮次出现；用标题去重容易把改标题的旧条目重复收录，也可能把同名新条目错误丢弃。

下面用 feedparser 解析 RSS，并以“订阅源 + 条目 ID”作为 SQLite 唯一键，验证重复导入不会增加记录。RSS 是人为构造的演示输入，已在 Windows、Python 3.11.15、feedparser 6.0.14 环境运行；没有连接真实资讯源，也没有调用模型生成摘要。

先分清采集、去重与 AI 摘要

feedparser 基础文档展示将 RSS/Atom 内容解析为 feed 与 entries。本文完成的是智能体的资讯采集入口：接收一份订阅内容，解析条目并持久保存；定时请求、模型摘要和发布需要在后续步骤单独实现。

entry.id 参考说明该字段来自 RSS guid、Atom id 等标识。本例优先使用 entry.id，没有时退回 entry.link；两者都不存在就报错，不用当前时间或标题临时拼出一个“稳定 ID”。

唯一键同时包含 source_feed，避免不同订阅源使用相同 ID 时互相覆盖。这只在每个源内部去重；同一新闻被两个来源转载时，仍会保留两条来源记录。跨源判重是另一个任务，不宜直接把出处抹掉。

对异常订阅先停止导入

Bozo Detection 文档解释：解析器发现非良构 XML 时设置 bozo，并可能仍尝试解析部分内容。本例采取严格策略，bozo 为真就拒绝该输入，保留异常供排查，不把部分条目当完整抓取结果。

运行含重复条目的完整示例

安装 python -m pip install feedparser==6.0.14。将下面代码保存为 ingest_rss.py，执行 python ingest_rss.py。程序保存演示 XML，导入 SQLite 两次，并测试不完整 XML 的拒绝分支。

演示共 3 个 item，其中 article-001 出现两次。订阅网址 https://example.invalid/ai-news-feed.xml 是虚构来源标识，没有网络访问；条目内容也不是实际新闻。

from pathlib import Path
import json,sqlite3
import feedparser

feed_url='https://example.invalid/ai-news-feed.xml'
xml='''<?xml version="1.0" encoding="UTF-8"?><rss version="2.0"><channel><title>人为构造的AI资讯订阅</title><link>https://example.invalid/</link><description>仅用于验证采集链路。</description><item><guid isPermaLink="false">article-001</guid><title>演示条目一</title><link>https://example.invalid/a</link><description>这是一条虚构的资讯内容。</description></item><item><guid isPermaLink="false">article-002</guid><title>演示条目二</title><link>https://example.invalid/b</link><description>保留来源，不把摘要当成全文。</description></item><item><guid isPermaLink="false">article-001</guid><title>演示条目一</title><link>https://example.invalid/a</link><description>这是一条虚构的资讯内容。</description></item></channel></rss>'''
Path('sample_feed.xml').write_text(xml,encoding='utf-8')

def ingest(raw,source_url,connection):
    parsed=feedparser.parse(raw)
    if parsed.bozo:
        raise ValueError('Malformed feed: '+str(parsed.bozo_exception))
    inserted=0
    duplicates=0
    with connection:
        for entry in parsed.entries:
            stable_id=entry.get('id') or entry.get('link')
            if not stable_id:raise ValueError('Entry has neither a stable ID nor link')
            record={'title':entry.get('title',''),'link':entry.get('link',''),
                    'summary':entry.get('summary',''),'published':entry.get('published'),
                    'source_feed':source_url,'source_entry_id':stable_id}
            cursor=connection.execute('INSERT OR IGNORE INTO items(source_feed,entry_id,payload) VALUES(?,?,?)',
                (source_url,stable_id,json.dumps(record,ensure_ascii=False)))
            if cursor.rowcount==1:inserted+=1
            else:duplicates+=1
    return inserted,duplicates

connection=sqlite3.connect('rss_items.sqlite3')
connection.execute('CREATE TABLE IF NOT EXISTS items(source_feed TEXT,entry_id TEXT,payload TEXT,PRIMARY KEY(source_feed,entry_id))')
raw=Path('sample_feed.xml').read_bytes()
first=ingest(raw,feed_url,connection)
second=ingest(raw,feed_url,connection)
records=[json.loads(row[0]) for row in connection.execute('SELECT payload FROM items ORDER BY entry_id')]
assert len(records)==2
assert second==(0,3)
assert all(x['source_feed']==feed_url for x in records)
try:ingest(b'<rss><channel>',feed_url,connection)
except ValueError:print('malformed_feed_rejected:',True)
else:raise AssertionError('Malformed feed accepted')
Path('rss_source_records.json').write_text(json.dumps(records,ensure_ascii=False,indent=2),encoding='utf-8')
print('first_import_new_duplicate:',first)
print('second_import_new_duplicate:',second)
print('stored_items:',len(records))
print('saved: rss_source_records.json')
connection.close()

本地运行输出

malformed_feed_rejected: True
first_import_new_duplicate: (2, 1)
second_import_new_duplicate: (0, 3)
stored_items: 2
saved: rss_source_records.json

如何确认重复导入没有新增

在本例首次创建数据库时，第一轮新增 2 条、识别重复 1 条；第二轮新增 0 条、识别重复 3 条，库中最终为 2 条。程序还验证不完整 XML 被拒绝，并把已保存记录导出为 rss_source_records.json。

数据库是持久文件，因此再次执行整个脚本时，第一轮也可能新增 0 条。保留原库并检查第二轮新增为 0、存储总数仍为 2，就能判断重复导入是否正常；不用清空数据来追求固定的首次输出。

打开 rss_source_records.json，检查每条记录的 source_feed、source_entry_id、link、title 和 summary。本文样例没有发布时间，published 因而为空；实际源未提供时间时，也不能用采集时间冒充新闻发布时间。

接入真实订阅源时改什么

1. 删除创建演示 XML 的部分，保留 ingest 函数与数据库定义。
2. feed_url 设置为真实订阅网址，raw 改为获取后保存的原始订阅字节。下载阶段单独确认 HTTP 成功、超时和内容格式，失败时先停下，不把错误页当订阅导入。
3. 保留原始订阅和获取时间，先对少量真实条目检查 ID 与链接，尤其留意没有 guid 或会改变链接的来源。
4. 把同一个订阅地址作为稳定的 source_feed 使用；地址迁移时先制定映射，否则同一条目可能被当成来自新源。
5. 让后续摘要步骤读取新增记录，并把原链接、源 ID 与生成结果一并保存。模型摘要之后仍需核对原始资讯内容。

新增去重不等于追踪更正

本例使用 INSERT OR IGNORE，只处理首次新增；已有 ID 的文字被修改时，会保留旧 payload，而不会更新内容。如果需要跟踪更正，另存内容哈希和版本，再明确何时更新当前版本、何时保留历史。不能把第二轮没有新增解释为“全部资讯都没有变化”。

没有稳定 ID 而退回 link 时，链接变化可能产生新记录；标题变化不会改变已有 ID 的去重结果。一个源错误复用同一 ID 时，也会导致新条目被忽略，因此接入新源要用真实数据检查 ID 质量。

RSS summary 可能是摘要，也可能带 HTML，不保证是全文。做 AI 摘要前先确认它是否包含任务需要的信息；需要全文时，另行读取原文并保留出处，不让模型根据短摘要补造事实。本例没有提取链接指向的文章正文。

RSS 空列表能当成今天没有新资讯吗？

不能只凭 entries 为空就下结论。先确认请求成功、拿到的是订阅内容且没有解析异常，再判断源本身是否没有条目。采集失败和没有更新应记录成不同状态，后续智能体才不会生成误导性的“今日无更新”。