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