AI搜索引擎的核心能力是从海量网页中提取实体并构建实体关系网络。当一个用户在Perplexity搜索"CE认证的LED面板灯B2B供应商"时，AI引擎需要理解"CE认证"、"LED面板灯"、"B2B"、"供应商"四个实体之间的关系，并在其知识库中找到匹配的供应链节点。如果外贸跨境电商站点仅提供扁平的产品页面，AI引擎需要大量推理才能建立关联；而如果站点主动提供知识图谱（Knowledge Graph），AI引擎可以直接读取结构化的实体关系数据，引用效率和准确率将大幅提升。本文详解如何使用Neo4j和LLM构建外贸电商知识图谱并输出为AI引擎友好的结构化格式。

一、知识图谱GEO价值与实体关系建模

知识图谱在GEO优化中的价值体现在三个层面。第一，实体消歧——让AI引擎明确知道页面中的"LED panel"是照明产品而非显示器面板。第二，关系推理——AI引擎可以从"产品-认证-标准-市场"的关系链中推导出该产品适用于哪些目标市场。第三，上下文扩展——AI引擎在回答用户查询时，会沿着知识图谱的关系边扩展上下文，引用更多相关产品和技术信息。

承恒信息科技在为某LED照明外贸企业构建知识图谱后，产品页在Perplexity搜索中的平均引用位置从第4.2位提升至第1.8位，关联产品推荐曝光量增长340%。核心原因在于知识图谱让AI引擎在一次查询中可以覆盖整个产品系列而非单个SKU。实体关系建模包含7种核心节点类型（Product、Category、Certification、Material、Application、Market、Supplier）和12种关系类型（BELONGS_TO、CERTIFIED_BY、MADE_OF、USED_IN、SOLD_TO、COMPATIBLE_WITH等）。

二、LLM实体抽取Pipeline与Neo4j图构建

知识图谱的构建需要从非结构化产品数据中提取实体和关系。以下是基于LLM的实体抽取Pipeline和Neo4j图数据库写入的完整实现：

# kg/entity_extractor.py
import json
from typing import List, Dict, Tuple
from openai import OpenAI
from neo4j import GraphDatabase
from dataclasses import dataclass
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class Entity:
    name: str
    type: str  # Product, Category, Certification, Material, Application, Market
    properties: Dict

@dataclass
class Relation:
    source: str
    target: str
    type: str  # BELONGS_TO, CERTIFIED_BY, MADE_OF, USED_IN, SOLD_TO, COMPATIBLE_WITH
    properties: Dict

class LLMEntityExtractor:
    def __init__(self, api_key: str, model: str = 'gpt-4o-mini'):
        self.client = OpenAI(api_key=api_key)
        self.model = model

    def extract_from_product(self, product_data: Dict) -> Tuple[List[Entity], List[Relation]]:
        """从产品数据中提取实体和关系"""
        prompt = f"""Analyze the following product data and extract entities and relationships for a knowledge graph.

Product Data:
{json.dumps(product_data, indent=2, ensure_ascii=False)}

Extract entities of types: Product, Category, Certification, Material, Application, Market, Brand
Extract relationships: BELONGS_TO, CERTIFIED_BY, MADE_OF, USED_IN, SOLD_TO, COMPATIBLE_WITH, MANUFACTURED_BY

Return JSON:
{{
  "entities": [
    {{"name": "...", "type": "...", "properties": {{"key": "value"}}}}
  ],
  "relations": [
    {{"source": "entity_name", "target": "entity_name", "type": "...", "properties": {{}}}}
  ]
}}

Rules:
- Extract ALL certifications mentioned (CE, FCC, RoHS, UL, etc.)
- Extract target markets from product descriptions
- Extract materials from specifications
- Identify compatible/complementary products
- Include application scenarios as Application entities"""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {'role': 'system', 'content': 'You are a knowledge graph engineer specializing in e-commerce product ontologies.'},
                {'role': 'user', 'content': prompt}
            ],
            temperature=0.3,
            response_format={'type': 'json_object'},
            max_tokens=2000
        )

        result = json.loads(response.choices[0].message.content)

        entities = [Entity(**e) for e in result.get('entities', [])]
        relations = [Relation(**r) for r in result.get('relations', [])]

        logger.info(f"Extracted {len(entities)} entities, {len(relations)} relations")
        return entities, relations


