OpenClaw AI搜索技能：让 Agent 实时看见 Google AI Mode 数据层

前言

Google AI Mode 正在以肉眼可见的速度渗透全球搜索市场。如果你今天用 OpenClaw 构建 AI Agent，而你的 Agent 依赖 Google 搜索数据做决策，那么你可能正在面对一个隐蔽但实质严重的问题：你的 Agent 看不见 AI Overview。

AI Overview 是 Google AI Mode 下动态生成并注入到 SERP 页面顶部的 AI 摘要内容。它不在传统 DOM 结构里，普通 HTTP 爬虫拿不到，即便无头浏览器也面临渲染时序和反检测的双重挑战。对于依赖 SERP 数据的 Agent 工作流来说，这是一个系统性的数据盲区。

本文记录如何通过 Pangolinfo AI Mode API，为 OpenClaw Agent 配置一个生产级别的 AI搜索技能，实现对 Google AI Overview 内容的实时、稳定采集。

---

技术原理详解

为什么 AI Mode 数据无法用传统爬虫获取？

Google AI Mode（通过 udm=50 参数触发）在收到搜索请求后，会在后台执行一次 LLM 推理，将生成结果以 JSON-LD 格式内联注入 SERP 页面。这个过程：

1. 时序依赖：AI 生成有延迟，页面 DOM 的 AI Overview 节点在初始渲染后才填充
2. 动态注入：内容通过 JavaScript 动态插入，静态 HTML 爬虫不可见
3. 反检测强化：Google 对 AI Mode 页面的反机器人保护比普通搜索更严格
4. 多轮状态：支持 param 多轮对话，意味着每次追问都触发新的推理和页面更新

自建抓取方案需要维护：住宅代理池 + 指纹轮换 + 渲染等待策略 + 反检测追踪，这是一套持续消耗工程资源的基础设施。

Pangolinfo AI Mode API 的架构方案

AI Mode API 完整调用流程：从 OpenClaw Skill 发起请求到获取结构化 JSON 响应

Pangolinfo 将上述复杂性封装在服务端，通过以下架构实现稳定的 AI Mode 数据采集：

OpenClaw Agent
    ↓ POST /api/v2/scrape
Pangolinfo API Gateway
    ↓ 任务调度
浏览器渲染集群（住宅IP池 + 指纹管理）
    ↓ 渲染 + 等待 AI 内容加载
AI Overview 内容提取
    ↓ 结构化 JSON 解析
OpenClaw Agent ← JSON 响应

开发者的接入点只有一个：POST 请求。复杂性全在服务端。

---

完整代码实现

基础版：单次查询

import requests
import json
from typing import Optional


def query_google_ai_mode(
    query: str,
    api_key: str,
    follow_up: Optional[list[str]] = None,
    screenshot: bool = False
) -> dict:
    """
    查询 Google AI Mode，返回结构化 AI Overview 数据
    
    Args:
        query: 搜索关键词
        api_key: Pangolinfo API Key
        follow_up: 多轮对话提示词（最多5条，超出5条接口响应效率降低）
        screenshot: 是否返回页面截图 URL
    
    Returns:
        {
            "has_ai_overview": bool,
            "ai_content": list[str],
            "references": list[dict],
            "screenshot_url": str,
            "task_id": str
        }
    """
    
    # Google AI Mode URL 构造（udm=50 是必要参数）
    google_url = (
        f"https://www.google.com/search"
        f"?num=10&udm=50&q={requests.utils.quote(query)}"
    )
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    
    payload = {
        "url": google_url,
        "parserName": "googleAISearch",  # 固定值，选择 AI Mode 解析器
        "screenshot": screenshot
    }
    
    # 多轮对话：限制最多5条，避免响应延迟
    if follow_up:
        payload["param"] = follow_up[:5]
    
    response = requests.post(
        "https://scrapeapi.pangolinfo.com/api/v2/scrape",
        headers=headers,
        json=payload,
        timeout=30
    )
    response.raise_for_status()
    
    api_result = response.json()
    
    # 检查 API 级别错误
    if api_result.get("code") != 0:
        raise ValueError(
            f"Pangolinfo API 错误 [{api_result.get('code')}]: "
            f"{api_result.get('message')}"
        )
    
    data = api_result["data"]
    
    # 解析 AI Overview 内容
    parsed = {
        "has_ai_overview": bool(data.get("ai_overview", 0)),
        "ai_content": [],
        "references": [],
        "screenshot_url": data.get("screenshot", ""),
        "task_id": data.get("taskId", "")
    }
    
    items = data.get("json", {}).get("items", [])
    for item in items:
        if item.get("type") != "ai_overview":
            continue
            
        # 提取 AI 生成的文本内容
        for sub_item in item.get("items", []):
            if sub_item.get("type") == "ai_overview_elem":
                parsed["ai_content"].extend(sub_item.get("content", []))
        
        # 提取引用来源
        for ref in item.get("references", []):
            parsed["references"].append({
                "title": ref.get("title", ""),
                "url": ref.get("url", ""),
                "domain": ref.get("domain", "")
            })
    
    return parsed

