Feat/v1/project users

README.md

@@ -99,6 +99,16 @@ Aggregated → Reranked → Generated Answer
 
 Each branch gets the **exact same embedding model and storage path** used during indexing - automatically injected from Neo4j metadata.
 
+## Multi-Project Organization
+
+Organize pipelines into projects for better multi-tenancy:
+- **Hierarchy**: `User → Project → Pipelines` (e.g., `alice/rag_app/indexing_pipeline`)
+- **Storage**: Automatic path isolation at `data/{username}/{project}/{pipeline}/`
+- **Graph**: Components store project field for complete isolation
+- **Usage**: Simply add `project="my_app"` to factory and runner methods
+
+Example: `factory.build_pipeline_graphs_from_specs(username="alice", project="rag_app", ...)`
+
 ## Testing & Comparing Strategies
 
 This architecture enables systematic **comparison of different RAG strategies** using evaluation metrics:

agentic_rag/components/neo4j_manager.py

@@ -162,21 +162,39 @@ def get_component_nodes_by_ids(
             return [dict(r["c"]) for r in results]
 
     def get_components_by_pipeline(
-        self, pipeline_name: str, username: Optional[str] = None
+        self,
+        pipeline_name: str,
+        username: Optional[str] = None,
+        project: Optional[str] = None,
     ) -> List[Dict[str, Any]]:
         """
         Get all Component nodes for a specific pipeline.
 
         Args:
             pipeline_name: Name of the pipeline (e.g., 'index_1')
             username: Optional username filter for multi-tenant isolation
+            project: Optional project filter for multi-tenant isolation
 
         Returns:
             List of component node dictionaries with all properties
         """
         with self.driver.session(database=self.database) as session:
-            if username:
-                # Query with username filter for multi-tenant isolation
+            if username and project:
+                # Query with username and project filter through Project node
+                # Find components by traversing from Project or by matching project field directly
+                query = """
+                    MATCH (c:Component {pipeline_name: $pipeline_name, project: $project, author: $username})
+                    RETURN c
+                    ORDER BY c.id
+                """
+                results = session.run(
+                    query,
+                    pipeline_name=pipeline_name,
+                    username=username,
+                    project=project,
+                ).data()
+            elif username:
+                # Query with username filter only (backward compatible - searches all projects)
                 query = """
                     MATCH (c:Component {pipeline_name: $pipeline_name, author: $username})
                     RETURN c
@@ -207,28 +225,33 @@ def validate_user_exists(self, username: str) -> bool:
             return result is not None
 
     def get_pipeline_components_by_hash(
-        self, pipeline_hash: str, username: str
+        self, pipeline_hash: str, username: str, project: str = "default"
     ) -> List[Dict[str, object]]:
         """
         Traverse entire pipeline graph using DFS to get all connected components.
-        Only follows paths within the same pipeline.
+        Only follows paths within the same pipeline and project.
 
         Args:
             pipeline_hash: Single pipeline name/hash to load
             username: Username to validate permissions
+            project: Project name to filter by (defaults to "default")
 
         Returns:
             List of component dictionaries with all necessary data
         """
         with self.driver.session(database=self.database) as session:
             # First find the starting component(s) owned by the user for this pipeline
+            # Traverse through Project node using FLOWS_TO: User→Project→Component
             start_query = """
-                MATCH (u:User {username: $username})-[:OWNS]->(start:Component)
+                MATCH (u:User {username: $username})-[:OWNS]->(p:Project {name: $project})-[:FLOWS_TO]->(start:Component)
                 WHERE start.pipeline_name = $pipeline_hash
                 RETURN start.id AS start_id
             """
             start_results = session.run(
-                query=start_query, pipeline_hash=pipeline_hash, username=username
+                query=start_query,
+                pipeline_hash=pipeline_hash,
+                username=username,
+                project=project,
             ).data()
 
             if not start_results:

agentic_rag/config.py

@@ -146,6 +146,21 @@ def validate_lighthouse(self) -> bool:
         """
         return self.lighthouse_api_key is not None
 
+    def get_project_path(self, username: str, project: str) -> str:
+        """
+        Get storage path for a specific user project.
+
+        Args:
+            username: Username
+            project: Project name
+
+        Returns:
+            Path in format: {agentic_root_dir}/{username}/{project}/
+        """
+        import os
+
+        return os.path.join(self.agentic_root_dir, username, project)
+
     def to_dict(self) -> dict:
         """
         Export configuration as dictionary.

agentic_rag/pipeline/README.md

@@ -9,26 +9,29 @@ graph TB
     subgraph "Creation Time (Once)"
         User1[User] --> Factory[PipelineFactory]
         Factory --> Storage[GraphStorage]
-        Storage --> Neo4j[(Neo4j Database)]
+        Storage --> Neo4j[(Neo4j Graph)]
+        Neo4j --> Nodes["User → Project → Components"]
     end
 
     subgraph "Runtime (Many Times)"
         User2[User] --> Runner[PipelineRunner]
         Runner --> Neo4j
-        Runner --> Storage2[GraphStorage]
-        Storage2 --> Haystack[Haystack Pipeline]
+        Neo4j --> Components[Load Components]
+        Components --> Haystack[Build Haystack Pipeline]
         Haystack --> Results[Execution Results]
     end
-
-    subgraph "Legacy Path (Deprecated)"
-        User3[User] --> Runner2[PipelineRunner]
-        Runner2 -.-> Factory2[PipelineFactory]
-        Factory2 -.-> Storage3[GraphStorage]
-        Storage3 -.-> Neo4j
-        Storage3 -.-> Haystack2[Haystack Pipeline]
-    end
 ```
 
+## Project Hierarchy
+
+The system supports `User → Project → Pipelines` for multi-tenant organization:
+- **Graph**: `(User)-[:OWNS]->(Project)-[:FLOWS_TO]->(Component)-[:FLOWS_TO]->(Component)`
+- **Storage**: `data/{username}/{project}/{pipeline_name}/`
+- **Isolation**: Component IDs include project name, ensuring complete separation
+- **Usage**: Pass `project="my_app"` to factory and runner methods
+
+Example: `factory.build_pipeline_graphs_from_specs(username="alice", project="rag_app", ...)`
+
 ## Component Responsibilities
 
 ### 🏭 **PipelineFactory** (`factory.py`)
@@ -48,55 +51,56 @@ graph TB
 
 ### 🏃 **PipelineRunner** (`runner.py`)
 **Purpose**: Executes pipelines with data
-- **Preferred**: Load pre-built pipelines from Neo4j
-- **Legacy**: Create pipelines at runtime (deprecated)
+- Load pre-built pipelines from Neo4j
 - Execute indexing and retrieval operations
-- Handle pipeline input/output mapping
+- Handle multi-branch retrieval pipeline orchestration
+- Track metrics and performance
 
-## Usage Patterns
-
-### **Preferred: Build Once, Run Many**
+## Usage Pattern: Build Once, Run Many
 
 ```python
-# 1. BUILD TIME - Create pipeline once
-from agentic_rag.pipeline import PipelineFactory
-from agentic_rag.components import GraphStore
+from agentic_rag import Config, PipelineFactory, PipelineRunner
 
-graph_store = GraphStore()
-factory = PipelineFactory(graph_store)
+# Configuration
+config = Config(
+    neo4j_uri="bolt://localhost:7687",
+    neo4j_username="neo4j",
+    neo4j_password="password"
+)
 
-# Build and store in Neo4j
-factory.build_pipeline_graph(
-    components=[
-        {"type": "CONVERTER.PDF"},
+# 1. BUILD TIME - Create pipeline once
+factory = PipelineFactory(config=config)
+pipelines = factory.build_pipeline_graphs_from_specs(
+    pipeline_specs=[[
+        {"type": "CONVERTER.TEXT"},
         {"type": "CHUNKER.MARKDOWN_AWARE"},
-        {"type": "EMBEDDER.SENTENCE_TRANSFORMERS"},
-        {"type": "WRITER.CHROMA_DOCUMENT"}
-    ],
-    pipeline_name="pdf_indexing_pipeline"
+        {"type": "EMBEDDER.SENTENCE_TRANSFORMERS_DOC"},
+        {"type": "WRITER.CHROMA_DOCUMENT_WRITER"}
+    ]],
+    username="alice",
+    project="rag_app",  # Project organization
+    configs=[{"_pipeline_name": "my_pipeline"}]
 )
 
 # 2. RUNTIME - Load and execute many times
-from agentic_rag.pipeline import PipelineRunner
-
-runner = PipelineRunner(graph_store=graph_store)
-runner.load_from_graph("pdf_indexing_pipeline")  # Fast loading from Neo4j
+runner = PipelineRunner(config=config)
+runner.load_pipelines(
+    pipeline_names=["my_pipeline"],
+    username="alice",
+    project="rag_app"
+)
 
 # Execute multiple times
-results1 = runner.run("indexing", {"documents": documents1})
-results2 = runner.run("indexing", {"documents": documents2})
-```
-
-### **Legacy: Create at Runtime (Deprecated)**
-
-```python
-# Creates graph every time - inefficient
-runner = PipelineRunner()
-runner.load_pipeline(components, "my_pipeline")  # Slow - rebuilds graph
-results = runner.run("indexing", data)
+result = runner.run(
+    pipeline_name="my_pipeline",
+    username="alice",
+    project="rag_app",
+    type="indexing",
+    data_path="./documents"
+)
 ```
 
-## Component Interactions
+## Pipeline Flow
 
 ### Creation Flow
 ```mermaid
@@ -106,61 +110,33 @@ sequenceDiagram
     participant S as GraphStorage
     participant N as Neo4j
 
-    U->>F: build_pipeline_graph(components, name)
+    U->>F: build_pipeline_graphs_from_specs(username, project)
     F->>F: Parse & validate components
-    F->>F: Create PipelineSpec
-    F->>S: build_pipeline_graph(spec)
-    S->>S: Determine connections
-    S->>S: Apply component substitutions
-    S->>N: Store components & relationships
+    F->>F: Create PipelineSpec objects
+    F->>S: create_pipeline_graph(spec, project)
+    S->>N: Store User → Project → Components
+    S->>N: Create FLOWS_TO relationships
     S-->>F: Graph created
-    F-->>U: PipelineSpec
+    F-->>U: List[PipelineSpec]
 ```
 
-### Runtime Flow (Preferred)
+### Execution Flow
 ```mermaid
 sequenceDiagram
     participant U as User
     participant R as PipelineRunner
     participant N as Neo4j
-    participant H as Haystack
-
-    U->>R: load_from_graph("pipeline_name")
-    R->>N: Query pipeline components
-    N-->>R: Component data & connections
-    R->>R: Reconstruct PipelineSpec
-    R->>H: Build Haystack pipeline
-    H-->>R: Ready pipeline
+    participant H as Haystack Pipeline
 
-    U->>R: run("indexing", data)
-    R->>H: Execute with data
-    H-->>R: Results
-    R-->>U: Execution results
-```
+    U->>R: load_pipelines(names, username, project)
+    R->>N: Query: User → Project → Components
+    N-->>R: Component metadata & connections
+    R->>R: Build Haystack components from metadata
+    R->>H: Create connected pipeline
+    H-->>R: Ready
 
-### Runtime Flow (Legacy)
-```mermaid
-sequenceDiagram
-    participant U as User
-    participant R as PipelineRunner
-    participant F as PipelineFactory
-    participant S as GraphStorage
-    participant N as Neo4j
-    participant H as Haystack
-
-    U->>R: load_pipeline(components, name)
-    R->>F: build_pipeline_graph(components, name)
-    F->>S: build_pipeline_graph(spec)
-    S->>N: Store graph (expensive!)
-    S-->>F: Success
-    F-->>R: PipelineSpec
-    R->>S: build_haystack_pipeline(spec)
-    S->>H: Create pipeline
-    H-->>S: Ready pipeline
-    S-->>R: Haystack pipeline
-
-    U->>R: run("indexing", data)
-    R->>H: Execute with data
+    U->>R: run(pipeline_name, username, project, type)
+    R->>H: Execute with input data
     H-->>R: Results
     R-->>U: Execution results
 ```
@@ -178,30 +154,9 @@ pipeline/
 
 ## Key Benefits
 
-1. **Performance**: Build once, run many times
-2. **Persistence**: Pipelines survive application restarts
-3. **Scalability**: No graph rebuilding at runtime
-4. **Flexibility**: Load any stored pipeline by name
-5. **Clean Separation**: Creation vs execution concerns
-6. **Component Substitutions**: Automatic writer→retriever conversion for retrieval pipelines
-
-## Migration Guide
-
-**Old Pattern (Deprecated)**:
-```python
-runner = PipelineRunner()
-runner.load_pipeline(components, "my_pipeline")  # Slow
-results = runner.run("indexing", data)
-```
-
-**New Pattern (Recommended)**:
-```python
-# Build once (setup/deployment time)
-factory = PipelineFactory(graph_store)
-factory.build_pipeline_graph(components, "my_pipeline")
-
-# Run many times (application runtime)
-runner = PipelineRunner(graph_store=graph_store)
-runner.load_from_graph("my_pipeline")  # Fast
-results = runner.run("indexing", data)
-```
+1. **Multi-Tenancy**: User → Project → Pipelines hierarchy for complete isolation
+2. **Performance**: Build once, run many times - no graph rebuilding at runtime
+3. **Persistence**: Pipelines stored in Neo4j, survive application restarts
+4. **Flexibility**: Query multiple indexing pipelines in parallel for retrieval
+5. **Auto-Orchestration**: Retrieval pipelines auto-inject embedders/retrievers from indexing metadata
+6. **Path Isolation**: Automatic storage at `data/{username}/{project}/{pipeline}/`