从零构建基于Agent的知识库系统：知识图谱实战入门指南

1次阅读

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

在传统数据库（如 MySQL）中存储知识时，我们常遇到两个核心问题：

语义关联缺失 ：比如 ” 爱因斯坦 - 发明 - 相对论 ” 这条知识，在关系型数据库中需要拆分成多表关联查询，无法直接表达实体间关系
多跳推理困难 ：当需要回答 ” 爱因斯坦的老师是谁的学生 ” 这类问题时，需要编写复杂的 JOIN 查询，且性能随跳数增加急剧下降

关系型数据库 ：
优点：事务支持完善，适合结构化数据
局限：处理关联查询时需要预设 schema，灵活性差
图数据库 (Neo4j)：
优点：天然适合存储实体关系，支持路径查询
示例：MATCH (a:Person)-[:TEACHED_BY]->(b)-[:STUDENT_OF]->(c) RETURN c
向量数据库 ：
优点：适合语义相似度搜索
局限：无法处理明确的逻辑关系

使用 spaCy 进行实体和关系抽取的典型代码：

from typing import List
from pydantic import BaseModel
import spacy

class Entity(BaseModel):
    text: str
    label: str

class Relation(BaseModel):
    head: Entity
    tail: Entity
    rel_type: str

def extract_knowledge(text: str) -> List[Relation]:
    """时间复杂度 O(n^2)，主要消耗在依存句法分析"""
    nlp = spacy.load("en_core_web_sm")
    doc = nlp(text)

    relations = []
    for token in doc:  # O(n)
        if token.dep_ in ("nsubj", "dobj"):
            relations.append(Relation(head=Entity(text=token.head.text, label=token.head.ent_type_),
                tail=Entity(text=token.text, label=token.ent_type_),
                rel_type=token.dep_
            ))
    return relations

推荐的数据模型和批量插入优化方案：

from neo4j import GraphDatabase
from typing import Iterable

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

    def create_constraints(self):
        """必须创建的唯一性约束"""
        with self._driver.session() as session:
            session.run("CREATE CONSTRAINT IF NOT EXISTS FOR (e:Entity) REQUIRE e.id IS UNIQUE")

    @staticmethod
    def _create_entities(tx, batch: Iterable[Entity]):
        """批量插入的 Transaction 处理"""
        tx.run("""
        UNWIND $batch as item
        MERGE (e:Entity {id: item.text})
        SET e.label = item.label
        """, batch=[item.dict() for item in batch])

    def bulk_insert(self, entities: List[Entity]):
        """性能关键：每 1000 条作为一个批次"""
        with self._driver.session() as session:
            for i in range(0, len(entities), 1000):
                batch = entities[i:i+1000]
                session.write_transaction(self._create_entities, batch)

基于 Cypher 的简单推理规则示例：

def infer_chain(graph: KnowledgeGraph, start_entity: str):
    """实现两跳推理查询"""
    with graph._driver.session() as session:
        result = session.run("""
        MATCH (start:Entity {id: $name})-[:REL_TYPE*1..2]->(end)
        RETURN end.id as result
        """, name=start_entity)
        return [record["result"] for record in result]

批量插入优化 ：
使用 UNWIND 语句处理批量数据
合理设置 batch size（建议 500-2000）
索引策略 ：
为高频查询属性创建索引
复合查询使用复合索引
示例：CREATE INDEX FOR (e:Entity) ON (e.label, e.id)

当图谱中大部分节点只有 1 - 2 个连接时：

解决方案：
添加虚拟中间节点（如 ” 事件 ” 节点）
使用向量嵌入补充相似性连接

多数据源合并时的常见问题：

def align_entities(entity1: Entity, entity2: Entity) -> bool:
    """基于规则的简单对齐方案"""
    # 名称相似度
    if levenshtein(entity1.text, entity2.text) < 2:
        return True

    # 别名表查找
    alias_map = {"USA": "United States"}
    return alias_map.get(entity1.text) == entity2.text

现有知识图谱的局限在于：
– 规则引擎无法处理模糊推理
– 静态图谱难以适应动态知识

可能的结合方式：
1. 使用 LLM 生成候选推理路径
2. 将自然语言问题转为 Cypher 查询
3. 基于嵌入的混合检索方案

通过本教程，我们完成了从原始文本到可推理知识库的完整链路。知识图谱不是银弹，但确实是解决复杂语义关系的利器。期待看到大家更有创意的应用！

正文完