在使用 LangChain 构建基于知识图谱的 RAG 系统或智能体时，一个主要挑战是实现从非结构化数据中准确抽取节点和关系。尤其是使用本地化小型大语言模型 (LLM) 时，这一问题尤为显著。这往往会导致 AI 系统的性能不佳。
LangChain 的信息抽取能力受限于其对严格 JSON 解析的依赖。即使在使用大型模型或高度详细的提示词模板时，这种解析方式也可能失败。
与此形成对比的是，BAML(https://github.com/BoundaryML/baml) 采用了模糊解析方法。即使大语言模型的输出不是完美的 JSON 格式，它也能成功抽取数据。
本文将探讨 LangChain 使用小型量化模型进行抽取时的局限性，并展示 BAML 如何将其抽取成功率从约 25% 提升至接近 99%。
所有代码都已上传至我的 GitHub 仓库：
目录
• 初始化评估数据集 • 量化小型 LLaMA 模型 • 基于 LLMGraphTransformer 的方法 • 理解 LangChain 的问题所在 • 改进提示词是否能够解决问题？ • BAML 初始化与快速概览 • 整合 BAML 与 LangChain • 运行 BAML 实验 • 使用 Neo4j 分析 GraphRAG • 查找并链接相似实体 • 使用 Leiden 算法进行社区检测 • 分析最终图结构 • 结论
初始化评估数据集
为了深入理解问题并找到解决方案，我们需要准备评估数据集。通过对这些数据进行多项测试，我们可以全面掌握 BAML 改进 LangChain 知识图谱抽取效果的方式。
我们将使用 Tomasonjo(https://github.com/tomasonjo) 在 GitHub 上提供的 博客数据集(https://github.com/tomasonjo/blog-datasets)。首先，让我们加载这些数据。
# Import the pandas library for data manipulation and analysis
import pandas as pd
# Load the news articles dataset from a CSV file hosted on GitHub into a pandas DataFrame
news = pd.read_csv(
"https://raw.githubusercontent.com/tomasonjo/blog-datasets/main/news_articles.csv"
)
# Display the first 5 rows of the DataFrame
news.head()我们的 DataFrame 结构非常简单：包含新闻的标题和正文（即描述）。我们需要再添加一列，用于存储每篇新闻文章文本对应的总 token 数量。
为此，我们可以使用 OpenAI 的 tiktoken 库。通过循环遍历数据集来计算 token 数量的方法很直接，我们现在开始实现。
# Import the tiktoken library to count tokens from text
import tiktoken
# Define a function to calculate the number of tokens in a given string for a specific model
defnum_tokens_from_string(string: str, model: str = "gpt-4o") -> int:
"""Returns the number of tokens in a text string."""
# Get the encoding for the specified model
encoding = tiktoken.encoding_for_model(model)
# Encode the string into tokens and count them
num_tokens = len(encoding.encode(string))
# Return the total number of tokens
return num_tokens
# Create a new column 'tokens' in the DataFrame
# It calculates the number of tokens for the combined 'title' and 'text' of each article
news["tokens"] = [
num_tokens_from_string(f"{row['title']} {row['text']}")
for i, row in news.iterrows()
]计算 DataFrame 中所有文章的 token 数量仅需几秒钟。
以下是更新后的 DataFrame 展示。
# Display the first 5 rows of the DataFrame to show the new 'tokens' column
news.head()这些 token 稍后将在我们的评估和分析阶段中使用，这也是我们执行此步骤的原因。
量化小型 LLaMA 模型
为了使用 AI 模型将数据转换为知识图谱，我们将采用轻量级量化模型进行严格测试评估。
在生产环境中，开源大语言模型通常通过量化部署来降低成本和延迟。本文将使用 LLaMA 3.1 模型。
本文选择 Ollama 作为平台，但 LangChain 支持多种 API 和本地大语言模型提供商，因此可根据需求选择任一合适选项。
# ChatOllama 是 Ollama 语言模型的接口
from langchain_ollama import ChatOllama
# 定义要使用的模型名称
model = "llama3"
# 初始化 ChatOllama 语言模型
# 'temperature' 参数控制输出的随机性。
# 较低的值（如 0.001）使模型的响应更具确定性。
llm = ChatOllama(model=model, temperature=0.001)您还需要在系统上安装 Ollama。它支持 macOS、Windows 和 Linux 操作系统。
1. 访问 Ollama 官方网站：https://ollama.com/ 2. 下载适用于您操作系统的安装程序，并按照安装说明进行操作。
安装完成后，Ollama 将作为后台服务运行。
• 在 macOS 和 Windows 上，应用程序应自动启动并在后台运行（您可能会在菜单栏或系统托盘中看到图标）。 • 在 Linux 上，您可能需要使用 systemctl start ollama手动启动它。
要检查服务是否正在运行，请打开您的终端或命令提示符并输入：
# 检查可用模型
ollama list
#### 输出 ####
[ ] <-- 暂无模型如果服务正在运行但尚未有模型，您会看到一个空模型列表，这在当前阶段是完全正常的。如果您收到“命令未找到”错误，请确保 Ollama 已正确安装。如果您遇到连接错误，则表示服务器未运行。
您只需使用 pull 命令即可下载 llama3 模型。由于模型较大，这需要一些时间和数千兆字节的磁盘空间。
# 下载 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 响应流。这证实了服务器正在运行并能够提供模型服务。
评估数据和大语言模型均已准备就绪，下一步是将数据进行转换，以深入理解 LangChain 的内部问题。
基于 LLMGraphTransformer 的方法
使用 LangChain 或 LangGraph 将原始或结构化数据转换为知识图谱的推荐方法是借助其内置工具。其中最常见的方法之一是 langchain_experimental 库中的 LLMGraphTransformer。
该工具被设计为一体化解决方案：仅需提供文本和大语言模型，它便能自动处理提示词与解析，并返回图结构。
让我们看看它在我们的本地 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 模型）。
我们还需要告诉转换器，希望它为节点和关系抽取哪些额外信息，即“属性”。在这个例子中，我们只要求抽取 description 属性。
# 使用 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_content=text)
# 使用转换器将文档转换为 GraphDocument 对象列表
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="Processing documents"
):
graph_document = future.result()
graph_documents.extend(graph_document)运行代码后，进度条显示所有 20 篇文章都已处理完毕。
#### 输出 ####
Processing documents: 100%|██████████| 20/20 [01:32<00:00, 4.64s/it]理解 LangChain 的问题所在
那么，我们得到了什么？让我们检查一下 graph_documents 列表。
# 显示图文档列表
print(graph_documents)这是我们得到的输出：
#### 输出 ####
[GraphDocument(nodes=[], relationships=[], source=Document(metadata={}, page_content='XPeng Stock Rises...')),
GraphDocument(nodes=[], relationships=[], source=Document(metadata={}, page_content='Ryanair sacks chief pilot...')),
GraphDocument(nodes=[], relationships=[], source=Document(metadata={}, page_content='Dáil almost suspended...')),
GraphDocument(nodes=[Node(id='Jude Bellingham', type='Person', properties={}), Node(id='Real Madrid', type='Organization', properties={})], relationships=[], source=Document(metadata={}, page_content='Arsenal have Rice bid rejected...')),
...
]我们立刻发现了一个问题。许多
GraphDocument对象的nodes和relationships列表都是空的。
这意味着对于这些文章，大语言模型生成的输出存在两种问题：一是 LangChain 无法将其解析为有效的图结构；二是完全没有抽取到任何实体。
这就是使用小型量化大语言模型进行结构化数据抽取的根本挑战。它们常常难以遵循像 LLMGraphTransformer 这类工具所期望的严格 JSON 格式。即使是微小的错误——比如多余的逗号、遗漏的引号——都会导致解析失败，从而一无所获。
让我们量化一下这个失败率。我们将统计 20 份文档中有多少份未能成功生成图。
# 初始化无节点文档计数器
empty_count = 0
# 遍历生成的图文档
for doc in graph_documents:
# 如果 'nodes' 列表为空，则增加计数器
if not doc.nodes:
empty_count += 1现在，我们来计算失败的百分比。
# 计算并打印未能生成任何节点的文档百分比
print(f"Percentage missing: {empty_count/len(graph_documents)*100}")
#### 输出 ####
Percentage missing: 75.075% 的失败率。这完全不能接受。这意味着在我们的 20 篇样本文章中，只有 5 篇成功转换成了知识图谱。
对于任何生产系统来说，仅有 25% 的成功率是不可接受的。
症结所在即在此，这是一个普遍现象。标准方法对于小型大语言模型稍显不稳定的特性来说过于僵化。
改进提示词是否能够解决问题？
75% 的失败率是一个巨大的问题。作为开发者，当大语言模型表现不佳时，我们的第一反应通常是调整提示词。更明确的指令通常应带来更好的结果。LLMGraphTransformer 内部使用默认提示词，但我们无法轻易修改它。
因此，让我们使用 Langchain 的 ChatPromptTemplate 构建一个自己的简单链。这使我们能够完全控制发送给 llama3 的指令。我们可以更明确地“引导”模型，使其每次都生成正确的 JSON 格式。
首先，我们使用 Pydantic 模型来定义我们期望的输出结构。这在 Langchain 中是用于结构化输出的常见模式。
# 导入 Pydantic 模型以定义数据结构
from langchain_core.pydantic_v1 import BaseModel, Field
# 定义一个简单的节点结构
classNode(BaseModel):
id: str = Field(description="节点的唯一标识符。")
type: str = Field(description="节点的类型（例如：Person, Organization）。")
# 定义一个简单的关系结构
classRelationship(BaseModel):
source: Node = Field(description="关系源节点。")
target: Node = Field(description="关系目标节点。")
type: str = Field(description="关系的类型（例如：WORKS_FOR）。")
# 定义整体图结构
classKnowledgeGraph(BaseModel):
nodes: List[Node] = Field(description="图中的节点列表。")
relationships: List[Relationship] = Field(description="图中的关系列表。")接下来，我们将创建一个新的、更详细的提示词。这个提示词将明确包含由我们的 Pydantic 模型生成的 JSON Schema，并向大语言模型提供非常具体的指令。
我们的想法是：不给模型留下任何出错的可能性。
# 导入提示词模板和输出解析器
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="Processing with better prompt"):
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_content=text)
graph_documents_prompt_engineered.append(GraphDocument(nodes=nodes, relationships=relationships, source=doc))
except Exception as e:
# 如果大语言模型输出的 JSON 无效，解析器将失败。我们将捕获该错误。
errors.append(str(e))
doc = Document(page_content=text)
graph_documents_prompt_engineered.append(GraphDocument(nodes=[], relationships=[], source=doc))现在，我们来揭晓结果。让我们再次检查失败率。
# 初始化无节点文档计数器
empty_count_prompt_engineered = 0
# 遍历新结果
for doc in graph_documents_prompt_engineered:
ifnot 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% 略有改善，但仍然远未达到可靠的水平。每次失败的原因都是 llama3 生成了格式错误的 JSON，即使我们已尽力优化提示词，这仍导致 JsonOutputParser 抛出错误。
这表明了一个根本性的局限性：
单纯依靠提示词工程，无法完全解决小型大语言模型结构化输出不一致的问题。
那么，如果改进提示词不是解决办法，那是什么呢？我们需要一个工具，它不仅要求好的输出，而且足够智能，能够处理大语言模型给出的不完美输出。这正是 BAML 旨在解决的问题。
在接下来的章节中，我们将用 BAML 驱动的实现来替换整个链，看看它能带来怎样的改变。
BAML 初始化与快速概览
我们已经确定，即使经过精心设计的提示词工程，对于小型大语言模型来说，过度依赖严格的 JSON 解析也注定会失败。这些模型功能强大，但并非完美的格式化工具。
这就是 BAML (Basically, A Made-up Language) 发挥重要作用的地方。BAML 提供了两大关键优势，直接解决了我们的问题：
1. 简化 Schema： BAML 不使用冗长的 JSON Schema，而是采用一种简洁、类似 TypeScript 的语法来定义数据结构。这种语法既便于人类理解，也利于大语言模型解析，同时减少了 token 消耗并降低了出错风险。 2. 健壮的解析： BAML 的客户端内置了“模糊”或“与 Schema 对齐”的解析器。它不期望完美的 JSON。即使大语言模型出现常见的错误，如多余的逗号、遗漏的引号或额外文本，它仍能成功抽取数据。
首先，您需要安装 BAML 客户端及其 VS Code 扩展。
# 安装 baml 客户端
pip install baml-py在 VS Code 市场中搜索 baml 并安装该扩展。这个扩展功能强大，因为它提供了一个交互式平台，让您无需每次都运行 Python 代码即可测试提示词和 Schema。
接下来，我们将在 .baml 文件中定义图抽取逻辑。可以将其视为我们大语言模型调用的配置文件。我们将创建一个名为 extract_graph.baml 的文件：
// 定义图中的节点，包含 ID、类型和可选属性
classSimpleNode {
id string // 节点的唯一标识符
type string // 节点的类型/类别
properties Properties // 与节点关联的额外属性
}
// 定义节点或关系的可选属性结构
classProperties {
description string? // 可选的文本描述
}
// 定义两个节点之间的关系
classSimpleRelationship {
source_node_id string // 源节点的 ID
source_node_type string // 源节点的类型
target_node_id string // 目标节点的 ID
target_node_type string // 目标节点的类型
type string // 关系类型（例如：“连接到”、“属于”）
properties Properties // 关系的额外属性
}
// 定义由节点和关系组成的整体图结构
classDynamicGraph {
nodes SimpleNode[] // 图中所有节点的列表
relationships SimpleRelationship[] // 节点间所有关系的列表
}
// 从原始输入字符串中抽取 DynamicGraph 的函数
function ExtractGraph(graph: string) -> DynamicGraph {
client Ollama // 使用 Ollama 客户端解释输入
prompt #"
Extract fromthis content:
{{ ctx.output_format }}
{{ graph }} // 指导 Ollama 抽取图的提示词模板
}class 定义简洁易读。ExtractGraph 函数指示 BAML 使用 Ollama 客户端并提供一个 Jinja 提示词模板。特殊变量 {{ ctx.output_format }} 由 BAML 自动注入，用于定义简化的 Schema。
将 BAML 集成到 LangChain 工作流中
下面，我们将 BAML 函数集成到 LangChain 的工作流中。为此，我们需要编写辅助函数，将 BAML 输出转换为 LangChain 和 Neo4j 可识别的 GraphDocument 格式。
# Import necessary libraries
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
# Helper function to format nodes correctly (e.g., proper capitalization)
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
]
# Helper to map BAML's relationship output to Langchain's Relationship object
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
)
# Main helper to format all relationships
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
]
# Define a LangChain chainable function to call our BAML function
@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 关系转换为基本的 LangChainRelationship对象，通过将源节点和目标节点封装为Node对象。• _format_relationships(rels)：过滤无效关系，将其映射为 LangChainRelationship对象，并统一节点和关系的类型格式。• get_graph(message)：这是一个异步链式函数，负责将输入消息发送到 BAML API，调用ExtractGraph函数，并返回原始的图数据输出。
借助这些辅助函数，我们可以定义新的处理链。我们将使用一个更为简单的自定义提示词，因为 BAML 会替我们处理复杂的模式注入工作。
# Import the prompt template
from langchain_core.prompts import ChatPromptTemplate
# A simple, effective system prompt
system_prompt = """
You are a knowledgeable assistant skilled in extracting entities and their relationships from text.
Your goal is to create a knowledge graph.
"""
# The final prompt template
default_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
(
"human",
(
"Tip: Make sure to answer in the correct format and do not include any explanations. "
"Use the given format to extract information from the following input: {input}"
),
),
]
)
# Define the full BAML-powered chain
chain = default_prompt | llm | get_graph该系统提示模板指导模型从文本中提取实体和关系，以构建知识图谱：
• system_prompt：将模型的角色设定为实体-关系提取器。• default_prompt：结合系统指令和用户输入，并包含一个用于输入文本的占位符。• chain：将提示词通过语言模型运行，然后将其输出传递给get_graph函数进行图提取。
运行 BAML 实验
现在，我们将再次运行实验。这一次，我们将处理更大批量的文章，以真正测试这种新方法的可靠性。
由于时间限制，我在处理完 344 篇文章后停止了执行，但与最初的20篇相比，这是一个更具代表性的样本量。
为了实现并行处理，我们需要一些辅助函数。接下来，我们首先编写这些函数。
import asyncio
# Asynchronous function to process a single document
asyncdefaprocess_response(document: Document) -> GraphDocument:
# Invoke our BAML chain
resp = await chain.ainvoke({"input": document.page_content})
# Format the response into a GraphDocument
return GraphDocument(
nodes=_format_nodes(resp.nodes),
relationships=_format_relationships(resp.relationships),
source=document,
)
# Asynchronous function to process a list of documents
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
# Asynchronous function to process raw texts
asyncdefaprocess_text(texts: List[str]) -> List[GraphDocument]:
docs = [Document(page_content=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：将原始文本转换为文档并提取图数据。
现在，我们可以执行主循环来处理我们的文章。
# Initialize an empty list to store the resulting graph documents.
graph_documents_baml = []
# Set the total number of articles to be processed.
NUM_ARTICLES_BAML = 344
# Create a smaller DataFrame containing only the articles to be processed.
news_baml = news.head(NUM_ARTICLES_BAML)
# Extract titles and texts from the new DataFrame.
titles = news_baml["title"]
texts = news_baml["text"]
# Define the number of articles to process in each batch (chunk).
chunk_size = 4
# Iterate over the articles in chunks, using tqdm to display a progress bar.
for i in tqdm(range(0, len(titles), chunk_size), desc="Processing Chunks with BAML"):
# Get the titles for the current chunk.
title_chunk = titles[i : i + chunk_size]
# Get the texts for the current chunk.
text_chunk = texts[i : i + chunk_size]
# Combine the title and text for each article in the chunk into a single string.
combined_docs = [f"{title} {text}"for title, text inzip(title_chunk, text_chunk)]
try:
# Asynchronously process the combined documents to extract graph structures.
docs = await aprocess_text(combined_docs)
# Add the processed graph documents to the main list.
graph_documents_baml.extend(docs)
except Exception as e:
# Handle any errors that occur during processing and print an error message.
print(f"Error processing chunk starting at index {i}: {e}")
# After the loop, display the total number of graph documents successfully processed.
len(graph_documents_baml)这是我们得到的输出。
# Total number of graph documents
344
我们处理了344篇文章。现在，让我们进行与之前相同的故障分析。
# Initialize a counter for documents with no nodes
empty_count_baml = 0
# Iterate through the results from the BAML approach
for doc in graph_documents_baml:
if not doc.nodes:
empty_count_baml += 1
# Calculate and print the new failure percentage
print(f"Percentage missing with BAML: {empty_count_baml / len(graph_documents_baml) * 100}%")
#### OUTPUT ####
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
# Set up Neo4j connection details using environment variables
os.environ["NEO4J_URI"] = "bolt://localhost:7687"
os.environ["NEO4J_USERNAME"] = "neo4j"
os.environ["NEO4J_PASSWORD"] = "your_password" # Change this to your password
os.environ["DATABASE"] = "graphragdemo"
# Initialize the Neo4jGraph object
graph = Neo4jGraph()现在，我们可以将 graph_documents_baml 添加到数据库中。baseEntityLabel=True 参数会为所有节点添加 __Entity__ 标签，这对于后续查询非常有用。
# Add the graph documents to Neo4j
graph.add_graph_documents(graph_documents_baml, baseEntityLabel=True, include_source=True)数据加载完成后，我们可以运行一些 Cypher 查询来了解新知识图谱的结构。首先，我们来看看文章长度（以 token 计）与从中提取的实体数量之间的关系。
# Import libraries for plotting and data analysis
import matplotlib.pyplot as plt
import seaborn as sns
# Query Neo4j to get the entity count and token count for each document
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"]
]
# Create a scatter plot with a regression line
sns.lmplot(
x="token_count", y="entity_count", data=entity_dist_df, line_kws={"color": "red"}
)
plt.title("Entity Count vs Token Count Distribution")
plt.xlabel("Token Count")
plt.ylabel("Entity Count")
plt.show()
该图显示出明显的正相关关系：随着文章中 token 数量的增加，我们提取的实体数量也呈增长趋势。这正是我们预期的结果，也证实了我们的提取过程是符合逻辑的。
接下来，我们看看节点度分布。这能告诉我们实体之间的连接程度。在真实世界的网络中，少数高度连接的节点（中心节点）是常见的。
import numpy as np
# Query for the degree of each entity node
degree_dist = graph.query(
"""
MATCH (e:__Entity__)
RETURN count {(e)-[:!MENTIONS]-()} AS node_degree
"""
)
degree_dist_df = pd.DataFrame.from_records(degree_dist)
# Calculate statistics
mean_degree = np.mean(degree_dist_df["node_degree"])
percentiles = np.percentile(degree_dist_df["node_degree"], [25, 50, 75, 90])
# Plot the histogram with a log scale
plt.figure(figsize=(12, 6))
sns.histplot(degree_dist_df["node_degree"], bins=50, kde=False, color="blue")
plt.yscale("log")
plt.title("Node Degree Distribution")
plt.legend()
plt.show()
直方图显示出一种“长尾”分布，这在知识图谱中很常见。大多数实体只有少数连接（度数较低），而少数实体是高度连接的中心节点。
例如，90分位数的度数为4，但最大度数达到了37。这表明像“美国”或“微软”这样的实体很可能在我们的图谱中充当关键枢纽节点。
为了找到语义上相似（即使名称不同）的实体，我们需要为它们创建向量嵌入。嵌入是文本的数值表示。我们将为每个实体的 id 和 description 生成嵌入，并将它们存储在图谱中。
我们将使用 Ollama 提供的 llama3 模型进行嵌入，并利用 LangChain 的 Neo4jVector 来实现嵌入管理。
from langchain_community.vectorstores import Neo4jVector
from langchain_ollama import OllamaEmbeddings
# Create embeddings using our local llama3 model
embeddings = OllamaEmbeddings(model="llama3")
# Initialize the Neo4jVector instance to manage embeddings in the graph
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近邻（kNN）**算法来查找向量空间中彼此接近的节点。这是一种识别潜在重复或高度相关实体（例如，“曼联”和“曼彻斯特联”）的强大方法。
我们将为此使用 Neo4j 的图数据科学（GDS）库。
# Import the GraphDataScience library
from graphdatascience import GraphDataScience
# --- GDS Client Initialization ---
# Initialize the GraphDataScience client to connect to the Neo4j database.
# It uses connection details (URI, username, password) from environment variables.
gds = GraphDataScience(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"]),
)
# Set the specific database for the GDS operations.
gds.set_database(os.environ["DATABASE"])
# --- In-Memory Graph Projection ---
# Project a graph into memory for efficient processing by GDS algorithms.
# This projection is named 'entities'.
G, result = gds.graph.project(
"entities", # Name of the in-memory graph
"__Entity__", # Node label to project
"*", # Project all relationship types
nodeProperties=["embedding"] # Include the 'embedding' property for nodes
)
# --- Similarity Calculation using kNN ---
# Define the similarity threshold for creating relationships.
similarity_threshold = 0.95
# Use the k-Nearest Neighbors (kNN) algorithm to find similar nodes.
# This 'mutates' the in-memory graph by adding new relationships.
gds.knn.mutate(
G, # The in-memory graph to modify
nodeProperties=["embedding"], # Property to use for similarity calculation
mutateRelationshipType="SIMILAR", # The type of relationship to create
mutateProperty="score", # The property on the new relationship to store the similarity score
similarityCutoff=similarity_threshold, # Threshold to filter relationships
)我们正在为嵌入相似度分数高于 0.95 的节点之间创建 SIMILAR 关系。
kNN 算法帮助我们找到了重复项的候选对象，但仅凭文本相似性并不完美。我们可以通过查找不仅语义相似而且名称也高度相似（“编辑距离”较小）的实体来进一步完善这一点。
我们将查询这些候选实体，然后使用大型语言模型（LLM）来最终决定是否合并它们。
# Query for potential duplicates based on community and name similarity
word_edit_distance = 3
potential_duplicate_candidates = graph.query(
# ... (full Cypher query from the notebook) ...
"""
MATCH (e:`__Entity__`)
WHERE size(e.id) > 4
WITH e.wcc AS community, collect(e) AS nodes, count(*) AS count
WHERE count > 1
# ... (rest of the complex query) ...
RETURN distinct(combinedResult)
""",
params={"distance": word_edit_distance},
)
# Let's look at a few candidates
potential_duplicate_candidates[:5]上述代码的输出如下。
OUTPUT
[{'combinedResult': ['David Van', 'Davidvan']},
{'combinedResult': ['Cyb003', 'Cyb004']},
{'combinedResult': ['Delta Air Lines', 'Delta_Air_Lines']},
{'combinedResult': ['Elon Musk', 'Elonmusk']},
{'combinedResult': ['Market', 'Markets']}]这些看起来是明显的重复项。现在，我们可以使用另一个 BAML 函数，让 LLM 决定保留哪个名称。完成此解析过程后，我们将在 Neo4j 中合并这些节点。
# (Assuming 'merged_entities' is created by the LLM resolution process)
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 算法，这是一种先进的社区检测算法。
# Project the full graph, weighting relationships by their frequency
G, result = gds.graph.project(
"communities",
"__Entity__",
{
"_ALL_": {
"type": "*",
"orientation": "UNDIRECTED",
"properties": {"weight": {"property": "*", "aggregation": "COUNT"}},
}
},
)
# Run Leiden community detection and write the results back to the nodes
gds.leiden.write(
G,
writeProperty="communities",
includeIntermediateCommunities=True, # This creates hierarchical communities
relationshipWeightProperty="weight",
)这会为每个实体节点添加一个 communities 属性，它是一个社区ID列表，包含不同粒度层级的社区（从小型、紧密结合的组到更大、更广泛的主题）。
最后，我们通过创建 __Community__ 节点并将它们连接起来，在图谱中实现这种层次结构。这创建了一个可浏览的主题结构。
# Create a uniqueness constraint for community nodes
graph.query("CREATE CONSTRAINT IF NOT EXISTS FOR (c:__Community__) REQUIRE c.id IS UNIQUE;")
# Create community nodes and link entities and communities together
graph.query(
"""
MATCH (e:`__Entity__`)
UNWIND range(0, size(e.communities) - 1 , 1) AS index
// ... (full community creation query from notebook) ...
RETURN count(*)
"""
)这个复杂的查询创建了一个多层次的社区结构，例如：(Entity)-[:IN_COMMUNITY]->(Level_0_Community)-[:IN_COMMUNITY]->(Level_1_Community)。
分析最终图谱结构
经过所有这些工作，我们的知识图谱结构如何？让我们来分析每个层级的社区规模。
# Query for the size of each community at each level
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
"""
)
# Printing the processed dataframe
percentiles_df这个表格在这里很重要。它向我们展示了 Leiden 算法是如何将我们的 1,875 个实体进行分组的。
• 在 0级，我们有 858 个小型、高度集中的社区。其中90%的社区成员数量在4个或更少。 • 当我们提升到 3级时，算法已将这些社区合并为 732 个更大、更普遍的社区。此级别中最大的社区现在包含 77 个实体。
这种分层结构恰好满足了高效 GraphRAG 的需求。我们现在可以在不同抽象级别执行检索。
结论
实验结果明确表明，虽然标准的 LangChain 工具提供了快速入门的方法，但当与小型、开源 LLM 结合使用时，它们可能显得脆弱且不可靠。
通过引入 BAML，我们解决了提示词过于复杂和严格的 JSON 解析等核心问题。结果是成功率从 25% 大幅提升到 99% 以上，将一个失败的实验转变为一个健壮且可扩展的知识图谱构建管道。
以下是我们所采取的关键步骤的快速回顾：
• 首先，我们准备了新闻文章数据集，并使用 Ollama 设置了本地 llama3模型。• 我们首次使用 LangChain 的 LLMGraphTransformer进行测试，其 75% 的失败率源于严格的 JSON 解析。• 尝试通过高级提示词工程修复此问题，但失败率仅略微改善至约 62%。 • 随后，我们集成了 BAML，利用其简化的模式和健壮的解析器，在图提取方面取得了 99.4% 的成功率。 • 高质量的图数据被加载到 Neo4j 中进行结构化和分析。 • 我们通过为所有实体生成向量嵌入来丰富图谱，以捕获其语义信息。 • 利用 k 近邻（kNN）算法，我们识别并连接了语义相似的节点。 • 我们进一步完善了图谱，使用 LLM 智能地查找并合并重复实体。 • 最后，我们应用 Leiden 算法将实体组织成多层次的社区结构，为高级 GraphRAG 奠定了基础。
这种结合 LangChain 强大编排能力和 BAML 可靠结构化输出的方法，是构建强大且经济高效 AI 应用程序的有效途径。
