從零到一：大語言模型在知識圖譜構建中的實操指南

?作者|Ninja Geek

來源|神州問學

還沒有看過上篇的讀者可以閱讀《使用大語言模型從零構建知識圖譜（上）》了解整個系列的內容

通過創建一個自定義流程來自動上傳業務數據

在這一節，我會帶你創建一個自定義流程，通過大語言模型自動生成節點定義、關系和?Cypher 查詢，基于數據集進行操作。這種方法也適用于其他 DataFrame，同時該方法也能夠自動識別其 Schema。

需要注意的是，這種方法在性能上會是個問題，尤其是與?Langchain?的 LLMGraphTransformer 相比，我將在下一節中進行介紹。而本節主要幫助你理解如果從零開始構建該過程，從原理出發，幫助你有機會設計自己的 Graph-Builder。實際上，目前所謂最佳方法的主要限制來自于它對數據的天然含義和模式高度敏感。因此，需要跳出固有的思維模式就顯得至關重要，這樣才能夠幫助你從零開始設計 GraphRAG，或利用現有的，最佳實踐的 GraphRAG 來滿足你的業務需求。

現在，讓我們深入研究，設置我們將在接下來的練習中使用的大語言模型。你可以使用 Langchain 所支持的任何大語言模型，只要其性能能夠滿足你真是的業務需要。

這里我們有兩個可選的免費方案：DeepSeek-V3（注冊后可獲得 10 元的額度，有效期一個月）和?Ollama（可以讓你輕松的在本地運行開源模型）。對于這兩種方案我都進行了測試，盡管 DeepSeek-V3 提供了和 GPT-4o 類似的性能，我仍然推薦你選擇 Ollama 進行學習，這樣，你可以更深入的了解從模型下載到運行的整個過程。

在 Ollama 示例中，我們將使用?Qwen2.5-Coder:7B，它針對代碼任務進行了微調，并在代碼生成、推理和修復代碼錯誤方面表現出色。根據你本地計算機的配置來決定是否使用更高參數量的版本，如 14B 或 32B。

讓我們從初始化模型開始：

解釋

llm = OllamaLLM(model="qwen2.5-coder:latest")

讓我們開始提取數據集的結構，并定義節點及其屬性：

解釋

node_structure = "\n".join([f"{col}: {', '.join(map(str, movies[col].unique()[:3]))}..." for col in movies.columns
])
print(node_structure)

對于數據集中的每一列（例如：電影類型、導演），我們來展示一些樣本值。這將幫助大語言模型理解數據格式以及每一列的典型值。

解釋

 Release Year: 1907, 1908, 1909...
Title: Daniel boone, Laughing gas, The adventures of dollie...
Origin/Ethnicity: American...
Director: Wallace mccutcheon and ediwin s. porter, Edwin stanton porter, D. w. griffith...
Cast: William craven, florence lawrence, Bertha regustus, edward boulden, Arthur v. johnson, linda arvidson...
Genre: Biographical, Comedy, Drama...
Plot: Boone's daughter befriends an indian maiden as boone and his companion start out on a hunting expedition. while he is away, boone's cabin is attacked by the indians, who set it on fire and abduct boone's daughter. boone returns, swears vengeance, then heads out on the trail to the indian camp. his daughter escapes but is chased. the indians encounter boone, which sets off a huge fight on the edge of a cliff. a burning arrow gets shot into the indian camp. boone gets tied to the stake and tortured. the burning arrow sets the indian camp on fire, causing panic. boone is rescued by his horse, and boone has a knife fight in which he kills the indian chief.[2], The plot is that of a black woman going to the dentist for a toothache and being given laughing gas. on her way walking home, and in other situations, she can't stop laughing, and everyone she meets "catches" the laughter from her, including a vendor and police officers., On a beautiful summer day a father and mother take their daughter dollie on an outing to the river. the mother refuses to buy a gypsy's wares. the gypsy tries to rob the mother, but the father drives him off. the gypsy returns to the camp and devises a plan. they return and kidnap dollie while her parents are distracted. a rescue crew is organized, but the gypsy takes dollie to his camp. they gag dollie and hide her in a barrel before the rescue party gets to the camp. once they leave the gypsies and escapes in their wagon. as the wagon crosses the river, the barrel falls into the water. still sealed in the barrel, dollie is swept downstream in dangerous currents. a boy who is fishing in the river finds the barrel, and dollie is reunited safely with her parents...

