API 文档

组件

Component

class neo4j_graphrag.experimental.pipeline.component.Component[source]

所有组件都需要实现的接口。

async run(*args, **kwargs)[source]

运行组件并返回其结果。

注意：如果实现了 run_with_context，则此方法将不被使用。

参数:
返回类型:

DataModel

async run_with_context(context_, *args, **kwargs)[source]

此方法由管道编排器调用。context_ 参数包含有关管道运行的信息：run_id 和一个可用于从组件向管道回调发送事件的 notify 函数。

此功能将在未来版本中移至 run 方法。

它默认调用 run 方法以防止任何破坏性更改。

参数:
返回类型:

DataModel

DataLoader

class neo4j_graphrag.experimental.components.pdf_loader.DataLoader[source]

用于加载各种输入类型数据的接口。

get_document_metadata(text, metadata=None)[source]
参数:
返回类型:

Dict[str, str] | None

abstract async run(filepath, metadata=None)[source]

运行组件并返回其结果。

注意：如果实现了 run_with_context，则此方法将不被使用。

参数:
返回类型:

PdfDocument

PdfLoader

class neo4j_graphrag.experimental.components.pdf_loader.PdfLoader[source]
static load_file(file, fs)[source]

解析 PDF 文件并返回文本。

参数:

  • file (str)

  • fs (AbstractFileSystem)

返回类型:

str

async run(filepath, metadata=None, fs=None)[source]

运行组件并返回其结果。

注意：如果实现了 run_with_context，则此方法将不被使用。

参数:
返回类型:

PdfDocument

TextSplitter

class neo4j_graphrag.experimental.components.text_splitters.base.TextSplitter[source]

文本分割器的接口。

abstract async run(text)[source]

将一段文本分割成块。

参数:

text (str) – 要分割的文本。

返回:

块列表。

返回类型:

TextChunks

FixedSizeSplitter

class neo4j_graphrag.experimental.components.text_splitters.fixed_size_splitter.FixedSizeSplitter(chunk_size=4000, chunk_overlap=200, approximate=True)[source]
文本分割器，将输入文本分割成固定或近似固定大小的块，并可选择重叠。

chunk_size (int) – 每个块的字符数。

参数:

  • chunk_overlap (int) – 每个块与前一个块重叠的字符数。必须小于 chunk_size。

  • approximate (bool) – 如果为 True，则避免在块边界处将单词从中间分割。默认为 True。

  • 示例

async run(text)[source]

from neo4j_graphrag.experimental.components.text_splitters.fixed_size_splitter import FixedSizeSplitter
from neo4j_graphrag.experimental.pipeline import Pipeline

pipeline = Pipeline()
text_splitter = FixedSizeSplitter(chunk_size=4000, chunk_overlap=200, approximate=True)
pipeline.add_component(text_splitter, "text_splitter")
async run(text)[source]

将一段文本分割成块。

参数:

text (str) – 要分割的文本。

返回:

块列表。

返回类型:

TextChunks

LangChainTextSplitterAdapter

class neo4j_graphrag.experimental.components.text_splitters.langchain.LangChainTextSplitterAdapter(text_splitter)[source]

LangChain TextSplitter 的适配器。允许此类实例用于知识图谱构建器管道。

参数:

text_splitter (LangChainTextSplitter) – LangChain TextSplitter 类的一个实例。

async run(text)[source]

from langchain_text_splitters import RecursiveCharacterTextSplitter
from neo4j_graphrag.experimental.components.text_splitters.langchain import LangChainTextSplitterAdapter
from neo4j_graphrag.experimental.pipeline import Pipeline

pipeline = Pipeline()
text_splitter = LangChainTextSplitterAdapter(RecursiveCharacterTextSplitter())
pipeline.add_component(text_splitter, "text_splitter")
async run(text)[source]

将文本分割成块。

参数:

text (str) – 要分割的文本。

返回:

将文本分割成块。

返回类型:

TextChunks

LlamaIndexTextSplitterAdapter

class neo4j_graphrag.experimental.components.text_splitters.llamaindex.LlamaIndexTextSplitterAdapter(text_splitter)[source]

LlamaIndex TextSplitter 的适配器。允许此类实例用于知识图谱构建器管道。

参数:

text_splitter (LlamaIndexTextSplitter) – LlamaIndex TextSplitter 类的一个实例。

async run(text)[source]

from llama_index.core.node_parser.text.sentence import SentenceSplitter
from neo4j_graphrag.experimental.components.text_splitters.llamaindex import (
    LlamaIndexTextSplitterAdapter,
)
from neo4j_graphrag.experimental.pipeline import Pipeline

pipeline = Pipeline()
text_splitter = LlamaIndexTextSplitterAdapter(SentenceSplitter())
pipeline.add_component(text_splitter, "text_splitter")
async run(text)[source]

将文本分割成块。

参数:

text (str) – 要分割的文本。

返回:

将文本分割成块。

返回类型:

TextChunks

TextChunkEmbedder

class neo4j_graphrag.experimental.components.embedder.TextChunkEmbedder(embedder)[source]

用于从文本块创建嵌入的组件。

参数:

embedder (Embedder) – 用于创建嵌入的嵌入器。

async run(text)[source]

from neo4j_graphrag.experimental.components.embedder import TextChunkEmbedder
from neo4j_graphrag.embeddings.openai import OpenAIEmbeddings
from neo4j_graphrag.experimental.pipeline import Pipeline

embedder = OpenAIEmbeddings(model="text-embedding-3-large")
chunk_embedder = TextChunkEmbedder(embedder)
pipeline = Pipeline()
pipeline.add_component(chunk_embedder, "chunk_embedder")
async run(text_chunks)[source]

嵌入文本块列表。

参数:

text_chunks (TextChunks) – 要嵌入的文本块。

返回:

输入文本块，每个块都有一个添加的嵌入。

返回类型:

TextChunks

LexicalGraphBuilder

class neo4j_graphrag.experimental.components.lexical_graph.LexicalGraphBuilder(config=LexicalGraphConfig(id_prefix='', document_node_label='Document', chunk_node_label='Chunk', chunk_to_document_relationship_type='FROM_DOCUMENT', next_chunk_relationship_type='NEXT_CHUNK', node_to_chunk_relationship_type='FROM_CHUNK', chunk_id_property='id', chunk_index_property='index', chunk_text_property='text', chunk_embedding_property='embedding'))[source]

构建要插入 Neo4j 的词法图。词法图包含： - 每个文档一个节点 - 每个块一个节点 - 每个块与其来源文档之间的关系 - 文档中每个块与下一个块之间的关系

参数:

config (LexicalGraphConfig)

async run(text_chunks, document_info=None)[source]

运行组件并返回其结果。

注意：如果实现了 run_with_context，则此方法将不被使用。

参数:
返回类型:

GraphResult

async process_chunk(graph, chunk, next_chunk, document_info=None)[source]

添加块及其之间的关系 (NEXT_CHUNK)

就地更新 graph。

参数:
返回类型:

None

create_document_node(document_info)[source]

创建具有“path”属性的 Document 节点。任何文档元数据也作为节点属性添加。

参数:

document_info (DocumentInfo)

返回类型:

Neo4jNode

create_chunk_node(chunk)[source]

创建带有属性“text”、“index”和在处理过程中添加的任何“metadata”的块节点。对于作为 embedding_property 添加的潜在块嵌入属性的特殊情况。

参数:

chunk (TextChunk)

返回类型:

Neo4jNode

create_chunk_to_document_rel(chunk, document_info)[source]

创建块与其所属文档之间的关系。

参数:
返回类型:

Neo4jRelationship

create_next_chunk_relationship(chunk, next_chunk)[source]

创建块与下一个块之间的关系

参数:
返回类型:

Neo4jRelationship

create_node_to_chunk_rel(node, chunk_id)[source]

创建块与该块中找到的实体之间的关系

参数:
返回类型:

Neo4jRelationship

async process_chunk_extracted_entities(chunk_graph, chunk)[source]

创建块与从其中提取的每个实体之间的关系。

就地更新 chunk_graph。

参数:
返回类型:

None

Neo4jChunkReader

class neo4j_graphrag.experimental.components.neo4j_reader.Neo4jChunkReader(driver, fetch_embeddings=False, neo4j_database=None)[source]

从 Neo4j 数据库读取文本块。

参数:

  • driver (neo4j.driver) – 用于连接数据库的 Neo4j 驱动程序。

  • fetch_embeddings (bool) – 如果为 True，则还会返回嵌入属性。默认为 False。

  • neo4j_database (Optional[str]) – Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.experimental.components.neo4j_reader import Neo4jChunkReader

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")
DATABASE = "neo4j"

driver = GraphDatabase.driver(URI, auth=AUTH)
reader = Neo4jChunkReader(driver=driver, neo4j_database=DATABASE)
await reader.run()
async run(lexical_graph_config=LexicalGraphConfig(id_prefix='', document_node_label='Document', chunk_node_label='Chunk', chunk_to_document_relationship_type='FROM_DOCUMENT', next_chunk_relationship_type='NEXT_CHUNK', node_to_chunk_relationship_type='FROM_CHUNK', chunk_id_property='id', chunk_index_property='index', chunk_text_property='text', chunk_embedding_property='embedding'))[source]

从 Neo4j 数据库读取文本块。

参数:

lexical_graph_config (LexicalGraphConfig) – 词法图的节点标签和关系类型。

返回类型:

TextChunks

SchemaBuilder

class neo4j_graphrag.experimental.components.schema.SchemaBuilder[source]

一个构建器类，用于从给定实体、关系及其在潜在模式中定义的相互关系构建 SchemaConfig 对象。

async run(text)[source]

from neo4j_graphrag.experimental.components.schema import (
    SchemaBuilder,
    SchemaEntity,
    SchemaProperty,
    SchemaRelation,
)
from neo4j_graphrag.experimental.pipeline import Pipeline

entities = [
    SchemaEntity(
        label="PERSON",
        description="An individual human being.",
        properties=[
            SchemaProperty(
                name="name", type="STRING", description="The name of the person"
            )
        ],
    ),
    SchemaEntity(
        label="ORGANIZATION",
        description="A structured group of people with a common purpose.",
        properties=[
            SchemaProperty(
                name="name", type="STRING", description="The name of the organization"
            )
        ],
    ),
]
relations = [
    SchemaRelation(
        label="EMPLOYED_BY", description="Indicates employment relationship."
    ),
]
potential_schema = [
    ("PERSON", "EMPLOYED_BY", "ORGANIZATION"),
]
pipe = Pipeline()
schema_builder = SchemaBuilder()
pipe.add_component(schema_builder, "schema_builder")
pipe_inputs = {
    "schema": {
        "entities": entities,
        "relations": relations,
        "potential_schema": potential_schema,
    },
    ...
}
pipe.run(pipe_inputs)
async run(entities, relations=None, potential_schema=None)[source]

异步构建并返回 SchemaConfig 对象。

参数:

  • entities (List[SchemaEntity]) – 实体对象列表。

  • relations (List[SchemaRelation]) – 关系对象列表。

  • potential_schema (Dict[str, List[str]]) – 将实体名称映射到关系名称列表的字典。

返回:

一个配置好的模式对象，异步构建。

