把构建换到应用时踩过的坑

知识图谱 Neo4j NLP 实体识别 图数据库
作者: 醉月思 发布日期: 2024-08-14 分类: 技术实践

这次做知识图谱项目，从 NER 到图数据库，中间折腾了三个月，踩的坑比想的多。

先试了开源的 NER 模型，发现中文场景下效果一般。

为什么做知识图谱

产品要做智能问答，传统的关键词匹配搞不定复杂推理，老板说上知识图谱。

需求倒不复杂：用户问「张三和李四有什么关系」，系统能从结构化数据里找出两个人之间的路径，还能给出中间经过的实体。

听起来简单，实际上从数据到可用图谱，中间的坑比想象多。

知识抽取

实体识别

先试了开源的 NER 模型，发现中文场景下效果一般。人名识别还行，但公司名、产品名经常分错。

# 使用 HanLP 进行中文实体识别
from pyhanlp import *

text = "张三在阿里巴巴集团工作，负责天猫商城的推荐算法。"

# 加载 NER 模型
ner_model = HanLP.load(HanLP.PretrainedNER.MSR)

result = ner_model(text)
print(result)

# 输出格式：[张三/nr, 在/p, 阿里巴巴集团/nt, 工作/v, ，/w, 负责/v, 天猫商城/nt, 的/u, 推荐算法/n, 。/w]

坑一：领域实体识别不准。业务里有很多专业术语，通用模型完全不认识。

解决：自己训练了一个 BERT-NER 模型，加了领域词典。

# 自定义实体识别模型
from transformers import BertTokenizer, BertForTokenClassification
import torch

tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')
model = BertForTokenClassification.from_pretrained('bert-base-chinese', num_labels=10)

# 添加自定义标签
label_list = ['O', 'B-PER', 'I-PER', 'B-ORG', 'I-ORG', 'B-PROD', 'I-PROD', 'B-LOC', 'I-LOC', 'B-TIME']

# 训练代码省略...
# 实际训练用了 5000 条标注数据，效果提升明显

关系抽取

实体找到了，还要知道实体之间什么关系。试过几种方案：

方案一：基于规则的关系抽取

import re

text = "张三在阿里巴巴集团担任高级工程师"

# 定义关系规则
patterns = {
    'WORKS_AT': r'(.+?)在(.+?)担任(.+)',
    'LOCATED_IN': r'(.+?)位于(.+)',
    'FOUNDED_BY': r'(.+?)由(.+?)创立'
}

for relation, pattern in patterns.items():
    match = re.search(pattern, text)
    if match:
        entity1 = match.group(1)
        entity2 = match.group(2)
        print(f"Relation: {relation}, Entity1: {entity1}, Entity2: {entity2}")

坑二：规则写不完，中文表达太灵活。

方案二：基于深度学习的关系抽取

# 使用 BERT 进行关系分类
from transformers import BertTokenizer, BertForSequenceClassification

tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')
model = BertForSequenceClassification.from_pretrained('bert-base-chinese', num_labels=6)

# 关系类别
relation_types = ['WORKS_AT', 'LOCATED_IN', 'FOUNDED_BY', 'PART_OF', 'RELATED_TO', 'NO_RELATION']

# 输入格式：[CLS] 实体1 [SEP] 实体2 [SEP] 上下文 [SEP]
def prepare_input(entity1, entity2, context):
    input_text = f"[CLS] {entity1} [SEP] {entity2} [SEP] {context} [SEP]"
    return tokenizer(input_text, return_tensors='pt')

# 推理
def predict_relation(entity1, entity2, context):
    inputs = prepare_input(entity1, entity2, context)
    outputs = model(**inputs)
    predicted_class = torch.argmax(outputs.logits, dim=1).item()
    return relation_types[predicted_class]

# 示例
relation = predict_relation("张三", "阿里巴巴集团", "张三在阿里巴巴集团担任高级工程师")
print(f"Predicted relation: {relation}")

坑三：训练数据不够，关系类别定义不清晰。

解决：简化关系类型，优先处理高频关系。

图数据库选择

Neo4j 方案

图数据库选了 Neo4j，主要考虑：

  1. 生态成熟，文档完善
  2. Cypher 查询语言相对直观
  3. 社区版功能够用
# 安装 Neo4j
docker run -d \
    --name neo4j \
    -p7474:7474 -p7687:7687 \
    -e NEO4J_AUTH=neo4j/password \
    neo4j:5.12.0