生成節點

接下來，我們使用大語言模型的提示詞模板來引導模型如何提取節點及其屬性。讓我們先看看完整的代碼：

解釋

# 設置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)def validate_node_definition(node_def: Dict) -> bool:"""驗證節點結構定義"""if not isinstance(node_def, dict):return Falsereturn all(isinstance(v, dict) and all(isinstance(k, str) for k in v.keys())for v in node_def.values())@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def get_node_definitions(chain, structure: str, example: Dict) -> Dict[str, Dict[str, str]]:"""使用重試邏輯來獲取節點定義"""try:# 從大語言模型獲得響應response = chain.invoke({"structure": structure, "example": example})# 解析響應node_defs = ast.literal_eval(response)# 驗證結構if not validate_node_definition(node_defs):raise ValueError("無效的節點結構定")return node_defsexcept (ValueError, SyntaxError) as e:logger.error(f"解析節點定義時出錯: {e}")raise# 更新節點定義模板
node_example = {"NodeLabel1": {"property1": "row['property1']", "property2": "row['property2']"},"NodeLabel2": {"property1": "row['property1']", "property2": "row['property2']"},"NodeLabel3": {"property1": "row['property1']", "property2": "row['property2']"},
}define_nodes_prompt = PromptTemplate(input_variables=["example", "structure"],template=("""分析以下數據集結構并提取節點的實體標簽及其屬性。\n節點屬性應基于數據集列和它們的值。\n返回的結果應為一個字典，其中鍵是節點標簽，值是節點屬性。\n\n示例: {example}\n\n數據集結構:\n{structure}\n\n確保包括所有可能的節點標簽及其屬性。\n如果某個屬性可以是其自己的節點，請將其作為單獨的節點標簽。\n請不要使用三重反引號標識代碼塊，只需返回元組的列表。\n僅返回包含節點標簽和屬性的字典，不要包含任何其他文本或引號。"""),
)# 帶有錯誤處理機制的執行過程
try:node_chain = define_nodes_prompt | llmnode_definitions = get_node_definitions(node_chain, structure=node_structure, example=node_example)logger.info(f"節點定義: {node_definitions}")
except Exception as e:logger.error(f"獲取節點定義失敗: {e}")raise

在這個代碼片段中，我們首先使用 logging 庫設置日志記錄， logging 是一個 Python 模塊，用于跟蹤執行過程中的事件（如錯誤或狀態更新）：

解釋

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

我們使用 basicConfig 配置日志記錄，以顯示 INFO 級別或更高的消息，并初始化日志記錄器實例，我將在代碼中用它來記錄消息。

這個步驟其實不是必需的，你也可以用 print 語句來代替它。然而，這是一個良好的工程實踐。

接下來，我將創建一個函數來驗證大語言模型生成的節點：

解釋

def validate_node_definition(node_def: Dict) -> bool:"""驗證節點結構定義"""if not isinstance(node_def, dict):return Falsereturn all(isinstance(v, dict) and all(isinstance(k, str) for k in v.keys())for v in node_def.values())

該函數的輸入是一個字典，其中鍵是節點標簽（例如：Movie），值是屬性的字典（例如：title、year）。

首先，函數檢查 node_def 是否是一個字典，并驗證字典中的每個值是否也是字典，并且這些字典中的所有鍵是否都是字符串。如果結構有效，則返回 True 。

接下來，創建一個函數來調用 LLM 鏈并實際生成節點：

解釋

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def get_node_definitions(chain, structure: str, example: Dict) -> Dict[str, Dict[str, str]]:"""獲取帶有重試邏輯的節點定義"""try:# 從大語言模型獲取響應response = chain.invoke({"structure": structure, "example": example})# 解析響應node_defs = ast.literal_eval(response)# 驗證結構if not validate_node_definition(node_defs):raise ValueError("無效的節點結構定義")return node_defs

