每天打开手机,热点一个接一个:房贷新政、人形机器人、航天突破、核聚变能、数字产业……但大多数人的动作只是停留在“扫一眼标题”,然后继续刷下一条。真正有用的不是这一条热搜,而是你能不能从一堆零散信息里快速抽出“别人没看到的变化”,也就是常说的信息差和认知差。
这次我们不聊怎么读新闻,聊一个更实际的问题:怎么用一套本地部署的小工具,每天自动抓取热点、按主题分类、调用大模型生成一份属于自己的“信息差简报”。整个方案对硬件要求很低,普通 CPU 电脑就能跑,支持批量处理,也提供 HTTP 接口,方便接到企业微信、钉钉或自己的服务里。下面直接进入实现。
1. 核心能力速览
先把这套“每日信息差简报生成器”的关键规格列出来,方便确认适不适合你。
| 能力项 | 说明 |
|---|---|
| 项目类型 | 本地部署的资讯聚合与 AI 摘要工具 |
| 主要功能 | 热点抓取、标签分类、AI 摘要、Markdown 报告输出 |
| 硬件门槛 | 无 GPU 也能运行,普通 4 核 CPU + 8G 内存即可 |
| 显存占用 | 如果调用云端大模型 API,本地显存占用为 0;如果使用本地大模型,显存需按模型单独评估 |
| 推荐系统 | Windows 10/11、Ubuntu 20.04+、macOS 12+ |
| 启动方式 | 命令行启动,或一键脚本启动 |
| 是否支持 API | 支持,使用 FastAPI 提供 HTTP 接口 |
| 是否支持批量任务 | 支持,可对多来源、多主题批量抓取和汇总 |
| 输出格式 | 控制台输出 + Markdown 文件 |
| 适合场景 | 个人每日资讯整理、行业研究员热点监控、内容创作者选题辅助 |
需要先说明一点:这篇文章不绑定任何商业化项目,而是给出一套可以自己复现的最小实现思路。代码里的项目名、目录名、端口都可以按需替换。
2. 适用场景与使用边界
这类工具适合三类人:
第一类是个人知识管理重度用户。每天需要关注多个行业,靠手动刷新闻效率太低,用脚本自动抓取 RSS 或新闻源,再让大模型生成每条热点的一句话逻辑推导,能明显节省早上的阅读时间。
第二类是行业研究员或投资分析人员。他们需要的不是“发生了什么”,而是“这事件可能影响什么”。通过在提示词里强调“分析影响链”,工具输出的信息差简报会更有价值。
第三类是内容创作者。无论是做视频还是写公众号,找选题是刚需。热点抓下来之后,大模型可以从不同角度生成选题角度和观点碰撞点,帮助快速判断哪个切入点有差异化。
使用边界也必须说清楚:
- 抓取新闻内容时要注意版权,只做摘要和链接引用,不要全文转载。
- 涉及政策、金融、医疗等敏感领域时,AI 摘要只能作为快速参考,不能替代专业人士判断。
- 工具生成的内容可能带有模型幻觉,发布前需要人工复核。
- 不要用这个工具批量生成虚假信息或恶意引导舆论。
合规底线是一票否决项。技术本身只是帮人提高信息处理效率,但使用方式和传播内容必须由使用者负责。
3. 环境准备与前置条件
这套系统不依赖重型框架,主要技术栈是 Python 3.10+、FastAPI、Feedparser 和 OpenAI SDK。如果你要用国内大模型接口,可以直接换成 DashScope 或通义 SDK,逻辑不变。
准备清单:
| 项目 | 要求 |
|---|---|
| 操作系统 | Windows 10/11、Ubuntu 20.04+、macOS 12+ |
| Python 版本 | 3.10 或更高 |
| 磁盘空间 | 至少 2G,主要用于日志和生成的简报文件 |
| GPU | 可选,非必须 |
| 网络 | 需要能访问 RSS 源和大模型 API |
| 大模型 API Key | 必选,推荐通义千问、DeepSeek 或 OpenAI 兼容接口 |
如果你不想用云 API,也可以接本地 Ollama 服务,只需要把base_url改成http://127.0.0.1:11434/v1。但本地大模型对显存和内存的要求会明显提高,后面会单独讲。
4. 安装部署与启动方式
先创建一个项目目录,命名为infogap-daily,然后进入目录创建虚拟环境。
mkdir infogap-daily cd infogap-daily python -m venv venvWindows 下激活虚拟环境:
venv\Scripts\activateLinux 和 macOS 下激活:
source venv/bin/activate接着安装依赖。这里只装最小依赖,不装多余的包。
pip install fastapi uvicorn feedparser requests openai pydantic如果你用的是国内大模型 SDK,可以额外安装:
pip install dashscope安装完成后,在项目目录下新建config.yaml,用来统一管理 RSS 源、模型参数和输出目录。这里不引入复杂的配置框架,直接用PyYAML读取,所以还要安装:
pip install pyyaml下面是一个最小配置示例:
# config.yaml rss_sources: - name: "科技媒体" url: "https://example.com/rss/tech.xml" - name: "财经媒体" url: "https://example.com/rss/finance.xml" topics: - "人形机器人" - "航天突破" - "核聚变能" - "数字产业" llm: provider: "openai" base_url: "https://api.openai.com/v1" api_key: "${OPENAI_API_KEY}" model: "gpt-4o-mini" temperature: 0.3 max_tokens: 2000 output: output_dir: "./outputs" file_prefix: "daily_brief"注意,配置里的 RSS 地址是占位符。实际使用时要替换成你能访问的合规新闻源或博客源。api_key也可以通过环境变量注入,避免写在配置里。如果你用环境变量,需要把配置里的api_key改成${OPENAI_API_KEY},然后在 Python 里读取。
5. 核心逻辑:抓取与摘要
5.1 写一个 RSS 抓取模块
新建fetcher.py,负责从多个 RSS 源抓取最近 24 小时的新闻条目。为了保持代码简单,这里直接用 Feedparser 处理 XML。
# fetcher.py import feedparser from datetime import datetime, timedelta from urllib.parse import urlparse def fetch_entries(rss_url, hours=24): parsed = feedparser.parse(rss_url) entries = [] cutoff = datetime.now() - timedelta(hours=hours) for entry in parsed.entries: published = entry.get("published_parsed") or entry.get("updated_parsed") if published is None: continue pub_time = datetime(*published[:6]) if pub_time < cutoff: continue entries.append({ "title": entry.get("title", "").strip(), "link": entry.get("link", "").strip(), "summary": entry.get("summary", "").strip()[:200], "source": urlparse(rss_url).netloc, "published": pub_time.strftime("%Y-%m-%d %H:%M:%S") }) return entries def fetch_all_sources(rss_sources, hours=24): all_entries = [] for source in rss_sources: try: entries = fetch_entries(source["url"], hours=hours) all_entries.extend(entries) print(f"[OK] {source['name']} -> {len(entries)} 条") except Exception as exc: print(f"[ERROR] {source['name']} 抓取失败: {exc}") return all_entries这段代码里,published_parsed是 Feedparser 解析后的标准时间结构。每个源抓完立即打印状态,方便排错。批量抓取多个源时,如果单个源超时,不会影响其他源。
5.2 写一个 AI 摘要模块
新建summarizer.py,调用大模型 API 生成“信息差摘要”。这里的关键是提示词设计。不是直接问“这条新闻讲了什么”,而是要求模型输出三个层次:事实、影响、认知差。
# summarizer.py from openai import OpenAI class InfoGapSummarizer: def __init__(self, config): base_url = config["llm"].get("base_url") api_key = config["llm"]["api_key"] self.model = config["llm"]["model"] self.temperature = config["llm"].get("temperature", 0.3) self.max_tokens = config["llm"].get("max_tokens", 2000) self.client = OpenAI(base_url=base_url, api_key=api_key) def summarize(self, entry): prompt = f""" 你是资深行业分析师。请根据下面这条热点新闻生成简报。 输出格式: 1. 核心事实:用 2 句话概括事件。 2. 影响判断:拆解这条信息对行业或普通人可能产生的影响。 3. 认知差:指出大多数人没注意到的细节或后续值得跟踪的方向。 新闻标题:{entry['title']} 新闻摘要:{entry['summary']} 发布时间:{entry['published']} 来源:{entry['source']} """ response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": "你是一名严谨的中文资讯分析师,输出内容必须基于给定新闻,不能编造事实。"}, {"role": "user", "content": prompt} ], temperature=self.temperature, max_tokens=self.max_tokens ) return response.choices[0].message.content如果你的 API 服务是 OpenAI 兼容格式,比如 DeepSeek、通义兼容模式或 Ollama,都可以用这一段代码。如果使用 DashScope 原生 SDK,则要改成相应的方法。
5.3 生成信息差简报
新建generate_daily_brief.py,把抓取、摘要、输出串成一个完整流程。
# generate_daily_brief.py import os import yaml from datetime import datetime from fetcher import fetch_all_sources from summarizer import InfoGapSummarizer def load_config(path="config.yaml"): with open(path, "r", encoding="utf-8") as f: return yaml.safe_load(f) def main(): config = load_config() summarizer = InfoGapSummarizer(config) print("开始抓取热点信息...") entries = fetch_all_sources(config["rss_sources"], hours=24) print(f"共抓取 {len(entries)} 条内容") lines = [] for idx, entry in enumerate(entries, 1): print(f"正在生成第 {idx}/{len(entries)} 条摘要...") try: brief = summarizer.summarize(entry) except Exception as exc: print(f"[ERROR] 第 {idx} 条摘要生成失败: {exc}") continue lines.append(f"### {entry['title']}\n") lines.append(f"- 来源:{entry['source']} | 时间:{entry['published']}\n") lines.append(f"- 原文链接:{entry['link']}\n") lines.append(brief) lines.append("\n---\n") os.makedirs(config["output"]["output_dir"], exist_ok=True) date_str = datetime.now().strftime("%Y%m%d") file_path = os.path.join( config["output"]["output_dir"], f"{config['output']['file_prefix']}_{date_str}.md" ) with open(file_path, "w", encoding="utf-8") as f: f.write("\n".join(lines)) print(f"简报已生成:{file_path}") if __name__ == "__main__": main()运行命令:
python generate_daily_brief.py成功后,会在outputs目录下看到一个 Markdown 文件,内容就是按时间排列的“信息差简报”。
6. 功能测试与效果验证
部署完成后,不要急着批量跑,先用一个小样本验证链路。这里给出一套可复用的验证流程。
6.1 验证 RSS 抓取是否正常
新建test_fetcher.py,只抓取一个源,打印前 3 条结果。
# test_fetcher.py from fetcher import fetch_entries entries = fetch_entries("https://example.com/rss/tech.xml", hours=48) for e in entries[:3]: print(e["title"]) print(e["link"]) print(e["published"]) print("---")运行:
python test_fetcher.py如果打印出标题和链接,说明 RSS 解析正常。如果输出为空,先确认 RSS 地址是否可直接访问,是否近期有更新。
6.2 验证 AI 摘要是否正常
修改generate_daily_brief.py里的抓取逻辑,或者直接写一个调用测试:
from summarizer import InfoGapSummarizer import yaml config = yaml.safe_load(open("config.yaml", encoding="utf-8")) summarizer = InfoGapSummarizer(config) test_entry = { "title": "人形机器人公司发布新一代灵巧手", "summary": "该灵巧手拥有多个自由度,可完成精细抓取动作。", "source": "example.com", "published": "2026-08-26 08:00:00", "link": "https://example.com/news/1" } print(summarizer.summarize(test_entry))判断是否成功的标准:
- 输出是否包含“核心事实”“影响判断”“认知差”三个板块。
- 内容是否与输入高度相关,而不是模型自己发挥。
- 摘要是否有明显事实错误。
- 响应时间是否在可接受范围内。
6.3 验证批量任务稳定性
如果 RSS 源很多,建议先跑 3 个源,看日志是否有大量失败。批量任务最怕两个问题:单源超时导致整体卡住,API 调用频率超限。处理方法是给抓取和摘要分别加超时和重试策略。
当前代码里已经用try/except隔离了单源错误,API 调用失败也不会中断整个流程。如果多个源加起来超过 50 条,建议在fetch_all_sources后加一个max_entries参数,避免全部发送给模型。
entries = fetch_all_sources(config["rss_sources"], hours=24) entries = entries[:50]这样能控制单次任务成本和耗时,适合批量任务跑定时调度。
7. 接口 API 调用与批量任务扩展
只跑命令行不够灵活,日常使用大概率需要把简报推到 webhook、企业微信群机器人或自己的前端页面。这里提供一个 FastAPI 接口示例。
新建api.py:
# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import yaml from fetcher import fetch_all_sources from summarizer import InfoGapSummarizer app = FastAPI(title="InfoGap Daily API") config = yaml.safe_load(open("config.yaml", encoding="utf-8")) class BriefRequest(BaseModel): rss_sources: list = None hours: int = 24 max_entries: int = 20 class BriefResponse(BaseModel): total: int brief: str @app.post("/api/brief", response_model=BriefResponse) def generate_brief(request: BriefRequest): rss_sources = request.rss_sources or config["rss_sources"] try: entries = fetch_all_sources(rss_sources, hours=request.hours) entries = entries[: request.max_entries] summarizer = InfoGapSummarizer(config) result_lines = [] for entry in entries: brief = summarizer.summarize(entry) result_lines.append(f"### {entry['title']}\n\n来源:{entry['source']}\n\n{brief}\n\n---\n") return BriefResponse(total=len(entries), brief="\n".join(result_lines)) except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8090)启动 API 服务:
python api.py接着在另一个终端调用接口:
curl -X POST "http://127.0.0.1:8090/api/brief" \ -H "Content-Type: application/json" \ -d '{"hours": 24, "max_entries": 5}'也可以使用 Python requests 调用:
import requests payload = { "hours": 24, "max_entries": 10, "rss_sources": None } resp = requests.post("http://127.0.0.1:8090/api/brief", json=payload, timeout=180) print(resp.json()["total"]) print(resp.json()["brief"][:500])这个接口可以直接接到企业微信/钉钉机器人的定时任务里。配合系统 cron 或 Windows 计划任务,每天上午 8 点自动请求一次,然后把推送结果发到群里。
批量任务还可以进一步做成分片处理:把同一主题下的多条新闻合并到一个 Prompt 中,让大模型输出统一的“主题信息差”,而不是逐条摘要。这样能显著降低 token 消耗,输出也更聚焦。
8. 资源占用与性能观察
这套方案在普通 CPU 环境下性能表现不错。如果使用云端 API,本地进程主要是网络请求和文本处理,内存占用通常在 200MB 以内。抓取 10 个 RSS 源、生成 20 条摘要,耗时主要取决于模型 API 响应速度,可能在 2 到 5 分钟之间。
如果使用本地 Ollama 服务,情况就不同了。以 7B 参数模型为例,运行量化版本需要约 6GB 显存,加载到 CPU 则内存占用可能超过 8GB。生成速度会比云端 API 慢很多,每条摘要可能需要 10 到 30 秒。因此建议优先使用云端 API 或兼容接口。
观察资源占用的方法:
- Windows 打开任务管理器,查看 Python 进程的 CPU 和内存。
- Linux 使用
htop或nvidia-smi。
命令示例:
htop nvidia-smi --query-gpu=memory.used,utilization.gpu --format=csv如果发现内存占用持续增长,可能是抓取列表没有裁剪,把历史数据全部加载到内存了。在fetch_all_sources之后加entries[-50:]或entries[:50],可以控制内存峰值。
调节性能的几个关键参数:
hours:抓取最近几小时,值越小,条目越少。max_entries:控制发送给模型的条目数量。temperature:调低到 0.2 左右,摘要更稳定,减少模型自由发挥。max_tokens:控制每条摘要的最大长度。不需要太长时建议设为 800,节省费用。
9. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 抓取结果为空 | RSS 源不可访问或 24 小时内无更新 | 用浏览器打开 RSS 地址,检查返回内容 | 更换更强的 RSS 源,或调大hours值 |
| 抓取报超时 | 目标网站响应慢 | 查看网络连接和源站状态 | 给抓取函数加超时参数,或跳过该源 |
| API 密钥报错 | api_key未正确读取或已失效 | 打印配置内容,确认环境变量是否注入 | 检查.env或环境变量设置 |
| 摘要内容与新闻无关 | 提示词不够明确或模型幻觉 | 检查输入新闻是否清晰,调低 temperature | 在 prompt 中加入“必须严格基于输入” |
| 批量任务卡在一个源 | 某个 RSS 源一直不返回 | 查看日志定位具体源 | 用超时和异常捕获隔离单源失败 |
| 端口 8090 被占用 | 其他服务占用了端口 | 查看端口占用情况 | 修改uvicorn.run的端口值 |
| 生成的 Markdown 乱码 | 文件编码不是 UTF-8 | 检查保存文件时是否指定编码 | 使用open(file, "w", encoding="utf-8") |
| 接口返回 500 | 请求参数或模型调用异常 | 查看服务端日志 | 根据日志堆栈修复具体逻辑 |
这里建议大家把日志完整打印出来。最简单的做法是在每个函数入口加一行print,比如print(f"正在处理: {entry['title']}")。后续如果任务失败,能很快定位是哪一段出了问题。
10. 最佳实践与使用建议
经过几轮迭代,我总结了几条非常实用的经验。
第一,第一次运行不要直接跑全量源。先用两个测试源、五条以内条目,确认命令、API、输出格式都正确,再放开批量任务。全量跑一次如果中途出错,排查成本会高很多。
第二,把源按主题拆分。不要把科技、财经、娱乐混在一个配置里。建议为每个主题建立单独的配置文件,比如config_robot.yaml、config_space.yaml、config_nuclear.yaml,然后通过命令行参数指定:
python generate_daily_brief.py --config config_space.yaml这样每个主题的 Prompt 可以独立设计。比如人形机器人主题的 prompt 偏重技术路线对比,核聚变能主题的 prompt 偏重工程进度和商业化节点,数字产业主题的 prompt 偏重政策影响和产业链变化。
第三,输出结果要分目录管理。简单的目录结构如下:
infogap-daily/ config/ config_default.yaml inputs/ rss_cache/ outputs/ 2026-08-26/ 2026-08-27/ logs/每天生成的文件按日期放入对应目录,方便回溯。
第四,定时任务一定要加锁和日志。使用 Python 的filelock防止重复运行,避免每天定时任务因为前一次未结束而后一次又启动,导致重复推送。
# 使用 crontab 每天 8:00 执行 0 8 * * * cd /path/to/infogap-daily && /usr/bin/python generate_daily_brief.py >> logs/cron.log 2>&1第五,涉及人脸、声音、版权素材和敏感领域时,必须确认授权并做人工复核。自动生成的简报只适合作为辅助材料,不能直接对外发布。
11. 总结与下一步
这个“信息差简报生成器”最大的价值不是抓取本身,而是把“抓取 + 分类 + 模型总结 + 接口推送”组合成一条完整流水线。RSS 负责稳定获取信息,大模型负责提炼认知差,FastAPI 负责把能力开放给其他系统。整套东西跑通之后,每天早上看到的不再是几十条孤立新闻,而是一份带有判断框架的主题简报。
最值得先验证的功能是单源抓取和单条摘要。这两个环节跑通,后面批量任务和接口调用只是组合问题。最容易踩的坑有两个:一是 RSS 源不稳定导致结果为空,二是模型提示词设计太弱导致摘要没有信息量。
后续可以继续扩展的方向有很多。比如:加入自定义主题关键词过滤,把与“人形机器人”“航天突破”“核聚变能”“数字产业”无关的新闻自动过滤掉;增加去重模块,避免多个源重复报道同一事件;接入向量数据库做历史热点回溯,判断当前事件是否已经炒过一轮;还可以把每日简报自动汇总到 Notion 或飞书文档,方便长期积累。
这套方案不需要昂贵的硬件,也不需要多复杂的架构,花一个下午就能搭完。如果你每天也被大量热点信息淹没,建议直接照着这份流程搭一个属于自己的“信息差系统”。