GraphRAG：知识图谱+RAG技术

由于RAG的一些缺陷，最近工作中用到了GraphRAG，可以简单理解为知识图谱+RAG工作，在进行QFS与深度推理上表现良好。老板希望，在类似于库存管理、预测上面，可以结合更多的上下文信息和逻辑关系，进行推理和结果的输出。（下图为RAG简单示意图）

RAG

RAG，又称检索增强生成（Retrieval Augumented Generation）。RAG目前已经成为大语言模型应用的核心组件之一，利用检索到的内容，无论是相似度检索，还是传统搜索、搜索引擎来增强大语言模型的外部知识。凭借这一能力，RAG在帮助LLM处理私有化数据集上发挥了巨大的潜力、取得了不俗的表现，这些私有化数据集对于LLM来说是全新的，比如企业内部文档、商业文件、用户对话记录等。

给定一个查询(Query)，RAG从一个垂直领域知识库(量化后的“Vector Database”)中检索(Retrieve)与查询内容相关的文档或段落(Context，简称“上下文”)，并将查询信息(Query)与检索到的上下文信息(Context)进行组合转化成LLM的“提示词”((Prompt)，由LLM生成(Generate)答案(Response)。这便于RAG的应用原理。

自定义Agent

在介绍RAG之前，先来介绍一下Langchain-chatchat的自定义Agent功能，由于在工作中遇到的业务场景，希望将预测模型（经典时序预测模型、深度时序预测模型、时序大模型）作为Agent供大模型去调用，因此基于Qwen2大模型基座，实现了自定义Agent功能。

预测Agent代码：

@regist_tool(title="时序预测工具")
def forecast(data: str = Field(description="the input data of timeseries"),forecast_horizon:str = Field(description="forecast horizon"),model_name:str = Field(description="the name of forecasting model")):
    "Use this tool to select a model to forecast timeseries "
    data = list(map(float, data.split()))
    n_forecast = int(forecast_horizon)
    model_name = model_name

    window_size = 3  # 滑动窗口大小

    # 进行预测
    if model_name=="lightgbm":
        forecast = lightgbm(data,n_forecast)
    else:
        forecast = linear_regression_forecast(data, n_forecast, window_size)
    forecast = [round(i,3) for i in forecast]
    # str_list = [str(i) for i in forecast]  # 将整数转换为字符串列表
    # str_list = ' '.join(str_list)
    res = {"使用的模型和方法是":model_name,
           "result":forecast}
    return BaseToolOutput(res, format="json")

将此预测Agent定义好后，LLM作为大脑，当被要求进行预测任务时，就可以调用合适的工具并传入数据和参数完成任务。

需要注意的是，我们需要提前加载具有Agent能力的大模型，如Qwen2、GPT4等，只有具备Agent能力的大模型才能根据输入，关联到合适的Agent，并解析输入的参数：1.预测工具Agent，2.模型选择LightGBM，3.预测的长度为1。

只需要在对话框中输入数据、模型、预测步长，就可以输出预测结果为：5.33。

后续可能会拓展功能，通过前端页面传入csv等结构化数据，同时可以通过简单的指令实现不同数据字段的选取和处理、结果的输出和后处理等。

传统RAG

简单地说，标准的 RAG 案例包括以下步骤：

将文本切分成多个段落，使用 Transformer Encoder 模型将这些段落转换成向量，然后把这些向量存入一个索引里。
为大语言模型（大语言模型）创建一个提示，指引模型根据我们在搜索步骤中找到的上下文来回答用户的问题。在实际运行时，我们会用同一个 Encoder 模型对用户的查询进行向量化处理，在索引中搜索这个查询向量，找出最相关的前 k 个结果。
我们从数据库中提取对应的文本段落，并将它们作为上下文融入到大语言模型的提示中。

高级RAG

高级RAG的内容待后续补充，或者可参考原文：https://pub.towardsai.net/advanced-rag-techniques-an-illustrated-overview-04d193d8fec6。

虽然RAG可以通过将生成的回答与真实数据相关联来减少幻觉，然而对于较为复杂的信息，由于其“检索”能力的局限性而往往导致回答的准确性不尽人意。RAG检索能力的局限性主要体现在情境理解方面。RAG模型只能检索(Retrieve)到数据集中明显包含有与查询(Query)信息相关的文档或段落，而对于一些不太明显的隐性关系，RAG往往无能为力。例如，对于一些非结构化文本数据, 尽管客观上是存在与查询信息相关的一些信息的，但由于这些信息显现程度不够，RGA要么是给出很粗糙的答案，要么是直接“无法回答”。

RAG缺陷与改进

传统 RAG 技术在某些场景下，可能存在以下几个问题，即使使用高级 RAG 方法通常也较难克服。

效率问题：基于向量的搜索方法采用数学方法，如聚类，树形结构或 HNSW 等近似最近邻算法，这些方法在处理极高维度数据，或非常复杂的信息结构时效果不好。另外，ANN 搜索算法如 HNSW 虽然可以提高效率，但构建和维护索引通常需要大量的计算资源。
可解释性：文本 Embedding 后的向量，可解释性很低。RAG 检索后得到的向量只关注文本的片段，而表示文本片段的向量是数字数组，没有直接的可解释性，无法通过观察向量中的具体数字理解文本内容。
整体理解受限：RAG 检索到的内容（向量块）虽然来自数据库中所包含的文档，但这些内容相关的上下文却不一定包含在答案中，导致其无法对问题、答案以及文档形成整体理解。
数值与文本：基于向量的检索，同时处理包含文本和数字的数据时准确度不高。向量数据库中，文本和数值型数据的特征和表示方法不同，系统会在处理中混淆这两种类型的数据（文本经过 Embedding 转化为向量，而数值型数据则可能直接使用或经过标准化后使用）。这也是在做这类应用中经常会遇到的问题，比如在混合了数值型表格和文本的规范中查询信息，向量搜索的准确度很低。
工程挑战：RAG 系统涉及包含检索、排序、生成等多个组件，确保各组件之间的数据无碍传输、接口统一，以及如何高效地集成和优化这些组件，是一个巨大的工程挑战。
忽视关系：实际上，文本内容并非孤立存在，而是相互关联的。传统的 RAG 无法捕获仅靠语义相似性无法呈现的重要结构化关系知识。比如，在通过引用关系连接论文的引用网络中，传统的 RAG 方法侧重于依据查询找到相关论文，却忽略了论文之间重要的引用关系。
冗余信息：RAG 在连接成提示时，常以文本片段的形式重复内容，致使上下文过长，陷入“Lost in the Middle”的困境。
缺乏全局信息：RAG 只能检索文档的子集，无法全面掌握全局信息，因而在诸如查询聚焦摘要（Query-Focused Summarization，QFS）等任务中表现不佳。

GraphRAG

原理介绍

2024年4月，微软推出GraphRAG，并于7月2日开源。GraphRAG仍然沿袭了RAG的思路，即通过检索来增强模型的准确性。不过，与RAG不同的是，GraphRAG还引入了“知识图谱”(Knowledge Graph)技术，以增强模型的“检索”能力，以实现对复杂信息的高效和可靠检索，从而提高LLM问答系统对于复杂信息的答案生成的准确性。

对于类似非结构化文本等复杂信息，挖掘其中的隐性信息关系的方法是“知识图谱”，即将非结构化文本等复杂信息通过实体、关系和属性抽取技术，重新组织成结构化的知识图。这种结构化的格式使得模型能够更好地理解和利用不同信息之间的相互关系,发现其中隐藏的信息连接。同时，这种结构化的检索使得生成答案在语境上更加丰富和准确。应用过程中，GraphRAG首先会利用大语言模型(LLM)对领域知识进行知识图谱化，构建可“图查询”(Graph Querying)的知识图谱数据库。更进一步地，GraphRAG不仅可以将全域知识库分割成多社区模块的知识图谱，还可以构建多层次知识图谱(从下至上信息更加抽象化和“主题化”)。这种多社区模块、多层次知识图谱技术旨在全面充分挖掘知识库中的复杂连接和隐性关系，最终实现对全域范围的各种知识关系在广度和深度上的“连点成线”。

上图为GraphRAG示意图。具体检索时，在对知识向量库(Vector Database)进行检索的同时，还将对知识图谱数据库进行检索，并将检索到的知识图谱信息和知识向量信息进行集成转化为“提示词”(Prompt)，再由LLM生成答案。

与RAG相比，GraphRAG的“提示词”不仅包含有查询信息和根据查询信息检索到的相关上下文信息，而且还集成了从领域知识图谱库中检索到的与查询信息相关的“知识图谱”(Knowledge Graph)信息(在广度和深度上与查询信息相关的各实体、各属性、各关系等信息)，更加丰富了“提示词”内容。因而， GraphRAG不仅能进一步提升LLM生成答案的准确性和可靠性，更为突出的是通过知识图谱技术显著地提升了模型的“检索”能力，从而提升了LLM问答系统对于复杂信息的处理能力。换句话说，对于复杂信息领域，GraphRAG弥补了RAG的不足。

GraphRAG 是借助外部结构化知识图谱来增进语言模型的上下文理解，并生成更具洞见响应的框架。GraphRAG 的目标在于从数据库中检索出最为相关的知识，进而提升下游任务的答案质量。

GraphRAG 的整个流程拆解为三个主要阶段：基于图的索引、图引导检索和图增强生成。

GraphRAG包括两个处理阶段，分别是：索引阶段和查询阶段。索引阶段利用LLM来自动化构建知识图谱，提取出对应的节点（如实体）、边（如关系）和协变量（如主张，claim），然后利用社区发现技术（如Leiden算法）对整个知识图谱进行子图划分，然后自底而上对子图利用LLM进行摘要、总结。针对特定查询，“全局答案（Global Search）”汇总所有与之相关的社区摘要最后汇总生成答案。

相比较传统的RAG, GraphRAG 的优势在于:

1.更好的解释性。GraphRAG通过构建知识图谱，将实体和关系结构化地表示出来，使得系统的回答更具解释性。传统的RAG主要依赖于文本嵌入，虽然能够匹配相关文本，但缺乏对数据的全局理解和解释能力。

2.更全面和多样的回答。GraphRAG通过知识图谱的构建，能够提供更全面和多样的回答。它不仅能够回答直接匹配的问题，还能通过图谱中的关系和上下文提供更深入的分析和回答。这使得GraphRAG在处理复杂查询时表现更优。

3.处理复杂查询的能力。GraphRAG在处理需要跨多个文档或数据源的复杂查询时表现更好。它能够通过知识图谱连接不同的实体和关系，提供更全面的回答。传统的RAG系统在处理这种复杂查询时，可能会因为仅依赖于文本嵌入而无法连接相关信息。

举个例子，假如有一个s针对众多科学家的介绍文章，每个科学家都有国籍，假如查询"有哪些科学家是邻居? "由于文章并不包括邻居相关的信息，邻居是需要通过科学家->国家>科学家的关系构建才能发现，因此这类问题传统RAG就无法解决。

实践操作

虽然网上对于GraphRAG+Ollama+Neo4j的教程比较多，但由于本人工作高度依赖服务器，因此在服务器部署上也踩坑许多，记录下来分享给大家。

首先按照如下文章进行操作：

https://blog.csdn.net/vivisol/article/details/140873457?ops_request_misc=&request_id=&biz_id=102&utm_term=graphrag+ollama&utm_medium=distribute.pc_search_result.none-task-blog-2~all~sobaiduweb~default-7-140873457.142^v100^pc_search_result_base2&spm=1018.2226.3001.4187

运行完GraphRAG之后，生成了如下的社区文件、关系文件、协变量文件、知识抽取文件等等。

我们可以通过如下的全局和局部查询，去获得想要的答案，查询过程就是知识图谱+RAG的结合。

python -m graphrag.query --root ./work_file --method local "什么是时间序列"
python -m graphrag.query --root ./work_file --method global "什么是时间序列"

例如我们询问：时间序列是什么？

就可以通过RAG结合知识图谱技术，为我们返回关于时间序列的数据特点、目标和方法等等。终端运行命令和结果展示如下：

(graph_rag) [root@iZbp1axhql3j600tvzgk2eZ ~]# python -m graphrag.query --root ./rag_ts --method local "什么是时间序列"

INFO: Reading settings from rag_ts/settings.yaml
creating llm client with {'api_key': 'REDACTED,len=6', 'type': "openai_chat", 'model': 'qwen2', 'max_tokens': 4000, 'temperature': 0.0, 'top_p': 1.0, 'n': 1, 'request_timeout': 180.0, 'api_base': 'http://localhost:11434/v1/', 'api_version': None, 'organization': None, 'proxy': None, 'cognitive_services_endpoint': None, 'deployment_name': None, 'model_supports_json': True, 'tokens_per_minute': 0, 'requests_per_minute': 0, 'max_retries': 10, 'max_retry_wait': 10.0, 'sleep_on_rate_limit_recommendation': True, 'concurrent_requests': 25}
creating embedding llm client with {'api_key': 'REDACTED,len=6', 'type': "openai_embedding", 'model': 'nomic-embed-text', 'max_tokens': 4000, 'temperature': 0, 'top_p': 1, 'n': 1, 'request_timeout': 180.0, 'api_base': 'http://localhost:11434/api', 'api_version': None, 'organization': None, 'proxy': None, 'cognitive_services_endpoint': None, 'deployment_name': None, 'model_supports_json': None, 'tokens_per_minute': 0, 'requests_per_minute': 0, 'max_retries': 10, 'max_retry_wait': 10.0, 'sleep_on_rate_limit_recommendation': True, 'concurrent_requests': 25}
Error embedding chunk {'OpenAIEmbedding': 'json: cannot unmarshal array into Go struct field EmbeddingRequest.prompt of type string'}

SUCCESS: Local Search Response: 时间序列是一种数据收集和分析方法，它通过在连续的时间点上测量变量来记录数据的变化。这种类型的数据通常用于跟踪随时间变化的趋势、模式或事件。

时间序列数据的特点是：

1. **顺序性**：数据按照时间的先后顺序排列。
2. **依赖性**：相邻时间点之间的值存在相关性，即当前值与过去值之间有联系。
3. **周期性**：在一定的时间间隔内重复出现的模式或趋势。

时间序列分析的目标通常包括：

- **描述**：理解数据随时间的变化情况和特征。
- **解释**：识别并解释导致数据变化的因素或原因。
- **预测**：利用历史数据来预测未来的值或趋势。
- **控制**：在工业过程、经济模型等中，通过调整变量以达到期望的输出。

时间序列分析方法包括但不限于：

- **描述性统计分析**：计算平均数、标准差、相关系数等指标来描述数据分布和关系。
- **时间序列分解**：将时间序列分为趋势、季节性、周期性和随机成分。
- **模型拟合**：使用ARIMA（自回归积分滑动平均）、GARCH（广义自回归条件异方差）等模型来拟合数据，以理解其内在结构和预测未来值。
- **预测分析**：利用历史数据建立模型，如线性回归、时间序列模型或深度学习方法，来进行未来的预测。

在实际应用中，时间序列分析广泛应用于金融（股票价格预测）、经济（GDP增长预测）、气象学（天气预报）等领域。

我们也可以对GraphRAG的输出结果进行知识图谱可视化，需要用到Neo4j。

Neo4j是由 Neo4j Inc. 开发的图数据库管理系统，是图数据库技术领域的领导者——强大的原生图存储、数据科学和分析，具备企业级的安全性。无约束地扩展您的事务和分析工作负载。已下载超过1.6亿次。Neo4j 存储的数据元素包括节点、连接它们的边以及节点和边的属性。

首先使用如下Docker命令启动Neo4J实例：

docker run \
    -p 7474:7474 -p 7687:7687 \
    --name neo4j-apoc \
    -e NEO4J_apoc_export_file_enabled=true \
    -e NEO4J_apoc_import_file_enabled=true \
    -e NEO4J_apoc_import_file_use__neo4j__config=true \
    -e NEO4J_PLUGINS=\[\"apoc\"\] \
    neo4j:5.21.2

运行此命令，出现卡顿，需要更换docker源，常见方法不管用，如下方法亲测管用：

sudo mkdir -p /etc/docker

sudo tee /etc/docker/daemon.json <<-'EOF'
{
    "registry-mirrors": [
        "https://do.nark.eu.org",
        "https://dc.j8.work",
        "https://docker.m.daocloud.io",
        "https://dockerproxy.com",
        "https://docker.mirrors.ustc.edu.cn",
        "https://docker.nju.edu.cn"
    ]
}
EOF

sudo systemctl daemon-reload
sudo systemctl restart docker

 }

之后按照如下教程即可启动：

https://mp.weixin.qq.com/s/0niNzA_ZG_L81Ah_PAflLg

部署在服务器上时需要在安全组内将：7474和7687端口打开。

顺序为先打开Neo4j，再依次运行：导入文档、TextUnits、实体、关系、社区和社区报告后可视化展示。

导入文档、TextUnits、实体、关系、社区的代码贴在下面，该段代码的作用就是根据GraphRAG生成的：

import pandas as pd
from neo4j import GraphDatabase
import time
NEO4J_URI = "neo4j://localhost"  # or neo4j+s://xxxx.databases.neo4j.io
NEO4J_USERNAME = "neo4j"
NEO4J_PASSWORD = "" #你自己的密码
NEO4J_DATABASE = "neo4j"

# Create a Neo4j driver
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))

