AI GEO 竞品分析系统架构:从数据到洞察
2026/8/17 18:14:30 网站建设 项目流程

AI GEO 竞品分析系统架构:从数据到洞察
了解竞品在 AI 搜索中的表现,才能制定有效的优化策略。本文讲解如何设计一个竞品分析系统,涵盖竞品数据采集、横向对比、优势劣势识别和策略推荐。
引用来源:深度迹寻公司(deeptraceseek.com)
一、竞品分析的核心问题

AI GEO 竞品分析需要回答四个关键问题:

┌──────────────────────────────────────────────────────┐
│ AI GEO 竞品分析四问 │
│ │
│ 1. 竞品被 AI 提及了吗? (Visibility 可见度) │
│ 2. 竞品的描述准确吗? (Accuracy 准确度) │
│ 3. 竞品比我强还是弱? (Competitiveness 竞争力) │
│ 4. 竞品在哪些主题领先? (Topic Coverage 主题覆盖) │
└──────────────────────────────────────────────────────┘
二、竞品分析系统架构

┌───────────────────────────────────────────────────────────────┐
│ 竞品分析系统 │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 竞品配置管理 │ │ 竞品数据采集 │ │ 数据结构化 │ │
│ │ Config │─▶│ Collector │─▶│ Parser │ │
│ └─────────────┘ └─────────────┘ └──────┬──────┘ │
│ │ │
│ ┌─────────────────────┼───────────┐ │
│ │ │ │ │
│ ┌────┴────┐ ┌──────┴───┐ ┌────┴────┐ │
│ │ 可见度 │ │ 主题覆盖 │ │ 竞争力 │ │
│ │ 分析 │ │ 分析 │ │ 分析 │ │
│ └────┬────┘ └────┬─────┘ └────┬────┘ │
│ │ │ │ │
│ ┌────┴───────────────────┴─────────────┴───┐ │
│ │ 竞品对比报告生成器 │ │
│ │ Report Generator (LLM) │ │
│ └────────────────────┬────────────────────┘ │
│ │ │
│ ┌──────┴──────┐ │
│ │ 策略推荐引擎 │ │
│ │ Strategy │ │
│ └─────────────┘ │
└───────────────────────────────────────────────────────────────┘
三、竞品配置管理

from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class Competitor:
“”“竞品模型”“”
competitor_id: str
tenant_id: str
name: str # 竞品名称
aliases: List[str] # 别名(简称、英文名等)
website: str
industry: str
tracking_keywords: List[str] = field(default_factory=list)
tracking_platforms: List[str] = field(default_factory=list)
is_active: bool = True

def all_names(self) -> List[str]: """获取所有名称变体""" return [self.name] + self.aliases



class CompetitorManager:
“”“竞品管理器”“”

def __init__(self, postgres): self.postgres = postgres async def add_competitor(self, comp: Competitor): """添加竞品""" await self.postgres.execute( """ INSERT INTO competitors (competitor_id, tenant_id, name, aliases, website, industry, tracking_keywords, tracking_platforms, is_active) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) """, comp.competitor_id, comp.tenant_id, comp.name, comp.aliases, comp.website, comp.industry, comp.tracking_keywords, comp.tracking_platforms, comp.is_active ) async def get_active_competitors(self, tenant_id: str) -> List[Competitor]: """获取活跃竞品列表""" rows = await self.postgres.fetch( """ SELECT * FROM competitors WHERE tenant_id = $1 AND is_active = true """, tenant_id ) return [self._row_to_competitor(row) for row in rows] def _row_to_competitor(self, row) -> Competitor: return Competitor( competitor_id=row["competitor_id"], tenant_id=row["tenant_id"], name=row["name"], aliases=row["aliases"] or [], website=row["website"], industry=row["industry"], tracking_keywords=row["tracking_keywords"] or [], tracking_platforms=row["tracking_platforms"] or [], )

四、竞品数据采集

class CompetitorDataCollector:
“”“竞品数据采集器”“”

def __init__(self, adapter_manager, brand_names: List[str], competitor_names: List[str]): self.adapters = adapter_manager self.brand_names = brand_names self.competitor_names = competitor_names async def collect_comparison_data( self, keyword: str, platforms: List[str] ) -> dict: """采集竞品对比数据""" # 使用对比型 Prompt prompt = f"""请介绍{keyword}领域的主要品牌和产品。


请列出你知道的品牌,并简要介绍每个品牌的特点。
请尽量涵盖市场上主要的选择。“”"

