从零搭建支持 ChatGPT 等多平台的 GEO 监控系统

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, "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：

总结

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

核心改进：

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

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