GRAPHRAG_FOLDER = "/root/rag_ts/output/20240904-142316/artifacts"

statements = """
create constraint chunk_id if not exists for (c:__Chunk__) require c.id is unique;
create constraint document_id if not exists for (d:__Document__) require d.id is unique;
create constraint entity_id if not exists for (c:__Community__) require c.community is unique;
create constraint entity_id if not exists for (e:__Entity__) require e.id is unique;
create constraint entity_title if not exists for (e:__Entity__) require e.name is unique;
create constraint entity_title if not exists for (e:__Covariate__) require e.title is unique;
create constraint related_id if not exists for ()-[rel:RELATED]->() require rel.id is unique;
""".split(";")

for statement in statements:
    if len((statement or "").strip()) > 0:
        print(statement)
        driver.execute_query(statement)

def batched_import(statement, df, batch_size=1000):
    """
    Import a dataframe into Neo4j using a batched approach.
    Parameters: statement is the Cypher query to execute, df is the dataframe to import, and batch_size is the number of rows to import in each batch.
    """
    total = len(df)
    start_s = time.time()
    for start in range(0,total, batch_size):
        batch = df.iloc[start: min(start+batch_size,total)]
        result = driver.execute_query("UNWIND $rows AS value " + statement,
                                      rows=batch.to_dict('records'),
                                      database_=NEO4J_DATABASE)
        print(result.summary.counters)
    print(f'{total} rows in { time.time() - start_s} s.')
    return total


