基于OpenClaw/QClaw与LLM的Reddit智能摘要系统构建实战

1. 项目概述与核心价值如果你和我一样每天泡在Reddit和各种技术社区里试图从海量的帖子、评论和新闻中淘出真正有价值的信息那你一定体会过那种“信息过载”的无力感。首页永远刷不完热帖里夹杂着大量水贴和重复讨论而真正能启发思路、解决实际问题的“干货”往往藏在某个不起眼的子版块或长篇讨论的回复里。手动筛选效率太低。用现成的RSS信息太杂且缺乏聚合与提炼。这正是我启动“Reddit Digest”这个个人项目的初衷打造一个自动化、智能化的信息消化工具它不只是简单的爬虫更是一个能理解内容、提炼重点、并按照我的兴趣偏好生成个性化摘要的“工作伙伴”。这个项目的核心我称之为“OpenClaw/QClaw”技术栈驱动下的Reddit信息处理流水线。它围绕几个关键词展开openclaw和qclaw代表了其底层的数据抓取与处理引擎reddit-scraper指明了其主要的数据来源而workbuddy则完美诠释了它的最终形态——一个能7x24小时为你工作的智能助手。它解决的问题非常具体为你节省每天至少一小时的“信息垃圾”筛选时间把浓缩后的、高相关性的技术动态、社区讨论精华和行业新闻准时推送到你面前。无论你是开发者、产品经理、技术爱好者还是任何需要从Reddit及类似社区获取前沿信息的从业者这个项目都能让你从被动的信息接收者转变为主动的信息管理者。2. 技术架构与核心组件解析一个稳定、高效且可维护的自动化摘要系统其背后的架构设计至关重要。我的“Reddit Digest”并非一个单一脚本而是一个由多个松耦合组件构成的微服务系统。这样设计的好处是显而易见的每个组件可以独立开发、测试和部署当某个环节比如Reddit API策略变更需要调整时不会牵一发而动全身。2.1 数据抓取层OpenClaw/QClaw引擎这是整个系统的“手”和“眼睛”。OpenClaw和QClaw是我对两类抓取策略的抽象命名它们共同构成了灵活的数据获取层。OpenClaw开放式抓取针对公开的、无需复杂交互的页面。对于Reddit这意味着使用其官方提供的JSON API端点。例如获取r/programming子版块的热门帖子可以直接请求https://www.reddit.com/r/programming/hot/.json。这种方式稳定、快速且符合Reddit的使用规范。我会为OpenClaw配置一系列这样的API模板并附带请求头如设置一个合理的User-Agent以模拟浏览器行为避免被简单的反爬机制拦截。QClaw队列式精准抓取这是针对需要深入挖掘内容的场景。例如当一个帖子被OpenClaw识别为高价值通过初始的分数、评论数等指标后QClaw会接手它的任务不仅仅是获取帖子正文还要深入抓取该帖子下的“最佳评论”或“深度讨论线程”。这里的关键在于“队列”和“精准”。系统会维护一个优先级队列QClaw从队列中取出任务针对单个帖子URL进行深度请求解析出完整的评论树。这个过程需要更复杂的HTML解析或对Reddit更多API的调用并且必须严格遵守访问频率限制因此“队列”机制保证了任务的有序和可控。注意API频率限制与礼貌爬取Reddit API对未认证请求和OAuth认证请求都有严格的频率限制。我的策略是对于OpenClaw的列表抓取使用较长的间隔如每分钟一次对于QClaw的详情抓取则在单个任务完成后主动休眠1-2秒。此外务必在User-Agent中清晰标识你的应用名称和联系方式这是一个负责任的开发者的基本素养。2.2 数据处理与存储层原始数据抓取回来后是未经加工的“矿石”。数据处理层的任务就是将其冶炼成“钢材”。内容清洗与标准化Reddit返回的JSON数据虽然结构清晰但包含了许多前端展示相关的元信息。我们需要提取核心字段帖子标题title、作者author、发布时间created_utc、分数score、评论数num_comments、正文文本selftext、链接url以及最重要的——所属子版块subreddit。对于评论则需要提取评论正文、作者、分数以及嵌套的回复结构。我会使用像pandas或直接操作Python字典/列表的方式将这些数据清洗并转换成一个结构化的字典或自定义的Post/Comment对象。向量化存储与索引为了让系统能“理解”内容并进行相似性检索文本的向量化是关键一步。我选用像sentence-transformers这样的库将帖子标题和正文转换为高维向量例如384维的向量。这些向量会存入专门的向量数据库例如ChromaDB或Qdrant。同时帖子的元数据标题、链接、时间等会存入一个关系型数据库如SQLite用于轻量级部署或PostgreSQL用于生产环境或文档数据库如MongoDB中。两者通过唯一的帖子ID进行关联。这样当我想查找所有关于“Python async”的讨论时既可以通过向量数据库进行语义搜索也可以通过关系数据库按子版块、时间进行过滤。2.3 摘要生成与推送层这是体现项目“智能”和“价值”的核心也是“WorkBuddy”得名的原因。摘要生成策略我尝试过多种方法最终形成一个混合策略。提取式摘要对于评论区的精华直接选取点赞数最高的前3-5条评论。这种方法简单有效能直接反映社区的共识。抽象式摘要对于较长的帖子正文或技术文章分享使用大语言模型LLMAPI如OpenAI GPT、Anthropic Claude或开源的Llama API进行总结。我会设计一个精心构造的提示词Prompt“请用简洁的段落总结以下Reddit帖子的核心观点和技术要点并指出其中提到的关键工具或库。保持中立不要添加个人评价。” 然后将帖子正文和热门评论的文本一起喂给模型。混合摘要将上述两者结合。先呈现LLM生成的段落式概要再附上“社区高亮评论”的引用块。这种格式信息密度最高既有提炼又有原汁原味的社区反馈。个性化过滤与推送不是所有r/programming的热帖我都关心。我建立了一个“兴趣配置文件”里面定义了关键词如“Kubernetes”“Rust”“PostgreSQL”、我关注的具体子版块列表、以及需要屏蔽的关键词如“求职”“meme”。在生成每日摘要前系统会用这个配置文件对抓取到的帖子进行过滤和排序。推送渠道我选择了最通用的方式电子邮件和Telegram Bot。电子邮件适合长篇摘要而Telegram Bot则能提供即时、交互式的提醒。推送内容会经过精美的Markdown格式化确保在各种客户端上都有良好的阅读体验。3. 系统搭建与核心环节实现理论说再多不如一行代码。下面我将拆解搭建这个系统的关键步骤和代码实现。我的技术栈选择是Python作为主力语言因为它拥有极其丰富的库来支持爬虫、数据处理和AI应用。3.1 环境准备与依赖安装首先创建一个干净的Python虚拟环境是良好实践的开始。# 创建项目目录并进入 mkdir reddit-digest cd reddit-digest # 创建虚拟环境这里使用venv python3 -m venv venv # 激活虚拟环境 # 在Linux/macOS上 source venv/bin/activate # 在Windows上 venv\Scripts\activate接下来初始化项目并安装核心依赖。我使用requirements.txt来管理依赖。# requirements.txt # 网络请求与爬虫 requests2.28.0 aiohttp3.8.0 # 用于异步抓取提升效率 beautifulsoup44.11.0 # 备用用于解析非JSON页面 # 数据处理 pandas1.5.0 numpy1.24.0 # 向量化与存储 sentence-transformers2.2.0 chromadb0.4.0 # 轻量级向量数据库 # 数据库以SQLite和MongoDB为例 pymongo4.0.0 # 如果选用MongoDB # 关系型数据库通常有标准库或如sqlalchemy # 调度与任务队列 apscheduler3.10.0 # 高级Python调度器 # 或者使用Celery Redis用于更复杂的分布式队列 # 推送通知 python-telegram-bot20.0 # Telegram Bot yagmail0.15.0 # 发送邮件的简易库 # 大语言模型接口以OpenAI为例 openai1.0.0使用pip安装pip install -r requirements.txt。3.2 OpenClaw/QClaw抓取器实现让我们实现一个基础但健壮的抓取模块。我将其设计为一个类便于管理配置和状态。# scraper/core.py import requests import time import logging from typing import Dict, List, Optional from dataclasses import dataclass import json logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) dataclass class RedditPost: Reddit帖子数据类 id: str title: str author: str subreddit: str score: int num_comments: int url: str selftext: str created_utc: float permalink: str class RedditScraper: Reddit抓取器实现OpenClaw策略 def __init__(self, user_agent: str, rate_limit_delay: float 1.0): self.session requests.Session() self.session.headers.update({User-Agent: user_agent}) self.rate_limit_delay rate_limit_delay self.base_url https://www.reddit.com def fetch_hot_posts(self, subreddit: str, limit: int 25) - List[RedditPost]: 获取指定子版块的热门帖子OpenClaw url f{self.base_url}/r/{subreddit}/hot/.json?limit{limit} logger.info(fFetching hot posts from r/{subreddit}) try: response self.session.get(url, timeout10) response.raise_for_status() # 检查HTTP错误 data response.json() posts [] for child in data[data][children]: post_data child[data] post RedditPost( idpost_data[id], titlepost_data[title], authorpost_data[author], subredditpost_data[subreddit], scorepost_data[score], num_commentspost_data[num_comments], urlpost_data[url], selftextpost_data[selftext], created_utcpost_data[created_utc], permalinkf{self.base_url}{post_data[permalink]} ) posts.append(post) time.sleep(self.rate_limit_delay) # 遵守礼貌爬取间隔 return posts except requests.exceptions.RequestException as e: logger.error(fFailed to fetch from r/{subreddit}: {e}) return [] except (KeyError, json.JSONDecodeError) as e: logger.error(fFailed to parse response from r/{subreddit}: {e}) return [] def fetch_post_comments(self, post_permalink: str, depth: int 1) - List[Dict]: 获取指定帖子的评论QClaw策略的简化版 这是一个更复杂的操作需要解析评论树。 此处返回一个评论列表的简化示例。 # 实际实现中需要处理Reddit评论API的复杂嵌套结构 # 这里仅示意流程 url f{post_permalink}.json logger.info(fFetching comments for post: {post_permalink}) try: response self.session.get(url, timeout10) response.raise_for_status() data response.json() # data[1] 通常是评论数据结构非常嵌套 # 需要递归遍历 data[1][data][children] 来提取评论 # 此处省略具体解析代码... simplified_comments self._simplify_comment_tree(data, depth) time.sleep(self.rate_limit_delay * 2) # 详情页抓取间隔更长 return simplified_comments except Exception as e: logger.error(fFailed to fetch comments: {e}) return [] def _simplify_comment_tree(self, raw_data: Dict, max_depth: int) - List[Dict]: 递归简化评论树提取核心内容。这是一个复杂但关键的函数。 # 伪代码遍历raw_data提取评论body, author, score, 以及回复(replies) # 当深度超过max_depth时停止。 comments [] # ... 具体实现逻辑 return comments # 使用示例 if __name__ __main__: USER_AGENT RedditDigestBot/1.0 (by /u/YourRedditUsername) Contact: youremail.com scraper RedditScraper(user_agentUSER_AGENT, rate_limit_delay1.5) # OpenClaw: 抓取列表 programming_posts scraper.fetch_hot_posts(programming, limit10) for post in programming_posts[:3]: # 打印前三个 print(f- {post.title} (Score: {post.score}, Comments: {post.num_comments})) # QClaw: 选择一个帖子抓取评论 if programming_posts: sample_post programming_posts[0] comments scraper.fetch_post_comments(sample_post.permalink, depth2) print(f\nFetched {len(comments)} top-level comments for post: {sample_post.title})这个RedditScraper类封装了基本的抓取逻辑。fetch_hot_posts是典型的OpenClaw操作批量获取帖子列表。fetch_post_comments则代表了QClaw针对单个资源进行深度挖掘。在实际项目中QClaw部分会更复杂需要处理评论的嵌套结构和分页。3.3 向量化存储与兴趣匹配抓取到的帖子需要被理解和检索。以下是使用sentence-transformers和chromadb进行向量化存储和搜索的示例。# processor/vector_store.py from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings import uuid from typing import List import logging logger logging.getLogger(__name__) class VectorStoreManager: 管理文本向量化与向量数据库存储 def __init__(self, model_name: str all-MiniLM-L6-v2, persist_dir: str ./chroma_db): # 加载嵌入模型。all-MiniLM-L6-v2是一个轻量且效果不错的通用模型 logger.info(fLoading embedding model: {model_name}) self.model SentenceTransformer(model_name) # 初始化Chroma客户端设置持久化目录 self.client chromadb.Client(Settings( chroma_db_implduckdbparquet, persist_directorypersist_dir )) # 获取或创建集合类似于数据库中的表 self.collection self.client.get_or_create_collection(namereddit_posts) def generate_embedding(self, text: str): 为单条文本生成向量 # 如果文本过长可以截断或分段处理。这里做简单处理。 if not text or text.strip() : text [Empty] return self.model.encode(text).tolist() def store_post(self, post_id: str, title: str, content: str, metadata: dict): 存储一个帖子到向量数据库 # 结合标题和正文作为被检索的文本 combined_text f{title}. {content[:500]} # 限制长度避免过长 embedding self.generate_embedding(combined_text) # 准备元数据 doc_metadata { post_id: post_id, subreddit: metadata.get(subreddit, ), score: str(metadata.get(score, 0)), url: metadata.get(url, ), **metadata # 包含其他传入的元数据 } # 存储到ChromaDB self.collection.add( embeddings[embedding], documents[combined_text], # 存储的原始文本可用于展示 metadatas[doc_metadata], ids[post_id] # 使用帖子ID作为唯一标识 ) logger.info(fStored post {post_id} in vector database.) def search_similar(self, query: str, n_results: int 5, where_filter: dict None) - List[dict]: 根据查询文本搜索相似的帖子 :param where_filter: 可选的过滤条件如 {subreddit: programming} query_embedding self.generate_embedding(query) # 执行查询 results self.collection.query( query_embeddings[query_embedding], n_resultsn_results, wherewhere_filter # 例如只搜索某个子版块 ) # 整理返回结果 returned_posts [] if results[ids]: for i in range(len(results[ids][0])): post_info { id: results[ids][0][i], document: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] # 余弦距离越小越相似 } returned_posts.append(post_info) return returned_posts # 使用示例 if __name__ __main__: vs_manager VectorStoreManager() # 模拟存储一些帖子 sample_posts [ { id: abc123, title: How to optimize Python code for data science, content: Using vectorized operations with NumPy can greatly speed up your calculations..., metadata: {subreddit: Python, score: 256, url: https://reddit.com/...} }, { id: def456, title: Introduction to async programming in Rust, content: Tokio is a popular runtime for writing asynchronous Rust applications..., metadata: {subreddit: rust, score: 189, url: https://reddit.com/...} } ] for post in sample_posts: vs_manager.store_post( post_idpost[id], titlepost[title], contentpost[content], metadatapost[metadata] ) # 进行语义搜索 query ways to make Python run faster similar vs_manager.search_similar(query, n_results3) print(fPosts similar to {query}:) for res in similar: print(f - {res[metadata][title]} (r/{res[metadata][subreddit]}, score: {res[metadata][score]}))这个VectorStoreManager类负责将文本转换为向量并存入ChromaDB。search_similar方法允许我们进行语义搜索where参数还能让我们结合元数据如子版块进行过滤这是实现个性化推荐的基础。3.4 摘要生成与推送集成最后我们将所有环节串联起来并加入LLM摘要和推送功能。# digest/generator.py import openai # 或其他LLM提供商 from scraper.core import RedditScraper, RedditPost from processor.vector_store import VectorStoreManager import yagmail from telegram import Bot from telegram.error import TelegramError from datetime import datetime import logging import json logger logging.getLogger(__name__) class DigestGenerator: 摘要生成与推送的核心类 def __init__(self, config_path: str): with open(config_path, r) as f: self.config json.load(f) # 初始化组件 self.scraper RedditScraper( user_agentself.config[scraper][user_agent], rate_limit_delayself.config[scraper][rate_limit_delay] ) self.vector_store VectorStoreManager( model_nameself.config[vector_store][model_name], persist_dirself.config[vector_store][persist_dir] ) # 初始化LLM客户端以OpenAI为例 openai.api_key self.config[llm][openai_api_key] self.llm_model self.config[llm][model] # 初始化推送客户端 self.email_sender yagmail.SMTP( userself.config[notification][email][sender], passwordself.config[notification][email][password] # 建议使用应用专用密码 ) self.telegram_bot Bot(tokenself.config[notification][telegram][bot_token]) self.telegram_chat_id self.config[notification][telegram][chat_id] # 加载兴趣配置 self.interests self.config[user_interests] def _filter_and_rank_posts(self, posts: List[RedditPost]) - List[RedditPost]: 根据兴趣配置文件过滤和排序帖子 filtered [] for post in posts: # 1. 子版块过滤 if post.subreddit.lower() not in [s.lower() for s in self.interests[subreddits]]: continue # 2. 关键词过滤兴趣词加分屏蔽词一票否决 title_lower post.title.lower() content_lower post.selftext.lower() combined_text f{title_lower} {content_lower} # 检查屏蔽词 if any(block_word in combined_text for block_word in self.interests.get(block_keywords, [])): continue # 计算兴趣匹配度简单版兴趣关键词出现次数 interest_score sum(keyword in combined_text for keyword in self.interests[keywords]) # 可以结合帖子分数、评论数进行综合排序 post.custom_score interest_score * 10 post.score * 0.1 post.num_comments * 0.05 filtered.append(post) # 按自定义分数降序排序 filtered.sort(keylambda x: x.custom_score, reverseTrue) return filtered[:self.interests.get(max_posts_per_digest, 10)] # 取前N个 def _generate_summary_with_llm(self, post: RedditPost, top_comments: List[Dict]) - str: 使用LLM生成帖子摘要 # 构建提示词 comments_text \n.join([f- {c[body][:200]}... for c in top_comments[:3]]) prompt f 请为以下Reddit帖子生成一个简洁、信息丰富的摘要。 帖子标题{post.title} 帖子正文{post.selftext[:1500]} # 限制长度 热门评论 {comments_text} 请用一段话不超过200字总结该帖子的核心讨论点、技术要点或主要观点。保持客观不要添加个人意见。如果提到了具体的工具、库或方法请明确指出。 try: response openai.chat.completions.create( modelself.llm_model, messages[ {role: system, content: 你是一个专注于技术社区内容总结的助手。}, {role: user, content: prompt} ], max_tokens300, temperature0.5 # 较低的温度使输出更稳定、聚焦 ) summary response.choices[0].message.content.strip() return summary except Exception as e: logger.error(fLLM summary generation failed for post {post.id}: {e}) # 降级方案返回一个基于规则的简单摘要 return f帖子讨论了关于{post.title}。社区关注度较高{post.score}分{post.num_comments}条评论。 def generate_daily_digest(self): 生成每日摘要的主流程 logger.info(Starting daily digest generation...) all_filtered_posts [] # 1. 抓取所有关注子版块的热帖 for subreddit in self.interests[subreddits]: logger.info(fScraping r/{subreddit}) posts self.scraper.fetch_hot_posts(subreddit, limit30) all_filtered_posts.extend(posts) # 短暂休眠避免请求过快 import time time.sleep(2) # 2. 过滤和排序 selected_posts self._filter_and_rank_posts(all_filtered_posts) logger.info(fSelected {len(selected_posts)} posts after filtering.) digest_content f# 你的Reddit技术摘要 {datetime.now().strftime(%Y-%m-%d)}\n\n # 3. 为每个选中的帖子生成摘要 for i, post in enumerate(selected_posts, 1): digest_content f## {i}. {post.title}\n digest_content f**子版块:** r/{post.subreddit} | **热度:** {post.score} ↑ | **评论:** {post.num_comments} \n digest_content f**原文链接:** [点击查看]({post.permalink})\n\n # 获取热门评论QClaw top_comments self.scraper.fetch_post_comments(post.permalink, depth1) # 生成LLM摘要 llm_summary self._generate_summary_with_llm(post, top_comments) digest_content f** 智能摘要:**\n{llm_summary}\n\n # 附上精选评论 if top_comments: digest_content ** 社区精选评论:**\n for comment in top_comments[:2]: # 取前两条最热评论 # 简单清理评论文本如移除换行符 clean_body comment[body].replace(\n, ).strip()[:300] digest_content f- *\{clean_body}...\* (by u/{comment.get(author, Unknown)}, ↑{comment.get(score, 0)})\n digest_content \n digest_content ---\n\n # 4. 存储到向量数据库以备后续检索 for post in selected_posts: self.vector_store.store_post( post_idpost.id, titlepost.title, contentpost.selftext, metadata{ subreddit: post.subreddit, score: post.score, url: post.permalink, date: datetime.fromtimestamp(post.created_utc).isoformat() } ) # 5. 推送 self._send_notifications(digest_content) logger.info(Daily digest generated and sent successfully.) def _send_notifications(self, content: str): 通过邮件和Telegram发送摘要 subject fReddit Digest - {datetime.now().strftime(%Y-%m-%d)} # 发送邮件 try: self.email_sender.send( toself.config[notification][email][recipient], subjectsubject, contentscontent ) logger.info(Email notification sent.) except Exception as e: logger.error(fFailed to send email: {e}) # 发送Telegram消息由于内容可能很长可以分段或发送摘要链接 # 这里简化处理发送一个提示和首条内容 telegram_msg f{subject}\n\n今日精选了{len(content.split(##)) - 1}个帖子。完整内容已发送至邮箱。\n\n第一条摘要\n first_summary_start content.find(** 智能摘要:**) first_summary_end content.find(** 社区精选评论:**, first_summary_start) if first_summary_start ! -1 and first_summary_end ! -1: telegram_msg content[first_summary_start:first_summary_end].replace(**, ).strip()[:500] ... try: self.telegram_bot.send_message(chat_idself.telegram_chat_id, texttelegram_msg) logger.info(Telegram notification sent.) except TelegramError as e: logger.error(fFailed to send Telegram message: {e}) # 配置文件示例 config.json { scraper: { user_agent: RedditDigestBot/1.0 (by /u/YourUsername), rate_limit_delay: 1.5 }, vector_store: { model_name: all-MiniLM-L6-v2, persist_dir: ./data/chroma_db }, llm: { openai_api_key: your-api-key-here, model: gpt-3.5-turbo }, notification: { email: { sender: your-emailgmail.com, password: your-app-password, recipient: recipient-emailexample.com }, telegram: { bot_token: YOUR_BOT_TOKEN, chat_id: YOUR_CHAT_ID } }, user_interests: { subreddits: [programming, Python, rust, datascience, webdev], keywords: [tutorial, guide, performance, async, database, framework], block_keywords: [hire, looking for, meme, shitpost], max_posts_per_digest: 8 } } 这个DigestGenerator类是整个系统的大脑。它协调抓取、过滤、摘要生成、存储和推送的全流程。_filter_and_rank_posts方法体现了“WorkBuddy”的个性化能力而_generate_summary_with_llm则是智能化的核心。通过配置文件你可以轻松调整关注的子版块、关键词和推送方式。4. 部署、优化与常见问题排查将代码跑起来只是第一步要让“WorkBuddy”稳定可靠地长期工作还需要考虑部署、监控和优化。4.1 部署方案选择对于个人使用有几种轻量级的部署方式本地Cron任务最简单在Linux/Mac的服务器或常年开机的电脑上使用crontab定时执行脚本。# 每天上午9点运行摘要生成 0 9 * * * cd /path/to/reddit-digest /path/to/venv/bin/python run_digest.py在run_digest.py中你只需要初始化DigestGenerator并调用generate_daily_digest()方法。云服务器Supervisor在VPS上部署使用Supervisor或systemd来管理进程确保脚本崩溃后能自动重启。这种方式更稳定适合生产环境。无服务器函数更现代使用AWS Lambda、Google Cloud Functions或Vercel Serverless Functions。将核心逻辑打包成函数通过CloudWatch Scheduler或Cron触发。优点是完全不用管理服务器按需付费。但需要注意函数运行时间限制和冷启动问题并且需要处理好数据库如ChromaDB的持久化存储可能需要改用云端的向量数据库服务。4.2 性能优化与稳定性提升异步抓取当前的RedditScraper使用同步的requests库。当需要抓取多个子版块时是串行执行的总耗时是各个请求之和。可以改用aiohttp进行异步并发抓取能大幅缩短数据获取时间。但务必注意异步并发更容易触发API的频率限制需要更精细的速率控制例如使用asyncio.Semaphore。错误处理与重试网络请求天生不稳定。必须为所有外部API调用Reddit API、LLM API添加健壮的重试机制。可以使用tenacity库以指数退避策略进行重试。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def fetch_posts_with_retry(subreddit): return self.scraper.fetch_hot_posts(subreddit)数据去重Reddit的热门帖子列表在不同时间抓取可能会有重叠。在存储或生成摘要前应根据帖子ID进行去重避免同一条信息在多次摘要中出现。摘要缓存对于已经生成过摘要的帖子可以将其摘要结果缓存起来例如存到数据库或本地文件。如果该帖子第二天仍然在热门列表中可以直接使用缓存节省LLM API的调用费用和等待时间。4.3 常见问题排查实录在实际运行中你肯定会遇到各种问题。以下是我踩过的一些坑和解决方案问题1抓取很快被Reddit限制或返回429错误。原因请求频率过高或User-Agent设置不当。排查检查代码中的rate_limit_delay是否足够长建议至少1秒。确保User-Agent字符串格式正确包含应用名、版本和你的Reddit用户名。解决增加请求间隔尤其是在QClaw深度抓取评论时。考虑使用Reddit的OAuth2认证认证后的请求会有更高的频率限制。对于非常重要的应用可以申请Reddit的API使用许可。问题2LLM生成的摘要有时偏离主题或包含奇怪内容。原因提示词Prompt不够精确或者帖子/评论内容本身噪声太大如包含大量代码、链接或非英语内容。排查打印出发送给LLM的完整提示词和原文内容检查是否有无关信息。解决优化提示词增加更明确的指令例如“只总结技术讨论部分忽略个人轶事和玩笑”。在将文本发送给LLM前进行预处理移除代码块用[code snippet]代替、截断过长的文本、过滤掉非英语内容如果只关注英文社区。问题3向量搜索返回的结果不相关。原因嵌入模型不适合该领域或者查询文本太短、太模糊。排查检查存储的“文档”文本帖子标题正文是否包含足够的信息。尝试用不同的查询语句。解决尝试不同的sentence-transformers模型例如all-mpnet-base-v2效果更好但更慢。对于查询可以尝试将用户的关键词扩展成更完整的句子例如将“Python async”改为“discussions about asynchronous programming in Python”。问题4Telegram Bot推送失败。原因Token或Chat ID错误消息内容过长Telegram有消息长度限制网络问题。排查检查配置文件的Token和Chat ID是否正确。尝试发送一条简单的测试消息。解决确保从BotFather获取了正确的Token。Chat ID可以通过向你的Bot发送/start消息然后访问https://api.telegram.org/botYOUR_BOT_TOKEN/getUpdates来查看。对于长内容需要将摘要分割成多条消息发送或者改为发送一个指向托管摘要网页的链接。问题5摘要内容突然变少或为空。原因兴趣过滤条件太严格某个子版块抓取失败Reddit API返回结构变化。排查增加日志输出查看每个子版块抓取到了多少帖子过滤前后各剩下多少。检查Reddit API的响应数据格式。解决放宽过滤条件例如减少关键词数量或增加max_posts_per_digest。为抓取函数添加更详细的异常日志以便快速定位是哪个子版块出了问题。定期检查代码因为Reddit偶尔会调整其API。将这个系统搭建并运行起来后它就像一位不知疲倦的“WorkBuddy”每天准时为你奉上量身定制的信息精华。你可以根据反馈不断微调兴趣配置优化摘要质量。随着时间的推移向量数据库里积累的数据还能让你进行更有趣的分析比如“过去一个月里社区讨论最热烈的Python框架是什么”这不仅是获取信息的工具更是你理解技术趋势的私人雷达。