如果你不熟悉 Python 中的裝飾器，可能會好奇 @retry(...) 這部分是做什么的，可以將其看作是一個包裝函數，圍繞著實際的 get_node_definitions 函數。在這種情況下，我調用了 retry 裝飾器，如果發生錯誤，它會自動重試該函數。

● stop_after_attempt(3) : 最多重試 3 次。

● wait_exponential : 在重試之間增加延遲的時長（例如：4 秒、8 秒、16 秒等等）。

函數的輸入是：

●chain : LangChain 管道（提示 + LLM）。我會在稍后定義這個管道。

●structure : 數據集結構（列和示例值）。

● example : 用于引導 LLM 的示例節點定義。

接下來，chain.invoke 將結構和示例發送給 LLM，并接收一個字符串作為響應。ast.literal_eval 將字符串響應轉換成 Python 字典。

我使用 validate_node_definition 檢查解析后的字典是否符合正確的格式，如果結構無效，它會引發 ValueError 。

解釋

except (ValueError, SyntaxError) as e:logger.error(f"Error parsing node definitions: {e}")raise

如果響應無法解析或驗證，會記錄錯誤信息，該函數會拋出異常。

接下來，我們為 LLM 提供一個提示詞模板，以引導其完成節點生成任務：

解釋

 define_nodes_prompt = PromptTemplate(input_variables=["example", "structure"],template=("""分析以下數據集結構并提取節點的實體標簽及其屬性。\n節點屬性應基于數據集列和它們的值。\n返回的結果應為一個字典，其中鍵是節點標簽，值是節點屬性。\n\n示例: {example}\n\n數據集結構:\n{structure}\n\n確保包括所有可能的節點標簽及其屬性。\n如果某個屬性可以是其自己的節點，請將其作為單獨的節點標簽。\n請不要使用三重反引號標識代碼塊，只需返回元組的列表。\n僅返回包含節點標簽和屬性的字典，不要包含任何其他文本或引號。"""),
)

請注意，我提供了本節開始時定義的節點結構，以及如何生成節點字典的示例：

解釋

node_example = {"NodeLabel1": {"property1": "row['property1']", "property2": "row['property2']"},"NodeLabel2": {"property1": "row['property1']", "property2": "row['property2']"},"NodeLabel3": {"property1": "row['property1']", "property2": "row['property2']"},
}

在示例中，鍵是節點標簽（例如：Movie、Director），值是映射到數據集列的屬性字典（例如：row[’property1’] ）。

接下來，讓我們執行鏈：

解釋

try:node_chain = define_nodes_prompt | llmnode_definitions = get_node_definitions(node_chain, structure=node_structure, example=node_example)logger.info(f"節點定義: {node_definitions}")
except Exception as e:logger.error(f"獲取節點定義失敗: {e}")raise

在 LangChain 中，我們使用結構化提示詞 | LLM | … 來創建一個鏈，將提示詞模板與 LLM 結合，形成一個管道。我們使用 get_node_definitions 來獲取并驗證節點定義。

如果過程中出現失敗，錯誤會被記錄，并且程序會引發異常。

如果過程成功，它將生成類似于以下內容的結果：

解釋

 
INFO:__main__:Node Definitions: {'Movie': {'Release Year': "row['Release Year']", 'Title': "row['Title']"}, 'Director': {'Name': "row['Director']"}, 'Cast': {'Actor': "row['Cast']"}, 'Genre': {'Type': "row['Genre']"}, 'Plot': {'Description': "row['Plot']"}}

生成關系

一旦節點被定義，我們就可以識別它們之間的關系。接下來，我們來看看完整的代碼是怎樣的：

解釋