# 验证安装
docker exec -it neo4j cypher-shell -u neo4j -p password

数据导入

从 CSV 批量导入数据：

from neo4j import GraphDatabase

class KnowledgeGraphImporter:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

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

    def import_entities(self, csv_file, entity_type):
        with self.driver.session() as session:
            session.run(f"""
            LOAD CSV WITH HEADERS FROM 'file:///{csv_file}' AS row
            MERGE (e:{entity_type} {{name: row.name}})
            SET e.id = row.id, e.properties = row.properties
            """)

    def import_relations(self, csv_file, relation_type):
        with self.driver.session() as session:
            session.run(f"""
            LOAD CSV WITH HEADERS FROM 'file:///{csv_file}' AS row
            MATCH (source {{name: row.source}})
            MATCH (target {{name: row.target}})
            MERGE (source)-[r:{relation_type}]->(target)
            SET r.weight = toFloat(row.weight), r.confidence = toFloat(row.confidence)
            """)

# 使用示例
importer = KnowledgeGraphImporter("bolt://localhost:7687", "neo4j", "password")
importer.import_entities("companies.csv", "Company")
importer.import_entities("persons.csv", "Person")
importer.import_relations("works_at.csv", "WORKS_AT")

坑四：CSV 导入速度慢，大文件容易超时。

解决：分批导入，使用 UNWIND 批处理。

def batch_import(self, entities, entity_type, batch_size=1000):
    with self.driver.session() as session:
        for i in range(0, len(entities), batch_size):
            batch = entities[i:i + batch_size]
            session.run(f"""
            UNWIND $batch AS row
            MERGE (e:{entity_type} {{name: row.name}})
            SET e.id = row.id, e.properties = row.properties
            """, batch=batch)

图谱查询

基础查询

查询两个实体之间的关系：

// 查询张三和李四之间的路径
MATCH path = shortestPath(
  (p1:Person {name: "张三"})-[*]-(p2:Person {name: "李四"})
)
RETURN path, length(path) as distance;

// 查询张三的所有同事
MATCH (p:Person {name: "张三"})-[:WORKS_AT]->(c:Company)<-[:WORKS_AT]-(colleague:Person)
RETURN colleague.name, c.name as company;

// 查询阿里巴巴集团的所有产品
MATCH (c:Company {name: "阿里巴巴集团"})-[:OWNS]->(p:Product)
RETURN p.name, p.category;

推理查询

// 查询张三的二度人脉
MATCH (p:Person {name: "张三"})-[:WORKS_AT]->(c:Company)<-[:WORKS_AT]-(colleague:Person)
MATCH (colleague)-[:WORKS_AT]->(other:Company)<-[:WORKS_AT]-(second_degree:Person)
WHERE second_degree.name <> "张三" AND second_degree.name <> colleague.name
RETURN DISTINCT second_degree.name, other.name as company;

// 查询可能存在的合作关系
MATCH (c1:Company {name: "阿里巴巴集团"})-[:INVESTED_IN]->(startup:Company)<-[:FOUNDED_BY]-(p:Person)
MATCH (p)-[:WORKS_AT]->(c2:Company)
WHERE c2.name <> "阿里巴巴集团"
RETURN DISTINCT c2.name as potential_partner, startup.name as investment;

坑五：复杂查询性能差，特别是涉及长路径。

解决：添加索引，限制查询深度。

// 创建索引
CREATE INDEX person_name_idx FOR (p:Person) ON (p.name);
CREATE INDEX company_name_idx FOR (c:Company) ON (c.name);

// 限制查询深度
MATCH path = (p:Person {name: "张三"})-[*1..3]-(other:Person)
WHERE other.name <> "张三"
RETURN other.name, length(path) as distance
ORDER BY distance
LIMIT 20;

应用场景

智能问答

