Pipeline from config file

examples/customize/build_graph/pipeline/simple_kg_pipeline_config.json

@@ -0,0 +1,57 @@
+{
+    "version_": "1",
+    "neo4j_config": {
+        "uri": {
+            "resolver_": "ENV",
+            "var_": "NEO4J_URI"
+        },
+        "user": {
+            "resolver_": "ENV",
+            "var_": "NEO4J_USER"
+        },
+        "password": {
+            "resolver_": "ENV",
+            "var_": "NEO4J_PASSWORD"
+        },
+        "database": {
+            "resolver_": "ENV",
+            "var_": "NEO4J_DATABASE"
+        }
+    },
+    "llm_config": {
+        "name_": "openai",
+        "class_": "OpenAILLM",
+        "params_": {
+            "api_key": {
+                "resolver_": "ENV",
+                "var_": "OPENAI_API_KEY"
+            },
+            "model_name": "gpt-4o"
+        }
+    },
+    "embedder_config": {
+        "name_": "openai",
+        "class_": "OpenAIEmbeddings",
+        "params_": {
+            "api_key": {
+                "resolver_": "ENV",
+                "var_": "OPENAI_API_KEY"
+            }
+        }
+    },
+    "from_pdf": false,
+    "entities": ["Person", {"label":  "Organization"}],
+    "relations": ["WORKS_FOR", {"label":  "DIRECTED_BY"}],
+    "potential_schema": [
+        ["Person", "WORKS_FOR", "Organization"],
+        ["Organization", "DIRECTED_BY", "Person"]
+    ],
+    "text_splitter": {
+        "class_": "fixed_size_splitter.FixedSizeSplitter",
+        "params_": {
+            "chunk_size":  100,
+            "chunk_overlap": 10
+        }
+    },
+    "perform_entity_resolution": false
+}

examples/customize/build_graph/pipeline/simple_kg_pipeline_from_config_file.py

@@ -0,0 +1,36 @@
+"""In this example, the pipeline is defined in a JSON file 'pipeline_config.json'.
+According to the configuration file, some parameters will be read from the env vars
+(Neo4j credentials and the OpenAI API key).
+"""
+
+import asyncio
+
+## If env vars in a .env file, uncomment:
+## (requires pip install python-dotenv)
+# from dotenv import load_dotenv
+# load_dotenv()
+# env vars manually set for testing:
+import os
+
+from neo4j_graphrag.experimental.pipeline.config.parser import SimpleKGPipelineBuilder
+from neo4j_graphrag.experimental.pipeline.pipeline import PipelineResult
+
+os.environ["NEO4J_URI"] = "bolt://localhost:7687"
+os.environ["NEO4J_USER"] = "neo4j"
+os.environ["NEO4J_PASSWORD"] = "password"
+# os.environ["OPENAI_API_KEY"] = "sk-..."
+
+
+# Text to process
+TEXT = """The son of Duke Leto Atreides and the Lady Jessica, Paul is the heir of House Atreides,
+an aristocratic family that rules the planet Caladan, the rainy planet, since 10191."""
+
+
+async def main() -> PipelineResult:
+    file_path = "examples/customize/build_graph/pipeline/simple_kg_pipeline_config.json"
+    pipeline = SimpleKGPipelineBuilder.from_config_file(file_path)
+    return await pipeline.run_async(text=TEXT)
+
+
+if __name__ == "__main__":
+    print(asyncio.run(main()))

src/neo4j_graphrag/experimental/components/schema.py

@@ -14,7 +14,7 @@
 #  limitations under the License.
 from __future__ import annotations
 
-from typing import Any, Dict, List, Literal, Optional, Tuple
+from typing import Any, Dict, List, Literal, Optional, Self, Tuple, Union
 
 from pydantic import BaseModel, ValidationError, model_validator, validate_call
 
@@ -55,6 +55,14 @@ class SchemaEntity(BaseModel):
     description: str = ""
     properties: List[SchemaProperty] = []
 
+    @classmethod
+    def from_text_or_dict(
+        cls, input: str | dict[str, Union[str, dict[str, str]]]
+    ) -> Self:
+        if isinstance(input, str):
+            return cls(label=input)
+        return cls.model_validate(input)
+
 
 class SchemaRelation(BaseModel):
     """
@@ -65,6 +73,14 @@ class SchemaRelation(BaseModel):
     description: str = ""
     properties: List[SchemaProperty] = []
 
+    @classmethod
+    def from_text_or_dict(
+        cls, input: str | dict[str, Union[str, dict[str, str]]]
+    ) -> Self:
+        if isinstance(input, str):
+            return cls(label=input)
+        return cls.model_validate(input)
+
 
 class SchemaConfig(DataModel):
     """

src/neo4j_graphrag/experimental/pipeline/config/__init__.py

@@ -0,0 +1,14 @@
+#  Copyright (c) "Neo4j"
+#  Neo4j Sweden AB [https://neo4j.com]
+#  #
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  #
+#      https://www.apache.org/licenses/LICENSE-2.0
+#  #
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.

src/neo4j_graphrag/experimental/pipeline/config/param_resolvers.py

@@ -0,0 +1,58 @@
+#  Copyright (c) "Neo4j"
+#  Neo4j Sweden AB [https://neo4j.com]
+#  #
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  #
+#      https://www.apache.org/licenses/LICENSE-2.0
+#  #
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import os
+from typing import Any, Optional
+
+from .types import ParamFromEnvConfig, ParamResolverEnum, ParamToResolveConfig
+
+
+class ParamResolver:
+    """A base class for all parameter resolvers."""
+
+    name: ParamResolverEnum
+
+    def resolve(self, param: ParamToResolveConfig) -> Any:
+        raise NotImplementedError
+
+
+class EnvParamResolver(ParamResolver):
+    """Resolve a parameter by reading its value
+    in the environment variables.
+
+    Example:
+
+    .. code-block:: python
+
+        import os
+        os.environ["MY_ENV_VAR"] = "LOCAL"
+
+        resolver = EnvParamResolver()
+        resolver.resolve("MY_ENV_VAR")
+        # Output: "LOCAL"
+    """
+
+    name = ParamResolverEnum.ENV
+
+    def resolve(self, param: ParamFromEnvConfig) -> Optional[str]:
+        return os.environ.get(param.var_)
+
+
+PARAM_RESOLVERS = {
+    resolver.name: resolver
+    for resolver in [
+        EnvParamResolver,
+    ]
+}