万字深度解析 last30days-skill：当 AI Agent 遇见跨平台趋势猎手——从信号聚合到 grounded summary、从多平台数据源到可扩展插件架构的完整技术指南（2026）

在信息过载的时代，Google 告诉你编辑想让你看到的，而 /last30days 告诉你真实的人在讨论什么。

目录

1. 问题背景：传统搜索的困境与 AI Agent 搜索的崛起
2. last30days-skill 是什么？核心设计哲学
3. 架构深度解析：四层架构与数据流动全景
4. 信号源全景：10 大平台数据接入技术详解
5. 核心模块代码实战
6. 安装与集成：5 大 AI 编程工具适配指南
7. 高级用法：自定义数据源与插件开发
8. 生产级部署：性能优化与错误处理
9. 实战案例：用 last30days 做技术调研
10. 与其他方案的对比：为什么 last30days 胜出
11. 未来展望：AI Agent 搜索的下一站
12. 总结

1. 问题背景：传统搜索的困境与 AI Agent 搜索的崛起

1.1 传统搜索引擎的结构性缺陷

当你在 Google 搜索 "best AI coding tool 2026" 时，你得到的是什么？

• SEO 操纵的内容：排名前列的往往是花了钱做 SEO 的营销页面
• 过时信息：索引延迟导致你看到的是 3 个月前的排名
• 编辑偏见：算法推荐的是"编辑认为你应该看的"，而非"真实用户在讨论的"
• 信息茧房：你无法感知 Reddit 上某个工具的吐槽、Hacker News 上的激烈辩论、X 上的真实用户反馈

传统搜索的信息流：
用户查询 → Google 索引 → SEO 优化的页面 → 排名算法 → 呈现给用户
                ↓
          编辑/营销内容占主导
          真实用户声音被稀释

1.2 AI Agent 编程时代的新型信息需求

2026 年，AI Agent（Claude Code、Cursor、Copilot 等）已成为开发者的日常工具。但一个核心问题浮现：

AI Agent 的知识截止日期是它的训练数据，它怎么知道"上周 RustFS 发布了 2.0 版本"？

传统做法是开发者手动去 GitHub Trending、Reddit、Hacker News 上翻，然后复制粘贴给 AI Agent。这不仅低效，而且信息碎片化严重。

last30days-skill 的核心价值：让 AI Agent 自己去做调研，然后把 grounded（有依据的）摘要返回给你。

last30days 的信息流：
用户查询 → AI Agent 调用 /last30days → 并行抓取 10 个平台 → 
信号聚合与打分 → 综合摘要生成 → 带引用来源的完整报告
                                                ↓
                                        真实用户声音
                                        最新趋势（≤30天）
                                        多维度交叉验证

1.3 为什么是"最近 30 天"？

项目名 last30days 揭示了一个关键设计决策：时效性优先。

在技术迭代以周为单位进行的今天（Go 1.26、Deno 2.9、Headroom 等项目的爆发式增长），30 天前的讨论可能已经过时。last30days 通过：

• 时间过滤：只保留最近 30 天的内容
• 热度排序：按 engagement（点赞、评论、分享）打分
• 信号交叉验证：同一话题在 Reddit、X、YouTube 上都被讨论 → 高置信度

2. last30days-skill 是什么？核心设计哲学

2.1 项目概览

2.2 核心设计哲学

哲学一：搜索"人"，而非搜索"网站"

项目作者 mvanhorn 的表述非常精准：

"Google 聚合的是编辑和 SEO 者的内容，而 /last30days 搜索的是真实的、正在发生的人。"

这不是一个搜索引擎替代品，而是一个社会信号聚合器。它回答的问题是："真实的人在最近 30 天里，对某个话题到底是怎么看的？"

哲学二：Grounded Summary（有依据的摘要）

传统 AI 生成内容的最大问题是"幻觉"——它看起来权威，但你不知道信息从哪来的。

last30days 的每份报告都包含：

• 引用来源：每条结论都标注来自哪个平台、哪条具体帖子
• 热度打分：这条信息有多少人认同（upvotes/comments/shares）
• 情绪标注：社区是支持、反对还是中立

哲学三：对 AI Agent 友好（Agent-Optimized Output）

输出格式专门为 AI Agent 消费设计：

• 结构化 JSON（方便程序解析）
• 降低 token 消耗（摘要而非全文）
• 支持深度参数（shallow/default/deep）控制返回粒度

哲学四：可扩展架构（Plugin-First）

每个数据源都是独立的 Python 模块（scripts/lib/reddit.py、scripts/lib/x.py 等），遵循统一的接口规范。添加新平台只需：

