API 文档¶
组件¶
Component¶
- class neo4j_graphrag.experimental.pipeline.component.Component[源代码]¶
所有组件都需要实现的接口。
- async run_with_context(context_, *args, **kwargs)[源代码]¶
此方法由流水线编排器调用。context_ 参数包含有关流水线运行的信息：run_id 和一个 notify 函数，该函数可用于将事件从组件发送到流水线回调。
此功能将在未来版本中移至 run 方法中。
默认调用 run 方法以防止任何破坏性变更。
- 参数:
context_ (RunContext)
args (Any)
kwargs (Any)
- 返回类型:
DataModel
DataLoader¶
PdfLoader¶
TextSplitter¶
FixedSizeSplitter¶
- class neo4j_graphrag.experimental.components.text_splitters.fixed_size_splitter.FixedSizeSplitter(chunk_size=4000, chunk_overlap=200, approximate=True)[源代码]¶
- 文本分割器，将输入文本分割为固定大小或近似固定大小的
块，并可选择重叠部分。
- 参数:
示例
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")
LangChainTextSplitterAdapter¶
- class neo4j_graphrag.experimental.components.text_splitters.langchain.LangChainTextSplitterAdapter(text_splitter)[源代码]¶
LangChain TextSplitters 的适配器。允许将此类的实例用于知识图谱构建流水线中。
- 参数:
text_splitter (LangChainTextSplitter) – LangChain TextSplitter 类的一个实例。
示例
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")
LlamaIndexTextSplitterAdapter¶
- class neo4j_graphrag.experimental.components.text_splitters.llamaindex.LlamaIndexTextSplitterAdapter(text_splitter)[源代码]¶
LlamaIndex TextSplitters 的适配器。允许将此类的实例用于知识图谱构建流水线中。
- 参数:
text_splitter (LlamaIndexTextSplitter) – LlamaIndex TextSplitter 类的一个实例。
示例
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")
TextChunkEmbedder¶
- class neo4j_graphrag.experimental.components.embedder.TextChunkEmbedder(embedder, max_concurrency=5)[源代码]¶
用于从文本块创建嵌入 (embeddings) 的组件。
示例
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)[源代码]¶
对一系列文本块进行嵌入处理。
- 参数:
text_chunks (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'))[源代码]¶
构建要插入 Neo4j 的词法图 (lexical graph)。词法图包含： - 每个文档一个节点 - 每个分块一个节点 - 每个分块与其来源文档之间的关系 - 分块与文档中下一个分块之间的关系
- 参数:
config (LexicalGraphConfig)
- async run(text_chunks, document_info=None)[源代码]¶
运行组件并返回其结果。
注意：如果实现了 run_with_context，则不会使用此方法。
- 参数:
text_chunks (TextChunks)
document_info (DocumentInfo | None)
- 返回类型:
GraphResult
- async process_chunk(graph, chunk, next_chunk, document_info=None)[源代码]¶
添加分块及其之间的关系 (NEXT_CHUNK)
就地更新 graph。
- 参数:
graph (Neo4jGraph)
chunk (TextChunk)
next_chunk (TextChunk | None)
document_info (DocumentInfo | None)
- 返回类型:
None
- create_document_node(document_info)[源代码]¶
创建一个带有 ‘path’ 属性的 Document 节点。任何文档元数据也会作为节点属性添加。
- 参数:
document_info (DocumentInfo)
- 返回类型:
- create_chunk_node(chunk)[源代码]¶
创建带有 ‘text’、‘index’ 属性以及过程中添加的任何 ‘metadata’ 的分块节点。针对潜在的分块嵌入属性，该属性将作为 embedding_property 添加，属于特殊情况。
- create_chunk_to_document_rel(chunk, document_info)[源代码]¶
创建分块与其所属文档之间的关系。
- 参数:
chunk (TextChunk)
document_info (DocumentInfo)
- 返回类型:
- async process_chunk_extracted_entities(chunk_graph, chunk)[源代码]¶
创建分块与从中提取出的每个实体之间的关系。
就地更新 chunk_graph。
- 参数:
chunk_graph (Neo4jGraph)
chunk (TextChunk)
- 返回类型:
None
Neo4jChunkReader¶
- class neo4j_graphrag.experimental.components.neo4j_reader.Neo4jChunkReader(driver, fetch_embeddings=False, neo4j_database=None)[源代码]¶
从 Neo4j 数据库读取文本块。
- 参数:
示例
from neo4j import GraphDatabase from neo4j_graphrag.experimental.components.neo4j_reader import Neo4jChunkReader URI = "neo4j://: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'))[源代码]¶
从 Neo4j 数据库读取文本块。
- 参数:
lexical_graph_config (LexicalGraphConfig) – 词法图的节点标签和关系类型。
- 返回类型:
SchemaBuilder¶
- class neo4j_graphrag.experimental.components.schema.SchemaBuilder[源代码]¶
一个构建器类，用于根据潜在 schema 中定义的给定实体、关系及其相互关系构造 GraphSchema 对象。
示例
from neo4j_graphrag.experimental.components.schema import ( SchemaBuilder, NodeType, PropertyType, RelationshipType, ) from neo4j_graphrag.experimental.pipeline import Pipeline node_types = [ NodeType( label="PERSON", description="An individual human being.", properties=[ PropertyType( name="name", type="STRING", description="The name of the person" ) ], ), NodeType( label="ORGANIZATION", description="A structured group of people with a common purpose.", properties=[ PropertyType( name="name", type="STRING", description="The name of the organization" ) ], ), ] relationship_types = [ RelationshipType( label="EMPLOYED_BY", description="Indicates employment relationship." ), ] patterns = [ ("PERSON", "EMPLOYED_BY", "ORGANIZATION"), ] pipe = Pipeline() schema_builder = SchemaBuilder() pipe.add_component(schema_builder, "schema_builder") pipe_inputs = { "schema": { "node_types": node_types, "relationship_types": relationship_types, "patterns": patterns, }, ... } pipe.run(pipe_inputs)
SchemaFromTextExtractor¶
- class neo4j_graphrag.experimental.components.schema.SchemaFromTextExtractor(llm, prompt_template=None, llm_params=None, use_structured_output=False)[源代码]¶
一个组件，用于从文本自动提取 schema 后，根据 LLM 的输出构造 GraphSchema 对象。
- 参数:
llm (LLMInterface) – 用于 schema 提取的语言模型。
prompt_template (Optional[PromptTemplate]) – 用于提取的自定义提示词模板。
llm_params (Optional[Dict[str, Any]]) – 传递给 LLM 的附加参数。
use_structured_output (bool) – 是否将结构化输出 (LLMInterfaceV2) 与 GraphSchema Pydantic 模型配合使用。仅支持 OpenAILLM 和 VertexAILLM。默认为 False（使用 V1 基于提示词的 JSON 提取）。
V1 示例（默认，基于提示词的 JSON）
from neo4j_graphrag.experimental.components.schema import SchemaFromTextExtractor from neo4j_graphrag.llm import OpenAILLM llm = OpenAILLM(model_name="gpt-5", model_params={"temperature": 0}) extractor = SchemaFromTextExtractor(llm=llm)
V2 示例（结构化输出）
from neo4j_graphrag.experimental.components.schema import SchemaFromTextExtractor from neo4j_graphrag.llm import OpenAILLM llm = OpenAILLM(model_name="gpt-5") extractor = SchemaFromTextExtractor(llm=llm, use_structured_output=True)
schema_visualization¶
- neo4j_graphrag.experimental.utils.schema.schema_visualization(schema)[源代码]¶
使用 neo4j-viz 库可视化 GraphSchema 的辅助函数。
用法
VG = schema_visualization(schema) html = VG.render() # in Jupyter: display(html) # to save the generated HTML with open("my_schema.html", "w") as f: f.write(html.data)
- 参数:
schema (dict[str, Any] | GraphSchema)
- 返回类型:
VisualizationGraph
EntityRelationExtractor¶
- class neo4j_graphrag.experimental.components.entity_relation_extractor.EntityRelationExtractor(*args, on_error=OnError.IGNORE, create_lexical_graph=True, **kwargs)[源代码]¶
实体关系提取组件的抽象类。
- 参数:
on_error (OnError) – 提取过程中发生错误时该怎么办。默认为引发错误。
create_lexical_graph (bool) – 除提取的实体和关系外，是否在图中包含文本块。默认为 True。
args (Any)
kwargs (Any)
- async run(chunks, document_info=None, lexical_graph_config=None, **kwargs)[源代码]¶
运行组件并返回其结果。
注意：如果实现了 run_with_context，则不会使用此方法。
- 参数:
chunks (TextChunks)
document_info (DocumentInfo | None)
lexical_graph_config (LexicalGraphConfig | None)
kwargs (Any)
- 返回类型:
- update_ids(graph, chunk)[源代码]¶
通过给节点 ID 添加唯一前缀，使其在分块、文档和流水线运行中保持唯一。
- 参数:
graph (Neo4jGraph)
chunk (TextChunk)
- 返回类型:
LLMEntityRelationExtractor¶
- class neo4j_graphrag.experimental.components.entity_relation_extractor.LLMEntityRelationExtractor(llm, prompt_template=<neo4j_graphrag.generation.prompts.ERExtractionTemplate object>, create_lexical_graph=True, on_error=OnError.RAISE, max_concurrency=5, use_structured_output=False)[源代码]¶
使用大语言模型从一系列文本块中提取知识图谱。
- 参数:
llm (LLMInterface) – 用于提取的语言模型。
prompt_template (ERExtractionTemplate | str) – 用于提取的自定义提示词模板。
create_lexical_graph (bool) – 除提取的实体和关系外，是否在图中包含文本块。默认为 True。
on_error (OnError) – 提取过程中发生错误时该怎么办。默认为引发错误。
max_concurrency (int) – 可用于向 LLM 发出请求的最大并发任务数。
use_structured_output (bool) – 是否将结构化输出 (LLMInterfaceV2) 与 Neo4jGraph Pydantic 模型配合使用。仅支持 OpenAILLM 和 VertexAILLM。默认为 False（使用 V1 基于提示词的 JSON 提取）。
V1 示例（默认，基于提示词的 JSON）
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-5", model_params={"temperature": 0, "response_format": {"type": "object"}}) extractor = LLMEntityRelationExtractor(llm=llm) pipe = Pipeline() pipe.add_component(extractor, "extractor")
V2 示例（结构化输出）
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-5", model_params={"temperature": 0}) extractor = LLMEntityRelationExtractor(llm=llm, use_structured_output=True) pipe = Pipeline() pipe.add_component(extractor, "extractor")
- async run(chunks, document_info=None, lexical_graph_config=None, schema=None, examples='', **kwargs)[源代码]¶
对列表中的所有块执行实体和关系提取。
（可选）通过添加节点和关系来表示返回图中的文档及其分块，从而创建“词法图”（更多详情，请参阅 词法图构建器文档 和 用户指南）
- 参数:
chunks (TextChunks) – 要从中提取实体和关系的文本块列表。
document_info (Optional[DocumentInfo], optional) – 分块来源的文档。用于词法图创建步骤。
lexical_graph_config (Optional[LexicalGraphConfig], optional) – 词法图配置，用于自定义词法图中的节点标签和关系类型。
schema (GraphSchema | None) – 用于指导 LLM 提取的 schema 定义。
examples (str) – 提示词中用于少样本学习 (few-shot learning) 的示例。
kwargs (Any)
- 返回类型:
KGWriter¶
- class neo4j_graphrag.experimental.components.kg_writer.KGWriter[源代码]¶
用于将知识图谱写入数据存储的抽象类。
- 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'))[源代码]¶
将图写入数据存储。
- 参数:
graph (Neo4jGraph) – 要写入数据存储的知识图谱。
lexical_graph_config (LexicalGraphConfig) – 词法图中的节点标签和关系类型。
- 返回类型:
Neo4jWriter¶
- class neo4j_graphrag.experimental.components.kg_writer.Neo4jWriter(driver, neo4j_database=None, batch_size=1000, clean_db=True)[源代码]¶
将知识图谱写入 Neo4j 数据库。
- 参数:
示例
from neo4j import GraphDatabase from neo4j_graphrag.experimental.components.kg_writer import Neo4jWriter from neo4j_graphrag.experimental.pipeline import Pipeline URI = "neo4j://: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'))[源代码]¶
将知识图谱更新插入 (upsert) 到 Neo4j 数据库中。
- 参数:
graph (Neo4jGraph) – 要更新插入数据库的知识图谱。
lexical_graph_config (LexicalGraphConfig) – 词法图的节点标签和关系类型。
- 返回类型:
SinglePropertyExactMatchResolver¶
- class neo4j_graphrag.experimental.components.resolver.SinglePropertyExactMatchResolver(driver, filter_query=None, resolve_property='name', neo4j_database=None)[源代码]¶
解析具有相同标签且属性完全相同（默认为“name”）的实体。
- 参数:
示例
from neo4j import GraphDatabase from neo4j_graphrag.experimental.components.resolver import SinglePropertyExactMatchResolver URI = "neo4j://: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
SpaCySemanticMatchResolver¶
- class neo4j_graphrag.experimental.components.resolver.SpaCySemanticMatchResolver(driver, filter_query=None, resolve_properties=None, similarity_threshold=0.8, nlp=None, spacy_model='en_core_web_lg', auto_download_spacy_model=True, neo4j_database=None)[源代码]¶
基于 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”）（参阅文档参考）。
nlp (Any)
auto_download_spacy_model (bool)
示例
from neo4j import GraphDatabase from neo4j_graphrag.experimental.components.resolver import SpaCySemanticMatchResolver URI = "neo4j://: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
FuzzyMatchResolver¶
流水线 (Pipelines)¶
Pipeline¶
- class neo4j_graphrag.experimental.pipeline.Pipeline(store=None, callback=None)[源代码]¶
这是主流水线类，用于定义组件及其执行顺序
- 参数:
store (Optional[ResultStore])
callback (Optional[EventCallbackProtocol])
SimpleKGPipeline¶
- class neo4j_graphrag.experimental.pipeline.kg_builder.SimpleKGPipeline(llm, driver, embedder, entities=None, relations=None, potential_schema=None, 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) – 用于从文本块生成块嵌入（chunk embeddings）的嵌入器实例。
schema (Optional[Union[GraphSchema, dict[str, list]]]) – 定义节点类型、关系类型和图模式的架构配置。
entities (Optional[List[Union[str, dict[str, str], NodeType]]]) –
已弃用。以下项之一的列表：
str: 实体标签
dict: 遵循 NodeType 架构，即包含 label（标签）、description（描述）和 properties（属性）键
自 1.7.1 版本起弃用：请改用 schema
relations (Optional[List[Union[str, dict[str, str], RelationshipType]]]) –
已弃用。以下项之一的列表：
str: 关系标签
dict: 遵循 RelationshipType 架构，即包含 label（标签）、description（描述）和 properties（属性）键
自 1.7.1 版本起弃用：请改用 schema
potential_schema (Optional[List[tuple]]) –
已弃用。潜在架构关系列表。
自 1.7.1 版本起弃用：请改用 schema
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])
配置文件¶
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, schema=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]¶
- 参数:
neo4j_config (dict[str, Neo4jDriverType])
embedder_config (dict[str, EmbedderType])
extras (dict[str, float | str | ParamFromEnvConfig | ParamFromKeyConfig | dict[str, Any]])
template_ (Literal[PipelineType.SIMPLE_KG_PIPELINE])
from_pdf (bool)
entities (Sequence[str | dict[str, str | list[dict[str, str]]]])
relations (Sequence[str | dict[str, str | list[dict[str, str]]]])
schema (GraphSchema | None)
on_error (OnError)
prompt_template (ERExtractionTemplate | str)
perform_entity_resolution (bool)
lexical_graph_config (LexicalGraphConfig | None)
neo4j_database (str | None)
pdf_loader (ComponentType | None)
kg_writer (ComponentType | None)
text_splitter (ComponentType | None)
PipelineRunner¶
- class neo4j_graphrag.experimental.pipeline.config.runner.PipelineRunner(pipeline_definition, config=None, do_cleaning=False)[source]¶
管道运行器（Pipeline runner）从不同的对象构建管道，并公开一个 run 方法来运行管道
管道可以从以下内容构建：- 一个 PipelineDefinition（`__init__` 方法）- 一个 PipelineConfig（`from_config` 方法）- 一个配置文件（`from_config_file` 方法）
- 参数:
pipeline_definition (PipelineDefinition)
config (Optional[AbstractPipelineConfig])
do_cleaning (bool)
检索器¶
RetrieverInterface¶
- class neo4j_graphrag.retrievers.base.Retriever(driver, neo4j_database=None)[source]¶
Neo4j 检索器的抽象类
- 参数:
driver (neo4j.Driver)
neo4j_database (Optional[str])
- VERIFY_NEO4J_VERSION = True¶
- search(*args, **kwargs)[source]¶
搜索方法。调用 `get_search_results` 方法（该方法返回 `neo4j.Record` 列表），并使用 `get_result_formatter` 返回的函数对其进行格式化，最后返回 `RetrieverResult`。
- 参数:
- 返回类型:
- abstract get_search_results(*args, **kwargs)[source]¶
此方法必须在每个子类中实现。它将接收提供给 `search` 方法公共接口的相同参数（在验证后）。它返回一个 `RawSearchResult` 对象，该对象由 `neo4j.Record` 对象列表和可选的 `metadata`（元数据）字典组成，该字典可包含检索器级别的信息。
注意，尽管此方法不打算从类外部调用，但我们将其设为公共，以便开发者清楚地了解它应该在子类中实现。
- 返回:
Neo4j 记录列表和可选的元数据字典
- 返回类型:
- 参数:
- get_result_formatter()[source]¶
返回用于将 neo4j.Record 转换为 RetrieverResultItem 的函数。
- 返回类型:
Callable[[Record], RetrieverResultItem]
- default_record_formatter(record)[source]¶
尽力推测节点到文本（node-to-text）的方法。继承类可以重写此方法以实现自定义文本格式。
- 参数:
record (Record)
- 返回类型:
VectorRetriever¶
- class neo4j_graphrag.retrievers.VectorRetriever(driver, index_name, embedder=None, return_properties=None, result_formatter=None, neo4j_database=None)[source]¶
提供使用嵌入向量搜索的检索方法。如果提供了嵌入器，它需要具有所需的 Embedder 类型。
示例
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]) – 用于嵌入查询文本的嵌入器对象。
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`，则不需要嵌入器。
- 参数:
- 抛出异常:
SearchValidationError – 如果输入参数验证失败。
EmbeddingRequiredError – 如果未提供嵌入器。
- 返回:
搜索查询的结果，表现为 neo4j.Record 列表和可选的元数据字典
- 返回类型:
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` 后进行图遍历。
示例
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]) – 用于嵌入查询文本的嵌入器对象。
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`，则不需要嵌入器。
- 参数:
- 抛出异常:
SearchValidationError – 如果输入参数验证失败。
EmbeddingRequiredError – 如果未提供嵌入器。
- 返回:
搜索查询的结果，表现为 neo4j.Record 列表和可选的元数据字典
- 返回类型:
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 类型。
示例
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]) – 用于嵌入查询文本的嵌入器对象。
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 之间。
- 抛出异常:
SearchValidationError – 如果输入参数验证失败。
EmbeddingRequiredError – 如果未提供嵌入器。
- 返回:
搜索查询的结果，表现为 neo4j.Record 列表和可选的元数据字典
- 返回类型:
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` 中使用，如下例所示。
示例
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]) – 用于嵌入查询文本的嵌入器对象。
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 之间。
- 抛出异常:
SearchValidationError – 如果输入参数验证失败。
EmbeddingRequiredError – 如果未提供嵌入器。
- 返回:
搜索查询的结果，表现为 neo4j.Record 列表和可选的元数据字典
- 返回类型:
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 模式（schema）。
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 数据库检索记录。
- 参数:
- 抛出异常:
SearchValidationError – 如果输入参数验证失败。
Text2CypherRetrievalError – 如果 LLM 无法生成正确的 Cypher 查询。
- 返回:
搜索查询的结果，表现为 neo4j.Record 列表和可选的元数据字典
- 返回类型:
ToolsRetriever¶
- class neo4j_graphrag.retrievers.ToolsRetriever(driver, llm, tools, neo4j_database=None, system_instruction=None)[source]¶
一个使用 LLM 根据用户输入选择适当工具进行检索的检索器。
该检索器接收一个 LLM 实例和一个 Tool 对象列表。执行搜索时，它使用 LLM 分析查询并确定应使用哪些工具（如果有）来检索必要的数据。然后它执行选定的工具并返回合并后的结果。
示例
import neo4j from neo4j_graphrag.retrievers import ToolsRetriever, VectorRetriever, Text2CypherRetriever from neo4j_graphrag.llm import OpenAILLM from neo4j_graphrag.embeddings import OpenAIEmbeddings driver = neo4j.GraphDatabase.driver("neo4j://:7687", auth=("neo4j", "password")) llm = OpenAILLM(model_name="gpt-5", api_key="your-api-key") embedder = OpenAIEmbeddings(model="text-embedding-3-small", api_key="your-api-key") # Create retrievers and convert them to tools vector_retriever = VectorRetriever(driver, "vector-index", embedder) vector_tool = vector_retriever.convert_to_tool( name="vector_search", description="Search for documents using semantic similarity" ) text2cypher_retriever = Text2CypherRetriever(driver, llm) cypher_tool = text2cypher_retriever.convert_to_tool( name="cypher_search", description="Generate and execute Cypher queries for structured data retrieval" ) # Initialize ToolsRetriever with the tools tools_retriever = ToolsRetriever( driver=driver, llm=llm, tools=[vector_tool, cypher_tool] ) # Use the retriever - the LLM will automatically select appropriate tools result = tools_retriever.search("What movies did Tom Hanks act in and what are their plots?")
- 参数:
driver (neo4j.Driver) – Neo4j Python 驱动程序。
llm (LLMInterface) – 用于选择和协调工具执行的 LLM 实例。
tools (Sequence[Tool]) – 可供选择的工具列表。所有工具必须具有唯一的名称。
neo4j_database (Optional[str]) – Neo4j 数据库名称。如果未提供，则默认为服务器的默认数据库（默认为 “neo4j”）。
system_instruction (Optional[str]) – 用于指导 LLM 选择工具的自定义系统指令。如果未提供，将使用默认指令。
- 抛出异常:
ValueError – 如果工具列表中发现重复的工具名称。
- search(query_text, message_history=None, **kwargs)¶
使用 LLM 根据查询选择并执行适当的工具。
- 参数:
query_text (str) – 用户的查询文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]], optional) – 之前的对话历史。默认为 None。
**kwargs (Any) – 传递给工具执行的附加参数。
- 返回:
来自执行工具的合并结果。
- 返回类型:
外部检索器¶
本节包含与 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, node_label_neo4j=None)[source]¶
提供使用 Weaviate 数据库对嵌入向量进行搜索的检索方法。如果提供了嵌入器，它需要具有所需的 Embedder 类型。
示例
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) – Weaviate 属性名称，其包含指向相应 Neo4j 节点 ID 属性的标识符。
id_property_neo4j (str) – 用作将 Weaviate 匹配项关联到 Neo4j 节点的标识符的 Neo4j 节点属性名称。
embedder (Optional[Embedder]) – 用于嵌入查询文本的嵌入器对象。
result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) – 将 neo4j.Record 转换为 RetrieverResultItem 的函数。
neo4j_database (Optional[str]) –
Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认情况下为“neo4j”）（参阅文档参考）。
node_label_neo4j (Optional[str]) – 要检索的 Neo4j 节点标签。如果需要，该标签必须正确转义，例如 “`Label with spaces`”。
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 中使用了向量化器（vectorizer）。
示例
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", node_label_neo4j="Document", ) biology_embedding = ... retriever.search(query_vector=biology_embedding, top_k=2)
- 参数:
- 抛出异常:
SearchValidationError – 如果输入参数验证失败。
- 返回:
搜索查询的结果，表现为 neo4j.Record 列表和可选的元数据字典
- 返回类型:
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, node_label_neo4j=None)[source]¶
提供使用 Pinecone 数据库对嵌入向量进行搜索的检索方法。如果提供了嵌入器，它需要具有所需的 Embedder 类型。
示例
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]) – 用于嵌入查询文本的嵌入器对象。
retrieval_query (str) – 附加的 Cypher 查询。
result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) – 将 neo4j.Record 转换为 RetrieverResultItem 的函数。
neo4j_database (Optional[str]) –
Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认情况下为“neo4j”）（参阅文档参考）。
node_label_neo4j (Optional[str]) – 要检索的 Neo4j 节点标签。如果需要，该标签必须正确转义，例如 “`Label with spaces`”。
- 抛出异常:
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()
示例
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", node_label_neo4j="Document", ) biology_embedding = ... retriever.search(query_vector=biology_embedding, top_k=2)
- 参数:
- 抛出异常:
SearchValidationError – 如果输入参数验证失败。
EmbeddingRequiredError – 如果在使用文本作为输入时未提供嵌入器。
- 返回:
搜索查询的结果，表现为 neo4j.Record 列表和可选的元数据字典
- 返回类型:
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, node_label_neo4j=None, id_property_getter=None)[source]¶
提供使用 Qdrant 数据库对嵌入向量进行搜索的检索方法。
示例
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) – Qdrant 有效负载（payload）属性名称，其包含指向相应 Neo4j 节点 ID 属性的标识符。
embedder (Optional[Embedder]) – 用于嵌入查询文本的嵌入器对象。
result_formatter (Optional[Callable[[neo4j.Record], RetrieverResultItem]]) – 将 neo4j.Record 转换为 RetrieverResultItem 的函数。
neo4j_database (Optional[str]) –
Neo4j 数据库的名称。如果未提供，则默认为服务器的默认数据库（默认情况下为“neo4j”）（参阅文档参考）。
node_label_neo4j (Optional[str]) – 要检索的 Neo4j 节点标签。如果需要，该标签必须正确转义，例如 “`Label with spaces`”。
id_property_getter (Optional[Callable[[ScoredPoint], str]]) – 从 ScoredPoint 获取 ID 属性的函数。默认为 point.payload.get(id_property_external, point.id)。
retrieval_query (Optional[str])
- 抛出异常:
RetrieverInitializationError – 如果输入参数验证失败。
- search(query_vector=None, query_text=None, top_k=5, **kwargs)¶
针对提供的 query_vector 或 query_text，使用 Qdrant 获取前 top_k 个最近邻嵌入。如果提供了 query_text，则使用提供的嵌入器（embedder）生成 query_vector。
参阅以下文档了解更多详细信息： - 查询向量索引 - db.index.vector.queryNodes() - db.index.fulltext.queryNodes()
示例
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", node_label_neo4j="Document", ) embedding = ... retriever.search(query_vector=embedding, top_k=2)
- 参数:
- 抛出异常:
SearchValidationError – 如果输入参数验证失败。
EmbeddingRequiredError – 如果在使用文本作为输入时未提供嵌入器。
- 返回:
搜索查询的结果，表现为 neo4j.Record 列表和可选的元数据字典
- 返回类型:
Embedder (嵌入器)¶
- class neo4j_graphrag.embeddings.base.Embedder(rate_limit_handler=None)[source]¶
嵌入模型的接口。传入检索器（retriever）的嵌入器必须实现此接口。
- 参数:
rate_limit_handler (Optional[RateLimitHandler]) – 速率限制处理器。默认为带指数退避的重试。
SentenceTransformerEmbeddings¶
OpenAIEmbeddings¶
- class neo4j_graphrag.embeddings.openai.OpenAIEmbeddings(model='text-embedding-ada-002', rate_limit_handler=None, **kwargs)[source]¶
OpenAI 嵌入类。此类使用 OpenAI Python 客户端为文本数据生成嵌入。
- 参数:
model (str) – 要使用的 OpenAI 嵌入模型的名称。默认为 “text-embedding-ada-002”。
kwargs (Any) – 所有其他参数将传递给 openai.OpenAI 初始化。
rate_limit_handler (Optional[RateLimitHandler])
AzureOpenAIEmbeddings¶
- class neo4j_graphrag.embeddings.openai.AzureOpenAIEmbeddings(model='text-embedding-ada-002', rate_limit_handler=None, **kwargs)[source]¶
Azure OpenAI 嵌入类。此类使用 Azure OpenAI Python 客户端为文本数据生成嵌入。
- 参数:
model (str) – 要使用的 Azure OpenAI 嵌入模型的名称。默认为 “text-embedding-ada-002”。
kwargs (Any) – 所有其他参数将传递给 openai.AzureOpenAI 初始化。
rate_limit_handler (Optional[RateLimitHandler])
OllamaEmbeddings¶
- class neo4j_graphrag.embeddings.ollama.OllamaEmbeddings(model, rate_limit_handler=None, **kwargs)[source]¶
Ollama 嵌入类。此类使用 ollama Python 客户端为文本数据生成向量嵌入。
- 参数:
model (str) – 要使用的 Mistral AI 文本嵌入模型名称。默认为 “mistral-embed”。
rate_limit_handler (Optional[RateLimitHandler])
kwargs (Any)
VertexAIEmbeddings¶
- class neo4j_graphrag.embeddings.vertexai.VertexAIEmbeddings(model='text-embedding-004', rate_limit_handler=None)[source]¶
Vertex AI 嵌入类。此类使用 Vertex AI Python 客户端为文本数据生成向量嵌入。
- 参数:
model (str) – 要使用的 Vertex AI 文本嵌入模型名称。默认为 “text-embedding-004”。
rate_limit_handler (Optional[RateLimitHandler])
- embed_query(text, task_type='RETRIEVAL_QUERY', **kwargs)[source]¶
使用 Vertex AI 文本嵌入模型为给定查询生成嵌入。
- 参数:
text (str) – 要生成嵌入的文本。
task_type (str) – 文本嵌入任务的类型。默认为 “RETRIEVAL_QUERY”。有关完整列表，请参见 https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/text-embeddings-api#tasktype。
**kwargs (Any) – 传递给 Vertex AI 客户端 get_embeddings 方法的额外关键字参数。
- 返回类型:
MistralAIEmbeddings¶
- class neo4j_graphrag.embeddings.mistral.MistralAIEmbeddings(model='mistral-embed', rate_limit_handler=None, **kwargs)[source]¶
Mistral AI 嵌入类。此类使用 Mistral AI Python 客户端为文本数据生成向量嵌入。
- 参数:
model (str) – 要使用的 Mistral AI 文本嵌入模型名称。默认为 “mistral-embed”。
rate_limit_handler (Optional[RateLimitHandler])
kwargs (Any)
CohereEmbeddings¶
- class neo4j_graphrag.embeddings.cohere.CohereEmbeddings(model='', rate_limit_handler=None, **kwargs)[source]¶
- 参数:
model (str)
rate_limit_handler (Optional[RateLimitHandler])
kwargs (Any)
Generation (生成)¶
LLM (大语言模型)¶
LLMInterface (LLM 接口)¶
- class neo4j_graphrag.llm.LLMInterface(model_name, model_params=None, rate_limit_handler=None, **kwargs)[source]¶
大语言模型的接口。
- 参数:
model_name (str) – 语言模型的名称。
model_params (Optional[dict]) – 发送文本到模型时传递的额外参数。默认为 None。
rate_limit_handler (Optional[RateLimitHandler]) – 速率限制处理器。默认为带指数退避的重试。
**kwargs (Any) – 初始化类时传递给模型的参数。默认为 None。
- supports_structured_output: bool = False¶
此 LLM 是否支持结构化输出（带有 Pydantic 模型或 JSON 模式的 response_format）。
- 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 的响应。
- 返回类型:
- 抛出异常:
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 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- invoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]¶
向 LLM 发送带有工具定义的文本输入，并获取工具调用响应。
这是一个默认实现，支持工具/函数调用的 LLM 提供商应当重写此方法。
- 参数:
input (str) – 发送给 LLM 的文本。
tools (Sequence[Tool]) – 供 LLM 选择的 Tools 序列。每个 LLM 实现都应处理向其特定格式的转换。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的包含工具调用的响应。
- 返回类型:
ToolCallResponse
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
NotImplementedError – 如果 LLM 提供商不支持工具调用。
- async ainvoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]¶
异步向 LLM 发送带有工具定义的文本输入，并获取工具调用响应。
这是一个默认实现，支持工具/函数调用的 LLM 提供商应当重写此方法。
- 参数:
input (str) – 发送给 LLM 的文本。
tools (Sequence[Tool]) – 供 LLM 选择的 Tools 序列。每个 LLM 实现都应处理向其特定格式的转换。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的包含工具调用的响应。
- 返回类型:
ToolCallResponse
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
NotImplementedError – 如果 LLM 提供商不支持工具调用。
OpenAILLM¶
- class neo4j_graphrag.llm.openai_llm.OpenAILLM(model_name, model_params=None, rate_limit_handler=None, **kwargs)[source]¶
OpenAI LLM。
- 参数:
model_name (str)
rate_limit_handler (Optional[RateLimitHandler])
kwargs (Any)
AzureOpenAILLM¶
- class neo4j_graphrag.llm.openai_llm.AzureOpenAILLM(model_name, model_params=None, system_instruction=None, rate_limit_handler=None, **kwargs)[source]¶
Azure OpenAI LLM。
- 参数:
model_name (str)
system_instruction (Optional[str])
rate_limit_handler (Optional[RateLimitHandler])
kwargs (Any)
OllamaLLM¶
- class neo4j_graphrag.llm.ollama_llm.OllamaLLM(model_name, model_params=None, rate_limit_handler=None, **kwargs)[source]¶
Ollama 模型的 LLM 包装器。
- 参数:
model_name (str)
rate_limit_handler (Optional[RateLimitHandler])
kwargs (Any)
- invoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[source]¶
- invoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str, Any] | None = None, **kwargs: Any) LLMResponse
向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- async ainvoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[source]¶
- async ainvoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str Any] | None = None, **kwargs: Any) LLMResponse
异步向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- get_messages(input, message_history=None, system_instruction=None)[source]¶
构建 Ollama 聊天 API 的消息列表。
- 参数:
input (str)
message_history (Optional[Union[List[LLMMessage], MessageHistory]])
system_instruction (Optional[str])
- 返回类型:
Sequence[Message]
- get_messages_v2(input)[source]¶
构建 Ollama 聊天 API 的消息列表。
- 参数:
input (list[LLMMessage])
- 返回类型:
Sequence[Message]
- invoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]¶
向 LLM 发送带有工具定义的文本输入，并获取工具调用响应。 :param input: 发送给 LLM 的文本。 :type input: str :param tools: 供 LLM 选择的工具列表。 :type tools: List[Tool] :param message_history: 先前消息的集合，
每条消息都分配有特定的角色。
- 参数:
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
input (str)
tools (Sequence[Tool])
message_history (Optional[Union[List[LLMMessage], MessageHistory]])
- 返回:
来自 LLM 的包含工具调用的响应。
- 返回类型:
ToolCallResponse
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- async ainvoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]¶
向 LLM 发送带有工具定义的文本输入，并获取工具调用响应。 :param input: 发送给 LLM 的文本。 :type input: str :param tools: 供 LLM 选择的工具列表。 :type tools: List[Tool] :param message_history: 先前消息的集合，
每条消息都分配有特定的角色。
- 参数:
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
input (str)
tools (Sequence[Tool])
message_history (Optional[Union[List[LLMMessage], MessageHistory]])
- 返回:
来自 LLM 的包含工具调用的响应。
- 返回类型:
ToolCallResponse
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
VertexAILLM¶
- class neo4j_graphrag.llm.vertexai_llm.VertexAILLM(model_name='gemini-1.5-flash-001', model_params=None, system_instruction=None, rate_limit_handler=None, **kwargs)[source]¶
Vertex AI 上大语言模型的接口。
- 参数:
model_name (str, optional) – 要使用的 LLM 名称。默认为 “gemini-1.5-flash-001”。
model_params (Optional[dict], optional) – 当向模型发送文本时，传递给 LLMInterface(V1) 的额外参数。默认为 None。
system_instruction (Optional[str]) – 设置对话中模型行为和上下文的额外指令。默认为 None。
rate_limit_handler (Optional[RateLimitHandler], optional) – LLMInterface(V1) 的速率限制处理器。默认为 None。
**kwargs (Any) – 初始化类时传递给模型的参数。默认为 None。
- 抛出异常:
LLMGenerationError – 如果在从模型生成响应时发生错误。
示例
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?")
- invoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[source]¶
- invoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str, Any] | None = None, **kwargs: Any) LLMResponse
向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- async ainvoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[source]¶
- async ainvoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str Any] | None = None, **kwargs: Any) LLMResponse
异步向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- invoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]¶
向 LLM 发送带有工具定义的文本输入，并获取工具调用响应。
这是一个默认实现，支持工具/函数调用的 LLM 提供商应当重写此方法。
- 参数:
input (str) – 发送给 LLM 的文本。
tools (Sequence[Tool]) – 供 LLM 选择的 Tools 序列。每个 LLM 实现都应处理向其特定格式的转换。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的包含工具调用的响应。
- 返回类型:
ToolCallResponse
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
NotImplementedError – 如果 LLM 提供商不支持工具调用。
- async ainvoke_with_tools(input, tools, message_history=None, system_instruction=None)[source]¶
异步向 LLM 发送带有工具定义的文本输入，并获取工具调用响应。
这是一个默认实现，支持工具/函数调用的 LLM 提供商应当重写此方法。
- 参数:
input (str) – 发送给 LLM 的文本。
tools (Sequence[Tool]) – 供 LLM 选择的 Tools 序列。每个 LLM 实现都应处理向其特定格式的转换。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的包含工具调用的响应。
- 返回类型:
ToolCallResponse
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
NotImplementedError – 如果 LLM 提供商不支持工具调用。
AnthropicLLM¶
- class neo4j_graphrag.llm.anthropic_llm.AnthropicLLM(model_name, model_params=None, rate_limit_handler=None, **kwargs)[source]¶
Anthropic 上大语言模型的接口。
- 参数:
model_name (str, optional) – 要使用的 LLM 名称。默认为 “gemini-1.5-flash-001”。
model_params (Optional[dict], optional) – 当向模型发送文本时，传递给 LLMInterface(V1) 的额外参数。默认为 None。
system_instruction – Optional[str], optional): 设置对话中模型行为和上下文的额外指令。默认为 None。
rate_limit_handler (Optional[RateLimitHandler], optional) – LLMInterface(V1) 的速率限制管理处理器。默认为 None。
**kwargs (Any) – 初始化类时传递给模型的参数。默认为 None。
- 抛出异常:
LLMGenerationError – 如果在从模型生成响应时发生错误。
示例
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?")
- invoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[source]¶
- invoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str, Any] | None = None, **kwargs: Any) LLMResponse
向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- async ainvoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[source]¶
- async ainvoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str Any] | None = None, **kwargs: Any) LLMResponse
异步向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- get_messages(input, message_history=None)[source]¶
根据输入和消息历史为 LLM 构建消息列表。
- 参数:
input (str)
message_history (Optional[Union[List[LLMMessage], MessageHistory]])
- 返回类型:
Iterable[MessageParam]
CohereLLM¶
- class neo4j_graphrag.llm.cohere_llm.CohereLLM(model_name='', model_params=None, rate_limit_handler=None, **kwargs)[source]¶
Cohere 平台上大语言模型的接口。
- 参数:
model_name (str, optional) – 要使用的 LLM 名称。默认为 “gemini-1.5-flash-001”。
model_params (Optional[dict], optional) – 当向模型发送文本时，传递给 LLMInterface(V1) 的额外参数。默认为 None。
system_instruction (Optional[str], optional) – 设置对话中模型行为和上下文的额外指令。默认为 None。
rate_limit_handler (Optional[RateLimitHandler], optional) – LLMInterface(V1) 的速率限制处理器，用于管理 API 速率限制。默认为 None。
**kwargs (Any) – 初始化类时传递给模型的参数。默认为 None。
- 抛出异常:
LLMGenerationError – 如果在从模型生成响应时发生错误。
示例
from neo4j_graphrag.llm import CohereLLM llm = CohereLLM(api_key="...") llm.invoke("Say something")
- invoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[source]¶
- invoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str, Any] | None = None, **kwargs: Any) LLMResponse
向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- async ainvoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[source]¶
- async ainvoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str Any] | None = None, **kwargs: Any) LLMResponse
异步向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- get_messages(input, message_history=None, system_instruction=None)[源代码]¶
将输入和消息历史转换为 Cohere 的 ChatMessages。
- 参数:
input (str)
message_history (Optional[Union[List[LLMMessage], MessageHistory]])
system_instruction (Optional[str])
- 返回类型:
ChatMessages
- get_messages_v2(input)[源代码]¶
将 LLMMessage 列表转换为 Cohere 的 ChatMessages。
- 参数:
input (list[LLMMessage])
- 返回类型:
ChatMessages
MistralAILLM¶
- class neo4j_graphrag.llm.mistralai_llm.MistralAILLM(model_name, model_params=None, rate_limit_handler=None, **kwargs)[源代码]¶
- 参数:
model_name (str)
rate_limit_handler (Optional[RateLimitHandler])
kwargs (Any)
- invoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[源代码]¶
- invoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str, Any] | None = None, **kwargs: Any) LLMResponse
向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- async ainvoke(input: str, message_history: List[LLMMessage] | MessageHistory | None = None, system_instruction: str | None = None) LLMResponse[源代码]¶
- async ainvoke(input: List[LLMMessage], response_format: Type[BaseModel] | dict[str Any] | None = None, **kwargs: Any) LLMResponse
异步向 LLM 发送文本输入并获取响应。
- 参数:
input (str) – 发送给 LLM 的文本。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
system_instruction (Optional[str]) – 针对此调用的可选 LLM 系统消息覆盖。
- 返回:
来自 LLM 的响应。
- 返回类型:
- 抛出异常:
LLMGenerationError – 如果出现任何错误。
- get_messages(input, message_history=None, system_instruction=None)[源代码]¶
为 Mistral 聊天补全模型构建消息列表。
- 参数:
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)]]
- get_messages_v2(input)[源代码]¶
为 Mistral 聊天补全模型构建消息列表。
- 参数:
input (list[LLMMessage])
- 返回类型:
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)]]
速率限制¶
RateLimitHandler¶
- class neo4j_graphrag.utils.rate_limit.RateLimitHandler[源代码]¶
速率限制处理策略的抽象基类。
- abstract handle_async(func)[源代码]¶
将速率限制处理应用于异步函数。
- 参数:
func (AF) – 要进行速率限制处理封装的异步函数。
- 返回:
封装后的异步函数。
- 返回类型:
AF
- is_retryable_exception(exception)[源代码]¶
如果提供的异常应该重试，则返回 True。
默认行为将速率限制错误视为可重试。
重写此方法（或提供自定义处理器实现）以扩展重试分类，而无需修改库代码。
- to_retryable_error(exception)[源代码]¶
将提供商异常转换为可重试的错误类型。
默认行为通过
convert_to_rate_limit_error()将提供商速率限制错误转换为RateLimitError。重写此方法以封装额外的瞬态提供商错误。
- 参数:
exception (Exception)
- 返回类型:
RetryableError
RetryRateLimitHandler¶
NoOpRateLimitHandler¶
PromptTemplate¶
- class neo4j_graphrag.generation.prompts.PromptTemplate(template=None, expected_inputs=None, system_instructions=None)[源代码]¶
此类用于生成参数化提示（Prompt）。它由一个字符串（模板，使用 Python 格式语法，参数置于花括号 {} 中）和所需输入列表定义。在将指令发送给 LLM 之前，调用 format 方法，该方法将用提供的值替换参数。如果缺少任何预期的输入，将引发 PromptMissingInputError。
- 参数:
- format(*args, **kwargs)[源代码]¶
此方法用于用提供的值替换参数。参数必须通过以下方式提供：- 作为关键字参数 (kwargs) - 如果使用与预期输入相同的顺序，则作为位置参数 (args)。
示例
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: # '''
RagTemplate¶
ERExtractionTemplate¶
- class neo4j_graphrag.generation.prompts.ERExtractionTemplate(template=None, expected_inputs=None, system_instructions=None)[源代码]¶
- 参数:
- DEFAULT_TEMPLATE: str = '\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（字符串），并在定义关系时重复使用该 ID。\n务必遵守关系对应的源节点和目标节点类型以及关系方向。\n\n请确保遵守以下规则以生成有效的 JSON 对象：\n- 除了其中的 JSON 之外，不要返回任何额外信息。\n- 省略 JSON 周围的任何反引号 - 只需输出 JSON 本身。\n- JSON 对象不得封装在列表中 - 它是一个独立的 JSON 对象。\n- 属性名必须用双引号括起来\n\n示例：\n{examples}\n\n输入文本：\n\n{text}\n'¶
SchemaExtractionTemplate¶
- class neo4j_graphrag.generation.prompts.SchemaExtractionTemplate(template=None, expected_inputs=None, system_instructions=None)[源代码]¶
- 参数:
- DEFAULT_TEMPLATE: str = '\n你是一个顶尖算法，旨在提取结构化格式的标签属性图模式。\n\n根据输入文本生成通用的图谱模式。识别关键节点类型、它们的关系类型以及属性类型。\n\n重要规则：\n1. 仅返回抽象模式信息，而不是具体的实例。\n2. 节点类型使用单数 PascalCase 标签（例如：Person, Company, Product）。\n3. 关系类型使用 UPPER_SNAKE_CASE 标签（例如：WORKS_FOR, MANAGES）。\n4. 仅在可以自信推断类型时才包含属性定义，否则请省略。\n5. 定义模式时，确保提到的每个节点标签和关系标签都存在于你的节点类型和关系类型列表中。\n6. 不要创建文本中未明确提到的节点类型。\n7. 保持模式简洁，并专注于文本中清晰可辨的模式。\n8. 唯一性约束：\n8.1 唯一性（UNIQUENESS）是可选的；每个 node_type 可能有也可能没有恰好一个唯一性约束。\n8.2 仅使用样本中似乎没有太多缺失值的属性。\n8.3 约束按标签引用 node_types，并指定哪个属性是唯一的。\n8.4 如果属性出现在唯一性约束中，它必须也作为属性出现在相应的 node_type 中。\n9. 必填属性：\n9.1 如果该节点/关系类型的每个实例都必须具有此属性（非空），则将属性标记为 "required": true。\n9.2 如果属性是可选的，并且在某些实例上可能不存在，则将属性标记为 "required": false。\n9.3 作为标识符、名称或基本特征的属性通常是必填的。\n9.4 作为补充信息（电话号码、描述、元数据）的属性通常是可选的。\n9.5 不确定时，默认为 "required": false。\n9.6 如果属性具有 UNIQUENESS 约束，它必须被标记为 "required": true。\n10. 严禁在节点标签或关系类型中使用双下划线（__）作为前缀或后缀（例如禁止使用 __Person__ 或 __KNOWS__）。\n\n可接受的属性类型有：BOOLEAN, DATE, DURATION, FLOAT, INTEGER, LIST, LOCAL_DATETIME, LOCAL_TIME, POINT, STRING, ZONED_DATETIME, ZONED_TIME。\n\n返回遵循此精确结构的有效 JSON 对象：\n{{\n "node_types": [\n {{\n "label": "Person",\n "properties": [\n {{\n "name": "name",\n "type": "STRING",\n "required": true\n }},\n {{\n "name": "email",\n "type": "STRING",\n "required": false\n }}\n ]\n }}\n ...\n ],\n "relationship_types": [\n {{\n "label": "WORKS_FOR"\n }}\n ...\n ],\n "patterns": [\n {{"source": "Person", "relationship": "WORKS_FOR", "target": "Company"}},\n ...\n ],\n "constraints": [\n {{\n "type": "UNIQUENESS",\n "node_type": "Person",\n "property_name": "name"\n }}\n ...\n ]\n}}\n\n示例：\n{examples}\n\n输入文本：\n{text}\n'¶
Text2CypherTemplate¶
- class neo4j_graphrag.generation.prompts.Text2CypherTemplate(template=None, expected_inputs=None, system_instructions=None)[源代码]¶
- 参数:
RAG¶
GraphRAG¶
- class neo4j_graphrag.generation.graphrag.GraphRAG(retriever, llm, prompt_template=<neo4j_graphrag.generation.prompts.RagTemplate object>)[源代码]¶
使用特定的检索器（Retriever）和 LLM 执行 GraphRAG 搜索。
示例
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, LLMInterfaceV2 或 LangChain 聊天模型) – 用于生成答案的 LLM。
prompt_template (RagTemplate) – 将与上下文和用户问题一起格式化并传递给 LLM 的提示模板。
- 抛出异常:
RagInitializationError – 如果输入参数验证失败。
- search(query_text='', message_history=None, examples='', retriever_config=None, return_context=None, response_fallback=None)[源代码]¶
警告
- ‘return_context’ 的默认值将在未来版本中由 ‘False’
更改为 ‘True’。
- 此方法执行完整的 RAG 搜索：
检索（Retrieval）：上下文检索
增强（Augmentation）：提示词格式化
生成（Generation）：使用 LLM 生成答案
- 参数:
query_text (str) – 用户问题。
message_history (Optional[Union[List[LLMMessage], MessageHistory]]) – 先前消息的集合，每条消息都分配有特定的角色。
examples (str) – 添加到 LLM 提示词中的示例。
retriever_config (Optional[dict]) – 传递给检索器 search 方法的参数；例如：top_k
return_context (bool) – 是否将检索器结果附加到最终结果中（默认值：False）。
response_fallback (Optional[str]) – 如果不为空，当上下文为空时，将返回此消息而不调用 LLM。
- 返回:
LLM 生成的答案。
- 返回类型:
数据库交互¶
- neo4j_graphrag.indexes.create_vector_index(driver, name, label, embedding_property, dimensions, similarity_fn, fail_if_exists=False, neo4j_database=None)[源代码]¶
此方法构建 Cypher 查询并执行它，以在 Neo4j 中创建一个新的向量索引。
请参阅关于创建向量索引的 Cypher 手册。
确保提供的索引名称在数据库上下文中是唯一的。
示例
from neo4j import GraphDatabase from neo4j_graphrag.indexes import create_vector_index URI = "neo4j://: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)[源代码]¶
此方法构建 Cypher 查询并执行它，以在 Neo4j 中创建一个新的全文索引。
请参阅关于创建全文索引的 Cypher 手册。
确保提供的索引名称在数据库上下文中是唯一的。
示例
from neo4j import GraphDatabase from neo4j_graphrag.indexes import create_fulltext_index URI = "neo4j://: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, )
- 参数:
- 抛出异常:
ValueError – 如果输入参数验证失败。
neo4j.exceptions.ClientError – 如果全文索引创建失败。
- 返回类型:
None
- neo4j_graphrag.indexes.drop_index_if_exists(driver, name, neo4j_database=None)[源代码]¶
此方法构建 Cypher 查询并在索引存在的情况下执行它以在 Neo4j 中删除该索引。请参阅关于删除向量索引的 Cypher 手册。
示例
from neo4j import GraphDatabase from neo4j_graphrag.indexes import drop_index_if_exists URI = "neo4j://: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, )
- neo4j_graphrag.indexes.upsert_vectors(driver, ids, embedding_property, embeddings, neo4j_database=None, entity_type=EntityType.NODE)[源代码]¶
此方法构建 Cypher 查询并执行它，以在一组节点或关系上更新或插入（upsert）嵌入向量。
示例
from neo4j import GraphDatabase from neo4j_graphrag.indexes import upsert_vectors URI = "neo4j://: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’)。默认为 ‘NODE’。
- 抛出异常:
ValueError – 如果 ID 列表和嵌入列表的长度不匹配，或者嵌入向量的维度不一致。
Neo4jInsertionError – 如果在 Neo4j 中尝试更新向量时发生错误。
- 返回类型:
None
- neo4j_graphrag.indexes.upsert_vector(driver, node_id, embedding_property, vector, neo4j_database=None)[源代码]¶
警告
‘upsert_vector’ 已被弃用，并将在未来版本中删除，请改用 ‘upsert_vectors’。
此方法构建 Cypher 查询并执行它，以在特定节点上更新或插入（upsert）向量属性。
示例
from neo4j import GraphDatabase from neo4j_graphrag.indexes import upsert_vector URI = "neo4j://: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=..., )
- neo4j_graphrag.indexes.upsert_vector_on_relationship(driver, rel_id, embedding_property, vector, neo4j_database=None)[源代码]¶
警告
‘upsert_vector_on_relationship’ 已被弃用，并将在未来版本中删除，请改用 ‘upsert_vectors’。
此方法构建 Cypher 查询并执行它，以在特定关系上更新或插入（upsert）向量属性。
示例
from neo4j import GraphDatabase from neo4j_graphrag.indexes import upsert_vector_on_relationship URI = "neo4j://: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=..., )
- async neo4j_graphrag.indexes.async_upsert_vector(driver, node_id, embedding_property, vector, neo4j_database=None)[源代码]¶
警告
‘async_upsert_vector’ 已被弃用，并将在未来版本中删除。
此方法构建 Cypher 查询并异步执行，以在特定节点上更新或插入（upsert）向量属性。
示例
from neo4j import AsyncGraphDatabase from neo4j_graphrag.indexes import upsert_vector URI = "neo4j://: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=..., )
- async neo4j_graphrag.indexes.async_upsert_vector_on_relationship(driver, rel_id, embedding_property, vector, neo4j_database=None)[源代码]¶
警告
‘async_upsert_vector_on_relationship’ 已被弃用，并将在未来版本中删除。
此方法构建 Cypher 查询并异步执行，以在特定关系上更新或插入（upsert）向量属性。
示例
from neo4j import AsyncGraphDatabase from neo4j_graphrag.indexes import upsert_vector_on_relationship URI = "neo4j://: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=..., )
- neo4j_graphrag.indexes.retrieve_vector_index_info(driver, index_name, label_or_type, embedding_property, neo4j_database=None)[源代码]¶
检查 Neo4j 数据库中是否存在向量索引并返回其信息。如果未找到匹配的索引，则返回 None。
- 参数:
- 返回:
如果找到，则返回包含第一个匹配索引信息的字典，否则返回 None。
- 返回类型:
Optional[Dict[str, Any]]
- neo4j_graphrag.indexes.retrieve_fulltext_index_info(driver, index_name, label_or_type, text_properties=[], neo4j_database=None)[源代码]¶
检查 Neo4j 数据库中是否存在全文索引并返回其信息。如果未找到匹配的索引，则返回 None。
- 参数:
- 返回:
如果找到，则返回包含第一个匹配索引信息的字典，否则返回 None。
- 返回类型:
Optional[Dict[str, Any]]
- neo4j_graphrag.schema.get_structured_schema(driver, is_enhanced=False, database=None, timeout=None, sanitize=False, sample=1000)[查看源码]¶
返回图的结构化模式。
返回具有以下格式的字典
{ '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': ['Person'], '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 过程。
警告
某些标签会从输出模式中排除
由此包中的 KG Builder 流水线创建的 __Entity__ 和 __KGBuilder__ 节点标签
一些与 Bloom 内部相关的标签。
- 参数:
driver (neo4j.Driver) – Neo4j Python 驱动实例。
is_enhanced (bool) – 标志，指示是以详细统计信息 (True) 还是以更简单的概览格式 (False) 格式化模式。
database (Optional[str]) – 要连接的数据库名称。默认为 ‘neo4j’。
timeout (Optional[float]) – 事务的超时时间（以秒为单位）。用于终止长时间运行的查询。默认情况下未设置超时。
sanitize (bool) – 指示是否从结果中删除包含超过 128 个元素的列表的标志。用于从数据库响应中删除类似嵌入（embedding）的属性。默认为 False。
sample (int) – 为 apoc.meta.data 过程采样的节点数量。将 sample 设置为 -1 将取消采样。默认为 1000。
- 返回:
结构化格式的图模式信息。
- 返回类型:
- neo4j_graphrag.schema.get_schema(driver, is_enhanced=False, database=None, timeout=None, sanitize=False, sample=1000)[查看源码]¶
以具有以下格式的字符串形式返回图的模式
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) – 标志，指示是以详细统计信息 (True) 还是以更简单的概览格式 (False) 格式化模式。
database (Optional[str]) – 要连接的数据库名称。默认为 ‘neo4j’。
timeout (Optional[float]) – 事务的超时时间（以秒为单位）。用于终止长时间运行的查询。默认情况下未设置超时。
sanitize (bool) – 指示是否从结果中删除包含超过 128 个元素的列表的标志。用于从数据库响应中删除类似嵌入（embedding）的属性。默认为 False。
sample (int) – 为 apoc.meta.data 过程采样的节点数量。将 sample 设置为 -1 将取消采样。默认为 1000。
- 返回:
序列化格式的图模式信息。
- 返回类型:
消息历史记录¶
- class neo4j_graphrag.message_history.InMemoryMessageHistory(messages=None)[查看源码]¶
存储在内存中的消息历史记录
示例
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)
- 参数:
messages (Optional[List[LLMMessage]]) – 用于初始化历史记录的消息列表。默认为 None。
- class neo4j_graphrag.message_history.Neo4jMessageHistory(session_id, driver, window=None, database=None)[查看源码]¶
存储在 Neo4j 数据库中的消息历史记录
示例
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)