class RelationshipIdentifier:"""識別圖數據庫中節點之間的關系。"""RELATIONSHIP_EXAMPLE = [("NodeLabel1", "RelationshipLabel", "NodeLabel2"),("NodeLabel1", "RelationshipLabel", "NodeLabel3"),("NodeLabel2", "RelationshipLabel", "NodeLabel3"),]PROMPT_TEMPLATE = PromptTemplate(input_variables=["structure", "node_definitions", "example"],template="""考慮以下數據集結構：\n{structure}\n\n考慮以下節點定義：\n{node_definitions}\n\n根據數據集結構和節點定義，識別節點之間的關系（邊）。\n以三元組的形式返回關系，其中每個三元組包含起始節點標簽、關系標簽和結束節點標簽，每個三元組是一個元組。\n請僅返回元組列表。請不要使用三重反引號標識代碼塊，只返回元組列表。\n\n示例:\n{example}"""
)def __init__(self, llm: Any, logger: logging.Logger = None):self.llm = llmself.logger = logger or logging.getLogger(__name__)self.chain = self.PROMPT_TEMPLATE | self.llmdef validate_relationships(self, relationships: List[Tuple]) -> bool:"""驗證關系結構"""return all(isinstance(rel, tuple) and len(rel) == 3 and all(isinstance(x, str) for x in rel)for rel in relationships)@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))def identify_relationships(self, structure: str, node_definitions: Dict) -> List[Tuple]:"""識別關系并應用重試邏輯"""try:response = self.chain.invoke({"structure": structure, "node_definitions": str(node_definitions), "example": str(self.RELATIONSHIP_EXAMPLE)})relationships = ast.literal_eval(response)if not self.validate_relationships(relationships):raise ValueError("無效的關系結構")self.logger.info(f"已驗證 {len(relationships)} 個關系")return relationshipsexcept Exception as e:self.logger.error(f"驗證關系時出現錯誤：{e}")raisedef get_relationship_types(self) -> List[str]:"""提取唯一的關系類型。"""return list(set(rel[1] for rel in self.identify_relationships()))# 用法
identifier = RelationshipIdentifier(llm=llm)
relationships = identifier.identify_relationships(node_structure, node_definitions)
print("關系:", relationships)

由于這段代碼需要進行比節點生成更多的操作，我們將代碼組織在一個類中 —— RelationshipIdentifier —— 以封裝所有關系提取、驗證和日志記錄的邏輯。我們使用類似的邏輯，因此我們提供一個關系示例：

解釋

 
RELATIONSHIP_EXAMPLE = [("NodeLabel1", "RelationshipLabel", "NodeLabel2"),("NodeLabel1", "RelationshipLabel", "NodeLabel3"),("NodeLabel2", "RelationshipLabel", "NodeLabel3"),
]

在這里，每個關系都是一個元組，包含以下內容：

●起始節點標簽：源節點的標簽（例如：Movie）。

●關系標簽：連接類型（例如：DIRECTED_BY）。

●結束節點標簽：目標節點的標簽（例如：Director）。

接下來，我們定義實際的提示詞模板：

解釋

PROMPT_TEMPLATE = PromptTemplate(input_variables=["structure", "node_definitions", "example"],template="""考慮以下數據集結構：\n{structure}\n\n考慮以下節點定義：\n{node_definitions}\n\n根據數據集結構和節點定義，識別節點之間的關系（邊）。\n以三元組的形式返回關系，其中每個三元組包含起始節點標簽、關系標簽和結束節點標簽，每個三元組是一個元組。\n請僅返回元組列表。請不要使用三重反引號標識代碼塊，只返回元組列表。\n\n示例:\n{example}"""
)

在這種情況下，我們有三個輸入變量：

●structure：數據集結構，列出了列和示例值。我在本節開始時定義了它。

● node_definitions ：節點標簽及其屬性的字典。這些節點是在上一節中由 LLM 生成的。

● example ：三元組格式的示例關系。

接下來，我將使用者三個屬性初始化類：

解釋

def __init__(self, llm: Any, logger: logging.Logger = None):self.llm = llmself.logger = logger or logging.getLogger(__name__)self.chain = self.PROMPT_TEMPLATE | self.llm

●llm ：用于處理提示的語言模型（例如：GPT-4o-mini）。

● logger ：可選參數，用于記錄進度和錯誤（如果未提供，則默認為標準日志記錄器）。

●self.chain ：將提示詞模板與 LLM 結合，創建一個可重用的管道。

類似之前的做法，我們創建一個方法來驗證生成的關系：

解釋

def validate_relationships(self, relationships: List[Tuple]) -> bool:"""驗證關系結構。"""return all(isinstance(rel, tuple) and len(rel) == 3 and all(isinstance(x, str) for x in rel)for rel in relationships)

