Neo4j + LangChain：如何构建基于知识图谱的最强RAG系统？

MATCH(n:Article) 
WHERE n.citation_count > 50
RETURN n.title, n.citation_count

query = "Find articles about climate change?"
vectorstore = Neo4jVector.from_existing_graph(**args)
vectorstore.similarity_search(query, k=3)

def route_question(state: GraphState):
    print("---ROUTE QUESTION---")
    question = state["question"]
    source = question_router.invoke({"question": question})
    if source.datasource == "vector search":
        print("---ROUTE QUESTION TO VECTOR SEARCH---")
        return "decomposer"
    elif source.datasource == "graph query":
        print("---ROUTE QUESTION TO GRAPH QA---")
        return "prompt_template"


workflow = StateGraph(GraphState)


workflow.add_node(PROMPT_TEMPLATE, prompt_template)
workflow.add_node(GRAPH_QA, graph_qa)


workflow.add_node(DECOMPOSER, decomposer)
workflow.add_node(VECTOR_SEARCH, vector_search)
workflow.add_node(PROMPT_TEMPLATE_WITH_CONTEXT, prompt_template_with_context)
workflow.add_node(GRAPH_QA_WITH_CONTEXT, graph_qa_with_context)


workflow.set_conditional_entry_point(
    route_question,
    {
        'decomposer': DECOMPOSER, 
        'prompt_template': PROMPT_TEMPLATE 
    },
)


workflow.add_edge(DECOMPOSER, VECTOR_SEARCH)
workflow.add_edge(VECTOR_SEARCH, PROMPT_TEMPLATE_WITH_CONTEXT)
workflow.add_edge(PROMPT_TEMPLATE_WITH_CONTEXT, GRAPH_QA_WITH_CONTEXT)
workflow.add_edge(GRAPH_QA_WITH_CONTEXT, END)


workflow.add_edge(PROMPT_TEMPLATE, GRAPH_QA)
workflow.add_edge(GRAPH_QA, END)


app = workflow.compile()


app.get_graph().draw_mermaid_png(output_file_path="graph.png")

class SubQuery(BaseModel):
    """将给定问题/查询分解为子查询"""


    sub_query: str = Field(
        ...,
        description="对原始问题的唯一释义。",
    )


system = """你是一名专家，能够将用户问题转换为 Neo4j Cypher 查询。
执行查询分解。给定用户问题，将其分解为两个独立的子查询，
你需要回答这些子查询以回答原始问题。


对于给定的输入问题，创建一个用于相似性搜索的查询，并创建一个用于执行 neo4j 图查询的查询。
以下是示例：
问题：查找关于光合作用的文章并返回其标题。
答案：
sub_query1：查找与光合作用相关的文章。
sub_query2：返回文章的标题
"""
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "{question}"),
    ]
)


llm_with_tools = llm.bind_tools([SubQuery])
parser = PydanticToolsParser(tools=[SubQuery])
query_analyzer = prompt | llm_with_tools | parser

template = f"""
    任务：生成用于查询图数据库的 Cypher 语句。
    说明：
    仅使用模式中提供的关系类型和属性。
    不要使用未提供的其他关系类型或属性。


    上下文来自于向量搜索 {context}


    使用上下文，创建 Cypher 语句并使用该语句进行图查询。
    """

context = state["article_ids"]


prefix = f"""
任务：生成用于查询图数据库的 Cypher 语句。
说明：
仅使用模式中提供的关系类型和属性。
不要使用未提供的其他关系类型或属性。
...
...
上下文以元组 ('a..', 'W..') 的形式提供
使用元组的第二个元素作为节点 ID，例如 'W..... 
以下是上下文：{context}


使用上述上下文中的节点 ID，创建 Cypher 语句并使用该语句进行图查询。
示例：以下是针对一些问题示例生成的 Cypher 语句示例：
"""


FEW_SHOT_PROMPT = FewShotPromptTemplate(
    example_selector = example_selector,
    example_prompt = example_prompt,
    prefix=prefix,
    suffix="问题：{question}, \\nCypher 查询：",
    input_variables =["question", "query"],
) 
return FEW_SHOT_PROMPT

template = f"""
任务：生成用于查询图数据库的 Cypher 语句。
说明：
仅使用模式中提供的关系类型和属性。
不要使用未提供的其他关系类型或属性。


上下文来自于向量搜索 {context}


使用上下文，创建 Cypher 语句并使用该语句进行图查询。
"""


