如何為 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、發現新排名頁面，並在重要搜尋結果變化時提醒團隊。