AI知识图谱构建实战：从零搭建企业级智能问答系统

1次阅读

没有评论

共计 3607 个字符，预计需要花费 10 分钟才能阅读完成。

在企业数据智能化的过程中，我们经常遇到这样的困扰：

数据散落在各个系统（CRM、ERP、文档系统等），格式千差万别
传统数据库只能回答「是什么」，无法理解「为什么」和「怎么样」
简单的关键词匹配问答经常给出答非所问的结果

这些问题正是知识图谱要解决的核心痛点。通过将数据转化为「实体 - 关系 - 实体」的三元组形式，我们能让机器真正理解业务知识。

在知识图谱存储方案中，主要分为两类：

RDF 三元组存储（如 Jena）
优点：W3C 标准，适合学术研究
缺点：查询性能较差，缺乏属性支持
属性图模型（如 Neo4j）
优点：
- 直观的图结构可视化
- 支持节点和关系的属性存储
- Cypher 查询语言易学易用
- 千万级节点仍保持良好性能

对于企业级应用，Neo4j 的工程友好性明显更胜一筹。下面这段 Cypher 语句可以快速创建一个人物关系图谱：

CREATE (张伟: 人物 {name:'张伟', age:35, title:'CTO'})
CREATE (字节跳动: 公司 {name:'字节跳动', industry:'互联网'})
CREATE (张伟)-[r: 任职 {since:2018}]->(字节跳动)

使用 BERT-BiLSTM-CRF 组合模型，既利用 BERT 的语义理解能力，又通过 CRF 保证标签合理性。关键代码段：

class NERModel(nn.Module):
    def __init__(self, bert_model, hidden_dim, tagset_size):
        super().__init__()
        self.bert = bert_model
        self.lstm = nn.LSTM(768, hidden_dim // 2, 
                           bidirectional=True)
        self.hidden2tag = nn.Linear(hidden_dim, tagset_size)
        self.crf = CRF(tagset_size)

    def forward(self, x):
        # BERT 输出 [batch_size, seq_len, 768]
        embeds = self.bert(x)[0]  
        # BiLSTM 输出 [seq_len, batch_size, hidden_dim]
        lstm_out, _ = self.lstm(embeds.permute(1,0,2))
        # 全连接层输出 [batch_size, seq_len, tagset_size]
        emissions = self.hidden2tag(lstm_out.permute(1,0,2))
        return emissions

在传统模型基础上加入注意力机制，让模型聚焦关键文本片段。模型结构示意图：

[文本输入] → [BERT 编码] → [注意力权重计算] → [关系分类]

关键改进点：

class AttentionLayer(nn.Module):
    def __init__(self, hidden_size):
        super().__init__()
        self.query = nn.Linear(hidden_size, hidden_size)
        self.key = nn.Linear(hidden_size, hidden_size)

    def forward(self, hidden_states):
        # 计算注意力分数 [batch_size, seq_len, seq_len]
        scores = torch.matmul(self.query(hidden_states), 
            self.key(hidden_states).transpose(1,2))
        # 归一化注意力权重
        attn_weights = F.softmax(scores, dim=-1)
        # 加权求和 [batch_size, seq_len, hidden_size]
        context = torch.matmul(attn_weights, hidden_states)
        return context

遵循这些最佳实践避免后期重构：

节点类型不超过 20 种，用标签分类（如: 人物: 高管）
关系类型保持原子性（避免 「参与 - 管理」 这种复合关系）

为高频查询路径创建索引：

CREATE INDEX FOR (p: 产品) ON (p. 名称)

完整的处理流程应该是：

原始数据 → 2. 实体识别 → 3. 关系抽取 → 4. 图谱存储 → 5. 质量验证

用 Airflow 可以方便地编排这个 DAG：

with DAG('kg_pipeline', schedule_interval='@daily') as dag:
    extract = PythonOperator(task_id='data_extract', python_callable=extract_data)
    transform = PythonOperator(task_id='entity_recognition', python_callable=run_ner)
    load = Neo4jOperator(task_id='import_to_neo4j', neo4j_conn_id='neo4j_default')

    extract >> transform >> load

当处理百万级数据时：

按数据来源分片处理
使用 Dask 或 Ray 进行分布式计算
Neo4j 批量导入建议使用apoc.load.json

CALL apoc.periodic.iterate(
  'UNWIND $data AS item RETURN item',
  'MERGE (p: 人物 {name:item.name})
   SET p += item.properties',
  {batchSize:1000, parallel:true, params:{data:$data}})

避免全量更新的开销：

为每个节点添加 更新时间戳
使用子图隔离变更集
通过事务保证一致性

def delta_update(update_records):
    with driver.session() as session:
        tx = session.begin_transaction()
        try:
            for record in update_records:
                # 使用 MERGE 实现存在则更新，不存在则创建
                tx.run("""
                MERGE (n: 实体 {id: $id})
                SET n += $properties, 
                    n.last_updated = timestamp()
                """, parameters)
            tx.commit()
        except Exception as e:
            tx.rollback()

遇到「苹果」可能是水果也可能是公司时：

结合上下文特征（相邻实体）
维护同义词词典

添加消歧属性：

CREATE (: 产品 {name:'苹果', disambig:'电子产品'})
CREATE (: 水果 {name:'苹果', disambig:'农产品'})

当图谱出现 A→B→C→A 的环路时，查询可能陷入死循环。解决方法：

设置遍历深度限制

MATCH path=(a)-[*1..3]->(b)
WHERE a.name='张三' AND NOT cyclic(path)
RETURN path

使用 APOC 路径扩展过程

CALL apoc.path.expandConfig($startNode, {
    relationshipFilter: '合作 | 投资',
    maxLevel: 5
})

知识图谱 + 大模型的融合方案：

检索增强生成（RAG）：
先用图谱查找相关实体
将结构化结果作为 prompt 上下文

def query_kg(question):
    # 实体链接
    entities = linker.extract(question)  
    # 图谱查询
    with driver.session() as session:
        result = session.run("""
        MATCH path=(e)-[r]->(t)
        WHERE e.name IN $entities
        RETURN path LIMIT 5
        """, {'entities': entities})
    return [dict(record) for record in result]

def generate_answer(question):
    context = query_kg(question)
    prompt = f"""基于以下知识回答问题：{context}\n 问题：{question}"""
    return llm.generate(prompt)