首先在数据集 - 开放知识图谱下载红楼梦的知识图谱，这个网站上有各种各样的知识图谱，可以挑你感兴趣的做( • ̀ω•́ )

这个知识图谱的作者们已经将三元组抽取出来了，我们可以直接用，如果你对三元组是如何生成的感兴趣，可以看看他们之前的步骤。对于这个项目，我们只需要打开neo_db文件夹，修改config文件并运行creat_graph文件即可在neo4j上生成图谱，这个数据集中还提供了红楼梦的txt文件，我们在后面的RAG部分可以用。

导包

from langchain_community.graphs import Neo4jGraph
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser

配置graph和llm

graph=Neo4jGraph(
    url="bolt://127.0.0.1:7687",
    username="neo4j",
    password="yourpassword",
    refresh_schema=False
)
llm=ChatOpenAI(
    model='your model',
    temperature=0,
    max_tokens=None,
    max_retries=2,
    api_key='your api_key',
    base_url='your url'
)

不使用GraphCypherQAChain，GraphCypherQAChain依赖 APOC，易出错，且不透明。

自己控制 Cypher 生成流程，更安全、更可控。

schema不需要列出所有关系，但必须包含与当前任务相关的“核心模式”（Relevant Schema）。

SCHEMA="""
- (:Entity {Name: STRING})
- (:Entity)-[:FATHER|MOTHER|SPOUSE|CHILD|LOVES]->(:Entity)
"""

cypher_prompt=PromptTemplate.from_template(
"""
你是一个Cypher专家，请根据以下schema，将自然语言问题转换成Cypher查询。
只返回Cypher语句，不要解释。

Schema:
{schema}

问题：{question}
Cypher:
"""
).partial(schema=SCHEMA)

graph_chain=cypher_prompt | llm | StrOutputParser()

question = '邢夫人和邢岫烟是什么关系?'
cypher=graph_chain.invoke({'question':question})
print(cypher)

使用双层路由，简单规则 + LLM fallback，兼顾效率与准确

def double_router(question):
    kg_keywords=['谁','关系','共同','一起','属于','路径']
    rag_keywords=['介绍','原理','解释','背景']
    
    kg_score=sum(1 for kw in kg_keywords if kw in question)
    rag_score=sum(1 for kw in rag_keywords if kw in question)
    
    if kg_score>rag_score:
        category='KG'
        print(category)
        return category
    elif rag_score>0:
        category='RAG'
        print(category)
        return category
    router_prompt=PromptTemplate.from_template(
        """
        你是一个问题分类器，请判断一下问题更适合通过哪种方式回答：
        -RAG:基于文档检索的答案（如定义，描述，解释）
        -KG:基于知识图谱的关系查询（如人物关系）
        只需返回RAG或KG
        问题：{question}
        类别：
        """
    )
    )
    router_chain=router_prompt | llm | StrOutputParser()
    category=router_chain.invoke({'question':question})
    print(category)
    return category
category=double_router(question)

对相同或相似问题使用缓存，提升响应速度。

先查图谱找核心实体 → 用实体去文档中找更多背景信息

@lru_cache(maxsize=100)
def cached_kg_query(question):
    cypher=graph_chain.invoke({'question':question})
    return graph.query(cypher)
def answer(question,category):
    if category=='KG':
        try:
            result=cached_kg_query(question)
            if result:
                relationship=result[0].get("type(r)") or result[0].get("relationship")
                enhanced_query=f"{question},涉及{relationship}的背景信息"
                context=with_message_history.invoke(
        {'question':enhanced_query},
        config=config
    )
                answer="{}\n{}".format(relationship,context)
                category='KG+RAG'
            else:
                answer="图谱中未找到"
        except Exception as e:
            answer = f"查询出错：{str(e)}"
        print('result:{},source:{}'.format(answer,category))
    else:
        response=with_message_history.invoke(
        {'question':question},
        config=config
    )
        print('result:{},source:{}'.format(response,category))