QAnything与Neo4j集成：文档知识图谱构建

1. 引言

在日常工作中，我们经常需要处理大量的文档资料，比如技术文档、产品手册、研究报告等。这些文档中包含大量有价值的信息，但如何快速找到相关信息并理解其中的关联关系，一直是个头疼的问题。

传统的文档检索方式往往只能找到包含关键词的片段，却无法展示信息之间的内在联系。比如，你想了解某个技术概念在不同文档中的演变过程，或者想找出所有讨论特定技术方案的文档，传统方法就显得力不从心了。

这就是为什么我们需要将QAnything的文档解析能力与Neo4j的图数据库优势结合起来。通过这种集成，我们不仅能快速提取文档中的关键信息，还能以可视化的方式展示信息之间的复杂关系，让知识发现变得直观而高效。

2. 环境准备与快速部署

2.1 系统要求

在开始之前，确保你的系统满足以下基本要求：

• 操作系统：Linux、macOS或Windows
• 内存：至少8GB RAM（建议16GB以上）
• 存储：至少10GB可用空间
• Python版本：3.8或更高版本

2.2 安装必要的库

首先安装所需的Python包：

pip install qanything-kernel neo4j py2neo pandas numpy

如果你还没有安装Docker，建议先安装Docker，因为Neo4j通常通过Docker容器运行：

# 对于Ubuntu/Debian系统
sudo apt-get update
sudo apt-get install docker.io

# 对于CentOS/RHEL系统
sudo yum install docker
sudo systemctl start docker

2.3 启动Neo4j数据库

使用Docker快速启动Neo4j实例：

docker run \
    --name neo4j \
    -p 7474:7474 \
    -p 7687:7687 \
    -d \
    -v neo4j_data:/data \
    -v neo4j_logs:/logs \
    -v neo4j_import:/var/lib/neo4j/import \
    --env NEO4J_AUTH=neo4j/password123 \
    neo4j:latest

这样就在本地启动了一个Neo4j数据库，可以通过浏览器访问 http://localhost:7474 来管理数据库。

3. QAnything文档解析基础

3.1 初始化QAnything

首先让我们初始化QAnything来处理文档：

from qanything_kernel.core.local_doc_qa import LocalDocQA
from qanything_kernel.configs.model_config import *

# 初始化QAnything
local_doc_qa = LocalDocQA()
local_doc_qa.init_cfg(
    embedding_model=EMBEDDING_MODEL,
    embedding_device=EMBEDDING_DEVICE,
    rerank_model=RERANK_MODEL,
    rerank_device=RERANK_DEVICE
)

3.2 文档解析示例

让我们用一个简单的PDF文档来测试解析功能：

import os
from pathlib import Path

# 准备示例文档
doc_path = "sample_technical.pdf"

# 解析文档
def parse_document(file_path):
    if not os.path.exists(file_path):
        print(f"文档 {file_path} 不存在")
        return None
    
    # 使用QAnything解析文档
    documents = local_doc_qa.load_file(file_path)
    
    print(f"成功解析文档，共提取 {len(documents)} 个文本块")
    for i, doc in enumerate(documents[:3]):  # 显示前3个文本块
        print(f"文本块 {i+1}: {doc.page_content[:100]}...")
    
    return documents

# 执行解析
parsed_docs = parse_document(doc_path)

4. 构建知识图谱模型

4.1 设计图数据库 schema

在将文档内容导入Neo4j之前，我们需要设计合适的数据模型。一个典型的文档知识图谱可能包含以下节点类型和关系：

• 节点类型：

  • Document（文档）
  • Entity（实体）
  • Concept（概念）
  • Topic（主题）

• 关系类型：

  • CONTAINS（包含）
  • MENTIONS（提及）
  • RELATED_TO（相关）
  • DESCRIBES（描述）

4.2 实体识别与提取

从解析的文档中提取关键实体和概念：

import re
from collections import defaultdict

