基于AI 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.10+
• LLM 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\activate

2.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，写入依赖：

httpx>=0.27.0 feedparser>=6.0.11 apscheduler>=3.10.4 pyyaml>=6.0.1 openai>=1.30.0

安装依赖：

pip install -r requirements.txt

4.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_url=config["base_url"], api_key=config["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( model=self.model, messages=[ {"role": "system", "content": "你是一个精准的内容推荐评分器。"}, {"role": "user", "content": prompt}, ], temperature=self.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": f"LLM 评分失败：{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(key=lambda x: x["score"], reverse=True) 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 = f"https://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, params=params, headers=headers, timeout=10) 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": f"https://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", encoding="utf-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, start=1): print(f"{idx}. {item['title']} (评分：{item['score']})") print(f" 理由：{item['reason']}") print(f" 链接：{item['url']}\n") # 可进一步输出为 Markdown 文件 with open("daily_report.md", "w", encoding="utf-8") as f: f.write("# 今日个性化推荐\n\n") for idx, item in enumerate(results, start=1): 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", minutes=60) 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”；代码中做字符串清理，去掉代码块标记
评分全部为 0	API 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, key=lambda x: x["score"], reverse=True)

8. 总结与动手方向

到这里，一个基于 Agent 的个性化信息流系统已经完整落地了。回顾一下核心内容：

• Agent 能够主动获取内容、理解内容、判断相关性，替代传统 RSS 的被动订阅模式。
• 多平台适配器是接入 B 站、小红书、YouTube、Twitter 等数据源的核心设计。
• LLM 评分不是越复杂越好，关键是 Prompt 设计和成本控制。
• 去重、缓存、降级是长期运行稳定性的保障。

下一步你可以从这几个方向继续深入：

• 把推荐报告通过企业微信、钉钉、Telegram Bot 推送，每天 8 点自动发送。
• 增加“反馈机制”，看到不感兴趣的内容可以标记负反馈，Agent 下次评分时参考。
• 接入本地大模型（例如 Ollama），实现完全离线运行，避免数据外流。
• 用 Agent 自动抓取完整正文内容，生成摘要式简报，形成你的“每日早报”。

如果你也受够了被平台推荐算法支配的感觉，可以动手把这个系统搭起来。第一次跑通时，你大概率会遇到 JSON 解析失败、接口返回结构变化这类小坑，这些都是 Agent 开发的必经之路。把排查过程记录下来，本身就是一笔很好的技术积累。