返回类型:

SchemaConfig

EntityRelationExtractor

class neo4j_graphrag.experimental.components.entity_relation_extractor.EntityRelationExtractor(*args, on_error=OnError.IGNORE, create_lexical_graph=True, **kwargs)[source]

实体关系提取组件的抽象类。

参数:

  • on_error (OnError) – 提取过程中发生错误时要执行的操作。默认为抛出错误。

  • create_lexical_graph (bool) – 除了提取的实体和关系外，是否将文本块包含在图中。默认为 True。

  • args (Any)

  • kwargs (Any)

async run(chunks, document_info=None, lexical_graph_config=None, **kwargs)[source]

运行组件并返回其结果。

注意：如果实现了 run_with_context，则此方法将不被使用。

参数:
返回类型:

Neo4jGraph

update_ids(graph, chunk)[source]

通过为节点 ID 添加唯一前缀，使它们在块、文档和管道运行中保持唯一。

参数:
返回类型:

Neo4jGraph

LLMEntityRelationExtractor

class neo4j_graphrag.experimental.components.entity_relation_extractor.LLMEntityRelationExtractor(llm, prompt_template=<neo4j_graphrag.generation.prompts.ERExtractionTemplate object>, create_lexical_graph=True, enforce_schema=SchemaEnforcementMode.NONE, on_error=OnError.RAISE, max_concurrency=5)[source]

使用大型语言模型从一系列文本块中提取知识图谱。

参数:

  • llm (LLMInterface) – 用于提取的语言模型。

  • prompt_template (ERExtractionTemplate | str) – 用于提取的自定义提示模板。

  • create_lexical_graph (bool) – 除了提取的实体和关系外，是否将文本块包含在图中。默认为 True。

  • enforce_schema (SchemaEnforcementMode) – 是否根据提供的模式验证提取的实体/关系。默认为 None。

  • on_error (OnError) – 提取过程中发生错误时要执行的操作。默认为抛出错误。

  • max_concurrency (int) – 用于向 LLM 发出请求的最大并发任务数。

async run(text)[source]

from neo4j_graphrag.experimental.components.entity_relation_extractor import LLMEntityRelationExtractor
from neo4j_graphrag.llm import OpenAILLM
from neo4j_graphrag.experimental.pipeline import Pipeline

llm = OpenAILLM(model_name="gpt-4o", model_params={"temperature": 0, "response_format": {"type": "object"}})

extractor = LLMEntityRelationExtractor(llm=llm)
pipe = Pipeline()
pipe.add_component(extractor, "extractor")
async run(chunks, document_info=None, lexical_graph_config=None, schema=None, examples='', **kwargs)[source]

对列表中的所有块执行实体和关系提取。

可选地，通过向返回的图中添加节点和关系以表示文档及其块来创建“词法图”（有关更多详细信息，请参见词法图构建器文档和用户指南）

参数:

  • chunks (TextChunks) – 要从中提取实体和关系的文本块列表。

  • document_info (Optional[DocumentInfo], optional) – 块来源的文档。用于词法图创建步骤。

  • lexical_graph_config (Optional[LexicalGraphConfig], optional) – 词法图配置，用于自定义词法图中的节点标签和关系类型。

  • schema (SchemaConfig | None) – 用于指导 LLM 提取的模式定义。

  • examples (str) – 提示中用于 Few-Shot Learning 的示例。

  • kwargs (Any)

返回类型:

Neo4jGraph

KGWriter

class neo4j_graphrag.experimental.components.kg_writer.KGWriter[source]

用于将知识图谱写入数据存储的抽象类。

abstract async run(graph, lexical_graph_config=LexicalGraphConfig(id_prefix='', document_node_label='Document', chunk_node_label='Chunk', chunk_to_document_relationship_type='FROM_DOCUMENT', next_chunk_relationship_type='NEXT_CHUNK', node_to_chunk_relationship_type='FROM_CHUNK', chunk_id_property='id', chunk_index_property='index', chunk_text_property='text', chunk_embedding_property='embedding'))[source]

将图写入数据存储。

参数:

  • graph (Neo4jGraph) – 要写入数据存储的知识图谱。

  • lexical_graph_config (LexicalGraphConfig) – 词法图中的节点标签和关系类型。

返回类型:

KGWriterModel

Neo4jWriter

class neo4j_graphrag.experimental.components.kg_writer.Neo4jWriter(driver, neo4j_database=None, batch_size=1000)[source]

将知识图谱写入 Neo4j 数据库。

参数:

  • driver (neo4j.driver) – 用于连接数据库的 Neo4j 驱动程序。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

  • batch_size (int) – 每次批处理写入数据库的节点或关系数量。默认为 1000。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.experimental.components.kg_writer import Neo4jWriter
from neo4j_graphrag.experimental.pipeline import Pipeline

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")
DATABASE = "neo4j"

driver = GraphDatabase.driver(URI, auth=AUTH)
writer = Neo4jWriter(driver=driver, neo4j_database=DATABASE)

pipeline = Pipeline()
pipeline.add_component(writer, "writer")
async run(graph, lexical_graph_config=LexicalGraphConfig(id_prefix='', document_node_label='Document', chunk_node_label='Chunk', chunk_to_document_relationship_type='FROM_DOCUMENT', next_chunk_relationship_type='NEXT_CHUNK', node_to_chunk_relationship_type='FROM_CHUNK', chunk_id_property='id', chunk_index_property='index', chunk_text_property='text', chunk_embedding_property='embedding'))[source]

将知识图谱插入 Neo4j 数据库。

参数:

  • graph (Neo4jGraph) – 要插入数据库的知识图谱。

  • lexical_graph_config (LexicalGraphConfig) – 词法图的节点标签和关系类型。

返回类型:

KGWriterModel

SinglePropertyExactMatchResolver

class neo4j_graphrag.experimental.components.resolver.SinglePropertyExactMatchResolver(driver, filter_query=None, resolve_property='name', neo4j_database=None)[source]

解析具有相同标签和完全相同属性（默认为“name”）的实体。

参数:

  • driver (neo4j.Driver) – 用于连接数据库的 Neo4j 驱动程序。

  • filter_query (Optional[str]) – 为了缩小解析范围，添加一个 Cypher WHERE 子句。

  • resolve_property (str) – 将被比较的属性（默认值：“name”）。如果值完全匹配，则合并实体。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.experimental.components.resolver import SinglePropertyExactMatchResolver

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")
DATABASE = "neo4j"

driver = GraphDatabase.driver(URI, auth=AUTH)
resolver = SinglePropertyExactMatchResolver(driver=driver, neo4j_database=DATABASE)
await resolver.run()  # no expected parameters
async run()[source]

基于以下规则解析实体：对于每个实体标签，具有相同“resolve_property”值（精确匹配）的实体将被分组到一个节点中。

  • 属性：如果已设置，第一个节点的属性将保留，否则列表中的第一个属性将被写入。

  • 关系：合并具有相同类型和目标节点的关系。

有关更多详细信息，请参见 apoc.refactor.mergeNodes 文档。

返回类型:

ResolutionStats

SpaCySemanticMatchResolver

class neo4j_graphrag.experimental.components.resolver.SpaCySemanticMatchResolver(driver, filter_query=None, resolve_properties=None, similarity_threshold=0.8, spacy_model='en_core_web_lg', neo4j_database=None)[source]

根据 spaCy 的静态嵌入和余弦相似度，解析具有相同标签和相似文本属性集（默认为 [“name”]）的实体。

参数:

  • driver (neo4j.Driver) – 用于连接数据库的 Neo4j 驱动程序。

  • filter_query (Optional[str]) – 可选的 Cypher WHERE 子句以缩小解析范围。

  • resolve_properties (Optional[List[str]]) – 考虑用于嵌入的属性列表。默认为 [“name”]。

  • similarity_threshold (float) – 节点合并的相似度阈值。默认为 0.8。

  • spacy_model (str) – 要加载的 spaCy 模型的名称。默认为“en_core_web_lg”。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.experimental.components.resolver import SpaCySemanticMatchResolver

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")
DATABASE = "neo4j"

driver = GraphDatabase.driver(URI, auth=AUTH)
resolver = SpaCySemanticMatchResolver(driver=driver, neo4j_database=DATABASE)
await resolver.run()  # no expected parameters
async run()[source]

运行组件并返回其结果。

注意：如果实现了 run_with_context，则此方法将不被使用。

返回类型:

ResolutionStats

FuzzyMatchResolver

class neo4j_graphrag.experimental.components.resolver.FuzzyMatchResolver(driver, filter_query=None, resolve_properties=None, similarity_threshold=0.8, neo4j_database=None)[source]

使用 RapidFuzz 进行模糊匹配，解析具有相同标签和相似文本属性集的实体。相似度分数被归一化为 0 到 1 之间。

参数:

  • driver (Driver)

  • filter_query (str | None)

  • resolve_properties (List[str] | None)

  • similarity_threshold (float)

  • neo4j_database (str | None)

async run()[source]

运行组件并返回其结果。

注意：如果实现了 run_with_context，则此方法将不被使用。

返回类型:

ResolutionStats

管道

Pipeline

class neo4j_graphrag.experimental.pipeline.Pipeline(store=None, callback=None)[source]

这是主管道，其中定义了组件及其执行顺序。

参数:
draw(path, layout='dot', hide_unused_outputs=True)[source]

将管道图渲染到指定路径的 HTML 文件。

参数:

  • path (str)

  • layout (str)

  • hide_unused_outputs (bool)

返回类型:

Any

add_component(component, name)[source]

添加新组件。组件由其名称唯一标识。如果“name”已存在于管道中，则会引发 ValueError。

参数:
返回类型:

None

connect(start_component_name, end_component_name, input_config=None)[source]

将一个组件连接到另一个组件。

参数:

  • start_component_name (str) – 在 add_component 方法中定义的组件名称

  • end_component_name (str) – 在 add_component 方法中定义的组件名称

  • input_config (Optional[dict[str, str]]) – 结束组件输入配置：传播以前的组件输出。

引发:

PipelineDefinitionError – 如果提供的组件不在管道中，或者此连接将创建的图是循环的。

返回类型:

None

async run(data)[source]
参数:

data (dict[str, Any])

返回类型:

PipelineResult

SimpleKGPipeline

class neo4j_graphrag.experimental.pipeline.kg_builder.SimpleKGPipeline(llm, driver, embedder, entities=None, relations=None, potential_schema=None, enforce_schema='NONE', from_pdf=True, text_splitter=None, pdf_loader=None, kg_writer=None, on_error='IGNORE', prompt_template=<neo4j_graphrag.generation.prompts.ERExtractionTemplate object>, perform_entity_resolution=True, lexical_graph_config=None, neo4j_database=None)[source]

一个简化从文本文档构建知识图谱过程的类。它抽象了设置管道及其组件的复杂性。

