介绍如何使用 Python 构建一个通用的 GEO（生成引擎优化）多平台监控系统。系统架构包含任务调度、平台查询和结果分析模块，支持 ChatGPT、豆包、Kimi、文心一言等多个 AI 平台。通过定义基类实现平台解耦，利用 schedule 库配置定时任务自动执行监控并生成报告。代码示例展示了如何集成各平台 API、提取品牌信息以及保存监控数据到 JSON 文件，旨在解决单平台监控扩展性差的问题，提升自动化效率。
随着 AI 模型快速迭代，单一平台监控脚本维护成本较高。痛定思痛，决定写一套通用的 GEO 多平台监控方案。
本文分享完整代码，支持：ChatGPT、豆包、Kimi、文心一言、通义千问。
┌─────────────────────────────────────────────────────────┐
│ GEO 多平台监控系统 │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 任务调度 │→ │ 平台查询 │→ │ 结果分析 │ │
│ └───────────┘ └───────────┘ └───────────┘ │
│ ↑ ↓ ↓ │
│ └──── 告警通知 ←────── 报告生成 ←────── │
│ │
│ 支持平台： │
│ ✅ ChatGPT ✅ 豆包 ✅ Kimi │
│ ✅ 文心一言 ✅ 通义千问 │
└─────────────────────────────────────────────────────────┘
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Optional, Dict
from datetime import datetime
import time
import random
@dataclass
class GEOResult:
"""GEO 查询结果"""
platform: str
query: str
brand: str
mentioned: bool
position: int
context: str
timestamp: str
response_time_ms: float
error: Optional[str] = None
class BasePlatform(ABC):
"""AI 平台基类"""
def __init__(self, name: str, api_key: str = None):
self.name = name
self.api_key = api_key
self.base_delay = 1.0 # 基础延迟 (秒)
@abstractmethod
def query(self, search_query: str) -> str:
"""执行查询，返回 AI 响应文本"""
pass
@abstractmethod
def extract_brand_info(self, response: str, brand: str) -> Dict:
"""从响应中提取品牌相关信息"""
pass
def check_brand(self, search_query: str, brand: str) -> GEOResult:
"""检查品牌在 AI 响应中的出现情况"""
start_time = time.time()
try:
# 执行查询
response = self.query(search_query)
# 提取品牌信息
brand_info = self.extract_brand_info(response, brand)
response_time = (time.time() - start_time) * 1000
return GEOResult(
platform=self.name,
query=search_query,
brand=brand,
mentioned=brand_info["mentioned"],
position=brand_info["position"],
context=brand_info["context"],
timestamp=datetime.now().isoformat(),
response_time_ms=response_time
)
except Exception as e:
response_time = (time.time() - start_time) * 1000
return GEOResult(
platform=self.name,
query=search_query,
brand=brand,
mentioned=False,
position=0,
context="",
timestamp=datetime.now().isoformat(),
response_time_ms=response_time,
error=str(e)
)
import os
import requests
class ChatGPTPlatform(BasePlatform):
"""ChatGPT 平台"""
def __init__(self, api_key: str = None):
super().__init__("ChatGPT", api_key or os.getenv("OPENAI_API_KEY"))
self.api_base = "https://api.openai.com/v1"
self.model = "gpt-4"
def query(self, search_query: str) -> str:
"""查询 ChatGPT"""
url = f"{self.api_base}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [{"role": "user", "content": f"作为一个专业的 AI 助手，请回答：{search_query}"}],
"temperature": 0.7,
"max_tokens": 500
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
return result["choices"][0][][]
() -> :
mentioned = brand response
mentioned:
{: , : , : }
position = response.index(brand)
start = (, position - )
end = ((response), position + (brand) + )
context = response[start:end]
{: , : , : }
class DoubaoPlatform(BasePlatform):
"""豆包平台 (字节跳动)"""
def __init__(self, api_key: str = None):
super().__init__("豆包", api_key or os.getenv("DOUBAO_API_KEY"))
self.api_base = "https://ark.cn-beijing.volces.com/api/v3"
self.model = "doubao-seed-250615"
def query(self, search_query: str) -> str:
"""查询豆包"""
url = f"{self.api_base}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [{"role": "user", "content": search_query}]
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
return result["choices"][0]["message"]["content"]
def extract_brand_info(self, response: str, brand: str) -> Dict:
mentioned = brand response
mentioned:
{: , : , : }
position = response.index(brand)
start = (, position - )
end = ((response), position + (brand) + )
{: , : , : }
class KimiPlatform(BasePlatform):
"""Kimi 平台 (月之暗面)"""
def __init__(self, api_key: str = None):
super().__init__("Kimi", api_key or os.getenv("KIMI_API_KEY"))
self.api_base = "https://api.moonshot.cn/v1"
self.model = "kimi-flash"
def query(self, search_query: str) -> str:
"""查询 Kimi"""
url = f"{self.api_base}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [{"role": "user", "content": search_query}]
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
return result["choices"][0]["message"]["content"]
def extract_brand_info(self, response: str, brand: str) -> Dict:
mentioned = brand response
mentioned:
{: , : , : }
idx = response.index(brand)
{: , : , : }
from typing import List, Dict
import json
class GEOMultiPlatformMonitor:
"""GEO 多平台监控器"""
def __init__(self, brand: str):
self.brand = brand
self.platforms: List[BasePlatform] = []
self.results: List[GEOResult] = []
def add_platform(self, platform: BasePlatform):
"""添加监控平台"""
self.platforms.append(platform)
def check_all_platforms(self, queries: List[str]) -> List[GEOResult]:
"""检查所有平台"""
all_results = []
for platform in self.platforms:
for query in queries:
print(f"正在检查 {platform.name} - {query}...")
result = platform.check_brand(query, self.brand)
all_results.append(result)
# 添加延迟，避免请求过快
time.sleep(random.uniform(1.0, 2.0))
self.results = all_results
return all_results
def generate_report(self) -> str:
.results:
report =
total = (.results)
mentioned = ( r .results r.mentioned)
success = ( r .results r.error)
report +=
report +=
report +=
report +=
report +=
platform_name (r.platform r .results):
platform_results = [r r .results r.platform == platform_name]
p_mentioned = ( r platform_results r.mentioned)
p_total = (platform_results)
p_success = ( r platform_results r.error)
emoji = p_mentioned >
report +=
report +=
report +=
report +=
report +=
errors = [r r .results r.error]
errors:
report +=
r errors:
report +=
report
():
filename:
filename =
data = [
{
: r.platform,
: r.query,
: r.mentioned,
: r.position,
: r.context,
: r.timestamp,
: r.response_time_ms,
: r.error
}
r .results
]
(filename, , encoding=) f:
json.dump(data, f, ensure_ascii=, indent=)
()
import schedule
def daily_geo_check():
"""每日 GEO 检查任务"""
monitor = GEOMultiPlatformMonitor("你的品牌")
# 添加平台
monitor.add_platform(ChatGPTPlatform())
monitor.add_platform(DoubaoPlatform())
monitor.add_platform(KimiPlatform())
# 设置查询词
queries = ["推荐一个 XX 工具", "XX 服务商哪家好", "怎么选择 XX 供应商", "XX 工具排行榜"]
# 执行检查
results = monitor.check_all_platforms(queries)
# 生成报告
report = monitor.generate_report()
print(report)
# 保存结果
monitor.save_to_json()
# 配置定时任务
schedule.every().day.at("09:00").do(daily_geo_check)
print("GEO 多平台监控系统已启动，每天 9:00 自动检查")
# 运行
while True:
schedule.run_pending()
time.sleep(60)
pip install requests schedule
export OPENAI_API_KEY="your-openai-key"
export DOUBAO_API_KEY="your-doubao-key"
export KIMI_API_KEY="your-kimi-key"
# 修改监控的品牌
monitor = GEOMultiPlatformMonitor("你的品牌")
# 修改查询词
queries = ["推荐一个 XX 工具", "XX 服务商哪家好"]
python geo_monitor.py
用这套系统监控 GEO：
| 指标 | 之前 | 之后 |
|---|---|---|
| 监控平台 | 1 个 | 4 个+ |
| 手动操作 | 30 分钟 | 0 分钟 |
| 数据留存 | 无 | JSON 永久保存 |
| 告警机制 | 无 | 可配置 |
从单平台监控，变成多平台监控。
核心改进：
完整代码可以直接使用，只需要配置 API Key 即可。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online