def extract_entities(text):
    """
    从文本中提取可能的实体和概念
    这是一个简化的示例，实际应用中可以使用NLP库如spaCy
    """
    # 提取大写字母开头的名词（简单启发式规则）
    entities = re.findall(r'\b[A-Z][a-z]+\b', text)
    
    # 提取技术术语（包含大写字母的单词）
    tech_terms = re.findall(r'\b[A-Za-z]{3,}[A-Z][A-Za-z]*\b', text)
    
    # 合并结果
    all_entities = list(set(entities + tech_terms))
    
    return all_entities

def process_documents(documents):
    """
    处理所有文档，提取实体和关系
    """
    knowledge_graph = {
        'documents': [],
        'entities': defaultdict(list),
        'relationships': []
    }
    
    for doc in documents:
        doc_id = f"doc_{hash(doc.metadata.get('source', 'unknown'))}"
        doc_content = doc.page_content
        
        # 记录文档信息
        knowledge_graph['documents'].append({
            'id': doc_id,
            'content': doc_content[:200] + "...",  # 只存储前200字符
            'metadata': doc.metadata
        })
        
        # 提取实体
        entities = extract_entities(doc_content)
        for entity in entities:
            knowledge_graph['entities'][entity].append(doc_id)
            knowledge_graph['relationships'].append({
                'from': doc_id,
                'to': entity,
                'type': 'MENTIONS'
            })
    
    return knowledge_graph

# 处理解析的文档
kg_data = process_documents(parsed_docs)
print(f"提取了 {len(kg_data['entities'])} 个唯一实体")

5. 导入Neo4j图数据库

5.1 连接Neo4j数据库

from neo4j import GraphDatabase

class Neo4jConnector:
    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def close(self):
        self.driver.close()
    
    def create_document_node(self, doc_id, content, metadata):
        with self.driver.session() as session:
            result = session.execute_write(
                self._create_and_return_document, doc_id, content, metadata
            )
            return result
    
    @staticmethod
    def _create_and_return_document(tx, doc_id, content, metadata):
        query = (
            "CREATE (d:Document {id: $doc_id, content: $content, source: $source}) "
            "RETURN d.id AS id"
        )
        result = tx.run(query, doc_id=doc_id, content=content, 
                       source=metadata.get('source', 'unknown'))
        return result.single()["id"]
    
    def create_entity_node(self, entity_name):
        with self.driver.session() as session:
            result = session.execute_write(
                self._create_and_return_entity, entity_name
            )
            return result
    
    @staticmethod
    def _create_and_return_entity(tx, entity_name):
        query = (
            "MERGE (e:Entity {name: $name}) "
            "RETURN e.name AS name"
        )
        result = tx.run(query, name=entity_name)
        return result.single()["name"]
    
    def create_relationship(self, from_id, to_entity, rel_type):
        with self.driver.session() as session:
            result = session.execute_write(
                self._create_and_return_relationship, from_id, to_entity, rel_type
            )
            return result
    
    @staticmethod
    def _create_and_return_relationship(tx, from_id, to_entity, rel_type):
        query = (
            "MATCH (d:Document {id: $from_id}) "
            "MATCH (e:Entity {name: $to_entity}) "
            "MERGE (d)-[r:MENTIONS]->(e) "
            "SET r.type = $rel_type "
            "RETURN d.id, type(r), e.name"
        )
        result = tx.run(query, from_id=from_id, to_entity=to_entity, rel_type=rel_type)
        return result.single()

# 初始化连接
neo4j_conn = Neo4jConnector("bolt://localhost:7687", "neo4j", "password123")

5.2 批量导入数据

def import_to_neo4j(kg_data, connector):
    """
    将知识图谱数据导入Neo4j
    """
    # 导入文档节点
    for doc in kg_data['documents']:
        connector.create_document_node(
            doc['id'], doc['content'], doc['metadata']
        )
    
    # 导入实体节点和关系
    for entity, doc_ids in kg_data['entities'].items():
        connector.create_entity_node(entity)
        for doc_id in doc_ids:
            connector.create_relationship(doc_id, entity, 'MENTIONS')
    
    print("数据导入完成")