参数:

  • llm (LLMInterface) – 用于实体和关系提取的 LLM 实例。

  • driver (neo4j.Driver) – 用于数据库连接的 Neo4j 驱动程序实例。

  • embedder (Embedder) – 用于从文本块生成块嵌入的嵌入器实例。

  • entities (Optional[List[Union[str, dict[str, str], SchemaEntity]]]) –

    一个列表，包含以下类型：

    • str: 实体标签

    • dict: 遵循 SchemaEntity 模式，即包含 label、description 和 properties 键

  • relations (Optional[List[Union[str, dict[str, str], SchemaRelation]]]) –

    一个列表，包含以下类型：

    • str: 关系标签

    • dict: 遵循 SchemaRelation 模式，即包含 label、description 和 properties 键

  • potential_schema (Optional[List[tuple]]) – 潜在模式关系列表。

  • enforce_schema (str) – 对提取的实体/关系根据提供的模式进行验证。默认为“NONE”，在这种情况下，即使提供了模式，模式强制执行也会被忽略。可能的值有“None”或“STRICT”。

  • from_pdf (bool) – 确定是否在管道中包含 PdfLoader。如果为 True，则在 run 方法中需要 file_path 输入。如果为 False，则在 run 方法中需要 text 输入。

  • text_splitter (Optional[TextSplitter]) – 文本分割组件。默认为 FixedSizeSplitter()。

  • pdf_loader (Optional[DataLoader]) – PDF 加载组件。默认为 PdfLoader()。

  • kg_writer (Optional[KGWriter]) – 知识图谱写入组件。默认为 Neo4jWriter()。

  • on_error (str) – 实体和关系提取器的错误处理策略。默认为“IGNORE”，如果提取失败，则忽略块。可能的值：“RAISE”或“IGNORE”。

  • perform_entity_resolution (bool) – 合并具有相同标签和名称的实体。默认值：True

  • prompt_template (str) – 用于提取的自定义提示模板。

  • lexical_graph_config (Optional[LexicalGraphConfig], optional) – 词法图配置，用于自定义词法图中的节点标签和关系类型。

  • neo4j_database (Optional[str])

async run_async(file_path=None, text=None)[source]

异步运行知识图构建过程。

参数:

  • file_path (Optional[str]) – 要处理的 PDF 文件的路径。如果 from_pdf 为 True，则为必填项。

  • text (Optional[str]) – 要处理的文本内容。如果 from_pdf 为 False，则为必填项。

返回:

流水线执行的结果。

返回类型:

PipelineResult

配置文件

SimpleKGPipelineConfig

class neo4j_graphrag.experimental.pipeline.config.template_pipeline.simple_kg_builder.SimpleKGPipelineConfig(*, neo4j_config={}, llm_config={}, embedder_config={}, extras={}, template_=PipelineType.SIMPLE_KG_PIPELINE, from_pdf=False, entities=[], relations=[], potential_schema=None, enforce_schema=SchemaEnforcementMode.NONE, on_error=OnError.IGNORE, prompt_template=<neo4j_graphrag.generation.prompts.ERExtractionTemplate object>, perform_entity_resolution=True, lexical_graph_config=None, neo4j_database=None, pdf_loader=None, kg_writer=None, text_splitter=None)[source]
参数:

PipelineRunner

class neo4j_graphrag.experimental.pipeline.config.runner.PipelineRunner(pipeline_definition, config=None, do_cleaning=False)[source]

流水线运行器从不同的对象构建流水线，并公开一个 run 方法来运行流水线。

流水线可以从以下方式构建：- 一个 PipelineDefinition（__init__ 方法）- 一个 PipelineConfig（from_config 方法）- 一个配置文件（from_config_file 方法）

参数:

  • pipeline_definition (PipelineDefinition)

  • config (AbstractPipelineConfig | None)

  • do_cleaning (bool)

检索器

RetrieverInterface

class neo4j_graphrag.retrievers.base.Retriever(driver, neo4j_database=None)[source]

Neo4j 检索器的抽象类。

参数:

  • driver (neo4j.Driver)

  • neo4j_database (Optional[str])

index_name: str
VERIFY_NEO4J_VERSION = True
search(*args, **kwargs)[source]

搜索方法。调用 get_search_results 方法，该方法返回一个 neo4j.Record 列表，并使用 get_result_formatter 返回的函数对其进行格式化，以返回 RetrieverResult。

参数:
返回类型:

RetrieverResult

abstract get_search_results(*args, **kwargs)[source]

此方法必须在每个子类中实现。它将接收通过 search 方法提供给公共接口的相同参数，并在验证后。它返回一个 RawSearchResult 对象，该对象包含一个 neo4j.Record 对象列表和一个可选的 metadata 字典，其中可以包含检索器级别的信息。

请注意，尽管此方法不打算从类外部调用，但我们将其公开是为了让开发人员更清楚地知道它应该在子类中实现。

返回:

Neo4j 记录列表和可选的元数据字典

返回类型:

RawSearchResult

参数:
get_result_formatter()[source]

返回用于将 neo4j.Record 转换为 RetrieverResultItem 的函数。

返回类型:

Callable[[Record], RetrieverResultItem]

default_record_formatter(record)[source]

尽力猜测节点到文本的方法。继承的类可以重写此方法以实现自定义文本格式。

参数:

record (Record)

返回类型:

RetrieverResultItem

VectorRetriever

class neo4j_graphrag.retrievers.VectorRetriever(driver, index_name, embedder=None, return_properties=None, result_formatter=None, neo4j_database=None)[source]

提供使用向量搜索嵌入的检索方法。如果提供了嵌入器，它需要具有所需的 Embedder 类型。

async run(text)[source]

import neo4j
from neo4j_graphrag.retrievers import VectorRetriever

driver = neo4j.GraphDatabase.driver(URI, auth=AUTH)

retriever = VectorRetriever(driver, "vector-index-name", custom_embedder)
retriever.search(query_text="Find me a book about Fremen", top_k=5)

或者如果查询文本的向量嵌入可用

retriever.search(query_vector=..., top_k=5)
参数:

  • driver (neo4j.Driver) – Neo4j Python 驱动程序。

  • index_name (str) – 向量索引名称。

  • embedder (Optional[Embedder]) – 用于嵌入查询文本的 Embedder 对象。

  • return_properties (Optional[list[str]]) – 要返回的节点属性列表。

  • result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) –

    提供的自定义函数，用于将 neo4j.Record 转换为 RetrieverResultItem。

    neo4j.Record 中提供了两个变量

    • node: 表示从向量索引搜索中检索到的节点。

    • score: 表示相似度分数。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

RetrieverInitializationError – 如果输入参数验证失败。

search(query_vector=None, query_text=None, top_k=5, effective_search_ratio=1, filters=None)

获取提供给 query_vector 或 query_text 的 top_k 最近邻嵌入。更多详情请参阅以下文档

要通过文本查询，实例化类时必须提供一个嵌入器。如果传递了 query_vector，则不需要嵌入器。

参数:

  • query_vector (Optional[list[float]]) – 用于获取最近邻的向量嵌入。默认为 None。

  • query_text (Optional[str]) – 用于获取最近邻的文本。默认为 None。

  • top_k (int) – 要返回的邻居数量。默认为 5。

  • effective_search_ratio (int) – 通过将 top_k 相乘来控制候选池大小，以平衡查询准确性和性能。默认为 1。

  • filters (Optional[dict[str, Any]]) – 用于元数据预过滤的过滤器。默认为 None。

引发:
返回:

搜索查询结果，为 neo4j.Record 列表和可选的元数据字典。

返回类型:

RawSearchResult

VectorCypherRetriever

class neo4j_graphrag.retrievers.VectorCypherRetriever(driver, index_name, retrieval_query, embedder=None, result_formatter=None, neo4j_database=None)[source]

提供使用向量相似性并由 Cypher 查询增强的检索方法。此检索器基于 VectorRetriever 构建。如果提供了嵌入器，它需要具有所需的 Embedder 类型。

注意：node 是基础查询中的一个变量，可以在 retrieval_query 中使用，如下例所示。

retrieval_query 是额外的 Cypher，它允许在检索 node 后进行图遍历。

async run(text)[source]

import neo4j
from neo4j_graphrag.retrievers import VectorCypherRetriever

driver = neo4j.GraphDatabase.driver(URI, auth=AUTH)

retrieval_query = "MATCH (node)-[:AUTHORED_BY]->(author:Author)" "RETURN author.name"
retriever = VectorCypherRetriever(
  driver, "vector-index-name", retrieval_query, custom_embedder
)
retriever.search(query_text="Find me a book about Fremen", top_k=5)
参数:

  • driver (neo4j.Driver) – Neo4j Python 驱动程序。

  • index_name (str) – 向量索引名称。

  • retrieval_query (str) – 追加的 Cypher 查询。

  • embedder (Optional[Embedder]) – 用于嵌入查询文本的 Embedder 对象。

  • result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) – 提供的自定义函数，用于将 neo4j.Record 转换为 RetrieverResultItem。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

在用户指南中了解更多信息。

search(query_vector=None, query_text=None, top_k=5, effective_search_ratio=1, query_params=None, filters=None)

获取提供给 query_vector 或 query_text 的 top_k 最近邻嵌入。更多详情请参阅以下文档

要通过文本查询，实例化类时必须提供一个嵌入器。如果传递了 query_vector，则不需要嵌入器。

参数:

  • query_vector (Optional[list[float]]) – 用于获取最近邻的向量嵌入。默认为 None。

  • query_text (Optional[str]) – 用于获取最近邻的文本。默认为 None。

  • top_k (int) – 要返回的邻居数量。默认为 5。

  • effective_search_ratio (int) – 通过将 top_k 相乘来控制候选池大小，以平衡查询准确性和性能。默认为 1。

  • query_params (Optional[dict[str, Any]]) – Cypher 查询的参数。默认为 None。

  • filters (Optional[dict[str, Any]]) – 用于元数据预过滤的过滤器。默认为 None。

引发:
返回:

搜索查询结果，为 neo4j.Record 列表和可选的元数据字典。

返回类型:

RawSearchResult

HybridRetriever

class neo4j_graphrag.retrievers.HybridRetriever(driver, vector_index_name, fulltext_index_name, embedder=None, return_properties=None, result_formatter=None, neo4j_database=None)[source]

提供结合向量搜索嵌入和全文搜索的检索方法。如果提供了嵌入器，它需要具有所需的 Embedder 类型。

async run(text)[source]

import neo4j
from neo4j_graphrag.retrievers import HybridRetriever

driver = neo4j.GraphDatabase.driver(URI, auth=AUTH)

retriever = HybridRetriever(
    driver, "vector-index-name", "fulltext-index-name", custom_embedder
)
retriever.search(query_text="Find me a book about Fremen", top_k=5)
参数:

  • driver (neo4j.Driver) – Neo4j Python 驱动程序。

  • vector_index_name (str) – 向量索引名称。

  • fulltext_index_name (str) – 全文索引名称。

  • embedder (Optional[Embedder]) – 用于嵌入查询文本的 Embedder 对象。

  • return_properties (Optional[list[str]]) – 要返回的节点属性列表。

  • result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) – 提供的自定义函数，用于将 neo4j.Record 转换为 RetrieverResultItem。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

    neo4j.Record 中提供了两个变量

    • node: 表示从向量索引搜索中检索到的节点。

    • score: 表示相似度分数。

search(query_text, query_vector=None, top_k=5, effective_search_ratio=1, ranker=HybridSearchRanker.NAIVE, alpha=None)

