AI Agent实时搜索Skill：聚合Reddit、X与YouTube信息

先看结论：这类"搜索引擎 Skill"不是给你本地跑个大模型，而是把 Reddit、X、YouTube 这些平台的搜索能力封装成 Agent 可以自动调用的工具集。装上之后，你的 Agent 能在拿到任务时主动去各平台抓实时内容，再回来汇总成结构化结果。整个过程不需要 GPU、不需要高显存，主要消耗的是各平台的 API 配额和网络请求时间。

这篇是"每天认识一个 AI Agent Skill"系列里非常实用的一篇。我建议所有做 AI Agent 开发、舆情分析、内容选题、竞品调研的人都收藏一下。本文会从 Skill 的核心能力、目录结构、环境准备、安装方式、功能测试、接口调用、批量任务到常见问题排查，完整带你过一遍。

1. 核心能力速览

从能力边界看，这类 Skill 解决的是大模型"训练数据截止时间"造成的最大短板：实时信息获取。大模型再强，它也不会知道你昨天在某个平台发生了什么讨论。搜索引擎类 Skill 就是补上这块拼图的。

2. 适用场景与使用边界

2.1 适合谁

• Agent 开发者：想给自己的 Agent 增加实时信息检索能力，而不是让它只靠记忆回答。
• 内容运营和小编：需要快速知道某个话题在 Reddit、X、YouTube 上分别是什么热度、什么观点。
• 产品和市场研究人员：做竞品分析、用户反馈收集、行业趋势判断。
• 个人效率工具爱好者：想做一个"今天全网都在聊什么"的自动汇总脚本。

2.2 能解决什么问题

• 跨平台信息聚合：一个查询词，同时搜 Reddit 帖子、X 推文、YouTube 视频，一次拿全。
• 降低信息获取成本：不用手动开三个平台挨个搜索，Agent 直接帮你检索并返回摘要。
• 结构化输出：搜索结果不再是网页碎片，而是带标题、链接、发布时间、摘要的结构化 JSON。
• 定时批量追踪：对同一组关键词定期执行，形成趋势曲线。

2.3 不适合什么场景

• 不适合做大规模爬虫抓取，这会直接违反平台条款。
• 不适合对数据权威性要求极高的金融、医疗决策，搜索结果噪声大，必须人工复核。
• 不适合需要完整全文内容的场景，搜索 API 通常只返回摘要和元数据，不返回正文。

2.4 合规与安全边界

这里要专门提醒：Reddit、X、YouTube 都有自己的 API 使用条款和速率限制。个人学习和测试没问题，但如果是商业项目，必须确认你的使用方式在平台允许范围内。涉及用户生成内容时，要注意隐私保护，不要下载和转售他人数据。涉及版权的视频、图片、文章，不要直接复制用于商业用途。文章、评论、推文的作者信息，在聚合展示时要保留来源链接。

3. 环境准备与前置条件

这类 Skill 的部署不算复杂，但前置条件比一般 Web 项目多一点，主要是 API Key 和网络可达性。

3.1 运行环境

• Python 3.10 或更高版本。
• pip 包管理器。
• 一个支持 Agent Skill 的框架，例如 Claude Code、OpenAI Agent SDK，或者你自己写的 Agent 编排代码。
• 建议使用虚拟环境，避免依赖冲突。

3.2 API Key 准备

三个平台的 API Key 获取方式不同，在使用前必须逐个申请：

申请时注意，X 的免费层级调用次数极其有限，YouTube 的配额也是按天计算的。Reddit 的 API 相对宽松，但依然有每分钟 / 每小时的请求上限。

如果你不想分别对接三个平台，也可以选择封装好的第三方搜索 API（如 Tavily、Brave Search、Serper），它们通常已经聚合了多平台结果，只需要一个 Key。缺点是平台覆盖面以网页为主，对 Reddit、X、YouTube 的专项搜索能力不一定比官方 API 细。

以官方 API 为例说明环境变量配置，建议不要把 Key 写死在代码里：

# credentials.env 示例 REDDIT_CLIENT_ID=your_reddit_client_id REDDIT_CLIENT_SECRET=your_reddit_client_secret REDDIT_USER_AGENT="your-app-name/1.0 by your-username" X_BEARER_TOKEN=your_x_bearer_token YOUTUBE_API_KEY=your_youtube_api_key