# 执行导入
import_to_neo4j(kg_data, neo4j_conn)

6. Cypher查询优化技巧

6.1 基础查询示例

让我们学习一些常用的Cypher查询：

def run_cypher_query(connector, query, parameters=None):
    """
    执行Cypher查询
    """
    with connector.driver.session() as session:
        result = session.run(query, parameters or {})
        return [record for record in result]

# 查询所有文档及其提及的实体
query_all = """
MATCH (d:Document)-[r:MENTIONS]->(e:Entity)
RETURN d.id AS document, e.name AS entity, r.type AS relationship
ORDER BY document
"""

results = run_cypher_query(neo4j_conn, query_all)
for record in results[:5]:  # 显示前5条结果
    print(f"文档 {record['document']} 提及了实体 {record['entity']}")

6.2 高级查询技巧

# 查找提及特定实体的所有文档
def find_documents_by_entity(entity_name):
    query = """
    MATCH (d:Document)-[:MENTIONS]->(e:Entity {name: $entity_name})
    RETURN d.id AS document_id, d.content AS preview
    ORDER BY document_id
    """
    return run_cypher_query(neo4j_conn, query, {'entity_name': entity_name})

# 查找实体共现关系
def find_entity_cooccurrence():
    query = """
    MATCH (d:Document)-[:MENTIONS]->(e1:Entity)
    MATCH (d:Document)-[:MENTIONS]->(e2:Entity)
    WHERE e1.name < e2.name  // 避免重复对
    RETURN e1.name AS entity1, e2.name AS entity2, count(d) AS cooccurrence_count
    ORDER BY cooccurrence_count DESC
    LIMIT 10
    """
    return run_cypher_query(neo4j_conn, query)

# 执行查询
ai_docs = find_documents_by_entity("AI")
print(f"找到 {len(ai_docs)} 篇提及AI的文档")

cooccurrences = find_entity_cooccurrence()
print("最常见的实体共现关系：")
for record in cooccurrences:
    print(f"{record['entity1']} 和 {record['entity2']}: {record['cooccurrence_count']}次")

7. 子图提取与可视化探索

7.1 提取相关子图

def extract_entity_subgraph(entity_name, depth=2):
    """
    提取以某个实体为中心的子图
    """
    query = """
    MATCH path = (e:Entity {name: $entity_name})<-[*1..$depth]-(d:Document)
    WITH e, d, relationships(path) AS rels
    RETURN e.name AS center_entity, 
           d.id AS document, 
           [r IN rels | type(r)] AS relationships
    """
    return run_cypher_query(neo4j_conn, query, 
                          {'entity_name': entity_name, 'depth': depth})

# 提取AI相关的子图
ai_subgraph = extract_entity_subgraph("AI")
print(f"AI实体相关的子图包含 {len(ai_subgraph)} 个节点")

7.2 可视化探索

虽然Neo4j自带了浏览器可视化界面，但我们也可以使用Python库进行可视化：

import matplotlib.pyplot as plt
import networkx as nx
from pyvis.network import Network