获取提供给 query_vector 或 query_text 的 top_k 最近邻嵌入。query_vector 和 query_text 都可以提供。如果提供了 query_vector，则在向量搜索中将优先使用它而不是嵌入的 query_text。

更多详情请参阅以下文档

要通过文本查询，实例化类时必须提供一个嵌入器。

参数:

  • query_text (str) – 用于获取最近邻的文本。

  • query_vector (Optional[list[float]], optional) – 用于获取最近邻的向量嵌入。默认为 None。

  • top_k (int, optional) – 要返回的邻居数量。默认为 5。

  • effective_search_ratio (int) – 通过将 top_k 相乘来控制向量索引的候选池大小，以平衡查询准确性和性能。默认为 1。

  • ranker (str, HybridSearchRanker) – 用于对检索结果进行排序的排名器类型。

  • alpha (Optional[float]) – 使用线性排名器时向量分数的权重。全文索引分数乘以 (1 - alpha)。使用线性排名器时为必填项；必须在 0 到 1 之间。

引发:
返回:

搜索查询结果，为 neo4j.Record 列表和可选的元数据字典。

返回类型:

RawSearchResult

HybridCypherRetriever

class neo4j_graphrag.retrievers.HybridCypherRetriever(driver, vector_index_name, fulltext_index_name, retrieval_query, embedder=None, result_formatter=None, neo4j_database=None)[source]

提供结合向量搜索嵌入和全文搜索的检索方法，并由 Cypher 查询增强。此检索器基于 HybridRetriever 构建。如果提供了嵌入器，它需要具有所需的 Embedder 类型。

注意：node 是基础查询中的一个变量，可以在 retrieval_query 中使用，如下例所示。

async run(text)[source]

import neo4j
from neo4j_graphrag.retrievers import HybridCypherRetriever

driver = neo4j.GraphDatabase.driver(URI, auth=AUTH)

retrieval_query = "MATCH (node)-[:AUTHORED_BY]->(author:Author)" "RETURN author.name"
retriever = HybridCypherRetriever(
    driver, "vector-index-name", "fulltext-index-name", retrieval_query, custom_embedder
)
retriever.search(query_text="Find me a book about Fremen", top_k=5)

要通过文本查询，实例化类时必须提供一个嵌入器。

参数:

  • driver (neo4j.Driver) – Neo4j Python 驱动程序。

  • vector_index_name (str) – 向量索引名称。

  • fulltext_index_name (str) – 全文索引名称。

  • retrieval_query (str) – 追加的 Cypher 查询。

  • embedder (Optional[Embedder]) – 用于嵌入查询文本的 Embedder 对象。

  • result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) – 提供的自定义函数，用于将 neo4j.Record 转换为 RetrieverResultItem。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

RetrieverInitializationError – 如果输入参数验证失败。

search(query_text, query_vector=None, top_k=5, effective_search_ratio=1, query_params=None, ranker=HybridSearchRanker.NAIVE, alpha=None)

获取提供给 query_vector 或 query_text 的 top_k 最近邻嵌入。query_vector 和 query_text 都可以提供。如果提供了 query_vector，则在向量搜索中将优先使用它而不是嵌入的 query_text。

更多详情请参阅以下文档

参数:

  • query_text (str) – 用于获取最近邻的文本。

  • query_vector (Optional[list[float]]) – 用于获取最近邻的向量嵌入。默认为 None。

  • top_k (int) – 要返回的邻居数量。默认为 5。

  • effective_search_ratio (int) – 通过将 top_k 相乘来控制向量索引的候选池大小，以平衡查询准确性和性能。默认为 1。

  • query_params (Optional[dict[str, Any]]) – Cypher 查询的参数。默认为 None。

  • ranker (str, HybridSearchRanker) – 用于对检索结果进行排序的排名器类型。

  • alpha (Optional[float]) – 使用线性排名器时向量分数的权重。全文索引分数乘以 (1 - alpha)。使用线性排名器时为必填项；必须在 0 到 1 之间。

引发:
返回:

搜索查询结果，为 neo4j.Record 列表和可选的元数据字典。

返回类型:

RawSearchResult

Text2CypherRetriever

class neo4j_graphrag.retrievers.Text2CypherRetriever(driver, llm, neo4j_schema=None, examples=None, result_formatter=None, custom_prompt=None, neo4j_database=None)[source]

允许使用自然语言从 Neo4j 数据库中检索记录。使用 LLM 将用户的自然语言查询转换为 Cypher 查询，然后使用生成的 Cypher 查询从 Neo4j 数据库中检索记录。

参数:

  • driver (neo4j.Driver) – Neo4j Python 驱动程序。

  • llm (neo4j_graphrag.generation.llm.LLMInterface) – 用于生成 Cypher 查询的 LLM 对象。

  • neo4j_schema (Optional[str]) – 用于生成 Cypher 查询的 Neo4j 模式。

  • examples (Optional[list[str], optional) – 供 LLM 用作示例的可选用户输入/查询对。

  • custom_prompt (Optional[str]) – 可选的自定义提示，用于替代自动生成的提示。如果提供了 neo4j_schema 和 examples，则将包含它们作为提示参数。

  • result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]])

  • neo4j_database (Optional[str])

引发:

RetrieverInitializationError – 如果输入参数验证失败。

search(query_text, prompt_params=None)
使用 LLM 将 query_text 转换为 Cypher 查询。

使用生成的 Cypher 查询从 Neo4j 数据库中检索记录。

参数:

  • query_text (str) – 用于搜索 Neo4j 数据库的自然语言查询。

  • prompt_params (Dict[str, Any]) – 额外的键值，如果提供了自定义提示，则注入其中。如果指定了 schema 或 examples 参数，它将覆盖初始化期间传递的相应值。示例：{‘schema’: ‘这是图模式’}

引发:
返回:

搜索查询结果，为 neo4j.Record 列表和可选的元数据字典。

返回类型:

RawSearchResult

外部检索器

本节包括与 Neo4j 外部数据库集成的检索器。

WeaviateNeo4jRetriever

class neo4j_graphrag.retrievers.external.weaviate.weaviate.WeaviateNeo4jRetriever(driver, client, collection, id_property_external, id_property_neo4j, embedder=None, return_properties=None, retrieval_query=None, result_formatter=None, neo4j_database=None)[source]

提供使用 Weaviate 数据库对嵌入进行向量搜索的检索方法。如果提供了嵌入器，它需要具有所需的 Embedder 类型。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.retrievers import WeaviateNeo4jRetriever
from weaviate.connect.helpers import connect_to_local

with GraphDatabase.driver(NEO4J_URL, auth=NEO4J_AUTH) as neo4j_driver:
    with connect_to_local() as w_client:
        retriever = WeaviateNeo4jRetriever(
            driver=neo4j_driver,
            client=w_client,
            collection="Jeopardy",
            id_property_external="neo4j_id",
            id_property_neo4j="id"
        )

        result = retriever.search(query_text="biology", top_k=2)
参数:

  • driver (neo4j.Driver) – Neo4j Python 驱动程序。

  • client (WeaviateClient) – Weaviate 客户端对象。

  • collection (str) – 共享相同数据结构的 Weaviate 对象集的名称。

  • id_property_external (str) – 具有引用相应 Neo4j 节点 id 属性的标识符的 Weaviate 属性的名称。

  • id_property_neo4j (str) – 用作将 Weaviate 匹配项关联到 Neo4j 节点的标识符的 Neo4j 节点属性的名称。

  • embedder (Optional[Embedder]) – 用于嵌入查询文本的 Embedder 对象。

  • return_properties (Optional[list[str]]) – 要返回的节点属性列表。

  • result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) – 将 neo4j.Record 转换为 RetrieverResultItem 的函数。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

  • retrieval_query (Optional[str])

引发:

RetrieverInitializationError – 如果输入参数验证失败。

search(query_vector=None, query_text=None, top_k=5, **kwargs)

使用 Weaviate 获取提供给 query_vector 或 query_text 的 top_k 最近邻嵌入。query_vector 和 query_text 都可以提供。如果提供了 query_vector，则在向量搜索中将优先使用它而不是嵌入的 query_text。如果提供了 query_text，则它将检查是否提供了嵌入器并使用它来生成 query_vector。如果没有提供嵌入器，则假定在 Weaviate 中使用了向量化器。

async run(text)[source]

import neo4j
from neo4j_graphrag.retrievers import WeaviateNeo4jRetriever

driver = neo4j.GraphDatabase.driver(URI, auth=AUTH)

retriever = WeaviateNeo4jRetriever(
    driver=driver,
    client=weaviate_client,
    collection="Jeopardy",
    id_property_external="neo4j_id",
    id_property_neo4j="id",
)

biology_embedding = ...
retriever.search(query_vector=biology_embedding, top_k=2)
参数:

  • query_text (Optional[str]) – 用于获取最近邻的文本。

  • query_vector (Optional[list[float]]) – 用于获取最近邻的向量嵌入。默认为 None。

  • top_k (int) – 要返回的邻居数量。默认为 5。

  • kwargs (Any)

引发:

SearchValidationError – 如果输入参数验证失败。

返回:

搜索查询结果，为 neo4j.Record 列表和可选的元数据字典。

返回类型:

RawSearchResult

PineconeNeo4jRetriever

class neo4j_graphrag.retrievers.external.pinecone.pinecone.PineconeNeo4jRetriever(driver, client, index_name, id_property_neo4j, embedder=None, return_properties=None, retrieval_query=None, result_formatter=None, neo4j_database=None)[source]

提供使用 Pinecone 数据库对嵌入进行向量搜索的检索方法。如果提供了嵌入器，它需要具有所需的 Embedder 类型。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.retrievers import PineconeNeo4jRetriever
from pinecone import Pinecone

with GraphDatabase.driver(NEO4J_URL, auth=NEO4J_AUTH) as neo4j_driver:
    pc_client = Pinecone(PC_API_KEY)
    embedder = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

    retriever = PineconeNeo4jRetriever(
        driver=neo4j_driver,
        client=pc_client,
        index_name="jeopardy",
        id_property_neo4j="id",
        embedder=embedder,
    )

    result = retriever.search(query_text="biology", top_k=2)
参数:

  • driver (neo4j.Driver) – Neo4j Python 驱动程序。

  • client (Pinecone) – Pinecone 客户端对象。

  • index_name (str) – Pinecone 索引的名称。

  • id_property_neo4j (str) – 用作将 Pinecone 匹配项关联到 Neo4j 节点的标识符的 Neo4j 节点属性的名称。

  • embedder (Optional[Embedder]) – 用于嵌入查询文本的 Embedder 对象。

  • return_properties (Optional[list[str]]) – 要返回的节点属性列表。

  • retrieval_query (str) – 追加的 Cypher 查询。

  • result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) – 将 neo4j.Record 转换为 RetrieverResultItem 的函数。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

RetrieverInitializationError – 如果输入参数验证失败。

search(query_vector=None, query_text=None, top_k=5, **kwargs)

使用 Pinecone 获取提供给 query_vector 或 query_text 的 top_k 最近邻嵌入。query_vector 和 query_text 都可以提供。如果提供了 query_vector，则在向量搜索中将优先使用它而不是嵌入的 query_text。如果提供了 query_text，则它将检查是否提供了嵌入器并使用它来生成 query_vector。

