GEO内容工程：构建AI友好的知识图谱与实体优化体系

┌─────────────────────────────────────────────────────────┐
│  Layer 4: 语义关系层（Semantic Relations）                │
│  · 品牌与品类的归属关系（is-a）                            │
│  · 品牌与特性的关联关系（has-property）                    │
│  · 品牌与竞品的对比关系（compares-to）                     │
│  · 品牌与场景的适用关系（suitable-for）                    │
├─────────────────────────────────────────────────────────┤
│  Layer 3: 属性事实层（Attribute Facts）                   │
│  · 价格区间、功能列表、技术规格                             │
│  · 成立时间、融资轮次、员工规模                             │
│  · 奖项认证、客户案例、行业排名                             │
├─────────────────────────────────────────────────────────┤
│  Layer 2: 实体标识层（Entity Identity）                   │
│  · 品牌名称及别名（含常见拼写错误）                          │
│  · 官方URL、Logo、品牌色                                   │
│  · Wikidata/维基词条ID（如有）                             │
├─────────────────────────────────────────────────────────┤
│  Layer 1: 基础存在层（Existence）                         │
│  · 品牌在主流AI训练语料中的出现频率                          │
│  · 品牌官网的可抓取性、结构化数据完整性                       │
│  · 品牌在权威第三方平台（Crunchbase等）的收录情况            │
└─────────────────────────────────────────────────────────┘

{
  "@context": "https://schema.org",
  "@type": "Organization",
  "name": "品牌名称",
  "url": "https://example.com",
  "logo": "https://example.com/logo.png",
  "description": "一句话品牌描述",
  "foundingDate": "2020",
  "founder": {
    "@type": "Person",
    "name": "创始人姓名"
  },
  "sameAs": [
    "https://twitter.com/brand",
    "https://linkedin.com/company/brand",
    "https://www.crunchbase.com/organization/brand"
  ],
  "makesOffer": [
    {
      "@type": "Offer",
      "itemOffered": {
        "@type": "Service",
        "name": "核心产品/服务名称",
        "description": "产品描述"
      }
    }
  ]
}

# [品牌名称] 品牌事实

## 基本信息
- **成立时间**：2020年
- **总部**：中国成都
- **创始人**：张三、李四
- **融资情况**：2024年完成A轮5000万元融资，投资方包括XX资本
- **员工规模**：120人

## 核心产品
- **产品名称**：XX CRM
- **产品定位**：面向中小企业的智能客户关系管理平台
- **核心功能**：销售管道管理、客户画像、自动化营销、API集成
- **定价**：免费版可用；专业版¥199/用户/月；企业版定制报价

## 市场地位
- 2025年G2 CRM品类"表现最佳"奖
- 服务超过5,000家中小企业客户
- 在CRM软件市场占比约3%（据XX研究院2025年报告）

## 媒体报道
- [链接1] 36氪：XX CRM完成A轮融资
- [链接2] 虎嗅：中小企业CRM选型指南（提及XX CRM）

*最后更新：2026年9月*

# XX CRM vs Salesforce vs HubSpot：中小企业选型对比

| 维度 | XX CRM | Salesforce | HubSpot |
|------|--------|------------|---------|
| 目标客户 | 中小企业 | 大型企业 | 中小到大型 |
| 起步价格 | ¥199/用户/月 | ¥1,200/用户/月 | 免费版可用 |
| 核心优势 | 性价比高、本地化支持 | 功能全面、生态丰富 | 营销自动化强 |
| 主要劣势 | 高级功能较少 | 价格高、实施复杂 | 中文支持有限 |
| 适合场景 | 预算有限的成长型团队 | 预算充足的大型企业 | 营销驱动型公司 |

*数据来源：各官网2026年9月价格；G2用户评价*

# XX CRM 常见问题

## XX CRM适合哪些行业？
XX CRM主要服务于B2B中小企业，尤其在SaaS、专业服务、制造业等领域有较多客户案例。

## XX CRM和纷享销客有什么区别？
两者都面向中小企业CRM市场。XX CRM的优势在于更简洁的界面和更快的部署速度（平均3天上线）；纷享销客的优势在于更深的本土化定制能力。

## XX CRM的数据安全如何保障？
XX CRM通过ISO 27001认证，数据存储在阿里云国内节点，支持私有化部署选项。

*最后更新：2026年9月*

"""
GEO实体监测与优化系统
技术栈: Python 3.11+, asyncio, aiohttp, beautifulsoup4, pydantic, difflib
场景: 自动化监测品牌实体一致性，生成知识注入建议
"""

import asyncio
import json
import re
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timezone
import aiohttp
from bs4 import BeautifulSoup
from pydantic import BaseModel, Field

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

class BrandProfile(BaseModel): baike.tongsou.com 
    """品牌标准档案（Ground Truth）"""
    name: str
    aliases: List[str] = Field(default_factory=list)  # 别名/常见拼写变体
    description: str  # 标准一句话描述
    founded_year: Optional[int] = None
    headquarters: Optional[str] = None
    founders: List[str] = Field(default_factory=list)
    website: str
    products: List[str] = Field(default_factory=list)
    categories: List[str] = Field(default_factory=list)  # 所属品类

