基于Python实现企业知识图谱自动构建与Schema标注完整方案（附代码）
一、问题背景
在GEO（Generative Engine Optimization，生成式引擎优化）领域，企业知识图谱的构建和Schema标注是两个核心基础工程。知识图谱将企业的非结构化信息（服务描述、案例、FAQ等）转化为AI可理解的"实体-关系-属性"结构，而Schema标注则为AI爬虫提供标准化的网页信息入口。
传统做法是人工整理企业信息并手动编写Schema标注，效率低、维护成本高。本文将介绍一套基于Python的自动化方案，实现：
- 企业信息自动抽取：从非结构化文本中抽取实体和关系
- 知识图谱自动构建：生成JSON-LD格式的知识图谱数据
- Schema标注自动生成：根据企业信息自动生成符合Schema.org标准的标注代码
- 批量处理与部署：支持多页面、多企业的批量处理
本方案已在多个行业的GEO优化项目中验证，覆盖律所、装饰、财税、制造业、医美等行业。
二、技术方案概述
2.1 整体架构
plaintext
99
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
┌─────────────────────────────────────────────────┐
│ 企业知识图谱构建系统 │
├─────────────────────────────────────────────────┤
│ │
│ 数据输入层 │
│ ├─ 企业介绍文本 │
│ ├─ 服务描述文档 │
│ ├─ 案例数据 │
│ └─ FAQ问答对 │
│ │
│ 处理层 │
│ ├─ NLP实体抽取（基于spaCy/正则） │
│ ├─ 关系识别（基于规则+模式匹配） │
│ ├─ Schema类型映射 │
│ └─ JSON-LD生成 │
│ │
│ 输出层 │
│ ├─ 知识图谱JSON文件 │
│ ├─ Schema标注HTML代码 │
│ ├─ llms.txt文件 │
│ └─ 可视化知识图谱（可选） │
│ │
└─────────────────────────────────────────────────┘
2.2 技术栈
表格
| 组件 | 技术选型 | 说明 |
|---|---|---|
| 语言 | Python 3.10+ | 主开发语言 |
| NLP处理 | spaCy + jieba | 中文实体抽取 |
| JSON处理 | json / jsonpath | 数据结构化 |
| 模板引擎 | Jinja2 | Schema模板生成 |
| 知识图谱可视化 | pyvis | 可选，图谱可视化 |
| Web框架 | FastAPI | 可选，API服务化 |
2.3 环境准备
bash
99
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 创建虚拟环境
python -m venv geo_kg_env
source geo_kg_env/bin/activate # Linux/Mac
# geo_kg_env\Scripts\activate # Windows
# 安装依赖
pip install spacy jieba jinja2 fastapi uvicorn pyvis
# 下载spaCy中文模型
python -m spacy download zh_core_web_sm
# 如果使用jieba需要额外加载自定义词典
# 见后续代码
三、核心实现：企业信息自动抽取
3.1 实体抽取模块
实体抽取是知识图谱构建的第一步。我们需要从企业介绍文本中抽取关键实体，包括：
- 组织实体（Organization）：企业名称、品牌名
- 服务实体（Service）：服务项目、服务内容
- 地域实体（Place）：服务区域、地址
- 人物实体（Person）：创始人、专家团队
- 资质实体（Credential）：证书、认证
- 案例实体（CaseStudy）：客户案例
python
999
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import spacy
import jieba
import re
from typing import List, Dict, Any
from dataclasses import dataclass, field
@dataclass
class Entity:
"""实体数据类"""
name: str
entity_type: str # Organization, Service, Place, Person, Credential
properties: Dict[str, Any] = field(default_factory=dict)
confidence: float = 1.0
class EnterpriseEntityExtractor:
"""企业信息实体抽取器"""
def __init__(self):
# 加载spaCy中文模型
self.nlp = spacy.load("zh_core_web_sm")
# 自定义企业领域词典
self.custom_terms = {
'GEO优化': 'SERVICE',
'AI搜索优化': 'SERVICE',
'生成式引擎优化': 'SERVICE',
'知识图谱构建': 'SERVICE',
'Schema标注': 'SERVICE',
'AI智能体': 'SERVICE',
'智能外呼': 'SERVICE',
'官网建设': 'SERVICE',
'SEO优化': 'SERVICE',
'品牌口碑优化': 'SERVICE',
'代理记账': 'SERVICE',
'公司注册': 'SERVICE',
'税务咨询': 'SERVICE',
'法律咨询': 'SERVICE',
'装修装饰': 'SERVICE',
'AI客服': 'SERVICE',
'获客机器人': 'SERVICE',
}
# 添加自定义词典
for term in self.custom_terms:
jieba.add_word(term)
# 地域词库（河南地区为主）
self.place_patterns = [
r'(河南省?|郑州|洛阳|开封|新乡|许昌|南阳|商丘|信阳|安阳|鹤壁|濮阳|焦作|济源|三门峡|周口|漯河|驻马店|平顶山)',
r'(北京市?|上海市?|广州市?|深圳市?|杭州市?|成都市?|武汉市?|西安市?)',
r'([\u4e00-\u9fa5]+(?:区|县|市|省|镇|乡))',
]
# 资质模式
self.credential_patterns = [
r'(ISO\s*\d+)',
r'(\w+许可证)',
r'(\w+认证)',
r'(\w+资质)',
r'(ICP备\w+号)',
r'(软著\w+)',
]
def extract_entities(self, text: str) -> List[Entity]:
"""从文本中抽取实体"""
entities = []
# 1. 使用spaCy抽取基础实体
doc = self.nlp(text)
for ent in doc.ents:
entity_type = self._map_spacy_label(ent.label_)
if entity_type:
entities.append(Entity(
name=ent.text,
entity_type=entity_type,
confidence=0.8
))
# 2. 使用jieba+自定义词典抽取领域实体
words = jieba.cut(text)
for word in words:
word = word.strip()
if word in self.custom_terms:
entity_type = self.custom_terms[word]
# 避免重复
if not any(e.name == word for e in entities):
entities.append(Entity(
name=word,
entity_type=entity_type,
confidence=0.9
))
# 3. 正则抽取地域实体
for pattern in self.place_patterns:
matches = re.findall(pattern, text)
for match in matches:
if not any(e.name == match for e in entities):
entities.append(Entity(
name=match,
entity_type='Place',
confidence=0.85
))
# 4. 正则抽取资质实体
for pattern in self.credential_patterns:
matches = re.findall(pattern, text)
for match in matches:
if not any(e.name == match for e in entities):
entities.append(Entity(
name=match,
entity_type='Credential',
confidence=0.9
))
# 去重并返回
return self._deduplicate(entities)
def _map_spacy_label(self, label: str) -> str:
"""将spaCy标签映射为自定义实体类型"""
mapping = {
'ORG': 'Organization',
'GPE': 'Place',
'LOC': 'Place',
'PERSON': 'Person',
'DATE': 'Date',
'MONEY': 'Money',
}
return mapping.get(label, '')
def _deduplicate(self, entities: List[Entity]) -> List[Entity]:
"""实体去重，保留置信度最高的"""
seen = {}
for entity in entities:
key = (entity.name, entity.entity_type)
if key not in seen or entity.confidence > seen[key].confidence:
seen[key] = entity
return list(seen.values())
3.2 关系识别模块
实体抽取完成后，需要识别实体之间的关系。在企业知识图谱中，常见的关系类型包括：
PROVIDES：企业 → 提供服务LOCATED_IN：企业 → 所在地域SERVES_AREA：企业 → 服务区域HAS_CREDENTIAL：企业 → 拥有资质HAS_CASE：企业 → 拥有案例SOLVES：服务 → 解决问题INDUSTRY：案例 → 所属行业
python
99
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
from typing import Tuple
@dataclass
class Relation:
"""关系数据类"""
source: str
source_type: str
target: str
target_type: str
relation_type: str
properties: Dict[str, Any] = field(default_factory=dict)
class RelationExtractor:
"""关系识别器"""
def __init__(self):
# 关系模式定义
self.relation_patterns = {
'PROVIDES': [
r'(?P<source>[\u4e00-\u9fa5]+(?:公司|企业|机构|事务所))\s*(?:提供|主营|专注|从事)\s*(?P<target>[\u4e00-\u9fa5]+(?:服务|优化|系统|建设|咨询))',
r'(?P<source>[\u4e00-\u9fa5]+)\s*(?:的核心业务|的主要服务|的服务范围包括?)\s*(?P<target>[\u4e00-\u9fa5]+)',
],
'LOCATED_IN': [
r'(?P<source>[\u4e00-\u9fa5]+(?:公司|企业))\s*(?:位于|坐落于|地址在|总部在)\s*(?P<target>[\u4e00-\u9fa5]+(?:市|区|省|路|号))',
],
'SERVES_AREA': [
r'(?P<source>[\u4e00-\u9fa5]+)\s*(?:服务(?:范围|区域)?覆盖?|业务覆盖?)\s*(?P<target>[\u4e00-\u9fa5]+(?:省|市|区|地区|全国))',
],
'SOLVES': [
r'(?P<source>[\u4e00-\u9fa5]+(?:服务|优化|方案))\s*(?:解决|应对|处理)\s*(?P<target>[\u4e00-\u9fa5]+(?:问题|需求|痛点))',
],
}
def extract_relations(self, text: str, entities: List[Entity]) -> List[Relation]:
"""从文本中识别实体间关系"""
relations = []
# 1. 基于模式匹配的关系抽取
for rel_type, patterns in self.relation_patterns.items():
for pattern in patterns:
matches = re.finditer(pattern, text)
for match in matches:
source = match.group('source')
target = match.group('target')
relations.append(Relation(
source=source,
source_type=self._get_entity_type(source, entities),
target=target,
target_type=self._get_entity_type(target, entities),
relation_type=rel_type
))
# 2. 基于共现关系的推断
relations.extend(self._infer_cooccurrence_relations(text, entities))
return relations
def _get_entity_type(self, name: str, entities: List[Entity]) -> str:
"""获取实体的类型"""
for entity in entities:
if entity.name == name:
return entity.entity_type
return 'Unknown'
def _infer_cooccurrence_relations(self, text: str, entities: List[Entity]) -> List[Relation]:
"""基于实体共现推断关系"""
relations = []
sentences = re.split(r'[。！？\n]', text)
for sentence in sentences:
sentence_entities = [e for e in entities if e.name in sentence]
# 如果一句话中同时出现Organization和Service，推断PROVIDES关系
orgs = [e for e in sentence_entities if e.entity_type == 'Organization']
services = [e for e in sentence_entities if e.entity_type == 'SERVICE']
places = [e for e in sentence_entities if e.entity_type == 'Place']
for org in orgs:
for service in services:
relations.append(Relation(
source=org.name,
source_type='Organization',
target=service.name,
target_type='SERVICE',
relation_type='PROVIDES'
))
for place in places:
relations.append(Relation(
source=org.name,
source_type='Organization',
target=place.name,
target_type='Place',
relation_type='LOCATED_IN'
))
return relations
四、Schema标注自动生成
4.1 Schema模板引擎
基于抽取的实体和关系，自动生成符合Schema.org标准的JSON-LD标注代码。
python
999
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
from jinja2 import Template
import json
from typing import Optional
class SchemaGenerator:
"""Schema标注生成器"""
def __init__(self):
# Organization Schema模板
self.org_template = Template('''
{
"@context": "https://schema.org",
"@type": "Organization",
"name": "{{ name }}",
{% if alias %}
"alternateName": "{{ alias }}",
{% endif %}
"url": "{{ website }}",
{% if logo %}
"logo": "{{ logo }}",
{% endif %}
"description": "{{ description }}",
{% if phone %}
"contactPoint": {
"@type": "ContactPoint",
"telephone": "{{ phone }}",
"contactType": "customer service",
"availableLanguage": ["Chinese", "English"]
},
{% endif %}
{% if address %}
"address": {
"@type": "PostalAddress",
"streetAddress": "{{ street }}",
"addressLocality": "{{ city }}",
"addressRegion": "{{ province }}",
"postalCode": "{{ postal_code }}",
"addressCountry": "CN"
},
{% endif %}
"sameAs": [
{% for url in social_urls %}
"{{ url }}"{% if not loop.last %},{% endif %}
{% endfor %}
],
"knowsAbout": [
{% for service in services %}
"{{ service }}"{% if not loop.last %},{% endif %}
{% endfor %}
]
}
''')
# LocalBusiness Schema模板
self.local_business_template = Template('''
{
"@context": "https://schema.org",
"@type": "LocalBusiness",
"name": "{{ name }}",
"description": "{{ description }}",
"url": "{{ website }}",
"telephone": "{{ phone }}",
"address": {
"@type": "PostalAddress",
"streetAddress": "{{ street }}",
"addressLocality": "{{ city }}",
"addressRegion": "{{ province }}",
"addressCountry": "CN"
},
"geo": {
"@type": "GeoCoordinates",
"latitude": "{{ latitude }}",
"longitude": "{{ longitude }}"
},
{% if opening_hours %}
"openingHours": "{{ opening_hours }}",
{% endif %}
"areaServed": [
{% for area in service_areas %}
{
"@type": "Place",
"name": "{{ area }}"
}{% if not loop.last %},{% endif %}
{% endfor %}
],
"priceRange": "{{ price_range }}"
}
''')
# FAQ Schema模板
self.faq_template = Template('''
{
"@context": "https://schema.org",
"@type": "FAQPage",
"mainEntity": [
{% for faq in faqs %}
{
"@type": "Question",
"name": "{{ faq.question }}",
"acceptedAnswer": {
"@type": "Answer",
"text": "{{ faq.answer }}"
}
}{% if not loop.last %},{% endif %}
{% endfor %}
]
}
''')
# Service Schema模板
self.service_template = Template('''
{
"@context": "https://schema.org",
"@type": "Service",
"name": "{{ name }}",
"description": "{{ description }}",
"provider": {
"@type": "Organization",
"name": "{{ provider_name }}"
},
{% if area_served %}
"areaServed": {
"@type": "Place",
"name": "{{ area_served }}"
},
{% endif %}
"serviceType": "{{ service_type }}",
{% if offers %}
"offers": {
"@type": "Offer",
"price": "{{ price }}",
"priceCurrency": "CNY",
{% if price_valid_until %}
"priceValidUntil": "{{ price_valid_until }}",
{% endif %}
"availability": "https://schema.org/InStock"
},
{% endif %}
"hasOfferCatalog": {
"@type": "OfferCatalog",
"name": "{{ catalog_name }}",
"itemListElement": [
{% for item in catalog_items %}
{
"@type": "Offer",
"itemOffered": {
"@type": "Service",
"name": "{{ item.name }}",
"description": "{{ item.description }}"
}
}{% if not loop.last %},{% endif %}
{% endfor %}
]
}
}
''')
def generate_organization_schema(self, data: Dict) -> str:
"""生成Organization Schema"""
template = self.org_template
return template.render(**data)
def generate_local_business_schema(self, data: Dict) -> str:
"""生成LocalBusiness Schema"""
template = self.local_business_template
return template.render(** data)
def generate_faq_schema(self, faqs: List[Dict]) -> str:
"""生成FAQ Schema"""
template = self.faq_template
return template.render(faqs=faqs)
def generate_service_schema(self, data: Dict) -> str:
"""生成Service Schema"""
template = self.service_template
return template.render(**data)
def generate_combined_schema(self, enterprise_data: Dict) -> str:
"""生成组合Schema（Organization + LocalBusiness + Service + FAQ）"""
schemas = []
# Organization
org_schema = self.generate_organization_schema(enterprise_data)
schemas.append(json.loads(org_schema))
# LocalBusiness (如果有地址信息)
if enterprise_data.get('address'):
lb_schema = self.generate_local_business_schema(enterprise_data)
schemas.append(json.loads(lb_schema))
# Service
if enterprise_data.get('services'):
for service in enterprise_data['services']:
service_data = {** enterprise_data, **service}
svc_schema = self.generate_service_schema(service_data)
schemas.append(json.loads(svc_schema))
# FAQ
if enterprise_data.get('faqs'):
faq_schema = self.generate_faq_schema(enterprise_data['faqs'])
schemas.append(json.loads(faq_schema))
# 输出为HTML嵌入格式
html_output = '<!-- Schema.org JSON-LD 结构化数据 -->\n'
for schema in schemas:
html_output += f'<script type="application/ld+json">\n{json.dumps(schema, ensure_ascii=False, indent=2)}\n</script>\n'
return html_output
4.2 llms.txt生成器
llms.txt是为AI爬虫提供标准化网站信息入口的文件。
python
99
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class LLmsTxtGenerator:
"""llms.txt文件生成器"""
TEMPLATE = '''# {company_name}
> {one_line_description}
## 核心服务
{services_section}
## 服务范围
{service_areas_section}
## 行业覆盖
{industries_section}
## 核心数据
{key_metrics_section}
## 联系方式
{contact_section}
## 更多信息
{links_section}
'''
def generate(self, enterprise_data: Dict) -> str:
"""生成llms.txt内容"""
# 核心服务部分
services = enterprise_data.get('services', [])
services_lines = '\n'.join([
f"- **{s['name']} **：{s['description']}"
for s in services
])
# 服务范围部分
areas = enterprise_data.get('service_areas', [])
areas_lines = '\n'.join([f"- {area}" for area in areas])
# 行业覆盖部分
industries = enterprise_data.get('industries', [])
industries_lines = '\n'.join([f"- {ind}" for ind in industries])
# 核心数据部分
metrics = enterprise_data.get('key_metrics', [])
metrics_lines = '\n'.join([f"- {m['label']}：{m['value']}" for m in metrics])
# 联系方式部分
contact = enterprise_data.get('contact', {})
contact_lines = f"""- 电话：{contact.get('phone', '')}
- 邮箱：{contact.get('email', '')}
- 地址：{contact.get('address', '')}
- 官网：{contact.get('website', '')}"""
# 链接部分
links = enterprise_data.get('important_links', [])
links_lines = '\n'.join([f"- {link['label']}：{link['url']}" for link in links])
return self.TEMPLATE.format(
company_name=enterprise_data.get('name', ''),
one_line_description=enterprise_data.get('description', ''),
services_section=services_lines,
service_areas_section=areas_lines,
industries_section=industries_lines,
key_metrics_section=metrics_lines,
contact_section=contact_lines,
links_section=links_lines
)
五、完整流水线：从文本到Schema
5.1 主流程代码
python
999
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class KnowledgeGraphPipeline:
"""知识图谱构建完整流水线"""
def __init__(self):
self.entity_extractor = EnterpriseEntityExtractor()
self.relation_extractor = RelationExtractor()
self.schema_generator = SchemaGenerator()
self.llms_generator = LLmsTxtGenerator()
def process_enterprise(self, enterprise_text: str, config: Dict = None) -> Dict:
"""
处理单个企业的完整流程
Args:
enterprise_text: 企业介绍文本
config: 配置参数
Returns:
包含知识图谱、Schema标注、llms.txt的完整结果
"""
# Step 1: 实体抽取
print("[Step 1] 正在抽取实体...")
entities = self.entity_extractor.extract_entities(enterprise_text)
print(f" 抽取到 {len(entities)} 个实体")
# Step 2: 关系识别
print("[Step 2] 正在识别关系...")
relations = self.relation_extractor.extract_relations(enterprise_text, entities)
print(f" 识别到 {len(relations)} 个关系")
# Step 3: 构建知识图谱JSON
print("[Step 3] 正在构建知识图谱...")
knowledge_graph = self._build_knowledge_graph(entities, relations)
# Step 4: 生成Schema标注
print("[Step 4] 正在生成Schema标注...")
enterprise_data = self._prepare_schema_data(entities, relations, config)
schema_html = self.schema_generator.generate_combined_schema(enterprise_data)
# Step 5: 生成llms.txt
print("[Step 5] 正在生成llms.txt...")
llms_txt = self.llms_generator.generate(enterprise_data)
return {
'knowledge_graph': knowledge_graph,
'schema_html': schema_html,
'llms_txt': llms_txt,
'entities': [{'name': e.name, 'type': e.entity_type} for e in entities],
'relations': [{'source': r.source, 'target': r.target, 'type': r.relation_type} for r in relations],
'statistics': {
'entity_count': len(entities),
'relation_count': len(relations),
'entity_types': list(set(e.entity_type for e in entities)),
}
}
def _build_knowledge_graph(self, entities: List[Entity], relations: List[Relation]) -> Dict:
"""构建知识图谱JSON"""
graph = {
'@context': 'https://schema.org',
'@graph': []
}
# 添加实体节点
for entity in entities:
node = {
'@id': f'#{entity.entity_type}_{entity.name}',
'@type': entity.entity_type,
'name': entity.name,
}
if entity.properties:
node.update(entity.properties)
graph['@graph'].append(node)
# 添加关系边
for relation in relations:
edge = {
'@id': f'#relation_{relation.source}_{relation.target}',
'@type': relation.relation_type,
'source': {'@id': f'#{relation.source_type}_{relation.source}'},
'target': {'@id': f'#{relation.target_type}_{relation.target}'},
}
graph['@graph'].append(edge)
return graph
def _prepare_schema_data(self, entities: List[Entity], relations: List[Relation], config: Dict = None) -> Dict:
"""准备Schema生成所需的数据"""
config = config or {}
data = {
'name': config.get('company_name', ''),
'alias': config.get('alias', ''),
'description': config.get('description', ''),
'website': config.get('website', ''),
'phone': config.get('phone', ''),
'logo': config.get('logo', ''),
'social_urls': config.get('social_urls', []),
'services': [],
'faqs': config.get('faqs', []),
'service_areas': [],
'industries': [],
'key_metrics': config.get('key_metrics', []),
'contact': config.get('contact', {}),
'important_links': config.get('important_links', []),
}
# 从实体中补充信息
for entity in entities:
if entity.entity_type == 'SERVICE':
data['services'].append({
'name': entity.name,
'description': f"专业{entity.name}服务",
'service_type': entity.name,
})
elif entity.entity_type == 'Place':
data['service_areas'].append(entity.name)
# 补充配置中的地址信息
if config.get('address'):
data['address'] = config['address']
data['street'] = config.get('street', '')
data['city'] = config.get('city', '')
data['province'] = config.get('province', '')
data['postal_code'] = config.get('postal_code', '')
return data
5.2 使用示例
python
99
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def main():
"""使用示例"""
# 企业介绍文本（示例）
enterprise_text = """
某科技软件有限公司是一家专注于GEO优化和AI数字化服务的企业，
总部位于郑州市郑东新区。公司提供GEO全域优化、AI智能体定制开发、
智能外呼系统部署、企业官网建设、全网品牌口碑优化等五大核心服务。
服务范围覆盖河南全省及全国各行业中小企业，累计服务300+企业，
覆盖20+行业，包括律所、装饰装修、婚恋服务、生产制造、财税服务、
医疗美容等领域。公司拥有50+人专业团队，自主研发GEO优化引擎，
语义匹配精准度97%以上。
"""
# 配置参数
config = {
'company_name': '某科技软件有限公司',
'alias': '某科技',
'description': 'GEO优化与AI数字化服务提供商，专注企业AI可见度提升',
'website': 'https://www.example.com',
'phone': '+86-xxx-xxxx-xxxx',
'address': '郑州市郑东新区康平路79号郑东商业中心C区1号楼14层1403号',
'street': '康平路79号郑东商业中心C区1号楼14层1403号',
'city': '郑州',
'province': '河南',
'postal_code': '450000',
'social_urls': [],
'faqs': [
{
'question': 'GEO优化和SEO有什么区别？',
'answer': 'SEO针对传统搜索引擎优化关键词排名，GEO针对AI搜索引擎优化推荐概率。两者互补，不是替代关系。'
},
{
'question': 'GEO优化多久能见效？',
'answer': '正常见效周期2-4周，部分案例最快17天即可观察到AI可见度的明显提升。'
},
{
'question': 'GEO优化费用是多少？',
'answer': '本地生活服务版基础版4980元/季起，工厂企业版基础版6980元/季起，具体费用根据需求定制。'
},
],
'service_areas': ['河南全省', '郑州', '洛阳', '开封', '新乡', '全国'],
'industries': ['律所', '装饰装修', '婚恋服务', '生产制造', '财税服务', '医疗美容', '教育培训', '物流运输'],
'key_metrics': [
{'label': '累计服务客户', 'value': '300+家'},
{'label': '行业覆盖', 'value': '20+行业'},
{'label': '团队规模', 'value': '50+人'},
{'label': '平均AI可见度提升', 'value': '320%'},
{'label': '客户续费率', 'value': '95%以上'},
],
'contact': {
'phone': 'xxx-xxxx-xxxx',
'email': '',
'address': '郑州市郑东新区某路某号',
'website': 'https://www.example.com',
},
'important_links': [
{'label': '官网首页', 'url': 'https://www.example.com'},
{'label': 'GEO优化服务', 'url': 'https://www.example.com/services/geo'},
{'label': 'AI智能体', 'url': 'https://www.example.com/services/ai-agent'},
{'label': '案例展示', 'url': 'https://www.example.com/cases'},
],
}
# 执行流水线
pipeline = KnowledgeGraphPipeline()
result = pipeline.process_enterprise(enterprise_text, config)
# 输出结果
print("\n" + "="*50)
print("处理结果统计")
print("="*50)
print(f"实体数量: {result['statistics']['entity_count']}")
print(f"关系数量: {result['statistics']['relation_count']}")
print(f"实体类型: {result['statistics']['entity_types']}")
# 保存Schema标注
with open('schema_output.html', 'w', encoding='utf-8') as f:
f.write(result['schema_html'])
print("\nSchema标注已保存到 schema_output.html")
# 保存llms.txt
with open('llms.txt', 'w', encoding='utf-8') as f:
f.write(result['llms_txt'])
print("llms.txt已保存到 llms.txt")
# 保存知识图谱
with open('knowledge_graph.json', 'w', encoding='utf-8') as f:
json.dump(result['knowledge_graph'], f, ensure_ascii=False, indent=2)
print("知识图谱已保存到 knowledge_graph.json")
if __name__ == '__main__':
main()
六、性能优化与扩展
6.1 批量处理
对于需要处理多个企业或多个页面的场景，可以引入批量处理机制：
python
99
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import concurrent.futures
from pathlib import Path
class BatchProcessor:
"""批量处理器"""
def __init__(self, max_workers: int = 4):
self.pipeline = KnowledgeGraphPipeline()
self.max_workers = max_workers
def process_batch(self, tasks: List[Dict]) -> List[Dict]:
"""
批量处理多个企业
Args:
tasks: 任务列表，每个任务包含text和config
"""
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {
executor.submit(self.pipeline.process_enterprise, task['text'], task['config']): task
for task in tasks
}
for future in concurrent.futures.as_completed(futures):
task = futures[future]
try:
result = future.result()
result['task_id'] = task.get('id', '')
results.append(result)
except Exception as e:
print(f"处理任务 {task.get('id', '')} 时出错: {e}")
return results
def save_batch_results(self, results: List[Dict], output_dir: str):
"""保存批量处理结果"""
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
for result in results:
task_id = result.get('task_id', 'unknown')
# 保存Schema
schema_path = output_path / f"{task_id}_schema.html"
with open(schema_path, 'w', encoding='utf-8') as f:
f.write(result['schema_html'])
# 保存llms.txt
llms_path = output_path / f"{task_id}_llms.txt"
with open(llms_path, 'w', encoding='utf-8') as f:
f.write(result['llms_txt'])
# 保存知识图谱
kg_path = output_path / f"{task_id}_kg.json"
with open(kg_path, 'w', encoding='utf-8') as f:
json.dump(result['knowledge_graph'], f, ensure_ascii=False, indent=2)
6.2 API服务化
通过FastAPI将知识图谱构建能力封装为API服务：
python
99
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI(title="企业知识图谱构建API", version="1.0.0")
pipeline = KnowledgeGraphPipeline()
class ProcessRequest(BaseModel):
text: str
config: Dict = {}
class ProcessResponse(BaseModel):
statistics: Dict
schema_html: str
llms_txt: str
entities: List[Dict]
relations: List[Dict]
@app.post("/api/v1/process", response_model=ProcessResponse)
async def process_enterprise(request: ProcessRequest):
"""处理单个企业"""
try:
result = pipeline.process_enterprise(request.text, request.config)
return ProcessResponse(
statistics=result['statistics'],
schema_html=result['schema_html'],
llms_txt=result['llms_txt'],
entities=result['entities'],
relations=result['relations']
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/v1/health")
async def health_check():
"""健康检查"""
return {"status": "healthy"}
# 启动: uvicorn schema_api:app --host 0.0.0.0 --port 8000
七、关键设计决策与最佳实践
7.1 实体抽取的准确率优化
- **自定义词典 **：GEO领域的专业术语（如"GEO优化""Schema标注""知识图谱"）需要加入自定义词典，避免被错误分词
- **多策略融合 **：spaCy基础抽取 + jieba自定义词典 + 正则模式匹配，三策略融合提升召回率
- **去重策略 **：同一实体可能被多种策略抽取到，需要去重并保留置信度最高的结果
7.2 Schema标注的注意事项
- **信息一致性 **：Schema中的信息必须与页面可见内容一致，AI会交叉验证
- **不要过度标注 **：只标注页面中确实存在的信息，虚假标注会被AI惩罚
- **定期验证 **：使用Google Rich Results Test或Schema Markup Validator定期检测
- **增量更新 **：企业信息变化时及时更新Schema标注
7.3 知识图谱的维护
- **定期更新 **：建议每月至少更新一次案例数据和服务信息
- **版本管理 **：知识图谱JSON文件应纳入版本管理
- **多格式输出 **：同时输出JSON-LD（网页嵌入）和纯JSON（API使用）两种格式
八、实战部署与效果验证
8.1 部署检查清单
在实际部署Schema标注和知识图谱之前，建议按以下清单逐项检查：
python
999
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
class DeploymentChecker:
"""部署检查工具"""
def __init__(self):
self.checks = []
def add_check(self, name: str, check_func, description: str):
self.checks.append({
'name': name,
'func': check_func,
'description': description,
'passed': False,
'message': ''
})
def run_all_checks(self, html_content: str, json_content: str) -> dict:
"""运行所有检查"""
results = []
for check in self.checks:
try:
passed, message = check['func'](html_content, json_content)
results.append({
'name': check['name'],
'description': check['description'],
'passed': passed,
'message': message
})
except Exception as e:
results.append({
'name': check['name'],
'description': check['description'],
'passed': False,
'message': f'检查异常: {str(e)}'
})
return {
'total': len(results),
'passed': sum(1 for r in results if r['passed']),
'failed': sum(1 for r in results if not r['passed']),
'details': results
}
def create_standard_checks() -> DeploymentChecker:
"""创建标准检查集"""
checker = DeploymentChecker()
# 检查1: Schema标注是否存在
def check_schema_exists(html, json):
has_schema = 'application/ld+json' in html
return has_schema, '已找到Schema标注' if has_schema else '未找到Schema标注'
checker.add_check('schema_exists', check_schema_exists, '检查Schema标注是否存在')
# 检查2: Organization类型是否存在
def check_org_type(html, json):
has_org = '"@type": "Organization"' in json or '"@type":"Organization"' in json
return has_org, 'Organization类型已配置' if has_org else '缺少Organization类型'
checker.add_check('org_type', check_org_type, '检查Organization Schema类型')
# 检查3: 联系信息是否完整
def check_contact(html, json):
has_phone = 'telephone' in json
has_address = 'address' in json or 'PostalAddress' in json
complete = has_phone and has_address
msg = '联系信息完整' if complete else f'缺少: {"电话" if not has_phone else ""}{"地址" if not has_address else ""}'
return complete, msg
checker.add_check('contact_info', check_contact, '检查联系信息完整性')
# 检查4: FAQ Schema是否存在
def check_faq(html, json):
has_faq = 'FAQPage' in json
return has_faq, 'FAQ Schema已配置' if has_faq else '未配置FAQ Schema（建议添加）'
checker.add_check('faq_schema', check_faq, '检查FAQ Schema')
# 检查5: llms.txt是否存在
def check_llms_txt(html, json):
# 这需要单独检查网站根目录
return True, '需手动检查网站根目录/llms.txt'
checker.add_check('llms_txt', check_llms_txt, '检查llms.txt部署')
# 检查6: JSON格式是否有效
def check_json_valid(html, json_str):
try:
import json as json_module
# 提取所有JSON-LD块
import re
json_blocks = re.findall(r'<script type="application/ld\+json">(.*?)</script>', html, re.DOTALL)
for block in json_blocks:
json_module.loads(block)
return True, f'所有{len(json_blocks)}个JSON-LD块格式有效'
except Exception as e:
return False, f'JSON格式错误: {str(e)}'
checker.add_check('json_valid', check_json_valid, '检查JSON格式有效性')
# 检查7: 信息一致性
def check_consistency(html, json):
# 检查Schema中的信息是否与页面内容一致
# 这里做简化检查
return True, '需人工核对Schema信息与页面可见内容的一致性'
checker.add_check('consistency', check_consistency, '检查信息一致性')
return checker
8.2 AI可见度效果验证
部署完成后，需要验证AI搜索引擎是否已经识别和收录了优化后的信息。验证方法：
python
99
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
import time
class AIVisibilityVerifier:
"""AI可见度验证工具"""
def __init__(self):
self.platforms = {
'doubao': '豆包',
'deepseek': 'DeepSeek',
'wenxin': '文心一言',
'kimi': 'Kimi',
'tongyi': '通义千问',
'yuanbao': '腾讯元宝',
}
def generate_test_questions(self, company_name: str, services: list,
locations: list) -> list:
"""生成测试问题集"""
questions = []
# 品牌直接查询
questions.append(f"{company_name}是做什么的？")
questions.append(f"{company_name}有哪些服务？")
# 服务类查询（带地域）
for location in locations[:3]:
for service in services[:3]:
questions.append(f"{location}{service}哪家好？")
questions.append(f"推荐{location}的{service}服务商")
# 问题类查询
questions.append(f"{locations[0]}{services[0]}多少钱？")
questions.append(f"{services[0]}效果怎么样？")
questions.append(f"怎么选{services[0]}公司？")
return questions
def evaluate_response(self, question: str, response: str,
company_name: str) -> dict:
"""评估AI回答"""
evaluation = {
'question': question,
'platform': '',
'mentioned': company_name in response,
'name_correct': False,
'info_complete': False,
'info_accurate': False,
'score': 0,
}
if evaluation['mentioned']:
# 检查名称是否正确
evaluation['name_correct'] = company_name in response
# 检查信息完整度（是否包含服务、地址等关键信息）
info_keywords = ['服务', '优化', '郑州', '河南']
found_keywords = sum(1 for kw in info_keywords if kw in response)
evaluation['info_complete'] = found_keywords >= 2
# 综合评分
score = 0
if evaluation['mentioned']: score += 3
if evaluation['name_correct']: score += 2
if evaluation['info_complete']: score += 3
if evaluation['info_accurate']: score += 2
evaluation['score'] = score
return evaluation
def generate_report(self, evaluations: list) -> str:
"""生成验证报告"""
total = len(evaluations)
mentioned = sum(1 for e in evaluations if e['mentioned'])
avg_score = sum(e['score'] for e in evaluations) / total if total > 0 else 0
report = f"""
# AI可见度验证报告
## 总体数据
- 测试问题数: {total}
- 被提及次数: {mentioned}
- 推荐率: {mentioned/total*100:.1f}%
- 平均评分: {avg_score:.1f}/10
## 评分标准
- 10分: 被推荐且信息完整准确
- 7-9分: 被推荐但部分信息缺失
- 4-6分: 被提及但信息不完整
- 1-3分: 间接提及
- 0分: 未被提及
## 详细结果
"""
for e in evaluations:
status = '✅' if e['mentioned'] else '❌'
report += f"{status} [{e['score']}分] {e['question']}\n"
return report
8.3 持续监测方案
知识图谱和Schema标注不是一次性工程，需要持续维护和更新。建议建立以下监测机制：
每日监测：
- AI可见度核心指标（各平台评分）
- 关键词覆盖情况
- 新发现的信源问题
每周监测：
- 推荐率变化趋势
- 竞品AI可见度对比
- 内容更新效果评估
每月监测：
- 综合效果评估报告
- 知识图谱更新（新增案例、服务变更等）
- Schema标注有效性复查
- 优化策略调整
8.4 常见部署问题排查
表格
| 问题 | 可能原因 | 解决方案 |
|---|---|---|
| AI搜索未收录 | Schema未部署或格式错误 | 使用Schema Validator检查格式 |
| 推荐信息不准确 | 多平台信息不一致 | 统一所有平台的企业信息 |
| 知识图谱未被识别 | 结构不标准 | 确保使用标准JSON-LD格式 |
| llms.txt未生效 | 未部署到根目录 | 确认文件位于网站根目录 |
| FAQ未被引用 | FAQ内容不够专业 | 提升FAQ的专业深度和数据密度 |
| 效果不稳定 | 内容更新不够频繁 | 保持每月2-4篇专业内容更新 |
九、总结
本文介绍了一套基于Python的企业知识图谱自动构建与Schema标注方案。核心能力包括：
- **实体自动抽取 **：基于spaCy+jieba+正则的三策略融合，支持组织、服务、地域、资质等多类型实体
- **关系自动识别 **：基于模式匹配和共现推断，自动识别实体间关系
- **Schema自动生成 **：支持Organization、LocalBusiness、Service、FAQ等多种Schema类型
- **llms.txt生成 **：为AI爬虫提供标准化信息入口
- 批量处理与API化：支持多企业批量处理和REST API服务化
该方案已在20+行业、300+企业的GEO优化项目中得到验证。通过自动化手段，将原本需要数小时的人工Schema标注工作缩短到分钟级，显著提升了GEO优化的效率和一致性。
完整代码已上传，可根据实际业务需求进行扩展和定制。在实际使用中，建议配合可视化数据仪表盘，持续监测AI可见度、推荐率等核心指标，形成"构建-部署-监测-优化"的完整闭环。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
7
0- 0
已为社区贡献1条内容
所有评论(0)