更多详情请参阅以下文档：- 查询向量索引 - db.index.vector.queryNodes() - db.index.fulltext.queryNodes()

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.retrievers import PineconeNeo4jRetriever
from pinecone import Pinecone

with GraphDatabase.driver(NEO4J_URL, auth=NEO4J_AUTH) as neo4j_driver:
    pc_client = Pinecone(PC_API_KEY)
    retriever = PineconeNeo4jRetriever(
        driver=neo4j_driver,
        client=pc_client,
        index_name="jeopardy",
        id_property_neo4j="id"
    )
    biology_embedding = ...
    retriever.search(query_vector=biology_embedding, top_k=2)
参数:

  • query_text (str) – 用于获取最近邻的文本。

  • query_vector (Optional[list[float]], optional) – 用于获取最近邻的向量嵌入。默认为 None。

  • top_k (Optional[int]) – 要返回的邻居数量。默认为 5。

  • kwargs (Any)

引发:
返回:

搜索查询结果，为 neo4j.Record 列表和可选的元数据字典。

返回类型:

RawSearchResult

QdrantNeo4jRetriever

class neo4j_graphrag.retrievers.external.qdrant.qdrant.QdrantNeo4jRetriever(driver, client, collection_name, id_property_neo4j, id_property_external='id', using=None, embedder=None, return_properties=None, retrieval_query=None, result_formatter=None, neo4j_database=None)[source]

提供使用 Qdrant 数据库对嵌入进行向量搜索的检索方法。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.retrievers import QdrantNeo4jRetriever
from qdrant_client import QdrantClient

with GraphDatabase.driver(NEO4J_URL, auth=NEO4J_AUTH) as neo4j_driver:
    client = QdrantClient()
    retriever = QdrantNeo4jRetriever(
        driver=neo4j_driver,
        client=client,
        collection_name="my_collection",
        using="my_vector",
        id_property_external="neo4j_id"
    )
    embedding = ...
    retriever.search(query_vector=embedding, top_k=2)
参数:

  • driver (neo4j.Driver) – Neo4j Python 驱动程序。

  • client (QdrantClient) – Qdrant 客户端对象。

  • collection_name (str) – 要使用的 Qdrant 集合的名称。

  • using (str) – 多向量集合中包含在您的集合中的 Qdrant 向量的名称。

  • id_property_neo4j (str) – 用作将 Qdrant 匹配项关联到 Neo4j 节点的标识符的 Neo4j 节点属性的名称。

  • id_property_external (str) – 具有引用相应 Neo4j 节点 id 属性的标识符的 Qdrant 有效载荷属性的名称。

  • embedder (Optional[Embedder]) – 用于嵌入查询文本的 Embedder 对象。

  • return_properties (Optional[list[str]]) – 要返回的节点属性列表。

  • result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) – 将 neo4j.Record 转换为 RetrieverResultItem 的函数。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

  • retrieval_query (Optional[str])

引发:

RetrieverInitializationError – 如果输入参数验证失败。

search(query_vector=None, query_text=None, top_k=5, **kwargs)

使用 Qdrant 获取提供给 query_vector 或 query_text 的 top_k 最近邻嵌入。如果提供了 query_text，则使用提供的嵌入器生成 query_vector。

更多详情请参阅以下文档：- 查询向量索引 - db.index.vector.queryNodes() - db.index.fulltext.queryNodes()

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.retrievers import QdrantNeo4jRetriever
from qdrant_client import QdrantClient

with GraphDatabase.driver(NEO4J_URL, auth=NEO4J_AUTH) as neo4j_driver:
    client = QdrantClient()
    retriever = QdrantNeo4jRetriever(
        driver=neo4j_driver,
        client=client,
        collection_name="my_collection",
        id_property_external="neo4j_id"
    )
    embedding = ...
    retriever.search(query_vector=embedding, top_k=2)
参数:

  • query_text (str) – 用于获取最近邻的文本。

  • query_vector (Optional[list[float]], optional) – 用于获取最近邻的向量嵌入。默认为 None。

  • top_k (Optional[int]) – 要返回的邻居数量。默认为 5。

  • kwargs (Any) – 传递给 QdrantClient#query() 的其他关键字参数。

引发:
返回:

搜索查询结果，为 neo4j.Record 列表和可选的元数据字典。

返回类型:

RawSearchResult

嵌入器

class neo4j_graphrag.embeddings.base.Embedder[source]

嵌入模型接口。传递给检索器的嵌入器必须实现此接口。

abstract embed_query(text)[source]

嵌入查询文本。

参数:

text (str) – 要转换为向量嵌入的文本。

返回:

一个向量嵌入。

返回类型:

list[float]

SentenceTransformerEmbeddings

class neo4j_graphrag.embeddings.sentence_transformers.SentenceTransformerEmbeddings(model='all-MiniLM-L6-v2', *args, **kwargs)[source]
参数:
embed_query(text)[source]

嵌入查询文本。

参数:

text (str) – 要转换为向量嵌入的文本。

返回:

一个向量嵌入。

返回类型:

list[float]

OpenAIEmbeddings

class neo4j_graphrag.embeddings.openai.OpenAIEmbeddings(model='text-embedding-ada-002', **kwargs)[source]

OpenAI 嵌入类。此类使用 OpenAI Python 客户端为文本数据生成嵌入。

参数:

  • model (str) – 要使用的 OpenAI 嵌入模型的名称。默认为“text-embedding-ada-002”。

  • kwargs (Any) – 所有其他参数将传递给 openai.OpenAI 初始化。

AzureOpenAIEmbeddings

class neo4j_graphrag.embeddings.openai.AzureOpenAIEmbeddings(model='text-embedding-ada-002', **kwargs)[source]

Azure OpenAI 嵌入类。此类使用 Azure OpenAI Python 客户端为文本数据生成嵌入。

参数:

  • model (str) – 要使用的 Azure OpenAI 嵌入模型的名称。默认为“text-embedding-ada-002”。

  • kwargs (Any) – 所有其他参数将传递给 openai.AzureOpenAI 初始化。

OllamaEmbeddings

class neo4j_graphrag.embeddings.ollama.OllamaEmbeddings(model, **kwargs)[source]

Ollama 嵌入类。此类使用 ollama Python 客户端为文本数据生成向量嵌入。

参数:

  • model (str) – 要使用的 Mistral AI 文本嵌入模型的名称。默认为“mistral-embed”。

  • kwargs (Any)

embed_query(text, **kwargs)[source]

使用 Ollama 文本嵌入模型为给定查询生成嵌入。

参数:

  • text (str) – 要生成嵌入的文本。

  • **kwargs (Any) – 传递给 Ollama 客户端的其他关键字参数。

返回类型:

list[float]

VertexAIEmbeddings

class neo4j_graphrag.embeddings.vertexai.VertexAIEmbeddings(model='text-embedding-004')[source]

Vertex AI 嵌入类。此类使用 Vertex AI Python 客户端为文本数据生成向量嵌入。

参数:

model (str) – 要使用的 Vertex AI 文本嵌入模型的名称。默认为“text-embedding-004”。

embed_query(text, task_type='RETRIEVAL_QUERY', **kwargs)[source]

使用 Vertex AI 文本嵌入模型为给定查询生成嵌入。

参数:
返回类型:

list[float]

MistralAIEmbeddings

class neo4j_graphrag.embeddings.mistral.MistralAIEmbeddings(model='mistral-embed', **kwargs)[source]

Mistral AI 嵌入类。此类使用 Mistral AI Python 客户端为文本数据生成向量嵌入。

参数:

  • model (str) – 要使用的 Mistral AI 文本嵌入模型的名称。默认为“mistral-embed”。

  • kwargs (Any)

embed_query(text, **kwargs)[source]

使用 Mistral AI 文本嵌入模型为给定查询生成嵌入。

参数:

  • text (str) – 要生成嵌入的文本。

  • **kwargs (Any) – 传递给 Mistral AI 客户端的其他关键字参数。

返回类型:

list[float]

CohereEmbeddings

class neo4j_graphrag.embeddings.cohere.CohereEmbeddings(model='', **kwargs)[source]
参数:

  • model (str)

  • kwargs (Any)

embed_query(text, **kwargs)[source]

嵌入查询文本。

参数:

  • text (str) – 要转换为向量嵌入的文本。

  • kwargs (Any)

返回:

一个向量嵌入。

返回类型:

list[float]

生成

LLM

LLMInterface

class neo4j_graphrag.llm.LLMInterface(model_name, model_params=None, **kwargs)[source]

大型语言模型接口。

参数:

  • model_name (str) – 语言模型的名称。

  • model_params (Optional[dict]) – 将文本发送到模型时传递给模型的附加参数。默认为 None。

  • **kwargs (Any) – 类初始化时传递给模型的参数。默认为 None。

abstract invoke(input, message_history=None, system_instruction=None)[source]

向 LLM 发送文本输入并检索响应。

参数:

  • input (str) – 发送到 LLM 的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

LLM 的响应。

返回类型:

LLMResponse

引发:

LLMGenerationError – 如果出现任何问题。

abstract async ainvoke(input, message_history=None, system_instruction=None)[source]

异步向 LLM 发送文本输入并检索响应。

参数:

  • input (str) – 发送到 LLM 的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

LLM 的响应。

返回类型:

LLMResponse

引发:

LLMGenerationError – 如果出现任何问题。

invoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]

向 LLM 发送包含工具定义的文本输入并检索工具调用响应。

这是一个默认实现，应由支持工具/函数调用的 LLM 提供商覆盖。

参数:

  • input (str) – 发送到 LLM 的文本。

  • tools (Sequence[Tool]) – 供 LLM 选择的工具序列。每个 LLM 实现应处理将其转换为特定格式。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

包含工具调用的 LLM 响应。

返回类型:

ToolCallResponse

引发:
async ainvoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]

异步向 LLM 发送包含工具定义的文本输入并检索工具调用响应。

这是一个默认实现，应由支持工具/函数调用的 LLM 提供商覆盖。

参数:

  • input (str) – 发送到 LLM 的文本。

  • tools (Sequence[Tool]) – 供 LLM 选择的工具序列。每个 LLM 实现应处理将其转换为特定格式。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

包含工具调用的 LLM 响应。

返回类型:

ToolCallResponse

引发:

OpenAILLM

class neo4j_graphrag.llm.openai_llm.OpenAILLM(model_name, model_params=None, **kwargs)[source]
参数:

  • model_name (str)

  • model_params (Optional[dict[str, Any]])

  • kwargs (Any)

AzureOpenAILLM

class neo4j_graphrag.llm.openai_llm.AzureOpenAILLM(model_name, model_params=None, system_instruction=None, **kwargs)[source]
参数:

  • model_name (str)

  • model_params (Optional[dict[str, Any]])

  • system_instruction (Optional[str])

  • kwargs (Any)

OllamaLLM

class neo4j_graphrag.llm.ollama_llm.OllamaLLM(model_name, model_params=None, **kwargs)[source]
参数:

  • model_name (str)

  • model_params (Optional[dict[str, Any]])

  • kwargs (Any)

get_messages(input, message_history=None, system_instruction=None)[source]
参数:

  • input (str)

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]])

  • system_instruction (Optional[str])