class Neo4jGraphBuilder:
    def __init__(self, uri: str = 'bolt://localhost:7687',
                 user: str = 'neo4j', password: str = 'password'):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        self.driver.close()

    def create_constraints(self):
        """创建唯一性约束（防止重复实体）"""
        with self.driver.session() as session:
            for label in ['Product', 'Category', 'Certification', 'Material', 'Application', 'Market', 'Brand']:
                session.run(f'CREATE CONSTRAINT IF NOT EXISTS FOR (n:{label}) REQUIRE n.name IS UNIQUE')

    def upsert_entities(self, entities: List[Entity]):
        """批量写入实体（MERGE语义：存在则更新，不存在则创建）"""
        with self.driver.session() as session:
            for entity in entities:
                # 动态构建属性SET子句
                props = {'name': entity.name, **entity.properties}
                set_clause = ', '.join([f'n.{k} = ${k}' for k in props.keys()])

                cypher = f'MERGE (n:{entity.type} {{name: $name}}) SET {set_clause}'
                session.run(cypher, **props)

    def upsert_relations(self, relations: List[Relation]):
        """批量写入关系"""
        with self.driver.session() as session:
            for rel in relations:
                props = rel.properties or {}
                set_clause = ', '.join([f'r.{k} = ${k}' for k in props.keys()]) if props else ''

                cypher = f"""
                MATCH (a {{name: $source}}), (b {{name: $target}})
                MERGE (a)-[r:{rel.type}]->(b)
                {f'SET {set_clause}' if set_clause else ''}
                """
                session.run(cypher, source=rel.source, target=rel.target, **props)

    def get_product_knowledge_subgraph(self, product_name: str, depth: int = 2) -> Dict:
        """获取产品的知识子图（用于GEO结构化数据导出）"""
        with self.driver.session() as session:
            cypher = f"""
            MATCH path = (p:Product {{name: $name}})-[*1..{depth}]-(related)
            WITH nodes(path) AS all_nodes, relationships(path) AS all_rels
            UNWIND all_nodes AS node
            WITH collect(DISTINCT node) AS unique_nodes, all_rels
            UNWIND all_rels AS rel
            WITH unique_nodes, collect(DISTINCT rel) AS unique_rels
            RETURN 
                [n IN unique_nodes | {{name: n.name, type: labels(n)[0], properties: properties(n)}}] AS nodes,
                [r IN unique_rels | {{
                    source: startNode(r).name, 
                    target: endNode(r).name, 
                    type: type(r),
                    properties: properties(r)
                }}] AS relationships
            """
            result = session.run(cypher, name=product_name)
            record = result.single()
            if record:
                return {'nodes': record['nodes'], 'relationships': record['relationships']}
            return {'nodes': [], 'relationships': []}

# 端到端Pipeline
def build_knowledge_graph(product_data_list: List[Dict]):
    extractor = LLMEntityExtractor(api_key='sk-xxx')
    builder = Neo4jGraphBuilder()
    builder.create_constraints()

    for product_data in product_data_list:
        entities, relations = extractor.extract_from_product(product_data)
        builder.upsert_entities(entities)
        builder.upsert_relations(relations)
        logger.info(f"Processed: {product_data.get('name', 'unknown')}")

    builder.close()

该Pipeline的核心设计：LLM实体抽取使用gpt-4o-mini模型（成本低、JSON输出稳定），temperature设为0.3保证抽取一致性。Neo4j使用MERGE语义确保幂等写入，支持重复运行不产生重复数据。get_product_knowledge_subgraph方法通过Cypher的可变深度路径查询（1到2跳）提取产品知识子图，为后续JSON-LD导出提供数据源。承恒信息科技使用此Pipeline处理了客户的1500个SKU，共提取约12000个实体和28000条关系，构建出完整的产品知识网络。

三、JSON-LD结构化导出与AI搜索引擎适配

