企业如何建立品牌AI曝光度监测体系

┌────────────────────────────────────────────────────┐
│                 品牌AI曝光度监测体系                  │
├──────────┬──────────┬──────────┬────────────────────┤
│ 问题引擎  │ 采集引擎  │ 分析引擎  │   洞察与应用层      │
├──────────┼──────────┼──────────┼────────────────────┤
│·场景库   │·自动化   │·情感分析  │·可视化看板          │
│·问题库   │  提问采集 │·竞品识别  │·趋势预警            │
│·变体生成 │·多平台   │·事实核验  │·策略建议            │
│·优先级   │  适配    │·曝光度   │·自动报告            │
│  管理    │·反爬处理  │  计算    │·优化闭环            │
└──────────┴──────────┴──────────┴────────────────────┘

示例：
- “XX品牌是做什么的？”
- “XX品牌是什么档次？”
- “XX品牌的口碑怎么样？”
- “XX品牌是哪里的牌子？”
- “XX品牌靠谱吗？”

示例：
- “XX品牌有哪些产品？”
- “XX品牌的XX产品怎么样？”
- “XX品牌的XX产品适合什么肤质？”
- “XX品牌XX产品的成分安全吗？”
- “XX品牌XX产品多少钱？”

示例：
- “XX和YY哪个更好？”
- “预算500，推荐哪个牌子的面霜？”
- “敏感肌用什么牌子的洗面奶？”
- “2025年最值得买的母婴品牌有哪些？”
- “XX和YY的面膜有什么区别？”

示例：
- “XX品牌有负面新闻吗？”
- “XX品牌的用户真实评价怎么样？”
- “XX品牌的售后好吗？”
- “XX品牌被曝光过什么问题？”
- “XX品牌的成分是不是真的安全？”

示例：
- “宝宝湿疹用什么护肤品？”
- “夏天油皮用什么面霜不闷痘？”
- “产后脱发用什么洗发水？”
- “军训防晒用什么好？”
- “送女朋友什么护肤品好？”

from playwright.sync_api import sync_playwright
import time
import json
from datetime import datetime
from abc import ABC, abstractmethod

class AIPlatformMonitor(ABC):
    """AI平台监测抽象基类"""
    
    def __init__(self, platform_name, auth_file):
        self.platform_name = platform_name
        self.auth_file = auth_file
        self.browser = None
        self.context = None
        self.page = None
    
    def init_browser(self):
        """初始化浏览器实例"""
        self.playwright = sync_playwright().start()
        self.browser = self.playwright.chromium.launch(
            headless=True,
            args=['--disable-blink-features=AutomationControlled']
        )
        
        # 尝试加载已保存的登录态
        try:
            self.context = self.browser.new_context(
                storage_state=self.auth_file
            )
        except:
            self.context = self.browser.new_context()
        
        self.page = self.context.new_page()
    
    @abstractmethod
    def ask(self, question: str) -> dict:
        """向AI提问并获取回答，子类必须实现"""
        pass
    
    def batch_ask(self, questions: list, interval: int = 30) -> list:
        """批量提问"""
        results = []
        for i, q in enumerate(questions):
            print(f"[{self.platform_name}] 进度 {i+1}/{len(questions)}: {q[:40]}...")
            try:
                result = self.ask(q)
                results.append(result)
            except Exception as e:
                results.append({
                    "platform": self.platform_name,
                    "question": q,
                    "answer": None,
                    "error": str(e),
                    "timestamp": datetime.now().isoformat()
                })
            time.sleep(interval)  # 控制频率
        return results
    
    def close(self):
        """清理资源"""
        if self.page:
            self.page.close()
        if self.context:
            self.context.close()
        if self.browser:
            self.browser.close()
        if self.playwright:
            self.playwright.stop()