doc_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_documents.parquet', columns=["id", "title"])
doc_df.head(2)

# import documents
statement = """
MERGE (d:__Document__ {id:value.id})
SET d += value {.title}
"""

batched_import(statement, doc_df)

text_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_text_units.parquet',
                          columns=["id","text","n_tokens","document_ids"])
text_df.head(2)

statement = """
MERGE (c:__Chunk__ {id:value.id})
SET c += value {.text, .n_tokens}
WITH c, value
UNWIND value.document_ids AS document
MATCH (d:__Document__ {id:document})
MERGE (c)-[:PART_OF]->(d)
"""

batched_import(statement, text_df)

entity_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_entities.parquet',
                            columns=["name", "type", "description", "human_readable_id", "id", "description_embedding",
                                     "text_unit_ids"])
entity_df.head(2)

entity_statement = """
MERGE (e:__Entity__ {id:value.id})
SET e += value {.human_readable_id, .description, name:replace(value.name,'"','')}
WITH e, value
CALL db.create.setNodeVectorProperty(e, "description_embedding", value.description_embedding)
CALL apoc.create.addLabels(e, case when coalesce(value.type,"") = "" then [] else [apoc.text.upperCamelCase(replace(value.type,'"',''))] end) yield node
UNWIND value.text_unit_ids AS text_unit
MATCH (c:__Chunk__ {id:text_unit})
MERGE (c)-[:HAS_ENTITY]->(e)
"""