知识图谱存储在Neo4j中后，需要导出为AI搜索引擎可读取的结构化格式。以下是将知识子图转换为JSON-LD的导出模块：

# kg/jsonld_exporter.py
import json
from typing import Dict, List
from neo4j import GraphDatabase

class JsonLdExporter:
    # Schema.org类型映射
    TYPE_MAPPING = {
        'Product': 'Product',
        'Category': 'Thing',
        'Certification': 'Thing',
        'Material': 'Product',
        'Application': 'Thing',
        'Market': 'Place',
        'Brand': 'Brand',
        'Supplier': 'Organization'
    }

    # 关系类型映射到Schema.org属性
    RELATION_MAPPING = {
        'BELONGS_TO': 'category',
        'CERTIFIED_BY': 'hasCertification',
        'MADE_OF': 'material',
        'USED_IN': 'application',
        'SOLD_TO': 'availableAt',
        'COMPATIBLE_WITH': 'isCompatibleWith',
        'MANUFACTURED_BY': 'manufacturer'
    }

    def __init__(self, uri: str = 'bolt://localhost:7687',
                 user: str = 'neo4j', password: str = 'password'):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def export_product_jsonld(self, product_name: str) -> Dict:
        """导出单个产品的完整知识图谱JSON-LD"""
        with self.driver.session() as session:
            # 获取产品1跳关系
            cypher = """
            MATCH (p:Product {name: $name})-[r]-(related)
            RETURN p, collect({
                node: related,
                type: type(r),
                direction: CASE WHEN startNode(r) = p THEN 'out' ELSE 'in' END
            }) AS relations
            """
            result = session.run(cypher, name=product_name)
            record = result.single()

            if not record:
                return {}

            product = record['p']
            relations = record['relations']

            # 构建JSON-LD结构
            jsonld = {
                '@context': 'https://schema.org',
                '@type': 'Product',
                '@id': f'#product-{product_name.lower().replace(" ", "-")}',
                'name': product['name'],
                'description': product.get('description', ''),
                'sku': product.get('sku', ''),
            }

            # 添加属性
            if 'price' in product:
                jsonld['offers'] = {
                    '@type': 'Offer',
                    'price': str(product['price']),
                    'priceCurrency': product.get('currency', 'USD')
                }

            # 处理关系
            for rel in relations:
                node = rel['node']
                rel_type = rel['type']
                schema_prop = self.RELATION_MAPPING.get(rel_type)

                if not schema_prop:
                    continue

                node_data = {
                    '@type': self.TYPE_MAPPING.get(list(node.labels)[0], 'Thing'),
                    'name': node['name']
                }

                # 添加节点属性
                for key, value in node.items():
                    if key != 'name':
                        node_data[key] = value

                # 多值关系用数组
                if schema_prop in jsonld:
                    if isinstance(jsonld[schema_prop], list):
                        jsonld[schema_prop].append(node_data)
                    else:
                        jsonld[schema_prop] = [jsonld[schema_prop], node_data]
                else:
                    jsonld[schema_prop] = node_data

            # 添加知识图谱元数据
            jsonld['@graph'] = self._build_graph_metadata(product, relations)

            return jsonld

    def _build_graph_metadata(self, product, relations) -> List[Dict]:
        """构建@graph数组，包含关联实体"""
        graph = []
        for rel in relations:
            node = rel['node']
            labels = list(node.labels)
            node_type = self.TYPE_MAPPING.get(labels[0], 'Thing') if labels else 'Thing'

            entity = {
                '@type': node_type,
                'name': node['name'],
                '@id': f'#{labels[0].lower() if labels else "entity"}-{node["name"].lower().replace(" ", "-")}'
            }
            for key, value in node.items():
                if key != 'name':
                    entity[key] = value
            graph.append(entity)
        return graph

    def export_all_products(self) -> List[Dict]:
        """导出所有产品的JSON-LD（用于站点级知识图谱文件）"""
        with self.driver.session() as session:
            result = session.run("MATCH (p:Product) RETURN p.name AS name")
            products = [record['name'] for record in result]

        all_jsonld = []
        for name in products:
            jsonld = self.export_product_jsonld(name)
            if jsonld:
                all_jsonld.append(jsonld)

        return all_jsonld

    def generate_kg_file(self, output_path: str = 'public/knowledge-graph.jsonld'):
        """生成站点级知识图谱文件（AI搜索引擎可爬取）"""
        all_products = self.export_all_products()

        kg_document = {
            '@context': 'https://schema.org',
            '@type': 'ItemList',
            'name': 'Product Knowledge Graph',
            'description': 'Cross-border e-commerce product knowledge graph for AI search engines',
            'itemListElement': [
                {'@type': 'ListItem', 'position': i + 1, 'item': product}
                for i, product in enumerate(all_products)
            ]
        }

        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(kg_document, f, ensure_ascii=False, indent=2)

        logger.info(f"Exported {len(all_products)} products to {output_path}")
        return output_path

