关键词研究的第一步往往不是买昂贵工具，而是直接从Google自身的建议系统中提取数据。本文将系统讲解如何抓取Google People Also Ask和Autocomplete用于关键词研究——这是SEO开发者和数据工程师构建关键词工具时最实用的技术路径之一。我们会覆盖三个免费数据源、种子词扩展策略、代理防封方案，以及完整的Python实现代码。
三大免费关键词数据源及其搜索意图映射
Google在搜索过程中会生成三种结构化建议数据，每种对应不同的搜索意图阶段：
1. Autocomplete建议（suggestqueries.google.com）
Autocomplete是用户在搜索框输入时实时弹出的建议词。它的端点格式为：
https://suggestqueries.google.com/complete/search?client=chrome&q=你的关键词返回JSON数组，第二个元素是建议词列表。Autocomplete词反映早期探索意图——用户还在搜索框打字阶段，意图尚未完全明确。例如输入"running shoes"，建议词可能包含"running shoes for flat feet"、"running shoes on sale"，分别对应信息型和交易型意图。
2. People Also Ask（PAA）问题
PAA是搜索结果页面中以手风琴式折叠展示的问答模块。每个PAA问题对应信息型意图，非常适合FAQ页面和内容选题。PAA的独特之处在于它是递归的——点击展开一个问题后，会动态加载新的相关问题，形成无限扩展链。
3. Related Searches（相关搜索）
页面底部的相关搜索词反映搜索细化意图——用户已完成一次搜索但未完全满足，准备调整查询方向。这些词适合作为内容集群的主题锚点。
| 数据源 | 意图类型 | 数据结构 | 获取难度 |
|---|---|---|---|
| Autocomplete | 探索/交易 | JSON数组 | 低（HTTP请求） |
| People Also Ask | 信息型 | 嵌套DOM | 中（需渲染） |
| Related Searches | 细化/导航 | DOM列表 | 低（HTTP+解析） |
种子词到长尾词的扩展策略
单个种子词通过Autocomplete只能获得约10个建议。要生成数百个长尾词，需要组合两种策略：
a-z字母前缀扩展
对种子词逐一添加a到z的前缀，每次请求Autocomplete端点。例如"running shoes a"、"running shoes b"……理论上一个种子词可产生约260个候选词（26个字母 × 约10个建议词/次）。
修饰词前缀扩展
除了字母，还可以添加常见修饰词前缀：
- how — how to, how do, how much
- what — what is, what are, what causes
- best — best running shoes, best cheap running shoes
- vs — nike vs adidas running shoes
- near me — running shoes near me
将字母扩展与修饰词扩展结合，一个种子词可扩展出300-500个候选长尾词。
递归PAA扩展
PAA模块默认展示3-4个问题。点击展开一个问题后，Google会动态注入新的相关问题。通过递归展开，可以从单个种子词的PAA模块中提取20-50个嵌套问题。每个问题本身又可以作为新的种子词再次查询Autocomplete和PAA，形成指数级扩展。
为什么Autocomplete和PAA抓取需要住宅代理
直接用数据中心IP批量请求Autocomplete端点或PAA页面，会遇到两个核心问题：
1. 每IP速率限制
Google对suggestqueries.google.com端点施加了每IP速率限制。经验表明，单IP在短时间内发送超过约100-200次请求就可能触发临时封锁（HTTP 429或空响应）。对于需要a-z × 多种子词 × 多地区的扩展任务，单IP远远不够。
2. 地区偏见与本地化
Autocomplete建议和PAA问题具有强烈的地区依赖性。同一关键词"best running shoes"在美国、德国、日本的建议词完全不同。如果用美国数据中心IP查询德国用户的建议词，结果会被严重偏移。根据Google官方文档，搜索结果和建议受用户位置、语言和搜索历史影响（参见 Google搜索API文档）。
住宅代理通过提供真实ISP分配的IP地址，使请求看起来来自真实用户。配合国家/城市级地理定位，可以获取准确的本地化建议数据。
ProxyHat代理配置详解
ProxyHat住宅代理网关地址为 gate.proxyhat.com，HTTP端口 8080，SOCKS5端口 1080。地理定位参数通过用户名传递：
# 德国柏林住宅代理
http://user-country-DE-city-berlin:pass@gate.proxyhat.com:8080
# 美国纽约住宅代理
http://user-country-US-city-newyork:pass@gate.proxyhat.com:8080
# 日本东京住宅代理（SOCKS5）
socks5://user-country-JP-city-tokyo:pass@gate.proxyhat.com:1080使用-session-xxx参数可保持会话粘性，确保同一IP用于同一搜索会话：
http://user-country-DE-city-berlin-session-key1:pass@gate.proxyhat.com:8080更多配置细节可参考 ProxyHat官方文档。定价信息见 ProxyHat定价页。
实战代码：用httpx抓取Autocomplete JSON
以下示例使用httpx通过ProxyHat住宅代理抓取Autocomplete建议词，包含错误处理和重试逻辑：
import httpx
import time
import logging
import json
from typing import Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# ProxyHat 住宅代理配置
PROXY_URL = "http://user-country-DE-city-berlin:YOUR_PASSWORD@gate.proxyhat.com:8080"
AUTOCOMPLETE_URL = "https://suggestqueries.google.com/complete/search"
# Chrome client 参数返回 JSON 而非 XML
DEFAULT_PARAMS = {"client": "chrome", "hl": "de", "gl": "de"}
MAX_RETRIES = 3
RETRY_DELAY = 2.0 # 秒
def fetch_autocomplete(
keyword: str,
client: httpx.Client,
retries: int = MAX_RETRIES
) -> Optional[list[str]]:
"""抓取单个关键词的Autocomplete建议词列表"""
params = {**DEFAULT_PARAMS, "q": keyword}
for attempt in range(1, retries + 1):
try:
resp = client.get(
AUTOCOMPLETE_URL,
params=params,
timeout=10.0,
headers={
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
),
"Accept-Language": "de-DE,de;q=0.9,en;q=0.8",
},
)
resp.raise_for_status()
data = resp.json()
# Google返回格式: [原始查询, [建议词列表], ...]
suggestions = data[1] if len(data) > 1 else []
logger.info(f"'{keyword}' -> {len(suggestions)} suggestions")
return suggestions
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
logger.warning(f"Rate limited on '{keyword}', attempt {attempt}/{retries}")
time.sleep(RETRY_DELAY * attempt)
else:
logger.error(f"HTTP {e.response.status_code} for '{keyword}'")
return None
except (httpx.RequestError, json.JSONDecodeError) as e:
logger.error(f"Request failed for '{keyword}': {e}")
time.sleep(RETRY_DELAY)
return None
def expand_with_alpha(seed: str, client: httpx.Client) -> list[str]:
"""a-z字母前缀扩展"""
all_suggestions = []
for ch in "abcdefghijklmnopqrstuvwxyz":
query = f"{seed} {ch}"
suggestions = fetch_autocomplete(query, client)
if suggestions:
all_suggestions.extend(suggestions)
time.sleep(0.5) # 礼貌延迟
return all_suggestions
# 使用示例
with httpx.Client(proxy=PROXY_URL) as client:
seed = "running shoes"
results = expand_with_alpha(seed, client)
print(f"总共获取 {len(results)} 个建议词")
print(results[:10])
上述代码通过德国柏林的住宅代理发出请求，hl=de和gl=de参数确保返回德语建议词。每个字母查询之间有0.5秒延迟，避免触发速率限制。
用修饰词前缀进一步扩展
import httpx
import time
import logging
from itertools import chain
logger = logging.getLogger(__name__)
MODIFIER_PREFIXES = [
"how", "what", "why", "when", "where", "which",
"best", "top", "cheap", "buy", "review",
"vs", "alternative", "near me", "for",
]
SECOND_MODIFIERS = ["to", "is", "are", "does", "much", "many", "long"]
def expand_with_modifiers(
seed: str,
client: httpx.Client,
fetch_fn
) -> list[str]:
"""使用修饰词前缀扩展种子词"""
all_suggestions = []
# 单修饰词: "best running shoes"
for mod in MODIFIER_PREFIXES:
query = f"{mod} {seed}"
suggestions = fetch_fn(query, client)
if suggestions:
all_suggestions.extend(suggestions)
time.sleep(0.4)
# 双修饰词: "how to running shoes", "what is running shoes"
for mod1 in ["how", "what", "why", "when"]:
for mod2 in SECOND_MODIFIERS:
query = f"{mod1} {mod2} {seed}"
suggestions = fetch_fn(query, client)
if suggestions:
all_suggestions.extend(suggestions)
time.sleep(0.4)
return all_suggestions
# 组合字母扩展和修饰词扩展
# total = expand_with_alpha(seed, client) + expand_with_modifiers(seed, client, fetch_autocomplete)
用Playwright递归展开PAA问题
PAA问题嵌在动态DOM中，需要浏览器渲染才能提取。以下代码使用Playwright通过ProxyHat代理加载Google搜索页，递归点击PAA手风琴并提取问题及引用来源：
import asyncio
import logging
from playwright.async_api import async_playwright
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class PAAQuestion:
question: str
answer_snippet: str = ""
source_url: str = ""
source_name: str = ""
PROXY_SERVER = "http://gate.proxyhat.com:8080"
PROXY_USER = "user-country-DE-city-berlin"
PROXY_PASS = "YOUR_PASSWORD"
MAX_PAA_DEPTH = 8 # 递归展开层数
SCROLL_WAIT = 2.0 # 秒
def build_paa_url(keyword: str, gl: str = "de", hl: str = "de") -> str:
return f"https://www.google.com/search?q={keyword}&gl={gl}&hl={hl}"
async def scrape_paa(keyword: str, max_depth: int = MAX_PAA_DEPTH) -> list[PAAQuestion]:
"""递归抓取PAA问题及其引用来源"""
results: list[PAAQuestion] = []
seen_questions: set[str] = set()
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=True,
proxy={
"server": PROXY_SERVER,
"username": PROXY_USER,
"password": PROXY_PASS,
},
)
context = await browser.new_context(
locale="de-DE",
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
),
)
page = await context.new_page()
try:
url = build_paa_url(keyword)
await page.goto(url, wait_until="domcontentloaded", timeout=30000)
await page.wait_for_timeout(SCROLL_WAIT * 1000)
for depth in range(max_depth):
# 查找所有PAA问题元素
paa_elements = page.locator(
"div[jsname='Cpkphb'] >> div[class*='BA0z6e']"
)
count = await paa_elements.count()
if count == 0:
logger.info(f"Depth {depth}: No PAA elements found")
break
# 点击最后一个未展开的问题以触发新问题加载
for i in range(count):
try:
el = paa_elements.nth(i)
text = await el.inner_text()
question_text = text.strip().split("\n")[0]
if question_text and question_text not in seen_questions:
seen_questions.add(question_text)
# 点击展开获取答案
await el.click()
await page.wait_for_timeout(1500)
# 提取答案摘要和引用来源
answer_el = page.locator(
"div[data-attrid='wa:/description']"
).first
if await answer_el.count() > 0:
answer_text = await answer_el.inner_text()
else:
answer_text = ""
# 提取引用来源链接
source_link = page.locator(
"a[data-ved] >> span"
).first
source_url = ""
source_name = ""
if await source_link.count() > 0:
source_name = await source_link.inner_text()
href = await page.locator(
"a[data-ved]"
).first.get_attribute("href")
source_url = href or ""
results.append(PAAQuestion(
question=question_text,
answer_snippet=answer_text[:500],
source_url=source_url,
source_name=source_name,
))
logger.info(f"Depth {depth}: {question_text[:60]}")
except Exception as e:
logger.debug(f"Error at depth {depth}, item {i}: {e}")
continue
# 滚动到底部加载更多
await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
await page.wait_for_timeout(2000)
except Exception as e:
logger.error(f"PAA scraping failed: {e}")
finally:
await browser.close()
return results
# 运行
# results = asyncio.run(scrape_paa("running shoes"))
# for r in results[:5]:
# print(f"Q: {r.question}")
# print(f"A: {r.answer_snippet[:100]}")
# print(f"Source: {r.source_name} - {r.source_url}")
# print()
该脚本通过ProxyHat德国柏林住宅代理加载Google搜索页，递归展开PAA手风琴，提取每个问题的文本、答案摘要和引用来源URL。设置最大深度为8层，避免无限递归。
异步并发抓取多种子词
当需要处理多个种子词时，使用asyncio并发可以大幅提升效率。以下示例展示如何用httpx AsyncClient配合ProxyHat会话粘性代理实现并发抓取：
import asyncio
import httpx
import logging
from typing import Optional
logger = logging.getLogger(__name__)
AUTOCOMPLETE_URL = "https://suggestqueries.google.com/complete/search"
def build_proxy_url(country: str, city: str, session_id: str) -> str:
"""构建ProxyHat会话粘性代理URL"""
return (
f"http://user-country-{country}-city-{city}-session-{session_id}"
f":YOUR_PASSWORD@gate.proxyhat.com:8080"
)
async def fetch_autocomplete_async(
keyword: str,
country: str = "DE",
city: str = "berlin",
hl: str = "de",
gl: str = "de",
) -> Optional[list[str]]:
"""异步抓取Autocomplete，每个关键词使用独立会话IP"""
session_id = f"kw{abs(hash(keyword)) % 100000}"
proxy_url = build_proxy_url(country, city, session_id)
params = {"client": "chrome", "hl": hl, "gl": gl, "q": keyword}
async with httpx.AsyncClient(
proxy=proxy_url,
timeout=httpx.Timeout(15.0, connect=10.0),
limits=httpx.Limits(max_connections=5, max_keepalive_connections=2),
headers={
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
),
"Accept-Language": f"{hl}-{country.upper()},{hl};q=0.9",
},
) as client:
try:
resp = await client.get(AUTOCOMPLETE_URL, params=params)
resp.raise_for_status()
data = resp.json()
suggestions = data[1] if len(data) > 1 else []
logger.info(f"[{country}/{city}] '{keyword}' -> {len(suggestions)} suggestions")
return suggestions
except Exception as e:
logger.error(f"Failed for '{keyword}': {e}")
return None
async def batch_expand(
seeds: list[str],
country: str = "DE",
city: str = "berlin",
max_concurrent: int = 5,
) -> dict[str, list[str]]:
"""批量并发扩展多种子词"""
semaphore = asyncio.Semaphore(max_concurrent)
async def bounded_fetch(seed: str) -> tuple[str, list[str]]:
async with semaphore:
# 字母扩展
all_suggestions = []
for ch in "abcdefghijklmnopqrstuvwxyz":
query = f"{seed} {ch}"
suggestions = await fetch_autocomplete_async(
query, country, city
)
if suggestions:
all_suggestions.extend(suggestions)
await asyncio.sleep(0.3) # 礼貌延迟
return seed, list(set(all_suggestions))
tasks = [bounded_fetch(seed) for seed in seeds]
results = await asyncio.gather(*tasks, return_exceptions=True)
output = {}
for result in results:
if isinstance(result, tuple):
output[result[0]] = result[1]
else:
logger.error(f"Task failed: {result}")
return output
# 运行示例
# seeds = ["running shoes", "trail running", "marathon training"]
# results = asyncio.run(batch_expand(seeds, country="DE", city="berlin"))
# for seed, suggestions in results.items():
# print(f"{seed}: {len(suggestions)} suggestions")
使用信号量控制并发数为5，每个关键词分配独立的会话ID以获得不同IP。更多关于Web抓取最佳实践，可参考 ProxyHat Web抓取用例。
去重、意图聚类与CSV导出
抓取完成后，需要对结果去重、按意图聚类并导出为可用的CSV格式：
import csv
import re
from dataclasses import dataclass
from collections import defaultdict
from typing import Optional
@dataclass
class KeywordEntry:
keyword: str
source: str # "autocomplete" | "paa" | "related"
country: str
intent: str = ""
parent_seed: str = ""
paa_source_url: str = ""
# 意图分类规则
INTENT_PATTERNS = {
"informational": [
r"^how\b", r"^what\b", r"^why\b", r"^when\b",
r"^where\b", r"^which\b", r"\bguide\b", r"\btutorial\b",
],
"transactional": [
r"\bbuy\b", r"\bprice\b", r"\bsale\b", r"\bcheap\b",
r"\bdeal\b", r"\bcoupon\b", r"\bdiscount\b",
],
"navigational": [
r"\blogin\b", r"\bsign\s*up\b", r"\bofficial\b",
r"\bnear\s*me\b",
],
"comparison": [
r"\bvs\b", r"\balternative\b", r"\bcompare\b",
r"\bor\b",
],
}
def classify_intent(keyword: str) -> str:
"""基于关键词模式分类搜索意图"""
kw_lower = keyword.lower()
for intent, patterns in INTENT_PATTERNS.items():
for pattern in patterns:
if re.search(pattern, kw_lower):
return intent
return "commercial" # 默认归为商业型
def dedup_and_cluster(
entries: list[KeywordEntry],
) -> dict[str, list[KeywordEntry]]:
"""去重并按意图聚类"""
seen = set()
clustered: dict[str, list[KeywordEntry]] = defaultdict(list)
for entry in entries:
normalized = entry.keyword.lower().strip()
if normalized in seen:
continue
seen.add(normalized)
entry.intent = classify_intent(entry.keyword)
clustered[entry.intent].append(entry)
return clustered
def export_to_csv(
clustered: dict[str, list[KeywordEntry]],
filepath: str = "keyword_research.csv",
) -> None:
"""导出为CSV，按意图分组"""
with open(filepath, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([
"keyword", "intent", "source", "country",
"parent_seed", "paa_source_url",
])
for intent, entries in sorted(clustered.items()):
for entry in entries:
writer.writerow([
entry.keyword,
entry.intent,
entry.source,
entry.country,
entry.parent_seed,
entry.paa_source_url,
])
total = sum(len(v) for v in clustered.values())
print(f"导出完成: {total} 个关键词 -> {filepath}")
for intent, entries in clustered.items():
print(f" {intent}: {len(entries)} 个")
# 使用示例
# entries = [
# KeywordEntry("how to choose running shoes", "autocomplete", "DE", parent_seed="running shoes"),
# KeywordEntry("best running shoes 2024", "autocomplete", "DE", parent_seed="running shoes"),
# KeywordEntry("running shoes vs walking shoes", "paa", "DE", parent_seed="running shoes"),
# KeywordEntry("buy nike running shoes", "autocomplete", "DE", parent_seed="running shoes"),
# ]
# clustered = dedup_and_cluster(entries)
# export_to_csv(clustered, "running_shoes_keywords_de.csv")
导出的CSV可直接导入内容管理工具或用于FAQ页面规划。PAA来源URL字段可帮助内容团队直接引用权威来源。
使用curl快速验证Autocomplete
在编写完整脚本前，可以用curl快速验证代理和端点是否正常工作：
# 通过ProxyHat德国代理抓取Autocomplete建议
curl -x "http://user-country-DE-city-berlin:YOUR_PASSWORD@gate.proxyhat.com:8080" \
"https://suggestqueries.google.com/complete/search?client=chrome&hl=de&gl=de&q=running+shoes" \
-H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0" \
-H "Accept-Language: de-DE,de;q=0.9" \
--connect-timeout 10 \
--max-time 15
# 使用SOCKS5代理（端口1080）
curl -x "socks5://user-country-US-city-newyork:YOUR_PASSWORD@gate.proxyhat.com:1080" \
"https://suggestqueries.google.com/complete/search?client=chrome&hl=en&gl=us&q=best+running+shoes" \
-H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0" \
--connect-timeout 10伦理与合规考量
Autocomplete建议和PAA问题是公开可见的搜索建议数据，但大规模自动化抓取仍需注意以下原则：
- 控制请求频率：每个IP保持每秒不超过1-2次请求，避免对Google服务造成压力。
- 优先使用官方API：如果项目规模较大（每日超过10,000次请求），应考虑使用 Google Custom Search API 或 SerpAPI 等官方/第三方API服务。
- 遵守robots.txt：检查
suggestqueries.google.com/robots.txt的规则，虽然Autocomplete端点通常未被禁止，但仍应遵守。 - GDPR合规：如果处理涉及欧盟用户的数据，确保不收集个人可识别信息。PAA问题本身是聚合搜索行为数据，不涉及个人隐私。
- 合理使用：抓取的数据应用于关键词研究和内容规划，而非重新发布为搜索服务。
关于SERP追踪的更多技术细节，可参考 ProxyHat SERP追踪用例。
关键要点总结
三大数据源各有侧重：Autocomplete适合探索型长尾词发现，PAA适合信息型FAQ内容选题，Related Searches适合内容集群规划。三者结合可覆盖完整搜索意图漏斗。
住宅代理是规模化前提：单IP在约100-200次请求后可能触发限制。使用ProxyHat住宅代理配合国家/城市地理定位，可获取准确的本地化建议数据，同时分散请求避免封锁。
a-z + 修饰词扩展是核心策略：一个种子词通过字母前缀和修饰词前缀可扩展出300-500个候选词。递归PAA展开可额外获得20-50个嵌套问题。
去重和意图聚类是关键后处理步骤：基于正则模式的意图分类可将关键词分为informational/transactional/navigational/comparison四类，直接指导内容策略。
礼貌抓取是可持续之道：控制并发、添加延迟、遵守robots.txt，在规模较大时迁移到官方API。
准备好开始构建你的关键词研究工具了吗？查看 ProxyHat支持的全球代理位置，选择你需要的国家和地区开始本地化关键词扩展。
