基于Langchain在NebulaGraph实现GraphRAG

项目地址：https://github.com/EvannZhongg/NebulaGraphQAChain.git

1. 这个代码是做什么的？

  • 把自然语言问题转化为图数据库NebulaGraph查询语句
  • 查询 NebulaGraph（确保图数据库中已经存在数据，本代码不再额外构建数据插入图数据库）
  • 返回查询结果

简单来说：
你可以用“人话”问它问题，它会自动 翻译成图数据库语言，然后去 图数据库找答案，将查询到的结果与提问相结合重新用“人话”回答问题。

2. 代码执行的流程是什么？

  1. 读取 API 配置（告诉代码如何访问 LLM）
  2. 调用 大模型（LLM），让它帮我们翻译问题
  3. 连接 NebulaGraph 数据库
  4. 执行 数据库查询，获得结果
  5. 返回最终的查询结果

3. 代码分解讲解

3.1 读取环境变量（API 配置）

1
2
3
4
5
6
7
8
9
10
11
from dotenv import load_dotenv
import os

# 读取 .env 文件
load_dotenv()

CHAT_API_URL = os.getenv("CHAT_API_URL")
API_KEY = os.getenv("API_KEY")

if not CHAT_API_URL or not API_KEY:
    raise ValueError("请在 .env 文件中正确配置 CHAT_API_URL 和 API_KEY")

作用：

  • 读取 API 地址 和 密钥，让代码知道如何访问大模型。
  • 如果没有配置 API，就会报错，提醒用户。

🔹 什么是 .env 文件？

.env 文件用于存放 API 密钥，这样我们就不用在代码里写明白，保证安全，你可以在项目目录中新建一个.env 文件，本文使用的是siliconflow平台提供的API服务，你可以在siliconflow平台进行注册申请API KEY填入下方的API_KEY变量中。

1
2
CHAT_API_URL=https://api.siliconflow.cn/v1/chat/completions
API_KEY=sk-***************************

3.2 调用 LLM API

代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def call_chat_api(prompt):
    """
    调用 LLM API 生成 Cypher 查询
    """
    try:
        response = requests.post(
            CHAT_API_URL,
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={
                "model": "deepseek-ai/DeepSeek-V3",  # 选择模型
                "messages": [{"role": "user", "content": prompt}]
            }
        )
        
        if response.status_code == 200:
            data = response.json()
            result = data["choices"][0]["message"]["content"]
            
            # 处理 Markdown 代码块
            if result.startswith("```nebula") and result.endswith("```"):
                result = result[len("```nebula"):-len("```")].strip()
            elif result.startswith("```cypher") and result.endswith("```"):
                result = result[len("```cypher"):-len("```")].strip()
            elif result.startswith("```sql") and result.endswith("```"):
                result = result[len("```sql"):-len("```")].strip()
            elif result.startswith("```") and result.endswith("```"):
                result = result[len("```"):-len("```")].strip()
            
            return result
        else:
            print(f"API 调用失败: {response.status_code} {response.text}")
            return None
    except Exception as e:
        print(f"API 调用出错: {e}")
        return None

作用：

  1. 发送请求 给 LLM（大模型），让它把用户问题翻译成 数据库查询语句。
  2. 解析返回结果，提取真正的数据库查询语句。
  3. 如果出错，就打印错误信息。
  • 代码中的以下部分用于处理大模型输出含有代码标记块的情况，如果你的大模型输出含有代码标记块则查询语句会出现失败，在下方代码中添加代码标记块名称，避免查询出错.
    1
    2
    if result.startswith("```代码标记块内容") and result.endswith("```"):
      result = result[len("```"):-len("```")].strip()

示例：

  • 你输入：“列出某个电子器件的所有相关参数”
  • LLM 可能返回：
    1
    MATCH (n:Component) RETURN n.parameters
    这个就是 Cypher 查询语句，可以直接在图数据库执行。

3.3 自定义 LLM 处理类生成查询语句

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from langchain_core.runnables import Runnable
from langchain_core.prompt_values import StringPromptValue

class CustomChatModel(Runnable):
    def __init__(self):
        super().__init__()

    def invoke(self, input, config: dict = None, **kwargs):
        if isinstance(input, StringPromptValue):
            input = input.to_string()
        
        if not isinstance(input, str):
            raise ValueError(f"Expected input to be a string, got {type(input)} instead.")
        
        result = call_chat_api(input)
        if result is None:
            raise ValueError("API 调用未返回有效响应")
        return result

作用：

  • 让 LangChain 框架 知道如何调用 LLM。
  • 处理输入数据，确保 LLM 可以正确理解问题。
  • 调用 call_chat_api()，获取 LLM 生成的数据库查询语句。

3.4 连接 NebulaGraph

代码：

1
2
3
4
5
6
7
8
9
10
from langchain_community.graphs import NebulaGraph

graph = NebulaGraph(
    space="SPACE",  #将SPACE修改为自己的图空间名称
    username="root",
    password="nebula",
    address="127.0.0.1",
    port=9669,
    session_pool_size=30,
)

作用：

  • 连接 NebulaGraph 图数据库，让代码可以查询数据。
  • 配置 数据库地址、用户名、密码。

3.5 运行查询示例

代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def query_graph(question):
    """
    运行 NebulaGraph QAChain，处理自然语言查询
    """
    try:
        answer = chain.run(question)
        return answer
    except Exception as e:
        print(f"查询执行出错: {e}")
        return None

if __name__ == "__main__":
    test_question = input("请输入您的查询问题: ")  # 让用户在终端输入问题
    response = query_graph(test_question)
    print("查询结果:", response)

作用：

  • 让用户输入问题。
  • 调用 chain.run(question) 处理查询。
  • 返回查询语句和结果。

📌 示例：

  1. 用户输入：“1N4736AT-D的应用有什么”
  2. 代码调用 LLM，获取 Cypher 语句
  3. 执行数据库查询，返回结果。
    Query Execution
  4. 我们可以在图数据库中进行验证。
    Nebula Graph Query
  5. 这里制作了一个简易的前端页面，可以查看效果和后端日志，详细代码可在本文开头的项目地址中获取。
    image

该项目代码参考 LangChain 和 NebulaGraphQAChain。