class DoubaoMonitor(AIPlatformMonitor):
    """豆包监测器"""
    
    def __init__(self):
        super().__init__("doubao", "doubao_auth.json")
    
    def ask(self, question: str) -> dict:
        self.page.goto("https://www.doubao.com/chat/", wait_until="networkidle")
        time.sleep(2)
        
        # 输入问题
        input_box = self.page.wait_for_selector(
            "textarea[placeholder*='输入']", timeout=10000
        )
        input_box.fill(question)
        
        # 点击发送
        send_btn = self.page.query_selector("button[type='submit']")
        if not send_btn:
            send_btn = self.page.query_selector(".send-btn")
        send_btn.click()
        
        # 等待回答完成（监测停止生成按钮消失）
        try:
            self.page.wait_for_selector(".stop-generating", state="detached", timeout=60000)
        except:
            time.sleep(15)  # 兜底等待
        
        # 提取回答
        answer_blocks = self.page.query_selector_all(".answer-content, .message-content")
        answer_text = answer_blocks[-1].inner_text() if answer_blocks else ""
        
        # 提取引用来源
        sources = []
        source_links = self.page.query_selector_all(".source-link a, .citation a")
        for link in source_links:
            href = link.get_attribute("href")
            if href:
                sources.append(href)
        
        return {
            "platform": self.platform_name,
            "question": question,
            "answer": answer_text,
            "sources": sources,
            "timestamp": datetime.now().isoformat(),
            "status": "success"
        }


class KimiMonitor(AIPlatformMonitor):
    """Kimi监测器"""
    
    def __init__(self):
        super().__init__("kimi", "kimi_auth.json")
    
    def ask(self, question: str) -> dict:
        self.page.goto("https://kimi.moonshot.cn/", wait_until="networkidle")
        time.sleep(2)
        
        # Kimi的输入框
        input_box = self.page.wait_for_selector(
            "div[contenteditable='true']", timeout=10000
        )
        input_box.click()
        input_box.fill(question)
        
        # 发送
        self.page.click("button[type='submit']")
        
        # 等待回答
        time.sleep(10)
        self.page.wait_for_selector(".kimi-answer", timeout=30000)
        
        answer = self.page.query_selector(".kimi-answer").inner_text()
        
        return {
            "platform": self.platform_name,
            "question": question,
            "answer": answer,
            "sources": self._extract_sources(),
            "timestamp": datetime.now().isoformat(),
            "status": "success"
        }
    
    def _extract_sources(self):
        """提取Kimi的引用来源"""
        sources = []
        elements = self.page.query_selector_all(".reference-item a")
        for el in elements:
            href = el.get_attribute("href")
            if href:
                sources.append(href)
        return sources


class MonitorOrchestrator:
    """监测协调器：统一调度多平台监测"""
    
    def __init__(self):
        self.monitors = {}
        self._register_monitors()
    
    def _register_monitors(self):
        """注册所有平台的监测器"""
        self.monitors["doubao"] = DoubaoMonitor()
        self.monitors["kimi"] = KimiMonitor()
        # 可扩展：文心一言、元宝、通义千问等
    
    def run_full_monitoring(self, questions: list) -> dict:
        """全平台执行监测"""
        all_results = {}
        
        for platform_name, monitor in self.monitors.items():
            print(f"\n{'='*50}")
            print(f"开始监测平台：{platform_name}")
            print(f"{'='*50}")
            
            try:
                monitor.init_browser()
                results = monitor.batch_ask(questions, interval=20)
                all_results[platform_name] = results
            except Exception as e:
                print(f"平台 {platform_name} 监测异常: {str(e)}")
                all_results[platform_name] = {"error": str(e)}
            finally:
                monitor.close()
        
        return all_results


