寫給小白看的使用LangChain構(gòu)建基于知識圖譜的RAG系統(tǒng)實(shí)戰(zhàn)教程
在構(gòu)建基于知識圖譜的RAG系統(tǒng)或使用LangChain的智能體時，最大的挑戰(zhàn)之一是從非結(jié)構(gòu)化數(shù)據(jù)中準(zhǔn)確提取節(jié)點(diǎn)和關(guān)系。特別是當(dāng)使用較小的、量化的本地LLM時，這一點(diǎn)尤其困難，結(jié)果往往是AI系統(tǒng)表現(xiàn)不佳。
LangChain提取功能的一個關(guān)鍵問題是它依賴嚴(yán)格的JSON解析，即使使用更大的模型或非常詳細(xì)的提示模板，也可能失敗。相比之下，BAML使用一種模糊解析（fuzzy parsing）方法，即使LLM的輸出不是完美的JSON格式，也能成功提取數(shù)據(jù)。
在這篇博客中，我們將探討在使用較小的量化模型時LangChain提取的局限性，并展示BAML如何將提取成功率從大約25%提升到超過99%。
所有代碼都可以在這個GitHub倉庫中找到：??https://github.com/FareedKhan-dev/langchain-graphrag-baml??
目錄
? 初始化評估數(shù)據(jù)集
? 量化的小型LLaMA模型
? 基于LLMGraphTransformer的方法
? 理解LangChain的問題
? 改進(jìn)提示能解決問題嗎？
? BAML的初始化和快速概覽
? 將BAML與LangChain集成
? 運(yùn)行BAML實(shí)驗(yàn)
? 使用Neo4j分析GraphRAG
? 查找和鏈接相似實(shí)體
? 使用Leiden算法進(jìn)行社區(qū)檢測
? 分析最終圖譜結(jié)構(gòu)
? 結(jié)論
初始化評估數(shù)據(jù)集
為了理解問題及其解決方案，我們需要一個評估數(shù)據(jù)集來進(jìn)行多次測試，以了解BAML如何改進(jìn)LangChain知識圖譜。
我們將使用Tomasonjo的博客數(shù)據(jù)集，托管在GitHub上，先加載這些數(shù)據(jù)。
# 導(dǎo)入pandas庫用于數(shù)據(jù)操作和分析
import pandas as pd
# 從GitHub上的CSV文件加載新聞文章數(shù)據(jù)集到pandas DataFrame
news = pd.read_csv(
"https://raw.githubusercontent.com/tomasonjo/blog-datasets/main/news_articles.csv"
)
# 顯示DataFrame的前5行
news.head()
我們的DataFrame很簡單（包含標(biāo)題和文本，文本是新聞的描述）。我們還需要一列來存儲每篇新聞文章文本對應(yīng)的總token數(shù)。
為此，我們可以使用OpenAI的tiktoken庫來計(jì)算token，方法很簡單，用循環(huán)來處理數(shù)據(jù)集。
# 導(dǎo)入tiktoken庫來計(jì)算文本的token數(shù)
import tiktoken
# 定義一個函數(shù)，計(jì)算給定字符串在指定模型中的token數(shù)
defnum_tokens_from_string(string: str, model: str = "gpt-4o") -> int:
"""返回文本字符串中的token數(shù)。"""
# 獲取指定模型的編碼
encoding = tiktoken.encoding_for_model(model)
# 將字符串編碼為token并計(jì)數(shù)
num_tokens = len(encoding.encode(string))
# 返回總token數(shù)
return num_tokens
# 在DataFrame中創(chuàng)建新列'tokens'
# 計(jì)算每篇文章標(biāo)題和文本組合的token數(shù)
news["tokens"] = [
num_tokens_from_string(f"{row['title']} {row['text']}")
for i, row in news.iterrows()
]計(jì)算DataFrame的token只需要幾秒鐘。
這是更新后的DataFrame。
# 顯示DataFrame的前5行，展示新的'tokens'列
news.head()
這些token將在后續(xù)的評估和分析階段使用，因此我們進(jìn)行了這一步。
量化的小型LLaMA模型
為了將數(shù)據(jù)轉(zhuǎn)換為知識圖譜，我們將使用一個低級別量化的模型來進(jìn)行嚴(yán)格的測試。
在生產(chǎn)環(huán)境中，開源LLM通常以量化形式部署，以降低成本和延遲。本博客使用LLaMA 3.1。
這里選擇Ollama作為平臺，但LangChain支持多種API和本地LLM提供商，可以選擇任何合適的選項(xiàng)。
# ChatOllama是Ollama語言模型的接口
from langchain_ollama import ChatOllama
# 定義要使用的模型名稱
model = "llama3"
# 初始化ChatOllama語言模型
# 'temperature'參數(shù)控制輸出的隨機(jī)性
# 低值（如0.001）使模型的響應(yīng)更確定
llm = ChatOllama(model=model, temperature=0.001)你還需要在系統(tǒng)上安裝Ollama，它支持macOS、Windows和Linux。
訪問Ollama官方網(wǎng)站：https://ollama.com/下載適合你操作系統(tǒng)的安裝程序并按照說明安裝。安裝后，Ollama會作為后臺服務(wù)運(yùn)行。
在macOS和Windows上，應(yīng)用程序應(yīng)自動啟動并在后臺運(yùn)行（你可能在菜單欄或系統(tǒng)托盤中看到一個圖標(biāo)）。在Linux上，你可能需要用??systemctl start ollama??手動啟動。
要檢查服務(wù)是否運(yùn)行，打開終端或命令提示符并輸入：
# 檢查可用模型
ollama list輸出
[ ] <-- No models如果服務(wù)在運(yùn)行但沒有模型，你會看到一個空的模型列表，這在這個階段是正常的。如果出現(xiàn)“command not found”錯誤，確保Ollama已正確安裝。如果出現(xiàn)連接錯誤，說明服務(wù)器未運(yùn)行。
你可以通過pull命令簡單下載llama3模型。這需要一些時間和幾GB的磁盤空間，因?yàn)槟Ｐ秃艽蟆?/p>
# 下載llama3模型
ollama pull llama3這些命令完成后，再次運(yùn)行??ollama list??，你應(yīng)該能看到模型已列出。
# 向本地Ollama API發(fā)送請求以生成文本
curl http://localhost:11434/api/generate \
# 設(shè)置Content-Type頭以指示JSON負(fù)載
-H "Content-Type: application/json" \
# 提供請求數(shù)據(jù)
-d '{
"model": "llama3",
"prompt": "Why is the sky blue?"
}'輸出
{
"model": "llama3",
"created_at": "2025-08-03T12:00:00Z",
"response": "The sky appears blue be ... blue.",
"done": true
}如果成功，你會在終端看到一串JSON響應(yīng)，確認(rèn)服務(wù)器正在運(yùn)行并能提供模型服務(wù)。
現(xiàn)在評估數(shù)據(jù)和LLM都準(zhǔn)備好了，下一步是將數(shù)據(jù)轉(zhuǎn)換以更好地理解LangChain中的問題。
基于LLMGraphTransformer的方法
使用LangChain或LangGraph將原始或結(jié)構(gòu)化數(shù)據(jù)轉(zhuǎn)換為知識圖譜的正確方法是使用它們提供的方法。最常見的方法之一是langchain_experimental庫中的LLMGraphTransformer。
這個工具設(shè)計(jì)為一體化的解決方案：提供文本和LLM，它會處理提示和解析，返回圖譜結(jié)構(gòu)。
讓我們看看它與本地llama3模型的表現(xiàn)如何。
首先，我們需要導(dǎo)入所有必要的組件。
# 從LangChain的實(shí)驗(yàn)庫中導(dǎo)入主要的圖譜轉(zhuǎn)換器
from langchain_experimental.graph_transformers import LLMGraphTransformer
# 導(dǎo)入圖譜和文檔的數(shù)據(jù)結(jié)構(gòu)
from langchain_community.graphs.graph_document import GraphDocument, Node, Relationship
from langchain_core.documents import Document現(xiàn)在，初始化轉(zhuǎn)換器。我們將使用之前創(chuàng)建的llm對象（即llama3模型）。
我們還需要告訴轉(zhuǎn)換器我們希望為節(jié)點(diǎn)和關(guān)系提取哪些額外信息或“屬性”。在這個例子中，我們只要求描述。
# 使用llama3模型初始化LLMGraphTransformer
# 指定我們希望節(jié)點(diǎn)和關(guān)系都有'description'屬性
llm_transformer = LLMGraphTransformer(
llm=llm,
node_properties=["description"],
relationship_properties=["description"]
)為了讓流程可重復(fù)且整潔，我們將創(chuàng)建一個簡單的輔助函數(shù)。這個函數(shù)將接受一個文本字符串，將其包裝成LangChain的Document格式，然后傳遞給llm_transformer以獲取圖譜結(jié)構(gòu)。
# 導(dǎo)入List類型用于類型提示
from typing import List
# 定義一個函數(shù)，處理單個文本字符串并將其轉(zhuǎn)換為圖譜文檔
def process_text(text: str) -> List[GraphDocument]:
# 從原始文本創(chuàng)建LangChain Document對象
doc = Document(page_cnotallow=text)
# 使用轉(zhuǎn)換器將文檔轉(zhuǎn)換為圖譜文檔列表
return llm_transformer.convert_to_graph_documents([doc])一切設(shè)置好后，是時候運(yùn)行實(shí)驗(yàn)了。為了保持可管理性并突出核心問題，我們將處理數(shù)據(jù)集中的20篇文章樣本。
我們將使用ThreadPoolExecutor并行運(yùn)行處理，以加快工作流程。
# 導(dǎo)入并發(fā)處理和進(jìn)度條的庫
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
# 設(shè)置并行工作者的數(shù)量和要處理的文章數(shù)量
MAX_WORKERS = 10
NUM_ARTICLES = 20
# 這個列表將存儲生成的圖譜文檔
graph_documents = []
# 使用ThreadPoolExecutor并行處理文章
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
# 為樣本中的每篇文章提交處理任務(wù)
futures = [
executor.submit(process_text, f"{row['title']} {row['text']}")
for i, row in news.head(NUM_ARTICLES).iterrows()
]
# 每當(dāng)任務(wù)完成時，獲取結(jié)果并添加到列表中
for future in tqdm(
as_completed(futures), total=len(futures), desc="處理文檔"
):
graph_document = future.result()
graph_documents.extend(graph_document)運(yùn)行代碼后，進(jìn)度條顯示所有20篇文章都已處理。
輸出
處理文檔: 100%|██████████| 20/20 [01:32<00:00, 4.64s/it]理解LangChain的問題
那么，我們得到了什么？讓我們檢查graph_documents列表。
# 顯示圖譜文檔列表
print(graph_documents)這是我們得到的輸出：
輸出
[GraphDocument(nodes=[], relatinotallow=[], source=Document(metadata={}, page_cnotallow='XPeng Stock Rises...')),
GraphDocument(nodes=[], relatinotallow=[], source=Document(metadata={}, page_cnotallow='Ryanair sacks chief pilot...')),
GraphDocument(nodes=[], relatinotallow=[], source=Document(metadata={}, page_cnotallow='Dáil almost suspended...')),
GraphDocument(nodes=[Node(id='Jude Bellingham', type='Person', properties={}), Node(id='Real Madrid', type='Organization', properties={})], relatinotallow=[], source=Document(metadata={}, page_cnotallow='Arsenal have Rice bid rejected...')),
...
]立刻就能看出問題。許多GraphDocument對象的節(jié)點(diǎn)和關(guān)系列表是空的。
這意味著對于這些文章，LLM要么生成了LangChain無法解析成有效圖譜結(jié)構(gòu)的輸出，要么完全無法提取任何實(shí)體。
這就是使用較小的量化LLM進(jìn)行結(jié)構(gòu)化數(shù)據(jù)提取的核心挑戰(zhàn)。它們往往難以遵循像LLMGraphTransformer這樣的工具所期望的嚴(yán)格JSON格式。如果有一個小小的錯誤——比如多余的逗號、缺少引號——解析就會失敗，我們什么也得不到。
讓我們量化這個失敗率。我們將統(tǒng)計(jì)20篇文檔中有多少篇生成了空的圖譜。
# 初始化一個計(jì)數(shù)器，用于統(tǒng)計(jì)沒有節(jié)點(diǎn)的文檔
empty_count = 0
# 遍歷生成的圖譜文檔
for doc in graph_documents:
# 如果'nodes'列表為空，計(jì)數(shù)器加1
if not doc.nodes:
empty_count += 1現(xiàn)在，計(jì)算失敗的百分比。
# 計(jì)算并打印失敗生成節(jié)點(diǎn)的文檔百分比
print(f"Percentage missing: {empty_count/len(graph_documents)*100}")輸出
Percentage missing: 75.075%的失敗率。這太糟糕了。這意味著在我們的20篇文章樣本中，只有5篇成功轉(zhuǎn)換成了知識圖譜。
25%的成功率對于任何生產(chǎn)系統(tǒng)來說都是不可接受的。
這就是問題的所在，而且這是一個常見問題。標(biāo)準(zhǔn)方法對于較小LLM略顯不可預(yù)測的特性來說過于嚴(yán)格。
改進(jìn)提示能解決問題嗎？
75%的失敗率是個大問題。作為開發(fā)者，當(dāng)LLM表現(xiàn)不佳時，我們的第一反應(yīng)往往是調(diào)整提示。更好的指令應(yīng)該帶來更好的結(jié)果，對吧？LLMGraphTransformer內(nèi)部使用默認(rèn)提示，但我們無法輕易修改它。
所以，我們用LangChain的ChatPromptTemplate構(gòu)建自己的簡單鏈。這讓我們可以完全控制發(fā)送給llama3的指令。我們可以更明確地“引導(dǎo)”模型每次生成正確的JSON格式。
我們先用Pydantic模型定義我們想要的輸出結(jié)構(gòu)。這是LangChain中結(jié)構(gòu)化輸出的常見模式。
# 導(dǎo)入Pydantic模型以定義數(shù)據(jù)結(jié)構(gòu)
from langchain_core.pydantic_v1 import BaseModel, Field
# 定義一個簡單的節(jié)點(diǎn)結(jié)構(gòu)
classNode(BaseModel):
id: str = Field(descriptinotallow="節(jié)點(diǎn)的唯一標(biāo)識符。")
type: str = Field(descriptinotallow="節(jié)點(diǎn)類型（例如，Person, Organization）。")
# 定義一個簡單的關(guān)系結(jié)構(gòu)
classRelationship(BaseModel):
source: Node = Field(descriptinotallow="關(guān)系的源節(jié)點(diǎn)。")
target: Node = Field(descriptinotallow="關(guān)系的目標(biāo)節(jié)點(diǎn)。")
type: str = Field(descriptinotallow="關(guān)系類型（例如，WORKS_FOR）。")
# 定義整體圖譜結(jié)構(gòu)
classKnowledgeGraph(BaseModel):
nodes: List[Node] = Field(descriptinotallow="圖譜中的節(jié)點(diǎn)列表。")
relationships: List[Relationship] = Field(descriptinotallow="圖譜中的關(guān)系列表。")接下來，我們創(chuàng)建一個更詳細(xì)的提示。這個提示將明確包含從Pydantic模型生成的JSON schema，并給LLM非常具體的指令。
我們的目標(biāo)是盡量減少錯誤。
# 導(dǎo)入提示模板和輸出解析器
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers.json import JsonOutputParser
# 創(chuàng)建我們期望輸出結(jié)構(gòu)的實(shí)例
parser = JsonOutputParser(pydantic_object=KnowledgeGraph)
# 創(chuàng)建一個詳細(xì)的提示模板，包含明確指令
template = """
你是一個頂級算法，擅長以結(jié)構(gòu)化格式提取信息。
從給定的輸入文本中提取知識圖譜，包括節(jié)點(diǎn)和關(guān)系。
你的目標(biāo)是盡可能全面，提取所有相關(guān)實(shí)體及其連接。
將輸出格式化為帶有'nodes'和'relationships'鍵的JSON對象。
嚴(yán)格遵循以下JSON schema：
{schema}
以下是輸入文本：
--------------------
{text}
--------------------
"""
prompt = ChatPromptTemplate.from_template(
template,
partial_variables={"schema": parser.get_format_instructions()},
)
# 創(chuàng)建完整的提取鏈
chain = prompt | llm | parser這個新鏈比LLMGraphTransformer更明確。我們給模型提供了詳細(xì)的schema和清晰的指令。讓我們再次運(yùn)行20篇文章樣本，看看成功率是否有所提高。
# 這個列表將存儲新結(jié)果
graph_documents_prompt_engineered = []
errors = []
for i, row in tqdm(news.head(NUM_ARTICLES).iterrows(), total=NUM_ARTICLES, desc="使用改進(jìn)提示處理"):
text = f"{row['title']} {row['text']}"
try:
# 調(diào)用我們改進(jìn)的新鏈
graph_data = chain.invoke({"text": text})
# 手動將解析的JSON轉(zhuǎn)換回GraphDocument格式
nodes = [Node(id=node['id'], type=node['type']) for node in graph_data.get('nodes', [])]
relationships = [Relationship(source=Node(id=rel['source']['id'], type=rel['source']['type']),
target=Node(id=rel['target']['id'], type=rel['target']['type']),
type=rel['type']) for rel in graph_data.get('relationships', [])]
doc = Document(page_cnotallow=text)
graph_documents_prompt_engineered.append(GraphDocument(nodes=nodes, relatinotallow=relationships, source=doc))
except Exception as e:
# 如果LLM輸出不是有效的JSON，解析器會失敗。我們捕獲這個錯誤。
errors.append(str(e))
doc = Document(page_cnotallow=text)
graph_documents_prompt_engineered.append(GraphDocument(nodes=[], relatinotallow=[], source=doc))現(xiàn)在是關(guān)鍵時刻。讓我們再次檢查失敗率。
# 初始化一個計(jì)數(shù)器，用于統(tǒng)計(jì)沒有節(jié)點(diǎn)的文檔
empty_count_prompt_engineered = 0
# 遍歷新結(jié)果
for doc in graph_documents_prompt_engineered:
if not doc.nodes:
empty_count_prompt_engineered += 1
# 計(jì)算并打印新的失敗百分比
print(f"Percentage missing with improved prompt: {empty_count_prompt_engineered / len(graph_documents_prompt_engineered) * 100}%")
print(f"Number of JSON parsing errors: {len(errors)}")輸出
Percentage missing with improved prompt: 62.0%
Number of JSON parsing errors: 13結(jié)果呢？失敗率約為62%。雖然比最初的75%略有改進(jìn)，但仍遠(yuǎn)不夠可靠。我們?nèi)匀粺o法從20篇文章中的13篇提取圖譜。JsonOutputParser每次都拋出錯誤，因?yàn)楸M管我們盡力優(yōu)化了提示，llama3仍然生成了格式錯誤的JSON。
這表明了一個根本性限制：
僅靠提示工程無法完全解決較小LLM生成不一致結(jié)構(gòu)化輸出的問題。
那么，如果更好的提示不是答案，那是什么？我們需要一個工具，不僅要求好的輸出，還能聰明地處理LLM給出的不完美輸出。這正是BAML設(shè)計(jì)來解決的問題。
在接下來的部分，我們將用BAML驅(qū)動的實(shí)現(xiàn)替換整個鏈，看看它帶來的變化。
BAML的初始化和快速概覽
我們已經(jīng)確定，即使小心進(jìn)行提示工程，依賴嚴(yán)格的JSON解析與較小的LLM一起使用是失敗的秘訣。模型很強(qiáng)大，但不是完美的格式化工具。
這正是BAML（Basically, A Made-up Language）非常重要的地方。BAML提供了兩個關(guān)鍵優(yōu)勢，直接解決了我們的問題：
?簡化的Schema：BAML不用冗長的JSON schema，而是使用類似TypeScript的簡潔語法定義數(shù)據(jù)結(jié)構(gòu)。這對人類和LLM都更容易理解，減少token使用和混淆的可能性。
?魯棒的解析：BAML的客戶端帶有“模糊”或“schema對齊”的解析器。它不期望完美的JSON，能處理LLM常見的錯誤，如多余的逗號、缺少引號或多余文本，仍然成功提取數(shù)據(jù)。
首先，你需要安裝BAML客戶端和它的VS Code擴(kuò)展。
# 安裝BAML客戶端
pip install baml-py在VS Code市場中搜索BAML并安裝擴(kuò)展。這個擴(kuò)展很棒，因?yàn)樗峁┝艘粋€交互式游樂場，讓你無需每次運(yùn)行Python代碼即可測試提示和schema。
接下來，我們在一個.baml文件中定義圖譜提取邏輯。將其視為LLM調(diào)用的配置文件。我們創(chuàng)建一個名為??extract_graph.baml??的文件：
// 定義圖譜中的節(jié)點(diǎn)，包含ID、類型和可選屬性
class SimpleNode {
id string // 節(jié)點(diǎn)的唯一標(biāo)識符
type string // 節(jié)點(diǎn)的類型/類別
properties Properties // 與節(jié)點(diǎn)相關(guān)的附加屬性
}
// 定義節(jié)點(diǎn)或關(guān)系的可選屬性結(jié)構(gòu)
class Properties {
description string? // 可選的文本描述
}
// 定義兩個節(jié)點(diǎn)之間的關(guān)系
class SimpleRelationship {
source_node_id string // 源節(jié)點(diǎn)的ID
source_node_type string // 源節(jié)點(diǎn)的類型
target_node_id string // 目標(biāo)節(jié)點(diǎn)的ID
target_node_type string // 目標(biāo)節(jié)點(diǎn)的類型
type string // 關(guān)系類型（例如，"connects_to", "belongs_to"）
properties Properties // 關(guān)系的附加屬性
}
// 定義包含節(jié)點(diǎn)和關(guān)系的整體圖譜
class DynamicGraph {
nodes SimpleNode[] // 圖譜中的所有節(jié)點(diǎn)列表
relationships SimpleRelationship[] // 節(jié)點(diǎn)之間的所有關(guān)系列表
}
// 從原始輸入字符串提取DynamicGraph的函數(shù)
function ExtractGraph(graph: string) -> DynamicGraph {
client Ollama // 使用Ollama客戶端解釋輸入
prompt #"
Extract from this content:
{{ ctx.output_format }}
{{ graph }} // 提示模板，指導(dǎo)Ollama提取圖譜
}類定義簡單易讀。??ExtractGraph???函數(shù)告訴BAML使用Ollama客戶端，并提供了一個Jinja提示模板。特殊的??{{ ctx.output_format }}??變量是BAML自動注入我們簡化schema定義的地方。
將BAML與LangChain集成
現(xiàn)在，我們將這個BAML函數(shù)集成到LangChain工作流程中。我們需要一些輔助函數(shù)，將BAML的輸出轉(zhuǎn)換為LangChain和Neo4j理解的GraphDocument格式。
# 導(dǎo)入必要的庫
from typing importAny, List
import baml_client as client
from langchain_community.graphs.graph_document import GraphDocument, Node, Relationship
from langchain_core.runnables import chain
# 輔助函數(shù)，正確格式化節(jié)點(diǎn)（例如，適當(dāng)大寫）
def_format_nodes(nodes: List[Node]) -> List[Node]:
return [
Node(
id=el.id.title() ifisinstance(el.id, str) else el.id,
type=el.type.capitalize() if el.typeelseNone,
properties=el.properties
)
for el in nodes
]
# 輔助函數(shù)，將BAML的關(guān)系輸出映射到LangChain的Relationship對象
defmap_to_base_relationship(rel: Any) -> Relationship:
source = Node(id=rel.source_node_id, type=rel.source_node_type)
target = Node(id=rel.target_node_id, type=rel.target_node_type)
return Relationship(
source=source, target=target, type=rel.type, properties=rel.properties
)
# 主要輔助函數(shù)，格式化所有關(guān)系
def_format_relationships(rels) -> List[Relationship]:
relationships = [
map_to_base_relationship(rel)
for rel in rels
if rel.typeand rel.source_node_id and rel.target_node_id
]
return [
Relationship(
source=_format_nodes([el.source])[0],
target=_format_nodes([el.target])[0],
type=el.type.replace(" ", "_").upper(),
properties=el.properties,
)
for el in relationships
]
# 定義一個LangChain可鏈?zhǔn)秸{(diào)用的函數(shù)，調(diào)用我們的BAML函數(shù)
@chain
asyncdefget_graph(message):
graph = await client.b.ExtractGraph(graph=message.content)
return graph讓我們了解每個輔助函數(shù)的目的：
- ??
?_format_nodes(nodes)??：通過大寫ID和類型來標(biāo)準(zhǔn)化節(jié)點(diǎn)格式，返回格式整潔的Node對象列表。 - ??
?map_to_base_relationship(rel)??：將原始BAML關(guān)系轉(zhuǎn)換為基本的LangChain Relationship對象，將源和目標(biāo)包裝為Node對象。 - ??
?_format_relationships(rels)??：過濾無效關(guān)系，將其映射到LangChain Relationship對象，并格式化節(jié)點(diǎn)類型和關(guān)系類型以保持一致性。 - ??
?get_graph(message)??：一個異步鏈函數(shù)，將輸入消息發(fā)送到BAML API，調(diào)用ExtractGraph，并返回原始圖譜輸出。
有了這些輔助函數(shù)，我們可以定義新的處理鏈。我們將使用一個更簡單的自定義提示，因?yàn)锽AML為我們處理了復(fù)雜的schema注入。
# 導(dǎo)入提示模板
from langchain_core.prompts import ChatPromptTemplate
# 一個簡單有效的系統(tǒng)提示
system_prompt = """
你是一個知識淵博的助手，擅長從文本中提取實(shí)體及其關(guān)系。
你的目標(biāo)是創(chuàng)建知識圖譜。
"""
# 最終提示模板
default_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
(
"human",
(
"提示：確保以正確格式回答，不要包含任何解釋。 "
"使用給定格式從以下輸入中提取信息：{input}"
),
),
]
)
# 定義完整的BAML驅(qū)動鏈
chain = default_prompt | llm | get_graph這個提示模板指導(dǎo)模型提取實(shí)體和關(guān)系以構(gòu)建知識圖譜：
???system_prompt??：將模型角色設(shè)置為實(shí)體-關(guān)系提取器。
???default_prompt??：結(jié)合系統(tǒng)和人類消息，帶有輸入文本的占位符。
???chain??：通過語言模型運(yùn)行提示，然后將輸出傳遞給get_graph進(jìn)行圖譜提取。
運(yùn)行BAML實(shí)驗(yàn)
現(xiàn)在是再次運(yùn)行實(shí)驗(yàn)的時候了。這次我們將處理更大的文章批次，以真正測試新方法的可靠性。
由于時間限制，我在處理344篇文章后停止了執(zhí)行，但這比最初的20篇樣本要穩(wěn)健得多。
在執(zhí)行并行處理之前，需要一些輔助函數(shù)，我們先來寫這些函數(shù)。
import asyncio
# 異步函數(shù)，處理單個文檔
asyncdefaprocess_response(document: Document) -> GraphDocument:
# 調(diào)用我們的BAML鏈
resp = await chain.ainvoke({"input": document.page_content})
# 將響應(yīng)格式化為GraphDocument
return GraphDocument(
nodes=_format_nodes(resp.nodes),
relatinotallow=_format_relationships(resp.relationships),
source=document,
)
# 異步函數(shù)，處理文檔列表
asyncdefaconvert_to_graph_documents(
documents: List[Document],
) -> List[GraphDocument]:
tasks = [asyncio.create_task(aprocess_response(document)) for document in documents]
results = await asyncio.gather(*tasks)
return results
# 異步函數(shù)，處理原始文本
asyncdefaprocess_text(texts: List[str]) -> List[GraphDocument]:
docs = [Document(page_cnotallow=text) for text in texts]
graph_docs = await aconvert_to_graph_documents(docs)
return graph_docs讓我們分解每個異步函數(shù)的目的：
???aprocess_response??：處理一個文檔并返回GraphDocument。
???aconvert_to_graph_documents??：并行處理多個文檔并返回圖譜結(jié)果。
???aprocess_text??：將原始文本轉(zhuǎn)換為文檔并提取圖譜數(shù)據(jù)。
現(xiàn)在，我們可以簡單地執(zhí)行主循環(huán)來處理文章。
# 初始化一個空列表，存儲生成的圖譜文檔
graph_documents_baml = []
# 設(shè)置要處理的文章總數(shù)
NUM_ARTICLES_BAML = 344
# 創(chuàng)建一個僅包含要處理的文章的較小DataFrame
news_baml = news.head(NUM_ARTICLES_BAML)
# 從新DataFrame中提取標(biāo)題和文本
titles = news_baml["title"]
texts = news_baml["text"]
# 定義每批（chunk）處理的文章數(shù)量
chunk_size = 4
# 使用tqdm顯示進(jìn)度條，逐批迭代文章
for i in tqdm(range(0, len(titles), chunk_size), desc="使用BAML處理分塊"):
# 獲取當(dāng)前分塊的標(biāo)題
title_chunk = titles[i : i + chunk_size]
# 獲取當(dāng)前分塊的文本
text_chunk = texts[i : i + chunk_size]
# 將每篇文章的標(biāo)題和文本合并為單個字符串
combined_docs = [f"{title} {text}"for title, text inzip(title_chunk, text_chunk)]
try:
# 異步處理合并的文檔以提取圖譜結(jié)構(gòu)
docs = await aprocess_text(combined_docs)
# 將處理好的圖譜文檔添加到主列表
graph_documents_baml.extend(docs)
except Exception as e:
# 處理處理過程中發(fā)生的任何錯誤并打印錯誤消息
print(f"處理從索引{i}開始的分塊時出錯：{e}")
# 循環(huán)結(jié)束后，顯示成功處理的圖譜文檔總數(shù)
len(graph_documents_baml)這是我們得到的輸出。
# 圖譜文檔總數(shù)
344
我們處理了344篇文章。現(xiàn)在，讓我們運(yùn)行之前做的失敗分析。
# 初始化一個計(jì)數(shù)器，用于統(tǒng)計(jì)沒有節(jié)點(diǎn)的文檔
empty_count_baml = 0
# 遍歷BAML方法的處理結(jié)果
for doc in graph_documents_baml:
if not doc.nodes:
empty_count_baml += 1
# 計(jì)算并打印新的失敗百分比
print(f"Percentage missing with BAML: {empty_count_baml / len(graph_documents_baml) * 100}%")輸出
Percentage missing with BAML: 0.5813953488372093%這是一個驚人的結(jié)果。我們的失敗率從75%下降到僅0.58%。這意味著我們的成功率現(xiàn)在是99.4%！
通過簡單地將嚴(yán)格的LLMGraphTransformer替換為BAML驅(qū)動的鏈，我們從一個失敗的原型轉(zhuǎn)變?yōu)橐粋€穩(wěn)健的生產(chǎn)就緒流程。
這表明瓶頸不是小型LLM理解任務(wù)的能力，而是系統(tǒng)對完美JSON的脆弱期望。
使用Neo4j分析GraphRAG
僅僅提取實(shí)體是不夠的。GraphRAG的真正力量在于結(jié)構(gòu)化這些知識，找到隱藏的聯(lián)系，并總結(jié)相關(guān)信息的社區(qū)。
我們現(xiàn)在將高質(zhì)量的圖譜數(shù)據(jù)加載到Neo4j中，并使用圖數(shù)據(jù)科學(xué)技術(shù)來豐富它。
首先，我們設(shè)置與Neo4j數(shù)據(jù)庫的連接。
import os
from langchain_community.graphs import Neo4jGraph
# 使用環(huán)境變量設(shè)置Neo4j連接詳情
os.environ["NEO4J_URI"] = "bolt://localhost:7687"
os.environ["NEO4J_USERNAME"] = "neo4j"
os.environ["NEO4J_PASSWORD"] = "your_password" # 將此更改為你的密碼
os.environ["DATABASE"] = "graphragdemo"
# 初始化Neo4jGraph對象
graph = Neo4jGraph()現(xiàn)在，我們可以將graph_documents_baml添加到數(shù)據(jù)庫中。??baseEntityLabel=True???參數(shù)為所有節(jié)點(diǎn)添加??__Entity__??標(biāo)簽，便于后續(xù)查詢。
# 將圖譜文檔添加到Neo4j
graph.add_graph_documents(graph_documents_baml, baseEntityLabel=True, include_source=True)數(shù)據(jù)加載后，我們可以運(yùn)行一些Cypher查詢來了解新知識圖譜的結(jié)構(gòu)。讓我們從查看文章長度（以token計(jì)）和從中提取的實(shí)體數(shù)量之間的關(guān)系開始。
# 導(dǎo)入繪圖和數(shù)據(jù)分析的庫
import matplotlib.pyplot as plt
import seaborn as sns
# 查詢Neo4j以獲取每個文檔的實(shí)體數(shù)量和token數(shù)量
entity_dist = graph.query(
"""
MATCH (d:Document)
RETURN d.text AS text,
count {(d)-[:MENTIONS]->()} AS entity_count
"""
)
entity_dist_df = pd.DataFrame.from_records(entity_dist)
entity_dist_df["token_count"] = [
num_tokens_from_string(str(el)) for el in entity_dist_df["text"]
]
# 創(chuàng)建帶回歸線的散點(diǎn)圖
sns.lmplot(
x="token_count", y="entity_count", data=entity_dist_df, line_kws={"color": "red"}
)
plt.title("實(shí)體數(shù)量與Token數(shù)量分布")
plt.xlabel("Token數(shù)量")
plt.ylabel("實(shí)體數(shù)量")
plt.show()
實(shí)體數(shù)量與Token數(shù)量
該圖顯示了一個明顯的正相關(guān)：隨著文章中token數(shù)量的增加，提取的實(shí)體數(shù)量也傾向于增加。這正是我們期望的，證實(shí)了我們的提取過程表現(xiàn)得很合理。
接下來，我們看看節(jié)點(diǎn)度分布。這告訴我們實(shí)體的連接程度。在現(xiàn)實(shí)世界的網(wǎng)絡(luò)中，少數(shù)高度連接的節(jié)點(diǎn)（中心節(jié)點(diǎn)）是常見的。
import numpy as np
# 查詢每個實(shí)體節(jié)點(diǎn)的度
degree_dist = graph.query(
"""
MATCH (e:__Entity__)
RETURN count {(e)-[:!MENTIONS]-()} AS node_degree
"""
)
degree_dist_df = pd.DataFrame.from_records(degree_dist)
# 計(jì)算統(tǒng)計(jì)數(shù)據(jù)
mean_degree = np.mean(degree_dist_df["node_degree"])
percentiles = np.percentile(degree_dist_df["node_degree"], [25, 50, 75, 90])
# 繪制對數(shù)尺度的直方圖
plt.figure(figsize=(12, 6))
sns.histplot(degree_dist_df["node_degree"], bins=50, kde=False, color="blue")
plt.yscale("log")
plt.title("節(jié)點(diǎn)度分布")
plt.legend()
plt.show()
節(jié)點(diǎn)度分布
直方圖顯示了一個“長尾”分布，這是知識圖譜的典型特征。大多數(shù)實(shí)體只有少數(shù)連接（低度），而少數(shù)實(shí)體是高度連接的中心節(jié)點(diǎn)。
例如，第90百分位的度數(shù)是4，但最大度數(shù)是37。這表明像“USA”或“Microsoft”這樣的實(shí)體可能是圖譜中的中心點(diǎn)。
為了找到語義上相似的實(shí)體（即使名稱不同），我們需要為它們創(chuàng)建向量嵌入（embedding）。嵌入是文本的數(shù)字表示。我們將為每個實(shí)體的ID和描述生成嵌入，并存儲在圖譜中。
我們將通過Ollama使用llama3模型進(jìn)行嵌入，并使用LangChain的Neo4jVector來處理這個過程。
from langchain_community.vectorstores import Neo4jVector
from langchain_ollama import OllamaEmbeddings
# 使用本地llama3模型創(chuàng)建嵌入
embeddings = OllamaEmbeddings(model="llama3")
# 初始化Neo4jVector實(shí)例以管理圖譜中的嵌入
vector = Neo4jVector.from_existing_graph(
embeddings,
node_label="__Entity__",
text_node_properties=["id", "description"],
embedding_node_property="embedding",
database=os.environ["DATABASE"],
)此命令遍歷Neo4j中的所有??__Entity__??節(jié)點(diǎn)，為其屬性生成嵌入，并將其存儲回節(jié)點(diǎn)的embedding屬性中。
查找和鏈接相似實(shí)體
有了嵌入，我們現(xiàn)在可以使用k-Nearest Neighbors（kNN）算法找到向量空間中彼此接近的節(jié)點(diǎn)。這是識別潛在重復(fù)或高度相關(guān)實(shí)體（例如，“Man United”和“Manchester United”）的強(qiáng)大方法。
我們將使用Neo4j的Graph Data Science（GDS）庫來實(shí)現(xiàn)這一點(diǎn)。
# 導(dǎo)入GraphDataScience庫
from graphdatascience import GraphDataScience
# --- GDS客戶端初始化 ---
# 初始化GraphDataScience客戶端以連接到Neo4j數(shù)據(jù)庫
# 使用環(huán)境變量中的連接詳情（URI、用戶名、密碼）
gds = GraphDataScience(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"]),
)
# 為GDS操作設(shè)置特定數(shù)據(jù)庫
gds.set_database(os.environ["DATABASE"])
# --- 內(nèi)存圖投影 ---
# 將圖譜投影到內(nèi)存中以便GDS算法高效處理
# 此投影命名為'entities'
G, result = gds.graph.project(
"entities", # 內(nèi)存圖的名稱
"__Entity__", # 要投影的節(jié)點(diǎn)標(biāo)簽
"*", # 投影所有關(guān)系類型
nodeProperties=["embedding"] # 包含節(jié)點(diǎn)的'embedding'屬性
)
# --- 使用kNN計(jì)算相似性 ---
# 定義創(chuàng)建關(guān)系的相似性閾值
similarity_threshold = 0.95
# 使用k-Nearest Neighbors（kNN）算法找到相似節(jié)點(diǎn)
# 這會通過添加新關(guān)系“變異”內(nèi)存圖
gds.knn.mutate(
G, # 要修改的內(nèi)存圖
nodeProperties=["embedding"], # 用于相似性計(jì)算的屬性
mutateRelatinotallow="SIMILAR", # 要創(chuàng)建的關(guān)系類型
mutateProperty="score", # 新關(guān)系上存儲相似性分?jǐn)?shù)的屬性
similarityCutoff=similarity_threshold, # 過濾關(guān)系的閾值
)我們?yōu)榍度胂嗨菩苑謹(jǐn)?shù)高于0.95的節(jié)點(diǎn)創(chuàng)建??SIMILAR??關(guān)系。
kNN算法幫助我們找到了潛在的重復(fù)實(shí)體，但僅靠文本相似性并不完美。我們可以通過尋找不僅語義相似而且名稱非常相似的實(shí)體（低“編輯距離”）進(jìn)一步優(yōu)化。
我們將查詢這些候選實(shí)體，然后使用LLM做出最終的合并決定。
# 根據(jù)社區(qū)和名稱相似性查詢潛在重復(fù)實(shí)體
word_edit_distance = 3
potential_duplicate_candidates = graph.query(
"""
MATCH (e:`__Entity__`)
WHERE size(e.id) > 4
WITH e.wcc AS community, collect(e) AS nodes, count(*) AS count
WHERE count > 1
# ... (筆記本中的完整Cypher查詢) ...
RETURN distinct(combinedResult)
""",
params={"distance": word_edit_distance},
)
# 看看幾個候選實(shí)體
potential_duplicate_candidates[:5]上述代碼的輸出如下。
輸出
[{'combinedResult': ['David Van', 'Davidvan']},
{'combinedResult': ['Cyb003', 'Cyb004']},
{'combinedResult': ['Delta Air Lines', 'Delta_Air_Lines']},
{'combinedResult': ['Elon Musk', 'Elonmusk']},
{'combinedResult': ['Market', 'Markets']}]這些看起來明顯是重復(fù)的。我們現(xiàn)在可以使用另一個BAML函數(shù)讓LLM決定保留哪個名稱。運(yùn)行這個分辨過程后，我們在Neo4j中合并這些節(jié)點(diǎn)。
# (假設(shè)'merged_entities'由LLM分辨過程創(chuàng)建)
graph.query(
"""
UNWIND $data AS candidates
CALL {
WITH candidates
MATCH (e:__Entity__) WHERE e.id IN candidates
RETURN collect(e) AS nodes
}
CALL apoc.refactor.mergeNodes(nodes, {properties: {'`.*`': 'discard'}})
YIELD node
RETURN count(*)
""",
params={"data": merged_entities},
)使用Leiden算法進(jìn)行社區(qū)檢測
現(xiàn)在是GraphRAG的核心：將相關(guān)實(shí)體分組為社區(qū)。
我們將投影整個圖譜（包括所有原始關(guān)系）到內(nèi)存中，并運(yùn)行Leiden算法，這是一個最先進(jìn)的社區(qū)檢測算法。
# 投影整個圖譜，按關(guān)系頻率加權(quán)
G, result = gds.graph.project(
"communities",
"__Entity__",
{
"_ALL_": {
"type": "*",
"orientation": "UNDIRECTED",
"properties": {"weight": {"property": "*", "aggregation": "COUNT"}},
}
},
)
# 運(yùn)行Leiden社區(qū)檢測并將結(jié)果寫回節(jié)點(diǎn)
gds.leiden.write(
G,
writeProperty="communities",
includeIntermediateCommunities=True, # 這會創(chuàng)建層次社區(qū)
relatinotallow="weight",
)這會為每個實(shí)體節(jié)點(diǎn)添加一個communities屬性，這是一個不同粒度級別的社區(qū)ID列表（從小型緊密群體到更大的廣泛主題）。
最后，我們通過創(chuàng)建??__Community__??節(jié)點(diǎn)并將它們鏈接起來，將這個層次結(jié)構(gòu)具體化在圖譜中。這創(chuàng)建了一個可瀏覽的主題結(jié)構(gòu)。
# 為社區(qū)節(jié)點(diǎn)創(chuàng)建唯一性約束
graph.query("CREATE CONSTRAINT IF NOT EXISTS FOR (c:__Community__) REQUIRE c.id IS UNIQUE;")
# 創(chuàng)建社區(qū)節(jié)點(diǎn)并將實(shí)體和社區(qū)鏈接起來
graph.query(
"""
MATCH (e:`__Entity__`)
UNWIND range(0, size(e.communities) - 1 , 1) AS index
// ... (筆記本中的完整社區(qū)創(chuàng)建查詢) ...
RETURN count(*)
"""
)這個復(fù)雜查詢創(chuàng)建了一個多級社區(qū)結(jié)構(gòu)，例如：??(Entity)-[:IN_COMMUNITY]->(Level_0_Community)-[:IN_COMMUNITY]->(Level_1_Community)??。
分析最終圖譜結(jié)構(gòu)
經(jīng)過所有這些工作，我們的知識圖譜是什么樣的？讓我們分析每一級的社區(qū)規(guī)模。
# 查詢每一級社區(qū)的大小
community_size = graph.query(
"""
MATCH (c:__Community__)<-[:IN_COMMUNITY*]-(e:__Entity__)
WITH c, count(distinct e) AS entities
RETURN split(c.id, '-')[0] AS level, entities
"""
)
# 打印處理后的DataFrame
percentiles_df
百分位DataFrame
這個表格很重要。它顯示了Leiden算法如何對我們的1,875個實(shí)體進(jìn)行分組。
在Level 0，我們有858個小型、聚焦的社區(qū)，其中90%包含4個或更少的成員。到Level 3，算法將這些合并為732個更大、更廣泛的社區(qū)，這一級的最大社區(qū)包含77個實(shí)體。
這種層次結(jié)構(gòu)正是我們進(jìn)行有效GraphRAG所需的。我們現(xiàn)在可以在不同抽象級別進(jìn)行檢索。
結(jié)論
結(jié)果很明顯。雖然標(biāo)準(zhǔn)的LangChain工具提供了一個快速入門的途徑，但它們在與較小的開源LLM一起使用時可能不穩(wěn)定且不可靠。
通過引入BAML，我們解決了過于復(fù)雜的提示和嚴(yán)格JSON解析的核心問題。結(jié)果是將成功率從25%大幅提升到超過99%，將一個失敗的實(shí)驗(yàn)轉(zhuǎn)變?yōu)橐粋€穩(wěn)健且可擴(kuò)展的知識圖譜構(gòu)建流程。
以下是我們采取的關(guān)鍵步驟的快速回顧：
1. 我們從準(zhǔn)備新聞文章數(shù)據(jù)集和使用Ollama設(shè)置本地llama3模型開始。
2. 使用LangChain的LLMGraphTransformer進(jìn)行的第一次測試有75%的失敗率，因?yàn)閲?yán)格的JSON解析。
3. 嘗試通過高級提示工程修復(fù)，失敗率僅略微改善到約62%。
4. 然后我們集成了BAML，利用其簡化的schema和魯棒解析器實(shí)現(xiàn)了99.4%的圖譜提取成功率。
5. 將高質(zhì)量的圖譜數(shù)據(jù)加載到Neo4j中進(jìn)行結(jié)構(gòu)化和分析。
6. 通過為所有實(shí)體生成向量嵌入來豐富圖譜，捕捉語義含義。
7. 使用k-Nearest Neighbors（kNN）算法識別并鏈接語義相似的節(jié)點(diǎn)。
8. 進(jìn)一步通過LLM智能查找和合并重復(fù)實(shí)體來優(yōu)化圖譜。
9. 最后，應(yīng)用Leiden算法將實(shí)體組織成多級社區(qū)層次結(jié)構(gòu)，為高級GraphRAG奠定了基礎(chǔ)。
這種使用LangChain進(jìn)行強(qiáng)大編排和BAML進(jìn)行可靠結(jié)構(gòu)化輸出的方法是構(gòu)建強(qiáng)大且成本效益高的AI應(yīng)用的制勝組合。
本文轉(zhuǎn)載自????????PyTorch研習(xí)社????，作者：AI研究生