class KnowledgeGraphQA:
    def __init__(self, neo4j_uri, user, password):
        self.driver = GraphDatabase.driver(neo4j_uri, auth=(user, password))
        self.templates = {
            "relationship": "MATCH (p1:Person {{name: $entity1}})-[r]-(p2:Person {{name: $entity2}}) RETURN type(r) as relation",
            "colleagues": "MATCH (p:Person {{name: $name}})-[:WORKS_AT]->(c:Company)<-[:WORKS_AT]-(colleague:Person) RETURN colleague.name",
            "products": "MATCH (c:Company {{name: $name}})-[:OWNS]->(p:Product) RETURN p.name"
        }

    def answer(self, question):
        # 解析问题（简化版，实际需要 NLP）
        if "有什么关系" in question:
            entity1, entity2 = self._extract_entities(question)
            return self._query_relation(entity1, entity2)
        elif "同事" in question:
            name = self._extract_name(question)
            return self._query_colleagues(name)
        else:
            return "抱歉，我不理解这个问题。"

    def _query_relation(self, entity1, entity2):
        with self.driver.session() as session:
            result = session.run(
                self.templates["relationship"],
                entity1=entity1, entity2=entity2
            )
            relations = [record["relation"] for record in result]
            if relations:
                return f"{entity1} 和 {entity2} 之间有 {', '.join(relations)} 关系"
            else:
                return f"没有找到 {entity1} 和 {entity2} 之间的关系"

    def _query_colleagues(self, name):
        with self.driver.session() as session:
            result = session.run(
                self.templates["colleagues"],
                name=name
            )
            colleagues = [record["colleague.name"] for record in result]
            if colleagues:
                return f"{name} 的同事有：{', '.join(colleagues)}"
            else:
                return f"没有找到 {name} 的同事信息"

# 使用示例
qa = KnowledgeGraphQA("bolt://localhost:7687", "neo4j", "password")

# 用户提问
print(qa.answer("张三和李四有什么关系？"))
print(qa.answer("张三的同事有哪些？"))

推荐系统

# 基于知识图谱的推荐
class KnowledgeGraphRecommender:
    def __init__(self, neo4j_uri, user, password):
        self.driver = GraphDatabase.driver(neo4j_uri, auth=(user, password))

    def recommend_products(self, user_name, top_n=5):
        with self.driver.session() as session:
            # 查询用户购买过的产品
            result = session.run("""
            MATCH (u:User {name: $name})-[:PURCHASED]->(p:Product)
            RETURN p.name, p.category
            """, name=user_name)

            purchased = list(result)

            # 基于相似用户推荐
            recommendations = session.run("""
            MATCH (u1:User {name: $name})-[:PURCHASED]->(p:Product)<-[:PURCHASED]-(u2:User)
            MATCH (u2)-[:PURCHASED]->(p2:Product)
            WHERE NOT (u1)-[:PURCHASED]->(p2)
            RETURN p2.name, count(*) as score
            ORDER BY score DESC
            LIMIT $limit
            """, name=user_name, limit=top_n)

            return [record["p2.name"] for record in recommendations]

# 使用示例
recommender = KnowledgeGraphRecommender("bolt://localhost:7687", "neo4j", "password")
products = recommender.recommend_products("张三")
print(f"推荐给张三的产品：{products}")

踩过的坑总结

坑六：实体消歧

同一个实体名称对应多个真实实体，比如「苹果」可能是公司、水果、品牌。

解决：添加上下文信息，用实体类型限定。

def disambiguate_entity(entity_name, context):
    # 分析上下文确定实体类型
    if "公司" in context or "企业" in context:
        entity_type = "Company"
    elif "水果" in context or "食物" in context:
        entity_type = "Food"
    else:
        entity_type = "Unknown"

    # 查询图谱
    with driver.session() as session:
        result = session.run("""
        MATCH (e {name: $name})
        WHERE labels(e) = [$type]
        RETURN e
        """, name=entity_name, type=entity_type)
        return result.single()

坑七：知识更新困难

业务数据变化快，图谱更新不及时会出错。

解决：建立更新机制，定期同步。

class KnowledgeGraphUpdater:
    def __init__(self, neo4j_uri, user, password):
        self.driver = GraphDatabase.driver(neo4j_uri, auth=(user, password))

    def update_entity(self, entity_type, entity_name, properties):
        with self.driver.session() as session:
            session.run(f"""
            MERGE (e:{entity_type} {{name: $name}})
            SET e += $properties
            SET e.updated_at = datetime()
            """, name=entity_name, properties=properties)

    def delete_entity(self, entity_type, entity_name):
        with self.driver.session() as session:
            session.run(f"""
            MATCH (e:{entity_type} {{name: $name}})
            DETACH DELETE e
            """, name=entity_name)

    def batch_update(self, updates):
        with self.driver.session() as session:
            for update in updates:
                if update["action"] == "update":
                    self.update_entity(update["type"], update["name"], update["properties"])
                elif update["action"] == "delete":
                    self.delete_entity(update["type"], update["name"])

坑八：图谱规模大时查询慢

