知识图谱构建
与品牌语义网络建模

# 品牌实体与关系
(CloudShield, 提供, 云安全平台)
(CloudShield, 位于, 深圳)
(CloudShield, 获得, ISO27001认证)
(云安全平台, 包含, 威胁检测引擎)
(云安全平台, 支持, 等保2.0合规)

# 实体属性
(CloudShield, 成立时间, "2018年")
(CloudShield, 员工规模, "500+")
(CloudShield, 融资轮次, "B轮")

import spacy
from collections import defaultdict

# 加载中文NLP模型
nlp = spacy.load("zh_core_web_trained")

# 品牌实体类型映射
ENTITY_TYPE_MAP = {
    "ORG": "品牌/组织",
    "PRODUCT": "产品",
    "PERSON": "人物",
    "GPE": "地理位置",
    "TECH": "技术术语",
}

def extract_entities(text: str) -> dict:
    """从文本中抽取品牌相关实体"""
    doc = nlp(text)
    entities = defaultdict(list)

    for ent in doc.ents:
        entity_type = ENTITY_TYPE_MAP.get(ent.label_, "其他")
        entities[entity_type].append({
            "text": ent.text,
            "label": ent.label_,
            "start": ent.start_char,
            "end": ent.end_char,
        })

    return dict(entities)

# 关系抽取：基于依存句法分析
def extract_relations(text: str) -> list:
    """从文本中抽取实体间关系"""
    doc = nlp(text)
    relations = []

    for token in doc:
        if token.dep_ in ["nsubj", "dobj", "pobj"]:
            subject = [t for t in token.head.lefts if t.ent_type_]
            obj = [t for t in token.head.rights if t.ent_type_]
            if subject and obj:
                relations.append({
                    "subject": subject[0].text,
                    "predicate": token.head.lemma_,
                    "object": obj[0].text,
                })

    return relations

# 查询所有获得ISO认证的品牌
PREFIX brand: <http://example.org/brand/>
PREFIX cert: <http://example.org/certification/>

SELECT ?brand ?brandName ?certName ?date
WHERE {
    ?brand a brand:Brand ;
           brand:name ?brandName ;
           brand:hasCertification ?cert .
    ?cert cert:name ?certName ;
          cert:date ?date .
    FILTER (CONTAINS(?certName, "ISO"))
}
ORDER BY DESC(?date)

{
  "@context": "https://schema.org",
  "@graph": [
    {
      "@id": "#brand",
      "@type": "Organization",
      "name": "CloudShield",
      "description": "企业级云安全解决方案提供商",
      "foundingDate": "2018",
      "address": {
        "@type": "PostalAddress",
        "addressLocality": "深圳",
        "addressCountry": "CN"
      },
      "makesOffer": { "@id": "#product" }
    },
    {
      "@id": "#product",
      "@type": "SoftwareApplication",
      "name": "CloudShield 云安全平台",
      "applicationCategory": "SecurityApplication",
      "featureList": [
        "威胁检测引擎",
        "合规审计系统",
        "数据加密服务"
      ]
    }
  ]
}

// 创建品牌知识图谱
CREATE (b:Brand {name:'CloudShield', founded:2018, location:'深圳'})
CREATE (p:Product {name:'云安全平台', type:'SaaS'})
CREATE (c:Certification {name:'ISO 27001', issuer:'ISO'})
CREATE (i:Industry {name:'金融', standard:'等保2.0'})
CREATE (b)-[:PROVIDES]->(p)
CREATE (b)-[:HAS_CERT]->(c)
CREATE (p)-[:APPLICABLE_TO]->(i)

// 查询：找到提供满足等保2.0的云安全产品的品牌
MATCH (b:Brand)-[:PROVIDES]->(p:Product)-[:APPLICABLE_TO]->(i:Industry)
WHERE i.standard = '等保2.0'
RETURN b.name AS 品牌, p.name AS 产品, i.name AS 适用行业

// 查询：品牌完整画像（含所有关联实体）
MATCH (b:Brand {name:'CloudShield'})-[r]-(n)
RETURN b.name AS 品牌,
       type(r) AS 关系类型,
       labels(n)[0] AS 实体类型,
       n.name AS 关联实体