def visualize_subgraph(entity_name):
    """
    使用pyvis可视化子图
    """
    # 提取子图数据
    query = """
    MATCH (e:Entity {name: $entity_name})
    MATCH (e)<-[:MENTIONS]-(d:Document)
    OPTIONAL MATCH (d)-[:MENTIONS]->(other:Entity)
    WHERE other.name <> $entity_name
    RETURN d.id AS document, other.name AS entity, 
           count(*) AS strength
    """
    
    results = run_cypher_query(neo4j_conn, query, {'entity_name': entity_name})
    
    # 创建网络图
    net = Network(height="600px", width="100%", bgcolor="#222222", font_color="white")
    
    # 添加中心节点
    net.add_node(entity_name, label=entity_name, color="#ff0000", size=30)
    
    # 添加文档和实体节点
    document_nodes = set()
    entity_nodes = set()
    
    for record in results:
        doc_id = record['document']
        entity = record['entity']
        strength = record['strength']
        
        if doc_id not in document_nodes:
            net.add_node(doc_id, label=doc_id[:10]+"...", color="#00ff00", size=15)
            document_nodes.add(doc_id)
        
        if entity and entity not in entity_nodes:
            net.add_node(entity, label=entity, color="#0000ff", size=20)
            entity_nodes.add(entity)
        
        if entity:  # 文档到实体的关系
            net.add_edge(doc_id, entity, value=strength)
        
        # 文档到中心实体的关系
        net.add_edge(doc_id, entity_name, value=strength)
    
    # 生成HTML文件
    net.show(f"{entity_name}_subgraph.html")
    print(f"可视化结果已保存到 {entity_name}_subgraph.html")

# 生成AI实体的子图可视化
visualize_subgraph("AI")

8. 实际应用案例

8.1 技术文档分析

假设我们有一组技术文档，想要分析其中的技术概念关联：

def analyze_technical_documents():
    """
    分析技术文档中的概念关联
    """
    # 查找最常被讨论的技术概念
    query = """
    MATCH (e:Entity)<-[:MENTIONS]-(d:Document)
    WHERE d.source CONTAINS 'technical'
    RETURN e.name AS technology, count(d) AS document_count
    ORDER BY document_count DESC
    LIMIT 10
    """
    
    top_techs = run_cypher_query(neo4j_conn, query)
    print("最常被讨论的技术概念：")
    for i, record in enumerate(top_techs, 1):
        print(f"{i}. {record['technology']}: {record['document_count']}篇文档")
    
    return top_techs

# 执行分析
top_technologies = analyze_technical_documents()

8.2 研究趋势分析

def analyze_research_trends():
    """
    分析研究趋势和概念演变
    """
    # 查找概念共现网络中的核心节点
    query = """
    MATCH (e:Entity)<-[:MENTIONS]-(:Document)
    WITH e, count(*) as doc_count
    WHERE doc_count > 2  // 只考虑出现在多篇文档中的实体
    MATCH (e)<-[:MENTIONS]-(d:Document)
    MATCH (d)-[:MENTIONS]->(other:Entity)
    WHERE other <> e AND size((other)<-[:MENTIONS]-()) > 2
    WITH e, other, count(d) AS cooccurrence
    WHERE cooccurrence > 1
    RETURN e.name AS entity1, other.name AS entity2, cooccurrence
    ORDER BY cooccurrence DESC
    LIMIT 15
    """
    
    trends = run_cypher_query(neo4j_conn, query)
    print("核心概念关联网络：")
    for record in trends:
        print(f"{record['entity1']} ←→ {record['entity2']} ({record['cooccurrence']}次)")
    
    return trends

# 分析研究趋势
research_trends = analyze_research_trends()

9. 总结

通过将QAnything的文档解析能力与Neo4j的图数据库优势相结合，我们创建了一个强大的文档知识图谱系统。这个系统不仅能够提取文档中的关键信息，还能以可视化的方式展示信息之间的复杂关系，大大提升了知识发现的效率。

实际使用下来，这种集成方案确实很实用。文档解析部分处理得很干净，提取的实体质量也不错。Neo4j的图数据库特性让关系查询变得特别方便，特别是当你需要找出隐藏的模式或者关联时，图查询的优势就体现出来了。

如果你正在处理大量的技术文档或研究资料，建议试试这种方法。开始时可以从小的文档集入手，熟悉了整个流程后再扩展到更大的数据集。实践中可能会遇到一些数据质量的问题，比如实体识别不够准确，这时候可以考虑引入更先进的NLP技术来提升识别精度。

这个方案还有很多可以扩展的地方，比如加入时间维度来分析概念的演变，或者整合多语言文档的处理能力。希望这个教程能为你构建自己的知识管理系统提供一些有用的思路。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。