AI搜索引擎的核心能力是从海量网页中提取实体并构建实体关系网络。当一个用户在Perplexity搜索CE认证的LED面板灯B2B供应商时AI引擎需要理解CE认证、LED面板灯、B2B、供应商四个实体之间的关系并在其知识库中找到匹配的供应链节点。如果外贸跨境电商站点仅提供扁平的产品页面AI引擎需要大量推理才能建立关联而如果站点主动提供知识图谱Knowledge GraphAI引擎可以直接读取结构化的实体关系数据引用效率和准确率将大幅提升。本文详解如何使用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.pyimport jsonfrom typing import List, Dict, Tuplefrom openai import OpenAIfrom neo4j import GraphDatabasefrom dataclasses import dataclassimport logginglogging.basicConfig(levellogging.INFO)logger logging.getLogger(__name__)dataclassclass Entity:name: strtype: str # Product, Category, Certification, Material, Application, Marketproperties: Dictdataclassclass Relation:source: strtarget: strtype: str # BELONGS_TO, CERTIFIED_BY, MADE_OF, USED_IN, SOLD_TO, COMPATIBLE_WITHproperties: Dictclass LLMEntityExtractor:def __init__(self, api_key: str, model: str gpt-4o-mini):self.client OpenAI(api_keyapi_key)self.model modeldef extract_from_product(self, product_data: Dict) - Tuple[List[Entity], List[Relation]]:从产品数据中提取实体和关系prompt fAnalyze the following product data and extract entities and relationships for a knowledge graph.Product Data:{json.dumps(product_data, indent2, ensure_asciiFalse)}Extract entities of types: Product, Category, Certification, Material, Application, Market, BrandExtract relationships: BELONGS_TO, CERTIFIED_BY, MADE_OF, USED_IN, SOLD_TO, COMPATIBLE_WITH, MANUFACTURED_BYReturn 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 entitiesresponse self.client.chat.completions.create(modelself.model,messages[{role: system, content: You are a knowledge graph engineer specializing in e-commerce product ontologies.},{role: user, content: prompt}],temperature0.3,response_format{type: json_object},max_tokens2000)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(fExtracted {len(entities)} entities, {len(relations)} relations)return entities, relationsclass 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(fCREATE 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([fn.{k} ${k} for k in props.keys()])cypher fMERGE (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([fr.{k} ${k} for k in props.keys()]) if props else cypher fMATCH (a {{name: $source}}), (b {{name: $target}})MERGE (a)-[r:{rel.type}]-(b){fSET {set_clause} if set_clause else }session.run(cypher, sourcerel.source, targetrel.target, **props)def get_product_knowledge_subgraph(self, product_name: str, depth: int 2) - Dict:获取产品的知识子图用于GEO结构化数据导出with self.driver.session() as session:cypher fMATCH path (p:Product {{name: $name}})-[*1..{depth}]-(related)WITH nodes(path) AS all_nodes, relationships(path) AS all_relsUNWIND all_nodes AS nodeWITH collect(DISTINCT node) AS unique_nodes, all_relsUNWIND all_rels AS relWITH unique_nodes, collect(DISTINCT rel) AS unique_relsRETURN[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 relationshipsresult session.run(cypher, nameproduct_name)record result.single()if record:return {nodes: record[nodes], relationships: record[relationships]}return {nodes: [], relationships: []}# 端到端Pipelinedef build_knowledge_graph(product_data_list: List[Dict]):extractor LLMEntityExtractor(api_keysk-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(fProcessed: {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.pyimport jsonfrom typing import Dict, Listfrom neo4j import GraphDatabaseclass 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-LDwith 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 relationsresult session.run(cypher, nameproduct_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:continuenode_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 jsonlddef _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 Thingentity {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] valuegraph.append(entity)return graphdef 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_jsonlddef 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, encodingutf-8) as f:json.dump(kg_document, f, ensure_asciiFalse, indent2)logger.info(fExported {len(all_products)} products to {output_path})return output_pathJSON-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.pyfrom neo4j import GraphDatabasefrom typing import Dict, Listfrom datetime import datetime, timedeltaimport scheduleimport timeclass GraphAnalytics:def __init__(self, uribolt://localhost:7687, userneo4j, passwordpassword):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 labelMATCH (n) WHERE label IN labels(n)RETURN label, count(n) AS countORDER 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 countORDER 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(fMATCH (n)-[r]-()WITH n, count(r) AS degree, labels(n) AS labelsWHERE degree 3RETURN n.name AS name, labels[0] AS type, degreeORDER BY degree DESCLIMIT {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 skuLIMIT 50)# 有Product但无SOLD_TO关系no_market session.run(MATCH (p:Product)WHERE NOT (p)-[:SOLD_TO]-()RETURN p.name AS product, p.sku AS skuLIMIT 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 100if stats[avg_degree] 3:health_score - 20if stats[isolated_nodes] stats[total_nodes] * 0.1:health_score - 15if len(gaps[missing_certifications]) 50:health_score - 25if len(gaps[missing_markets]) 50:health_score - 25return {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(fKnowledge 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开发、图分析运维等全栈技术咨询服务。
