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, fieldfrom typing import List, Optional​dataclassclass Competitor:“”“竞品模型”“”competitor_id: strtenant_id: strname: str # 竞品名称aliases: List[str] # 别名简称、英文名等website: strindustry: strtracking_keywords: List[str] field(default_factorylist)tracking_platforms: List[str] field(default_factorylist)is_active: bool Truedef 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_idrow[competitor_id], tenant_idrow[tenant_id], namerow[name], aliasesrow[aliases] or [], websiterow[website], industryrow[industry], tracking_keywordsrow[tracking_keywords] or [], tracking_platformsrow[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( keywordprompt, platformadapter.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(), keylambda 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 dataclassfrom typing import Dict, List​dataclassclass VisibilityReport:“”“可见度报告”“”keyword: strplatform: strbrand_visibility: float # 0-1competitor_visibility: Dict[str, float] # {竞品名: 可见度}brand_rank: int # 品牌在提及顺序中的排名total_brands_mentioned: intshare_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( keywordcollection_data[keyword], platformcollection_data.get(platform, ), brand_visibilitybrand_visibility, competitor_visibilitycompetitor_visibility, brand_rankbrand_rank, total_brands_mentionedtotal, share_of_voiceshare_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​dataclassclass TopicCoverageReport:“”“主题覆盖报告”“”keyword: strbrand_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( keywordkeyword, brand_topicsbrand_topics, competitor_topicscompetitor_topics, unique_brand_topicsunique_brand, unique_competitor_topicsunique_competitor, overlap_topicsoverlap, coverage_gapcoverage_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}], temperature0.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-2周内可执行中期策略1-2个月长期策略3-6个月​每条策略请包含策略名称具体行动预期效果优先级高/中/低​格式JSON“”response await self.llm.chat( messages[{role: user, content: prompt}], temperature0.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 驱动的可执行竞争策略报告生成聚合多关键词的完整竞品报告