返回类型:

Sequence[Message]

invoke(input, message_history=None, system_instruction=None)[source]

向LLM发送文本并返回响应。

参数:

  • input (str) – 要发送给LLM的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

LLM 的响应。

返回类型:

LLMResponse

async ainvoke(input, message_history=None, system_instruction=None)[source]

异步地将文本输入发送到OpenAI聊天完成模型并返回响应内容。

参数:

  • input (str) – 发送到 LLM 的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

来自OpenAI的响应。

返回类型:

LLMResponse

引发:

LLMGenerationError – 如果出现任何问题。

VertexAILLM

class neo4j_graphrag.llm.vertexai_llm.VertexAILLM(model_name='gemini-1.5-flash-001', model_params=None, system_instruction=None, **kwargs)[source]

Vertex AI上大型语言模型的接口

参数:

  • model_name (str, optional) – 要使用的LLM名称。默认为“gemini-1.5-flash-001”。

  • model_params (Optional[dict], optional) – 向模型发送文本时传递的额外参数。默认为None。

  • system_instruction (Optional[str]) – Optional[str], optional): 用于设置模型在对话中的行为和上下文的额外指令。默认为None。

  • **kwargs (Any) – 类初始化时传递给模型的参数。默认为 None。

引发:

LLMGenerationError – 如果模型生成响应时出错。

async run(text)[source]

from neo4j_graphrag.llm import VertexAILLM
from vertexai.generative_models import GenerationConfig

generation_config = GenerationConfig(temperature=0.0)
llm = VertexAILLM(
    model_name="gemini-1.5-flash-001", generation_config=generation_config
)
llm.invoke("Who is the mother of Paul Atreides?")
get_messages(input, message_history=None)[source]
参数:
返回类型:

list[Content]

invoke(input, message_history=None, system_instruction=None)[source]

向LLM发送文本并返回响应。

参数:

  • input (str) – 要发送给LLM的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

LLM 的响应。

返回类型:

LLMResponse

async ainvoke(input, message_history=None, system_instruction=None)[source]

异步地向LLM发送文本并返回响应。

参数:

  • input (str) – 要发送给LLM的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

LLM 的响应。

返回类型:

LLMResponse

async ainvoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]

异步向 LLM 发送包含工具定义的文本输入并检索工具调用响应。

这是一个默认实现，应由支持工具/函数调用的 LLM 提供商覆盖。

参数:

  • input (str) – 发送到 LLM 的文本。

  • tools (Sequence[Tool]) – 供 LLM 选择的工具序列。每个 LLM 实现应处理将其转换为特定格式。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

包含工具调用的 LLM 响应。

返回类型:

ToolCallResponse

引发:
invoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]

向 LLM 发送包含工具定义的文本输入并检索工具调用响应。

这是一个默认实现，应由支持工具/函数调用的 LLM 提供商覆盖。

参数:

  • input (str) – 发送到 LLM 的文本。

  • tools (Sequence[Tool]) – 供 LLM 选择的工具序列。每个 LLM 实现应处理将其转换为特定格式。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

包含工具调用的 LLM 响应。

返回类型:

ToolCallResponse

引发:

AnthropicLLM

class neo4j_graphrag.llm.anthropic_llm.AnthropicLLM(model_name, model_params=None, **kwargs)[source]

Anthropic平台上大型语言模型的接口

参数:

  • model_name (str, optional) – 要使用的LLM名称。默认为“gemini-1.5-flash-001”。

  • model_params (Optional[dict], optional) – 向模型发送文本时传递的额外参数。默认为None。

  • system_instruction – Optional[str], optional): 用于设置模型在对话中的行为和上下文的额外指令。默认为None。

  • **kwargs (Any) – 类初始化时传递给模型的参数。默认为 None。

引发:

LLMGenerationError – 如果模型生成响应时出错。

async run(text)[source]

from neo4j_graphrag.llm import AnthropicLLM

llm = AnthropicLLM(
    model_name="claude-3-opus-20240229",
    model_params={"max_tokens": 1000},
    api_key="sk...",   # can also be read from env vars
)
llm.invoke("Who is the mother of Paul Atreides?")
get_messages(input, message_history=None)[source]
参数:

  • input (str)

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]])

返回类型:

Iterable[MessageParam]

invoke(input, message_history=None, system_instruction=None)[source]

向LLM发送文本并返回响应。

参数:

  • input (str) – 要发送给LLM的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

LLM 的响应。

返回类型:

LLMResponse

async ainvoke(input, message_history=None, system_instruction=None)[source]

异步地向LLM发送文本并返回响应。

参数:

  • input (str) – 要发送给LLM的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

LLM 的响应。

返回类型:

LLMResponse

CohereLLM

class neo4j_graphrag.llm.cohere_llm.CohereLLM(model_name='', model_params=None, **kwargs)[source]

Cohere平台上大型语言模型的接口

参数:

  • model_name (str, optional) – 要使用的LLM名称。默认为“gemini-1.5-flash-001”。

  • model_params (Optional[dict], optional) – 向模型发送文本时传递的额外参数。默认为None。

  • system_instruction – Optional[str], optional): 用于设置模型在对话中的行为和上下文的额外指令。默认为None。

  • **kwargs (Any) – 类初始化时传递给模型的参数。默认为 None。

引发:

LLMGenerationError – 如果模型生成响应时出错。

async run(text)[source]

from neo4j_graphrag.llm import CohereLLM

llm = CohereLLM(api_key="...")
llm.invoke("Say something")
get_messages(input, message_history=None, system_instruction=None)[source]
参数:

  • input (str)

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]])

  • system_instruction (Optional[str])

返回类型:

ChatMessages

invoke(input, message_history=None, system_instruction=None)[source]

向LLM发送文本并返回响应。

参数:

  • input (str) – 要发送给LLM的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

LLM 的响应。

返回类型:

LLMResponse

async ainvoke(input, message_history=None, system_instruction=None)[source]

异步地向LLM发送文本并返回响应。

参数:

  • input (str) – 要发送给LLM的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

LLM 的响应。

返回类型:

LLMResponse

MistralAILLM

class neo4j_graphrag.llm.mistralai_llm.MistralAILLM(model_name, model_params=None, **kwargs)[source]
参数:

  • model_name (str)

  • model_params (Optional[dict[str, Any]])

  • kwargs (Any)

get_messages(input, message_history=None, system_instruction=None)[source]
参数:

  • input (str)

  • message_history (List[LLMMessage] | MessageHistory | None)

  • system_instruction (str | None)

返回类型:

list[Annotated[Annotated[AssistantMessage, Tag(tag=assistant)] | Annotated[SystemMessage, Tag(tag=system)] | Annotated[ToolMessage, Tag(tag=tool)] | Annotated[UserMessage, Tag(tag=user)], Discriminator(discriminator=~mistralai.models.chatcompletionrequest.<lambda>, custom_error_type=None, custom_error_message=None, custom_error_context=None)]]

invoke(input, message_history=None, system_instruction=None)[source]

将文本输入发送到Mistral聊天完成模型并返回响应内容。

参数:

  • input (str) – 发送到 LLM 的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

来自MistralAI的响应。

返回类型:

LLMResponse

引发:

LLMGenerationError – 如果出现任何问题。

async ainvoke(input, message_history=None, system_instruction=None)[source]

异步地将文本输入发送到MistralAI聊天完成模型并返回响应内容。

参数:

  • input (str) – 发送到 LLM 的文本。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • system_instruction (Optional[str]) – 在此调用中覆盖 LLM 系统消息的选项。

返回:

来自MistralAI的响应。

返回类型:

LLMResponse

引发:

LLMGenerationError – 如果出现任何问题。

PromptTemplate

class neo4j_graphrag.generation.prompts.PromptTemplate(template=None, expected_inputs=None, system_instructions=None)[source]

此类别用于生成参数化提示。它使用Python格式语法（花括号{}中的参数）和所需输入列表从字符串（模板）定义。在将指令发送到LLM之前，调用format方法，该方法将用提供的值替换参数。如果任何预期输入缺失，则会引发PromptMissingInputError。

参数:

  • template (Optional[str])

  • expected_inputs (Optional[list[str]])

  • system_instructions (Optional[str])

DEFAULT_SYSTEM_INSTRUCTIONS: str = ''
DEFAULT_TEMPLATE: str = ''
EXPECTED_INPUTS: list[str] = []
format(*args, **kwargs)[source]

此方法用于将参数替换为提供的值。参数必须通过以下方式提供： - 作为关键字参数（kwargs） - 如果使用与预期输入相同的顺序，则作为位置参数（args）

async run(text)[source]

prompt_template = PromptTemplate(
    template='''Explain the following concept to {target_audience}:
    Concept: {concept}
    Answer:
    ''',
    expected_inputs=['target_audience', 'concept']
)
prompt = prompt_template.format('12 yo children', concept='graph database')
print(prompt)

# Result:
# '''Explain the following concept to 12 yo children:
# Concept: graph database
# Answer:
# '''
参数:
返回类型:

str

RagTemplate

class neo4j_graphrag.generation.prompts.RagTemplate(template=None, expected_inputs=None, system_instructions=None)[source]
参数:

  • template (Optional[str])

  • expected_inputs (Optional[list[str]])

  • system_instructions (Optional[str])

DEFAULT_SYSTEM_INSTRUCTIONS: str = '使用提供的上下文回答用户问题。'
DEFAULT_TEMPLATE: str = 'Context:\n{context}\n\nExamples:\n{examples}\n\nQuestion:\n{query_text}\n\nAnswer:\n'
EXPECTED_INPUTS: list[str] = ['context', 'query_text', 'examples']

ERExtractionTemplate

class neo4j_graphrag.generation.prompts.ERExtractionTemplate(template=None, expected_inputs=None, system_instructions=None)[source]
参数:

  • template (Optional[str])

  • expected_inputs (Optional[list[str]])

  • system_instructions (Optional[str])

DEFAULT_TEMPLATE: str = '\n你是一个顶级的算法，旨在以结构化格式提取\n信息以构建知识图谱。\n\n从以下文本中提取实体（节点）并指定其类型。\n同时提取这些节点之间的关系。\n\n使用以下JSON格式返回结果：\n{{"nodes": [ {{"id": "0", "label": "Person", "properties": {{"name": "John"}} }}],\n"relationships": [{{"type": "KNOWS", "start_node_id": "0", "end_node_id": "1", "properties": {{"since": "2024-08-01"}} }}]}\n\n只使用以下节点和关系（如果提供）：\n{schema}\n\n为每个节点分配一个唯一的ID（字符串），并重新使用它来定义关系。\n请遵循关系的源节点和目标节点类型以及\n关系方向。\n\n请确保遵守以下规则以生成有效的JSON对象：\n- 除JSON本身外，不要返回任何额外信息。\n- 省略JSON周围的任何反引号 - 只输出JSON本身。\n- JSON对象不得封装在列表中 - 它是一个独立的JSON对象。\n- 属性名称必须用双引号括起来\n\n示例：\n{examples}\n\n输入文本：\n\n{text}\n'
EXPECTED_INPUTS: list[str] = ['text']

Text2CypherTemplate

