【项目】在AUTODL上使用langchain实现《红楼梦》知识图谱和RAG混合检索（三）知识图谱和路由部分
首先在数据集 - 开放知识图谱下载红楼梦的知识图谱，这个网站上有各种各样的知识图谱，可以挑你感兴趣的做( • ̀ω•́ )
这个知识图谱的作者们已经将三元组抽取出来了，我们可以直接用，如果你对三元组是如何生成的感兴趣，可以看看他们之前的步骤。对于这个项目，我们只需要打开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 categoryelif rag_score>0:category='RAG'print(category)return categoryrouter_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))