加载到当前 shell：

set -a source credentials.env set +a

3.3 网络可达性

调用 Reddit、X、YouTube 的 API，本质是 HTTPS 请求。如果你所在网络环境访问不了这些 API 域名，Skill 跑不通。排查顺序是：

1. 确认 API 域名可以解析。
2. 确认 HTTPS 端口 443 可以连通。
3. 确认你的请求没有被防火墙策略拦截。

如果 API 端点不可达，先解决网络策略问题再继续测试。

4. 安装部署与启动方式

不同框架对 Skill 的目录规范有差异，但大体结构相通。下面给出一套通用目录结构，实际使用时按你的框架文档调整。

4.1 Skill 目录结构

web-search-agent/ ├── SKILL.md ├── requirements.txt ├── config.yaml ├── src/ │ ├── __init__.py │ ├── search_reddit.py │ ├── search_x.py │ ├── search_youtube.py │ └── aggregator.py ├── credentials.env └── examples/ └── demo_query.json

• SKILL.md：Skill 的核心描述文件，Agent 靠它来理解这个技能什么时候用、怎么用。
• config.yaml：配置各平台的搜索参数、速率限制、缓存策略。
• src/：实际执行搜索的代码模块。
• credentials.env：存放 API Key，注意不要提交到 Git。

4.2 编写 SKILL.md

SKILL.md是 Agent 理解 Skill 用法的关键。不同框架的格式有差异，下面是一个通用模板：

--- name: platform-search description: 搜索 Reddit、X、YouTube 的实时内容，用于获取最新讨论、热点、视频信息。当用户需要了解某话题在社交平台上的实时动态时使用。 --- # Platform Search Skill ## 功能 - search_reddit(query, limit=10) - search_x(query, max_results=10) - search_youtube(query, max_results=10) - search_all(query, limit=10) ## 使用示例 用户提问：今天 AI Agent 话题在 Reddit 上有什么热门讨论？ 调用 search_all("AI Agent", limit=5)，优先返回 Reddit 结果。 ## 注意事项 - 搜索结果需要保留原始链接。 - 如果某个平台返回空结果，不要报错，继续搜索其他平台。 - 聚合结果按相关度和发布时间排序。

4.3 安装依赖

cd web-search-agent python -m venv .venv source .venv/bin/activate pip install -r requirements.txt

requirements.txt内容示例：

requests>=2.31.0 PyYAML>=6.0 python-dotenv>=1.0.0

4.4 配置 config.yaml

platforms: reddit: enabled: true default_limit: 10 time_filter: week sort: relevance x: enabled: true default_limit: 10 recency: "7d" youtube: enabled: true default_limit: 10 order: relevance cache: enabled: true ttl_seconds: 3600 cache_dir: ./cache output: format: json include_links: true

4.5 注册到 Agent 框架

把 Skill 目录放入框架指定的 skills 目录，例如：

# Claude Code 风格 cp -r web-search-agent ~/.claude/skills/platform-search/ # OpenAI Agent SDK 风格 cp -r web-search-agent ~/.agents/skills/platform-search/

具体路径以你的框架文档为准。注册完成后，Agent 就能在合适的时候自动调用这个 Skill 了。想验证 Skill 是否被识别，可以重启 Agent 会话后问一句"你能搜索 Reddit 吗"，看模型回答里是否提到了这个能力。

5. 功能测试与效果验证

Skill 装好之后，先别急着接业务，按下面的测试维度逐个验证。

5.1 Reddit 搜索测试

测试目的：确认 Reddit OAuth2 鉴权成功，能按关键词返回帖子标题、链接、分数和摘要。

使用 Reddit API 的搜索端点：

import requests client_id = "your_client_id" client_secret = "your_client_secret" user_agent = "your-app-name/1.0 by your-username" # 1. 获取 access token auth = requests.auth.HTTPBasicAuth(client_id, client_secret) data = { "grant_type": "client_credentials", "username": "your_username", "password": "your_password" } headers = {"User-Agent": user_agent} resp = requests.post( "https://www.reddit.com/api/v1/access_token", auth=auth, data=data, headers=headers ) token = resp.json()["access_token"] # 2. 搜索帖子 search_headers = { "User-Agent": user_agent, "Authorization": f"Bearer {token}" } params = { "q": "AI Agent", "limit": 5, "sort": "relevance", "time_filter": "week" } search_resp = requests.get( "https://oauth.reddit.com/search", params=params, headers=search_headers, timeout=15 ) posts = search_resp.json()["data"]["children"] for post in posts: p = post["data"] print(f"[{p['score']}] {p['title']} (r/{p['subreddit']})") print(f" https://www.reddit.com{p['permalink']}")

