这篇能做出什么
先看成果。做完这套流水线，你会得到一个每天自动更新的表：每个监控问题在 AI 搜索答案里引用了哪些链接、来自哪些域名、自有域名有没有被引用、排在第几位。
举个虚构的例子，格式如下（数字仅示意）：
| 问题 | 引用条数 | 来源域名数 | 自有域引用 | 首位引用 |
|---|---|---|---|---|
| 家用净水器怎么挑 | 6 | 5 | 1 | 否 |
| 净水器滤芯多久换一次 | 4 | 3 | 2 | 是 |
围绕这张表，还能算出四个 GEO（Generative Engine Optimization，生成式引擎优化）指标：
- 引用覆盖率：含至少一条自有域引用的答案数 ÷ 答案总数
- 引用份额：自有域引用条数 ÷ 总引用条数
- 首位引用率：自有域出现在第 1 条引用的比例
- 提及率：品牌名出现在答案正文（哪怕没有链接）的比例
再加一步差分，就能看到「这周新增了哪些引用、哪些引用消失了」，形成可持续观测的闭环。
前置条件清单
- Python 3（以官方文档当前版本为准），会装
requests、pandas、PyYAML即可 - 一个能拿到 AI 答案的入口：官方 API、产品页面的复制导出、或你有权访问的页面
- 一个待监控的问题清单（10～50 条起步就够）
- 一张自有域名 + 竞品域名表
- 基本的正则和 HTML 常识
- 合规前提：只抓你有权访问的内容，遵守目标站点的 robots.txt 与服务条款，控制请求频率
拿不到 API 时，最稳的做法是手动导出：在 AI 搜索页面复制答案，粘进文本框存成 Markdown。规模小但零风险，先把解析和统计跑通，再考虑自动化。
分步骤
第 1 步：统一数据入口，先定 schema
不管答案从哪来，先落成一个统一的 JSONL 文件，一行一条答案。后面所有脚本只认这个格式。
```json
{"answer_id":"q001-2024-06-01","query":"家用净水器怎么挑","engine":"example-engine","collected_at":"2024-06-01T09:00:00+08:00","answer_markdown":"选购时先看滤芯类型[1]，再看通量[2]。","answer_html":""}
```
建目录：
```bash
mkdir -p geo-monitor/{data/raw,config,src,out,logs}
cd geo-monitor
```
三个关键约定：
answer_id用问题编号 + 日期，方便按时间分组answer_markdown存原始文本，不要预先清洗，方便回溯- 采集环节不要解析引用，解析单独一步做，出错好排查
第 2 步：解析行内引用，三种格式分别处理
AI 答案里的引用通常长这样：
1. Markdown 行内链接：某媒体
2. 上标编号 + 文末参考文献：正文 [1]，末尾 [1] https://example.com/a
3. 裸 URL，或 HTML 卡片式 <sup><a href="...">1</a></sup>
写一个解析器，按「先链接、再编号、最后裸链」的顺序处理，避免重复计数。
```python
src/parse.py
import re
MD_LINK = re.compile(r'\[([^\]\n]{0,200})\]\((https?://[^\s)]+)\)')
ANGLE_LINK = re.compile(r'<(https?://[^\s>]+)>')
BARE_URL = re.compile(r'(?<![\w(\[<])(https?://[^\s<>"\')\]]+)')
NUM_MARK = re.compile(r'\[(\d{1,3})\]')
URL_IN_TEXT = re.compile(r'https?://[^\s<>"\')\]]+')
def build_ref_map(text: str) -> dict:
"""从文末参考文献区建立 编号 -> URL 的映射"""
refs = {}
for line in text.splitlines():
m = re.match(r'^\s*\[?(\d{1,3})\]?[\.\)、:]?\s+(.*)$', line)
if not m:
continue
url_m = URL_IN_TEXT.search(m.group(2))
if url_m:
refs[m.group(1)] = url_m.group(0)
return refs
def extract_citations(text: str) -> list:
cits, order, rest = [], 0, text
1) Markdown 行内链接
for m in MD_LINK.finditer(rest):
order += 1
cits.append({"order": order, "anchor": m.group(1).strip(),
"url": m.group(2), "kind": "inline_link"})
rest = MD_LINK.sub(lambda m: m.group(1), rest)
2) 尖括号链接
for m in ANGLE_LINK.finditer(rest):
order += 1
cits.append({"order": order, "anchor": "", "url": m.group(1), "kind": "angle"})
rest = ANGLE_LINK.sub("", rest)
3) 编号式引用（需配合参考文献区）
refs = build_ref_map(text)
for m in NUM_MARK.finditer(rest):
idx = m.group(1)
if idx in refs:
order += 1
cits.append({"order": order, "anchor": "", "url": refs[idx], "kind": "numref"})
4) 兜底：裸 URL
for m in BARE_URL.finditer(rest):
order += 1
cits.append({"order": order, "anchor": "", "url": m.group(1), "kind": "bare"})
return cits
```
kind 字段很有用：后面排查「为什么某条引用漏了」时，先按 kind 分组看数量分布。
第 3 步：URL 规范化，别让同一个页面算成两个来源
追踪参数、末尾斜杠、大小写都会把同一个页面拆成多条记录。归因之前统一清洗一遍。
```python
src/attribute.py
from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode
TRACKING = {"utm_source", "utm_medium", "utm_campaign", "utm_term",
"utm_content", "gclid", "fbclid", "spm", "ref", "from"}
def normalize_url(url: str) -> str:
p = urlparse(url)
host = p.netloc.lower()
if host.startswith("www."):
host = host[4:]
path = p.path.rstrip("/") or "/"
query = [(k, v) for k, v in parse_qsl(p.query, keep_blank_values=True)
if k.lower() not in TRACKING]
return urlunparse((p.scheme.lower(), host, path, "", urlencode(query), ""))
def domain_of(url: str) -> str:
return urlparse(normalize_url(url)).netloc
```
然后是来源分类。用一张 CSV 维护映射，代码只读配置，改规则不用改代码。
```csv
config/domains.csv
domain,source_type,is_owned,note
example.com,media,false,示例媒体
docs.example.org,official_docs,false,示例官方文档
yourbrand.com,owned,true,自有主站
```
```python
import csv
def load_domain_map(path="config/domains.csv") -> dict:
dmap = {}
with open(path, newline="", encoding="utf-8") as f:
for row in csv.DictReader(l for l in f if not l.startswith("#")):
dmap[row["domain"].lower()] = {
"source_type": row["source_type"],
"is_owned": row["is_owned"].strip().lower() == "true",
}
return dmap
def classify(domain: str, dmap: dict) -> dict:
if domain in dmap:
return dmap[domain]
未收录：按后缀粗略分类，并打标记等人工复核
if domain.endswith(".gov.cn") or domain.endswith(".edu.cn"):
return {"source_type": "authority", "is_owned": False}
return {"source_type": "unknown", "is_owned": False}
```
一定要打 unknown 标记。每周把 unknown 的域名捞出来补进 CSV，两三轮之后映射表就够用了。
第 4 步：统计来源占比，算出 GEO 指标
把每条引用展开成一行，用 pandas 聚合。
```python
src/metrics.py
import json, pandas as pd
from attribute import normalize_url, domain_of, load_domain_map, classify
def build_frame(raw_jsonl: str) -> pd.DataFrame:
dmap = load_domain_map()
rows = []
with open(raw_jsonl, encoding="utf-8") as f:
for line in f:
rec = json.loads(line)
from parse import extract_citations
cits = extract_citations(rec["answer_markdown"])
brand = rec.get("brand", "")
for c in cits:
url = normalize_url(c["url"])
dom = domain_of(url)
meta = classify(dom, dmap)
rows.append({
"answer_id": rec["answer_id"],
"query": rec["query"],
"engine": rec["engine"],
"date": rec["collected_at"][:10],
"order": c["order"],
"kind": c["kind"],
"anchor": c["anchor"],
"url": url,
"domain": dom,
"source_type": meta["source_type"],
"is_owned": meta["is_owned"],
"brand_in_text": brand in rec["answer_markdown"] if brand else False,
})
return pd.DataFrame(rows)
def geo_metrics(df: pd.DataFrame) -> pd.DataFrame:
out = []
for (query, engine, date), g in df.groupby(["query", "engine", "date"]):
answers = g["answer_id"].nunique()
owned = g[g["is_owned"]]
out.append({
"query": query, "engine": engine, "date": date,
"answers": answers,
"citations": len(g),
"domains": g["domain"].nunique(),
"coverage": owned["answer_id"].nunique() / answers,
"share": len(owned) / len(g),
"first_pos_rate": (owned["order"] == 1).sum() / max(owned["answer_id"].nunique(), 1),
"mention_rate": g.groupby("answer_id")["brand_in_text"].first().mean(),
})
return pd.DataFrame(out)
```
来源占比一条命令就能看：
```python
df.groupby(["source_type", "domain"]).size().reset_index(name="n") \
.sort_values("n", ascending=False)
```
第 5 步：串成定时流水线，并加差分
目录固定成流水线骨架：
```text
geo-monitor/
├── data/raw/2024-06-01.jsonl
├── config/{domains.csv,queries.txt}
├── src/{parse.py,attribute.py,metrics.py,report.py}
├── out/weekly.md
└── logs/
```
写一个总入口：
```python
src/report.py
import pandas as pd
from metrics import build_frame, geo_metrics
df = build_frame("data/raw/current.jsonl")
m = geo_metrics(df)
m.to_csv("out/metrics.csv", index=False)
prev = pd.read_csv("out/metrics_prev.csv")
cur_keys = set(df["answer_id"] + "|" + df["domain"])
prev_df = build_frame("data/raw/prev.jsonl")
prev_keys = set(prev_df["answer_id"] + "|" + prev_df["domain"])
new_cites = cur_keys - prev_keys
lost_cites = prev_keys - cur_keys
with open("out/weekly.md", "w", encoding="utf-8") as f:
f.write("# GEO 周报\n\n")
f.write(f"新增引用 {len(new_cites)} 条，消失引用 {len(lost_cites)} 条。\n\n")
f.write(m.to_markdown(index=False))
```
定时任务用 cron，每天清晨跑一次：
```bash
crontab -e
0 8 * * * cd /opt/geo-monitor && /usr/bin/env python3 src/report.py >> logs/run.log 2>&1
```
跑之前记得把 current.jsonl 备份成 prev.jsonl，否则差分永远是空的。这一步放在脚本开头：
```python
import shutil, os
if os.path.exists("data/raw/prev.jsonl"):
shutil.copy("data/raw/current.jsonl", "data/raw/prev.jsonl")
```
第 6 步：把数据回灌到 GEO 动作
监测不落到动作上等于白做。拿到周报后固定问三个问题：
1. 被引最多的是哪类来源？ 如果 source_type 集中在官方文档和问答社区，就去这两类渠道经营内容，而不是只改官网。
2. 引用的锚文本在说什么？ 把你被引用时的 anchor 字段单独拉出来看，AI 用什么措辞指代你，就照着这个措辞优化页面标题和小标题。
3. 哪些问题一直零引用？ 这些是内容缺口，优先补。
常见坑与排错
编号与参考文献错位。 有的答案脚注从 0 开始，有的跨段落复用同一编号，还有的参考文献区和正文用两种括号。排查方法：打印 build_ref_map 的结果，和原文末尾的人工比对一遍。
直接请求拿不到引用。 页面是前端渲染的，requests 拿到的 HTML 里根本没有链接。要么改用官方 API，要么用浏览器自动化渲染后再取 page.content()，注意控制频率并遵守条款。
跳转链接没解码。 引用常被包装成 /redirect?url=... 或短链。归因前先跟随重定向拿到最终 URL，但同时保留原始 URL，方便对账。
追踪参数把来源拆散。 同一页面带不同 utm_* 会统计成多条，务必先过 normalize_url。
全角字符导致正则失效。 中文答案里常见 【1】、（链接） 这类全角写法，正则里补上全角分支，或者先做一次字符归一化。
口径变了，趋势就断了。 今天按「引用条数」算份额，明天改按「答案数」算，曲线一定跳变。在 out/metrics.csv 里加一列 schema_version，改口径就升版本，并在周报里注明。
映射表过期。 网站换域名、媒体被收购都会让分类失真，每月人工抽查一次 unknown 列表。
答案里的引用是卡片而非链接。 部分产品把来源渲染成结构化卡片，正文里没有可见 URL。这种情况改为解析 HTML 里的 href 属性，或直接调 API 拿结构化字段。
下一步建议
- 扩问题库：从 10 条扩到 100 条，按「认知类 / 对比类 / 购买类」分组，不同组别的引用结构通常差异很大。
- 接多引擎对照：同一批问题在不同 AI 搜索产品里各跑一遍，对比引用份额，找出各自的偏好来源。
- 加截图存证：每次采集顺手存一张页面截图，答案改版时能回溯当时的样子。
- 打通 SEO 数据：把自然搜索排名和 AI 引用率放在一张表里，能看出哪些页面在两种入口表现背离。
- 做内容实验：挑三个零引用的问题，针对性改写页面后观察四周的覆盖率变化，这是验证 GEO 动作是否有效最朴素的办法。
整套东西的核心其实就三件事：把引用解析准、把来源归对类、把口径固定住。剩下的都是围绕这三件事加自动化和展示。先用手动导出的十条答案跑通全流程，再谈规模化。