PROMPT = PromptTemplate(
    input_variables =["question", "context"],
    template = template,
) 

chain = (
    { 
     "question": RunnablePassthrough(),
     "context" : RetrievalQA.from_chain_type(),
    }
    | PROMPT
    | GraphCypherQAChain() 
)

class RouteQuery(BaseModel):
    """将用户查询路由到最相关的数据源。"""


    datasource: Literal["vector search", "graph query"] = Field(
        ...,
        description="给定用户问题选择将其路由到向量存储或图数据库。",
    )
    
llm = ChatOpenAI(temperature=0)
structured_llm_router = llm.with_structured_output(RouteQuery)


system = """你是一名专家，能够将用户问题路由以执行向量搜索或图查询。
向量存储包含与文章标题、摘要和主题相关的文档。以下是三个路由情况：
如果用户问题涉及相似性搜索，请执行向量搜索。用户查询可能包含类似“相似”、“相关”、“相关性”、“相同”、“最近”等术语，表明向量搜索。对于其他情况，请使用图查询。


向量搜索案例的问题示例：
    查找关于光合作用的文章
    查找与氧化应激相关的类似文章


图数据库查询的问题示例：
    MATCH (n:Article) RETURN COUNT(n)
    MATCH (n:Article) RETURN n.title


图 QA 链的问题示例：
    查找特定年份发表的文章并返回其标题、作者
    查找来自位于特定国家（例如日本）的机构的作者
"""


route_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "{question}")
    ]
)


question_router = route_prompt | structured_llm_router


def route_question(state: GraphState):
    print("---ROUTE QUESTION---")
    question = state["question"]
    source = question_router.invoke({"question": question})
    if source.datasource == "vector search":
        print("---ROUTE QUESTION TO VECTOR SEARCH---")
        return "decomposer"
    elif source.datasource == "graph query":
        print("---ROUTE QUESTION TO GRAPH QA---")
        return "prompt_template"

class GraphState(TypedDict):
    """
    表示我们图的状态。


    属性：
        question: 问题
        documents: 链的结果
        article_ids: 来自向量搜索的文章 ID 列表
        prompt: 提示模板对象
        prompt_with_context: 来自向量搜索的带上下文的提示模板
        subqueries: 分解的查询
    """


    question: str
    documents: dict
    article_ids: List[str]
    prompt: object
    prompt_with_context: object
    subqueries: object

def prompt_template_with_context(state: GraphState):


    question = state["question"]  
    queries = state["subqueries"] 


    prompt_with_context = create_few_shot_prompt_with_context(state)
    
    return {"prompt_with_context": prompt_with_context, "question":question, "subqueries": queries}

app.invoke({"question": "查找引用次数最多的前 5 篇文章并返回其标题"})


---ROUTE QUESTION---
---ROUTE QUESTION TO GRAPH QA---


> 正在进入新的 GraphCypherQAChain 链...
生成的 Cypher：
MATCH (a:Article) WITH a ORDER BY a.citation_count DESC RETURN a.title LIMIT 5


> 链完成。


graph_qa_result['documents']


{'query': '查找引用次数最多的前 5 篇文章并返回其标题',
 'result': [{'a.title': '从蚯蚓堆肥中分离出的腐殖酸增强了玉米根的根伸长、侧根出现和质膜 H+-ATPase 活性'},
  {'a.title': '快速估算相对含水量'},
  {'a.title': 'ARAMEMNON，一个用于阿拉伯芥整合膜蛋白的新数据库'},
  {'a.title': '植物生理学中的多胺。'},
  {'a.title': '对阿拉伯芥根和芽中硝酸盐反应的微阵列分析揭示了 1000 多个快速反应基因以及与葡萄糖、海藻糖-6-磷酸、铁和硫酸盐代谢的新联系'}]}

app.invoke({"question": "查找关于氧化应激的文章。返回最相关文章的标题"})


---ROUTE QUESTION---
---ROUTE QUESTION TO VECTOR SEARCH---


> 正在进入新的 RetrievalQA 链...


> 链完成。


graph_qa_result['documents']


{'query': '返回最相关文章的标题。',
 'result': [{'a.title': '苔藓和谷类之间对脱落酸和应激的分子反应的保守性'}]}


graph_qa_result.keys()
dict_keys(['question', 'documents', 'article_ids', 'prompt_with_context', 'subqueries'])


graph_qa_result['subqueries']


[SubQuery(sub_query='查找与氧化应激相关的文章。'),
 SubQuery(sub_query='返回最相关文章的标题。')]