进阶版：封装为 OpenClaw Skill 类

import asyncio
import aiohttp
import time
from dataclasses import dataclass, field


@dataclass
class AIOverviewResult:
    """AI Overview 查询结果数据类"""
    query: str
    has_ai_overview: bool
    ai_content: list[str] = field(default_factory=list)
    references: list[dict] = field(default_factory=list)
    screenshot_url: str = ""
    task_id: str = ""
    credits_consumed: int = 0
    
    def to_agent_context(self) -> str:
        """将结果格式化为 OpenClaw Agent 可直接使用的上下文字符串"""
        if not self.has_ai_overview:
            return f"查询 '{self.query}' 没有 AI Overview 结果。"
        
        context = f"## Google AI Overview: {self.query}\n\n"
        for i, content in enumerate(self.ai_content, 1):
            context += f"{i}. {content}\n"
        
        if self.references:
            context += "\n### 参考来源\n"
            for ref in self.references:
                context += f"- [{ref['domain']}] {ref['title']}: {ref['url']}\n"
        
        return context


class PangolinAISearchSkill:
    """
    OpenClaw AI搜索技能
    基于 Pangolinfo AI Mode API 实现 Google AI Overview 实时采集
    
    积点消耗：每次成功获取 AI Overview = 2 积点
    """
    
    API_URL = "https://scrapeapi.pangolinfo.com/api/v2/scrape"
    GOOGLE_AI_MODE_URL = "https://www.google.com/search?num=10&udm=50&q={query}"
    
    def __init__(self, api_key: str, rate_limit_rps: float = 2.0):
        """
        Args:
            api_key: Pangolinfo API Key（从控制台获取）
            rate_limit_rps: 每秒最大请求数（默认2，避免触发限流）
        """
        self.api_key = api_key
        self.rate_limit_rps = rate_limit_rps
        self._last_request_time = 0.0
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}"
        }
    
    def _rate_limit(self):
        """简单速率控制"""
        elapsed = time.time() - self._last_request_time
        min_interval = 1.0 / self.rate_limit_rps
        if elapsed < min_interval:
            time.sleep(min_interval - elapsed)
        self._last_request_time = time.time()
    
    def search(
        self,
        query: str,
        follow_up: list[str] = None,
        screenshot: bool = False
    ) -> AIOverviewResult:
        """同步搜索接口"""
        self._rate_limit()
        
        import requests
        
        search_url = self.GOOGLE_AI_MODE_URL.format(
            query=requests.utils.quote(query)
        )
        payload = {
            "url": search_url,
            "parserName": "googleAISearch",
            "screenshot": screenshot
        }
        if follow_up:
            payload["param"] = follow_up[:5]
        
        try:
            resp = requests.post(
                self.API_URL,
                headers=self.headers,
                json=payload,
                timeout=30
            )
            resp.raise_for_status()
            return self._parse_to_result(query, resp.json())
        except Exception as e:
            raise RuntimeError(f"AI 搜索技能调用失败: {e}") from e
    
    async def search_async(
        self,
        query: str,
        session: aiohttp.ClientSession,
        follow_up: list[str] = None,
        screenshot: bool = False
    ) -> AIOverviewResult:
        """异步搜索接口（批量查询推荐）"""
        from urllib.parse import quote
        
        search_url = self.GOOGLE_AI_MODE_URL.format(query=quote(query))
        payload = {
            "url": search_url,
            "parserName": "googleAISearch",
            "screenshot": screenshot
        }
        if follow_up:
            payload["param"] = follow_up[:5]
        
        async with session.post(
            self.API_URL,
            headers=self.headers,
            json=payload
        ) as resp:
            data = await resp.json()
            return self._parse_to_result(query, data)
    
    def batch_search(
        self,
        queries: list[str],
        max_concurrent: int = 5
    ) -> list[AIOverviewResult]:
        """批量异步搜索（推荐大规模使用）"""
        
        async def _batch():
            connector = aiohttp.TCPConnector(limit=max_concurrent)
            async with aiohttp.ClientSession(connector=connector) as session:
                tasks = [
                    self.search_async(q, session) 
                    for q in queries
                ]
                return await asyncio.gather(*tasks, return_exceptions=True)
        
        return asyncio.run(_batch())
    
    def _parse_to_result(self, query: str, api_data: dict) -> AIOverviewResult:
        if api_data.get("code") != 0:
            raise ValueError(f"API 错误: {api_data.get('message')}")
        
        data = api_data["data"]
        result = AIOverviewResult(
            query=query,
            has_ai_overview=bool(data.get("ai_overview", 0)),
            screenshot_url=data.get("screenshot", ""),
            task_id=data.get("taskId", ""),
            credits_consumed=2 if data.get("ai_overview") else 1
        )
        
        for item in data.get("json", {}).get("items", []):
            if item.get("type") == "ai_overview":
                for sub in item.get("items", []):
                    if sub.get("type") == "ai_overview_elem":
                        result.ai_content.extend(sub.get("content", []))
                result.references = [
                    {"title": r.get("title"), "url": r.get("url"), "domain": r.get("domain")}
                    for r in item.get("references", [])
                ]
        
        return result