预期结果：控制台输出 5 条帖子标题、分数和链接。

判断标准：

• 返回 200 且children列表非空。
• 每个帖子有title、permalink、score字段。

常见失败原因：

• 401 Unauthorized：token 获取失败，检查 client_id / client_secret。
• 403 Forbidden：User-Agent 不合规，Reddit 要求自定义且不能是 "python"。
• 空结果：关键词太宽泛或time_filter设置过短，换个词再试。

5.2 X 搜索测试

测试目的：确认 X API v2 的 Bearer Token 有效，能按关键词返回推文。

X API v2 的 recent search 端点只返回最近 7 天的推文（免费层级），示例：

import requests bearer_token = "your_bearer_token" url = "https://api.x.com/2/tweets/search/recent" headers = { "Authorization": f"Bearer {bearer_token}" } params = { "query": "AI Agent -is:retweet lang:en", "max_results": 10, "tweet.fields": "created_at,public_metrics,author_id" } resp = requests.get(url, headers=headers, params=params, timeout=15) if resp.status_code == 200: data = resp.json() # 注意：x 的 API 没有标准分页，使用 next_token 翻页 tweets = data.get("data", []) meta = data.get("meta", {}) print(f"获取 {len(tweets)} 条推文") print(f"下一页 token: {meta.get('next_token', '无')}") else: print(f"错误 {resp.status_code}: {resp.text}")

预期结果：返回 10 条推文 JSON。

判断标准：

• 返回 200，data列表有内容。
• tweet 对象含id、text、created_at。

常见失败原因：

• 401 Unauthorized：Bearer Token 无效或过期。
• 429 Too Many Requests：免费层级速率限制太严格，建议降低请求频率。

注意区分：X 的搜索结果默认可能包含大量无关推文，需要用-is:retweet排除转推，lang:en限制语言，必要时加from:过滤指定用户。

5.3 YouTube 搜索测试

测试目的：确认 YouTube Data API v3 能按关键词返回视频列表，关键是能正确解析视频 ID 并拼出链接。

import requests api_key = "your_youtube_api_key" url = "https://www.googleapis.com/youtube/v3/search" params = { "part": "snippet", "q": "AI Agent 教程", "maxResults": 5, "type": "video", "order": "relevance", "key": api_key } resp = requests.get(url, params=params, timeout=15) if resp.status_code == 200: items = resp.json().get("items", []) for item in items: video_id = item["id"]["videoId"] title = item["snippet"]["title"] channel = item["snippet"]["channelTitle"] url_link = f"https://www.youtube.com/watch?v={video_id}" print(f"[{channel}] {title}") print(f" {url_link}") else: print(f"错误 {resp.status_code}: {resp.text}")

预期结果：输出视频标题、频道名和可点击的视频链接。

判断标准：

• 返回 200，items非空。
• 每个 item 能正确提取videoId并生成链接。

常见失败原因：

• 403 quotaExceeded：每日配额用完，只有第二天重置。
• 400 invalidParameter：type传错，必须是video、channel、playlist之一。

5.4 聚合搜索测试

测试目的：验证三个平台的搜索结果能否合并成一份统一结构。这一步是 Skill 的核心价值所在，因为 Agent 最终要的是"一个话题全网怎么看"，而不是三块割裂的数据。

def search_all(query, limit=10): results = { "query": query, "platforms": { "reddit": search_reddit(query, limit), "x": search_x(query, limit), "youtube": search_youtube(query, limit) }, "total_count": 0, "generated_at": None } for platform, items in results["platforms"].items(): results["total_count"] += len(items) return results

预期结果：返回一个包含三平台结果列表的 JSON，每条结果都带source、title、link、published_time、summary。

6. 接口 API 与批量任务