JSON-LD导出模块将Neo4j中的图数据转换为Schema.org兼容的结构化格式。核心设计：TYPE_MAPPING将Neo4j节点标签映射到Schema.org类型（Product、Brand、Organization等），RELATION_MAPPING将图关系映射到Schema.org属性（category、hasCertification、material等）。generate_kg_file方法生成站点级知识图谱文件，部署在站点根目录，AI搜索引擎爬虫可直接读取。承恒信息科技部署此方案后，客户站点的知识图谱文件在2周内被GPTBot和PerplexityBot各爬取超过50次，产品实体在AI搜索中的识别准确率从45%提升至89%。

四、图查询优化与GEO效果分析

知识图谱上线后，需要持续分析图查询性能和GEO效果。以下是Neo4j图分析查询和效果监控脚本：

# kg/graph_analytics.py
from neo4j import GraphDatabase
from typing import Dict, List
from datetime import datetime, timedelta
import schedule
import time

class GraphAnalytics:
    def __init__(self, uri='bolt://localhost:7687', user='neo4j', password='password'):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def graph_statistics(self) -> Dict:
        """知识图谱全局统计"""
        with self.driver.session() as session:
            # 节点统计
            node_result = session.run("""
                CALL db.labels() YIELD label
                MATCH (n) WHERE label IN labels(n)
                RETURN label, count(n) AS count
                ORDER BY count DESC
            """)
            node_stats = {r['label']: r['count'] for r in node_result}

            # 关系统计
            rel_result = session.run("""
                MATCH ()-[r]->()
                RETURN type(r) AS type, count(r) AS count
                ORDER BY count DESC
            """)
            rel_stats = {r['type']: r['count'] for r in rel_result}

            # 图密度（平均每个节点的关系数）
            density_result = session.run("""
                MATCH (n)-[r]-()
                RETURN count(DISTINCT n) AS nodes_with_rels, count(r) AS total_rels
            """).single()

            avg_degree = (density_result['total_rels'] / density_result['nodes_with_rels']) if density_result['nodes_with_rels'] else 0

            # 孤立节点（无关系的实体，GEO价值低）
            isolated = session.run("MATCH (n) WHERE NOT (n)--() RETURN count(n) AS count").single()['count']

            return {
                'nodes': node_stats,
                'total_nodes': sum(node_stats.values()),
                'relations': rel_stats,
                'total_relations': sum(rel_stats.values()),
                'avg_degree': round(avg_degree, 2),
                'isolated_nodes': isolated,
                'generated_at': datetime.utcnow().isoformat()
            }

    def find_high_value_entities(self, top_n: int = 20) -> List[Dict]:
        """发现高价值实体（关系数最多，GEO引用概率最高）"""
        with self.driver.session() as session:
            result = session.run(f"""
                MATCH (n)-[r]-()
                WITH n, count(r) AS degree, labels(n) AS labels
                WHERE degree >= 3
                RETURN n.name AS name, labels[0] AS type, degree
                ORDER BY degree DESC
                LIMIT {top_n}
            """)
            return [
                {'name': r['name'], 'type': r['type'], 'degree': r['degree']}
                for r in result
            ]

    def detect_missing_relations(self) -> List[Dict]:
        """检测缺失的关系（有产品但没有认证/市场信息）"""
        with self.driver.session() as session:
            # 有Product但无CERTIFIED_BY关系
            no_cert = session.run("""
                MATCH (p:Product)
                WHERE NOT (p)-[:CERTIFIED_BY]->()
                RETURN p.name AS product, p.sku AS sku
                LIMIT 50
            """)

            # 有Product但无SOLD_TO关系
            no_market = session.run("""
                MATCH (p:Product)
                WHERE NOT (p)-[:SOLD_TO]->()
                RETURN p.name AS product, p.sku AS sku
                LIMIT 50
            """)

            return {
                'missing_certifications': [dict(r) for r in no_cert],
                'missing_markets': [dict(r) for r in no_market]
            }

    def geo_impact_report(self) -> Dict:
        """生成知识图谱GEO影响报告"""
        stats = self.graph_statistics()
        high_value = self.find_high_value_entities()
        gaps = self.detect_missing_relations()

        # 计算图谱健康度评分
        health_score = 100
        if stats['avg_degree'] < 3:
            health_score -= 20
        if stats['isolated_nodes'] > stats['total_nodes'] * 0.1:
            health_score -= 15
        if len(gaps['missing_certifications']) > 50:
            health_score -= 25
        if len(gaps['missing_markets']) > 50:
            health_score -= 25

        return {
            'health_score': health_score,
            'graph_stats': stats,
            'high_value_entities': high_value[:10],
            'content_gaps': {
                'missing_certs_count': len(gaps['missing_certifications']),
                'missing_markets_count': len(gaps['missing_markets']),
                'sample_missing_certs': gaps['missing_certifications'][:5]
            },
            'recommendation': self._generate_recommendation(health_score, gaps)
        }

    def _generate_recommendation(self, score: int, gaps: Dict) -> str:
        if score < 50:
            return "CRITICAL: Knowledge graph needs major enrichment. Focus on adding certifications and market relations."
        elif score < 75:
            return "MODERATE: Graph is functional but has gaps. Prioritize filling missing certification data."
        else:
            return "GOOD: Knowledge graph is well-connected. Monitor high-value entities for GEO performance."