# 使用示例
if __name__ == "__main__":
    orchestrator = MonitorOrchestrator()
    
    # 加载监测问题
    questions = [
        "XX品牌的口红怎么样？",
        "XX和YY哪个牌子更好？",
        "预算300元，推荐什么牌子的口红？",
        "XX品牌有什么负面评价吗？"
    ]
    
    results = orchestrator.run_full_monitoring(questions)
    
    # 保存结果
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    with open(f"monitoring_report_{timestamp}.json", "w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    
    print("\n全平台监测完成！")

# crontab 示例：每天早上9点和晚上9点执行监测
0 9 * * * cd /app && python run_monitor.py >> /var/log/monitor.log 2>&1
0 21 * * * cd /app && python run_monitor.py >> /var/log/monitor.log 2>&1

# 云函数版本（适配腾讯云函数）
def main_handler(event, context):
    """云函数入口"""
    orchestrator = MonitorOrchestrator()
    
    # 从环境变量或配置中心获取问题列表
    questions = json.loads(event.get("questions", "[]"))
    
    # 执行监测
    results = orchestrator.run_full_monitoring(questions)
    
    # 结果写入云数据库
    save_to_tencent_db(results)
    
    # 触发告警检查
    check_alerts(results)
    
    return {
        "code": 0,
        "message": f"监测完成，共{len(questions)}个问题，{len(results)}个平台",
        "timestamp": datetime.now().isoformat()
    }

# serverless.yml
functions:
  aiMonitor:
    handler: index.main_handler
    events:
      - timer:
          name: dailyMonitor
          cronExpression: "0 0 9 * * * *"  # 每天9点
          enable: true
      - timer:
          name: eveningMonitor
          cronExpression: "0 0 21 * * * *" # 每天21点
          enable: true

FROM python:3.11-slim

RUN apt-get update && apt-get install -y \
    chromium \
    chromium-driver \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

ENV PYTHONUNBUFFERED=1
ENV CHROMIUM_PATH=/usr/bin/chromium

CMD ["python", "run_monitor.py"]

import json
import openai
from datetime import datetime, timedelta
from collections import Counter, defaultdict

class ExposureAnalyzer:
    """AI曝光度分析引擎"""
    
    def __init__(self, llm_api_key: str):
        self.client = openai.OpenAI(api_key=llm_api_key)
    
    def analyze_answer(self, answer: str, brand_name: str, 
                       competitors: list) -> dict:
        """使用LLM对单个回答进行多维分析"""
        
        prompt = f"""
        请分析以下AI助手回答中关于品牌"{brand_name}"的曝光情况。
        
        回答内容：
        {answer}
        
        竞品列表：{competitors}
        
        请严格按JSON格式返回，包含以下字段：
        {{
            "brand_mentioned": true/false,
            "mention_type": "主动推荐/客观提及/对比提及/负面提及/未提及",
            "sentiment_score": -1.0到1.0之间的数值,
            "recommendation_strength": "强力推荐/推荐/中立/不推荐/强烈不推荐",
            "competitors_mentioned": ["被提及的竞品列表"],
            "brand_position": "首选/并列/次选/未被推荐",
            "key_descriptions": ["对品牌的关键描述词"],
            "positive_signals": ["正面信号"],
            "negative_signals": ["负面信号"],
            "factual_errors": [{{"claim": "错误陈述", "correction": "正确信息"}}],
            "citation_sources": ["引用来源域名"],
            "answer_quality": "高/中/低",
            "summary": "一句话总结本次曝光情况"
        }}
        """
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0.1  # 低温度保证一致性
        )
        
        return json.loads(response.choices[0].message.content)
    
    def calculate_exposure_index(self, analysis_results: list) -> dict:
        """计算综合曝光指数"""
        
        total = len(analysis_results)
        if total == 0:
            return {}
        
        mentioned_count = sum(1 for r in analysis_results if r.get("brand_mentioned"))
        
        # 曝光广度指数（被提及的比例）
        breadth = mentioned_count / total * 100
        
        # 曝光深度指数（推荐强度的加权平均）
        strength_map = {
            "强力推荐": 1.0, "推荐": 0.75, "中立": 0.5,
            "不推荐": 0.25, "强烈不推荐": 0.0, "未提及": 0.0
        }
        depth = sum(strength_map.get(r.get("recommendation_strength", "未提及"), 0) 
                    for r in analysis_results) / total * 100
        
        # 情感健康度
        sentiments = [r.get("sentiment_score", 0) for r in analysis_results 
                      if r.get("brand_mentioned")]
        avg_sentiment = sum(sentiments) / len(sentiments) if sentiments else 0
        sentiment_health = (avg_sentiment + 1) / 2 * 100  # 映射到0-100
        
        # 竞争定位指数
        preferred = sum(1 for r in analysis_results 
                       if r.get("brand_position") in ["首选", "并列"])
        competitive_position = preferred / total * 100 if total > 0 else 0
        
        # 综合曝光指数（加权综合）
        composite = (
            breadth * 0.25 +
            depth * 0.30 +
            sentiment_health * 0.25 +
            competitive_position * 0.20
        )
        
        return {
            "composite_exposure_index": round(composite, 2),
            "exposure_breadth": round(breadth, 2),
            "exposure_depth": round(depth, 2),
            "sentiment_health": round(sentiment_health, 2),
            "competitive_position": round(competitive_position, 2),
            "total_questions_analyzed": total,
            "brand_mentioned_times": mentioned_count
        }
    
    def detect_anomalies(self, current_results: list, 
                         historical_baseline: dict) -> list:
        """异常检测：发现显著变化"""
        alerts = []
        
        current_index = self.calculate_exposure_index(current_results)
        
        # 对比基线数据
        for metric in ["exposure_breadth", "exposure_depth", "sentiment_health"]:
            current_val = current_index.get(metric, 0)
            baseline_val = historical_baseline.get(metric, current_val)
            
            change_pct = ((current_val - baseline_val) / baseline_val * 100) if baseline_val > 0 else 0
            
            if abs(change_pct) > 20:
                direction = "上升" if change_pct > 0 else "下降"
                severity = "warning" if abs(change_pct) > 30 else "info"
                alerts.append({
                    "metric": metric,
                    "current_value": round(current_val, 2),
                    "baseline_value": round(baseline_val, 2),
                    "change_percentage": round(change_pct, 2),
                    "direction": direction,
                    "severity": severity,
                    "message": f"{metric} {direction}了{abs(round(change_pct, 2))}%"
                })
        
        return alerts

