如何为 AI Agent 构建 Google Search 监测器

AI Agent 如果能知道“什么变了”，会更有用。

普通搜索流程通常只回答某一刻的问题。Google Search monitor 则不同：它会定期检查重要 query，检测新的或变化的搜索结果，并在 AI Agent 生成答案或触发 workflow 前，提供最新 source signals。

要为 AI Agent 构建 Google Search monitor，通常需要四个部分：monitored queries、返回结构化 Google results 的 SERP API、保存历史结果的 storage layer，以及用于检测新 URL、排名变化或 domain 变化的 comparison function。

为什么 AI Agent 需要搜索监测？

AI Agent 通常会把搜索当作 source discovery layer。

简单 workflow 如下：

User question
→ Generate search query
→ Get SERP JSON
→ Select useful sources
→ Fetch pages
→ Generate grounded answer

这适合一次性问题。但很多商业场景是持续发生的：

• 监测竞争对手

• 追踪新的产品页

• 观察 pricing page 变化

• 发现新的 review 或 news coverage

• 跟进法规更新

• 更新 RAG source list

• 找到新进入目标 keyword 排名的页面

在这类场景中，Agent 不应每次都从零开始，而应知道哪些来源之前出现过、哪些是新的，以及哪些 domain 持续出现在同一个 query 下。

Google Search Monitor 应该追踪什么？

一个实用 monitor 不需要保存所有 SERP 字段。先从 Agent 真正能用的字段开始。

对 AI Agent 和 RAG pipeline 来说，link、domain、snippet 和 timestamp 特别重要。它们可以帮助 Agent 判断哪些页面值得抓取，哪些来源可能已过期。

基础架构

简单 Google Search monitor 可以这样设计：

Monitored query list
→ SERP API request
→ Normalize SERP JSON
→ Save current snapshot
→ Compare with previous snapshot
→ Send changes to AI agent

输出应该是一份干净的 change list：

{
  "query": "best CRM software for startups",
  "new_results": [
    {
      "position": 3,
      "title": "Best CRM Tools for Startup Teams",
      "link": "https://example.com/startup-crm",
      "domain": "example.com",
      "snippet": "Compare CRM platforms for early-stage teams..."
    }
  ]
}

这能让 Agent 收到聚焦的 update，而不是一整份原始 search page。

Python 示例：监测 Google Search Results

下面代码使用 provider-neutral SERP API 结构。你可以将 SERP_API_ENDPOINT 换成所选 provider 的 endpoint，并按需调整 authentication。

import os
import json
import time
from pathlib import Path
from urllib.parse import urlparse
from datetime import datetime, timezone

import requests


SERP_API_KEY = os.getenv("SERP_API_KEY")
SERP_API_ENDPOINT = os.getenv("SERP_API_ENDPOINT")
SNAPSHOT_FILE = Path("google_search_snapshot.json")


MONITORED_QUERIES = [
    {
        "query": "best CRM software for startups",
        "location": "United States",
        "language": "en",
        "device": "desktop"
    },
    {
        "query": "AI search API",
        "location": "United States",
        "language": "en",
        "device": "desktop"
    }
]


def get_domain(url: str | None) -> str:
    if not url:
        return ""
    parsed = urlparse(url)
    return parsed.netloc.replace("www.", "") if parsed.netloc else ""


def clean_text(value: str | None) -> str:
    if not value:
        return ""
    return " ".join(value.split())


def fetch_google_serp(query_config: dict) -> dict:
    params = {
        "engine": "google",
        "query": query_config["query"],
        "location": query_config["location"],
        "language": query_config["language"],
        "device": query_config["device"],
        "page": 1,
        "output": "json"
    }

    headers = {
        "Authorization": f"Bearer {SERP_API_KEY}"
    }

    response = requests.get(
        SERP_API_ENDPOINT,
        params=params,
        headers=headers,
        timeout=30
    )

    response.raise_for_status()
    return response.json()


def normalize_results(serp_json: dict, query_config: dict) -> list[dict]:
    organic_results = serp_json.get("organic_results", [])
    timestamp = datetime.now(timezone.utc).isoformat()

    normalized = []

    for index, item in enumerate(organic_results, start=1):
        link = item.get("link") or item.get("url")

        if not link:
            continue

        normalized.append({
            "query": query_config["query"],
            "location": query_config["location"],
            "language": query_config["language"],
            "device": query_config["device"],
            "position": item.get("position") or index,
            "title": clean_text(item.get("title")),
            "link": link,
            "domain": get_domain(link),
            "snippet": clean_text(item.get("snippet") or item.get("description")),
            "timestamp": timestamp
        })

    return normalized


def load_previous_snapshot() -> dict:
    if not SNAPSHOT_FILE.exists():
        return {}

    with open(SNAPSHOT_FILE, "r", encoding="utf-8") as file:
        return json.load(file)


def save_snapshot(snapshot: dict) -> None:
    with open(SNAPSHOT_FILE, "w", encoding="utf-8") as file:
        json.dump(snapshot, file, indent=2, ensure_ascii=False)


