之前刷 B 站、小红书、YouTube、Twitter 的时候我总有一种感觉首页推荐越来越“懂我”但也越来越让我害怕。你以为是自己主动选择的内容其实背后是平台算法在猜你想看什么。猜得准一点就多留你一会儿猜得偏一点你就刷到根本不想看的内容。最近 GitHub 上这类“把推荐流换成你自己的”Agent 项目热度很高本文涉及的方向已经有开源项目累计拿到 1500 Stars也是 B 站 AI 创造公开赛和各类 Agent 黑客松里常见的方向。这篇文章会完整拆解一个可运行的“个人推荐流 Agent”实现方案从内容采集、大模型内容理解、用户画像维护到最终生成一份属于你自己的推荐清单。你可以 10 分钟内在本地跑通不需要付费 API不需要 GPU 也能玩。核心代码全部给出照着复制就能运行。无论你是 AI Agent 初学者、前端/后端开发者还是对推荐算法感兴趣的产品同学这篇文章都能帮你理解 Agent 应用的完整架构并给你一个可以直接扩展的项目骨架。1. 背景为什么我们需要“自己的推荐流”1.1 推荐算法为什么是黑盒各大内容平台的推荐系统本质上是一个多目标优化问题平台要同时优化你的停留时长、点击率、互动率、广告收入等指标。所以它推荐的未必是“对你最有价值”的内容而是“最可能让你继续刷”的内容。这就是为什么你会遇到下面这些情况不小心点了一次“明星八卦”之后首页连续三天全是类似内容。你明明只对技术教程感兴趣但推荐流里混入了大量带货视频。内容越推越窄陷入所谓的“信息茧房”很难发现新领域的好内容。平台算法的真实目标和你作为用户的目标并不完全一致。你想“高效获取有价值的信息”平台想“让你停留更久”。1.2 Agent 能做什么Agent智能体是一个能够自主完成“感知→决策→行动”闭环的程序。当它被应用到推荐场景时思路就变了传统的推荐流平台算法决定你看到什么。Agent 推荐流你决定规则Agent 帮你把各平台的内容抓回来用大模型理解内容再按照你自己的兴趣画像和过滤规则重新排序最后生成一份只属于你的推荐清单。这意味着你可以把 B 站、小红书、YouTube、Twitter 的内容源统一收拢到一个管道里。用大模型判断每篇内容是否值得看而不是只看标题关键词。维护自己的“兴趣画像”今天想重点关注什么就改什么。生成 Markdown 格式的阅读周报、每日推荐清单或者直接推送到自己的 Bot。1.3 这类项目为什么值得关注GitHub 上推荐流 Agent 项目热度上升主要有几个原因。第一大模型让“内容理解”变得足够便宜。以前要做语义分析、主题分类得训练专门的模型现在直接调用大模型就能完成。第二Agent 生态成熟了。从 LangChain、LangGraph到各种 Harness 框架开发者可以用很短的代码组合出完整流程。第三个人数据主权意识增强。很多人不希望自己的浏览行为被无限分析更希望自己掌控信息获取方式。这类项目也是各类 AI 创造比赛的热门选题比如 B 站 AI 创造公开赛里很多参赛作品都是基于 Agent 的个人助手、信息流管理工具。1.4 先分清 Agent、Harness 与框架在阅读开源项目时你经常看到 Agent、Harness、Framework 这几个词不少人会混淆。简单理解概念作用类比Agent完成任务的智能体包含模型、提示词、工具调用逻辑员工Harness承载 Agent 运行的运行时环境处理输入输出、重试、日志、工具注册工位 办公流程Framework帮助开发者构建 Agent 的开发框架公司制度 工具集我们这篇文章为了让你看清原理会从零写一个非常轻量的 Agent不依赖重量级框架。理解了底层逻辑之后再切换到 LangGraph、Dify、Coze 这类生产级框架会容易得多。2. 环境准备与项目初始化2.1 运行环境要求这个项目非常轻量普通笔记本即可运行。推荐环境如下操作系统Windows 10/11、macOS、Linux 均可Python3.10 及以上大模型可选两种方案本地模型安装 Ollama拉取一个 7B 左右的模型如qwen2.5:7b完全离线免费云端 API使用 OpenAI 兼容接口比如 DeepSeek、通义千问、智谱等注意额度这里说一句实话如果你想完全 0 成本就用本地 Ollama 方案如果追求更好的理解效果可以使用云端 API大部分厂商注册会送免费额度小规模使用基本够。2.2 项目结构我们创建一个名为personal-recommender-agent的项目目录结构如下personal-recommender-agent/ ├── config.yaml # 配置文件模型地址、内容源、兴趣画像 ├── requirements.txt # Python 依赖 ├── main.py # 入口调度程序 ├── collector.py # 内容采集模块 ├── analyzer.py # 内容理解模块 ├── profile.py # 用户画像模块 └── recommender.py # 推荐生成模块2.3 安装依赖新建文件requirements.txt内容如下httpx0.27 feedparser6.0 pyyaml6.0 openai1.0然后在项目目录执行pip install -r requirements.txt如果你使用的是国内网络pip 下载慢时可以临时切换为清华镜像源pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple如果你想把 GitHub 上的开源项目克隆到本地网络慢时可以使用--depth 1做浅克隆只拉取最新一次提交体积会小很多git clone --depth 1 https://github.com/xxx/xxx.git3. 核心原理拆解这个 Agent 的完整工作闭环可以拆成五步采集 → 理解 → 画像 → 排序 → 输出。内容源(Feed) - 采集器 - 标准化条目 标准化条目 - 大模型分析 - 标签评分 标签评分 - 用户画像过滤 - 候选列表 候选列表 - 多样性与时间衰减 - 最终推荐 最终推荐 - Markdown/HTML - 推送给用户下面分别拆解每一步。3.1 内容采集层统一 Feed 入口不同平台数据格式不同因此我们要在最外层做一层“标准化”。采集器负责把各类内容源转换成统一的ContentItem数据结构包含标题、链接、摘要、发布时间、来源平台等字段。数据源选择方面要遵守平台规则。推荐使用以下合规方式公开 RSSYouTube 频道、博客、播客都支持 RSS。官方开放 API部分平台提供搜索 API比如 YouTube Data API。自建内容源比如你自己收藏的链接、邮件订阅、GitHub Star 项目等。不建议去破解平台的内部接口也不建议频繁爬取页面否则容易触发风控和法律风险。这也是成熟开源项目通常会优先使用 RSS 的原因。3.2 内容理解层大模型判断价值拿到一篇内容后我们要判断这篇文章值不值得推荐这个判断不能只看标题关键词因为标题党太多。更好的做法是把标题、摘要、正文片段拼接成一段 Prompt交给大模型让它返回内容主题分类关键词标签相关度打分0-100是否值得阅读为了节省 token可以把摘要截断到 300-500 字。对于本地模型这个量级的速度是可以接受的。3.3 用户画像层你的偏好就是规则用户画像可以是一份简单的配置包含兴趣关键词列表排除关键词列表各兴趣领域的权重这个画像不由平台算法“学习”得到而是你主动维护的。想多看点 Agent 相关内容就把AI Agent的权重调高不想看某类内容就加入排除列表。3.4 排序与多样化输出排序时我们把大模型相关度评分作为基础分再叠加两个因子时间衰减内容发布时间越新权重越高。来源多样性防止某个平台内容刷屏限制同一来源的推荐数量。一句话总结推荐 相关度评分 × 时间衰减 × 来源多样性约束。4. 完整实战案例搭建一个个人推荐 Agent下面进入实操环节。我们一步一步写出这个项目的所有代码。4.1 配置文件 config.yaml创建config.yaml内容如下agent: name: personal-recommender interval_minutes: 720 llm: provider: openai-compatible base_url: http://localhost:11434/v1 model: qwen2.5:7b api_key: EMPTY temperature: 0.3 feeds: - name: Google Developers url: https://www.youtube.com/feeds/videos.xml?channel_idUC_x5XG1OV2P6uZZ5FSM9Ttw - name: GitHub Blog url: https://github.blog/feed/ - name: 示例博客 url: https://www.ruanyifeng.com/blog/atom.xml profile: interests: ai_agent: 1.0 big_model: 0.9 open_source: 0.8 python: 0.7 dislike_topics: - 娱乐八卦 - 无意义挑战 max_items_per_source: 8 top_n: 10注意这里的base_url指向本地 Ollama 服务。如果你使用云端 API把base_url换成对应服务商地址model换成对应模型名api_key填你的密钥即可。4.2 内容采集模块 collector.py创建collector.py负责抓取 RSS 并转换为统一数据结构# 文件路径collector.py import time import feedparser from dataclasses import dataclass, field from typing import List dataclass class ContentItem: title: str link: str summary: str source: str published: str tags: List[str] field(default_factorylist) score: float 0.0 class ContentCollector: def __init__(self, feeds: list): self.feeds feeds def fetch(self) - List[ContentItem]: items: List[ContentItem] [] for feed in self.feeds: try: parsed feedparser.parse(feed[url]) for entry in parsed.entries[:10]: summary entry.get(summary, )[:500] items.append( ContentItem( titleentry.get(title, ), linkentry.get(link, ), summarysummary, sourcefeed[name], publishedentry.get(published, ), ) ) print(f[collector] {feed[name]} 抓取完成共 {len(parsed.entries)} 条) except Exception as e: print(f[collector] {feed[name]} 抓取失败{e}) time.sleep(1) return items这里的feedparser.parse()会读取 RSS 内容entry是每篇文章或视频的条目。4.3 内容理解模块 analyzer.py创建analyzer.py使用 OpenAI 兼容接口调用大模型。# 文件路径analyzer.py import json from openai import OpenAI from collector import ContentItem class ContentAnalyzer: def __init__(self, llm_config: dict): self.client OpenAI( base_urlllm_config[base_url], api_keyllm_config.get(api_key, EMPTY), ) self.model llm_config[model] self.temperature llm_config.get(temperature, 0.3) def analyze(self, item: ContentItem): prompt f 你是一个内容推荐助手。请分析以下内容判断它是否值得推荐。 标题{item.title} 摘要{item.summary} 来源{item.source} 请严格按照下面的 JSON 格式返回不要有多余内容 {{ topic: 内容主题, keywords: [关键词1, 关键词2], score: 0-100之间的整数, reason: 一句话推荐理由 }} try: resp self.client.chat.completions.create( modelself.model, messages[ {role: system, content: 你是一个严谨的内容分析引擎。}, {role: user, content: prompt}, ], temperatureself.temperature, response_format{type: json_object}, ) result json.loads(resp.choices[0].message.content) item.tags result.get(keywords, []) item.score float(result.get(score, 0)) return result except Exception as e: print(f[analyzer] 分析失败{e}) # 降级策略当大模型不可用时直接用关键词模糊匹配打分 return self._fallback_analyze(item) def _fallback_analyze(self, item: ContentItem): # 本地离线兜底逻辑 keywords [AI, Agent, 大模型, Python, 开源, GPT, 编程, 效率] hit sum(1 for kw in keywords if kw.lower() in item.title.lower()) item.score min(100, hit * 20) topic 未分类 if hit 0: topic 技术内容 return { topic: topic, keywords: keywords[:3], score: item.score, reason: 离线关键词匹配, }这里有一个重要细节response_format{type: json_object}是 OpenAI 兼容接口的常见用法但部分本地模型可能不支持。如果你使用的模型不支持可以去掉这个参数改成正则提取 JSON。4.4 用户画像与推荐模块 profile.py创建profile.py# 文件路径profile.py from typing import List class UserProfile: def __init__(self, config: dict): self.interests config.get(interests, {}) self.dislike_topics config.get(dislike_topics, []) self.max_items_per_source config.get(max_items_per_source, 5) self.top_n config.get(top_n, 10) def filter(self, items: List): filtered [] source_count {} for item in items: # 排除不喜欢的内容 if any(dislike in item.title or dislike in (item.tags or []) for dislike in self.dislike_topics): continue # 来源数量限制 src item.source source_count[src] source_count.get(src, 0) 1 if source_count[src] self.max_items_per_source: continue filtered.append(item) return filtered def sort_items(self, items: List): # 基础分来自大模型时间权重简化为固定排序 return sorted(items, keylambda x: x.score, reverseTrue)[: self.top_n]4.5 推荐输出模块 recommender.py创建recommender.py生成 Markdown 推荐报告# 文件路径recommender.py from datetime import datetime from typing import List from collector import ContentItem class Recommender: def __init__(self, output_file: str recommend.md): self.output_file output_file def generate_report(self, items: List[ContentItem]): lines [ # 今日推荐清单, , f 生成时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}, , ] for idx, item in enumerate(items, start1): lines.append(f## {idx}. {item.title}) lines.append() lines.append(f- 来源{item.source}) lines.append(f- 评分{item.score}) if item.tags: lines.append(f- 标签{, .join(item.tags)}) lines.append(f- 链接{item.link}) if item.summary: lines.append() lines.append(f- 摘要{item.summary[:100]}...) lines.append() report \n.join(lines) with open(self.output_file, w, encodingutf-8) as f: f.write(report) print(f[recommender] 推荐报告已生成{self.output_file}) return report4.6 入口调度 main.py最后创建main.py# 文件路径main.py import yaml from collector import ContentCollector from analyzer import ContentAnalyzer from profile import UserProfile from recommender import Recommender def load_config(pathconfig.yaml): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main(): config load_config() print([main] 开始内容采集...) collector ContentCollector(config[feeds]) items collector.fetch() if not items: print([main] 没有抓到任何内容请检查 feeds 配置或网络连接。) return print(f[main] 共采集到 {len(items)} 条内容开始分析...) analyzer ContentAnalyzer(config[llm]) for item in items: analyzer.analyze(item) print([main] 开始用户画像过滤...) profile UserProfile(config[profile]) filtered profile.filter(items) sorted_items profile.sort_items(filtered) print([main] 生成推荐报告...) recommender Recommender() recommender.generate_report(sorted_items) print([main] 全部完成。) for item in sorted_items: print(f - {item.title} ({item.score})) if __name__ __main__: main()4.7 运行与验证确保本地 Ollama 服务已启动如果使用本地模型方案ollama serve然后拉取模型ollama pull qwen2.5:7b最后运行主程序python main.py预期输出大致如下[collector] Google Developers 抓取完成共 15 条 [collector] GitHub Blog 抓取完成共 10 条 [collector] 示例博客 抓取完成共 20 条 [main] 共采集到 45 条内容开始分析... [analyzer] 分析失败... [analyzer] 降级为离线关键词匹配 [main] 开始用户画像过滤... [main] 生成推荐报告... [main] 全部完成。 - Introducing Agent Harness: Production-Ready AI Agents (85.0) - Building a Personal Recommendation Agent (82.0)如果没有启动 Ollama程序不会崩溃会自动降级到离线关键词匹配模式。这就是“0 成本可玩”的关键设计。5. 常见问题与排查思路在跑这个项目时你大概率会遇到下面这些问题。问题现象常见原因解决思路采集内容为空Feed URL 不正确或网络不通先用浏览器打开 URL确认是 XML 格式大模型调用报错Ollama 未启动或模型名错误执行ollama list查看已安装模型JSON 解析失败本地模型不支持response_format去掉该参数改用正则提取 JSON推荐结果不相关兴趣画像关键词太少在profile.interests中增加关键词同一来源内容刷屏未设置来源数量限制调低max_items_per_sourcepip 下载依赖慢网络原因使用国内镜像源安装5.1 大模型返回的不是 JSON这是最常见的问题。本地小模型的指令遵循能力相对较弱不保证严格输出 JSON。解决方案有两种方案一在 Prompt 中给出示例输出格式并且标注“只输出 JSON”。方案二用正则表达式从返回文本中提取 JSON 片段import re import json text resp.choices[0].message.content match re.search(r\{.*\}, text, re.S) if match: result json.loads(match.group())5.2 本地模型太慢怎么办本地 7B 模型在 CPU 上确实比较慢。优化思路有切换到更小的模型比如qwen2.5:3b或llama3.2:3b。截断摘要把summary[:500]改为summary[:200]。启用批量处理把多条内容合并成一个 Prompt 交给模型。批量处理示例思路batch_prompt 请逐条分析以下内容返回 JSON 数组\n for i, item in enumerate(items[:5]): batch_prompt f{i1}. {item.title} - {item.summary[:200]}\n5.3 RSS 中有些字段为空不同平台的 RSS 字段名称可能不一样比如摘要可能是summary也可能是description。建议代码里做兼容summary entry.get(summary) or entry.get(description) or 5.4 GitHub 项目克隆或下载慢在国内网络环境下克隆开源项目时可以使用 GitHub 镜像加速域名或者使用浅克隆只拉取最新代码。注意不要使用来路不明的“加速脚本”尽量选择知名度高的镜像站点。6. 最佳实践与工程建议6.1 数据获取合规边界这是最重要的建议。无论做哪个平台的推荐 Agent都要遵守以下原则优先使用官方的 RSS、API 或导出功能。不绕过登录验证、不破解签名参数、不高频请求。尊重平台的 robots 协议和用户协议。只抓取公开数据不采集用户隐私信息。如果某个平台没有任何公开数据源建议放弃该平台而不是想办法绕过限制。技术能力不应该用来越过法律和规则红线。6.2 大模型成本优化如果你使用云端 API成本是必须考虑的。推荐做法使用缓存同一条内容的分析结果保存到本地避免重复调用。样本截断摘要控制在 300 字以内标题和链接必传。使用 Batch API部分厂商提供异步批量接口价格更低。本地优先敏感内容和个人阅读偏好尽量使用本地小模型处理。6.3 用户画像与隐私保护个人推荐 Agent 表面上是个工具本质上是一个持续收集你阅读偏好的系统。因此不要把 API Key 硬编码到代码里使用环境变量。画像数据保存在本地数据库或本地文件不要上传云端。如果使用云端大模型不要在 Prompt 中上传过于私密的信息。6.4 Agent 工程化上线建议本教程为了讲原理直接用while和requests串起流程。生产环境建议参考以下升级路径任务调度使用 APScheduler 或系统 cron。消息通知推荐结果推送至钉钉、飞书、微信机器人或邮件。可观测性接入 LangSmith、MLflow 或自定义日志记录每次 Agent 运行的输入输出。流程编排从手写循环迁移到 LangGraph利用节点、条件和人工审核能力。向量存储当内容量变大时把历史内容向量化存入本地向量库避免重复推荐。6.5 推荐算法的基本均衡纯按大模型打分排序容易出现“某平台内容霸榜”或“推荐结果太单一”的问题。可以在排序函数中加入一个简单的多样性惩罚def diversified_sort(items, source_weight0.3): seen {} scored [] for item in items: seen[item.source] seen.get(item.source, 0) 1 penalty source_weight * (seen[item.source] - 1) final_score item.score - penalty scored.append((final_score, item)) scored.sort(keylambda x: x[0], reverseTrue) return [item for _, item in scored]这个思路类似推荐系统里的 MMR最大边际相关能保证推荐结果既有相关性又有一定覆盖面。7. 下一步学习方向到这里你已经完成了一个完整的“个人推荐流 Agent”最小实现。你掌握了Agent 的感知-决策-行动闭环。多来源内容采集与统一数据结构设计。大模型内容理解与 Prompt 调用技巧。用户画像过滤、排序与多样性控制。项目模块化拆分和离线降级策略。如果这篇文章对你有帮助可以收藏备用。接下来想继续深入可以按以下路线学习用 LangGraph 重构整个流程加入条件分支和人工审核节点。接入向量数据库做基于 Embedding 的相似内容去重与推荐。把推荐结果做成一个定时推送的 Bot或一个本地 Web 阅读面板。研究推荐系统经典算法协同过滤、矩阵分解、MMR 多样性排序。关注 B 站 AI 创造公开赛优秀作品观察成熟 Agent 项目的工程组织方式。推荐算法不该是平台单方面给你的“黑盒”。用 Agent 把推荐权拿回自己手里这件事技术上并不复杂复杂的是你愿不愿意花 10 分钟配置好第一份兴趣画像。动手试试你会打开一个全新的信息获取方式。