-- 原始回答存储表
CREATE TABLE ai_monitor_raw_answers (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    monitor_batch_id VARCHAR(64) NOT NULL COMMENT '批次ID',
    platform VARCHAR(50) NOT NULL COMMENT 'AI平台',
    brand_name VARCHAR(100) NOT NULL,
    question_category VARCHAR(50) COMMENT '问题分类',
    question_text TEXT NOT NULL,
    answer_text MEDIUMTEXT,
    sources JSON COMMENT '引用来源',
    monitor_time DATETIME NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_batch (monitor_batch_id),
    INDEX idx_brand_time (brand_name, monitor_time),
    INDEX idx_platform_time (platform, monitor_time)
) COMMENT 'AI监测原始回答表';

-- 分析结果表
CREATE TABLE ai_exposure_analysis (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    raw_answer_id BIGINT NOT NULL COMMENT '关联原始回答ID',
    brand_mentioned TINYINT DEFAULT 0,
    mention_type VARCHAR(50),
    sentiment_score DECIMAL(3,2),
    recommendation_strength VARCHAR(30),
    competitors_mentioned JSON,
    brand_position VARCHAR(20),
    key_descriptions JSON,
    positive_signals JSON,
    negative_signals JSON,
    factual_errors JSON,
    exposure_score DECIMAL(5,2) COMMENT '单次曝光评分',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (raw_answer_id) REFERENCES ai_monitor_raw_answers(id),
    INDEX idx_brand_mentioned (brand_mentioned),
    INDEX idx_sentiment (sentiment_score)
) COMMENT 'AI曝光分析结果表';

-- 曝光指数汇总表（按日/周/月汇总）
CREATE TABLE ai_exposure_summary (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    brand_name VARCHAR(100) NOT NULL,
    period_type ENUM('daily','weekly','monthly') NOT NULL,
    period_start DATE NOT NULL,
    platform VARCHAR(50) COMMENT '平台，ALL表示全平台',
    question_category VARCHAR(50) COMMENT '问题分类，ALL表示全分类',
    composite_index DECIMAL(5,2) COMMENT '综合曝光指数',
    exposure_breadth DECIMAL(5,2),
    exposure_depth DECIMAL(5,2),
    sentiment_health DECIMAL(5,2),
    competitive_position DECIMAL(5,2),
    total_questions INT,
    brand_mentioned_count INT,
    alerts_triggered INT DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_period (brand_name, period_type, period_start, platform, question_category),
    INDEX idx_brand_period (brand_name, period_start)
) COMMENT 'AI曝光指数汇总表';

-- Grafana常用查询示例

-- 1. 曝光指数趋势
SELECT 
    period_start as time,
    composite_index as value,
    platform as metric
FROM ai_exposure_summary 
WHERE brand_name = '$brand' 
  AND period_type = 'daily'
  AND period_start >= DATE_SUB(NOW(), INTERVAL 30 DAY)
ORDER BY period_start;

-- 2. 各平台推荐率对比
SELECT 
    platform,
    SUM(brand_mentioned_count) / SUM(total_questions) * 100 as mention_rate,
    AVG(competitive_position) as avg_position
FROM ai_exposure_summary
WHERE brand_name = '$brand'
  AND period_type = 'weekly'
  AND period_start >= DATE_SUB(NOW(), INTERVAL 4 WEEK)
GROUP BY platform;

-- 3. 问题分类曝光矩阵
SELECT 
    question_category,
    platform,
    AVG(composite_index) as avg_index
FROM ai_exposure_summary
WHERE brand_name = '$brand'
  AND question_category != 'ALL'
  AND period_type = 'monthly'
GROUP BY question_category, platform;

问题发现 → 问题分级 → 根因分析 → 制定方案 → 执行 → 效果验证
    ↑                                                      ↓
    └──────────────── 闭环反馈 ────────────────────────────┘