# ==================== 使用示例 ====================

if __name__ == "__main__":
    skill = PangolinAISearchSkill(api_key="YOUR_PANGOLINFO_API_KEY")
    
    # 1. 单次查询
    result = skill.search("最好的 Python 数据爬虫库 2026")
    print(result.to_agent_context())
    print(f"消耗积点: {result.credits_consumed}")
    
    # 2. 带追问的多轮对话
    result2 = skill.search(
        query="asyncio vs threading",
        follow_up=["哪个更适合 I/O 密集型任务？", "给我看一个对比示例"]
    )
    
    # 3. 批量查询（生产推荐）
    keywords = [
        "OpenClaw Agent 配置",
        "Google AI Mode API",
        "Pangolinfo 评测",
        "电商数据采集方案 2026"
    ]
    batch_results = skill.batch_search(keywords, max_concurrent=3)
    
    for r in batch_results:
        if isinstance(r, Exception):
            print(f"查询失败: {r}")
        else:
            print(f"[{r.query}] AI Overview: {r.has_ai_overview}")

---

常见问题与解决方案

Q: 请求返回了结果但 ai_overview 字段为 0？

A: 并非所有查询都有 AI Overview。Google 对 AI Mode 摘要的触发有内部策略，信息型查询（How/What/Why）触发率较高，导航型和纯品牌词触发率低。这种情况属正常，不是 API 问题。

Q: param 超过 5 条怎么办？

A: API 文档明确指出超过 5 条会导致响应效率降低。建议把长对话链拆成多次独立请求，每次最多携带 3-4 条上下文，保持响应速度。

Q: 积点消耗验证：如何知道是否消耗了 2 积点？

A: 响应 data.ai_overview 字段为 1 时表示成功提取 AI Overview，此时消耗 2 积点。为 0 时（无 AI Overview）消耗标准积点。

Q: 如何在 OpenClaw 配置文件中注册这个 Skill？

A: 将 PangolinAISearchSkill 类按照 OpenClaw Skill 接口规范封装后，在 config.yaml 的 skills 字段注册，并在 Agent Prompt 中描述技能的触发条件即可。

---

性能优化建议

• 并发控制：批量请求时推荐 max_concurrent=5，过高会导致超时集中爆发
• 缓存策略：相同关键词在短时间内重复查询收益递减，建议对热词设置 1 小时缓存
• 失败重试：网络层加指数退避重试（最多 3 次），timeout 建议设为 30 秒
• 优先级队列：区分高优先级实时查询和低优先级批量任务，分开调用通道

---

总结

AI Mode 数据采集的核心难点是架构层面的，不是反爬层面的。Pangolinfo AI Mode API 通过托管浏览器集群和专用解析器，把复杂性留在服务端，让 OpenClaw Skill 的接入保持极简。配合本文提供的生产级代码，5 分钟可以完成基础集成，30 分钟可以搭完带缓存和并发控制的完整 Skill。