最近总能在各种技术群里看到类似“算法推荐把我困在信息茧房里了”的吐槽。刷 B 站全是重复的影视解说打开小红书全是广告软文油管和推特更是被同质化内容塞满。平台推荐算法的核心目标并不是“让你看到你真正想看的”而是“让你停留更久”。于是我们看到的推荐流经常和真实兴趣背道而驰。我一直想做一个自己的信息流过滤器把所有平台内容抓下来用 AI 重新排序只看自己想看的东西。折腾了一段时间后发现最顺手的方案是用 Agent 架构来做这件事。最近在 GitHub 上看到一个相关方向的 Agent 项目已经积累了不少 Stars思路非常适合做二次开发。这篇文章就把整个落地思路整理出来包括 Agent 的工作机制、核心代码结构、以及如何接入 B 站、小红书、油管、推特等多个内容源帮你 10 分钟搭出属于你自己的个性化推荐系统。文章适合对 Agent 开发感兴趣的读者也适合被算法推荐困扰、想自己做信息流过滤的开发者。读完你不仅理解 Agent 项目的基本架构还能实际跑通一个最小可用的个性化信息流系统。1. 背景与核心概念1.1 平台推荐算法与个人信息流之间的矛盾先聊一个基础问题为什么平台推荐总是不对味B 站、小红书、YouTube、Twitter 这类内容平台的推荐系统本质上是一个“多目标优化”系统。平台要同时兼顾用户时长、广告收入、内容生态、新内容冷启动等多个指标。因此推荐列表里的内容并不完全等于“你最想看的内容”而是“平台认为能让你停留更久的内容”。举个例子你可能只是一个偶尔看游戏视频的用户但只要你点过几次游戏视频推荐流里就会塞满大量游戏内容。这是因为平台认为“游戏内容能提高你的留存概率”哪怕你自己已经觉得“最近不想看游戏了”。这种推荐逻辑的滞后性和单一性促使一部分用户开始寻找“自己掌控推荐”的方案。1.2 AI Agent 能做什么AI Agent智能体是一个能够自主完成多步任务的程序。和普通的脚本不同Agent 不只是“按固定流程执行”它可以调用大语言模型LLM进行判断、决策和优化。在信息流场景下Agent 可以做到以下几件事定时从多个平台拉取指定用户、话题或关键词下的内容。对内容进行去重、打分、过滤把垃圾内容和低相关度内容剔除。根据你自己的兴趣标签和历史行为用 LLM 重新排序。生成统一格式的摘要、标签、推荐理由形成你自己的“每日信息流”。简单来说传统 RSS 只是把内容聚合到一起而 Agent 能做的是“读内容、理解内容、判断你是否想看到它”。这篇文章介绍的 Agent 项目就是把“信息获取 大模型判断 个人偏好排序”串成了一个完整的自动化流程。你只需要配置好兴趣源和偏好它就能每天定时生成一份“为你定制”的信息流。1.3 Agent 与传统爬虫、RSS 的区别很多人会问这不就是爬虫吗确实Agent 底层也会抓取内容但二者有本质区别。传统爬虫是“数据搬运工”它关心的是“怎么把页面内容抓下来”一般不做语义理解。RSS 订阅则是“内容聚合”它只是把不同源的更新放在一起排序规则非常简单通常是按发布时间倒序。Agent 的核心能力在于“理解”和“决策”。能力传统爬虫RSS 订阅AI Agent内容获取✅✅✅内容解析部分简单文本✅ 完整语义理解兴趣匹配❌❌✅ LLM 判断自动排序❌按时间✅ 按偏好主动学习❌❌✅ 可迭代优化多平台统一输出❌有限支持✅这也是为什么 Agent 方案更适合做个人信息流它不只是“把内容搬过来”而是“帮你读内容再把值得看的挑出来”。2. 环境准备与项目结构2.1 技术栈选型这个 Agent 项目目前比较主流的实现方式是 Python 技术栈因为它生态成熟适合快速做原型验证。核心组件包括Python 3.10LLM API支持 OpenAI 格式或本地部署的模型均可Feedparser / httpx 用于内容获取APScheduler 用于定时调度SQLite 做轻量级本地缓存与历史去重如果你只用本地小模型也可以跑通只是摘要和排序效果会有一定差异。项目本身设计成了“模型无关”你可以把底层 API 切换成任何兼容 OpenAI 接口的模型服务。版本说明项目对 Python 版本的要求并不严格3.8 以上基本都能运行。但由于部分依赖库的新版本要求 Python 3.9建议直接用 3.10 或 3.11 版本避免环境问题。具体依赖版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。2.2 运行环境无论你使用的是 Windows、macOS 还是 Linux都可以运行该项目。建议使用虚拟环境venv 或 conda隔离依赖。python -m venv feed-agent-env source feed-agent-env/bin/activate # Windows 使用 feed-agent-env\Scripts\activate2.3 项目目录结构为了让代码结构清晰我建议把整个 Agent 拆成下面这样的目录feed-agent/ ├── agent/ │ ├── __init__.py │ ├── core.py # Agent 核心调度逻辑 │ ├── router.py # 内容源路由与适配 │ ├── llm.py # LLM 调用与提示词管理 │ └── filters.py # 去重、评分、过滤逻辑 ├── sources/ │ ├── __init__.py │ ├── base.py # 数据源抽象基类 │ ├── bilibili.py # B站适配器 │ ├── xiaohongshu.py # 小红书适配器 │ ├── youtube.py # YouTube 适配器 │ └── twitter.py # Twitter 适配器 ├── config/ │ └── config.yaml # 配置文件 ├── storage/ │ └── db.py # SQLite 存储 ├── main.py # 程序入口 ├── requirements.txt # 依赖清单 └── README.md这个结构把“数据获取”“内容理解”“调度执行”“存储”四层分离开来后续要增加新的内容平台只需要新增一个 source 适配器即可。3. 核心原理拆解Agent 是如何工作的3.1 Agent 的工作流程整个信息流 Agent 的运行流程可以拆成四个阶段第一阶段内容收集。Agent 根据你配置的数据源列表从 B 站、小红书、YouTube、Twitter 等平台拉取指定用户、话题或关键词下的内容。这里需要说明不同平台的开放程度不一样。B 站和 YouTube 都有相对稳定的接口或 RSS 通道Twitter 和部分平台由于接口限制较多可能需要借助官方 API 或合规的第三方数据服务。个人项目建议优先使用官方公开接口或模拟数据做演示。第二阶段内容标准化。不同平台返回的数据格式差异巨大。B 站视频有 BV 号、分P、UP 主等信息小红书笔记有标题、正文、标签、图片链接YouTube 视频有时长、频道名、观看次数推特推文有转发数、点赞数、话题标签。Agent 需要把这些异构数据统一转换成内部标准格式后续处理才能一致。第三阶段LLM 理解与评分。这是整个 Agent 最核心的部分。把标准化后的内容标题、摘要、标签拼装成 Prompt发送给 LLM让模型判断这条内容与你的兴趣标签的匹配程度输出一个分数比如 0-100以及推荐理由。第四阶段排序与输出。根据 LLM 评分对内容进行排序过滤掉低分内容和历史重复内容最终生成一个 Markdown 或 HTML 格式的信息流报告。3.2 LLM 调用与提示词设计LLM 的质量直接决定了推荐效果。设计提示词时需要把“你的兴趣标签”和“待判断内容”一起传进去。下面是一个参考提示词结构你是一个个人内容推荐助手。用户对以下主题感兴趣 - AI 编程 - 独立开发者工具 - 技术复盘 请根据以下内容判断它与用户兴趣的相关程度。 内容标题{标题} 内容摘要{摘要} 内容标签{标签} 请返回严格 JSON 格式的结果 { reason: 这条内容与 AI 编程相关且是实战教程对用户有参考价值, score: 92 }注意这里必须强制让模型输出 JSON 格式否则后续解析会很麻烦。实际项目中可以在 Prompt 中加上“只输出 JSON不要输出其他内容”同时在代码里做一次 JSON 解析异常兜底。3.3 多平台适配器设计适配器模式是这个项目最有复用价值的部分。每个平台对应一个适配器类只需要实现固定的接口方法即可接入。基础抽象类大致长这样# 文件路径sources/base.py from abc import ABC, abstractmethod class BaseSource(ABC): 所有内容源适配器的抽象基类 abstractmethod def fetch(self, config: dict) - list[dict]: 从平台获取内容列表 返回格式 [ { platform: bilibili, content_id: BV1xxxx, title: 视频标题, summary: 摘要或简介, tags: [AI, 编程], url: https://www.bilibili.com/video/BV1xxxx, published_at: 2025-01-01 12:00:00, }, ] pass有了这个抽象基类新增一个平台只需要实现一个 fetch 方法返回统一格式的字典列表就可以直接参与后续的过滤和排序流程。3.4 历史去重与缓存信息流系统最容易忽略的问题就是“重复推荐”。如果你的 Agent 每小时跑一次同一个视频就会被重复推好几次。解决方案是在 SQLite 里维护一张内容表用platform content_id作为唯一键。每次拉取到新内容后先查询数据库判断是否已经存在如果存在就跳过。同时还可以保存模型评分和推荐理由避免重复调用 LLM 浪费费用。CREATE TABLE IF NOT EXISTS content ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, content_id TEXT NOT NULL, title TEXT, summary TEXT, tags TEXT, url TEXT, score REAL, reason TEXT, published_at TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(platform, content_id) );4. 完整实战本地跑通第一版4.1 创建项目与依赖先创建项目目录和虚拟环境mkdir feed-agent cd feed-agent python -m venv venv source venv/bin/activate创建requirements.txt写入依赖httpx0.27.0 feedparser6.0.11 apscheduler3.10.4 pyyaml6.0.1 openai1.30.0安装依赖pip install -r requirements.txt4.2 配置文件创建config/config.yaml。这个配置文件包括三部分模型配置、兴趣配置、数据源配置。# 文件路径config/config.yaml llm: provider: openai # 使用 OpenAI 兼容接口 base_url: https://api.example.com/v1 # 根据你的模型服务地址调整 api_key: sk-xxx # 建议从环境变量读取 model: gpt-4o-mini # 按实际可用模型调整 temperature: 0.3 # 评分场景尽量低 interests: tags: - AI 编程 - 独立开发 - 开源项目 - 技术复盘 min_score: 60 # 低于该分数直接过滤 sources: bilibili: enabled: true # 关注 UP 主的 UID 列表可以通过 B站页面获取 up_uids: - 123456789 xiaohongshu: enabled: false youtube: enabled: true channel_ids: - UCxxxxxxxxxxxx twitter: enabled: false schedule: interval_minutes: 60 # 每小时执行一次注意配置文件里的api_key不应该硬编码提交到 Git 仓库。生产环境建议使用环境变量注入例如import os api_key os.getenv(LLM_API_KEY, )4.3 核心代码实现先来实现 LLM 调用模块。这个模块负责把内容列表批量发送给大模型并获得评分结果。# 文件路径agent/llm.py import json from openai import OpenAI class LLMScorer: def __init__(self, config: dict): self.client OpenAI( base_urlconfig[base_url], api_keyconfig[api_key], ) self.model config[model] self.temperature config.get(temperature, 0.3) self.interests config.get(interests, []) def build_prompt(self, item: dict) - str: interests_text \n.join(f- {tag} for tag in self.interests) return f 你是一个个人内容推荐助手。用户对以下主题感兴趣 {interests_text} 请判断以下内容与用户兴趣的相关程度 标题{item[title]} 摘要{item.get(summary, )[:200]} 标签{, .join(item.get(tags, [])[:5])} 请只返回 JSON不要返回任何其他文本格式如下 {{ reason: 简要说明理由, score: 0 }} 评分规则 - 80-100强烈推荐与用户兴趣高度相关 - 60-79值得一看有一定关联 - 40-59一般关联较弱 - 0-39不推荐 def score_item(self, item: dict) - dict: prompt self.build_prompt(item) try: response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: 你是一个精准的内容推荐评分器。}, {role: user, content: prompt}, ], temperatureself.temperature, ) text response.choices[0].message.content # 清理可能的 Markdown 代码块标记 text text.strip().removeprefix(json).removeprefix().removesuffix() result json.loads(text) return { score: float(result.get(score, 0)), reason: result.get(reason, ), } except Exception as e: return { score: 0.0, reason: fLLM 评分失败{e}, }这里需要说明不同模型返回 JSON 的稳定性差异比较大。为了减少解析失败最好在 Prompt 中明确“只返回 JSON”同时在代码里做好异常兜底。接下来实现 Agent 核心逻辑。这个模块负责把“获取内容 → LLM 评分 → 排序过滤 → 入库”串起来。# 文件路径agent/core.py import sqlite3 from datetime import datetime from agent.llm import LLMScorer from storage.db import init_db, is_duplicate, save_content class FeedAgent: def __init__(self, config: dict): self.config config self.interests config[interests][tags] self.min_score config[interests][min_score] self.llm_scorer LLMScorer({ **config[llm], interests: self.interests, }) self.sources [] init_db() def register_source(self, source): 注册内容源适配器 self.sources.append(source) def run_once(self) - list[dict]: 执行一轮完整的信息流处理 all_items [] for source in self.sources: items source.fetch(self.config.get(sources, {})) all_items.extend(items) print(f[{source.__class__.__name__}] 获取到 {len(items)} 条内容) results [] for item in all_items: # 1. 去重检查 if is_duplicate(item[platform], item[content_id]): continue # 2. LLM 评分 scored self.llm_scorer.score_item(item) # 3. 分数过滤 if scored[score] self.min_score: continue # 4. 保存结果 save_content(item, scored) results.append({**item, **scored}) # 按分数从高到低排序 results.sort(keylambda x: x[score], reverseTrue) return results再来看一个 B 站数据源的实现示例。这里用 B 站用户公开的 RSS 通道或页面接口做演示实际使用时需要根据平台当前可用的数据获取方式调整。# 文件路径sources/bilibili.py import httpx from sources.base import BaseSource class BilibiliSource(BaseSource): B站内容源适配器 def fetch(self, config: dict) - list[dict]: bilibili_config config.get(bilibili, {}) if not bilibili_config.get(enabled): return [] up_uids bilibili_config.get(up_uids, []) items [] for uid in up_uids: # 这里以 UP 主动态接口为例实际请以 B站当前公开接口为准 # 注意生产环境请遵守平台条款控制请求频率 url fhttps://api.bilibili.com/x/polymer/web-dynamic/v1/feed/space params {host_mid: uid} headers {User-Agent: Mozilla/5.0} try: resp httpx.get(url, paramsparams, headersheaders, timeout10) data resp.json() # 简化处理从动态列表中提取视频或图文内容 for card in data.get(data, {}).get(items, [])[:10]: item self._parse_card(card) if item: items.append(item) except Exception as e: print(f[Bilibili] 获取账号 {uid} 内容失败{e}) return items def _parse_card(self, card: dict) - dict | None: 解析动态卡片提取统一格式信息 try: module card.get(modules, {}) author module.get(author, {}) content module.get(archive, {}) or module.get(dynamic, {}) return { platform: bilibili, content_id: card.get(id_str, ), title: content.get(title) or content.get(description, 无标题), summary: (content.get(description) or )[:200], tags: [], # B站动态没有统一标签字段可后续提取 url: fhttps://www.bilibili.com/video/{content.get(bvid, )}, published_at: str(module.get(pub_time, )), } except Exception: return None最后实现程序入口# 文件路径main.py import yaml from sources.bilibili import BilibiliSource from agent.core import FeedAgent def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main(): config load_config(config/config.yaml) agent FeedAgent(config) agent.register_source(BilibiliSource()) results agent.run_once() print(\n 今日推荐 ) for idx, item in enumerate(results, start1): print(f{idx}. {item[title]} (评分{item[score]})) print(f 理由{item[reason]}) print(f 链接{item[url]}\n) # 可进一步输出为 Markdown 文件 with open(daily_report.md, w, encodingutf-8) as f: f.write(# 今日个性化推荐\n\n) for idx, item in enumerate(results, start1): f.write(f## {idx}. {item[title]}\n\n) f.write(f- 平台{item[platform]}\n) f.write(f- 评分{item[score]}\n) f.write(f- 理由{item[reason]}\n) f.write(f- 链接{item[url]}\n\n) if __name__ __main__: main()4.4 运行与验证在项目根目录执行python main.py如果配置正确你会看到类似下面的输出[BilibiliSource] 获取到 10 条内容 [LLM] 评分完成3/10 [LLM] 评分完成5/10 [LLM] 评分完成8/10 [LLM] 评分完成10/10 今日推荐 1. 【AI编程实战】用 Agent 自动写单元测试 (评分95.0) 理由内容与用户“AI 编程”兴趣标签高度匹配并涉及工具实战。 链接https://www.bilibili.com/video/BVxxxx ...同时项目目录下会生成daily_report.md里面是排版好的 Markdown 推荐列表。4.5 结果说明第一版虽然简单但整个 Agent 的核心链路已经跑通了。数据源获取内容 → LLM 理解内容 → 按兴趣过滤排序 → 输出个人推荐列表四个环节全部打通。后续你可以把调度器加上让 Agent 定时执行# 文件路径schedule_demo.py from apscheduler.schedulers.blocking import BlockingScheduler from agent.core import FeedAgent scheduler BlockingScheduler() def job(): print(开始执行信息流抓取任务...) agent FeedAgent(config) agent.register_source(BilibiliSource()) results agent.run_once() print(f处理完成共 {len(results)} 条推荐。) scheduler.add_job(job, interval, minutes60) scheduler.start()5. 进阶接入多个平台内容源5.1 适配器模式的优势前面提到的适配器模式在这个过程中优势非常明显。你不需要改动核心逻辑只需要新增一个文件实现同样的fetch方法然后在main.py里注册即可。下面以“模拟小红书数据源”为例演示如何快速新增一个内容源。由于小红书官方开放接口有限这里使用模拟数据来演示适配器写法实际使用时可以替换为合规的数据获取方式。# 文件路径sources/xiaohongshu.py from sources.base import BaseSource class XiaohongshuSource(BaseSource): 小红书内容源适配器演示用模拟实现 def fetch(self, config: dict) - list[dict]: xhs_config config.get(xiaohongshu, {}) if not xhs_config.get(enabled): return [] # 实际项目中这里应替换为合规的数据获取逻辑 # 例如官方开放平台接口、自己账号授权后的数据导出等 mock_items [ { platform: xiaohongshu, content_id: demo-001, title: 分享一个提升开发效率的AI工具, summary: 最近发现一款AI工具配合 Agent 使用非常顺手..., tags: [AI工具, 效率], url: https://www.xiaohongshu.com/explore/demo-001, published_at: 2025-01-10 10:00:00, }, { platform: xiaohongshu, content_id: demo-002, title: 周末学习打卡LLM 入门笔记, summary: 整理了最近学习大语言模型的知识点..., tags: [LLM, 学习], url: https://www.xiaohongshu.com/explore/demo-002, published_at: 2025-01-11 20:30:00, }, ] return mock_items然后在入口注册# main.py 中新增 from sources.xiaohongshu import XiaohongshuSource agent.register_source(BilibiliSource()) agent.register_source(XiaohongshuSource())通过这种扩展方式你可以轻松叠加 YouTube、Twitter、即刻、少数派、知乎等任意内容来源。需要做的只有两件事写一个 fetch 方法统一数据格式。5.2 统一数据格式的重要性可能有读者会问为什么非要统一成固定的 dict 结构因为后续的“去重”、“LLM 评分”、“排序”都是无差别处理每一条内容的。如果每个平台返回的数据结构都不一样核心逻辑里会堆满各种if platform bilibili分支代码很快会变得腐烂。统一格式的本质是定义了一个“内部协议”让所有外部数据源在进入核心处理链路之前完成格式转换。这是整个 Agent 架构中最重要的设计决策之一。6. 常见问题与排查思路实际开发中最容易踩坑的点不在 Agent 核心逻辑而是在数据获取和模型调用这两层。下面列出一份排查清单。问题现象常见原因解决思路获取内容为空数据源接口变更或未授权先用浏览器或 curl 验证接口是否可用检查返回状态码和数据结构LLM 返回 JSON 解析失败模型输出包含额外文字Prompt 中强制“只返回 JSON”代码中做字符串清理去掉代码块标记评分全部为 0API Key 错误或模型名称不对单独写一个测试脚本验证模型连接是否正常重复内容很多未启用去重或历史数据丢失检查 SQLite 表是否存在确认唯一键设置是否正确抓取频率过高被封请求间隔太短或缺少限速在适配器中增加限速逻辑控制单次请求间隔和单日调用量内存占用持续增长内容列表过大未做分页限制每个数据源限制单次拉取条数建议 10-20 条以内定时任务不触发时区或调度配置错误检查 APScheduler 时区设置确认interval_minutes是否读取成功这里重点展开两个高频问题。问题一LLM 评分结果不稳定。同一个内容跑两次评分可能得到完全不同的分数。原因是 LLM 本身具有概率性尤其在 temperature 较高时随机性更大。解决方法是将 temperature 调低到 0.2-0.3让输出更稳定。对同一条内容采样多次取平均值一般 2-3 次即可。在 Prompt 中给出明确评分标准和参考示例减少歧义。问题二数据平台接口经常变动。B 站、小红书等平台的页面结构和接口并不稳定可能几个月就调整一次。如果你的适配器突然拿不到数据优先检查接口返回的 JSON 是否还是原来的结构。是否新增了登录校验或风控。是否要求配置 Cookie 或 Token。生产使用中建议把数据获取做成“可降级”的。获取失败时使用上次缓存的数据继续跑保证信息流不会中断。7. 最佳实践与工程建议7.1 模型选型与成本控制如果每天只跑一次、每次 50 条内容LLM 的调用量其实很小。但如果你把调度频率提高到每小时成本会快速上升。一些成本控制建议先用便宜的模型跑全量过滤再用高质量模型对 Top 20 的内容做精细摘要。对同一内容源的内容做关键词前置过滤明显不相关的直接跳过 LLM 评分。内容入库后如果已经评分过不再重复调用 LLM。7.2 隐私与安全边界个人信息流系统会采集你的观看历史、兴趣标签和关注列表这些数据非常敏感。需要注意数据尽量本地存储不要上传到第三方服务器。如果使用云端 LLM API避免把包含个人隐私的完整内容发送给模型只发送摘要和标题片段。不要把 API Key 提交到公开仓库使用环境变量或本地密钥文件管理。定期清理 SQLite 中的历史数据避免长期积累。7.3 平台合规建议不同平台对数据抓取的政策差异很大。在搭建系统时请务必注意优先使用平台官方提供的 API、RSS、开放接口。控制请求频率避免对平台服务器造成压力。仅用于个人学习和研究不要将抓取内容进行二次分发。如果涉及商业用途必须获得平台书面授权。这篇文章的示例使用模拟数据演示就是为了避开平台接口限制问题。你在实际接入时需要自行确认目标平台的最新政策和技术方案。7.4 可维护性设计Agent 项目虽然小但长期运行后维护成本会逐渐凸显。几个建议每个数据源独立成文件保持单一职责。配置与代码分离兴趣标签、模型参数统一放 YAML。日志要完整至少记录每次抓取的源、条数、耗时、评分结果。用dataclass定义统一的内容结构代替纯字典提升代码可读性。7.5 失败降级与重试机制Agent 是无人值守运行的程序网络波动、接口异常、模型服务不可用都可能发生。核心逻辑需要做好降级# 伪代码降级逻辑示意 def run_once(self): results [] for source in self.sources: try: items source.fetch(self.config) except Exception as e: print(f数据源 {source} 获取失败使用缓存数据。原因{e}) items load_from_cache(source) for item in items: scored self.llm_scorer.score_item(item) if scored[score] self.min_score: results.append({**item, **scored}) return sorted(results, keylambda x: x[score], reverseTrue)8. 总结与动手方向到这里一个基于 Agent 的个性化信息流系统已经完整落地了。回顾一下核心内容Agent 能够主动获取内容、理解内容、判断相关性替代传统 RSS 的被动订阅模式。多平台适配器是接入 B 站、小红书、YouTube、Twitter 等数据源的核心设计。LLM 评分不是越复杂越好关键是 Prompt 设计和成本控制。去重、缓存、降级是长期运行稳定性的保障。下一步你可以从这几个方向继续深入把推荐报告通过企业微信、钉钉、Telegram Bot 推送每天 8 点自动发送。增加“反馈机制”看到不感兴趣的内容可以标记负反馈Agent 下次评分时参考。接入本地大模型例如 Ollama实现完全离线运行避免数据外流。用 Agent 自动抓取完整正文内容生成摘要式简报形成你的“每日早报”。如果你也受够了被平台推荐算法支配的感觉可以动手把这个系统搭起来。第一次跑通时你大概率会遇到 JSON 解析失败、接口返回结构变化这类小坑这些都是 Agent 开发的必经之路。把排查过程记录下来本身就是一笔很好的技术积累。
最近总能在各种技术群里看到类似“算法推荐把我困在信息茧房里了”的吐槽。刷 B 站全是重复的影视解说打开小红书全是广告软文油管和推特更是被同质化内容塞满。平台推荐算法的核心目标并不是“让你看到你真正想看的”而是“让你停留更久”。于是我们看到的推荐流经常和真实兴趣背道而驰。我一直想做一个自己的信息流过滤器把所有平台内容抓下来用 AI 重新排序只看自己想看的东西。折腾了一段时间后发现最顺手的方案是用 Agent 架构来做这件事。最近在 GitHub 上看到一个相关方向的 Agent 项目已经积累了不少 Stars思路非常适合做二次开发。这篇文章就把整个落地思路整理出来包括 Agent 的工作机制、核心代码结构、以及如何接入 B 站、小红书、油管、推特等多个内容源帮你 10 分钟搭出属于你自己的个性化推荐系统。文章适合对 Agent 开发感兴趣的读者也适合被算法推荐困扰、想自己做信息流过滤的开发者。读完你不仅理解 Agent 项目的基本架构还能实际跑通一个最小可用的个性化信息流系统。1. 背景与核心概念1.1 平台推荐算法与个人信息流之间的矛盾先聊一个基础问题为什么平台推荐总是不对味B 站、小红书、YouTube、Twitter 这类内容平台的推荐系统本质上是一个“多目标优化”系统。平台要同时兼顾用户时长、广告收入、内容生态、新内容冷启动等多个指标。因此推荐列表里的内容并不完全等于“你最想看的内容”而是“平台认为能让你停留更久的内容”。举个例子你可能只是一个偶尔看游戏视频的用户但只要你点过几次游戏视频推荐流里就会塞满大量游戏内容。这是因为平台认为“游戏内容能提高你的留存概率”哪怕你自己已经觉得“最近不想看游戏了”。这种推荐逻辑的滞后性和单一性促使一部分用户开始寻找“自己掌控推荐”的方案。1.2 AI Agent 能做什么AI Agent智能体是一个能够自主完成多步任务的程序。和普通的脚本不同Agent 不只是“按固定流程执行”它可以调用大语言模型LLM进行判断、决策和优化。在信息流场景下Agent 可以做到以下几件事定时从多个平台拉取指定用户、话题或关键词下的内容。对内容进行去重、打分、过滤把垃圾内容和低相关度内容剔除。根据你自己的兴趣标签和历史行为用 LLM 重新排序。生成统一格式的摘要、标签、推荐理由形成你自己的“每日信息流”。简单来说传统 RSS 只是把内容聚合到一起而 Agent 能做的是“读内容、理解内容、判断你是否想看到它”。这篇文章介绍的 Agent 项目就是把“信息获取 大模型判断 个人偏好排序”串成了一个完整的自动化流程。你只需要配置好兴趣源和偏好它就能每天定时生成一份“为你定制”的信息流。1.3 Agent 与传统爬虫、RSS 的区别很多人会问这不就是爬虫吗确实Agent 底层也会抓取内容但二者有本质区别。传统爬虫是“数据搬运工”它关心的是“怎么把页面内容抓下来”一般不做语义理解。RSS 订阅则是“内容聚合”它只是把不同源的更新放在一起排序规则非常简单通常是按发布时间倒序。Agent 的核心能力在于“理解”和“决策”。能力传统爬虫RSS 订阅AI Agent内容获取✅✅✅内容解析部分简单文本✅ 完整语义理解兴趣匹配❌❌✅ LLM 判断自动排序❌按时间✅ 按偏好主动学习❌❌✅ 可迭代优化多平台统一输出❌有限支持✅这也是为什么 Agent 方案更适合做个人信息流它不只是“把内容搬过来”而是“帮你读内容再把值得看的挑出来”。2. 环境准备与项目结构2.1 技术栈选型这个 Agent 项目目前比较主流的实现方式是 Python 技术栈因为它生态成熟适合快速做原型验证。核心组件包括Python 3.10LLM API支持 OpenAI 格式或本地部署的模型均可Feedparser / httpx 用于内容获取APScheduler 用于定时调度SQLite 做轻量级本地缓存与历史去重如果你只用本地小模型也可以跑通只是摘要和排序效果会有一定差异。项目本身设计成了“模型无关”你可以把底层 API 切换成任何兼容 OpenAI 接口的模型服务。版本说明项目对 Python 版本的要求并不严格3.8 以上基本都能运行。但由于部分依赖库的新版本要求 Python 3.9建议直接用 3.10 或 3.11 版本避免环境问题。具体依赖版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。2.2 运行环境无论你使用的是 Windows、macOS 还是 Linux都可以运行该项目。建议使用虚拟环境venv 或 conda隔离依赖。python -m venv feed-agent-env source feed-agent-env/bin/activate # Windows 使用 feed-agent-env\Scripts\activate2.3 项目目录结构为了让代码结构清晰我建议把整个 Agent 拆成下面这样的目录feed-agent/ ├── agent/ │ ├── __init__.py │ ├── core.py # Agent 核心调度逻辑 │ ├── router.py # 内容源路由与适配 │ ├── llm.py # LLM 调用与提示词管理 │ └── filters.py # 去重、评分、过滤逻辑 ├── sources/ │ ├── __init__.py │ ├── base.py # 数据源抽象基类 │ ├── bilibili.py # B站适配器 │ ├── xiaohongshu.py # 小红书适配器 │ ├── youtube.py # YouTube 适配器 │ └── twitter.py # Twitter 适配器 ├── config/ │ └── config.yaml # 配置文件 ├── storage/ │ └── db.py # SQLite 存储 ├── main.py # 程序入口 ├── requirements.txt # 依赖清单 └── README.md这个结构把“数据获取”“内容理解”“调度执行”“存储”四层分离开来后续要增加新的内容平台只需要新增一个 source 适配器即可。3. 核心原理拆解Agent 是如何工作的3.1 Agent 的工作流程整个信息流 Agent 的运行流程可以拆成四个阶段第一阶段内容收集。Agent 根据你配置的数据源列表从 B 站、小红书、YouTube、Twitter 等平台拉取指定用户、话题或关键词下的内容。这里需要说明不同平台的开放程度不一样。B 站和 YouTube 都有相对稳定的接口或 RSS 通道Twitter 和部分平台由于接口限制较多可能需要借助官方 API 或合规的第三方数据服务。个人项目建议优先使用官方公开接口或模拟数据做演示。第二阶段内容标准化。不同平台返回的数据格式差异巨大。B 站视频有 BV 号、分P、UP 主等信息小红书笔记有标题、正文、标签、图片链接YouTube 视频有时长、频道名、观看次数推特推文有转发数、点赞数、话题标签。Agent 需要把这些异构数据统一转换成内部标准格式后续处理才能一致。第三阶段LLM 理解与评分。这是整个 Agent 最核心的部分。把标准化后的内容标题、摘要、标签拼装成 Prompt发送给 LLM让模型判断这条内容与你的兴趣标签的匹配程度输出一个分数比如 0-100以及推荐理由。第四阶段排序与输出。根据 LLM 评分对内容进行排序过滤掉低分内容和历史重复内容最终生成一个 Markdown 或 HTML 格式的信息流报告。3.2 LLM 调用与提示词设计LLM 的质量直接决定了推荐效果。设计提示词时需要把“你的兴趣标签”和“待判断内容”一起传进去。下面是一个参考提示词结构你是一个个人内容推荐助手。用户对以下主题感兴趣 - AI 编程 - 独立开发者工具 - 技术复盘 请根据以下内容判断它与用户兴趣的相关程度。 内容标题{标题} 内容摘要{摘要} 内容标签{标签} 请返回严格 JSON 格式的结果 { reason: 这条内容与 AI 编程相关且是实战教程对用户有参考价值, score: 92 }注意这里必须强制让模型输出 JSON 格式否则后续解析会很麻烦。实际项目中可以在 Prompt 中加上“只输出 JSON不要输出其他内容”同时在代码里做一次 JSON 解析异常兜底。3.3 多平台适配器设计适配器模式是这个项目最有复用价值的部分。每个平台对应一个适配器类只需要实现固定的接口方法即可接入。基础抽象类大致长这样# 文件路径sources/base.py from abc import ABC, abstractmethod class BaseSource(ABC): 所有内容源适配器的抽象基类 abstractmethod def fetch(self, config: dict) - list[dict]: 从平台获取内容列表 返回格式 [ { platform: bilibili, content_id: BV1xxxx, title: 视频标题, summary: 摘要或简介, tags: [AI, 编程], url: https://www.bilibili.com/video/BV1xxxx, published_at: 2025-01-01 12:00:00, }, ] pass有了这个抽象基类新增一个平台只需要实现一个 fetch 方法返回统一格式的字典列表就可以直接参与后续的过滤和排序流程。3.4 历史去重与缓存信息流系统最容易忽略的问题就是“重复推荐”。如果你的 Agent 每小时跑一次同一个视频就会被重复推好几次。解决方案是在 SQLite 里维护一张内容表用platform content_id作为唯一键。每次拉取到新内容后先查询数据库判断是否已经存在如果存在就跳过。同时还可以保存模型评分和推荐理由避免重复调用 LLM 浪费费用。CREATE TABLE IF NOT EXISTS content ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, content_id TEXT NOT NULL, title TEXT, summary TEXT, tags TEXT, url TEXT, score REAL, reason TEXT, published_at TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(platform, content_id) );4. 完整实战本地跑通第一版4.1 创建项目与依赖先创建项目目录和虚拟环境mkdir feed-agent cd feed-agent python -m venv venv source venv/bin/activate创建requirements.txt写入依赖httpx0.27.0 feedparser6.0.11 apscheduler3.10.4 pyyaml6.0.1 openai1.30.0安装依赖pip install -r requirements.txt4.2 配置文件创建config/config.yaml。这个配置文件包括三部分模型配置、兴趣配置、数据源配置。# 文件路径config/config.yaml llm: provider: openai # 使用 OpenAI 兼容接口 base_url: https://api.example.com/v1 # 根据你的模型服务地址调整 api_key: sk-xxx # 建议从环境变量读取 model: gpt-4o-mini # 按实际可用模型调整 temperature: 0.3 # 评分场景尽量低 interests: tags: - AI 编程 - 独立开发 - 开源项目 - 技术复盘 min_score: 60 # 低于该分数直接过滤 sources: bilibili: enabled: true # 关注 UP 主的 UID 列表可以通过 B站页面获取 up_uids: - 123456789 xiaohongshu: enabled: false youtube: enabled: true channel_ids: - UCxxxxxxxxxxxx twitter: enabled: false schedule: interval_minutes: 60 # 每小时执行一次注意配置文件里的api_key不应该硬编码提交到 Git 仓库。生产环境建议使用环境变量注入例如import os api_key os.getenv(LLM_API_KEY, )4.3 核心代码实现先来实现 LLM 调用模块。这个模块负责把内容列表批量发送给大模型并获得评分结果。# 文件路径agent/llm.py import json from openai import OpenAI class LLMScorer: def __init__(self, config: dict): self.client OpenAI( base_urlconfig[base_url], api_keyconfig[api_key], ) self.model config[model] self.temperature config.get(temperature, 0.3) self.interests config.get(interests, []) def build_prompt(self, item: dict) - str: interests_text \n.join(f- {tag} for tag in self.interests) return f 你是一个个人内容推荐助手。用户对以下主题感兴趣 {interests_text} 请判断以下内容与用户兴趣的相关程度 标题{item[title]} 摘要{item.get(summary, )[:200]} 标签{, .join(item.get(tags, [])[:5])} 请只返回 JSON不要返回任何其他文本格式如下 {{ reason: 简要说明理由, score: 0 }} 评分规则 - 80-100强烈推荐与用户兴趣高度相关 - 60-79值得一看有一定关联 - 40-59一般关联较弱 - 0-39不推荐 def score_item(self, item: dict) - dict: prompt self.build_prompt(item) try: response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: 你是一个精准的内容推荐评分器。}, {role: user, content: prompt}, ], temperatureself.temperature, ) text response.choices[0].message.content # 清理可能的 Markdown 代码块标记 text text.strip().removeprefix(json).removeprefix().removesuffix() result json.loads(text) return { score: float(result.get(score, 0)), reason: result.get(reason, ), } except Exception as e: return { score: 0.0, reason: fLLM 评分失败{e}, }这里需要说明不同模型返回 JSON 的稳定性差异比较大。为了减少解析失败最好在 Prompt 中明确“只返回 JSON”同时在代码里做好异常兜底。接下来实现 Agent 核心逻辑。这个模块负责把“获取内容 → LLM 评分 → 排序过滤 → 入库”串起来。# 文件路径agent/core.py import sqlite3 from datetime import datetime from agent.llm import LLMScorer from storage.db import init_db, is_duplicate, save_content class FeedAgent: def __init__(self, config: dict): self.config config self.interests config[interests][tags] self.min_score config[interests][min_score] self.llm_scorer LLMScorer({ **config[llm], interests: self.interests, }) self.sources [] init_db() def register_source(self, source): 注册内容源适配器 self.sources.append(source) def run_once(self) - list[dict]: 执行一轮完整的信息流处理 all_items [] for source in self.sources: items source.fetch(self.config.get(sources, {})) all_items.extend(items) print(f[{source.__class__.__name__}] 获取到 {len(items)} 条内容) results [] for item in all_items: # 1. 去重检查 if is_duplicate(item[platform], item[content_id]): continue # 2. LLM 评分 scored self.llm_scorer.score_item(item) # 3. 分数过滤 if scored[score] self.min_score: continue # 4. 保存结果 save_content(item, scored) results.append({**item, **scored}) # 按分数从高到低排序 results.sort(keylambda x: x[score], reverseTrue) return results再来看一个 B 站数据源的实现示例。这里用 B 站用户公开的 RSS 通道或页面接口做演示实际使用时需要根据平台当前可用的数据获取方式调整。# 文件路径sources/bilibili.py import httpx from sources.base import BaseSource class BilibiliSource(BaseSource): B站内容源适配器 def fetch(self, config: dict) - list[dict]: bilibili_config config.get(bilibili, {}) if not bilibili_config.get(enabled): return [] up_uids bilibili_config.get(up_uids, []) items [] for uid in up_uids: # 这里以 UP 主动态接口为例实际请以 B站当前公开接口为准 # 注意生产环境请遵守平台条款控制请求频率 url fhttps://api.bilibili.com/x/polymer/web-dynamic/v1/feed/space params {host_mid: uid} headers {User-Agent: Mozilla/5.0} try: resp httpx.get(url, paramsparams, headersheaders, timeout10) data resp.json() # 简化处理从动态列表中提取视频或图文内容 for card in data.get(data, {}).get(items, [])[:10]: item self._parse_card(card) if item: items.append(item) except Exception as e: print(f[Bilibili] 获取账号 {uid} 内容失败{e}) return items def _parse_card(self, card: dict) - dict | None: 解析动态卡片提取统一格式信息 try: module card.get(modules, {}) author module.get(author, {}) content module.get(archive, {}) or module.get(dynamic, {}) return { platform: bilibili, content_id: card.get(id_str, ), title: content.get(title) or content.get(description, 无标题), summary: (content.get(description) or )[:200], tags: [], # B站动态没有统一标签字段可后续提取 url: fhttps://www.bilibili.com/video/{content.get(bvid, )}, published_at: str(module.get(pub_time, )), } except Exception: return None最后实现程序入口# 文件路径main.py import yaml from sources.bilibili import BilibiliSource from agent.core import FeedAgent def load_config(path: str) - dict: with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main(): config load_config(config/config.yaml) agent FeedAgent(config) agent.register_source(BilibiliSource()) results agent.run_once() print(\n 今日推荐 ) for idx, item in enumerate(results, start1): print(f{idx}. {item[title]} (评分{item[score]})) print(f 理由{item[reason]}) print(f 链接{item[url]}\n) # 可进一步输出为 Markdown 文件 with open(daily_report.md, w, encodingutf-8) as f: f.write(# 今日个性化推荐\n\n) for idx, item in enumerate(results, start1): f.write(f## {idx}. {item[title]}\n\n) f.write(f- 平台{item[platform]}\n) f.write(f- 评分{item[score]}\n) f.write(f- 理由{item[reason]}\n) f.write(f- 链接{item[url]}\n\n) if __name__ __main__: main()4.4 运行与验证在项目根目录执行python main.py如果配置正确你会看到类似下面的输出[BilibiliSource] 获取到 10 条内容 [LLM] 评分完成3/10 [LLM] 评分完成5/10 [LLM] 评分完成8/10 [LLM] 评分完成10/10 今日推荐 1. 【AI编程实战】用 Agent 自动写单元测试 (评分95.0) 理由内容与用户“AI 编程”兴趣标签高度匹配并涉及工具实战。 链接https://www.bilibili.com/video/BVxxxx ...同时项目目录下会生成daily_report.md里面是排版好的 Markdown 推荐列表。4.5 结果说明第一版虽然简单但整个 Agent 的核心链路已经跑通了。数据源获取内容 → LLM 理解内容 → 按兴趣过滤排序 → 输出个人推荐列表四个环节全部打通。后续你可以把调度器加上让 Agent 定时执行# 文件路径schedule_demo.py from apscheduler.schedulers.blocking import BlockingScheduler from agent.core import FeedAgent scheduler BlockingScheduler() def job(): print(开始执行信息流抓取任务...) agent FeedAgent(config) agent.register_source(BilibiliSource()) results agent.run_once() print(f处理完成共 {len(results)} 条推荐。) scheduler.add_job(job, interval, minutes60) scheduler.start()5. 进阶接入多个平台内容源5.1 适配器模式的优势前面提到的适配器模式在这个过程中优势非常明显。你不需要改动核心逻辑只需要新增一个文件实现同样的fetch方法然后在main.py里注册即可。下面以“模拟小红书数据源”为例演示如何快速新增一个内容源。由于小红书官方开放接口有限这里使用模拟数据来演示适配器写法实际使用时可以替换为合规的数据获取方式。# 文件路径sources/xiaohongshu.py from sources.base import BaseSource class XiaohongshuSource(BaseSource): 小红书内容源适配器演示用模拟实现 def fetch(self, config: dict) - list[dict]: xhs_config config.get(xiaohongshu, {}) if not xhs_config.get(enabled): return [] # 实际项目中这里应替换为合规的数据获取逻辑 # 例如官方开放平台接口、自己账号授权后的数据导出等 mock_items [ { platform: xiaohongshu, content_id: demo-001, title: 分享一个提升开发效率的AI工具, summary: 最近发现一款AI工具配合 Agent 使用非常顺手..., tags: [AI工具, 效率], url: https://www.xiaohongshu.com/explore/demo-001, published_at: 2025-01-10 10:00:00, }, { platform: xiaohongshu, content_id: demo-002, title: 周末学习打卡LLM 入门笔记, summary: 整理了最近学习大语言模型的知识点..., tags: [LLM, 学习], url: https://www.xiaohongshu.com/explore/demo-002, published_at: 2025-01-11 20:30:00, }, ] return mock_items然后在入口注册# main.py 中新增 from sources.xiaohongshu import XiaohongshuSource agent.register_source(BilibiliSource()) agent.register_source(XiaohongshuSource())通过这种扩展方式你可以轻松叠加 YouTube、Twitter、即刻、少数派、知乎等任意内容来源。需要做的只有两件事写一个 fetch 方法统一数据格式。5.2 统一数据格式的重要性可能有读者会问为什么非要统一成固定的 dict 结构因为后续的“去重”、“LLM 评分”、“排序”都是无差别处理每一条内容的。如果每个平台返回的数据结构都不一样核心逻辑里会堆满各种if platform bilibili分支代码很快会变得腐烂。统一格式的本质是定义了一个“内部协议”让所有外部数据源在进入核心处理链路之前完成格式转换。这是整个 Agent 架构中最重要的设计决策之一。6. 常见问题与排查思路实际开发中最容易踩坑的点不在 Agent 核心逻辑而是在数据获取和模型调用这两层。下面列出一份排查清单。问题现象常见原因解决思路获取内容为空数据源接口变更或未授权先用浏览器或 curl 验证接口是否可用检查返回状态码和数据结构LLM 返回 JSON 解析失败模型输出包含额外文字Prompt 中强制“只返回 JSON”代码中做字符串清理去掉代码块标记评分全部为 0API Key 错误或模型名称不对单独写一个测试脚本验证模型连接是否正常重复内容很多未启用去重或历史数据丢失检查 SQLite 表是否存在确认唯一键设置是否正确抓取频率过高被封请求间隔太短或缺少限速在适配器中增加限速逻辑控制单次请求间隔和单日调用量内存占用持续增长内容列表过大未做分页限制每个数据源限制单次拉取条数建议 10-20 条以内定时任务不触发时区或调度配置错误检查 APScheduler 时区设置确认interval_minutes是否读取成功这里重点展开两个高频问题。问题一LLM 评分结果不稳定。同一个内容跑两次评分可能得到完全不同的分数。原因是 LLM 本身具有概率性尤其在 temperature 较高时随机性更大。解决方法是将 temperature 调低到 0.2-0.3让输出更稳定。对同一条内容采样多次取平均值一般 2-3 次即可。在 Prompt 中给出明确评分标准和参考示例减少歧义。问题二数据平台接口经常变动。B 站、小红书等平台的页面结构和接口并不稳定可能几个月就调整一次。如果你的适配器突然拿不到数据优先检查接口返回的 JSON 是否还是原来的结构。是否新增了登录校验或风控。是否要求配置 Cookie 或 Token。生产使用中建议把数据获取做成“可降级”的。获取失败时使用上次缓存的数据继续跑保证信息流不会中断。7. 最佳实践与工程建议7.1 模型选型与成本控制如果每天只跑一次、每次 50 条内容LLM 的调用量其实很小。但如果你把调度频率提高到每小时成本会快速上升。一些成本控制建议先用便宜的模型跑全量过滤再用高质量模型对 Top 20 的内容做精细摘要。对同一内容源的内容做关键词前置过滤明显不相关的直接跳过 LLM 评分。内容入库后如果已经评分过不再重复调用 LLM。7.2 隐私与安全边界个人信息流系统会采集你的观看历史、兴趣标签和关注列表这些数据非常敏感。需要注意数据尽量本地存储不要上传到第三方服务器。如果使用云端 LLM API避免把包含个人隐私的完整内容发送给模型只发送摘要和标题片段。不要把 API Key 提交到公开仓库使用环境变量或本地密钥文件管理。定期清理 SQLite 中的历史数据避免长期积累。7.3 平台合规建议不同平台对数据抓取的政策差异很大。在搭建系统时请务必注意优先使用平台官方提供的 API、RSS、开放接口。控制请求频率避免对平台服务器造成压力。仅用于个人学习和研究不要将抓取内容进行二次分发。如果涉及商业用途必须获得平台书面授权。这篇文章的示例使用模拟数据演示就是为了避开平台接口限制问题。你在实际接入时需要自行确认目标平台的最新政策和技术方案。7.4 可维护性设计Agent 项目虽然小但长期运行后维护成本会逐渐凸显。几个建议每个数据源独立成文件保持单一职责。配置与代码分离兴趣标签、模型参数统一放 YAML。日志要完整至少记录每次抓取的源、条数、耗时、评分结果。用dataclass定义统一的内容结构代替纯字典提升代码可读性。7.5 失败降级与重试机制Agent 是无人值守运行的程序网络波动、接口异常、模型服务不可用都可能发生。核心逻辑需要做好降级# 伪代码降级逻辑示意 def run_once(self): results [] for source in self.sources: try: items source.fetch(self.config) except Exception as e: print(f数据源 {source} 获取失败使用缓存数据。原因{e}) items load_from_cache(source) for item in items: scored self.llm_scorer.score_item(item) if scored[score] self.min_score: results.append({**item, **scored}) return sorted(results, keylambda x: x[score], reverseTrue)8. 总结与动手方向到这里一个基于 Agent 的个性化信息流系统已经完整落地了。回顾一下核心内容Agent 能够主动获取内容、理解内容、判断相关性替代传统 RSS 的被动订阅模式。多平台适配器是接入 B 站、小红书、YouTube、Twitter 等数据源的核心设计。LLM 评分不是越复杂越好关键是 Prompt 设计和成本控制。去重、缓存、降级是长期运行稳定性的保障。下一步你可以从这几个方向继续深入把推荐报告通过企业微信、钉钉、Telegram Bot 推送每天 8 点自动发送。增加“反馈机制”看到不感兴趣的内容可以标记负反馈Agent 下次评分时参考。接入本地大模型例如 Ollama实现完全离线运行避免数据外流。用 Agent 自动抓取完整正文内容生成摘要式简报形成你的“每日早报”。如果你也受够了被平台推荐算法支配的感觉可以动手把这个系统搭起来。第一次跑通时你大概率会遇到 JSON 解析失败、接口返回结构变化这类小坑这些都是 Agent 开发的必经之路。把排查过程记录下来本身就是一笔很好的技术积累。
