【实战】从零搭建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 @dataclassclassGEOResult:"""GEO查询结果""" platform:str query:str brand:str mentioned:bool position:int context:str timestamp:str response_time_ms:float error: Optional[str]=NoneclassBasePlatform(ABC):"""AI平台基类"""def__init__(self, name:str, api_key:str=None): self.name = name self.api_key = api_key self.base_delay =1.0# 基础延迟(秒)@abstractmethoddefquery(self, search_query:str)->str:"""执行查询，返回AI响应文本"""pass@abstractmethoddefextract_brand_info(self, response:str, brand:str)-> Dict:"""从响应中提取品牌相关信息"""passdefcheck_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)*1000return 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)*1000return 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 classChatGPTPlatform(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"defquery(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"]defextract_brand_info(self, response:str, brand:str)-> Dict:"""提取品牌信息""" mentioned = brand in response ifnot 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}..."}豆包平台
classDoubaoPlatform(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"defquery(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"]defextract_brand_info(self, response:str, brand:str)-> Dict:"""提取品牌信息""" mentioned = brand in response ifnot 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平台
classKimiPlatform(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"defquery(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"]defextract_brand_info(self, response:str, brand:str)-> Dict:"""提取品牌信息""" mentioned = brand in response ifnot 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 classGEOMultiPlatformMonitor:"""GEO多平台监控器"""def__init__(self, brand:str): self.brand = brand self.platforms: List[BasePlatform]=[] self.results: List[GEOResult]=[]defadd_platform(self, platform: BasePlatform):"""添加监控平台""" self.platforms.append(platform)defcheck_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 defgenerate_report(self)->str:"""生成监控报告"""ifnot self.results:return"暂无监控数据" report =f"""# GEO多平台监控报告 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 监控品牌: {self.brand} --- ## 📊 总体概览 """ total =len(self.results) mentioned =sum(1for r in self.results if r.mentioned) success =sum(1for r in self.results ifnot 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 inset(r.platform for r in self.results): platform_results =[r for r in self.results if r.platform == platform_name] p_mentioned =sum(1for r in platform_results if r.mentioned) p_total =len(platform_results) p_success =sum(1for r in platform_results ifnot r.error) emoji ="✅"if p_mentioned >0else"❌" 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 defsave_to_json(self, filename:str=None):"""保存结果到JSON"""ifnot 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 ]withopen(filename,"w", encoding="utf-8")as f: json.dump(data, f, ensure_ascii=False, indent=2)print(f"报告已保存: {filename}")4. 定时任务配置
import schedule defdaily_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自动检查")# 运行whileTrue: schedule.run_pending() time.sleep(60)使用方法
1. 安装依赖
pip install requests schedule 2. 配置API Key
# 设置环境变量exportOPENAI_API_KEY="your-openai-key"exportDOUBAO_API_KEY="your-doubao-key"exportKIMI_API_KEY="your-kimi-key"3. 修改品牌名和查询词
# 修改监控的品牌 monitor = GEOMultiPlatformMonitor("你的品牌")# 修改查询词 queries =["推荐一个XX工具","XX服务商哪家好"]4. 运行
python geo_monitor.py 效果
用这套系统监控GEO：
| 指标 | 之前 | 之后 |
|---|---|---|
| 监控平台 | 1个 | 4个+ |
| 手动操作 | 30分钟 | 0分钟 |
| 数据留存 | 无 | JSON永久保存 |
| 告警机制 | 无 | 可配置 |
总结
Sora死了，但监控代码升级了。
从单平台监控，变成多平台监控。
核心改进：
- 平台解耦，易于扩展
- 数据结构化，便于分析
- 定时任务，自动运行
- 报告生成，直观展示
完整代码可以直接使用，只需要配置API Key即可。
有问题欢迎评论区交流。
#Python #GEO #多平台监控 #ChatGPT #Kimi #豆包
