GEO自动化监测与度量系统：用代码量化品牌的“AI可见性得分”

AVS = (W1 × 提及率) + (W2 × 推荐率) + (W3 × 引用顺位得分) + (W4 × 情感/事实准确度)

默认权重建议（可根据业务调整）：
W1 (提及率) = 0.30  // 基础门槛：AI知道你
W2 (推荐率) = 0.35  // 核心转化：AI推荐你
W3 (引用顺位) = 0.20 // 注意力分配：你在回答中的位置
W4 (事实准确度) = 0.15 // 品牌安全：AI描述你的信息是否正确

[1. Prompt 策略库] 
   ├── 品类词 (如: "最好的CRM软件")
   ├── 场景词 (如: "适合中小企业的客户管理工具")
   ├── 竞品词 (如: "Salesforce的平替")
   └── 对比词 (如: "HubSpot vs 纷享销客")
          │
          ▼
[2. 并发请求引擎] ──调用──> [主流AI API (豆包/文心/Kimi/ChatGPT)]
          │
          ▼
[3. LLM 裁判模型 (Judge LLM)] 
   ├── 实体识别：是否提及目标品牌？
   ├── 意图分类：是中立提及还是正向推荐？
   ├── 顺位提取：在列表中的排名？
   └── 事实核查：描述是否有幻觉/错误？
          │
          ▼
[4. 数据聚合与AVS计算] ──输出──> [GEO 监测仪表盘]

"""
GEO自动化监测与AVS评估引擎
技术栈: Python 3.11+, asyncio, aiohttp, pydantic
场景: 批量测试Prompt，评估品牌在AI回答中的可见性得分(AVS)
依赖: pip install aiohttp pydantic
"""

import asyncio
import json
import re
from typing import List, Dict, Optional
from datetime import datetime, timezone
from pydantic import BaseModel, Field
import aiohttp

# ============ 数据模型 ============

class GeoPrompt(BaseModel): 5030.baike.tongsou.com 
    prompt_text: str
    category: str  # "category", "scenario", "competitor", "comparison"
    intent: str    # "informational", "transactional"

class AIResponse(BaseModel):
    prompt: str
    model_name: str
    response_text: str
    latency_ms: int

class EvaluationResult(BaseModel): 5038.baike.tongsou.com  
    prompt: str
    model_name: str
    is_mentioned: bool
    is_recommended: bool
    list_position: Optional[int] = None  # 如果在列表中，排第几（None表示不在列表中）
    fact_accuracy_score: float  # 0.0 - 1.0
    raw_evaluation: str

class AVSReport(BaseModel): 5052.baike.tongsou.com 
    brand_name: str
    test_date: str
    total_prompts: int
    mention_rate: float
    recommendation_rate: float
    avg_position_score: float
    avg_fact_accuracy: float
    final_avs_score: float

# ============ 核心引擎 ============