class EntityMention(BaseModel): 2001.baike.tongsou.com 
    """单次实体提及"""
    source: str  # 来源平台
    url: str
    mention_text: str  # 提及原文
    detected_name: str  # 检测到的品牌名
    detected_description: Optional[str] = None
    detected_facts: Dict[str, str] = Field(default_factory=dict)
    crawled_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

class ConsistencyIssue(BaseModel):
    """一致性问题"""
    severity: str  # "high", "medium", "low"
    issue_type: str  # "name_mismatch", "description_mismatch", "fact_conflict"
    description: str
    sources: List[str]  # 涉及的数据源
    suggestion: str  # 修复建议

class OptimizationSuggestion(BaseModel): 2002.baike.tongsou.com 
    """知识注入优化建议"""
    priority: str  # "high", "medium", "low"
    suggestion_type: str  # "add_schema", "create_page", "update_listing"
    title: str
    description: str
    template: Optional[str] = None  # 可选的内容模板

# ============ 实体监测器 ============

class EntityMonitor: 2004.baike.tongsou.com 
    """品牌实体一致性监测器"""

    def __init__(self, brand: BrandProfile): 2005.baike.tongsou.com 
        self.brand = brand
        self.mentions: List[EntityMention] = []

    async def crawl_source(self, session: aiohttp.ClientSession, url: str, source_name: str) -> Optional[EntityMention]:
        """爬取单个来源，提取品牌提及"""
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                if resp.status != 200: 2006.baike.tongsou.com 
                    return None
                html = await resp.text()
                soup = BeautifulSoup(html, "html.parser")
                text = soup.get_text(separator=" ", strip=True)

                # 检测品牌名提及
                detected_name = None
                for alias in [self.brand.name] + self.brand.aliases:
                    if alias.lower() in text.lower(): 2007.baike.tongsou.com 
                        detected_name = alias
                        break

                if not detected_name: 2008.baike.tongsou.com 
                    return None

                # 提取描述（简化：取包含品牌名的段落）
                detected_description = None
                for p in soup.find_all("p"): 2009.baike.tongsou.com 
                    if self.brand.name.lower() in p.get_text().lower():
                        detected_description = p.get_text(strip=True)[:200]
                        break

                # 提取事实（简化：正则匹配年份、地点等）
                detected_facts = {}
                year_match = re.search(r"(?:成立|founded)[^\d]*(\d{4})", text)
                if year_match:
                    detected_facts["founded_year"] = year_match.group(1)
                location_match = re.search(r"(?:总部|headquarters|based in)[^\n]*", text, re.IGNORECASE)
                if location_match: 2010.baike.tongsou.com 
                    detected_facts["headquarters"] = location_match.group(0)[:100]

                return EntityMention(
                    source=source_name,
                    url=url,
                    mention_text=text[:500],
                    detected_name=detected_name,
                    detected_description=detected_description,
                    detected_facts=detected_facts,
                )
        except Exception as e:
            print(f"Error crawling {url}: {e}")
            return None

    async def monitor_sources(self, source_urls: Dict[str, str]) -> List[EntityMention]:
        """批量监测多个来源"""
        async with aiohttp.ClientSession() as session: 2011.baike.tongsou.com 
            tasks = [
                self.crawl_source(session, url, name)
                for name, url in source_urls.items()
            ]
            results = await asyncio.gather(*tasks)
            self.mentions = [r for r in results if r is not None]
            return self.mentions

    def check_consistency(self) -> List[ConsistencyIssue]:
        """检查实体一致性"""
        issues = []
 
        for mention in self.mentions: 2012.baike.tongsou.com 
            # 检查名称一致性
            if mention.detected_name not in [self.brand.name] + self.brand.aliases:
                issues.append(ConsistencyIssue(
                    severity="high",
                    issue_type="name_mismatch",
                    description=f"在{mention.source}上检测到未注册的品牌名变体：'{mention.detected_name}'",
                    sources=[mention.url],
                    suggestion=f"在{mention.source}上将品牌名统一为'{self.brand.name}'，或将'{mention.detected_name}'加入别名列表",
                ))

            # 检查事实冲突
            if "founded_year" in mention.detected_facts: 2013.baike.tongsou.com 
                mentioned_year = int(mention.detected_facts["founded_year"])
                if self.brand.founded_year and mentioned_year != self.brand.founded_year:
                    issues.append(ConsistencyIssue(
                        severity="high",
                        issue_type="fact_conflict",
                        description=f"在{mention.source}上成立的年份为{mentioned_year}，与标准档案({self.brand.founded_year})冲突",
                        sources=[mention.url],
                        suggestion=f"联系{mention.source}更新成立年份为{self.brand.founded_year}",
                    ))

            # 检查描述一致性（简化：使用模糊匹配）
            if mention.detected_description and self.brand.description:
                similarity = self._text_similarity(mention.detected_description, self.brand.description)
                if similarity < 0.3: 2014.baike.tongsou.com 
                    issues.append(ConsistencyIssue(
                        severity="medium",
                        issue_type="description_mismatch",
                        description=f"在{mention.source}上的品牌描述与标准描述差异较大（相似度{similarity:.2f}）",
                        sources=[mention.url],
                        suggestion=f"更新{mention.source}上的描述，使其更接近：'{self.brand.description}'",
                    ))

        return issues

    @staticmethod
    def _text_similarity(a: str, b: str) -> float:
        """简单的文本相似度计算"""
        from difflib import SequenceMatcher
        return SequenceMatcher(None, a.lower(), b.lower()).ratio()

