首先在數據集 - 開放知識圖譜下載紅樓夢的知識圖譜，這個網站上有各種各樣的知識圖譜，可以挑你感興趣的做( ? ?ω?? )
這個知識圖譜的作者們已經將三元組抽取出來了，我們可以直接用，如果你對三元組是如何生成的感興趣，可以看看他們之前的步驟。對于這個項目，我們只需要打開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))
)
