先说说我为什么非要自己动手。每天刷 AI 资讯的姿势我曾经是早上看一遍科技媒体中午刷 Hacker News晚上再扫一圈 arXiv 和 GitHub Trending。听起来很自律实际大部分时间花在重复阅读上——同一篇新闻换个标题出现五六次真正有深度的长文被淹没在快讯里。所以我花了两周时间自己搭了一个 AI 资讯聚合平台RSS 和 API 自动抓取、正文抽取、去重、AI 摘要、自动打标签最后统一在一个网页里浏览。这篇就把完整的架构设计、代码实现、部署和踩坑过程写清楚适合有点编程基础、想重新夺回信息流控制权的人也适合想给团队做 AI 情报系统的朋友抄作业。1. 为什么自己搭信息的“策展权”不能全交给别人1.1 现成资讯服务的三个痛点“聚合”这件事市面上已经有太多产品但用下来总有三个绕不开的痛点。第一个是信息过载。平台为了覆盖足够多的用户会把各种来源、各种质量的资讯涌到你面前算法推荐还会不断投喂同质化内容。你看到的不是“今天 AI 领域发生了什么”而是“平台希望你觉得 AI 领域发生了什么”。第二个是延迟和选择性。很多平台更新靠编辑搬运热点出来两小时后才上架如果你想抢在第一时间看到重要论文或开源项目根本等不起。第三个是最致命的你没法沉淀自己的筛选标准。这个源质量高、那个源偶尔有独家你希望用某种规则把内容排序希望把某些主题置顶希望根据团队需求生成定制简报这类需求在现成服务里几乎做不了。1.2 自己搭的平台到底该长什么样动手之前先别想着做一个“今日头条”而是想清楚“每天能省下多少时间”。我给自己定义的形态是从固定信息源自动抓取内容清洗后去重再用大模型生成统一格式的摘要和标签最后展示在一个极简的网页上按时间倒序排列支持标签和关键词筛选。这套系统不追求实时半小时更新一次就够不追求海量控制在几十个经过筛选的源不追求复杂交互能搜、能筛、能点开原文就够了。换句话说它更像是“给第二天的自己准备一份干净的早餐”而不是“给你一个永远刷不完的瀑布流”。1.3 谁适合参考这套方案如果你刚接触编程但已经会一点 Python照着我下面的代码抄一遍也能跑起来。如果你是产品经理或研发负责人想给团队搭一套 AI 情报监控可以先按最小版本跑通再加上邮件或群机器人推送。如果你已经有一定开发经验我更想分享的是其中的架构思路和踩坑经验代码里的很多细节是我跑了半年之后才稳定下来的逻辑。2. 整体设计先画架构再聊代码2.1 核心数据流整个系统可以拆成一条流水线采集、清洗、存储、增强、分发。采集层负责从 RSS、API、网页定时拉数据。清洗层做正文抽取、URL 去重、标题相似度去重。存储层用 SQLite 保存结构化条目。增强层调用大模型生成摘要、标签、重要度评分顺便做文本向量化。分发层就是一个 Web 页面再加可选的通知推送。为什么把增强层放在存储层后面因为原始抓下来的内容质量参差不齐如果先做 AI 摘要很多脏文本会浪费 token还会重复计算。等入库、去重之后再增强一条内容只处理一次成本可控。2.2 技术选型为什么是 Python FastAPI SQLite个人项目的第一原则是怎么简单怎么来怎么容易维护怎么来。Python 生态做这件事太舒服了。RSS 解析有 feedparser正文抽取有 trafilaturaWeb 框架用 FastAPI异步性能足够一台小水管服务器扛日更几千条资讯数据库用 SQLite 单文件备份直接 cp 一下就行。有人会问为什么不用爬虫框架 Scrapy。答案是杀鸡用牛刀。我们的主要数据源是 RSS 和 API不是需要复杂抓取的站点Scrapy 的学习和维护成本对个人项目来说偏高。至于为什么不用 Celery单机任务、API 调用也不重APScheduler 就够了。组件我的选择备选方案理由RSS 解析feedparser自己写 XML 解析稳定处理编码和日期格式很省心正文抽取trafilaturanewspaper3k、readability在边缘页面上更稳定Web 框架FastAPIFlask、Django异步、自动文档、模板渲染也方便数据库SQLitePostgreSQL单文件、零运维数据量合适任务调度APSchedulercron、Celery单机场景够用不引入消息队列部署Docker Composesystemd、裸跑迁移方便依赖隔离2.3 关键功能模块模块化不是为了炫技是为了出问题时知道去哪排查。我分成这几块采集器负责从各个源拉数据并解析出结构化条目清洗器负责正文抽取、去掉无效字符、统一时间格式去重器负责 URL 和标题相似度判断增强器调用大模型生成 AI 摘要、标签、重要度分数查询与展示层负责搜索、筛选和页面渲染。模块之间通过数据库状态字段协作不需要复杂的消息队列。2.4 数据规模与升级路径如果你的资讯量到了每天几万条SQLite 可能不够看多个进程同时写会出现锁繁忙。这时可以平滑迁移到 PostgreSQL把数据库连接层换成 SQLAlchemy 即可。向量检索也建议从内存计算换成 pgvector 或专门的向量库。但起步阶段SQLite 完全能扛住别过度设计。3. 信息源管理聚合的第一步是把“源头”理清3.1 信息源分类与推荐渠道我把自己常用的信息源分成了四类媒体类、社区类、论文类和项目类。媒体类比如机器之心、量子位、InfoQ 中文它们有官方 RSS一般在 /feed 或 /rss.xml 能找到。社区类比如 Hacker News 提供官方 APIReddit 的 r/MachineLearning 用 .rss 后缀也能导出 feedGitHub Trending 可以配合搜索 API 按时间排序。论文类就是 arXiv 的 cs.AI、cs.CL、cs.LG 分区RSS 地址非常规整。项目类可以看 GitHub 的 releases 和 trending也可以通过第三方 hub 生成订阅源。找 RSS 有个笨但有效的方法在网址后面试探 /feed、/rss、/atom.xml很多系统都默认支持这几个路径。找不到了再用 RSSHub 这类工具生成生态很大。3.2 抓取与解析的坑很多 RSS 源只给摘要不给全文。这时候就需要正文抽取。我选 trafilatura 而不是 newspaper3k因为它在边界页面上的稳定性更好。它的核心思路是看 DOM 节点里的文本密度和标点比例找出最像正文的区块。局限也很明显如果是纯 JavaScript 渲染的页面直接抓 HTML 是抓不到的得上无头浏览器但这套复杂度我建议先别碰。抓取频率控制也容易忽略。有些站点对 RSS 抓取不设限但你每 10 分钟去拉一次IP 很快会被封。我的做法是每个源设置不同的抓取间隔重要源 30 分钟一次普通源 6 小时一次并且带着合法的 User-Agent 和联系邮箱尽量不给对方服务器增加压力。3.3 多级去重策略如果只靠 URL 去重同一个新闻在不同来源会被重复收录。所以我去重分三层第一层是 URL 精确去重入库前把 URL 去掉常见跟踪参数再算 hash第二层是标题规范化去重把标题转小写、去掉末尾感叹号问号、去掉【】里的栏目名然后算编辑距离或 simhash相似度超过阈值就认为是同一条第三层是内容 hash如果正文能抽出来就基于正文前 200 字做 hash。阈值我一般设在 0.78 左右。为什么不是 0.8 也不是 0.7太低容易误杀不同新闻太高漏掉同题报道。这个值要根据你自己的信息源测试多跑几次看结果再调。3.4 字段设计与存储我在 items 表里预留了这些字段source_name、url、url_hash、title、content_text、content_html、summary、tags、ai_summary、ai_tags、importance、published_at、fetched_at、status。url_hash 要做唯一索引这是去重的基础。content_html 不是必须的但保留下来方便以后对接前端或做富文本展示。ai_summary 和 ai_tags 一开始可以先设空等大模型处理完再回填。一条完整的资讯库里至少要能回答五个问题来自哪、链接到哪、标题是什么、正文讲什么、什么时候发的。剩下的字段都是加分项。4. AI 能力接入用大模型给资讯做摘要和标签4.1 为什么需要 AI 摘要而不是只取首段我最初觉得 RSS 自带的 summary 字段够用但很快发现两个问题第一很多源不提供 summary第二有的摘要就是文章第一段完全看不出这篇到底推出了什么新东西。AI 摘要的价值不是“用高级模型重写一遍”而是信息筛选。我给自己定的标准80 字以内说清楚三件事——谁或哪个团队、做了什么、为什么重要。实际写提示词的时候我会把这三件事直接写进要求让模型输出固定 JSON里面包含 summary、tags 和 importance。有了统一格式网页渲染和后续筛选都会方便很多。4.2 两种接入方式API 与本地模型接入大模型有两条路我在代码里做了同一个接口用环境变量切换。第一条是接大模型的 API比如 DeepSeek、Kimi、通义千问这类服务它们大多提供兼容接口。优点是零部署、效果好模型理解能力强缺点是长文本要计费密钥需要保管好内容也会发给第三方。第二条是本地跑 Ollama拉一个 qwen2.5 7B 级别的模型用 /api/generate 接口调用。优点是不花钱、数据不出门缺点是 7B 模型的摘要质量弱于商业 API速度也慢一些。我个人的建议是每天处理量在 300 条以内直接用 API 更省心如果完全不想把内容发给第三方就上本地模型。两条路切换成本很低只要把 base_url、model、api_key 配置化就行。4.3 提示词设计与成本控制提示词不需要写得像论文但一定要结构化。我用的模板大概是你是一名资深 AI 领域编辑。请为下面的资讯写一段不超过 80 字的中文摘要并给出 3 到 5 个标签和一个重要度分数0 到 10。只输出 JSON格式为 {summary: ..., tags: [...], importance: 8.5} 文章标题: {title} 文章内容: {content}这里有两个关键点一是输出格式必须用 JSON 约束方便程序解析二是内容截断取前 3000 字就够多数新闻核心信息在前面。温度参数我控制在 0.2 以下防止模型自由发挥。成本粗算一下按 4 个字符约 1 token 估算3000 字中文大约 1500 token加输出总共约 2000 token。300 条一天就是 60 万 token按当前主流 API 价格一天大概几块钱个人可接受。如果抓的文章量大还可以加规则只有标题命中关键词才做 AI 摘要能省不少钱。4.4 从标签推荐到多 AI 协作单条资讯处理好了下一步就是推荐。最简单的做法是标签匹配用户点了某个标签把相同 tags 但不同来源的文章优先展示。体验还行但不够聪明。进阶做法是用 embedding 生成向量相似度把文章编码成向量存起来查询时找相近内容。再往上走就是我最近在折腾的 AI Agent 工作流。可以把整个聚合平台看成三个 Agent采集 Agent 负责定时抓取和质量初筛摘要 Agent 负责理解并输出结构化摘要筛选 Agent 根据你的兴趣模型给每条内容评分。三个 Agent 之间通过数据库状态字段协作这就是很多人说的“多 AI 协作”。实际落地时不需要很重的框架用 Python SQLite 状态机就能跑起来。5. 实操实现一个最小可跑的聚合服务5.1 项目结构与依赖我采用的目录结构非常简单为了让你能直接抄ai-aggregator/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── feeds.py │ ├── ai_enhance.py │ ├── templates/ │ │ └── index.html ├── data/ ├── requirements.txt └── docker-compose.yml依赖文件 requirements.txtfastapi0.115.6 uvicorn[standard]0.34.0 feedparser6.0.11 httpx0.28.1 jinja23.1.4 apscheduler3.10.4 trafilatura1.12.0在开始写功能之前先把 virtualenv 建好然后安装依赖python -m venv venv source venv/bin/activate pip install -r requirements.txt5.2 采集入库feedparser 加 SQLite下面这段代码实现了三个最基础的动作初始化数据库、解析 RSS、批量入库。先建表再抓取最后用 INSERT OR IGNORE 保证 URL 唯一。import calendar import hashlib import sqlite3 import time from datetime import datetime, timezone import feedparser def init_db(db_path: str): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS items ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_name TEXT NOT NULL, url TEXT NOT NULL, url_hash TEXT UNIQUE NOT NULL, title TEXT NOT NULL, content_text TEXT, content_html TEXT, summary TEXT, tags TEXT, ai_summary TEXT, ai_tags TEXT, importance REAL DEFAULT 0, published_at TEXT, fetched_at TEXT, status TEXT DEFAULT new ) ) conn.commit() conn.close() def _published_to_iso(entry): parsed entry.get(published_parsed) or entry.get(updated_parsed) if parsed: return datetime.fromtimestamp(calendar.timegm(parsed), tztimezone.utc).isoformat() return datetime.now(timezone.utc).isoformat() def fetch_rss(feed_url: str) - list[dict]: data feedparser.parse(feed_url) entries [] for e in data.entries: entries.append({ source_name: data.feed.get(title, feed_url), title: e.get(title, ).strip(), link: e.get(link, ).strip(), summary: e.get(summary, ), content: e.get(content, [{}])[0].get(value, ) if e.get(content) else , published: _published_to_iso(e), }) return entries def url_hash(url: str) - str: return hashlib.sha256(url.strip().encode(utf-8)).hexdigest() def save_entries(db_path: str, entries: list[dict]): conn sqlite3.connect(db_path) for item in entries: h url_hash(item[link]) conn.execute( INSERT OR IGNORE INTO items (source_name, url, url_hash, title, content_text, content_html, summary, published_at, fetched_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) , ( item[source_name], item[link], h, item[title], item[content] or item[summary], item[content], item[summary], item[published], datetime.now(timezone.utc).isoformat(), )) conn.commit() conn.close()入库前把 URL 的跟踪参数去掉很重要否则一个带 utm_source、一个不带会当成两条。我在正式代码里会先做一次 clean_url把 ?utm、?from、?ref 这些参数删掉再算 hash。5.3 AI 增强调用本地 Ollama 或兼容 API增量处理未做摘要的条目是保证系统长期运行的关键。下面这段代码封装了统一的调用入口同时兼容 Ollama 和远程 API。import json import os import sqlite3 from concurrent.futures import ThreadPoolExecutor import httpx OLLAMA_URL os.getenv(OLLAMA_URL, http://localhost:11434/api/generate) LLM_API_URL os.getenv(LLM_API_URL, ) LLM_API_KEY os.getenv(LLM_API_KEY, ) LLM_MODEL os.getenv(LLM_MODEL, qwen2.5:7b) PROMPT 你是一名资深AI领域编辑。请为下面的资讯写一段不超过80字的中文摘要 并给出3到5个标签和一个重要度分数(0到10)。只输出JSON格式为: {summary: ..., tags: [...], importance: 8.5} 文章标题: {title} 文章内容: {content[:3000]} def call_llm(title: str, content: str) - dict: user_content PROMPT.format(titletitle, contentcontent) if LLM_API_URL: headers {Authorization: fBearer {LLM_API_KEY}} if LLM_API_KEY else {} resp httpx.post( f{LLM_API_URL}/chat/completions, json{model: LLM_MODEL, messages: [{role: user, content: user_content}], temperature: 0.2}, headersheaders, timeout60, ) resp.raise_for_status() return json.loads(resp.json()[choices][0][message][content]) else: resp httpx.post( OLLAMA_URL, json{model: LLM_MODEL, prompt: user_content, stream: False, format: json}, timeout120, ) resp.raise_for_status() return json.loads(resp.json()[response]) def enhance_pending_items(db_path: str, limit: int 50): conn sqlite3.connect(db_path) rows conn.execute( SELECT id, title, content_text FROM items WHERE ai_summary IS NULL AND statusnew ORDER BY published_at DESC LIMIT ?, (limit,) ).fetchall() def worker(row): rid, title, content_text row try: result call_llm(title, content_text or title) except Exception: return conn.execute( UPDATE items SET ai_summary?, ai_tags?, importance?, statusdone WHERE id?, (result.get(summary, ), ,.join(result.get(tags, [])), float(result.get(importance, 0)), rid), ) conn.commit() # 控制在 3 个并发避免触发 API 限流 with ThreadPoolExecutor(max_workers3) as pool: pool.map(worker, rows) conn.close()强调一下并发数。第一次跑的时候我直接开 10 个并发结果一半请求超时。后来把 max_workers 降到 3再加上简单的失败重试才稳定下来。5.4 Web 展示层FastAPI 加 Jinja2后端只需要一个首页路由读数据库后把数据传给模板。from pathlib import Path import sqlite3 from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from .feeds import init_db BASE_DIR Path(__file__).resolve().parent DB_PATH Path(__file__).resolve().parent.parent / data / aggregator.db app FastAPI() templates Jinja2Templates(directorystr(BASE_DIR / templates)) app.on_event(startup) def startup(): init_db(DB_PATH) def query_items(search: str , limit: int 50): conn sqlite3.connect(DB_PATH) sql SELECT * FROM items WHERE 11 params [] if search: sql AND (title LIKE ? OR ai_summary LIKE ? OR tags LIKE ?) params.extend([f%{search}%] * 3) sql ORDER BY published_at DESC LIMIT ? params.append(limit) rows conn.execute(sql, params).fetchall() cols [d[0] for d in conn.execute(SELECT * FROM items LIMIT 0).description] conn.close() return [dict(zip(cols, r)) for r in rows] app.get(/, response_classHTMLResponse) def home(request: Request, q: str ): items query_items(searchq.strip()) return templates.TemplateResponse(index.html, {request: request, items: items, q: q.strip()})模板 index.html 很长我只留下最核心的部分样式可以按自己审美重写!DOCTYPE html html langzh head meta charsetutf-8 meta nameviewport contentwidthdevice-width, initial-scale1 titleAI 资讯聚合/title /head body form methodget input typetext nameq value{{ q }} placeholder搜索标题 / 摘要 / 标签 button typesubmit搜索/button /form {% for item in items %} div classitem h3a href{{ item[url] }} target_blank{{ item[title] }}/a/h3 div classmeta{{ item[source_name] }} · {{ item[published_at] }} · {{ item[ai_tags] }}/div p{{ item[ai_summary] or item[summary] }}/p /div {% else %} p暂无内容请稍后回来。/p {% endfor %} /body /html启动方式是uvicorn app.main:app --host 0.0.0.0 --port 8000浏览器打开http://localhost:8000就能看到聚合列表。5.5 定时任务与自动更新页面跑起来之后还需要定时触发采集和 AI 增强。我用 APScheduler 在 FastAPI 启动时挂一个后台任务from apscheduler.schedulers.background import BackgroundScheduler scheduler BackgroundScheduler(timezoneAsia/Shanghai) SOURCES [ (机器之心, https://www.jiqizhixin.com/rss), (量子位, https://www.qbitai.com/feed), (InfoQ 中文, https://www.infoq.cn/feed), (Hacker News, https://news.ycombinator.com/rss), (Arxiv AI, http://export.arxiv.org/rss/cs.AI), ] def pipeline(): for name, url in SOURCES: try: entries fetch_rss(url) save_entries(DB_PATH, entries) except Exception: continue enhance_pending_items(DB_PATH) def start_scheduler(): scheduler.add_job(pipeline, interval, minutes30, idaggregate_pipeline) scheduler.start()在 startup 事件里调用 start_scheduler() 就行。如果你不想让进程常驻也可以写一个独立脚本让 cron 每半小时跑一次两种方式效果差不多。5.6 Docker 部署部署这块我直接上 Docker Compose原因是迁移方便服务器上不用装 Python 环境。DockerfileFROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app RUN mkdir -p /data CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000]docker-compose.ymlservices: aggregator: build: . ports: - 8000:8000 volumes: - ./data:/data environment: - DATA_PATH/data/aggregator.db - OLLAMA_URLhttp://host.docker.internal:11434/api/generate注意这里把数据库目录挂载到宿主机容器重建不会丢数据。OLLAMA_URL 用了 host.docker.internal是为了让容器能访问宿主机上的 Ollama。如果你用的是远程 API就把 OLLAMA_URL 换成你自己的服务地址。6. 常见问题排查与避坑实录6.1 抓取下来乱码怎么办RSS 大多是 XMLfeedparser 会按文档声明的编码解析一般不会出问题。但有些站点会返回压缩后的响应如果你直接用 requests 抓要设置 Accept-Encoding。还有一次我遇到某博客的 charset 声明和实际内容不一致页面上看正常抓回来全是乱码后来强制用 UTF-8 解码才解决。遇到乱码先别改代码先用 curl 看一眼响应头里的 content-type很多问题一眼就能定位。6.2 明明有新文章却不入库最可能的原因是 INSERT OR IGNORE 加 url_hash 把内容拦住了。同一个 URL 第一次入库后后面所有重复请求都会被忽略。这对去重是好事但如果源更新了同一篇的标题或正文你想抓最新版本就难了。解法是保留一个 updated_at 字段相同 url_hash 但标题变化超过一定比例时做 UPDATE。另一种情况是 URL 带了 utm 跟踪参数导致同一篇文章每次 URL 都不同。入库前要把常见跟踪参数删掉再算 hash。6.3 AI 摘要调用频繁报错常见原因有三个并发太高触发限流、超时时间太短、模型上下文不够。我遇到过 API 同时并发 8 个请求一半超时。后来改成线程池 max_workers3加失败重试两次平稳很多。还有一个很容易被忽略的坑模型返回的内容不一定每次都是合法 JSON它可能在 JSON 外面包一层代码块标记。解析的时候要先用正则把代码块标记去掉再做 json.loads否则会直接抛异常。6.4 搜索和过滤慢当数据量到几千条之后LIKE %关键词% 会开始变慢。SQLite 可以用 FTS5 全文索引解决。建一张虚拟表把 title、content_text、ai_summary 放进去用 MATCH 查询。如果不想动表结构也可以在写入时把整理好的标签字符串存在一个字段里过滤时用 tags LIKE 也能凑合但数据量大了还是建议上 FTS5。6.5 噪音还是太多AI 资讯圈的信息噪点特别多我的经验是三层过滤。第一层是源级别白名单只加真正持续输出高质量内容的源。第二层是内容规则排除纯转载、标题带“重磅”“震惊”但没有实质内容、视频转文字的低质量稿件。第三层才是模型打分importance 低于 5 的内容在列表里默认置灰只在标签页里出现。把规则下沉到采集层比在展示层手动屏蔽要省心得多。6.6 避坑速查表现象常见原因建议抓回来内容乱码编码声明错误或响应头不对检查 content-type强制设置 encoding新文章重复入库URL 跟踪参数导致 hash 不同入库前清理 utm/from/ref 参数同题新闻反复出现只做了 URL 去重加标题规范化 simhash 相似度去重AI 摘要超时并发太高或 timeout 太短控制到 3 并发设置 60 秒起返回内容解析失败模型输出不标准先去掉代码块标记再做 json.loads搜索越来越慢缺少全文索引建 FTS5 虚拟表定时期限内没抓到源被限流或请求太频繁降低抓取频率带 User-Agent这套系统我自己跑了半年多最大的感受是自己搭资讯平台的重点不是“我有多少信息源”而是“我如何定义什么值得看”。AI 摘要和自动化只是帮你省时间真正的筛选规则必须沉淀成代码和评分模型。你如果也想做别一上来就想得多完整先用 5 个 RSS 源加几十行代码把最小闭环跑通再加 AI 摘要再加推荐。跑顺了后面加能力都是水到渠成的事。
自建AI资讯聚合平台：RSS采集、去重与大模型摘要的完整架构实践
先说说我为什么非要自己动手。每天刷 AI 资讯的姿势我曾经是早上看一遍科技媒体中午刷 Hacker News晚上再扫一圈 arXiv 和 GitHub Trending。听起来很自律实际大部分时间花在重复阅读上——同一篇新闻换个标题出现五六次真正有深度的长文被淹没在快讯里。所以我花了两周时间自己搭了一个 AI 资讯聚合平台RSS 和 API 自动抓取、正文抽取、去重、AI 摘要、自动打标签最后统一在一个网页里浏览。这篇就把完整的架构设计、代码实现、部署和踩坑过程写清楚适合有点编程基础、想重新夺回信息流控制权的人也适合想给团队做 AI 情报系统的朋友抄作业。1. 为什么自己搭信息的“策展权”不能全交给别人1.1 现成资讯服务的三个痛点“聚合”这件事市面上已经有太多产品但用下来总有三个绕不开的痛点。第一个是信息过载。平台为了覆盖足够多的用户会把各种来源、各种质量的资讯涌到你面前算法推荐还会不断投喂同质化内容。你看到的不是“今天 AI 领域发生了什么”而是“平台希望你觉得 AI 领域发生了什么”。第二个是延迟和选择性。很多平台更新靠编辑搬运热点出来两小时后才上架如果你想抢在第一时间看到重要论文或开源项目根本等不起。第三个是最致命的你没法沉淀自己的筛选标准。这个源质量高、那个源偶尔有独家你希望用某种规则把内容排序希望把某些主题置顶希望根据团队需求生成定制简报这类需求在现成服务里几乎做不了。1.2 自己搭的平台到底该长什么样动手之前先别想着做一个“今日头条”而是想清楚“每天能省下多少时间”。我给自己定义的形态是从固定信息源自动抓取内容清洗后去重再用大模型生成统一格式的摘要和标签最后展示在一个极简的网页上按时间倒序排列支持标签和关键词筛选。这套系统不追求实时半小时更新一次就够不追求海量控制在几十个经过筛选的源不追求复杂交互能搜、能筛、能点开原文就够了。换句话说它更像是“给第二天的自己准备一份干净的早餐”而不是“给你一个永远刷不完的瀑布流”。1.3 谁适合参考这套方案如果你刚接触编程但已经会一点 Python照着我下面的代码抄一遍也能跑起来。如果你是产品经理或研发负责人想给团队搭一套 AI 情报监控可以先按最小版本跑通再加上邮件或群机器人推送。如果你已经有一定开发经验我更想分享的是其中的架构思路和踩坑经验代码里的很多细节是我跑了半年之后才稳定下来的逻辑。2. 整体设计先画架构再聊代码2.1 核心数据流整个系统可以拆成一条流水线采集、清洗、存储、增强、分发。采集层负责从 RSS、API、网页定时拉数据。清洗层做正文抽取、URL 去重、标题相似度去重。存储层用 SQLite 保存结构化条目。增强层调用大模型生成摘要、标签、重要度评分顺便做文本向量化。分发层就是一个 Web 页面再加可选的通知推送。为什么把增强层放在存储层后面因为原始抓下来的内容质量参差不齐如果先做 AI 摘要很多脏文本会浪费 token还会重复计算。等入库、去重之后再增强一条内容只处理一次成本可控。2.2 技术选型为什么是 Python FastAPI SQLite个人项目的第一原则是怎么简单怎么来怎么容易维护怎么来。Python 生态做这件事太舒服了。RSS 解析有 feedparser正文抽取有 trafilaturaWeb 框架用 FastAPI异步性能足够一台小水管服务器扛日更几千条资讯数据库用 SQLite 单文件备份直接 cp 一下就行。有人会问为什么不用爬虫框架 Scrapy。答案是杀鸡用牛刀。我们的主要数据源是 RSS 和 API不是需要复杂抓取的站点Scrapy 的学习和维护成本对个人项目来说偏高。至于为什么不用 Celery单机任务、API 调用也不重APScheduler 就够了。组件我的选择备选方案理由RSS 解析feedparser自己写 XML 解析稳定处理编码和日期格式很省心正文抽取trafilaturanewspaper3k、readability在边缘页面上更稳定Web 框架FastAPIFlask、Django异步、自动文档、模板渲染也方便数据库SQLitePostgreSQL单文件、零运维数据量合适任务调度APSchedulercron、Celery单机场景够用不引入消息队列部署Docker Composesystemd、裸跑迁移方便依赖隔离2.3 关键功能模块模块化不是为了炫技是为了出问题时知道去哪排查。我分成这几块采集器负责从各个源拉数据并解析出结构化条目清洗器负责正文抽取、去掉无效字符、统一时间格式去重器负责 URL 和标题相似度判断增强器调用大模型生成 AI 摘要、标签、重要度分数查询与展示层负责搜索、筛选和页面渲染。模块之间通过数据库状态字段协作不需要复杂的消息队列。2.4 数据规模与升级路径如果你的资讯量到了每天几万条SQLite 可能不够看多个进程同时写会出现锁繁忙。这时可以平滑迁移到 PostgreSQL把数据库连接层换成 SQLAlchemy 即可。向量检索也建议从内存计算换成 pgvector 或专门的向量库。但起步阶段SQLite 完全能扛住别过度设计。3. 信息源管理聚合的第一步是把“源头”理清3.1 信息源分类与推荐渠道我把自己常用的信息源分成了四类媒体类、社区类、论文类和项目类。媒体类比如机器之心、量子位、InfoQ 中文它们有官方 RSS一般在 /feed 或 /rss.xml 能找到。社区类比如 Hacker News 提供官方 APIReddit 的 r/MachineLearning 用 .rss 后缀也能导出 feedGitHub Trending 可以配合搜索 API 按时间排序。论文类就是 arXiv 的 cs.AI、cs.CL、cs.LG 分区RSS 地址非常规整。项目类可以看 GitHub 的 releases 和 trending也可以通过第三方 hub 生成订阅源。找 RSS 有个笨但有效的方法在网址后面试探 /feed、/rss、/atom.xml很多系统都默认支持这几个路径。找不到了再用 RSSHub 这类工具生成生态很大。3.2 抓取与解析的坑很多 RSS 源只给摘要不给全文。这时候就需要正文抽取。我选 trafilatura 而不是 newspaper3k因为它在边界页面上的稳定性更好。它的核心思路是看 DOM 节点里的文本密度和标点比例找出最像正文的区块。局限也很明显如果是纯 JavaScript 渲染的页面直接抓 HTML 是抓不到的得上无头浏览器但这套复杂度我建议先别碰。抓取频率控制也容易忽略。有些站点对 RSS 抓取不设限但你每 10 分钟去拉一次IP 很快会被封。我的做法是每个源设置不同的抓取间隔重要源 30 分钟一次普通源 6 小时一次并且带着合法的 User-Agent 和联系邮箱尽量不给对方服务器增加压力。3.3 多级去重策略如果只靠 URL 去重同一个新闻在不同来源会被重复收录。所以我去重分三层第一层是 URL 精确去重入库前把 URL 去掉常见跟踪参数再算 hash第二层是标题规范化去重把标题转小写、去掉末尾感叹号问号、去掉【】里的栏目名然后算编辑距离或 simhash相似度超过阈值就认为是同一条第三层是内容 hash如果正文能抽出来就基于正文前 200 字做 hash。阈值我一般设在 0.78 左右。为什么不是 0.8 也不是 0.7太低容易误杀不同新闻太高漏掉同题报道。这个值要根据你自己的信息源测试多跑几次看结果再调。3.4 字段设计与存储我在 items 表里预留了这些字段source_name、url、url_hash、title、content_text、content_html、summary、tags、ai_summary、ai_tags、importance、published_at、fetched_at、status。url_hash 要做唯一索引这是去重的基础。content_html 不是必须的但保留下来方便以后对接前端或做富文本展示。ai_summary 和 ai_tags 一开始可以先设空等大模型处理完再回填。一条完整的资讯库里至少要能回答五个问题来自哪、链接到哪、标题是什么、正文讲什么、什么时候发的。剩下的字段都是加分项。4. AI 能力接入用大模型给资讯做摘要和标签4.1 为什么需要 AI 摘要而不是只取首段我最初觉得 RSS 自带的 summary 字段够用但很快发现两个问题第一很多源不提供 summary第二有的摘要就是文章第一段完全看不出这篇到底推出了什么新东西。AI 摘要的价值不是“用高级模型重写一遍”而是信息筛选。我给自己定的标准80 字以内说清楚三件事——谁或哪个团队、做了什么、为什么重要。实际写提示词的时候我会把这三件事直接写进要求让模型输出固定 JSON里面包含 summary、tags 和 importance。有了统一格式网页渲染和后续筛选都会方便很多。4.2 两种接入方式API 与本地模型接入大模型有两条路我在代码里做了同一个接口用环境变量切换。第一条是接大模型的 API比如 DeepSeek、Kimi、通义千问这类服务它们大多提供兼容接口。优点是零部署、效果好模型理解能力强缺点是长文本要计费密钥需要保管好内容也会发给第三方。第二条是本地跑 Ollama拉一个 qwen2.5 7B 级别的模型用 /api/generate 接口调用。优点是不花钱、数据不出门缺点是 7B 模型的摘要质量弱于商业 API速度也慢一些。我个人的建议是每天处理量在 300 条以内直接用 API 更省心如果完全不想把内容发给第三方就上本地模型。两条路切换成本很低只要把 base_url、model、api_key 配置化就行。4.3 提示词设计与成本控制提示词不需要写得像论文但一定要结构化。我用的模板大概是你是一名资深 AI 领域编辑。请为下面的资讯写一段不超过 80 字的中文摘要并给出 3 到 5 个标签和一个重要度分数0 到 10。只输出 JSON格式为 {summary: ..., tags: [...], importance: 8.5} 文章标题: {title} 文章内容: {content}这里有两个关键点一是输出格式必须用 JSON 约束方便程序解析二是内容截断取前 3000 字就够多数新闻核心信息在前面。温度参数我控制在 0.2 以下防止模型自由发挥。成本粗算一下按 4 个字符约 1 token 估算3000 字中文大约 1500 token加输出总共约 2000 token。300 条一天就是 60 万 token按当前主流 API 价格一天大概几块钱个人可接受。如果抓的文章量大还可以加规则只有标题命中关键词才做 AI 摘要能省不少钱。4.4 从标签推荐到多 AI 协作单条资讯处理好了下一步就是推荐。最简单的做法是标签匹配用户点了某个标签把相同 tags 但不同来源的文章优先展示。体验还行但不够聪明。进阶做法是用 embedding 生成向量相似度把文章编码成向量存起来查询时找相近内容。再往上走就是我最近在折腾的 AI Agent 工作流。可以把整个聚合平台看成三个 Agent采集 Agent 负责定时抓取和质量初筛摘要 Agent 负责理解并输出结构化摘要筛选 Agent 根据你的兴趣模型给每条内容评分。三个 Agent 之间通过数据库状态字段协作这就是很多人说的“多 AI 协作”。实际落地时不需要很重的框架用 Python SQLite 状态机就能跑起来。5. 实操实现一个最小可跑的聚合服务5.1 项目结构与依赖我采用的目录结构非常简单为了让你能直接抄ai-aggregator/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── feeds.py │ ├── ai_enhance.py │ ├── templates/ │ │ └── index.html ├── data/ ├── requirements.txt └── docker-compose.yml依赖文件 requirements.txtfastapi0.115.6 uvicorn[standard]0.34.0 feedparser6.0.11 httpx0.28.1 jinja23.1.4 apscheduler3.10.4 trafilatura1.12.0在开始写功能之前先把 virtualenv 建好然后安装依赖python -m venv venv source venv/bin/activate pip install -r requirements.txt5.2 采集入库feedparser 加 SQLite下面这段代码实现了三个最基础的动作初始化数据库、解析 RSS、批量入库。先建表再抓取最后用 INSERT OR IGNORE 保证 URL 唯一。import calendar import hashlib import sqlite3 import time from datetime import datetime, timezone import feedparser def init_db(db_path: str): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS items ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_name TEXT NOT NULL, url TEXT NOT NULL, url_hash TEXT UNIQUE NOT NULL, title TEXT NOT NULL, content_text TEXT, content_html TEXT, summary TEXT, tags TEXT, ai_summary TEXT, ai_tags TEXT, importance REAL DEFAULT 0, published_at TEXT, fetched_at TEXT, status TEXT DEFAULT new ) ) conn.commit() conn.close() def _published_to_iso(entry): parsed entry.get(published_parsed) or entry.get(updated_parsed) if parsed: return datetime.fromtimestamp(calendar.timegm(parsed), tztimezone.utc).isoformat() return datetime.now(timezone.utc).isoformat() def fetch_rss(feed_url: str) - list[dict]: data feedparser.parse(feed_url) entries [] for e in data.entries: entries.append({ source_name: data.feed.get(title, feed_url), title: e.get(title, ).strip(), link: e.get(link, ).strip(), summary: e.get(summary, ), content: e.get(content, [{}])[0].get(value, ) if e.get(content) else , published: _published_to_iso(e), }) return entries def url_hash(url: str) - str: return hashlib.sha256(url.strip().encode(utf-8)).hexdigest() def save_entries(db_path: str, entries: list[dict]): conn sqlite3.connect(db_path) for item in entries: h url_hash(item[link]) conn.execute( INSERT OR IGNORE INTO items (source_name, url, url_hash, title, content_text, content_html, summary, published_at, fetched_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) , ( item[source_name], item[link], h, item[title], item[content] or item[summary], item[content], item[summary], item[published], datetime.now(timezone.utc).isoformat(), )) conn.commit() conn.close()入库前把 URL 的跟踪参数去掉很重要否则一个带 utm_source、一个不带会当成两条。我在正式代码里会先做一次 clean_url把 ?utm、?from、?ref 这些参数删掉再算 hash。5.3 AI 增强调用本地 Ollama 或兼容 API增量处理未做摘要的条目是保证系统长期运行的关键。下面这段代码封装了统一的调用入口同时兼容 Ollama 和远程 API。import json import os import sqlite3 from concurrent.futures import ThreadPoolExecutor import httpx OLLAMA_URL os.getenv(OLLAMA_URL, http://localhost:11434/api/generate) LLM_API_URL os.getenv(LLM_API_URL, ) LLM_API_KEY os.getenv(LLM_API_KEY, ) LLM_MODEL os.getenv(LLM_MODEL, qwen2.5:7b) PROMPT 你是一名资深AI领域编辑。请为下面的资讯写一段不超过80字的中文摘要 并给出3到5个标签和一个重要度分数(0到10)。只输出JSON格式为: {summary: ..., tags: [...], importance: 8.5} 文章标题: {title} 文章内容: {content[:3000]} def call_llm(title: str, content: str) - dict: user_content PROMPT.format(titletitle, contentcontent) if LLM_API_URL: headers {Authorization: fBearer {LLM_API_KEY}} if LLM_API_KEY else {} resp httpx.post( f{LLM_API_URL}/chat/completions, json{model: LLM_MODEL, messages: [{role: user, content: user_content}], temperature: 0.2}, headersheaders, timeout60, ) resp.raise_for_status() return json.loads(resp.json()[choices][0][message][content]) else: resp httpx.post( OLLAMA_URL, json{model: LLM_MODEL, prompt: user_content, stream: False, format: json}, timeout120, ) resp.raise_for_status() return json.loads(resp.json()[response]) def enhance_pending_items(db_path: str, limit: int 50): conn sqlite3.connect(db_path) rows conn.execute( SELECT id, title, content_text FROM items WHERE ai_summary IS NULL AND statusnew ORDER BY published_at DESC LIMIT ?, (limit,) ).fetchall() def worker(row): rid, title, content_text row try: result call_llm(title, content_text or title) except Exception: return conn.execute( UPDATE items SET ai_summary?, ai_tags?, importance?, statusdone WHERE id?, (result.get(summary, ), ,.join(result.get(tags, [])), float(result.get(importance, 0)), rid), ) conn.commit() # 控制在 3 个并发避免触发 API 限流 with ThreadPoolExecutor(max_workers3) as pool: pool.map(worker, rows) conn.close()强调一下并发数。第一次跑的时候我直接开 10 个并发结果一半请求超时。后来把 max_workers 降到 3再加上简单的失败重试才稳定下来。5.4 Web 展示层FastAPI 加 Jinja2后端只需要一个首页路由读数据库后把数据传给模板。from pathlib import Path import sqlite3 from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from .feeds import init_db BASE_DIR Path(__file__).resolve().parent DB_PATH Path(__file__).resolve().parent.parent / data / aggregator.db app FastAPI() templates Jinja2Templates(directorystr(BASE_DIR / templates)) app.on_event(startup) def startup(): init_db(DB_PATH) def query_items(search: str , limit: int 50): conn sqlite3.connect(DB_PATH) sql SELECT * FROM items WHERE 11 params [] if search: sql AND (title LIKE ? OR ai_summary LIKE ? OR tags LIKE ?) params.extend([f%{search}%] * 3) sql ORDER BY published_at DESC LIMIT ? params.append(limit) rows conn.execute(sql, params).fetchall() cols [d[0] for d in conn.execute(SELECT * FROM items LIMIT 0).description] conn.close() return [dict(zip(cols, r)) for r in rows] app.get(/, response_classHTMLResponse) def home(request: Request, q: str ): items query_items(searchq.strip()) return templates.TemplateResponse(index.html, {request: request, items: items, q: q.strip()})模板 index.html 很长我只留下最核心的部分样式可以按自己审美重写!DOCTYPE html html langzh head meta charsetutf-8 meta nameviewport contentwidthdevice-width, initial-scale1 titleAI 资讯聚合/title /head body form methodget input typetext nameq value{{ q }} placeholder搜索标题 / 摘要 / 标签 button typesubmit搜索/button /form {% for item in items %} div classitem h3a href{{ item[url] }} target_blank{{ item[title] }}/a/h3 div classmeta{{ item[source_name] }} · {{ item[published_at] }} · {{ item[ai_tags] }}/div p{{ item[ai_summary] or item[summary] }}/p /div {% else %} p暂无内容请稍后回来。/p {% endfor %} /body /html启动方式是uvicorn app.main:app --host 0.0.0.0 --port 8000浏览器打开http://localhost:8000就能看到聚合列表。5.5 定时任务与自动更新页面跑起来之后还需要定时触发采集和 AI 增强。我用 APScheduler 在 FastAPI 启动时挂一个后台任务from apscheduler.schedulers.background import BackgroundScheduler scheduler BackgroundScheduler(timezoneAsia/Shanghai) SOURCES [ (机器之心, https://www.jiqizhixin.com/rss), (量子位, https://www.qbitai.com/feed), (InfoQ 中文, https://www.infoq.cn/feed), (Hacker News, https://news.ycombinator.com/rss), (Arxiv AI, http://export.arxiv.org/rss/cs.AI), ] def pipeline(): for name, url in SOURCES: try: entries fetch_rss(url) save_entries(DB_PATH, entries) except Exception: continue enhance_pending_items(DB_PATH) def start_scheduler(): scheduler.add_job(pipeline, interval, minutes30, idaggregate_pipeline) scheduler.start()在 startup 事件里调用 start_scheduler() 就行。如果你不想让进程常驻也可以写一个独立脚本让 cron 每半小时跑一次两种方式效果差不多。5.6 Docker 部署部署这块我直接上 Docker Compose原因是迁移方便服务器上不用装 Python 环境。DockerfileFROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app RUN mkdir -p /data CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000]docker-compose.ymlservices: aggregator: build: . ports: - 8000:8000 volumes: - ./data:/data environment: - DATA_PATH/data/aggregator.db - OLLAMA_URLhttp://host.docker.internal:11434/api/generate注意这里把数据库目录挂载到宿主机容器重建不会丢数据。OLLAMA_URL 用了 host.docker.internal是为了让容器能访问宿主机上的 Ollama。如果你用的是远程 API就把 OLLAMA_URL 换成你自己的服务地址。6. 常见问题排查与避坑实录6.1 抓取下来乱码怎么办RSS 大多是 XMLfeedparser 会按文档声明的编码解析一般不会出问题。但有些站点会返回压缩后的响应如果你直接用 requests 抓要设置 Accept-Encoding。还有一次我遇到某博客的 charset 声明和实际内容不一致页面上看正常抓回来全是乱码后来强制用 UTF-8 解码才解决。遇到乱码先别改代码先用 curl 看一眼响应头里的 content-type很多问题一眼就能定位。6.2 明明有新文章却不入库最可能的原因是 INSERT OR IGNORE 加 url_hash 把内容拦住了。同一个 URL 第一次入库后后面所有重复请求都会被忽略。这对去重是好事但如果源更新了同一篇的标题或正文你想抓最新版本就难了。解法是保留一个 updated_at 字段相同 url_hash 但标题变化超过一定比例时做 UPDATE。另一种情况是 URL 带了 utm 跟踪参数导致同一篇文章每次 URL 都不同。入库前要把常见跟踪参数删掉再算 hash。6.3 AI 摘要调用频繁报错常见原因有三个并发太高触发限流、超时时间太短、模型上下文不够。我遇到过 API 同时并发 8 个请求一半超时。后来改成线程池 max_workers3加失败重试两次平稳很多。还有一个很容易被忽略的坑模型返回的内容不一定每次都是合法 JSON它可能在 JSON 外面包一层代码块标记。解析的时候要先用正则把代码块标记去掉再做 json.loads否则会直接抛异常。6.4 搜索和过滤慢当数据量到几千条之后LIKE %关键词% 会开始变慢。SQLite 可以用 FTS5 全文索引解决。建一张虚拟表把 title、content_text、ai_summary 放进去用 MATCH 查询。如果不想动表结构也可以在写入时把整理好的标签字符串存在一个字段里过滤时用 tags LIKE 也能凑合但数据量大了还是建议上 FTS5。6.5 噪音还是太多AI 资讯圈的信息噪点特别多我的经验是三层过滤。第一层是源级别白名单只加真正持续输出高质量内容的源。第二层是内容规则排除纯转载、标题带“重磅”“震惊”但没有实质内容、视频转文字的低质量稿件。第三层才是模型打分importance 低于 5 的内容在列表里默认置灰只在标签页里出现。把规则下沉到采集层比在展示层手动屏蔽要省心得多。6.6 避坑速查表现象常见原因建议抓回来内容乱码编码声明错误或响应头不对检查 content-type强制设置 encoding新文章重复入库URL 跟踪参数导致 hash 不同入库前清理 utm/from/ref 参数同题新闻反复出现只做了 URL 去重加标题规范化 simhash 相似度去重AI 摘要超时并发太高或 timeout 太短控制到 3 并发设置 60 秒起返回内容解析失败模型输出不标准先去掉代码块标记再做 json.loads搜索越来越慢缺少全文索引建 FTS5 虚拟表定时期限内没抓到源被限流或请求太频繁降低抓取频率带 User-Agent这套系统我自己跑了半年多最大的感受是自己搭资讯平台的重点不是“我有多少信息源”而是“我如何定义什么值得看”。AI 摘要和自动化只是帮你省时间真正的筛选规则必须沉淀成代码和评分模型。你如果也想做别一上来就想得多完整先用 5 个 RSS 源加几十行代码把最小闭环跑通再加 AI 摘要再加推荐。跑顺了后面加能力都是水到渠成的事。