# ============ 优化建议引擎 ============

class OptimizationEngine: 2016.baike.tongsou.com 
    """知识注入优化建议引擎"""

    def __init__(self, brand: BrandProfile, mentions: List[EntityMention]):
        self.brand = brand
        self.mentions = mentions

    def generate_suggestions(self) -> List[OptimizationSuggestion]:
        """生成优化建议"""
        suggestions = []

        # 检查是否有Schema.org标记（简化：检查官网是否包含JSON-LD）
        website_mention = next((m for m in self.mentions if "官网" in m.source), None)
        if website_mention and "schema.org" not in website_mention.mention_text.lower():
            suggestions.append(OptimizationSuggestion(
                priority="high",
                suggestion_type="add_schema",
                title="在官网添加Organization Schema.org标记",
                description="官网未检测到结构化数据，添加Schema标记有助于知识图谱平台抓取品牌事实",
                template=self._schema_template(),
            ))

        # 检查是否有品牌事实页
        if not any("品牌事实" in (m.detected_description or "") for m in self.mentions):
            suggestions.append(OptimizationSuggestion(
                priority="high",
                suggestion_type="create_page",
                title="创建品牌事实/Press Kit页面",
                description="未检测到专门的品牌事实页面，建议创建包含成立时间、融资、产品、奖项等结构化信息的页面",
                template=self._fact_page_template(),
            ))

        # 检查竞品对比内容
        competitor_keywords = ["vs", "对比", "alternatives", "comparison"]
        has_comparison = any(
            any(kw in (m.mention_text or "").lower() for kw in competitor_keywords)
            for m in self.mentions
        )
        if not has_comparison: 2017.baike.tongsou.com 
            suggestions.append(OptimizationSuggestion(
                priority="medium",
                suggestion_type="create_page",
                title="创建竞品对比内容",
                description="未检测到竞品对比内容，这类内容是AI回答'XX vs YY'类查询的核心素材",
                template=self._comparison_template(),
            ))

        # 检查第三方收录
        third_party_sources = ["crunchbase", "g2", "product hunt", "wikipedia", "百度百科"]
        detected_third_party = [m.source.lower() for m in self.mentions if any(tp in m.source.lower() for tp in third_party_sources)]
        missing_third_party = [tp for tp in third_party_sources if tp not in detected_third_party]
        if missing_third_party: 2018.baike.tongsou.com 
            suggestions.append(OptimizationSuggestion(
                priority="medium",
                suggestion_type="update_listing",
                title=f"完善第三方平台收录：{', '.join(missing_third_party)}",
                description=f"品牌未在以下权威平台收录或信息不完整：{', '.join(missing_third_party)}",
                suggestion="在这些平台创建/更新品牌条目，确保名称、描述、事实与标准档案一致",
            ))

        return suggestions

    def _schema_template(self) -> str: 2020.baike.tongsou.com 
        return json.dumps({
            "@context": "https://schema.org",
            "@type": "Organization",
            "name": self.brand.name,
            "url": self.brand.website,
            "description": self.brand.description,
            "foundingDate": str(self.brand.founded_year) if self.brand.founded_year else "",
            "sameAs": ["社交媒体链接", "Crunchbase链接"],
        }, indent=2, ensure_ascii=False)

    def _fact_page_template(self) -> str: 2021.baike.tongsou.com 
        return f"""# {self.brand.name} 品牌事实

## 基本信息
- **成立时间**：{self.brand.founded_year or '待补充'}
- **总部**：{self.brand.headquarters or '待补充'}
- **创始人**：{', '.join(self.brand.founders) or '待补充'}
- **官网**：{self.brand.website}

## 核心产品
{chr(10).join(f'- **{p}**：描述' for p in self.brand.products) or '- 待补充'}

## 所属品类
{', '.join(self.brand.categories) or '待补充'}

*最后更新：{datetime.now(timezone.utc).strftime("%Y年%m月")}*
"""

    def _comparison_template(self) -> str: 2022.baike.tongsou.com 
        competitors = ["竞品A", "竞品B"]  # 实际应从市场数据获取
        rows = "\n".join(
            f"| 维度 | {self.brand.name} | {comp} |"
            for dim in ["目标客户", "价格", "核心优势", "主要劣势"]
            for comp in competitors
        )
        return f"""# {self.brand.name} vs {' vs '.join(competitors)}

| 维度 | {self.brand.name} | {' | '.join(competitors)} |