节点超过百万后，查询性能明显下降。

解决：

  1. 合理使用索引
  2. 查询限制深度
  3. 使用缓存
import hashlib
import json

class CachedQuery:
    def __init__(self, neo4j_driver, cache_ttl=3600):
        self.driver = neo4j_driver
        self.cache = {}
        self.cache_ttl = cache_ttl

    def query(self, cypher, params=None):
        cache_key = self._get_cache_key(cypher, params)

        if cache_key in self.cache:
            cached = self.cache[cache_key]
            if time.time() - cached["timestamp"] < self.cache_ttl:
                return cached["result"]

        with self.driver.session() as session:
            result = list(session.run(cypher, params or {}))
            self.cache[cache_key] = {
                "result": result,
                "timestamp": time.time()
            }
            return result

    def _get_cache_key(self, cypher, params):
        key_string = f"{cypher}:{json.dumps(params, sort_keys=True)}"
        return hashlib.md5(key_string.encode()).hexdigest()

性能优化

内存配置

Neo4j 默认内存配置不够，需要调整：

# neo4j.conf
dbms.memory.heap.initial_size=512m
dbms.memory.heap.max_size=2G
dbms.memory.pagecache.size=2G

# 重启
docker restart neo4j

查询优化

// 使用 PROFILE 分析查询
PROFILE MATCH path = (p1:Person {name: "张三"})-[*]-(p2:Person {name: "李四"})
RETURN path;

// 优化后的查询
MATCH (p1:Person {name: "张三"})
MATCH (p2:Person {name: "李四"})
MATCH path = shortestPath((p1)-[*]-(p2))
RETURN path;

数据质量

数据清洗

def clean_entity_name(name):
    """清洗实体名称"""
    # 去除前后空格
    name = name.strip()

    # 统一全角半角
    name = name.replace('（', '(').replace('）', ')')
    name = name.replace('：', ':').replace('，', ',')

    # 去除特殊字符
    name = re.sub(r'[^\w\s一-鿿]', '', name)

    return name

def normalize_relation(relation):
    """标准化关系名称"""
    relation_mapping = {
        '工作在': 'WORKS_AT',
        '任职于': 'WORKS_AT',
        '位于': 'LOCATED_IN',
        '坐落在': 'LOCATED_IN',
        '创立了': 'FOUNDED_BY'
    }
    return relation_mapping.get(relation, relation)

数据验证

def validate_graph_data(driver):
    """验证图谱数据质量"""
    with driver.session() as session:
        # 检查孤立节点
        result = session.run("""
        MATCH (n)
        WHERE NOT (n)-[]-()
        RETURN labels(n) as labels, count(*) as count
        """)
        isolated = [record for record in result]
        if isolated:
            print(f"发现孤立节点：{isolated}")

        # 检查重复关系
        result = session.run("""
        MATCH ()-[r]->()
        WITH type(r) as rel_type, startNode(r) as start, endNode(r) as end
        WITH rel_type, start, end, count(*) as count
        WHERE count > 1
        RETURN rel_type, start.name, end.name, count
        """)
        duplicates = [record for record in result]
        if duplicates:
            print(f"发现重复关系：{duplicates}")

        # 检查缺失属性
        result = session.run("""
        MATCH (n:Person)
        WHERE n.name IS NULL OR n.name = ""
        RETURN count(*) as count
        """)
        missing = [record for record in result]
        if missing[0]["count"] > 0:
            print(f"发现缺失名称的Person节点：{missing[0]['count']}个")

写在最后

知识图谱这东西，看着概念多，拆开做其实就几件事：实体抽取、关系抽取、图存储、查询优化。

解决了：

  • 结构化知识表示
  • 复杂关系推理
  • 智能问答基础

带来了：

  • 数据处理复杂度
  • 维护成本不低
  • 效果依赖数据质量

要不要上知识图谱，先想清楚：

  1. 业务真的需要复杂推理吗？
  2. 有足够的数据质量支撑吗？
  3. 有持续维护的能力吗？

不是所有场景都适合，但一旦用对了，价值确实明显。

这次知识图谱项目花了三个月，从 NLP 到图数据库踩了不少坑，最后系统上线后问答准确率达到 78%，比关键词匹配提升了 40 个点。

版权声明: 本文首发于 指尖魔法屋-把构建换到应用时踩过的坑（https://blog.thinkmoon.cn/post/163-knowledge-graph-construction-application-practice/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章