class neo4j_graphrag.generation.prompts.Text2CypherTemplate(template=None, expected_inputs=None, system_instructions=None)[source]
参数:

  • template (Optional[str])

  • expected_inputs (Optional[list[str]])

  • system_instructions (Optional[str])

DEFAULT_TEMPLATE: str = '\n任务：根据用户输入生成用于查询Neo4j图数据库的Cypher语句。\n\nSchema:\n{schema}\n\n示例（可选）：\n{examples}\n\n输入：\n{query_text}\n\n不要使用任何未包含在Schema中的属性或关系。\n在您的响应中，除了生成的Cypher语句外，不要包含三重反引号 ``` 或任何额外文本。\n\nCypher查询：\n'
EXPECTED_INPUTS: list[str] = ['query_text']

RAG

GraphRAG

class neo4j_graphrag.generation.graphrag.GraphRAG(retriever, llm, prompt_template=<neo4j_graphrag.generation.prompts.RagTemplate object>)[source]

使用特定的检索器和LLM执行GraphRAG搜索。

async run(text)[source]

import neo4j
from neo4j_graphrag.retrievers import VectorRetriever
from neo4j_graphrag.llm.openai_llm import OpenAILLM
from neo4j_graphrag.generation import GraphRAG

driver = neo4j.GraphDatabase.driver(URI, auth=AUTH)

retriever = VectorRetriever(driver, "vector-index-name", custom_embedder)
llm = OpenAILLM()
graph_rag = GraphRAG(retriever, llm)
graph_rag.search(query_text="Find me a book about Fremen")
参数:

  • retriever (Retriever) – 用于查找相关上下文并传递给LLM的检索器。

  • llm (LLMInterface) – 用于生成答案的LLM。

  • prompt_template (RagTemplate) – 将用上下文和用户问题格式化并传递给LLM的提示模板。

引发:

RagInitializationError – 如果输入参数验证失败。

search(query_text='', message_history=None, examples='', retriever_config=None, return_context=None)[source]

警告

在未来的版本中，'return_context'的默认值将从'False'变为'True'。

此方法执行完整的RAG搜索

  1. 检索：上下文检索

  2. 增强：提示格式化

  3. 生成：使用LLM生成答案

参数:

  • query_text (str) – 用户问题。

  • message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 以前消息的集合，每条消息都分配有特定角色。

  • examples (str) – 添加到LLM提示中的示例。

  • retriever_config (Optional[dict]) – 传递给检索器搜索方法的参数；例如：top_k

  • return_context (bool) – 是否将检索器结果附加到最终结果（默认值：False）。

返回:

LLM生成的答案。

返回类型:

RagResultModel

conversation_prompt(summary, current_query)[source]
参数:

  • summary (str)

  • current_query (str)

返回类型:

str

数据库交互

neo4j_graphrag.indexes.create_vector_index(driver, name, label, embedding_property, dimensions, similarity_fn, fail_if_exists=False, neo4j_database=None)[source]

此方法构建并执行Cypher查询，以在Neo4j中创建新的向量索引。

请参阅Cypher手册中关于创建向量索引的部分。

确保提供的索引名称在数据库上下文中是唯一的。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.indexes import create_vector_index

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")

INDEX_NAME = "vector-index-name"

# Connect to Neo4j database
driver = GraphDatabase.driver(URI, auth=AUTH)

# Creating the index
create_vector_index(
    driver,
    INDEX_NAME,
    label="Document",
    embedding_property="vectorProperty",
    dimensions=1536,
    similarity_fn="euclidean",
    fail_if_exists=False,
)
参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • name (str) – 索引的唯一名称。

  • label (str) – 要索引的节点标签。

  • embedding_property (str) – 包含嵌入值的节点属性键。

  • dimensions (int) – 向量嵌入维度

  • similarity_fn (str) – 向量相似性函数的不区分大小写值：euclidean 或 cosine。

  • fail_if_exists (bool) – 如果为True，则当索引已存在时引发错误。默认为False。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

  • ValueError – 如果输入参数验证失败。

  • neo4j.exceptions.ClientError – 如果创建向量索引失败。

返回类型:

None

neo4j_graphrag.indexes.create_fulltext_index(driver, name, label, node_properties, fail_if_exists=False, neo4j_database=None)[source]

此方法构建并执行Cypher查询，以在Neo4j中创建新的全文索引。

请参阅Cypher手册中关于创建全文索引的部分。

确保提供的索引名称在数据库上下文中是唯一的。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.indexes import create_fulltext_index

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")

INDEX_NAME = "fulltext-index-name"

# Connect to Neo4j database
driver = GraphDatabase.driver(URI, auth=AUTH)

# Creating the index
create_fulltext_index(
    driver,
    INDEX_NAME,
    label="Document",
    node_properties=["vectorProperty"],
    fail_if_exists=False,
)
参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • name (str) – 索引的唯一名称。

  • label (str) – 要索引的节点标签。

  • node_properties (list[str]) – 要在其上创建全文索引的节点属性。

  • fail_if_exists (bool) – 如果为True，则当索引已存在时引发错误。默认为False。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

  • ValueError – 如果输入参数验证失败。

  • neo4j.exceptions.ClientError – 如果创建全文索引失败。

返回类型:

None

neo4j_graphrag.indexes.drop_index_if_exists(driver, name, neo4j_database=None)[source]

此方法构建并执行Cypher查询，以在Neo4j中删除一个索引（如果存在）。请参阅Cypher手册中关于删除向量索引的部分。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.indexes import drop_index_if_exists

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")

INDEX_NAME = "fulltext-index-name"

# Connect to Neo4j database
driver = GraphDatabase.driver(URI, auth=AUTH)

# Dropping the index if it exists
drop_index_if_exists(
    driver,
    INDEX_NAME,
)
参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • name (str) – 要删除的索引名称。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

neo4j.exceptions.ClientError – 如果删除索引失败。

返回类型:

None

neo4j_graphrag.indexes.upsert_vectors(driver, ids, embedding_property, embeddings, neo4j_database=None, entity_type=EntityType.NODE)[source]

此方法构建并执行Cypher查询，以在节点或关系的集合上进行嵌入的upsert（插入或更新）操作。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.indexes import upsert_vectors

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")

# Connect to Neo4j database
driver = GraphDatabase.driver(URI, auth=AUTH)

# Upsert embeddings data for several nodes
upsert_vectors(
    driver,
    ids=['123', '456', '789'],
    embedding_property="vectorProperty",
    embeddings=[
        [0.12, 0.34, 0.56],
        [0.78, 0.90, 0.12],
        [0.34, 0.56, 0.78],
    ],
    neo4j_database="neo4j",
    entity_type='NODE',
)
参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • ids (List[int]) – 节点或关系的元素ID。

  • embedding_property (str) – 存储向量的属性名称。

  • embeddings (List[List[float]]) – 要存储的向量列表，每个ID一个。

  • neo4j_database (Optional[str]) – Neo4j数据库的名称。如果未提供，则默认为服务器的默认数据库。默认为“neo4j”。

  • entity_type (EntityType) – 指定是对节点（'NODE'）还是关系（'RELATIONSHIP'）执行upsert操作。默认为'NODE'。

引发:

  • ValueError – 如果ID和嵌入的长度不匹配，或者嵌入的维度不一致。

  • Neo4jInsertionError – 如果尝试在Neo4j中upsert向量时发生错误。

返回类型:

None

neo4j_graphrag.indexes.upsert_vector(driver, node_id, embedding_property, vector, neo4j_database=None)[source]

警告

‘upsert_vector’已弃用，并将在未来版本中移除，请改用‘upsert_vectors’。

此方法构建并执行Cypher查询，以在特定节点上进行向量属性的upsert（插入或更新）操作。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.indexes import upsert_vector

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")

# Connect to Neo4j database
driver = GraphDatabase.driver(URI, auth=AUTH)

# Upsert the vector data
upsert_vector(
    driver,
    node_id="nodeId",
    embedding_property="vectorProperty",
    vector=...,
)
参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • node_id (int) – 节点的元素ID。

  • embedding_property (str) – 存储向量的属性名称。

  • vector (list[float]) – 要存储的向量。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

Neo4jInsertionError – 如果向量的upsert操作失败。

返回类型:

None

neo4j_graphrag.indexes.upsert_vector_on_relationship(driver, rel_id, embedding_property, vector, neo4j_database=None)[source]

警告

‘upsert_vector_on_relationship’已弃用，并将在未来版本中移除，请改用‘upsert_vectors’。

此方法构建并执行Cypher查询，以在特定关系上进行向量属性的upsert（插入或更新）操作。

async run(text)[source]

from neo4j import GraphDatabase
from neo4j_graphrag.indexes import upsert_vector_on_relationship

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")

# Connect to Neo4j database
driver = GraphDatabase.driver(URI, auth=AUTH)

# Upsert the vector data
upsert_vector_on_relationship(
    driver,
    node_id="nodeId",
    embedding_property="vectorProperty",
    vector=...,
)
参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • rel_id (int) – 关系的元素ID。

  • embedding_property (str) – 存储向量的属性名称。

  • vector (list[float]) – 要存储的向量。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

Neo4jInsertionError – 如果向量的upsert操作失败。

返回类型:

None

async neo4j_graphrag.indexes.async_upsert_vector(driver, node_id, embedding_property, vector, neo4j_database=None)[source]

警告

‘async_upsert_vector’已弃用，并将在未来版本中移除。

此方法构建并异步执行Cypher查询，以在特定节点上进行向量属性的upsert（插入或更新）操作。

async run(text)[source]

from neo4j import AsyncGraphDatabase
from neo4j_graphrag.indexes import upsert_vector

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")

# Connect to Neo4j database
driver = AsyncGraphDatabase.driver(URI, auth=AUTH)

# Upsert the vector data
async_upsert_vector(
    driver,
    node_id="nodeId",
    embedding_property="vectorProperty",
    vector=...,
)
参数:

  • driver (neo4j.AsyncDriver) – Neo4j Python异步驱动程序实例。

  • node_id (int) – 节点的元素ID。

  • embedding_property (str) – 存储向量的属性名称。

  • vector (list[float]) – 要存储的向量。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

Neo4jInsertionError – 如果向量的upsert操作失败。

返回类型:

None

async neo4j_graphrag.indexes.async_upsert_vector_on_relationship(driver, rel_id, embedding_property, vector, neo4j_database=None)[source]

警告

‘async_upsert_vector_on_relationship’已弃用，并将在未来版本中移除。

此方法构建并异步执行Cypher查询，以在特定关系上进行向量属性的upsert（插入或更新）操作。

async run(text)[source]

from neo4j import AsyncGraphDatabase
from neo4j_graphrag.indexes import upsert_vector_on_relationship

URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")

# Connect to Neo4j database
driver = AsyncGraphDatabase.driver(URI, auth=AUTH)

# Upsert the vector data
async_upsert_vector_on_relationship(
    driver,
    node_id="nodeId",
    embedding_property="vectorProperty",
    vector=...,
)
参数:

  • driver (neo4j.AsyncDriver) – Neo4j Python异步驱动程序实例。

  • rel_id (int) – 关系的元素ID。

  • embedding_property (str) – 存储向量的属性名称。

  • vector (list[float]) – 要存储的向量。

  • neo4j_database (Optional[str]) –

    Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认为“neo4j”）（参见文档参考）。