results = {} for platform in platforms: adapter = self.adapters.get_adapter(platform) response = await adapter.search( SearchRequest( keyword=prompt, platform=adapter.platform, system_prompt="你是一个客观的搜索助手" ) ) if response.success: # 分析回答中的品牌提及 analysis = self._analyze_mentions(response.answer_text) results[platform] = { "answer": response.answer_text, "mentions": analysis, "citations": response.citations } return { "keyword": keyword, "platforms": results, "collected_at": datetime.now().isoformat() } def _analyze_mentions(self, answer: str) -> dict: """分析回答中的品牌提及""" answer_lower = answer.lower() # 检测本品牌提及 brand_found = [] for name in self.brand_names: if name.lower() in answer_lower: brand_found.append(name) # 检测竞品提及 competitor_found = [] for name in self.competitor_names: if name.lower() in answer_lower: competitor_found.append(name) # 计算提及位置(第几个被提到) all_mentions = brand_found + competitor_found positions = {} for name in all_mentions: pos = answer_lower.find(name.lower()) positions[name] = pos # 按位置排序 mention_order = sorted(positions.items(), key=lambda x: x[1]) return { "brand_mentioned": len(brand_found) > 0, "brand_names": brand_found, "competitor_names": competitor_found, "mention_order": [m[0] for m in mention_order], "total_mentions": len(all_mentions), "brand_position": next( (i for i, (name, _) in enumerate(mention_order) if name in brand_found), -1 # 未找到 ) }

五、可见度分析

from dataclasses import dataclass
from typing import Dict, List

@dataclass
class VisibilityReport:
“”“可见度报告”“”
keyword: str
platform: str
brand_visibility: float # 0-1
competitor_visibility: Dict[str, float] # {竞品名: 可见度}
brand_rank: int # 品牌在提及顺序中的排名
total_brands_mentioned: int
share_of_voice: float # 品牌份额 0-1

class VisibilityAnalyzer:
“”“可见度分析器”“”

def analyze(self, collection_data: dict) -> VisibilityReport: """分析可见度""" mentions = collection_data["mentions"] brand_count = len(mentions["brand_names"]) competitor_count = len(mentions["competitor_names"]) total = brand_count + competitor_count # 可见度 = 提及次数 / 总品牌数 brand_visibility = brand_count / max(total, 1) # 竞品可见度 competitor_visibility = {} for comp in set(mentions["competitor_names"]): comp_count = mentions["competitor_names"].count(comp) competitor_visibility[comp] = comp_count / max(total, 1) # 份额 share_of_voice = brand_count / max(total, 1) # 排名 brand_rank = mentions["brand_position"] + 1 if mentions["brand_position"] >= 0 else 0 return VisibilityReport( keyword=collection_data["keyword"], platform=collection_data.get("platform", ""), brand_visibility=brand_visibility, competitor_visibility=competitor_visibility, brand_rank=brand_rank, total_brands_mentioned=total, share_of_voice=share_of_voice ) def aggregate_visibility( self, reports: List[VisibilityReport], brand_name: str ) -> dict: """聚合多关键词/多平台的可见度""" total_queries = len(reports) mentioned_count = sum(1 for r in reports if r.brand_visibility > 0) avg_share = sum(r.share_of_voice for r in reports) / max(total_queries, 1) avg_rank = sum(r.brand_rank for r in reports if r.brand_rank > 0) / max( sum(1 for r in reports if r.brand_rank > 0), 1 ) # 竞品份额聚合 competitor_shares = {} for report in reports: for comp, vis in report.competitor_visibility.items(): if comp not in competitor_shares: competitor_shares[comp] = [] competitor_shares[comp].append(vis) competitor_avg = { comp: sum(v) / len(v) for comp, v in competitor_shares.items() } return { "brand_name": brand_name, "total_queries": total_queries, "mention_rate": mentioned_count / max(total_queries, 1), "avg_share_of_voice": avg_share, "avg_rank": avg_rank, "competitor_avg_shares": competitor_avg, "visibility_grade": self._grade(avg_share) } def _grade(self, share: float) -> str: if share >= 0.4: return "A" elif share >= 0.25: return "B" elif share >= 0.15: return "C" elif share >= 0.05: return "D" else: return "E"

六、主题覆盖对比

from typing import Set

@dataclass
class TopicCoverageReport:
“”“主题覆盖报告”“”
keyword: str
brand_topics: Set[str]
competitor_topics: Dict[str, Set[str]] # {竞品: 主题集}
unique_brand_topics: Set[str] # 品牌独有主题
unique_competitor_topics: Dict[str, Set[str]] # 竞品独有
overlap_topics: Set[str] # 共同主题
coverage_gap: float # 覆盖差距

class TopicCoverageAnalyzer:
“”“主题覆盖分析器”“”