def detect_new_results(previous: list[dict], current: list[dict]) -> list[dict]:
    previous_links = {item["link"] for item in previous}
    return [item for item in current if item["link"] not in previous_links]


def run_monitor() -> list[dict]:
    previous_snapshot = load_previous_snapshot()
    new_snapshot = {}
    change_report = []

    for config in MONITORED_QUERIES:
        query_key = f'{config["query"]}|{config["location"]}|{config["language"]}|{config["device"]}'

        serp_json = fetch_google_serp(config)
        current_results = normalize_results(serp_json, config)
        previous_results = previous_snapshot.get(query_key, [])

        new_results = detect_new_results(previous_results, current_results)

        if new_results:
            change_report.append({
                "query": config["query"],
                "location": config["location"],
                "new_results": new_results
            })

        new_snapshot[query_key] = current_results
        time.sleep(1)

    save_snapshot(new_snapshot)
    return change_report


if __name__ == "__main__":
    changes = run_monitor()

    if not changes:
        print("No new search results detected.")
    else:
        print(json.dumps(changes, indent=2, ensure_ascii=False))

这段代码会检查一组 Google queries，保存当前 SERP snapshot，并回报新出现的 URLs。

正式环境中，可以用 cron、GitHub Actions、cloud function 或自己的 backend worker 定时执行。

如何把变化交给 AI Agent？

Change report 应该小而结构化。除非真的需要，不要把完整 raw SERP JSON 丢给模型。

更好的 Agent prompt 可以这样写：

You are monitoring Google Search results for important business queries.

Review the new search results below. Identify:
1. new competitor pages
2. new high-authority sources
3. pages that should be added to our RAG index
4. pages that need human review

Return a short summary and recommended actions.

然后附上 JSON change report。

这能让 Agent 更聚焦。它可以决定是否抓取页面、总结变化、提醒团队、更新数据库，或触发内容工作流。

如何选择 SERP API？

对 search monitoring 来说，API 选型应关注数据质量和可重复性，而不只是价格。

你可以用 SerpApi、Bright Data、Serper.dev、SearchAPI 或 Talordata 测试这类 workflow。真正重要的是 response 是否足够干净，能让 AI Agent、SEO monitor 或 RAG pipeline 直接使用。

例如，如果你只需要为 AI prototype 快速接入 Google Search API，Serper.dev 可能容易测试。如果 SERP data 是企业级 scraping stack 的一部分，Bright Data 值得比较。如果你已在 production 中使用 SerpApi schema，migration cost 需要考虑。如果你需要 structured SERP data、multi-engine coverage、geo-targeted results、JSON / HTML output 和成本敏感型 monitoring workflow，Talordata 值得与其他工具一起测试。获取1000次免费测试额度>>

最佳实践

先从小 query set 开始。10 个重要 query 搭配 2 个 location，通常比几百个未验证 keyword 更有价值。

每条 result 都要保存 metadata。Query、location、language、device、position 和 timestamp 会让数据之后仍然有用。

分开比较 URL 和 domain。既有 domain 的新 URL 可能代表内容更新；新 domain 可能代表新的竞争者或数据来源。

不要让 Agent 抓取每个页面。先用 SERP title 和 snippet 进行筛选。

Raw JSON 可用于 debug，但分析时应保存 normalized rows。

根据业务风险设置监测频率。Pricing page 或 competitor page 可以每天检查；较稳定的 topic monitoring 可以每周检查。

FAQ

什么是 AI Agent 的 Google Search monitor？

Google Search monitor 是一个会定期检查指定 Google queries、保存 SERP results、检测变化，并把有用更新交给 AI Agent 的系统。它能让 Agent 使用最新 search signals，而不是只依赖一次性搜索结果。

应该从 SERP JSON 保存哪些数据？

建议保存 query、location、language、device、timestamp、position、title、link、domain 和 snippet。这些字段足以支持大多数 SEO monitoring、AI source discovery 和 RAG update workflow。

AI search monitor 应该多久执行一次？

取决于场景。Competitive keywords、pricing pages 或 news-sensitive topics 可以每天监测；较稳定的 topic 可以每周检查。建议先低频开始，只有当变化有价值时再提高频率。

AI Agent 应该阅读搜索结果中的每个页面吗？

不应该。Agent 应先用 title、snippet、domain、result type 和 position 筛选来源，只抓取最可能有用、最新且相关的页面。

结语

Google Search monitor 可以把 SERP data 转化为 AI Agent 的持续信号。

你不需要让 Agent 每次都从零开始搜索，而是可以提供一份结构化 feed，包含 new URLs、ranking changes 和 source updates。

核心流程很简单：定义 queries、采集 SERP JSON、normalize results、比较 snapshots，然后把有用变化交给 Agent。

一旦这个流程建立起来，AI Agent 就可以监测竞争对手、更新 RAG sources、发现新排名页面，并在重要搜索结果变化时提醒团队。