引发:

Neo4jInsertionError – 如果向量的upsert操作失败。

返回类型:

None

neo4j_graphrag.indexes.retrieve_vector_index_info(driver, index_name, label_or_type, embedding_property)[source]

检查Neo4j数据库中是否存在向量索引并返回其信息。如果未找到匹配的索引，则返回None。

参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • index_name (str) – 要查找的索引名称。

  • label_or_type (str) – 索引的标签（对于节点）或类型（对于关系）。

  • embedding_property (str) – 包含嵌入的属性名称。

返回:

如果找到，则包含第一个匹配索引信息的字典，否则为None。

返回类型:

Optional[Dict[str, Any]]

neo4j_graphrag.indexes.retrieve_fulltext_index_info(driver, index_name, label_or_type, text_properties=[])[source]

检查Neo4j数据库中是否存在全文索引并返回其信息。如果未找到匹配的索引，则返回None。

参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • index_name (str) – 要查找的索引名称。

  • label_or_type (str) – 索引的标签（对于节点）或类型（对于关系）。

  • text_properties (List[str]) – 被索引的文本属性名称。

返回:

如果找到，则包含第一个匹配索引信息的字典，否则为None。

返回类型:

Optional[Dict[str, Any]]

neo4j_graphrag.schema.get_structured_schema(driver, is_enhanced=False, database=None, timeout=None, sanitize=False)[source]

返回图的结构化Schema。

返回一个具有以下格式的字典

{
    'node_props': {
        'Person': [{'property': 'id', 'type': 'INTEGER'}, {'property': 'name', 'type': 'STRING'}]
    },
    'rel_props': {
        'KNOWS': [{'property': 'fromDate', 'type': 'DATE'}]
    },
    'relationships': [
        {'start': 'Person', 'type': 'KNOWS', 'end': 'Person'}
    ],
    'metadata': {
        'constraint': [
            {'id': 7, 'name': 'person_id', 'type': 'UNIQUENESS', 'entityType': 'NODE', 'labelsOrTypes': ['Persno'], 'properties': ['id'], 'ownedIndex': 'person_id', 'propertyType': None},
        ],
        'index': [
            {'label': 'Person', 'properties': ['name'], 'size': 2, 'type': 'RANGE', 'valuesSelectivity': 1.0, 'distinctValues': 2.0},
        ]
    }
}

注意

返回的字典的内部结构取决于apoc.meta.data和apoc.schema.nodes过程。

警告

某些标签从输出Schema中排除

  • 由本包中的KG Builder管道创建的__Entity__和__KGBuilder__节点标签

  • 与Bloom内部相关的某些标签。

参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • is_enhanced (bool) – 指示是否以详细统计信息格式化Schema（True）或以更简单的概览格式化（False）的标志。

  • database (Optional[str]) – 要连接的数据库名称。默认为“neo4j”。

  • timeout (Optional[float]) – 事务的超时时间（秒）。对于终止长时间运行的查询很有用。默认情况下，未设置超时。

  • sanitize (bool) – 指示是否从结果中移除包含超过128个元素的列表的标志。有助于从数据库响应中移除类似嵌入的属性。默认为False。

返回:

以结构化格式表示的图Schema信息。

返回类型:

dict[str, Any]

neo4j_graphrag.schema.get_schema(driver, is_enhanced=False, database=None, timeout=None, sanitize=False)[source]

以以下格式的字符串形式返回图的Schema

Node properties:
Person {id: INTEGER, name: STRING}
Relationship properties:
KNOWS {fromDate: DATE}
The relationships:
(:Person)-[:KNOWS]->(:Person)
参数:

  • driver (neo4j.Driver) – Neo4j Python驱动程序实例。

  • is_enhanced (bool) – 指示是否以详细统计信息格式化Schema（True）或以更简单的概览格式化（False）的标志。

  • database (Optional[str]) – 要连接的数据库名称。默认为“neo4j”。

  • timeout (Optional[float]) – 事务的超时时间（秒）。对于终止长时间运行的查询很有用。默认情况下，未设置超时。

  • sanitize (bool) – 指示是否从结果中移除包含超过128个元素的列表的标志。有助于从数据库响应中移除类似嵌入的属性。默认为False。

返回:

以序列化格式表示的图Schema信息。

返回类型:

str

neo4j_graphrag.schema.format_schema(schema, is_enhanced)[source]

将结构化Schema格式化为人类可读的字符串。

根据is_enhanced标志，此函数要么创建节点标签和关系类型的简洁列表及其属性，要么生成更详细、更冗长的表示，其中包含示例或可用值以及最小/最大统计信息等附加细节。它还包括一个格式化的现有关系列表。

参数:

  • schema (Dict[str, Any]) – 结构化Schema字典，包含节点和关系的属性以及关系定义。

  • is_enhanced (bool) – 指示是否以详细统计信息格式化Schema（True）或以更简单的概览格式化（False）的标志。

返回:

图Schema的格式化字符串表示，包括节点属性、关系属性和关系模式。

返回类型:

str

消息历史

class neo4j_graphrag.message_history.InMemoryMessageHistory(messages=None)[source]

存储在内存中的消息历史

async run(text)[source]

from neo4j_graphrag.message_history import InMemoryMessageHistory
from neo4j_graphrag.types import LLMMessage

history = InMemoryMessageHistory()

message = LLMMessage(role="user", content="Hello!")
history.add_message(message)
参数:

消息 (可选[列表[LLMMessage]]) – 用于初始化历史记录的消息列表。默认为 None。

class neo4j_graphrag.message_history.Neo4jMessageHistory(session_id, driver, window=None)[source]

存储在 Neo4j 数据库中的消息历史记录

async run(text)[source]

import neo4j
from neo4j_graphrag.message_history import Neo4jMessageHistory
from neo4j_graphrag.types import LLMMessage

driver = neo4j.GraphDatabase.driver(URI, auth=AUTH)

history = Neo4jMessageHistory(
    session_id="123", driver=driver, window=10
)

message = LLMMessage(role="user", content="Hello!")
history.add_message(message)
参数:

  • 会话 ID (联合[str, int]) – 聊天会话的唯一标识符。

  • 驱动程序 (neo4j.Driver) – Neo4j 驱动程序实例。

  • 节点标签 (str, 可选) – 用于 Neo4j 中会话节点的标签。默认为“Session”。

  • 窗口 (可选[PositiveInt], 可选) – 检索消息时返回的先前消息的数量。

错误

Neo4jGraphRagError

class neo4j_graphrag.exceptions.Neo4jGraphRagError[source]

基类: Exception

用于 neo4j-graphrag 包的全局异常。

RetrieverInitializationError

class neo4j_graphrag.exceptions.RetrieverInitializationError(errors)[source]

基类: Neo4jGraphRagError

当检索器初始化失败时抛出的异常。

参数:

错误 (列表[ErrorDetails])

SearchValidationError

class neo4j_graphrag.exceptions.SearchValidationError(errors)[source]

基类: Neo4jGraphRagError

在搜索过程中发生验证错误时抛出的异常。

参数:

错误 (列表[ErrorDetails])

FilterValidationError

class neo4j_graphrag.exceptions.FilterValidationError[source]

基类: Neo4jGraphRagError

当元数据过滤的输入验证失败时抛出的异常。

EmbeddingsGenerationError

class neo4j_graphrag.exceptions.EmbeddingsGenerationError[source]

基类: Neo4jGraphRagError

当嵌入生成失败时抛出的异常

EmbeddingRequiredError

class neo4j_graphrag.exceptions.EmbeddingRequiredError[source]

基类: Neo4jGraphRagError

当需要嵌入方法但未提供时抛出的异常。

InvalidRetrieverResultError

class neo4j_graphrag.exceptions.InvalidRetrieverResultError[source]

基类: Neo4jGraphRagError

当检索器未能返回结果时抛出的异常。

Neo4jIndexError

class neo4j_graphrag.exceptions.Neo4jIndexError[source]

基类: Neo4jGraphRagError

处理 Neo4j 索引失败时抛出的异常。

Neo4jInsertionError

class neo4j_graphrag.exceptions.Neo4jInsertionError[source]

基类: Neo4jGraphRagError

将数据插入 Neo4j 数据库失败时抛出的异常。

Neo4jVersionError

class neo4j_graphrag.exceptions.Neo4jVersionError[source]

基类: Neo4jGraphRagError

当 Neo4j 版本不符合最低要求时抛出的异常。

Text2CypherRetrievalError

class neo4j_graphrag.exceptions.Text2CypherRetrievalError[source]

基类: Neo4jGraphRagError

当 text-to-cypher 检索失败时抛出的异常。

SchemaFetchError

class neo4j_graphrag.exceptions.SchemaFetchError[source]

基类: Neo4jGraphRagError

当无法获取 Neo4jSchema 时抛出的异常。

RagInitializationError

class neo4j_graphrag.exceptions.RagInitializationError(errors)[source]

基类: Neo4jGraphRagError

参数:

错误 (列表[ErrorDetails])

PromptMissingInputError

class neo4j_graphrag.exceptions.PromptMissingInputError[source]

基类: Neo4jGraphRagError

当提示所需的输入缺失时抛出的异常。

LLMGenerationError

class neo4j_graphrag.exceptions.LLMGenerationError[source]

基类: Neo4jGraphRagError

当 LLM 的答案生成失败时抛出的异常。

SchemaValidationError

class neo4j_graphrag.exceptions.SchemaValidationError[source]

基类: Neo4jGraphRagError

模式配置错误自定义异常。

PdfLoaderError

class neo4j_graphrag.exceptions.PdfLoaderError[source]

基类: Neo4jGraphRagError

PDF 加载器错误自定义异常。

PromptMissingPlaceholderError

class neo4j_graphrag.exceptions.PromptMissingPlaceholderError[source]

基类: Neo4jGraphRagError

当提示缺少预期占位符时抛出的异常。

InvalidHybridSearchRankerError

class neo4j_graphrag.exceptions.InvalidHybridSearchRankerError[source]

基类: Neo4jGraphRagError

当为混合搜索提供了无效的排序器类型时抛出的异常。

SearchQueryParseError

class neo4j_graphrag.exceptions.SearchQueryParseError[source]

基类: Neo4jGraphRagError

当文本搜索字符串中存在查询解析错误时抛出的异常。

PipelineDefinitionError

class neo4j_graphrag.experimental.pipeline.exceptions.PipelineDefinitionError[source]

基类: Neo4jGraphRagError

当流水线图无效时抛出

PipelineMissingDependencyError

class neo4j_graphrag.experimental.pipeline.exceptions.PipelineMissingDependencyError[source]

基类: Neo4jGraphRagError

当任务已调度但其依赖项尚未完成时抛出

PipelineStatusUpdateError

class neo4j_graphrag.experimental.pipeline.exceptions.PipelineStatusUpdateError[source]

基类: Neo4jGraphRagError

尝试无效的状态更改时抛出（例如：DONE => DOING）

InvalidJSONError

class neo4j_graphrag.experimental.pipeline.exceptions.InvalidJSONError[source]

基类: Neo4jGraphRagError

当 JSON 修复未能生成有效的 JSON 时抛出。

© . All rights reserved.