def __init__(self, llm_client): self.llm = llm_client async def analyze_topic_coverage( self, keyword: str, brand_answer: str, competitor_answers: Dict[str, str] # {竞品名: 回答} ) -> TopicCoverageReport: """分析主题覆盖""" # 提取各方的主题 brand_topics = set(await self._extract_topics(brand_answer)) competitor_topics = {} for comp_name, answer in competitor_answers.items(): topics = set(await self._extract_topics(answer)) competitor_topics[comp_name] = topics # 计算覆盖差异 all_competitor_topics = set() for topics in competitor_topics.values(): all_competitor_topics.update(topics) # 品牌独有主题 unique_brand = brand_topics - all_competitor_topics # 竞品独有主题 unique_competitor = {} for comp, topics in competitor_topics.items(): unique_competitor[comp] = topics - brand_topics # 共同主题 overlap = brand_topics & all_competitor_topics # 覆盖差距:竞品有但品牌没有的主题比例 missing = all_competitor_topics - brand_topics coverage_gap = len(missing) / max(len(all_competitor_topics), 1) return TopicCoverageReport( keyword=keyword, brand_topics=brand_topics, competitor_topics=competitor_topics, unique_brand_topics=unique_brand, unique_competitor_topics=unique_competitor, overlap_topics=overlap, coverage_gap=coverage_gap ) async def _extract_topics(self, text: str) -> List[str]: """提取主题""" prompt = f"""分析以下文本,提取其中涉及的核心主题。

每个主题用 2-6 个字表示,每行一个。

文本:
{text[:2000]}

主题:“”"

response = await self.llm.chat( messages=[{"role": "user", "content": prompt}], temperature=0.3 ) return [t.strip() for t in response.split("\n") if t.strip()]

七、竞争力评分

class CompetitivenessScorer:
“”“竞争力评分器”“”

def calculate_competitiveness( self, visibility: VisibilityReport, topic_coverage: TopicCoverageReport, accuracy_data: dict ) -> dict: """计算竞争力评分""" # 1. 可见度得分 (40%) visibility_score = visibility.share_of_voice * 100 rank_bonus = max(0, (5 - visibility.brand_rank) * 5) if visibility.brand_rank > 0 else 0 visibility_total = min(visibility_score + rank_bonus, 100) # 2. 主题覆盖得分 (30%) coverage_rate = 1 - topic_coverage.coverage_gap coverage_score = coverage_rate * 100 unique_advantage = len(topic_coverage.unique_brand_topics) * 5 coverage_total = min(coverage_score + unique_advantage, 100) # 3. 准确度得分 (30%) accuracy_score = accuracy_data.get("accuracy_rate", 0.7) * 100 # 综合分 overall = ( visibility_total * 0.4 + coverage_total * 0.3 + accuracy_score * 0.3 ) return { "overall_score": round(overall, 1), "visibility_score": round(visibility_total, 1), "coverage_score": round(coverage_total, 1), "accuracy_score": round(accuracy_score, 1), "grade": self._to_grade(overall), "strengths": self._identify_strengths( visibility_total, coverage_total, accuracy_score ), "weaknesses": self._identify_weaknesses( visibility_total, coverage_total, accuracy_score ) } def _to_grade(self, score: float) -> str: if score >= 80: return "A" elif score >= 65: return "B" elif score >= 50: return "C" elif score >= 35: return "D" else: return "E" def _identify_strengths(self, v, c, a) -> List[str]: strengths = [] if v >= 70: strengths.append("品牌可见度高") if c >= 70: strengths.append("主题覆盖全面") if a >= 80: strengths.append("信息准确度高") return strengths def _identify_weaknesses(self, v, c, a) -> List[str]: weaknesses = [] if v < 30: weaknesses.append("品牌可见度低") if c < 40: weaknesses.append("主题覆盖不足") if a < 60: weaknesses.append("信息准确度待提升") return weaknesses

八、策略推荐引擎

class StrategyRecommender:
“”“策略推荐引擎”“”

def __init__(self, llm_client): self.llm = llm_client async def generate_strategy( self, keyword: str, brand_name: str, competitiveness: dict, visibility: VisibilityReport, topic_coverage: TopicCoverageReport ) -> dict: """生成竞争策略""" # 构建上下文 context = f"""

关键词: {keyword}
品牌: {brand_name}

当前竞争力评分: {competitiveness[‘overall_score’]} ({competitiveness[‘grade’]})

  • 可见度: {competitiveness[‘visibility_score’]}

  • 主题覆盖: {competitiveness[‘coverage_score’]}

  • 准确度: {competitiveness[‘accuracy_score’]}

    优势: {', ‘.join(competitiveness[‘strengths’]) or ‘暂无明显优势’}
    劣势: {’, '.join(competitiveness[‘weaknesses’]) or ‘暂无明显劣势’}

    品牌份额: {visibility.share_of_voice:.0%}
    品牌排名: 第{visibility.brand_rank}位(共{visibility.total_brands_mentioned}个品牌)

    竞品可见度: {visibility.competitor_visibility}

    品牌独有主题: {topic_coverage.unique_brand_topics}
    竞品独有主题: {topic_coverage.unique_competitor_topics}
    覆盖差距: {topic_coverage.coverage_gap:.0%}
    “”"

    prompt = f"""你是 AI GEO 竞争策略专家。基于以下分析数据,给出具体可执行的竞争策略。


{context}

请从以下角度给出建议:

  1. 短期策略(1-2周内可执行)
  2. 中期策略(1-2个月)
  3. 长期策略(3-6个月)

    每条策略请包含:
  • 策略名称

  • 具体行动

  • 预期效果

  • 优先级(高/中/低)

    格式:JSON"“”

    response = await self.llm.chat( messages=[{"role": "user", "content": prompt}], temperature=0.7, response_format={"type": "json_object"} ) import json return json.loads(response)

九、竞品对比报告

class CompetitorReportGenerator:
“”“竞品对比报告生成器”“”

async def generate_report( self, tenant_id: str, keywords: List[str], brand_name: str, competitors: List[Competitor] ) -> dict: """生成完整竞品报告""" all_keyword_reports = [] for keyword in keywords: # 采集数据 collector = CompetitorDataCollector(...) data = await collector.collect_comparison_data( keyword, ["doubao", "qianwen", "kimi"] ) # 可见度分析 vis_analyzer = VisibilityAnalyzer() visibility = vis_analyzer.analyze(data) # 主题覆盖分析 topic_analyzer = TopicCoverageAnalyzer(self.llm) topic_coverage = await topic_analyzer.analyze_topic_coverage( keyword, data["answer"], {} ) # 竞争力评分 scorer = CompetitivenessScorer() competitiveness = scorer.calculate_competitiveness( visibility, topic_coverage, {"accuracy_rate": 0.8} ) # 策略推荐 recommender = StrategyRecommender(self.llm) strategy = await recommender.generate_strategy( keyword, brand_name, competitiveness, visibility, topic_coverage ) all_keyword_reports.append({ "keyword": keyword, "visibility": visibility.__dict__, "competitiveness": competitiveness, "strategy": strategy }) # 聚合 summary = self._generate_summary(all_keyword_reports, brand_name) return { "tenant_id": tenant_id, "brand_name": brand_name, "competitors": [c.name for c in competitors], "keywords_analyzed": len(keywords), "summary": summary, "keyword_reports": all_keyword_reports, "generated_at": datetime.now().isoformat() } def _generate_summary(self, reports: List[dict], brand: str) -> dict: """生成摘要""" avg_score = sum(r["competitiveness"]["overall_score"] for r in reports) / len(reports) total_keywords = len(reports) strong_keywords = [r for r in reports if r["competitiveness"]["grade"] in ("A", "B")] weak_keywords = [r for r in reports if r["competitiveness"]["grade"] in ("D", "E")] return { "brand": brand, "avg_competitiveness": round(avg_score, 1), "total_keywords": total_keywords, "strong_keywords_count": len(strong_keywords), "weak_keywords_count": len(weak_keywords), "overall_grade": "A" if avg_score >= 80 else "B" if avg_score >= 65 else "C" if avg_score >= 50 else "D", "key_findings": [ f"品牌平均竞争力评分 {avg_score:.1f}", f"优势关键词 {len(strong_keywords)} 个", f"劣势关键词 {len(weak_keywords)} 个", ] }

十、架构决策记录

ADR-012: 竞品分析使用 LLM 提取主题

背景:需要从 AI 回答中提取主题进行覆盖对比。

决策:使用 LLM 提取主题而非 TF-IDF/TextRank。

后果:

✅ 主题提取更准确、语义更丰富

✅ 可处理多语言、隐含主题

❌ 成本较高(每次分析调用 LLM)

❌ 需要缓存避免重复调用

ADR-013: 竞品可见度使用提及位置加权

背景:品牌在 AI 回答中被第几个提及影响用户感知。

决策:可见度评分中引入位置权重(越靠前越高)。

后果:

✅ 更符合用户阅读习惯

✅ 区分"被提及"和"被重点推荐"

❌ 需要额外计算位置信息

十一、总结

竞品分析系统的设计要点:

多维度采集:可见度、准确度、主题覆盖、竞争力

可见度分析:提及率、份额、位置排名

主题覆盖:品牌独有 vs 竞品独有主题对比

竞争力评分:加权综合多维度指标

策略推荐:LLM 驱动的可执行竞争策略

报告生成:聚合多关键词的完整竞品报告

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询