【实战】从零搭建GEO多平台监控系统：支持ChatGPT、豆包、Kimi、文心一言

【实战】从零搭建GEO多平台监控系统：支持ChatGPT、豆包、Kimi、文心一言

背景

Sora死了。

我的第一反应不是"AI完了"，而是"我的监控代码要不要改"。

因为之前我专门写了Sora的监控脚本。

Sora一关，代码废了。

痛定思痛，我决定写一套通用的GEO多平台监控方案。

本文分享完整代码，支持：ChatGPT、豆包、Kimi、文心一言、通义千问。

系统架构

┌─────────────────────────────────────────────────────────┐
│ GEO多平台监控系统 │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 任务调度 │→ │ 平台查询 │→ │ 结果分析 │ │
│ └───────────┘ └───────────┘ └───────────┘ │
│ ↑ ↓ ↓ │
│ └──── 告警通知 ←────── 报告生成 ←────── │
│ │
│ 支持平台： │
│ ✅ ChatGPT ✅ 豆包 ✅ Kimi │
│ ✅ 文心一言 ✅ 通义千问 │
└─────────────────────────────────────────────────────────┘

核心代码实现

1. 平台基类

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)
)

2. 具体平台实现

ChatGPT平台

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]["message"]["content"]

def extract_brand_info(self, response: str, brand: str) –> Dict:
"""提取品牌信息"""
mentioned = brand in response

if not mentioned:
return {"mentioned": False, "position": 0, "context": ""}

# 查找品牌在响应中的位置
position = response.index(brand)

# 提取上下文(前后各50字符)
start = max(0, position – 50)
end = min(len(response), position + len(brand) + 50)
context = response[start:end]

return {
"mentioned": True,
"position": 1, # 简化：出现即为位置1
"context": f"…{context}…"
}

豆包平台

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 in response

if not mentioned:
return {"mentioned": False, "position": 0, "context": ""}

position = response.index(brand)
start = max(0, position – 50)
end = min(len(response), position + len(brand) + 50)

return {
"mentioned": True,
"position": 1,
"context": f"…{response[start:end]}…"
}

Kimi平台

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 in response

if not mentioned:
return {"mentioned": False, "position": 0, "context": ""}

return {
"mentioned": True,
"position": 1,
"context": f"…{response[max(0, response.index(brand)–50):response.index(brand)+len(brand)+50]}…"
}

3. 多平台监控器

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:
"""生成监控报告"""
if not self.results:
return "暂无监控数据"

report = f"""# GEO多平台监控报告
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
监控品牌: {self.brand}

—

## 📊 总体概览

"""

total = len(self.results)
mentioned = sum(1 for r in self.results if r.mentioned)
success = sum(1 for r in self.results if not r.error)

report += f"- 总查询次数: {total}\\n"
report += f"- 品牌出现次数: {mentioned}\\n"
report += f"- **整体可见性: {mentioned/total*100:.1f}%**\\n"
report += f"- 查询成功率: {success/total*100:.1f}%\\n\\n"

# 分平台统计
report += "## 🏠 分平台数据\\n\\n"

for platform_name in set(r.platform for r in self.results):
platform_results = [r for r in self.results if r.platform == platform_name]
p_mentioned = sum(1 for r in platform_results if r.mentioned)
p_total = len(platform_results)
p_success = sum(1 for r in platform_results if not r.error)

emoji = "✅" if p_mentioned > 0 else "❌"
report += f"### {emoji} {platform_name}\\n"
report += f"- 查询次数: {p_total}\\n"
report += f"- 品牌出现: {p_mentioned}次\\n"
report += f"- 可见性: {p_mentioned/p_total*100:.1f}%\\n"
report += f"- 成功率: {p_success/p_total*100:.1f}%\\n\\n"

# 错误统计
errors = [r for r in self.results if r.error]
if errors:
report += "## ⚠️ 错误统计\\n\\n"
for r in errors:
report += f"- {r.platform} – {r.query}: {r.error}\\n"

return report

def save_to_json(self, filename: str = None):
"""保存结果到JSON"""
if not filename:
filename = f"geo_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"

data = [
{
"platform": r.platform,
"query": r.query,
"brand_mentioned": r.mentioned,
"position": r.position,
"context": r.context,
"timestamp": r.timestamp,
"response_time_ms": r.response_time_ms,
"error": r.error
}
for r in self.results
]

with open(filename, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)

print(f"报告已保存: {filename}")

4. 定时任务配置

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)

使用方法

1. 安装依赖

pip install requests schedule

2. 配置API Key

# 设置环境变量
export OPENAI_API_KEY="your-openai-key"
export DOUBAO_API_KEY="your-doubao-key"
export KIMI_API_KEY="your-kimi-key"

3. 修改品牌名和查询词

# 修改监控的品牌
monitor = GEOMultiPlatformMonitor("你的品牌")

# 修改查询词
queries = [
"推荐一个XX工具",
"XX服务商哪家好"
]

4. 运行

python geo_monitor.py

效果

用这套系统监控GEO：

总结

Sora死了，但监控代码升级了。

从单平台监控，变成多平台监控。

核心改进：

• 平台解耦，易于扩展
• 数据结构化，便于分析
• 定时任务，自动运行
• 报告生成，直观展示

完整代码可以直接使用，只需要配置API Key即可。

有问题欢迎评论区交流。

#Python #GEO #多平台监控 #ChatGPT #Kimi #豆包