1. 在 scripts/lib/ 下新建模块
2. 实现 search_<platform>(topic, depth) 函数
3. 在 scripts/scan.py 中注册

3. 架构深度解析：四层架构与数据流动全景

3.1 整体架构图

┌─────────────────────────────────────────────────────────────────┐
│                    AI Agent 交互层                                │
│  /last30days <topic>  [Claude Code / Cursor / Codex / ...]     │
└───────────────────────────┬─────────────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────────────┐
│                  Level 1:触发层（.claude-plugin）                  │
│  超轻量级描述 → Agent 判断是否调用 → 触发 scan.py                │
└───────────────────────────┬─────────────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────────────┐
│                  Level 2:工具定义层（scripts/scan.py）            │
│  参数解析 → 深度控制 → 平台路由 → 并行调度                        │
└───────────────────────────┬─────────────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────────────┐
│                  Level 3:信号采集层（scripts/lib/*.py）            │
│  Reddit  │ X  │ YouTube  │ TikTok  │ HN  │ Polymarket  │ ...   │
│  每个模块：查询扩展 → API调用 → 数据标准化 → 热度打分             │
└───────────────────────────┬─────────────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────────────┐
│                  Level 4:综合层（scripts/synthesize.py）           │
│  去重 → 按平台分组 → 交叉验证 → 生成 grounded summary            │
└───────────────────────────┬─────────────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────────────┐
│                      输出层（Markdown / JSON）                     │
│  带引用来源的完整报告 → 返回给 AI Agent → 呈现给用户             │
└─────────────────────────────────────────────────────────────────┘

3.2 Level 1：触发层（.claude-plugin 目录）

这是 last30days 与 AI Agent 的"握手协议"。核心文件是 .claude-plugin/skill.md，它包含一个极其精简的描述：

# last30days

Research any topic across Reddit, X, YouTube, HN, Polymarket, and the web from the last 30 days. Returns a grounded summary with citations.

When to use:
- User asks about trends, tools, or opinions on a topic
- Need recent community feedback (last 30 days)
- Comparing tools or technologies

为什么描述要精简？

AI Agent 的 context window 是有限的。每次 Agent 加载技能列表时，都会读取这些描述。如果描述太长，会消耗大量 token，导致 Agent 倾向于"偷懒"不调用这个技能。

last30days 的描述控制在 50 词以内，确保 Agent 能快速判断是否需要调用。

3.3 Level 2：工具定义层（scripts/scan.py）

这是整个技能的入口点。让我们看核心逻辑：

# scripts/scan.py（简化版）
import sys
import json
from scripts.lib import reddit, x, youtube, hn, polymarket, web

def scan(topic: str, depth: str = "default", platforms: list = None):
    """
    topic: 调研主题
    depth: shallow / default / deep（控制返回粒度）
    platforms: 指定平台（默认全部）
    """
    results = {}
    
    # 平台路由表
    platform_handlers = {
        "reddit": reddit.search_reddit,
        "x": x.search_x,
        "youtube": youtube.search_youtube,
        "hackernews": hn.search_hn,
        "polymarket": polymarket.search_polymarket,
        "web": web.search_web,
    }
    
    # 并行调度（关键优化！）
    from concurrent.futures import ThreadPoolExecutor
    with ThreadPoolExecutor(max_workers=6) as executor:
        future_to_platform = {}
        for platform in (platforms or platform_handlers.keys()):
            handler = platform_handlers[platform]
            future = executor.submit(handler, topic, depth)
            future_to_platform[future] = platform
        
        for future in concurrent.futures.as_completed(future_to_platform):
            platform = future_to_platform[future]
            try:
                results[platform] = future.result()
            except Exception as e:
                results[platform] = {"error": str(e)}
    
    # 交给综合层
    return synthesize(results, topic)

if __name__ == "__main__":
    topic = sys.argv[1]
    depth = sys.argv[2] if len(sys.argv) > 2 else "default"
    result = scan(topic, depth)
    print(json.dumps(result, ensure_ascii=False, indent=2))

关键技术点：

1. 并行调度：6 个平台同时抓取，总耗时 ≈ 最慢的那个平台（而非累加）
2. 深度控制：shallow 只返回标题和热度；default 返回摘要；deep 返回完整评论树
3. 错误隔离：某个平台 API 失效不影响其他平台的结果

3.4 Level 3：信号采集层（以 Reddit 为例）

scripts/lib/reddit.py 是整个项目中最复杂的模块之一。让我们深入：

# scripts/lib/reddit.py（核心逻辑）
import requests
from datetime import datetime, timedelta

def search_reddit(topic: str, depth: str = "default") -> list:
    """
    Reddit 搜索的核心流程：
    1. 查询扩展：把 "AI coding" 扩展为 ["AI coding", "AI programming", "Claude Code", ...]
    2. 执行搜索：调用 Reddit 的 JSON API（无需官方 API key！）
    3. 时间过滤：只保留最近 30 天的帖子
    4. 数据标准化：统一字段名（platform, title, url, heat_score, ...）
    """
    
    # 步骤1：查询扩展（关键创新！）
    queries = expand_reddit_queries(topic)
    # 例如："AI agent search" → ["AI agent search", "AI agent research", 
    #       "last30days", "agent skills", ...]
    
    # 步骤2：执行搜索
    all_results = []
    for query in queries:
        url = f"https://www.reddit.com/search.json?q={query}&sort=hot&limit=25&t=month"
        headers = {"User-Agent": "last30days-skill/2.9.5"}
        resp = requests.get(url, headers=headers)
        posts = resp.json()["data"]["children"]
        
        for post in posts:
            p = post["data"]
            # 步骤3：时间过滤
            created = datetime.fromtimestamp(p["created_utc"])
            if datetime.now() - created > timedelta(days=30):
                continue
            
            # 步骤4：数据标准化
            all_results.append({
                "platform": "reddit",
                "title": p["title"],
                "url": "https://reddit.com" + p["permalink"],
                "heat_score": compute_heat(p),  # 自定义热度算法
                "engagement": {
                    "upvotes": p["ups"],
                    "comments": p["num_comments"],
                },
                "content": extract_content(p, depth),  # 根据深度返回不同粒度
                "created": created.isoformat(),
            })
    
    # 去重（同一帖子可能被多个查询命中）
    return deduplicate(all_results)

def compute_heat(post: dict) -> float:
    """
    热度打分算法（简化版）：
    heat = upvotes * 1.0 + comments * 3.0 + (近期加分)
    
    为什么评论权重更高？
    → 因为评论代表"真实讨论"，比单纯点赞更有价值
    """
    upvotes = post.get("ups", 0)
    comments = post.get("num_comments", 0)
    # 时间衰减：每过一天，热度乘以 0.95
    days_old = (datetime.now() - datetime.fromtimestamp(post["created_utc"])).days
    time_decay = 0.95 ** days_old
    
    return (upvotes * 1.0 + comments * 3.0) * time_decay

Reddit 模块的技术亮点：

1. 无需 API Key：使用 Reddit 的公开 JSON API（reddit.com/search.json），绕过官方 API 的严格限制
2. 查询扩展：通过 LLM（或规则）把用户输入扩展为多个相关查询，提高召回率
3. 热度算法：评论权重是点赞的 3 倍，因为评论代表深度讨论
4. 时间衰减：30 天前的内容热度会衰减到几乎为 0

3.5 Level 4：综合层（scripts/synthesize.py）

这是 last30days 的"大脑"——把来自 10 个平台的碎片化信息，综合成一份有逻辑的、带引用的报告。

# scripts/synthesize.py（核心逻辑）
def synthesize(platform_results: dict, topic: str) -> dict:
    """
    综合算法的核心步骤：
    1. 去重：跨平台去重（同一话题在不同平台被讨论）
    2. 按平台分组：Reddit 的讨论型内容 vs X 的实时型内容 vs YouTube 的深度型内容
    3. 交叉验证：如果同一结论在 ≥3 个平台都被提及 → 高置信度
    4. 生成 grounded summary
    """
    
    # 步骤1：跨平台去重
    seen_titles = set()
    unique_results = []
    for platform, items in platform_results.items():
        for item in items:
            title_key = normalize_title(item["title"])
            if title_key not in seen_titles:
                seen_titles.add(title_key)
                unique_results.append(item)
    
    # 步骤2：按平台特性分组
    grouped = group_by_platform(unique_results)
    # Reddit → "社区讨论"
    # X → "实时舆情"
    # YouTube → "深度解读"
    # Polymarket → "预测市场信号"
    
    # 步骤3：交叉验证
    cross_validation = {}
    for item in unique_results:
        key_points = extract_key_points(item["content"])
        for point in key_points:
            if point not in cross_validation:
                cross_validation[point] = []
            cross_validation[point].append(item["url"])
    
    # 步骤4：生成报告
    report = {
        "topic": topic,
        "generated_at": datetime.now().isoformat(),
        "summary": generate_summary(grouped, cross_validation),
        "by_platform": format_by_platform(grouped),
        "top_sources": rank_by_heat(unique_results[:10]),
        "cross_validated_claims": [
            {"claim": point, "sources": urls, "confidence": len(urls) / 6}
            for point, urls in cross_validation.items()
            if len(urls) >= 2  # 至少 2 个平台提及
        ],
    }
    
    return report

4. 信号源全景：10 大平台数据接入技术详解

last30days 目前支持 10 个信号源。让我们逐一解析技术接入方案：

4.1 Reddit：社区讨论的黄金矿藏

接入方式：Reddit 公开 JSON API（无需 API Key）

# 无需认证即可访问
curl "https://www.reddit.com/search.json?q=AI+agent&sort=hot&limit=10&t=month" \
  -H "User-Agent: last30days-skill/2.9.5"

技术要点：

• 速率限制：匿名访问约 30 次/分钟，建议加延迟
• 排序选项：sort=hot（热门）/ sort=new（最新）/ sort=top（最高分）
• 时间过滤：t=hour/day/week/month/year/all

数据价值：技术讨论的深度最高，评论区往往比正文更有价值。

4.2 X（Twitter）：实时舆情的脉搏

接入方式：使用 twitter-api-client（非官方但稳定）

# scripts/lib/x.py
from twitter_api_client import TwitterClient

def search_x(topic: str, depth: str) -> list:
    client = TwitterClient(
        auth_type="oauth2",
        consumer_key=os.getenv("TWITTER_CONSUMER_KEY"),
        consumer_secret=os.getenv("TWITTER_CONSUMER_SECRET"),
    )
    
    tweets = client.search_tweets(
        q=topic,
        count=50,
        result_type="recent",  # 只返回最近 30 天
    )
    
    return [format_tweet(t) for t in tweets]

技术要点：

• 需要 Twitter Developer Account（免费层级足够）
• 速率限制：450 次/15 分钟
• 高级技巧：使用 since_time 参数精确控制时间范围

数据价值：实时性强，适合捕捉"刚刚发生的事件"（如某个开源项目突然爆火）。

4.3 YouTube：深度解读与教程

接入方式：YouTube Data API v3

# scripts/lib/youtube.py
from googleapiclient.discovery import build

def search_youtube(topic: str, depth: str) -> list:
    youtube = build("youtube", "v3", api_key=os.getenv("YOUTUBE_API_KEY"))
    
    # 步骤1：搜索视频
    search_response = youtube.search().list(
        q=topic,
        part="snippet",
        maxResults=25,
        publishedAfter=(datetime.now() - timedelta(days=30)).isoformat(),
        type="video",
    ).execute()
    
    # 步骤2：获取视频详情（观看数、点赞数）
    video_ids = [item["id"]["videoId"] for item in search_response["items"]]
    stats_response = youtube.videos().list(
        id=",".join(video_ids),
        part="statistics",
    ).execute()
    
    # 步骤3：可选——获取字幕（深度模式）
    if depth == "deep":
        for video in stats_response["items"]:
            video["transcript"] = get_transcript(video["id"])
    
    return format_videos(stats_response["items"])

技术要点：

• API 配额：每天 10,000 单位（一次搜索消耗 100 单位）
• 字幕获取：使用 youtube-transcript-api 库，无需 API Key
• 热度算法：观看数 × 0.3 + 点赞数 × 1.0 + 评论数 × 2.0

数据价值：适合"如何使用某工具"的调研，字幕中往往包含作者的真实看法。

4.4 Hacker News：技术社区的精英视角

接入方式：Hacker News API（Algolia）

# scripts/lib/hn.py
import requests

def search_hn(topic: str, depth: str) -> list:
    """
    HN 使用 Algolia 提供的搜索 API，功能强大
    """
    url = "https://hn.algolia.com/api/v1/search"
    params = {
        "query": topic,
        "tags": "story",  # 只搜索帖子，不包含评论
        "numericFilters": f"created_at_i>{int((datetime.now() - timedelta(days=30)).timestamp())}",
        "hitsPerPage": 25,
    }
    
    resp = requests.get(url, params=params)
    hits = resp.json()["hits"]
    
    return [{
        "platform": "hackernews",
        "title": hit["title"],
        "url": f"https://news.ycombinator.com/item?id={hit['objectID']}",
        "heat_score": hit["points"],
        "engagement": {"comments": hit["num_comments"]},
        "content": hit.get("story_text", "")[:500 if depth == "shallow" else 5000],
    } for hit in hits]

技术要点：

• 无需 API Key
• 支持复杂的数值过滤（如 numericFilters）
• 可获取完整评论树（深度模式）

数据价值：HN 用户是技术精英，讨论质量极高，但注意"精英偏见"。

4.5 Polymarket：预测市场的集体智慧

接入方式：Polymarket API（基于 CLOB）

# scripts/lib/polymarket.py
import requests

def search_polymarket(topic: str, depth: str) -> list:
    """
    Polymarket 是预测市场平台，用户用真金白银投票
    → 这比点赞、评论更有说服力（"Talk is cheap, show me the money"）
    """
    url = "https://gamma-api.polymarket.com/markets"
    params = {
        "closed": "false",
        "limit": 25,
        "active": "true",
    }
    
    resp = requests.get(url, params=params)
    markets = resp.json()
    
    # 过滤与 topic 相关的市场
    relevant = [m for m in markets if topic.lower() in m["question"].lower()]
    
    return [{
        "platform": "polymarket",
        "title": m["question"],
        "url": f"https://polymarket.com/market/{m['conditionId']}",
        "probability": m["outcomePrices"][0],  # 0-1 之间
        "volume": m["volume"],  # 总交易量（美元）
        "interpretation": interpret_probability(m),
    } for m in relevant]

技术要点：

• 无需认证即可读取公开市场数据
• outcomePrices 是市场集体智慧的浓缩（0.7 = 70% 的人认为会发生）
• 交易量越大，信号越可靠

数据价值：Polymarket 的信号是"用钱投票"，在预测技术趋势方面出奇地准确。

4.6 其他平台概览

5. 核心模块代码实战

5.1 完整示例：添加一个自定义数据源

假设你想添加对 GitHub Discussions 的支持。步骤如下：

步骤 1：在 scripts/lib/ 下创建 github_discussions.py

# scripts/lib/github_discussions.py
import requests
from datetime import datetime, timedelta

def search_github_discussions(topic: str, depth: str = "default") -> list:
    """
    GitHub Discussions 搜索
    使用 GitHub REST API v3（无需认证，速率限制 60 次/小时）
    """
    
    # GitHub 的搜索 API
    url = "https://api.github.com/search/issues"
    params = {
        "q": f"{topic} is:discussion is:open created:>{thirty_days_ago()}",
        "sort": "comments",
        "order": "desc",
        "per_page": 25,
    }
    
    resp = requests.get(url, params=params)
    issues = resp.json()["items"]
    
    results = []
    for issue in issues:
        # 获取讨论的评论（深度模式）
        comments = []
        if depth == "deep":
            comments_resp = requests.get(issue["comments_url"])
            comments = [c["body"] for c in comments_resp.json()[:10]]
        
        results.append({
            "platform": "github_discussions",
            "title": issue["title"],
            "url": issue["html_url"],
            "heat_score": issue["comments"],
            "engagement": {
                "comments": issue["comments"],
                "reactions": sum(issue["reactions"].values()),
            },
            "content": issue["body"][:1000 if depth != "deep" else 5000],
            "comments": comments if depth == "deep" else [],
            "created": issue["created_at"],
        })
    
    return results

def thirty_days_ago() -> str:
    return (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")

步骤 2：在 scripts/scan.py 中注册

# scripts/scan.py（添加两行）
from scripts.lib import reddit, x, youtube, hn, polymarket, web, github_discussions

platform_handlers = {
    "reddit": reddit.search_reddit,
    "x": x.search_x,
    # ... 其他平台
    "github_discussions": github_discussions.search_github_discussions,  # 新增
}

步骤 3：在 .claude-plugin/skill.md 中更新平台列表

# last30days

Research any topic across Reddit, X, YouTube, HN, Polymarket, Web, **GitHub Discussions** from the last 30 days.

6. 安装与集成：5 大 AI 编程工具适配指南

last30days 支持 5 个主流 AI 编程工具。安装方式各有不同：

6.1 Claude Code

# 方式一：从 GitHub 克隆到 Claude 插件目录
git clone https://github.com/mvanhorn/last30days-skill.git ~/.claude/skills/last30days

# 方式二：使用 Claude Code 的内置命令（推荐）
/claude install-skill mvanhorn/last30days-skill

安装后，在 Claude Code 中直接使用：

/last30days RustFS vs MinIO performance

6.2 Cursor

Cursor 的技能系统与 Claude Code 兼容（都基于 .claude-plugin 规范）。

# 将 last30days 放到 Cursor 的 skills 目录
cp -r last30days-skill ~/.cursor/skills/last30days

然后在 Cursor 的 Chat 面板中：

@last30days 最近 30 天 AI Agent 框架的发展趋势

6.3 GitHub Copilot（VS Code）

Copilot 的扩展机制不同，需要通过 Copilot Skills 功能：

1. 在 VS Code 中安装 "Copilot Skills" 扩展
2. 在设置中指定技能路径：

// settings.json
{
  "copilot.skills.paths": [
    "~/.claude/skills/last30days"
  ]
}

6.4 OpenClaw

OpenClaw 使用 Skill Workshop 机制：

# 在 OpenClaw 中
skill_install last30days-skill

6.5 Gemini CLI

Gemini CLI 支持外部工具定义（Tool Schema）：

# 将 last30days 包装为 Gemini 工具
gemini tools add --name last30days --cmd "python /path/to/last30days/scripts/scan.py"

7. 高级用法：自定义数据源与插件开发

7.1 深入理解查询扩展机制

last30days 的核心竞争力之一是查询扩展——把用户的简短输入，扩展为多个相关查询，大幅提升召回率。

# scripts/lib/query_expansion.py
def expand_queries(topic: str, platform: str) -> list:
    """
    查询扩展的三种策略：
    1. 同义词扩展（使用 WordNet 或预定义映射表）
    2. 相关技术扩展（如 "AI agent" → "Claude Code", "Cursor", "AutoGPT"）
    3. LLM 辅助扩展（调用 LLM 生成相关查询）
    """
    
    # 策略1：预定义映射表（快速、无 API 调用）
    synonym_map = {
        "AI agent": ["AI assistant", "autonomous agent", "LLM agent"],
        "coding tool": ["programming tool", "developer tool", "code assistant"],
        # ...
    }
    
    # 策略2：相关技术映射
    tech_map = {
        "AI agent framework": ["LangChain", "AutoGPT", "BabyAGI", "CrewAI"],
        "vector database": ["Pinecone", "Weaviate", "Qdrant", "Milvus"],
        # ...
    }
    
    # 策略3：LLM 扩展（深度模式）
    if use_llm:
        prompt = f"""Generate 5 related search queries for: "{topic}"
        Each query should capture a different aspect of the topic.
        Output as a JSON array of strings."""
        expanded = call_llm(prompt)
        return json.loads(expanded)
    
    # 合并策略1和2
    expanded = [topic]
    for key, synonyms in synonym_map.items():
        if key in topic.lower():
            expanded.extend(synonyms)
    for key, related in tech_map.items():
        if key in topic.lower():
            expanded.extend(related)
    
    return list(set(expanded))  # 去重

7.2 热度打分算法的深入优化

基础版本的热度算法（upvotes + comments）有一个问题：容易被"水帖"带偏。

优化版本引入了评论情感分析和投票者可信度：

# scripts/lib/heat_scoring.py
from textblob import TextBlob

def compute_heat_v2(post: dict, platform: str) -> float:
    """
    优化后的热度打分：
    - 基础分 = upvotes * 1.0 + comments * 3.0
    - 情感加分：评论区整体正面 → +20%
    - 争议加分：upvote/downvote 比例接近 1:1 → +30%（说明话题有争议性，值得关注）
    - 可信度加分：发帖者账号年龄 > 1 年 → +10%
    """
    
    base = post.get("ups", 0) * 1.0 + post.get("num_comments", 0) * 3.0
    
    # 情感分析（采样评论）
    sentiment_bonus = 0
    if "top_comments" in post:
        sentiments = [TextBlob(c).sentiment.polarity for c in post["top_comments"][:5]]
        avg_sentiment = sum(sentiments) / len(sentiments)
        if avg_sentiment > 0.3:  # 明显正面
            sentiment_bonus = 0.2
    
    # 争议加分
    controversy_bonus = 0
    upvote_ratio = post.get("upvote_ratio", 0.5)
    if 0.4 < upvote_ratio < 0.6:  # 接近 1:1
        controversy_bonus = 0.3
    
    # 可信度加分
    credibility_bonus = 0
    author_created = post.get("author_created_utc", 0)
    if author_created:
        author_age_days = (datetime.now().timestamp() - author_created) / 86400
        if author_age_days > 365:
            credibility_bonus = 0.1
    
    return base * (1 + sentiment_bonus + controversy_bonus + credibility_bonus)

8. 生产级部署：性能优化与错误处理

8.1 速率限制与重试机制

在生产环境中，API 速率限制是最大的挑战。last30days 使用指数退避 + 抖动策略：

# scripts/lib/rate_limiter.py
import time
import random
from functools import wraps

def rate_limited(max_per_minute: int):
    """
    装饰器：限制函数调用频率
    """
    min_interval = 60.0 / max_per_minute
    
    def decorator(func):
        last_called = [0.0]
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait_time = max(0, min_interval - elapsed)
            if wait_time > 0:
                time.sleep(wait_time + random.uniform(0, 0.5))  # 加抖动
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        
        return wrapper
    return decorator

def exponential_backoff(func):
    """
    装饰器：指数退避重试
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        max_retries = 5
        for attempt in range(max_retries):
            try:
                return func(*args, **kwargs)
            except RateLimitError:
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited. Retrying in {wait:.1f}s...")
                time.sleep(wait)
        raise MaxRetriesExceeded()
    return wrapper

# 使用示例
@rate_limited(30)  # Reddit: 30 次/分钟
@exponential_backoff
def fetch_reddit(query: str):
    # ...
    pass

8.2 缓存策略：避免重复抓取

对于相同的查询，缓存可以大幅降低 API 调用次数：

# scripts/lib/cache.py
import sqlite3
import json
from datetime import datetime, timedelta

class ResultCache:
    def __init__(self, db_path="~/.last30days/cache.db"):
        self.conn = sqlite3.connect(os.path.expanduser(db_path))
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS cache (
                query_hash TEXT PRIMARY KEY,
                platform TEXT,
                result TEXT,
                cached_at TIMESTAMP
            )
        """)
    
    def get(self, query: str, platform: str, max_age_hours: int = 6) -> list:
        """
        获取缓存结果（6 小时内的有效）
        """
        query_hash = f"{platform}:{query}"
        row = self.conn.execute(
            "SELECT result, cached_at FROM cache WHERE query_hash = ?",
            (query_hash,)
        ).fetchone()
        
        if row:
            cached_at = datetime.fromisoformat(row[1])
            if datetime.now() - cached_at < timedelta(hours=max_age_hours):
                return json.loads(row[0])
        return None
    
    def set(self, query: str, platform: str, result: list):
        query_hash = f"{platform}:{query}"
        self.conn.execute(
            "INSERT OR REPLACE INTO cache VALUES (?, ?, ?, ?)",
            (query_hash, platform, json.dumps(result), datetime.now().isoformat()),
        )
        self.conn.commit()

8.3 错误监控与告警

在生产环境中，你需要知道哪个平台出了问题：

# scripts/lib/monitoring.py
import logging
from sentry_sdk import capture_exception

logging.basicConfig(
    filename="~/.last30days/errors.log",
    level=logging.WARNING,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

def safe_fetch(platform: str, fetch_func, *args, **kwargs):
    """
    安全抓取：捕获所有异常，记录日志，发送告警
    """
    try:
        return fetch_func(*args, **kwargs)
    except RateLimitError as e:
        logging.warning(f"[{platform}] Rate limited: {e}")
        return []
    except AuthenticationError as e:
        logging.error(f"[{platform}] Auth failed: {e}")
        capture_exception(e)  # 发送到 Sentry
        return []
    except Exception as e:
        logging.error(f"[{platform}] Unexpected error: {e}")
        capture_exception(e)
        return []

9. 实战案例：用 last30days 做技术调研

9.1 案例一：评估是否采用 RustFS 作为对象存储方案

用户输入：

/last30days RustFS vs MinIO performance benchmarks

last30days 返回的综合报告（简化版）：

# RustFS vs MinIO：社区视角（过去 30 天）

## 综合摘要

过去 30 天里，RustFS 和 MinIO 在社区中的讨论热度明显倾向 RustFS。主要观点：

1. **性能**：多位用户报告 RustFS 在特定场景下比 MinIO 快 2-3 倍（[来源1](...) [来源2](...)）
2. **成熟度**：MinIO 更稳定，生产环境首选（[来源3](...)）
3. **社区活跃度**：RustFS GitHub 提交频率是 MinIO 的 5 倍（[来源4](...)）

## 按平台分组

### Reddit（深度讨论）
- [RustFS Discord 讨论记录](...) - 234 点赞，67 评论
- 核心观点：RustFS 的 Rust 实现带来内存安全优势

### Hacker News（技术精英视角）
- [RustFS 2.0 发布](...) - 345 点，89 评论
- 争议点：是否应该为了性能放弃 MinIO 的生态？

### Polymarket（预测市场）
- "RustFS 会在 2026 年底超过 MinIO 的 GitHub Star 数吗？"
- 当前概率：68%（交易量 $45,000）

## 交叉验证的结论

以下结论在 ≥3 个平台上被提及：
- ✅ RustFS 读吞吐量更高（Reddit + HN + YouTube）
- ✅ MinIO 文档更完善（Reddit + HN）
- ⚠️ RustFS 的 S3 兼容性有边缘情况问题（Reddit + GitHub Issues）

## 推荐行动

如果你追求极致性能且可以接受早期采用者的风险 → 选择 RustFS
如果你需要稳定生产和丰富生态 → 选择 MinIO

AI Agent 基于这份报告给出的建议：

"根据过去 30 天社区的真实反馈，我建议你在非核心业务上先试点 RustFS，同时保留 MinIO 作为回退方案。具体实施步骤：..."

9.2 案例二：内容选题——找出开发者最关心的 AI 编程话题

用户输入（博客作者视角）：

/last30days AI programming tools what developers complain about

返回的关键洞察：

1. Claude Code 的上下文丢失问题（Reddit 上 5 个高赞帖子）
2. Cursor 的 Tab 补全越来越不准（X 上大量吐槽）
3. AI 生成代码的版权风险（HN 上的激烈辩论，Polymarket 有 $120k 赌注）

行动：选择"AI 生成代码的版权风险"作为下一篇博客的主题——因为这是社区最关心、争议最大、且尚未有深度文章覆盖的话题。

10. 与其他方案的对比：为什么 last30days 胜出

last30days 的独特优势：

1. 多维度信号：Google 只有网页，Perplexity 主要是新闻，而 last30days 有 Reddit（深度讨论）+ X（实时舆情）+ Polymarket（预测市场）+ YouTube（教程）= 360° 视角
2. Agent-Native：从第一天就是为 AI Agent 设计的，输出格式、token 消耗、错误隔离都考虑到了
3. 可扩展性：任何开发者都可以添加新平台支持

11. 未来展望：AI Agent 搜索的下一站

11.1 从"搜索"到"持续感知"

目前的 last30days 是被动触发的（用户输入 /last30days <topic> 才执行）。

下一步是自然演进为主动感知：

# 未来版本（概念代码）
class TrendMonitor:
    def __init__(self, topics: list):
        self.topics = topics
        self.last_scan = {}
    
    def run(self):
        """
        每小时自动扫描关注的话题，如果发现：
        - 热度突然飙升（24 小时内 +500%）
        - 出现新的高置信度结论
        → 主动通知 AI Agent
        """
        for topic in self.topics:
            current = scan(topic, depth="shallow")
            previous = self.last_scan.get(topic)
            
            if previous and detect_spike(previous, current):
                notify_agent(topic, current)
            
            self.last_scan[topic] = current

11.2 从"英文互联网"到"全球互联网"

目前的 last30days 主要覆盖英文平台。随着 last30days-skill-cn（中文版）的出现，未来可能会出现：

• 多语言信号融合：同一话题的中文讨论 + 英文讨论同时呈现
• 文化视角标注："这是中国开发者的典型看法" vs "这是硅谷的典型看法"

11.3 从"信号聚合"到"信号预测"

Polymarket 的集成已经展现了"预测"的雏形。未来可以进一步：

• 训练一个模型，根据过去 30 天的信号，预测某个开源项目未来 90 天的 GitHub Star 增长
• 把这个预测结果作为"置信度加分项"纳入综合报告

12. 总结

last30days-skill 代表了一个重要的范式转变：

从"搜索引擎告诉你编辑想让你看到的"到"AI Agent 帮你汇总真实的人在讨论什么"

它的核心价值不在于"搜索技术"本身（这些都是成熟技术），而在于信号源的选择和对 AI Agent 需求的深刻理解。

关键要点回顾

1. 四层架构：触发层 → 工具定义层 → 信号采集层 → 综合层，每一层都针对 AI Agent 的特性优化
2. 10 个信号源：Reddit（深度）+ X（实时）+ YouTube（教程）+ HN（精英）+ Polymarket（预测）+ ... = 360° 视角
3. Grounded Summary：每条结论都带引用来源，解决 AI 幻觉问题
4. 可扩展架构：添加新平台只需实现统一接口，30 分钟完成
5. 生产级优化：速率限制、缓存、错误监控一应俱全

行动建议

如果你是这样的开发者，last30days 适合你：

• ✅ 需要频繁做技术调研（评估工具、跟踪趋势）
• ✅ 使用 Claude Code / Cursor / Copilot 等 AI 编程工具
• ✅ 希望 AI Agent 的答案基于"最近 30 天的真实讨论"而非"训练数据截止日期前的知识"

快速开始：

# 1 clone
git clone https://github.com/mvanhorn/last30days-skill.git ~/.claude/skills/last30days

# 2. 安装依赖
cd ~/.claude/skills/last30days
pip install -r requirements.txt

# 3. 配置 API Key（需要哪些配哪些）
export TWITTER_CONSUMER_KEY=xxx
export YOUTUBE_API_KEY=xxx

# 4. 在 Claude Code 中测试
/claude
> /last30days Headroom AI context compression

参考资源

• 项目 GitHub：https://github.com/mvanhorn/last30days-skill
• 中文版：https://github.com/Jesseovo/last30days-skill-cn
• 作者 Twitter：@mvanhorn_
• 相关项目：Headroom（上下文压缩）、claude-mem（跨会话记忆）

本文撰写于 2026 年 7 月，基于 last30days-skill v2.9.5。项目正在快速迭代中，建议关注 GitHub 获取最新动态。