我用Claude API搭了一个AI资讯监控机器人，踩了这几个坑才跑稳

上周有人问我每天是怎么跟上AI行业节奏的。

我发了一张Telegram截图给他——密密麻麻的推送消息，每条都有标题、一句话摘要、分类标签和原文链接。他问：这是什么工具？

我说，是我自己搭的。

他问多久搭好的。我说：搭起来只用了一个下午，但让它稳定跑下去花了我两周。

这篇文章，就是那两周踩坑经验的完整复盘。

---

为什么不用现成的工具

不是没试过。

早期我用过几款AI资讯聚合产品，体验都差不多：噪音太多、分类混乱、推荐逻辑不透明。更关键的是，我关注的方向很窄——主要是模型能力进展、API生态动向、国内大厂的产品节奏——但这些工具总想给我推"AI写作"、"AI绘画"、"某某公司融资"。

过滤不掉，就会逐渐失去信任感，最后干脆不看。

另一个动机更私人：我想把推送直接打到Telegram，因为我的信息消费主要在那里。现成工具基本不支持这种定制化输出。

核心诉求说清楚就是三条：

  • 定向追踪：只看我关心的几个方向
  • 智能摘要：每条资讯一句话说清楚"这条有什么新的"
  • 推送到我的地方：Telegram，不是App，不是邮件

这三条用Claude API都能做到。于是就开始搭了。

---

整体架构：先有全局感，再看细节

整个工作流的链路很清晰：

数据源（RSS + 少量爬虫）

↓


内容抓取 & 清洗


↓


URL哈希去重（过滤已处理过的）


↓


Claude API（摘要 + 分类 + 重要性评分）


↓


二次过滤（评分低于阈值的丢弃）


↓


Telegram Bot 推送


↓


本地日志 & 错误告警

技术栈选择上没什么特别：Python为主，RSS解析用feedparser，调度用APScheduler，Telegram推送用python-telegram-bot。

选Claude API的理由只有两个：

第一，长文本理解能力强。有些源的文章动辄四五千字，我需要它在理解完整语境后再给摘要，而不是只看前几段。

第二，指令遵循稳定。我需要结构化输出（JSON格式），Claude在格式约束方面的一致性比我早期测试的其他方案要好。

整个系统每天定时跑4次，分别在早上7点、中午12点、下午5点、晚上10点。每次处理约50-120条原始条目，最终推送的大概在10-25条之间。

---

踩过的3个坑

这是全文最重要的部分。每个坑我都按"现象 → 排查 → 解法"的顺序说。

坑1：Prompt写得"太聪明"反而输出不稳定

现象：

最开始我写了一个"全能Prompt"，试图在一次调用里完成所有任务：摘要、分类、重要性评分、提取关键词、判断是否与竞品相关……一共要求输出7个字段的JSON。

前几天跑得挺好。但大概第三天开始，偶尔会出现JSON格式错误——有时候某个字段缺失，有时候直接输出了一段解释性文字而不是JSON，有时候字段顺序乱了导致解析失败。

失败率大概在8%-12%之间，看起来不高，但每天几百次调用累积下来，丢失的内容不少。

排查过程：

我把失败的case全部存了下来，发现有一个规律：失败几乎都发生在文章内容比较复杂、或者涉及多个话题的条目上。

问题根源是：我的Prompt在要求模型同时做太多判断，当输入内容本身有歧义时，模型会"想多了"，开始在输出里解释自己的判断逻辑，破坏了JSON结构。

解法：

拆分任务。把"摘要生成"和"分类+评分"拆成两次调用，每次只做一件事。同时在Prompt里加入更强的格式约束：

system_prompt = """你是一个资讯处理助手。

你的输出必须且只能是合法的JSON格式，不包含任何其他文字。


如果你无法完成任务，请输出：{"error": "无法处理"}


绝对不要输出JSON以外的任何内容。"""

同时把temperature设为0，让输出更确定性。

拆分之后，格式错误率从8%-12%降到了不足1%。

---

坑2：不限速导致Rate Limit集中爆发

现象：

搭好之后跑了几天很正常。直到某天早上，系统积压了一批内容（前一天晚上有个大新闻，RSS更新了很多条），早上7点定时任务触发，一次性要处理180多条内容，API调用在几分钟内密集发出，触发了Rate Limit，大量请求失败。

排查过程：

看了一下日志，问题很明确：没有做任何限速，所有调用几乎是并发的。Claude API有每分钟请求数和token数的双重限制，批量处理时很容易踩线。

解法：

加入队列机制 + 指数退避重试逻辑。核心代码如下：

import time

import random


from anthropic import Anthropic, RateLimitError, APIError



client = Anthropic(


base_url="https://api.884819.xyz",  # 兼容官方格式，直接换base_url即可


api_key="your_api_key"


)



def call_claude_with_retry(prompt: str, max_retries: int = 5) -> str:


"""带指数退避重试的Claude API调用"""


for attempt in range(max_retries):


try:


response = client.messages.create(


model="claude-opus-4-6",


max_tokens=512,


temperature=0,


system=SYSTEM_PROMPT,


messages=[{"role": "user", "content": prompt}]


)


return response.content[0].text



except RateLimitError:


if attempt == max_retries - 1:


raise


# 指数退避：1s, 2s, 4s, 8s, 16s，加随机抖动


wait_time = (2 ** attempt) + random.uniform(0, 1)


print(f"Rate limit触发，{wait_time:.1f}秒后重试（第{attempt+1}次）")


time.sleep(wait_time)