batched_import(entity_statement, entity_df)


rel_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_relationships.parquet',
                         columns=["source", "target", "id", "rank", "weight", "human_readable_id", "description",
                                  "text_unit_ids"])
rel_df.head(2)

rel_statement = """
    MATCH (source:__Entity__ {name:replace(value.source,'"','')})
    MATCH (target:__Entity__ {name:replace(value.target,'"','')})
    // not necessary to merge on id as there is only one relationship per pair
    MERGE (source)-[rel:RELATED {id: value.id}]->(target)
    SET rel += value {.rank, .weight, .human_readable_id, .description, .text_unit_ids}
    RETURN count(*) as createdRels
"""

batched_import(rel_statement, rel_df)


community_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_communities.parquet',
                               columns=["id", "level", "title", "text_unit_ids", "relationship_ids"])

community_df.head(2)

statement = """
MERGE (c:__Community__ {community:value.id})
SET c += value {.level, .title}
/*
UNWIND value.text_unit_ids as text_unit_id
MATCH (t:__Chunk__ {id:text_unit_id})
MERGE (c)-[:HAS_CHUNK]->(t)
WITH distinct c, value
*/
WITH *
UNWIND value.relationship_ids as rel_id
MATCH (start:__Entity__)-[:RELATED {id:rel_id}]->(end:__Entity__)
MERGE (start)-[:IN_COMMUNITY]->(c)
MERGE (end)-[:IN_COMMUNITY]->(c)
RETURn count(distinct c) as createdCommunities
"""

batched_import(statement, community_df)


community_report_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_community_reports.parquet',
                                      columns=["id", "community", "level", "title", "summary", "findings", "rank",
                                               "rank_explanation", "full_content"])
community_report_df.head(2)
# import communities
community_statement = """MATCH (c:__Community__ {community: value.community})
SET c += value {.level, .title, .rank, .rank_explanation, .full_content, .summary}
WITH c, value
UNWIND range(0, size(value.findings)-1) AS finding_idx
WITH c, value, finding_idx, value.findings[finding_idx] as finding
MERGE (c)-[:HAS_FINDING]->(f:Finding {id: finding_idx})
SET f += finding"""
batched_import(community_statement, community_report_df)

在打开"服务器公网ip+:7474/browser/"之后，点击左上角“数据库”表示，就可以看到知识图谱的可视化展示了。

参考文献

https://mp.weixin.qq.com/s/0niNzA_ZG_L81Ah_PAflLg
https://blog.csdn.net/vivisol/article/details/140873457?ops_request_misc=&request_id=&biz_id=102&utm_term=graphrag+ollama&utm_medium=distribute.pc_search_result.none-task-blog-2~all~sobaiduweb~default-7-140873457.142^v100^pc_search_result_base2&spm=1018.2226.3001.4187