該方法檢查每個項目是否是元組，確保每個元組包含三個元素，并且所有元素都是字符串（例如：節點標簽或關系類型）。最后，如果滿足這些條件，則返回 TRUE ，否則返回 FALSE 。

接下來，我們創建一個方法來調用鏈并生成關系：

解釋

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def identify_relationships(self, structure: str, node_definitions: Dict) -> List[Tuple]:"""識別關系并應用重試邏輯。"""try:response = self.chain.invoke({"structure": structure, "node_definitions": str(node_definitions), "example": str(self.RELATIONSHIP_EXAMPLE)})relationships = ast.literal_eval(response)if not self.validate_relationships(relationships):raise ValueError("無效的關系結構")self.logger.info(f"已驗證 {len(relationships)} 個關系")return relationships

我們再次使用 retry 裝飾器來在失敗時重新嘗試 LLM 鏈，并以類似于節點生成時的方式調用鏈。

此外，我們使用 ast.literal_eval 將 LLM 的字符串輸出轉換成 Python 列表，并使用 validate_relationships 來確保輸出格式正確。

解釋

except Exception as e:self.logger.error(f"Error identifying relationships: {e}")raise

如果該方法失敗，它會記錄錯誤并最多重試 3 次。

最后一個方法返回唯一的關系標簽（例如：DIRECTED_BY、ACTED_IN）：

解釋

 
def get_relationship_types(self) -> List[str]:"""Extract unique relationship types."""return list(set(rel[1] for rel in self.identify_relationships()))

它調用 identify_relationships 方法來獲取關系列表。然后，它提取每個元組中的第二個元素（關系標簽），使用 set 來去除重復項，并將結果轉換回列表。

現在，終于到了生成關系的時候了：

解釋

identifier = RelationshipIdentifier(llm=llm)
relationships = identifier.identify_relationships(node_structure, node_definitions)
print("Relationships:", relationships)

如果 LLM 在 3 次嘗試內成功，它將返回一個類似以下內容的關系列表，以元組格式表示：

解釋

INFO:__main__:Identified 4 relationships
Relationships: [('Movie', 'Directed By', 'Director'), ('Movie', 'Starring', 'Cast'), ('Movie', 'Has Genre', 'Genre'), ('Movie', 'Contains Plot', 'Plot')]

生成 Cypher 查詢

在節點和關系定義完成后，我創建了 Cypher 查詢將它們加載到 Neo4j 中。這個過程遵循與節點生成和關系生成類似的邏輯。然而，我們增加了幾個額外的步驟來進行驗證，因為生成的輸出將用于將數據加載到我們的知識圖譜中。因此，我們需要盡可能提高成功的概率。讓我們首先看看完整的代碼：

解釋