<!-- 组织信息：部署在首页 -->
<script type="application/ld+json">
{
  "@context": "https://schema.org",
  "@type": "Organization",
  "name": "CloudShield",
  "url": "https://cloudshield.example.com",
  "logo": "https://cloudshield.example.com/logo.png",
  "description": "企业级云安全解决方案提供商",
  "foundingDate": "2018",
  "founder": {
    "@type": "Person",
    "name": "张明",
    "jobTitle": "CEO"
  },
  "address": {
    "@type": "PostalAddress",
    "addressLocality": "深圳",
    "addressRegion": "广东",
    "addressCountry": "CN"
  },
  "sameAs": [
    "https://github.com/cloudshield",
    "https://twitter.com/cloudshield"
  ]
}
</script>

<!-- 产品信息：部署在产品页面 -->
<script type="application/ld+json">
{
  "@context": "https://schema.org",
  "@type": "SoftwareApplication",
  "name": "CloudShield 云安全平台",
  "applicationCategory": "SecurityApplication",
  "operatingSystem": "Cloud",
  "description": "企业级云安全平台，提供威胁检测、合规审计、数据加密等一体化安全防护",
  "featureList": [
    "实时威胁检测引擎",
    "自动化合规审计",
    "端到端数据加密",
    "等保2.0合规支持"
  ],
  "offers": {
    "@type": "Offer",
    "priceCurrency": "CNY",
    "price": "9800",
    "priceUnit": "年"
  },
  "provider": {
    "@type": "Organization",
    "name": "CloudShield"
  },
  "aggregateRating": {
    "@type": "AggregateRating",
    "ratingValue": "4.8",
    "reviewCount": "256"
  }
}
</script>

# 品牌支柱页面
/                              → Brand: CloudShield
/about/                        → Brand: 关于我们
/team/                         → Brand: 团队

# 产品集群页面
/products/                     → Product: 产品总览
/products/cloud-security/      → Product: 云安全平台
/products/compliance/          → Product: 合规审计系统

# 认证集群页面
/certifications/               → Certification: 资质总览
/certifications/iso27001/      → Certification: ISO 27001

# 行业关联页面
/industries/                   → Industry: 行业解决方案
/industries/finance/           → Industry: 金融行业
/industries/healthcare/        → Industry: 医疗行业

from neo4j import GraphDatabase
from openai import OpenAI

class BrandGraphRAG:
    def __init__(self, neo4j_uri, neo4j_auth, openai_key):
        self.driver = GraphDatabase.driver(neo4j_uri, auth=neo4j_auth)
        self.llm = OpenAI(api_key=openai_key)

    def query_subgraph(self, entities: list, max_depth: int = 2):
        """从知识图谱中检索相关子图"""
        with self.driver.session() as session:
            result = session.run("""
                MATCH path = (b:Brand)-[*1..2]-(n)
                WHERE b.name IN $entities
                RETURN nodes(path) AS entities,
                       relationships(path) AS relations
            """, entities=entities)
            return [record.data() for record in result]

    def generate_answer(self, query: str):
        """GraphRAG完整流程"""
        # Step 1: 从查询中提取实体
        extracted = self.llm.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user",
                       "content": f"提取查询中的品牌实体: {query}"}]
        )
        entities = extracted.choices[0].message.content

        # Step 2: 图谱检索
        subgraph = self.query_subgraph(entities)

        # Step 3: 融合生成
        answer = self.llm.chat.completions.create(
            model="gpt-4",
            messages=[{
                "role": "system",
                "content": f"基于以下知识图谱数据回答: {subgraph}"
            }, {
                "role": "user",
                "content": query
            }]
        )
        return answer.choices[0].message.content

{
  "@context": {
    "@vocab": "https://industry.example.com/vocab/",
    "skos": "http://www.w3.org/2004/02/skos/core#"
  },
  "@graph": [
    {
      "@id": "cloud-security",
      "@type": "skos:Concept",
      "skos:prefLabel": "云安全",
      "skos:definition": "保护云计算环境中数据、应用和基础设施安全的技术与实践",
      "skos:narrower": [
        { "@id": "data-security" },
        { "@id": "threat-protection" },
        { "@id": "compliance-governance" }
      ]
    },
    {
      "@id": "data-security",
      "@type": "skos:Concept",
      "skos:prefLabel": "数据安全",
      "skos:broader": { "@id": "cloud-security" },
      "skos:related": [
        { "@id": "data-encryption" },
        { "@id": "zero-trust" }
      ]
    }
  ]
}