except APIError as e:


if attempt == max_retries - 1:


raise


time.sleep(2 ** attempt)



return ""




def process_batch(items: list, interval: float = 1.5) -> list:


"""批量处理，每次调用之间强制间隔"""


results = []


for item in items:


result = call_claude_with_retry(item["content"])


results.append(result)


time.sleep(interval)  # 固定间隔，避免突发


return results

加了重试逻辑和调用间隔之后，失败率从高峰期的约30%降到了接近0。

💡 如果你还没有Claude API访问权限，或者被地区限制卡住了，我目前用的是 [api.884819.xyz](https://api.884819.xyz) 做中转——接口格式和官方完全兼容，上面的代码直接换一下 base_url 就能跑，不需要改其他任何东西。新用户注册即送体验token，国产模型（Deepseek/千问等）完全免费，按量付费没有月租。

---

坑3：没做去重，同一条新闻反复推送

现象：

某天下午，我收到了关于同一个模型发布的3条推送，内容几乎一样，只是来自不同的RSS源。同一件事被摘要了3次，推送了3次。

这不只是体验问题——还造成了不必要的token消耗。

排查过程：

问题很简单：我只做了"当次运行内"的去重，但没有跨运行的持久化去重。同一条新闻被不同RSS源收录后，下次定时任务触发，又会被当成新内容处理。

解法：

两层去重：

第一层（URL哈希）： 对每条内容的URL做MD5，存入本地SQLite。每次处理前先查库，已存在的跳过。 
import hashlib

import sqlite3



def get_url_hash(url: str) -> str:


return hashlib.md5(url.encode()).hexdigest()



def is_duplicate(url: str, db_path: str = "seen_urls.db") -> bool:


url_hash = get_url_hash(url)


conn = sqlite3.connect(db_path)


cursor = conn.cursor()


cursor.execute(


"CREATE TABLE IF NOT EXISTS seen (hash TEXT PRIMARY KEY, ts INTEGER)"


)


cursor.execute("SELECT 1 FROM seen WHERE hash = ?", (url_hash,))


exists = cursor.fetchone() is not None


if not exists:


import time


cursor.execute(


"INSERT INTO seen VALUES (?, ?)", (url_hash, int(time.time()))


)


conn.commit()


conn.close()


return exists
第二层（标题相似度）： 对于不同URL但内容高度相似的情况（转载、洗稿），用标题的简单字符相似度做二次过滤。轻量场景下不需要上向量数据库，difflib.SequenceMatcher就够用： 
from difflib import SequenceMatcher


def is_similar_title(new_title: str, recent_titles: list, threshold: float = 0.75) -> bool:


for title in recent_titles:


ratio = SequenceMatcher(None, new_title, title).ratio()


if ratio > threshold:


return True


return False

两层去重加上之后，重复推送问题基本消失。

---

现在跑得最稳的配置

给出关键参数配置清单，可以直接参考：

| 参数 | 推荐值 | 说明 | | temperature | 0 | 结构化输出必须确定性 | | max_tokens | 512 | 摘要任务够用，控制成本 | | 调用间隔 | 1.5s | 避免突发Rate Limit | | 最大重试次数 | 5 | 指数退避，上限约30s | | 去重窗口 | 7天 | 超过7天的URL哈希清理 | | 每批处理上限 | 60条 | 单次任务不要太贪 | | 日志级别 | INFO + 错误单独告警 | 方便排查不刷屏 | Prompt模板（摘要+分类二合一，当前生产版本）： 
你是一个AI行业资讯处理助手。

你的输出必须且只能是合法的JSON，不包含任何其他文字。



处理以下文章，输出格式如下：


{


"summary": "一句话摘要，说明这条资讯的核心新意（50字以内）",


"category": "模型进展/产品发布/行业动态/研究论文/其他 中选一个",


"importance": 1-5的整数，5最重要,


"is_relevant": true或false（是否与大模型/AI基础设施相关）


}



文章标题：{{title}}


文章内容：{{content}}

---

跑了两个月，真实效果怎样

几个实际数据（主观估算，非精确统计）：

  • 每天处理原始条目：约200-400条
  • 过滤后实际推送：约20-40条，过滤率约85%-90%
  • 重复推送发生率：加去重后接近0
  • API调用失败率：加重试后低于1%
  • 每天token消耗：约15万-25万token（摘要任务为主）
费用估算： 按Claude Sonnet级别的定价粗算，每天大概在1-3元人民币之间，一个月不到100元。对我来说完全可以接受。 还没解决的问题：
  • 小语种内容处理较差，日文、韩文的AI资讯摘要质量明显下降
  • 某几个垂直方向的RSS源不稳定，时不时抓不到内容
  • 重要性评分的校准还不够准，有时候一些真正重要的进展被打了低分

这些问题我知道在哪，但还没有排上优先级去修。

---

最后说一句

这套工作流目前只做到了"信息收集+摘要"，本质上还是帮我读资讯。

但我最近在想一件事：下一步能不能让它自动判断——哪条动态值得写成一篇文章，并给出初步的选题角度？

也就是说，让Claude不只是帮我读资讯，而是帮我做编辑决策的第一层过滤。

从"信息消费"升级到"内容生产辅助"，这个方向我已经有了一些思路。如果这篇的反响还不错，下篇写这个。

---

本文由8848AI原创，转载请注明出处。关注8848AI，带你从零开始学AI。

#AI工具 #Claude #自动化工作流 #Python #API开发 #8848AI #AI资讯 #Telegram机器人