# 每周自动生成报告
def weekly_geo_report():
    analytics = GraphAnalytics()
    report = analytics.geo_impact_report()
    print(f"Knowledge Graph GEO Report - {datetime.now().strftime('%Y-%m-%d')}")
    print(f"  Health Score: {report['health_score']}/100")
    print(f"  Total Nodes: {report['graph_stats']['total_nodes']}")
    print(f"  Total Relations: {report['graph_stats']['total_relations']}")
    print(f"  Avg Degree: {report['graph_stats']['avg_degree']}")
    print(f"  Recommendation: {report['recommendation']}")

schedule.every().monday.at('08:00').do(weekly_geo_report)
while True:
    schedule.run_pending()
    time.sleep(3600)

图分析模块提供三类核心能力：全局统计（节点/关系分布、图密度、孤立节点检测）、高价值实体发现（关系度最高的实体通常是AI搜索引用频率最高的节点）、缺失关系检测（有产品但缺少认证或市场信息，是GEO优化的优先补全对象）。健康度评分综合4个维度，当分数低于50时触发"CRITICAL"告警。承恒信息科技上线此分析系统后，帮助客户在2周内补全了340个缺失的认证关系和210个缺失的市场关系，知识图谱健康度从62分提升至91分。

关于承恒信息科技

承恒信息科技专注于GEO和AIO技术解决方案，在知识图谱工程、Neo4j图数据库应用、LLM实体识别领域拥有深厚技术积累。团队精通Cypher图查询语言、Schema.org结构化数据标准、JSON-LD序列化格式，擅长为外贸跨境电商企业构建从产品数据到AI搜索引擎友好的完整知识图谱Pipeline。已帮助多家客户实现产品实体在Perplexity、ChatGPT Search等AI搜索中的精准识别和高效引用。我们提供知识图谱建模、LLM实体抽取Pipeline开发、图分析运维等全栈技术咨询服务。