使用 BAML 模糊解析改进 LangChain 知识图谱提取：成功率从25%提升到99%
在构建基于知识图谱的检索增强生成（RAG）系统或智能代理时，从非结构化数据中准确提取节点和关系是一项核心挑战。特别是在使用经过量化处理的小型本地大语言模型（LLM）时，这一问题尤为突出，往往导致整体系统性能显著下降。
LangChain 提取框架的主要限制在于其对严格 JSON 解析的依赖，即使采用大规模模型或精心设计的提示模板，解析失败的情况仍然频繁发生。
相比之下，BAML（Basically, A Made-up Language）采用模糊解析策略，能够在 LLM 输出格式不完全符合 JSON 标准的情况下仍然成功提取结构化数据。
本文将深入分析小型量化模型在 LangChain 提取任务中的性能限制，并展示 BAML 技术如何将知识图谱提取成功率从约 25% 显著提升至 99% 以上。
评估数据集的构建与初始化
为了系统性地分析问题并验证解决方案的有效性，本研究构建了一个标准化评估数据集，以便在多个测试场景中评估 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"
) 数据集结构相对简单，包含文章标题和正文内容。为了后续的性能分析和评估，需要添加一个新列来记录每篇新闻文章对应的 token 总数。
此处采用 OpenAI 的
tiktoken
库进行 token 计算，通过循环遍历的方式为整个数据集计算 token 数量。
# 导入 tiktoken 库以从文本计算令牌数
import tiktoken # 定义一个函数来计算给定字符串中特定模型的令牌数
def num_tokens_from_string(string: str, model: str = "gpt-4o") -> int: """返回文本字符串中的令牌数。""" # 获取指定模型的编码encoding = tiktoken.encoding_for_model(model) # 将字符串编码为令牌并计算它们 num_tokens = len(encoding.encode(string)) # 返回令牌总数return num_tokens # 在 DataFrame 中创建一个新列 'tokens'
# 它为每篇文章的 'title' 和 'text' 组合计算令牌数
news["tokens"] = [ num_tokens_from_string(f"{row['title']} {row['text']}") for i, row in news.iterrows() ]
数据集的 token 计算过程耗时较短，完成后得到包含 token 信息的增强数据集。
# 显示 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 将作为后台服务运行。在 macOS 和 Windows 系统中，应用程序会自动启动并在后台运行，用户可在菜单栏或系统托盘中看到相应图标。在 Linux 系统中，可能需要通过
systemctl start ollama
命令手动启动服务。
服务状态检查可通过终端或命令提示符执行以下命令：
# 检查可用模型ollama list #### OUTPUT #### [ ] <-- 尚无模型
如果服务正常运行但尚未安装模型，将显示空的模型列表，这在初始阶段是正常现象。若出现"命令未找到"错误，需要检查 Ollama 安装状态；若出现连接错误，则表明服务未正常启动。
模型下载使用 pull 命令完成。由于模型文件较大，此过程需要较长时间和充足的磁盘空间。
# 下载 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?" }' #### OUTPUT ####
{ "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
模型）。同时需要指定希望为节点和关系提取的附加属性信息。在本实验中，仅要求提取
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 篇文章均已处理完毕。
#### OUTPUT #### Processingdocuments: 100%|██████████|20/20 [01:32<00:00, 4.64s/it]
LangChain 框架的核心问题分析
实验结果的分析是理解问题本质的关键步骤。通过检查
graph_documents
列表的内容来评估处理效果。
# 显示图文档列表print(graph_documents)
输出结果揭示了严重的问题：
#### OUTPUT #### [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
列表。
这种现象表明，对于这些文章，LLM 要么生成了 LangChain 无法解析为有效图结构的输出，要么完全未能提取任何实体信息。
这正是小型量化 LLM 在结构化数据提取任务中面临的核心挑战。这些模型经常难以严格遵循
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}") #### OUTPUT #### Percentage missing: 75.0
75% 的失败率表明系统性能严重不足。在 20 篇文章的测试样本中，仅有 5 篇成功转换为知识图谱。
25% 的成功率远低于生产系统的可接受标准。
这一结果暴露了当前标准方法的根本缺陷：对于小型 LLM 固有的输出不确定性，现有框架过于僵化，缺乏必要的容错机制。
提示工程方法的探索与局限性分析
面对 75% 的高失败率，技术人员的直觉反应通常是通过提示工程来改善模型性能。理论上，更精确的指令应该带来更好的输出质量。然而，
LLMGraphTransformer
使用内置的默认提示，用户无法直接进行修改。
因此，本研究构建了基于 LangChain
ChatPromptTemplate
的自定义处理链，以获得对模型指令的完全控制权。通过更明确的指导，尝试引导模型始终生成符合要求的 JSON 格式输出。
实验采用 Pydantic 模型定义期望的输出结构，这是 LangChain 框架中处理结构化输出的标准范式。
# 导入 Pydantic 模型以定义数据结构
from langchain_core.pydantic_v1 import BaseModel, Field # 定义一个简单的 Node 结构
class Node(BaseModel): id: str = Field(description="节点的唯一标识符。") type: str = Field(description="节点的类型（例如，Person、Organization）。") # 定义一个简单的 Relationship 结构
class Relationship(BaseModel): source: Node = Field(description="关系的源节点。") target: Node = Field(description="关系的目标节点。") type: str = Field(description="关系的类型（例如，WORKS_FOR）。") # 定义整体图结构
class KnowledgeGraph(BaseModel): nodes: List[Node] = Field(description="图中的节点列表。") relationships: List[Relationship] = Field(description="图中的关系列表。")
接下来构建详细的提示模板，明确包含从 Pydantic 模型生成的 JSON 模式，并为 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} 这是输入文本：
--------------------
{text}
--------------------
""" prompt = ChatPromptTemplate.from_template( template, partial_variables={"schema": parser.get_format_instructions()},
) # 创建完整的提取链chain = prompt | llm | parser
改进后的处理链相比
LLMGraphTransformer
具有更高的明确性。模型接收到详细的数据模式和清晰的操作指令。重新运行 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: # 如果 LLM 输出不是有效的 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: 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)}") #### OUTPUT ####
Percentage missing with improved prompt: 62.0% Number of JSON parsing errors: 13
结果显示失败率约为 62%。虽然相比初始的 75% 有所改善，但仍远未达到可靠性要求。在 20 篇文章中，仍有 13 篇无法成功提取图结构。
JsonOutputParser
持续抛出异常，表明
llama3
模型尽管接受了优化的提示，仍然产生格式不正确的 JSON 输出。
这一实验结果揭示了一个基本技术限制：
单纯的提示工程无法从根本上解决小型 LLM 结构化输出不一致的问题。
既然改进提示不是有效解决方案，那么需要寻找既能要求高质量输出，又能智能处理模型不完美输出的技术方案。这正是 BAML 技术设计要解决的核心问题。
在后续章节中，将使用 BAML 驱动的实现替换现有处理链，并分析其带来的性能改进。
BAML 技术架构与核心优势
前期实验证明，即使通过精心设计的提示工程，依赖严格 JSON 解析的小型 LLM 仍然表现不佳。虽然这些模型具有强大的理解能力，但在格式化输出方面存在固有缺陷。
BAML（Basically, A Made-up Language）技术在此背景下展现出重要价值。该技术提供两个直接针对现有问题的关键优势：首先是简化的模式定义，BAML 摒弃了冗长的 JSON 模式，采用简洁的类 TypeScript 语法定义数据结构，这种设计既便于人类理解，也降低了 LLM 的理解难度，同时减少了 token 使用量和产生歧义的风险。其次是强大的解析能力，BAML 客户端配备了"模糊"或"模式对齐"解析器，该解析器不要求完美的 JSON 格式，能够处理常见的 LLM 输出错误（如尾随逗号、缺失引号或冗余文本），仍然能够成功提取所需的结构化数据。
环境配置首先需要安装 BAML 客户端及其 VS Code 扩展。
# 安装 baml 客户端pip install baml-py
在 VS Code 扩展市场搜索
baml
并安装相应扩展。该扩展提供了交互式测试环境，允许用户在无需运行 Python 代码的情况下测试提示和数据模式。
接下来在
.baml
文件中定义图提取逻辑。该文件可视为 LLM 调用的配置文件。创建名为
extract_graph.baml
的配置文件：
// 定义图中的节点，具有 ID、类型和可选属性
class SimpleNode { id string // 节点的唯一标识符type string // 节点的类型/类别properties Properties // 与节点关联的附加属性
} // 定义节点或关系的可选属性结构
class Properties { description string? // 可选的文本描述
} // 定义两个节点之间的关系
class SimpleRelationship { source_node_id string // 源节点的 IDsource_node_type string // 源节点的类型target_node_id string // 目标节点的 IDtarget_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 提取图}
class
定义语法简洁直观。
function ExtractGraph
指定使用
Ollama
客户端并提供 Jinja 风格的提示模板。特殊变量
{{ ctx.output_format }}
是 BAML 自动注入简化模式定义的位置。
BAML 与 LangChain 框架的集成实现
将 BAML 功能集成到 LangChain 工作流需要构建一系列辅助函数，用于将 BAML 输出转换为 LangChain 和 Neo4j 兼容的
GraphDocument
格式。
# 导入必要的库
from typing import Any, 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() if isinstance(el.id, str) else el.id, type=el.type.capitalize() if el.type else None, properties=el.properties ) for el in nodes ] # 将 BAML 的关系输出映射到 Langchain 的 Relationship 对象的辅助函数
def map_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.type and 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
async def get_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 负责处理复杂的模式注入，自定义提示设计相对简化。
# 导入提示模板
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 # 处理单个文档的异步函数
async def aprocess_response(document: Document) -> GraphDocument: # 调用我们的 BAML 链resp = await chain.ainvoke({"input": document.page_content}) # 将响应格式化为 GraphDocumentreturn GraphDocument( nodes=_format_nodes(resp.nodes), relationships=_format_relationships(resp.relationships), source=document, ) # 处理文档列表的异步函数
async def aconvert_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 # 处理原始文本的异步函数
async def aprocess_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
完成原始文本到文档的转换并执行图数据提取。
主要处理流程的实现：
# 初始化一个空列表来存储生成的图文档。
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_size = 4 # 分块迭代文章，使用 tqdm 显示进度条。
for i in tqdm(range(0, len(titles), chunk_size), desc="Processing Chunks with BAML"): # 获取当前块的标题。title_chunk = titles[i : i + chunk_size] # 获取当前块的文本。text_chunk = texts[i : i + chunk_size] # 将块中每篇文章的标题和文本合并为单个字符串。combined_docs = [f"{title} {text}" for title, text in zip(title_chunk, text_chunk)] try: # 异步处理合并的文档以提取图结构。docs = await aprocess_text(combined_docs) # 将处理的图文档添加到主列表中。graph_documents_baml.extend(docs) except Exception as e: # 处理处理过程中发生的任何错误并打印错误消息。print(f"Error processing chunk starting at index {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}%") #### OUTPUT #### Percentage missing with BAML: 0.5813953488372093%
实验结果令人瞩目。失败率从 75% 大幅下降至仅 0.58%，这意味着成功率达到了 99.4%！
通过简单地将僵化的
LLMGraphTransformer
替换为 BAML 驱动的处理链，系统从实验室原型转变为具备生产部署能力的稳健管道。
这一成果证明了瓶颈并非小型 LLM 的任务理解能力不足，而是系统对完美 JSON 格式的苛刻要求导致的脆弱性。
基于 Neo4j 的知识图谱深度分析
单纯的实体提取并不足以发挥知识图谱的全部潜力。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__
标签，便于后续查询操作。
# 将图文档添加到 Neo4jgraph.add_graph_documents(graph_documents_baml, baseEntityLabel=True, include_source=True)
数据导入完成后，运行 Cypher 查询来分析新构建知识图谱的结构特征。首先分析文章长度（以 token 为单位）与提取实体数量之间的关系。
# 导入用于绘图和数据分析的库
import matplotlib.pyplot as plt
import seaborn as sns # 查询 Neo4j 以获取每个文档的实体计数和令牌计数
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("Entity Count vs Token Count Distribution")
plt.xlabel("Token Count")
plt.ylabel("Entity Count") plt.show()
图表显示了明显的正相关关系：随着文章 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("Node Degree Distribution")
plt.legend() plt.show()
直方图呈现典型的"长尾"分布，这是知识图谱的标准特征。大多数实体具有较少的连接（低度值），而少数实体作为高连接度的枢纽存在。
具体分析显示，90% 的节点度数为 4，但最大度数达到 37。这表明诸如"USA"或"Microsoft"等实体可能在图中扮演中心节点的角色。
为了识别语义相似的实体（即使名称不同），需要为实体创建向量嵌入。嵌入是文本片段的数值表示形式。本研究为每个实体的
id
和
description
生成嵌入并存储在图数据库中。
采用 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-最近邻（kNN）**算法识别向量空间中相互接近的节点。这是发现潜在重复或高度相关实体（例如"Man United"和"Manchester United"）的有效方法。
采用 Neo4j 的图数据科学（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-最近邻（kNN）算法查找相似节点。
# 这会通过添加新关系来"变异"内存图。
gds.knn.mutate( G, # 要修改的内存图nodeProperties=["embedding"], # 用于相似性计算的属性mutateRelationshipType="SIMILAR", # 要创建的关系类型mutateProperty="score", # 新关系上存储相似性得分的属性similarityCutoff=similarity_threshold, # 过滤关系的阈值)
系统在嵌入相似性得分超过 0.95 的节点之间建立
SIMILAR
关系。
kNN 算法识别出重复候选实体，但仅依靠文本相似性并不完美。可以通过寻找既语义相似又具有相近名称（低编辑距离）的实体来进一步优化结果。
查询这些候选实体，然后使用 LLM 做出最终的合并决策。
# 基于社区和名称相似性查询潜在重复项
word_edit_distance = 3
potential_duplicate_candidates = graph.query( # ... （来自笔记本的完整 Cypher 查询）...""" MATCH (e:`__Entity__`) WHERE size(e.id) > 4 WITH e.wcc AS community, collect(e) AS nodes, count(*) AS count WHERE count > 1 # ... （查询的其余复杂部分）...RETURN distinct(combinedResult) """, params={"distance": word_edit_distance},
) # 让我们看看几个候选者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 中合并这些节点。
# （假设 '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 算法进行社区检测。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, # 这创建了分层社区relationshipWeightProperty="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 """
) # 打印处理的数据框percentiles_df
这一统计表格具有重要意义，展示了 Leiden 算法如何组织我们的 1,875 个实体。
在 Level 0 层级，系统识别出 858 个小规模、高内聚的社区，其中 90% 的社区包含 4 个或更少的成员。随着层级上升到 Level 3，算法将这些细粒度社区合并为 732 个规模更大、覆盖面更广的社区。该层级的最大社区包含 77 个实体。
这种分层结构正是有效实施 GraphRAG 所需的理想架构。系统现在能够在不同抽象层级上执行信息检索操作。
总结
实验结果清晰地证明了技术方案的有效性。传统的 LangChain 工具虽然提供了快速入门的便利，但在与小型开源 LLM 协同工作时表现出脆弱性和不可靠性。
通过引入 BAML 技术，本研究成功解决了过度复杂的提示设计和严格 JSON 解析带来的核心问题。最终成功率从 25% 显著提升至 99% 以上，将实验室概念验证转化为稳健且可扩展的知识图谱构建管道。
关键技术步骤回顾包括：首先准备新闻文章数据集并使用 Ollama 配置本地 llama3 模型。随后使用 LangChain 的 LLMGraphTransformer 进行初步测试，发现由于严格 JSON 解析导致 75% 的失败率。尝试通过高级提示工程改善问题，仅将失败率降低至约 62%。接着集成 BAML 技术，利用其简化模式和强大解析器实现 99.4% 的图提取成功率。将高质量图数据导入 Neo4j 进行结构化分析和处理。通过为所有实体生成向量嵌入来增强图的语义表示能力。使用 k-最近邻算法识别并关联语义相似的节点。通过 LLM 智能识别和合并重复实体来进一步优化图质量。最终应用 Leiden 算法将实体组织成多层社区层次结构，为高级 GraphRAG 应用奠定基础。
这种结合 LangChain 强大编排能力和 BAML 可靠结构化输出的技术方案，为构建高性能且经济高效的 AI 应用系统提供了成功的实践范例。
代码：https://avoid.overfit.cn/post/9b87950cd66c41a4b97fa0a44ed9d0c5
作者：Fareed Khan