From 382add8e8cc7fd9abb93ad57c7b6a8a05a3e50ab Mon Sep 17 00:00:00 2001 From: Vardhan Shorewala Date: Thu, 13 Nov 2025 17:55:57 -0800 Subject: [PATCH 1/3] feat: add core logic for project creation for a user --- agentic_rag/components/neo4j_manager.py | 37 +++++-- agentic_rag/config.py | 15 +++ agentic_rag/pipeline/factory.py | 47 ++++++--- agentic_rag/pipeline/runner.py | 43 ++++++-- agentic_rag/pipeline/storage.py | 43 ++++++-- agentic_rag/types/__init__.py | 2 + agentic_rag/types/graph_relationships.py | 12 +-- agentic_rag/types/node_types.py | 49 ++++++++- examples/indexing_pipeline_example.py | 13 ++- examples/retrieval_pipeline_example.py | 13 ++- tests/test_graph_pipeline.py | 125 +++++++++++++++++++++-- 11 files changed, 329 insertions(+), 70 deletions(-) diff --git a/agentic_rag/components/neo4j_manager.py b/agentic_rag/components/neo4j_manager.py index e311564..d9e63a1 100644 --- a/agentic_rag/components/neo4j_manager.py +++ b/agentic_rag/components/neo4j_manager.py @@ -162,7 +162,10 @@ 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. @@ -170,13 +173,28 @@ def get_components_by_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: diff --git a/agentic_rag/config.py b/agentic_rag/config.py index c3b2098..3accea0 100644 --- a/agentic_rag/config.py +++ b/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. diff --git a/agentic_rag/pipeline/factory.py b/agentic_rag/pipeline/factory.py index aa48b47..c7989ec 100644 --- a/agentic_rag/pipeline/factory.py +++ b/agentic_rag/pipeline/factory.py @@ -100,6 +100,7 @@ def build_pipeline_graphs_from_specs( self, pipeline_specs: List[List[Dict[str, str]]], username: str, + project: str = "default", configs: Optional[List[Dict[str, Any]]] = None, pipeline_types: Optional[List[str]] = None, ) -> List[PipelineSpec]: @@ -110,6 +111,7 @@ def build_pipeline_graphs_from_specs( pipeline_specs: List of component specifications as dicts. Example: [[{"type": "CONVERTER.PDF"}, {"type": "CHUNKER.RECURSIVE"}]] username: Username for multi-tenant isolation + project: Project name (defaults to "default") configs: Optional list of configuration dicts for each pipeline pipeline_types: Optional list of pipeline types ("indexing" or "retrieval") Defaults to "indexing" for all pipelines @@ -132,7 +134,7 @@ def build_pipeline_graphs_from_specs( ) self.logger.info( - f"Building {len(pipeline_specs)} pipeline graphs for user: {username}" + f"Building {len(pipeline_specs)} pipeline graphs for user: {username}, project: {project}" ) pipeline_specs_list = [] @@ -149,7 +151,7 @@ def build_pipeline_graphs_from_specs( pipeline_name = config.get("_pipeline_name", f"pipeline_{i}") self.logger.debug(f"Building {pipeline_type} pipeline {i}: {pipeline_name}") pipeline_spec = self.build_pipeline_graph( - spec, pipeline_name, username, config, pipeline_type + spec, pipeline_name, username, project, config, pipeline_type ) pipeline_specs_list.append(pipeline_spec) @@ -161,6 +163,7 @@ def build_pipeline_graph( component_specs: List[Dict[str, str]], pipeline_name: str, username: str, + project: str = "default", config: Optional[Dict[str, Any]] = None, pipeline_type: str = "indexing", ) -> PipelineSpec: @@ -172,6 +175,7 @@ def build_pipeline_graph( Example: [{"type": "CONVERTER.PDF"}, {"type": "CHUNKER.RECURSIVE"}] pipeline_name: Name for the pipeline username: Username for multi-tenant isolation + project: Project name (defaults to "default") config: Optional configuration dict pipeline_type: Type of pipeline - "indexing" or "retrieval" (default: "indexing") @@ -183,11 +187,11 @@ def build_pipeline_graph( # Route to appropriate builder based on pipeline type if pipeline_type == "indexing": return self._build_indexing_pipeline( - component_specs, pipeline_name, username, config + component_specs, pipeline_name, username, project, config ) elif pipeline_type == "retrieval": return self._build_retrieval_pipeline( - component_specs, pipeline_name, username, config + component_specs, pipeline_name, username, project, config ) else: raise ValueError( @@ -199,6 +203,7 @@ def _build_indexing_pipeline( component_specs: List[Dict[str, str]], pipeline_name: str, username: str, + project: str, config: Dict[str, Any], branch_id: Optional[str] = None, pipeline_type: Optional[PipelineType] = None, @@ -210,6 +215,7 @@ def _build_indexing_pipeline( component_specs: List of component specifications pipeline_name: Name for the pipeline username: Username for multi-tenant isolation + project: Project name config: Configuration dict branch_id: Optional branch identifier for retrieval pipeline branches pipeline_type: Optional pipeline type override @@ -218,7 +224,7 @@ def _build_indexing_pipeline( PipelineSpec for indexing pipeline """ self.logger.info( - f"Building indexing pipeline: {pipeline_name} for user: {username}" + f"Building indexing pipeline: {pipeline_name} for user: {username}, project: {project}" ) # Parse component specifications and validate @@ -241,7 +247,9 @@ def _build_indexing_pipeline( user_config = user_config.copy() # Don't modify original config # Use agentic_root_dir from config if available root_dir = self.config.agentic_root_dir if self.config else "./data" - user_config["root_dir"] = f"{root_dir}/{username}/{pipeline_name}" + user_config["root_dir"] = ( + f"{root_dir}/{username}/{project}/{pipeline_name}" + ) self.logger.debug( f"Auto-generated root_dir for chroma_document_writer: {user_config['root_dir']}" ) @@ -265,7 +273,9 @@ def _build_indexing_pipeline( self.logger.info( f"Creating graph representation for indexing pipeline '{pipeline_name}'" ) - self.graph_storage.build_pipeline_graph(pipeline_spec, username, branch_id) + self.graph_storage.build_pipeline_graph( + pipeline_spec, username, project, branch_id + ) else: self.logger.warning("No graph store configured, pipeline graph not created") @@ -293,7 +303,7 @@ def _extract_indexing_pipelines(self, config: Dict[str, Any]) -> List[str]: return indexing_pipelines def _fetch_indexing_pipeline_components( - self, indexing_pipelines: List[str], username: str + self, indexing_pipelines: List[str], username: str, project: str ) -> Dict[str, List[Dict[str, Any]]]: """ Step 2: Fetch embedder and writer components from each indexing pipeline. @@ -303,6 +313,7 @@ def _fetch_indexing_pipeline_components( Args: indexing_pipelines: List of pipeline names to fetch components from username: Username for multi-tenant isolation + project: Project name for isolation """ if not self.graph_store: raise RuntimeError("Cannot build retrieval pipeline without a graph store") @@ -312,14 +323,14 @@ def _fetch_indexing_pipeline_components( for indexing_pipeline_name in indexing_pipelines: self.logger.debug(f"Querying components for: {indexing_pipeline_name}") - # Get all components for this pipeline + # Get all components for this pipeline with project filtering all_components = self.graph_store.get_components_by_pipeline( - pipeline_name=indexing_pipeline_name, username=username + pipeline_name=indexing_pipeline_name, username=username, project=project ) if not all_components: raise ValueError( - f"No components found for '{indexing_pipeline_name}' (user: {username})" + f"No components found for '{indexing_pipeline_name}' (user: {username}, project: {project})" ) # Filter for embedder and writer components (needed for retrieval) @@ -338,11 +349,10 @@ def _fetch_indexing_pipeline_components( relevant_components = [c for c in [embedder, writer] if c is not None] if not relevant_components: - self.logger.warning( - f"No embedder/writer components found for '{indexing_pipeline_name}'" + raise ValueError( + f"No embedder/writer components found for '{indexing_pipeline_name}' in project '{project}'. " + f"Make sure the indexing pipeline exists in the same project before creating a retrieval pipeline." ) - indexing_pipeline_components[indexing_pipeline_name] = [] - continue indexing_pipeline_components[indexing_pipeline_name] = relevant_components @@ -480,6 +490,7 @@ def _build_retrieval_pipeline( component_specs: List[Dict[str, str]], pipeline_name: str, username: str, + project: str, config: Dict[str, Any], ) -> PipelineSpec: """ @@ -493,6 +504,7 @@ def _build_retrieval_pipeline( component_specs: List of component specifications (e.g., generator) pipeline_name: Name for the pipeline username: Username for multi-tenant isolation + project: Project name config: Configuration dict with optional "_indexing_pipelines" key Returns: @@ -505,7 +517,7 @@ def _build_retrieval_pipeline( } """ self.logger.info( - f"Building retrieval pipeline: {pipeline_name} for user: {username}" + f"Building retrieval pipeline: {pipeline_name} for user: {username}, project: {project}" ) # Step 1: Extract indexing pipeline names @@ -513,7 +525,7 @@ def _build_retrieval_pipeline( # Step 2: Fetch embedder and writer components indexing_pipeline_components = self._fetch_indexing_pipeline_components( - indexing_pipelines, username + indexing_pipelines, username, project ) # Step 3: Build component specs for each pipeline @@ -540,6 +552,7 @@ def _build_retrieval_pipeline( component_specs=pipeline_spec, pipeline_name=pipeline_name, username=username, + project=project, config=pipeline_config, branch_id=indexing_pipeline_name, pipeline_type=PipelineType.RETRIEVAL, diff --git a/agentic_rag/pipeline/runner.py b/agentic_rag/pipeline/runner.py index 855056e..a33e1f3 100644 --- a/agentic_rag/pipeline/runner.py +++ b/agentic_rag/pipeline/runner.py @@ -81,21 +81,26 @@ def reset_instance(cls) -> None: cls._instance = None cls._initialized = False - def load_pipelines(self, pipeline_names: List[str], username: str) -> None: + def load_pipelines( + self, pipeline_names: List[str], username: str, project: str = "default" + ) -> None: """ Load and build pipelines for a specific user. Args: pipeline_names: List of pipeline names to load. username: Username to load pipelines for + project: Project name (defaults to "default") """ - self.logger.info(f"Loading pipelines for user {username}: {pipeline_names}") + self.logger.info( + f"Loading pipelines for user {username}, project {project}: {pipeline_names}" + ) for pipeline_name in pipeline_names: try: # Load pipeline graph self.logger.debug(f"Loading pipeline graph: {pipeline_name}") - self.load_pipeline_graph([pipeline_name], username) + self.load_pipeline_graph([pipeline_name], username, project) # Build Haystack components self.logger.debug(f"Building Haystack components: {pipeline_name}") @@ -154,7 +159,9 @@ def _get_pipeline_type(self, pipeline_name: str) -> str: return pipeline_type - def load_pipeline_graph(self, pipeline_hashes: List[str], username: str) -> None: + def load_pipeline_graph( + self, pipeline_hashes: List[str], username: str, project: str = "default" + ) -> None: """ Load pipeline metadata from Neo4j and store in _pipeline_graphs. @@ -163,6 +170,8 @@ def load_pipeline_graph(self, pipeline_hashes: List[str], username: str) -> None Args: pipeline_hashes: List of pipeline names to load (e.g., ['retrieval_pipeline']) + username: Username for permissions + project: Project name to filter by (defaults to "default") Raises: RuntimeError: If no graph store is configured @@ -232,7 +241,7 @@ def load_pipeline_graph(self, pipeline_hashes: List[str], username: str) -> None graph_storage = GraphStorage(self.graph_store, registry) pipelines_data = graph_storage.load_pipeline_by_hashes( - pipeline_hashes, username + pipeline_hashes, username, project ) # Store the graph representations for all loaded pipelines @@ -691,7 +700,14 @@ def create_haystack_pipeline_retrieval( return branch_pipelines - def run(self, pipeline_name: str, username: str, type: str, **kwargs: Any) -> Any: + def run( + self, + pipeline_name: str, + username: str, + type: str, + project: str = "default", + **kwargs: Any, + ) -> Any: """ Run a pipeline by name, dispatching to the appropriate execution method. @@ -699,22 +715,27 @@ def run(self, pipeline_name: str, username: str, type: str, **kwargs: Any) -> An pipeline_name: Name of the pipeline to run (e.g., 'pdf_indexing_pipeline', 'pdf_retrieval_pipeline') username: Username for metrics and logging type: Pipeline type - "indexing" or "retrieval" + project: Project name (defaults to "default") **kwargs: Pipeline-specific arguments Returns: Pipeline execution results """ if type == "indexing" or type == PipelineUsage.INDEXING.value: - return self._run_indexing_pipeline(pipeline_name, username, **kwargs) + return self._run_indexing_pipeline( + pipeline_name, username, project, **kwargs + ) elif type == "retrieval" or type == PipelineUsage.RETRIEVAL.value: - return self._run_retrieval_pipeline(pipeline_name, username, **kwargs) + return self._run_retrieval_pipeline( + pipeline_name, username, project, **kwargs + ) else: raise ValueError( f"Unknown pipeline type: {type}. " "Must be 'indexing' or 'retrieval'" ) def _run_indexing_pipeline( - self, pipeline_name: str, username: str, **kwargs: Any + self, pipeline_name: str, username: str, project: str = "default", **kwargs: Any ) -> Any: """ Execute an indexing pipeline. @@ -722,6 +743,7 @@ def _run_indexing_pipeline( Args: pipeline_name: Name of the indexing pipeline username: Username for metrics and logging + project: Project name (defaults to "default") **kwargs: Pipeline-specific arguments - data_path: Path to directory containing PDFs (required) - sources: Optional list of specific file paths to process @@ -826,7 +848,7 @@ def _run_indexing_pipeline( raise def _run_retrieval_pipeline( - self, pipeline_name: str, username: str, **kwargs: Any + self, pipeline_name: str, username: str, project: str = "default", **kwargs: Any ) -> Dict[str, Any]: """ Execute all branches of a retrieval pipeline and aggregate results. @@ -834,6 +856,7 @@ def _run_retrieval_pipeline( Args: pipeline_name: Name of the retrieval pipeline username: Username for metrics and logging + project: Project name (defaults to "default") **kwargs: Must include 'query' (str) Returns: diff --git a/agentic_rag/pipeline/storage.py b/agentic_rag/pipeline/storage.py index c0fe777..187c59a 100644 --- a/agentic_rag/pipeline/storage.py +++ b/agentic_rag/pipeline/storage.py @@ -11,6 +11,7 @@ GraphRelationship, PipelineSpec, PipelineType, + ProjectNode, UserNode, create_haystack_component, ) @@ -61,6 +62,7 @@ def create_pipeline_graph( spec: PipelineSpec, connections: List[Tuple[str, str]], username: str, + project: str = "default", branch_id: Optional[str] = None, ) -> None: """Create graph representation of the pipeline components. @@ -69,6 +71,7 @@ def create_pipeline_graph( spec: Pipeline specification connections: List of (source, target) component connections username: Username for pipeline ownership + project: Project name (defaults to "default") branch_id: Optional branch identifier for retrieval pipeline branches """ @@ -78,6 +81,7 @@ def create_pipeline_graph( node = ComponentNode( component_name=component_spec.name, pipeline_name=spec.name, + project=project, version="1.0.0", author=username, component_config=component_spec.get_config(), @@ -93,29 +97,47 @@ def create_pipeline_graph( # Add/update the owning user node self.logger.info( - f"Creating pipeline graph for user '{username}', pipeline '{spec.name}'" + f"Creating pipeline graph for user '{username}', project '{project}', pipeline '{spec.name}'" ) user_node = UserNode(username=username, display_name=username.title()) user_dict = user_node.to_dict() self.graph_store.add_nodes_batch([user_dict], "User") + # Add/update the project node + project_node = ProjectNode(name=project, username=username) + project_dict = project_node.to_dict() + self.graph_store.add_nodes_batch([project_dict], "Project") + + # Connect user to project (User -[:OWNS]-> Project) + self.graph_store.add_edges_batch( + [ + ( + user_dict["id"], + project_dict["id"], + GraphRelationship.OWNS.value, + ) + ], + source_label="User", + target_label="Project", + ) + # Add component nodes self.logger.debug(f"Adding {len(nodes)} component nodes to graph") self.graph_store.add_nodes_batch(nodes, "Component") - # Connect user to the first component in the pipeline + # Connect project to the first component (Project -[:FLOWS_TO]-> Component) if spec.components: first_component_id = node_id_by_name.get(spec.components[0].name) if first_component_id: self.graph_store.add_edges_batch( [ ( - user_dict["id"], + project_dict["id"], first_component_id, - GraphRelationship.OWNS.value, + GraphRelationship.FLOWS_TO.value, ) ], - source_label="User", + source_label="Project", target_label="Component", ) @@ -147,6 +169,7 @@ def build_pipeline_graph( self, spec: PipelineSpec, username: str = "test_user", + project: str = "default", branch_id: Optional[str] = None, ) -> None: """Build a graph representation of the pipeline specification. @@ -154,6 +177,7 @@ def build_pipeline_graph( Args: spec: Pipeline specification username: Username for pipeline ownership (defaults to "test_user") + project: Project name (defaults to "default") branch_id: Optional branch identifier for retrieval pipeline branches """ @@ -161,7 +185,7 @@ def build_pipeline_graph( connections = self._determine_connections(spec.components) # Create graph representation - self.create_pipeline_graph(spec, connections, username, branch_id) + self.create_pipeline_graph(spec, connections, username, project, branch_id) def build_haystack_pipeline(self, spec: PipelineSpec) -> Any: """Build a Haystack pipeline from a pipeline specification.""" @@ -203,7 +227,7 @@ def _determine_connections( return connections def load_pipeline_by_hashes( - self, pipeline_hashes: List[str], username: str + self, pipeline_hashes: List[str], username: str, project: str = "default" ) -> Dict[str, List[Dict[str, Any]]]: """ Retrieve components for each pipeline hash from Neo4j. @@ -211,6 +235,7 @@ def load_pipeline_by_hashes( Args: pipeline_hashes: List of pipeline names to load username: Username for permissions + project: Project name to filter by (defaults to "default") Returns: Dictionary mapping pipeline names to their component data @@ -227,7 +252,9 @@ def load_pipeline_by_hashes( # Call Neo4j for this specific pipeline hash (single hash method) component_data_list = self.graph_store.get_pipeline_components_by_hash( - pipeline_hash, username # Single pipeline hash + pipeline_hash, + username, + project, # Single pipeline hash with project filter ) print(f" Found {len(component_data_list)} components") diff --git a/agentic_rag/types/__init__.py b/agentic_rag/types/__init__.py index b65d989..fccb518 100644 --- a/agentic_rag/types/__init__.py +++ b/agentic_rag/types/__init__.py @@ -31,6 +31,7 @@ ComponentRelationship, DataPiece, ProcessedByRelationship, + ProjectNode, TransformedByRelationship, UserNode, ) @@ -48,6 +49,7 @@ "ComponentNode", "ComponentRelationship", "UserNode", + "ProjectNode", # DataPiece types (for InGate/OutGate caching) "DataPiece", "TransformedByRelationship", diff --git a/agentic_rag/types/graph_relationships.py b/agentic_rag/types/graph_relationships.py index ba4271f..8a63fc6 100644 --- a/agentic_rag/types/graph_relationships.py +++ b/agentic_rag/types/graph_relationships.py @@ -6,18 +6,16 @@ class GraphRelationship(Enum): """Defines all relationship types used in the Neo4j graph.""" - # Component to Component relationships - FLOWS_TO = "FLOWS_TO" # Data flows from one component to another + # Flow relationships + FLOWS_TO = "FLOWS_TO" # Data/control flows from one node to another + # Used for: Component->Component, Project->Component # Component to DocumentStore relationships WRITES_TO = "WRITES_TO" # Writer components write to DocumentStores READS_FROM = "READS_FROM" # Retriever components read from DocumentStores - # User to Pipeline relationships - OWNS = "OWNS" # User owns/created a pipeline - - # Pipeline to Component relationships - CONTAINS = "CONTAINS" # Pipeline contains components + # User to Project relationships + OWNS = "OWNS" # User owns/created a project # DataPiece transformation relationships (for InGate/OutGate) TRANSFORMED_BY = "TRANSFORMED_BY" # DataPiece transformed to another DataPiece diff --git a/agentic_rag/types/node_types.py b/agentic_rag/types/node_types.py index 5551c13..514cf81 100644 --- a/agentic_rag/types/node_types.py +++ b/agentic_rag/types/node_types.py @@ -14,6 +14,7 @@ class ComponentNode: version: str author: str component_config: Dict[str, Any] + project: str = "default" # Project name for multi-tenant isolation component_type: Optional[str] = None # e.g., "EMBEDDER.SENTENCE_TRANSFORMERS_DOC" pipeline_type: Optional[str] = None # "indexing" or "retrieval" branch_id: Optional[str] = ( @@ -25,11 +26,11 @@ class ComponentNode: def __post_init__(self) -> None: """Generate ID and cache_key if not provided.""" if self.id is None: - # Create deterministic hash from: component_name__pipeline_name__version__author__branch_id + # Create deterministic hash from: component_name__pipeline_name__project__version__author__branch_id import hashlib import json - combined = f"{self.component_name}__{self.pipeline_name}__{self.version}__{self.author}" + combined = f"{self.component_name}__{self.pipeline_name}__{self.project}__{self.version}__{self.author}" # Include branch_id if provided (for retrieval pipeline branches) if self.branch_id: @@ -63,6 +64,7 @@ def to_dict(self) -> Dict[str, Any]: "id": self.id, "component_name": self.component_name, "pipeline_name": self.pipeline_name, + "project": self.project, "version": self.version, "author": self.author, "component_config_json": config_json, @@ -91,6 +93,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "ComponentNode": return cls( component_name=data["component_name"], pipeline_name=data["pipeline_name"], + project=data.get("project", "default"), version=data["version"], author=data["author"], component_config=component_config, @@ -146,6 +149,48 @@ def to_dict(self) -> Dict[str, Any]: } +@dataclass +class ProjectNode: + """Represents a project that contains pipelines.""" + + name: str # Project name (e.g., "my_rag_app") + username: str # Owner username + description: Optional[str] = None + id: Optional[str] = None + created_at: Optional[datetime] = None + + def __post_init__(self) -> None: + """Generate ID from username and project name.""" + if self.id is None: + import hashlib + + # Create deterministic ID: project_{username}_{name} + combined = f"{self.username}__{self.name}" + hash_obj = hashlib.sha256(combined.encode("utf-8")) + self.id = f"proj_{hash_obj.hexdigest()[:12]}" + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary for Neo4j insertion.""" + result = { + "id": self.id, + "name": self.name, + "username": self.username, + } + if self.description: + result["description"] = self.description + return result + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "ProjectNode": + """Create ProjectNode from dictionary.""" + return cls( + name=data["name"], + username=data["username"], + description=data.get("description"), + id=data.get("id"), + ) + + @dataclass class DataPiece: """ diff --git a/examples/indexing_pipeline_example.py b/examples/indexing_pipeline_example.py index a86a34d..2f8378b 100644 --- a/examples/indexing_pipeline_example.py +++ b/examples/indexing_pipeline_example.py @@ -31,6 +31,7 @@ # Configuration USERNAME = "your_username" +PROJECT = "demo_rag_app" # Project name for organizing pipelines # Pipeline names for different indexing strategies FAST_PIPELINE = "fast_retrieval_index" @@ -115,10 +116,11 @@ def create_indexing_pipelines() -> List: # Build both pipelines and store them in Neo4j # Each pipeline gets its own ChromaDB collection automatically - # Username is now injected at method level for multi-tenant isolation + # Username and project are injected at method level for multi-tenant isolation pipelines = factory.build_pipeline_graphs_from_specs( pipeline_specs=pipeline_specs, username=USERNAME, + project=PROJECT, configs=configs, pipeline_types=["indexing", "indexing"], ) @@ -157,20 +159,23 @@ def run_indexing_pipelines(data_directory: str) -> Dict[str, Any]: config=config, ) - # Load pipelines with username injection + # Load pipelines with username and project injection runner.load_pipelines( - pipeline_names=[FAST_PIPELINE, SEMANTIC_PIPELINE], username=USERNAME + pipeline_names=[FAST_PIPELINE, SEMANTIC_PIPELINE], + username=USERNAME, + project=PROJECT, ) results = {} # Run each indexing pipeline on the directory - # Username is now injected at method level + # Username and project are injected at method level for pipeline_name in [FAST_PIPELINE, SEMANTIC_PIPELINE]: result = runner.run( pipeline_name=pipeline_name, username=USERNAME, type="indexing", + project=PROJECT, data_path=data_directory, ) results[pipeline_name] = result diff --git a/examples/retrieval_pipeline_example.py b/examples/retrieval_pipeline_example.py index 3e0b287..c4ca7b7 100644 --- a/examples/retrieval_pipeline_example.py +++ b/examples/retrieval_pipeline_example.py @@ -31,6 +31,7 @@ # Configuration USERNAME = "your_username" +PROJECT = "demo_rag_app" # Must match the project from indexing_pipeline_example.py RETRIEVAL_PIPELINE_NAME = "multi_source_retrieval" # Indexing pipelines to query (must already exist in Neo4j) @@ -145,10 +146,11 @@ def create_retrieval_pipeline() -> Any: } # Build the pipeline and store it in Neo4j - # Username is now injected at method level for multi-tenant isolation + # Username and project are injected at method level for multi-tenant isolation pipeline = factory.build_pipeline_graphs_from_specs( pipeline_specs=[pipeline_spec], username=USERNAME, + project=PROJECT, configs=[pipeline_config], pipeline_types=["retrieval"], ) @@ -192,11 +194,13 @@ def run_retrieval_pipeline( enable_caching=False, ) - # Load pipelines with username injection - runner.load_pipelines(pipeline_names=[RETRIEVAL_PIPELINE_NAME], username=USERNAME) + # Load pipelines with username and project injection + runner.load_pipelines( + pipeline_names=[RETRIEVAL_PIPELINE_NAME], username=USERNAME, project=PROJECT + ) # Run the pipeline with the query - # Username is now injected at method level + # Username and project are injected at method level # Execution flow per branch: # 1. Embed the query (using the branch's embedding model) # 2. Retrieve top 5 documents from the branch's ChromaDB @@ -207,6 +211,7 @@ def run_retrieval_pipeline( pipeline_name=RETRIEVAL_PIPELINE_NAME, username=USERNAME, type="retrieval", + project=PROJECT, query=query, ground_truth_answer=ground_truth_answer, # Optional for grounded evaluation relevant_doc_ids=relevant_doc_ids or [], # Optional for document recall diff --git a/tests/test_graph_pipeline.py b/tests/test_graph_pipeline.py index 2339765..055dfd8 100644 --- a/tests/test_graph_pipeline.py +++ b/tests/test_graph_pipeline.py @@ -64,9 +64,13 @@ def test_factory_builds_pipeline_graph(self, mock_graph_store, test_config): {"type": "CHUNKER.DOCUMENT_SPLITTER"}, ] - # Build pipeline graph (stores in Neo4j) - username now injected at method level + # Build pipeline graph (stores in Neo4j) - username and project now injected at method level spec = factory.build_pipeline_graph( - pipeline_spec, "test_pipeline", username="test_user", config={} + pipeline_spec, + "test_pipeline", + username="test_user", + project="test_project", + config={}, ) assert spec is not None @@ -85,9 +89,13 @@ def test_runner_loads_pipeline_graph(self, mock_graph_store, test_config): {"type": "CHUNKER.DOCUMENT_SPLITTER"}, ] - # Username now injected at method level + # Username and project now injected at method level factory.build_pipeline_graph( - pipeline_spec, "load_test_pipeline", username="test_user", config={} + pipeline_spec, + "load_test_pipeline", + username="test_user", + project="test_project", + config={}, ) # Mock the load response @@ -122,7 +130,9 @@ def test_runner_loads_pipeline_graph(self, mock_graph_store, test_config): ) runner.load_pipelines( - pipeline_names=["load_test_pipeline"], username="test_user" + pipeline_names=["load_test_pipeline"], + username="test_user", + project="test_project", ) # Verify graph data was loaded @@ -139,9 +149,13 @@ def test_runner_builds_haystack_components(self, mock_graph_store, test_config): {"type": "CHUNKER.DOCUMENT_SPLITTER"}, ] - # Username now injected at method level + # Username and project now injected at method level factory.build_pipeline_graph( - pipeline_spec, "component_test_pipeline", username="test_user", config={} + pipeline_spec, + "component_test_pipeline", + username="test_user", + project="test_project", + config={}, ) # Mock the load response @@ -178,7 +192,9 @@ def test_runner_builds_haystack_components(self, mock_graph_store, test_config): ) runner.load_pipelines( - pipeline_names=["component_test_pipeline"], username="test_user" + pipeline_names=["component_test_pipeline"], + username="test_user", + project="test_project", ) # Verify components were built @@ -212,9 +228,13 @@ def test_pipeline_with_config(self, mock_graph_store, test_config): } } - # Username now injected at method level + # Username and project now injected at method level spec = factory.build_pipeline_graph( - pipeline_spec, "config_test_pipeline", username="test_user", config=config + pipeline_spec, + "config_test_pipeline", + username="test_user", + project="test_project", + config=config, ) assert spec.components[0].get_config()["chunk_size"] == 500 @@ -241,7 +261,9 @@ def test_invalid_pipeline_hash_handling(self, mock_graph_store, test_config): ) runner.load_pipelines( - pipeline_names=["nonexistent_pipeline"], username="test_user" + pipeline_names=["nonexistent_pipeline"], + username="test_user", + project="test_project", ) # Should load empty/missing data for non-existent pipeline @@ -250,6 +272,87 @@ def test_invalid_pipeline_hash_handling(self, mock_graph_store, test_config): assert len(runner._pipeline_graphs["nonexistent_pipeline"]) == 0 # If not in dict, that's also fine - pipeline doesn't exist + def test_project_hierarchy_in_graph(self, mock_graph_store, test_config): + """Test that project nodes are created in the graph hierarchy.""" + + factory = PipelineFactory(graph_store=mock_graph_store, config=test_config) + + pipeline_spec = [ + {"type": "CONVERTER.TEXT"}, + {"type": "CHUNKER.DOCUMENT_SPLITTER"}, + ] + + # Build pipeline with username and project + spec = factory.build_pipeline_graph( + pipeline_spec, + "project_test_pipeline", + username="alice", + project="rag_app", + config={}, + ) + + # Verify pipeline was created + assert spec is not None + assert spec.name == "project_test_pipeline" + + # Verify that add_nodes_batch was called for User, Project, and Component nodes + add_nodes_calls = mock_graph_store.add_nodes_batch.call_args_list + + # Should have calls for User, Project, and Component nodes + assert len(add_nodes_calls) >= 3 + + # Check that User node was created + user_call = [call for call in add_nodes_calls if call[0][1] == "User"] + assert len(user_call) > 0 + user_data = user_call[0][0][0][0] + assert user_data["username"] == "alice" + + # Check that Project node was created + project_call = [call for call in add_nodes_calls if call[0][1] == "Project"] + assert len(project_call) > 0 + project_data = project_call[0][0][0][0] + assert project_data["name"] == "rag_app" + assert project_data["username"] == "alice" + + # Check that Component nodes were created + component_call = [call for call in add_nodes_calls if call[0][1] == "Component"] + assert len(component_call) > 0 + + def test_project_node_creation(self): + """Test ProjectNode creation and ID generation.""" + from agentic_rag.types import ProjectNode + + project = ProjectNode( + name="my_rag_app", username="bob", description="My RAG application" + ) + + assert project.name == "my_rag_app" + assert project.username == "bob" + assert project.description == "My RAG application" + assert project.id is not None + assert project.id.startswith("proj_") + + # Test dictionary conversion + project_dict = project.to_dict() + assert project_dict["name"] == "my_rag_app" + assert project_dict["username"] == "bob" + assert project_dict["description"] == "My RAG application" + assert "id" in project_dict + + def test_storage_paths_include_project(self, mock_graph_store, test_config): + """Test that storage paths include project in hierarchy.""" + from agentic_rag.config import Config + + config = Config(agentic_root_dir="./test_data") + + # Test project path generation + path = config.get_project_path("alice", "rag_app") + assert "alice" in path + assert "rag_app" in path + assert path.endswith("alice/rag_app") or path.endswith( + "alice\\rag_app" + ) # Handle Windows paths + if __name__ == "__main__": pytest.main([__file__, "-v"]) From 22d035fb31a8b97f558e53e038a3fcbfdd86d887 Mon Sep 17 00:00:00 2001 From: Vardhan Shorewala Date: Thu, 13 Nov 2025 17:57:23 -0800 Subject: [PATCH 2/3] readme update for project support --- agentic_rag/pipeline/README.md | 10 ++++++++++ examples/indexing_pipeline_example.py | 2 +- examples/retrieval_pipeline_example.py | 2 +- 3 files changed, 12 insertions(+), 2 deletions(-) diff --git a/agentic_rag/pipeline/README.md b/agentic_rag/pipeline/README.md index 99ee549..df2571d 100644 --- a/agentic_rag/pipeline/README.md +++ b/agentic_rag/pipeline/README.md @@ -29,6 +29,16 @@ graph TB 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`) diff --git a/examples/indexing_pipeline_example.py b/examples/indexing_pipeline_example.py index 2f8378b..e22570a 100644 --- a/examples/indexing_pipeline_example.py +++ b/examples/indexing_pipeline_example.py @@ -30,7 +30,7 @@ load_dotenv() # Configuration -USERNAME = "your_username" +USERNAME = "your_username_2" PROJECT = "demo_rag_app" # Project name for organizing pipelines # Pipeline names for different indexing strategies diff --git a/examples/retrieval_pipeline_example.py b/examples/retrieval_pipeline_example.py index c4ca7b7..8f2cdcb 100644 --- a/examples/retrieval_pipeline_example.py +++ b/examples/retrieval_pipeline_example.py @@ -30,7 +30,7 @@ load_dotenv() # Configuration -USERNAME = "your_username" +USERNAME = "your_username_2" PROJECT = "demo_rag_app" # Must match the project from indexing_pipeline_example.py RETRIEVAL_PIPELINE_NAME = "multi_source_retrieval" From 5b5c22c33a211dc8c48798050a6f80caa495e72a Mon Sep 17 00:00:00 2001 From: Vardhan Shorewala Date: Thu, 13 Nov 2025 18:20:37 -0800 Subject: [PATCH 3/3] - Update READMEs with project hierarchy documentation --- README.md | 10 ++ agentic_rag/pipeline/README.md | 179 ++++++++++++--------------------- examples/README.md | 16 +-- 3 files changed, 81 insertions(+), 124 deletions(-) diff --git a/README.md b/README.md index 0113497..85c3cfc 100644 --- a/README.md +++ b/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: diff --git a/agentic_rag/pipeline/README.md b/agentic_rag/pipeline/README.md index df2571d..f3d8a29 100644 --- a/agentic_rag/pipeline/README.md +++ b/agentic_rag/pipeline/README.md @@ -9,24 +9,17 @@ 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 @@ -58,55 +51,56 @@ Example: `factory.build_pipeline_graphs_from_specs(username="alice", project="ra ### 🏃 **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 @@ -116,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 + participant H as Haystack Pipeline - 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 + 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 - U->>R: run("indexing", data) - R->>H: Execute with data - H-->>R: Results - R-->>U: Execution results -``` - -### 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 ``` @@ -188,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}/` diff --git a/examples/README.md b/examples/README.md index 2236db1..4c0b5d4 100644 --- a/examples/README.md +++ b/examples/README.md @@ -31,20 +31,22 @@ poetry run python examples/retrieval_pipeline_example.py ### `indexing_pipeline_example.py` -Two indexing strategies: +Creates two indexing strategies in the same project: - **Fast**: 300 char chunks, all-MiniLM-L6-v2 - **Semantic**: 800 char chunks, all-mpnet-base-v2 +- Uses `project="demo_rag_app"` for organization ### `retrieval_pipeline_example.py` Multi-source retrieval with evaluation: -- Queries both indexing pipelines -- Re-ranks results +- Queries both indexing pipelines from the same project +- Re-ranks results using cross-encoder - Generates answers via OpenRouter - Evaluates with BLEU, ROUGE, coherence, readability +- **Important**: Must use same `project` as indexing pipelines -## Storage +## Storage & Organization -- **Neo4j**: Pipeline graphs, metadata, lineage -- **ChromaDB**: Vector embeddings at `./data/{username}/{pipeline_name}/` -- **IPFS**: Document content +- **Neo4j**: Pipeline graphs with `User → Project → Pipelines` hierarchy +- **ChromaDB**: Vector embeddings at `./data/{username}/{project}/{pipeline_name}/` +- **IPFS**: Document content with user-based access control