class CypherQueryBuilder:"""構建用于 Neo4j 圖數據庫的 Cypher 查詢。"""INPUT_EXAMPLE = """NodeLabel1: value1, value2NodeLabel2: value1, value2"""EXAMPLE_CYPHER = example_cypher = """CREATE (n1:NodeLabel1 {property1: "row['property1']", property2: "row['property2']"})CREATE (n2:NodeLabel2 {property1: "row['property1']", property2: "row['property2']"})CREATE (n1)-[:RelationshipLabel]->(n2);"""PROMPT_TEMPLATE = PromptTemplate(input_variables=["structure", "node_definitions", "relationships", "example"],template="""考慮以下節點定義：\n{node_definitions}\n\n考慮以下關系：\n{relationships}\n\n生成 Cypher 查詢以創建節點和關系，使用下面的節點定義和關系。記得用數據集中的實際數據替換占位符值。\n包括每個節點的所有屬性，按照節點定義，并創建關系。\n返回一個包含每個查詢用分號分隔的單個字符串。\n請不要在響應中包含任何其他文本或引號。\n請僅返回包含 Cypher 查詢的字符串。請不要使用三重反引號標識代碼塊。\n\n示例輸入：\n{input}\n\n示例輸出Cypher查詢：\n{cypher}"""
)def __init__(self, llm: Any, logger: logging.Logger = None):self.llm = llmself.logger = logger or logging.getLogger(__name__)# self.chain = LLMChain(llm=llm, prompt=self.PROMPT_TEMPLATE)self.chain = self.PROMPT_TEMPLATE | self.llmdef validate_cypher_query(self, query: str) -> bool:"""使用 LLM 和正則表達式模式驗證 Cypher 查詢語法。"""VALIDATION_PROMPT = PromptTemplate(input_variables=["query"],template="""驗證此Cypher查詢并返回 TRUE 或 FALSE：查詢: {query}檢查規則：1. 有效的 CREATE 語句2. 正確的屬性格式3. 有效的關系語法4. 無缺失的括號5. 有效的屬性名稱6. 有效的關系類型如果查詢有效，返回 TRUE；如果無效，返回 FALSE。""")try:# 基本模式驗證basic_valid = all(re.search(pattern, query) for pattern in [r'CREATE \(',  r'\{.*?\}',    r'\)-\[:.*?\]->'])if not basic_valid:return False#  LLM 驗證validation_chain = VALIDATION_PROMPT | self.llmresult = validation_chain.invoke({"query": query})# 解析結果is_valid = "TRUE" in result.upper()if not is_valid:self.logger.warning(f"LLM 驗證查詢失敗: {query}")return is_validexcept Exception as e:self.logger.error(f"驗證錯誤: {e}")return Falsedef sanitize_query(self, query: str) -> str:"""清理并格式化 Cypher 查詢"""return (query.strip().replace('\n', ' ').replace('  ', ' ').replace("'row[", "row['").replace("]'", "']"))@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))def build_queries(self, node_definitions: Dict, relationships: List) -> str:"""構建帶有重試邏輯的 Cypher 查詢。"""try:response = self.chain.invoke({"node_definitions": str(node_definitions),"relationships": str(relationships),"input": self.INPUT_EXAMPLE,"cypher": self.EXAMPLE_CYPHER})# 獲取位于三重反引號內的響應。if '```' in response:response = response.split('```')[1]# 清理響應queries = self.sanitize_query(response)# 驗證查詢if not self.validate_cypher_query(queries):raise ValueError("無效的 Cypher 查詢語法")self.logger.info("成功生成 Cypher 查詢")return queriesexcept Exception as e:self.logger.error(f"構建 Cypher 查詢出錯: {e}")raisedef split_queries(self, queries: str) -> List[str]:"""將組合的查詢拆分為單獨的語句。"""return [q.strip() for q in queries.split(';') if q.strip()]# 用法
builder = CypherQueryBuilder(llm=llm)
cypher_queries = builder.build_queries(node_definitions, relationships)
print("Cypher 查詢:", cypher_queries)

我們提供一個提示詞模板來幫助 LLM：

解釋

PROMPT_TEMPLATE = PromptTemplate(input_variables=["structure", "node_definitions", "relationships", "example"],template="""考慮以下節點定義：\n{node_definitions}\n\n考慮以下關系：\n{relationships}\n\n生成 Cypher 查詢以創建節點和關系，使用下面的節點定義和關系。記得用數據集中的實際數據替換占位符值。\n包括每個節點的所有屬性，按照節點定義，并創建關系。\n返回一個包含每個查詢用分號分隔的單個字符串。\n請不要在響應中包含任何其他文本或引號。\n請僅返回包含 Cypher 查詢的字符串。請不要使用三重反引號標識代碼塊。\n\n示例輸入：\n{input}\n\n示例輸出Cypher查詢：\n{cypher}"""
)

現在，我提供了四個變量給提示詞模板：

● structure ：數據集結構，作為上下文。

●node_definitions ：生成的節點及其屬性。

●relationships ：節點之間生成的關系。

●example ： 用于格式參考的示例查詢。

解釋

 def __init__(self, llm: Any, logger: logging.Logger = None):self.llm = llmself.logger = logger or logging.getLogger(__name__)self.chain = self.PROMPT_TEMPLATE | self.llm

我們以與關系類相同的方式初始化該類。

接下來，我定義了一個驗證方法來檢查生成的輸出：