class GEOMonitorEngine: 5055.baike.tongsou.com 
    """GEO监测与评估引擎"""

    def __init__(self, brand_name: str, brand_facts: Dict, target_ai_api_url: str, judge_ai_api_url: str, api_key: str):
        self.brand_name = brand_name
        self.brand_facts = brand_facts  # 品牌的Ground Truth事实字典
        self.target_ai_api_url = target_ai_api_url  # 被监测的AI接口
        self.judge_ai_api_url = judge_ai_api_url    # 用于评估的裁判AI接口
        self.api_key = api_key

    async def fetch_ai_response(self, session: aiohttp.ClientSession, prompt: str, model_name: str) -> Optional[AIResponse]:
        """调用目标AI获取回答（此处为通用伪代码，需根据具体API替换）"""
        start_time = asyncio.get_event_loop().time(5057.baike.tongsou.com)
        payload = {
            "model": model_name,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.3  # 降低随机性以保证测试稳定性
        }
        headers = {"Authorization": f"Bearer {self.api_key}"}
        
        try:
            async with session.post(self.target_ai_api_url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                data = await resp.json(6004.baike.tongsou.com)
                latency = int((asyncio.get_event_loop().time() - start_time) * 1000)
                # 假设标准OpenAI格式返回
                text = data["choices"][0]["message"]["content"]
                return AIResponse(prompt=prompt, model_name=model_name, response_text=text, latency_ms=latency)
        except Exception as e:
            print(f"获取AI回答失败 [{model_name}]: {e}")
            return None

    async def evaluate_response(self, session: aiohttp.ClientSession, response: AIResponse) -> EvaluationResult:
        """使用Judge LLM评估AI回答"""
        # 构建裁判Prompt
        judge_prompt = f"""你是一个严格的GEO（生成式引擎优化）评估专家。
请评估以下AI回答中对品牌 "{self.brand_name}" 的表现。

【品牌标准事实】：{json.dumps(self.brand_facts, ensure_ascii=False)}
【用户Prompt】：{response.prompt}
【AI回答】：{response.response_text}

请严格按照以下JSON格式输出评估结果，不要输出其他内容： 6007.baike.tongsou.com 
{{
  "is_mentioned": true/false,  // AI是否明确提及了该品牌（全称或公认简称）
  "is_recommended": true/false, // AI是否给出了正向推荐（如"推荐"、"适合"、"优势"等）
  "list_position": null/整数,   // 如果AI输出了推荐列表（如Top 5），该品牌排第几？不在列表中填null
  "fact_accuracy_score": 0.0-1.0, // AI描述该品牌的事实与【标准事实】的吻合度（1.0为完全准确，0.0为严重幻觉）
  "raw_evaluation": "简短的评估理由"
}}"""

        payload = {
            "model": "gpt-4o-mini",  # Judge模型不需要最强，但需要指令遵循能力强
            "messages": [{"role": "user", "content": judge_prompt}],
            "temperature": 0.0,
            "response_format": { "type": "json_object" }
        }
        headers = {"Authorization": f"Bearer {self.api_key}"}

        try:
            async with session.post(self.judge_ai_api_url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=20)) as resp:
                data = await resp.json()
                eval_text = data["choices"][0]["message"]["content"]
                eval_data = json.loads(eval_text)
                
                return EvaluationResult(
                    prompt=response.prompt,
                    model_name=response.model_name,
                    is_mentioned=eval_data.get("is_mentioned", False),
                    is_recommended=eval_data.get("is_recommended", False),
                    list_position=eval_data.get("list_position"),
                    fact_accuracy_score=eval_data.get("fact_accuracy_score", 0.0),
                    raw_evaluation=eval_data.get("raw_evaluation", "")
                )
        except Exception as e: 6008.baike.tongsou.com 
            print(f"评估失败: {e}")
            return EvaluationResult(
                prompt=response.prompt, model_name=response.model_name,
                is_mentioned=False, is_recommended=False, list_position=None,
                fact_accuracy_score=0.0, raw_evaluation=f"评估异常: {str(e)}"
            )

    async def run_campaign(self, prompts: List[GeoPrompt], model_name: str) -> AVSReport:
        """执行完整的监测战役并生成AVS报告"""
        results: List[EvaluationResult] = []
        
        async with aiohttp.ClientSession() as session: 14009.baike.tongsou.com 
            # 1. 并发获取AI回答并评估 (限制并发数避免Rate Limit)
            sem = asyncio.Semaphore(5) 
            
            async def process_prompt(p: GeoPrompt): 14068.baike.tongsou.com 
                async with sem: 14013.baike.tongsou.com 
                    resp = await self.fetch_ai_response(session, p.prompt_text, model_name)
                    if resp:
                        eval_res = await self.evaluate_response(session, resp)
                        results.append(eval_res)

            await asyncio.gather(*[process_prompt(p) for p in prompts])

        # 2. 计算AVS指标
        total = len(results)
        if total == 0:
            raise ValueError("未获取到任何有效评估结果")

        mentions = sum(1 for r in results if r.is_mentioned)
        recommendations = sum(1 for r in results if r.is_recommended)
        
        # 顺位得分计算：排第1得1.0，第2得0.8，第3得0.6，第4得0.4，第5得0.2，未上榜0
        position_scores = []
        for r in results: 14070.baike.tongsou.com 
            if r.list_position is not None and 1 <= r.list_position <= 5:
                position_scores.append(1.0 - (r.list_position - 1) * 0.2)
            else:
                position_scores.append(0.0)
                
        fact_scores = [r.fact_accuracy_score for r in results]

        mention_rate = mentions / total
        recommendation_rate = recommendations / total
        avg_position_score = sum(position_scores) / total
        avg_fact_accuracy = sum(fact_scores) / total

        # 3. 计算最终AVS
        avs = (0.30 * mention_rate + 0.35 * recommendation_rate + 
               0.20 * avg_position_score + 0.15 * avg_fact_accuracy) * 100

        return AVSReport(
            brand_name=self.brand_name,
            test_date=datetime.now(timezone.utc).isoformat(),
            total_prompts=total,
            mention_rate=round(mention_rate, 4),
            recommendation_rate=round(recommendation_rate, 4),
            avg_position_score=round(avg_position_score, 4),
            avg_fact_accuracy=round(avg_fact_accuracy, 4),
            final_avs_score=round(avs, 2)
        )

# ============ 运行示例 ============
async def main(): 14071.baike.tongsou.com 
    # 品牌标准事实 (Ground Truth)
    brand_facts = {
        "name": "云图CRM",
        "company": "成都云图科技有限公司",
        "price": "专业版199元/用户/月",
        "target_audience": "中小企业",
        "core_features": ["销售管道", "客户画像", "自动化营销"]
    }

    # 测试Prompt库
    prompts = [
        GeoPrompt(prompt_text="推荐3款适合中小企业的CRM软件", category="scenario", intent="transactional"),
        GeoPrompt(prompt_text="国内有哪些好用的SaaS CRM？", category="category", intent="informational"),
        GeoPrompt(prompt_text="Salesforce太贵了，有什么平替推荐吗？", category="competitor", intent="transactional"),
        GeoPrompt(prompt_text="云图CRM和纷享销客哪个更好？", category="comparison", intent="transactional"),
    ]

    engine = GEOMonitorEngine(
        brand_name="云图CRM",
        brand_facts=brand_facts,
        target_ai_api_url="https://api.openai.com/v1/chat/completions", # 替换为目标AI API
        judge_ai_api_url="https://api.openai.com/v1/chat/completions",  # 替换为Judge AI API
        api_key="your-api-key-here"
    )

    report = await engine.run_campaign(prompts, model_name="gpt-4o")
    print("\n========== GEO AVS 监测报告 ==========")
    print(json.dumps(report.model_dump(), ensure_ascii=False, indent=2))

if __name__ == "__main__": 14080.baike.tongsou.com 
    asyncio.run(main())