Skill 本身不一定要暴露 HTTP 接口，但在实际使用中，你大概率会把它集成到内部工具平台或自动化流程里。有两种常见做法。

6.1 作为 Agent 可调用工具

第一种方式最简单：让 Agent 在推理时直接调用 Skill 提供的方法。你不需要自己写 HTTP 封装，只要确认 Skill 的方法能被 Agent 的工具调用协议识别即可。查询词由用户自然语言提供，Agent 负责拆解、调用、汇总。

6.2 包装成 HTTP 接口

第二种方式是把 Skill 包装成一个轻量 HTTP 服务，方便其他系统调用。下面是一个基于 Flask 的通用模板：

from flask import Flask, request, jsonify from src.search_reddit import search_reddit from src.search_x import search_x from src.search_youtube import search_youtube app = Flask(__name__) @app.route("/api/search", methods=["POST"]) def search(): data = request.get_json() query = data.get("query") limit = data.get("limit", 10) platforms = data.get("platforms", ["reddit", "x", "youtube"]) results = {"query": query, "platforms": {}} if "reddit" in platforms: results["platforms"]["reddit"] = search_reddit(query, limit) if "x" in platforms: results["platforms"]["x"] = search_x(query, limit) if "youtube" in platforms: results["platforms"]["youtube"] = search_youtube(query, limit) return jsonify(results) if __name__ == "__main__": app.run(host="127.0.0.1", port=8000)

启动服务：

python app.py

调用测试：

curl -X POST http://127.0.0.1:8000/api/search \ -H "Content-Type: application/json" \ -d '{"query": "AI Agent", "limit": 5, "platforms": ["reddit", "youtube"]}'

用 Python 调用：

import requests url = "http://127.0.0.1:8000/api/search" payload = { "query": "AI Agent", "limit": 5, "platforms": ["reddit", "x", "youtube"] } resp = requests.post(url, json=payload, timeout=30) data = resp.json() print(data)

6.3 批量任务与队列设计

批量搜索的核心问题不是循环调用，而是速率限制。常见的做法是：

• 把查询词列表读入任务队列。
• 每个查询词按平台限速间隔执行。
• 结果落盘为独立 JSON 文件。
• 失败的任务记录日志并重试。

一个参考实现：

import time import json import os from src.aggregator import search_all queries = [ "AI Agent", "Agent Skill", "Search Engine Agent", "LLM Tools" ] output_dir = "./outputs" os.makedirs(output_dir, exist_ok=True) for i, query in enumerate(queries): print(f"[{i + 1}/{len(queries)}] 搜索: {query}") try: result = search_all(query, limit=5) file_path = os.path.join(output_dir, f"result_{i + 1}.json") with open(file_path, "w", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=2) except Exception as e: print(f"失败: {query}, 原因: {e}") # 控制请求频率，防止被限流 time.sleep(5)

批量任务最容易踩的坑是"查着查着就 429"了。稳妥的做法是每个查询之间至少间隔 3 到 5 秒，并实现指数退避重试：

import time import requests def request_with_retry(func, max_retries=3): for attempt in range(max_retries): try: return func() except requests.exceptions.HTTPError as e: if e.response.status_code == 429: wait_time = 2 ** attempt * 5 print(f"触发限速，等待 {wait_time} 秒后重试") time.sleep(wait_time) else: raise

7. 资源占用与性能观察

这类 Skill 和本地大模型类的资源占用完全不同，重点观察三个维度。

7.1 本地资源占用

Skill 本身只是 Python 脚本和网络请求，不加载模型。内存占用通常只有几百 MB，CPU 占用集中在 JSON 解析和结果聚合阶段。唯一要注意的是，如果你在同一个 Agent 环境里同时运行多个 Skill，每个 Skill 的依赖包可能互相影响，建议用虚拟环境隔离。

7.2 外部 API 配额消耗

这是这类 Skill 真正要盯的资源。三种主要消耗：

如果频繁触发 429，先做的不是加硬重试，而是检查自己的请求频率是否合理。建议在代码里加请求计数和配额统计，把每个 Key 的日消耗量打出来，方便预估什么时候会撞到上限。

7.3 性能观察方法