解釋

 def validate_cypher_query(self, query: str) -> bool:"""使用 LLM 和正則表達式模式驗證 Cypher 查詢語法。"""VALIDATION_PROMPT = PromptTemplate(input_variables=["query"],template="""驗證此Cypher查詢并返回 TRUE 或 FALSE：查詢: {query}檢查規則：1. 有效的 CREATE 語句2. 正確的屬性格式3. 有效的關系語法4. 無缺失的括號5. 有效的屬性名稱6. 有效的關系類型如果查詢有效，返回 TRUE；如果無效，返回 FALSE。""")try:# 基本模式驗證basic_valid = all(re.search(pattern, query) for pattern in [r'CREATE \(',  r'\{.*?\}',    r'\)-\[:.*?\]->'])if not basic_valid:return False# LLM 驗證validation_chain = VALIDATION_PROMPT | self.llmresult = validation_chain.invoke({"query": query})# 解析結果is_valid = "TRUE" in result.upper()if not is_valid:self.logger.warning(f"LLM 驗證查詢失敗: {query}")return is_validexcept Exception as e:self.logger.error(f"驗證錯誤：{e}")return False

該方法執行兩個驗證步驟。首先是使用正則表達式進行基本驗證：

解釋

 basic_valid = all(re.search(pattern, query) for pattern in [r'CREATE \(',  r'\{.*?\}',    r'\)-\[:.*?\]->'
])
if not basic_valid:return False

這確保查詢包含必要的 Cypher 語法：

●CREATE ：確保節點和關系正在被創建。

● {.*?} ：確保包含屬性。

● -: .*?→ ：確保關系格式正確。

然后，它使用 LLM 執行高級驗證：

解釋

validation_chain = VALIDATION_PROMPT | self.llm
result = validation_chain.invoke({"query": query})
is_valid = "TRUE" in result.upper()

驗證在提示中指定，我要求 LLM 確保以下幾點：

1. 有效的 CREATE 語句

2. 正確的屬性格式

3. 有效的關系語法

4. 無缺失的括號

5.有效的屬性名稱

6.有效的關系類型

到目前為止一切看上去工作的都還不錯，這里，讓我再添加一個方法，進一步清理生成的輸出：

解釋

 def sanitize_query(self, query: str) -> str:"""清理并格式化 Cypher 查詢。"""return (query.strip().replace('\n', ' ').replace('  ', ' ').replace("'row[", "row['").replace("]'", "']"))

我將移除不必要的空格以及換行符（\n），并修復與數據集引用相關的潛在格式問題（例如：row[’property1’]）。

請根據你所使用的大語言模型考慮更新此方法，較小參數量的模型可能需要更多的數據清理操作。

接下來，我來定義一個查詢調用方法：

解釋

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))def build_queries(self, node_definitions: Dict, relationships: List) -> str:"""構建帶有重試邏輯的 Cypher 查詢。"""try:response = self.chain.invoke({"node_definitions": str(node_definitions),"relationships": str(relationships),"input": self.INPUT_EXAMPLE,"cypher": self.EXAMPLE_CYPHER})# 獲取位于三重反引號內的響應if '```' in response:response = response.split('```')[1]# 清理響應queries = self.sanitize_query(response)# 驗證查詢if not self.validate_cypher_query(queries):raise ValueError("無效的 Cypher 查詢語法")self.logger.info("成功生成 Cypher 查詢")return queriesexcept Exception as e:self.logger.error(f"構建 Cypher 查詢時出錯: {e}")raise

這個方法與關系構建器類中的方法類似，唯一的不同之處是：

解釋

 if '```' in response:response = response.split('```')[1]

在這里，LLM 可能會提供額外的 Markdown 格式來指定它是一個代碼塊。如果 LLM 的響應中存在這種格式，我只會提取三重反引號內的代碼。

接下來，我定義一個方法，將單一的 Cypher 查詢字符串拆分成單獨的語句：

解釋

 def split_queries(self, queries: str) -> List[str]:"""將組合的查詢拆分為單獨的語句"""return [q.strip() for q in queries.split(';') if q.strip()]

例如，以下 Cypher 查詢：

解釋

 CREATE (n1:Movie {title: "Inception"}); CREATE (n2:Director {name: "Nolan"});

這將轉換成以下形式：

解釋

["CREATE (n1:Movie {title: 'Inception'})", "CREATE (n2:Director {name: 'Nolan'})"]

這將非常有用，因為可以遍歷查詢列表。

最后，初始化類并生成 Cypher 查詢：

解釋

 builder = CypherQueryBuilder(llm=llm)
cypher_queries = builder.build_queries(node_definitions, relationships)
print("Cypher 查詢:", cypher_queries)

成功時，輸出將如下所示：

解釋

 
INFO:__main__:Successfully generated Cypher queries
Cypher Queries: CREATE (m:Movie {Release_Year: "row['Release Year']", Title: "row['Title']"}) CREATE (d:Director {Name: "row['Director']"}) CREATE (c:Cast {Actor: "row['Cast']"}) CREATE (g:Genre {Type: "row['Genre']"}) CREATE (p:Plot {Description: "row['Plot']"}) CREATE (m)-[:Directed_By]->(d) CREATE (m)-[:Starring]->(c) CREATE (m)-[:Has_Genre]->(g) CREATE (m)-[:Contains_Plot]->(p)

最后，遍歷數據集，并為每一行執行生成的 Cypher 查詢。

解釋

logs = ""
total_rows = len(df)def sanitize_value(value):if isinstance(value, str):return value.replace('"', '')return str(value)for index, row in tqdm(df.iterrows(), total=total_rows,desc="正在加載數據到 Neo4j",position=0,leave=True):# 將占位符替換為實際的值cypher_query = cypher_queriesfor column in df.columns:cypher_query = cypher_query.replace(f"row['{column}']", f'{sanitize_value(row[column])}')try:# 執行查詢并更新進度conn.execute_query(cypher_query)except Exception as e:logs += f"在行 {index+1}: {str(e)} 出現錯誤\n"

請注意，我定義了一個空字符串變量 logs ，用于捕獲潛在的失敗。我還添加了一個清理函數，用于傳遞給每個行輸入的值：

解釋

def sanitize_value(value):if isinstance(value, str):return value.replace('"', '')return str(value)

將防止包含雙引號的字符串破壞查詢語法。

接下來，我們來遍歷數據集：

解釋

for index, row in tqdm(df.iterrows(), total=total_rows,desc="正在加載數據到 Neo4j",position=0,leave=True):# 將占位符替換為實際的值cypher_query = cypher_queriesfor column in df.columns:cypher_query = cypher_query.replace(f"row['{column}']", f'{sanitize_value(row[column])}')try:# 執行查詢并更新進度conn.execute_query(cypher_query)except Exception as e:logs += f"在行 {index+1}: {str(e)} 出現錯誤\n"

正如我在練習開始時提到的，我使用 tqdm 為進度條添加了一個漂亮的外觀，以可視化的方式顯示處理了多少行數據。我傳遞了 df.iterrows() 來遍歷 DataFrame，提供索引和行數據。total=total_rows 由 tqdm 用于計算進度。添加 desc=”正在加載數據到 Neo4j” 來為進度條提供標簽。最后，position=0, leave=True 確保進度條在控制臺中保持可見。

接下來，我將像 row[’column_name’] 這樣的占位符替換成實際的數據集值，將每個值傳遞給 sanitize_value 函數，并執行查詢。

讓我們檢查一下數據集是否已上傳。切換到 Neo4j，并運行以下 Cypher 查詢：

解釋

 
MATCH p=(m:Movie)-[r]-(n)
RETURN p
LIMIT 100;

在我的機器上，LLM 生成了以下圖表：

這與我們手動上傳的知識圖譜非常相似。對于一個簡單的 LLM 來說，這還不賴對吧。雖然這需要相當多的編碼工作，但我們現在可以將其重用于多個數據集，更重要的是，可以將其作為基礎，創建更復雜的 LLM 圖形構建器。

在我提供的示例中，還沒有通過提供實體、關系和屬性來幫助 LLM。然而，考慮將它們作為示例來提高 LLM 的性能。此外，更現代化的方法利用思維鏈來提出額外的節點和關系。這使得模型能夠順序推理并進一步改進結果。另一種策略是提供行樣本，以更好的適應每行中提供的值。

在本系列文章的最后一篇，也就是下篇中，你將看到使用 LangChain 實現的現代化 GraphRAG。