• 网络请求耗时：三个平台接口内部耗时差异较大，一般几十到几百毫秒都有，主要取决于网络状况。
• 聚合耗时：主要在 JSON 解析和去重，通常小于 100ms。
• 批量任务耗时：主要取决于查询词数量和限速间隔，查询词越多，整体耗时越长，因为速率限制是硬性的。

想降低消耗，最有效的办法是加缓存。同一个查询词在短时间内重复搜索，结果差异不大。建议对查询词加一个 1 小时左右的缓存：

import json import os import time def get_cached_or_search(query, ttl=3600): cache_key = hashlib.md5(query.encode()).hexdigest() cache_file = f"./cache/{cache_key}.json" if os.path.exists(cache_file): with open(cache_file, "r", encoding="utf-8") as f: data = json.load(f) if time.time() - data["cached_at"] < ttl: return data["result"] result = search_all(query, limit=5) os.makedirs("./cache", exist_ok=True) with open(cache_file, "w", encoding="utf-8") as f: json.dump({ "cached_at": time.time(), "result": result }, f, ensure_ascii=False, indent=2) return result

这样对重复查询词能省掉大量 API 配额。

8. 常见问题与排查方法

排查这类 Skill 问题有两条主线：第一是本地环境，确认 Python 依赖和配置文件加载正常；第二是 API 调用，确认 Key 有效、配额足够、请求格式正确。网络类问题优先用 curl 直接测接口，排除框架干扰。

9. 最佳实践与使用建议

这部分是工程落地时最容易踩坑的地方，提前做好能省很多事。

9.1 第一次先小参数测试

不要一上来就批量跑一百个查询。先用 limit=3、单平台、单查询跑通整个链路，确认 API Key 有效、结果能解析、Agent 能正确引用结果，再逐步加大参数。

9.2 保留一套最小可运行配置

把 SKILL.md、config.yaml、credentials.env.example 这三个文件作为最小配置集。新增功能时先备份这套基础配置，出问题能快速回滚。

9.3 模型文件、输入素材、输出结果分目录管理

批量任务的查询词列表放inputs/，结果 JSON 放outputs/，中间缓存放cache/，不要让脚本在所有目录乱写。目录清晰之后，定位问题会快很多。

9.4 批量任务必须加日志和失败重试

批量任务跑几十分钟，中途网络波动是必然的。给每个查询任务加一个状态：pending / running / success / failed。失败的任务单独记录到failed.txt，跑完后重试一遍。

9.5 接口服务要限制访问范围

如果包装成了 HTTP 接口，不要直接监听0.0.0.0暴露到公网。最好只监听127.0.0.1，或者加上简单的 API Token 校验。这类接口消耗的是你的 API 配额，被刷了会产生费用。

9.6 涉及人脸、声音、版权素材时必须确认授权

虽然这个 Skill 主要搜文字和视频元数据，但如果你把搜索结果用于内容生产、二次创作、商业报告，必须注意：视频缩略图、频道标志、用户头像、评论内容，都可能涉及肖像权和版权。聚合展示时保留来源链接，不直接搬运原图、不绕过平台内嵌播放，都是基本要求。

9.7 发布或商用前要做效果复核

搜索结果的准确性和时效性受平台算法影响很大。同一个关键词，今天搜和明天搜，结果差异可能非常大。商业场景下，建议对 Skill 输出的结果做抽样人工复核，确认没有错误归因和断章取义。

10. 总结与下一步

这个 Skill 最值得尝试的点，是它用一个很小的工程成本，让 Agent 获得了"实时信息检索"的能力，整个部署不需要 GPU，也不需要写复杂的模型推理代码，核心工作就是配好 API Key、写好聚合逻辑。

拿到这个 Skill，最先验证的应该是 Reddit 和 YouTube 两条链路，因为这两个平台对新用户的 API 门槛相对友好，能快速看到效果。X 平台受免费层级配额限制，不要指望大量测试。

最容易踩的坑有三个：一是 API Key 环境变量没加载导致 401，二是批量任务不考虑速率限制导致 429，三是 X 和 YouTube 的配额被一次性跑光。想进一步扩展方向，可以考虑加入 DuckDuckGo 或 GitHub 搜索，形成覆盖网页、社交平台、代码仓库的完整搜索矩阵。

建议收藏备用，特别是如果你正在做一个需要实时信息输入的 Agent，这套 Skill 的接入思路可以直接复用。