MCP and FAIR Schemas

.env.example

@@ -18,12 +18,17 @@ NEO4J_PASSWORD=your_neo4j_password_here
 # NEO4J_DATABASE=neo4j
 
 # ==============================================================================
-# REQUIRED: IPFS Storage (Lighthouse)
+# REQUIRED: Akave S3-Compatible Storage
 # ==============================================================================
-# Required for distributed storage of component outputs via IPFS
-# Get API key at: https://www.lighthouse.storage/
+# Required for distributed storage of component outputs via Akave
+# Akave provides decentralized S3-compatible storage with encryption
+# Get credentials at: https://akave.ai/
 
-LIGHTHOUSE_API_KEY=your_lighthouse_api_key_here
+AKAVE_ACCESS_KEY=your_akave_access_key_here
+AKAVE_SECRET_KEY=your_akave_secret_key_here
+AKAVE_BUCKET=agentic-rag
+# Optional: Override default endpoint (defaults to o3-rc3.akave.xyz)
+# AKAVE_ENDPOINT=https://o3-rc3.akave.xyz
 
 # ==============================================================================
 # OPTIONAL: LLM API Keys (only needed if using specific components)
@@ -66,7 +71,7 @@ OPENAI_API_KEY=your_openai_api_key_here
 # -----------------------
 # Required for all pipelines:
 #   - Neo4j: NEO4J_URI, NEO4J_USERNAME, NEO4J_PASSWORD
-#   - Lighthouse IPFS: LIGHTHOUSE_API_KEY
+#   - Akave Storage: AKAVE_ACCESS_KEY, AKAVE_SECRET_KEY
 #
 # Optional (only needed if using specific components):
 #   - GENERATOR.OPENROUTER requires: OPENROUTER_API_KEY

agentic_rag/__init__.py

@@ -2,6 +2,7 @@
 
 from .components import ComponentRegistry, get_default_registry
 from .config import Config, get_config, get_global_config, set_global_config
+from .export import JSONLDExporter
 from .pipeline import PipelineFactory, PipelineRunner
 from .types import (
     DOCUMENT_STORE,
@@ -34,4 +35,5 @@
     "ComponentType",
     "DOCUMENT_STORE",
     "list_available_components",
+    "JSONLDExporter",
 ]

agentic_rag/components/gates/gated_component.py

@@ -1,17 +1,49 @@
-"""Generic wrapper that adds caching gates to any Haystack component."""
+"""Generic wrapper that adds caching gates to any Haystack component.
+
+FAIR Compliance:
+- Supports content_type from ComponentSpec registry
+- Supports run_id for provenance tracking
+- Passes FAIR fields through to OutGate for storage
+"""
 
 import asyncio
 import contextvars
 import time
 from typing import Any, Dict, List, Optional
 
+from ...types.node_types import ContentType
 from ...utils.logger import get_logger
 from ...utils.metrics import MetricsCollector
 from ..neo4j_manager import GraphStore
 from .ingate import InGate
 from .outgate import OutGate
 
 
+def get_content_type_from_registry(component_name: str) -> Optional[ContentType]:
+    """
+    Get the FAIR content_type from the component registry.
+
+    Args:
+        component_name: Name of the component (e.g., "markdown_aware_chunker")
+
+    Returns:
+        ContentType enum value or None if not found/specified
+    """
+    from ..registry import get_default_registry
+
+    registry = get_default_registry()
+    spec = registry.get_component_spec(component_name)
+
+    if spec and spec.produces_content_type:
+        try:
+            return ContentType(spec.produces_content_type)
+        except ValueError:
+            pass
+
+    # Fallback to None if not in registry or not specified
+    return None
+
+
 class GatedComponent:
     """
     Generic wrapper that adds InGate/OutGate caching to any component.
@@ -52,7 +84,9 @@ def __init__(
         graph_store: GraphStore,
         username: str,
         cache_key: Optional[str] = None,
-        retrieve_from_ipfs: bool = True,
+        retrieve_from_storage: bool = True,
+        run_id: Optional[str] = None,
+        content_type: Optional[ContentType] = None,
     ):
         """
         Initialize gated component wrapper.
@@ -64,7 +98,9 @@ def __init__(
             graph_store: Neo4j graph store
             username: User who owns this pipeline/data
             cache_key: Pipeline-agnostic cache key for cache lookups
-            retrieve_from_ipfs: If True, retrieves cached data from IPFS (default: False)
+            retrieve_from_storage: If True, retrieves cached data from Akave (default: False)
+            run_id: FAIR provenance - Run ID to link all outputs to
+            content_type: FAIR semantic type (auto-inferred from component_name if not provided)
         """
         self.component = component
         self.component_id = component_id
@@ -77,13 +113,20 @@ def __init__(
         self.logger = get_logger(f"{__name__}.{component_name}", username=username)
         self.metrics = MetricsCollector(username=username)
 
-        # Create gates with IPFS retrieval enabled (use cache_key for lookups)
+        # FAIR compliance fields
+        self.run_id = run_id
+        # Get content_type from registry if not provided
+        self.content_type = content_type or get_content_type_from_registry(
+            component_name
+        )
+
+        # Create gates with storage retrieval enabled (use cache_key for lookups)
         self.ingate = InGate(
             graph_store=graph_store,
             component_id=self.cache_key,  # Use cache_key for lookups
             component_name=component_name,
             username=username,
-            retrieve_from_ipfs=True,  # Always retrieve from IPFS for cached results
+            retrieve_from_storage=True,  # Always retrieve from Akave for cached results
         )
 
         self.outgate = OutGate(
@@ -93,6 +136,17 @@ def __init__(
             username=username,
         )
 
+    def set_run_id(self, run_id: str) -> None:
+        """
+        Set the run_id for FAIR provenance tracking.
+
+        This can be called after construction to link outputs to a specific Run.
+
+        Args:
+            run_id: The Run ID to link all outputs to
+        """
+        self.run_id = run_id
+
     def run(self, **kwargs: Any) -> Dict[str, Any]:
         """
         Run component with caching.
@@ -173,11 +227,14 @@ def run(self, **kwargs: Any) -> Dict[str, Any]:
                     single_output_items = self._extract_output_data(single_output)
 
                     # Store: input → its outputs (handles both 1→1 and 1→N)
+                    # Include FAIR compliance fields
                     if single_output_items:
                         self.outgate.store(
                             input_data=input_item,
                             output_data=single_output_items,
                             component_config=component_config,
+                            content_type=self.content_type,
+                            run_id=self.run_id,
                         )
                 except Exception as e:
                     self.logger.warning(f"Failed to cache item: {e}")
@@ -316,11 +373,14 @@ async def run_async(self, **kwargs: Any) -> Dict[str, Any]:
                     single_output_items = self._extract_output_data(single_output)
 
                     # Store: input → its outputs (handles both 1→1 and 1→N) - use async version
+                    # Include FAIR compliance fields
                     if single_output_items:
                         await self.outgate.store_async(
                             input_data=input_item,
                             output_data=single_output_items,
                             component_config=component_config,
+                            content_type=self.content_type,
+                            run_id=self.run_id,
                         )
                 except Exception as e:
                     self.logger.warning(f"Failed to cache item: {e}")

agentic_rag/components/gates/ingate.py

@@ -8,7 +8,7 @@
 import json
 from typing import Any, Dict, List, Optional
 
-from ...utils.ipfs_client import LighthouseClient
+from ...utils.akave_client import AkaveClient
 from ...utils.logger import get_logger
 from ..neo4j_manager import GraphStore
 
@@ -35,8 +35,8 @@ def __init__(
         component_id: str,
         component_name: str,
         username: Optional[str] = None,
-        ipfs_client: Optional[LighthouseClient] = None,
-        retrieve_from_ipfs: bool = False,
+        storage_client: Optional[AkaveClient] = None,
+        retrieve_from_storage: bool = False,
     ):
         """
         Initialize InGate.
@@ -46,14 +46,14 @@ def __init__(
             component_id: Unique ID of the component this gate protects
             component_name: Human-readable component name
             username: Username for per-user logging
-            ipfs_client: Optional Lighthouse IPFS client (creates one if not provided)
-            retrieve_from_ipfs: If True, retrieves actual data from IPFS (default: False, returns metadata only)
+            storage_client: Optional Akave storage client (creates one if not provided)
+            retrieve_from_storage: If True, retrieves actual data from storage (default: False, returns metadata only)
         """
         self.graph_store = graph_store
         self.component_id = component_id
         self.component_name = component_name
-        self.ipfs_client = ipfs_client or LighthouseClient()
-        self.retrieve_from_ipfs = retrieve_from_ipfs
+        self.storage_client = storage_client or AkaveClient()
+        self.retrieve_from_storage = retrieve_from_storage
         self.logger = get_logger(f"{__name__}.{component_name}", username=username)
 
     def check_cache_batch(
@@ -77,8 +77,8 @@ def check_cache_batch(
             }
 
             cached_result format:
-            - If retrieve_from_ipfs=False: List[{fingerprint, ipfs_hash, data_type}]
-            - If retrieve_from_ipfs=True: List of actual data retrieved from IPFS
+            - If retrieve_from_storage=False: List[{fingerprint, object_key, data_type}]
+            - If retrieve_from_storage=True: List of actual data retrieved from storage
 
         Example (batch):
             >>> result = ingate.check_cache_batch(
@@ -128,22 +128,24 @@ def check_cache_batch(
                 # Found cached result
                 cached_metadata = cache_map[fp]
 
-                if self.retrieve_from_ipfs:
+                if self.retrieve_from_storage:
                     try:
                         cached_data = []
                         for output_meta in cached_metadata:
-                            ipfs_hash = output_meta["ipfs_hash"]
+                            object_key = output_meta[
+                                "ipfs_hash"
+                            ]  # Field name kept for Neo4j compat
                             data_type = output_meta.get("data_type")
-                            data = self._retrieve_from_ipfs(ipfs_hash, data_type)
+                            data = self._retrieve_from_storage(object_key, data_type)
                             cached_data.append(data)
                         cached.append((item, cached_data))
                         cache_hits += 1
                     except (ConnectionError, Exception) as e:
-                        # IPFS retrieval failed - treat as cache miss
-                        self.logger.warning(f"IPFS retrieval failed for {fp}: {e}")
+                        # Storage retrieval failed - treat as cache miss
+                        self.logger.warning(f"Storage retrieval failed for {fp}: {e}")
                         uncached.append(item)
                 else:
-                    # Just return metadata (fingerprints + IPFS hashes)
+                    # Just return metadata (fingerprints + object keys)
                     cached.append((item, cached_metadata))
                     cache_hits += 1
             else:
@@ -213,24 +215,26 @@ async def check_cache_batch_async(
                 # Found cached result
                 cached_metadata = cache_map[fp]
 
-                if self.retrieve_from_ipfs:
+                if self.retrieve_from_storage:
                     try:
                         cached_data = []
                         for output_meta in cached_metadata:
-                            ipfs_hash = output_meta["ipfs_hash"]
+                            object_key = output_meta[
+                                "ipfs_hash"
+                            ]  # Field name kept for Neo4j compat
                             data_type = output_meta.get("data_type")
-                            data = await self._retrieve_from_ipfs_async(
-                                ipfs_hash, data_type
+                            data = await self._retrieve_from_storage_async(
+                                object_key, data_type
                             )
                             cached_data.append(data)
                         cached.append((item, cached_data))
                         cache_hits += 1
                     except (ConnectionError, Exception) as e:
-                        # IPFS retrieval failed - treat as cache miss
-                        self.logger.warning(f"IPFS retrieval failed for {fp}: {e}")
+                        # Storage retrieval failed - treat as cache miss
+                        self.logger.warning(f"Storage retrieval failed for {fp}: {e}")
                         uncached.append(item)
                 else:
-                    # Just return metadata (fingerprints + IPFS hashes)
+                    # Just return metadata (fingerprints + object keys)
                     cached.append((item, cached_metadata))
                     cache_hits += 1
             else:
@@ -308,22 +312,52 @@ def hash_config(self, config: Dict[str, Any]) -> str:
         hash_obj = hashlib.sha256(config_str.encode("utf-8"))
         return f"cfg_{hash_obj.hexdigest()[:16]}"
 
-    def _retrieve_from_ipfs(
-        self, ipfs_hash: str, data_type: Optional[str] = None
+    def _retrieve_from_storage(
+        self, object_key: str, data_type: Optional[str] = None
     ) -> Any:
-        """Retrieve from IPFS and reconstruct as Document."""
+        """Retrieve from Akave storage and reconstruct as Document."""
         from haystack import Document
 
-        text = self.ipfs_client.retrieve_text(ipfs_hash)
+        # Try to retrieve as JSON first (may contain embedding)
+        try:
+            data = self.storage_client.retrieve_json(object_key)
+            if isinstance(data, dict) and "content" in data:
+                # Reconstruct Document with all fields
+                return Document(
+                    content=data.get("content", ""),
+                    meta=data.get("meta", {}),
+                    id=data.get("id"),
+                    embedding=data.get("embedding"),  # May be None or list
+                )
+        except Exception:
+            pass
+
+        # Fallback: retrieve as text
+        text = self.storage_client.retrieve_text(object_key)
         return Document(content=text)
 
-    async def _retrieve_from_ipfs_async(
-        self, ipfs_hash: str, data_type: Optional[str] = None
+    async def _retrieve_from_storage_async(
+        self, object_key: str, data_type: Optional[str] = None
     ) -> Any:
-        """Async version of _retrieve_from_ipfs."""
+        """Async version of _retrieve_from_storage."""
         from haystack import Document
 
-        text = await self.ipfs_client.retrieve_text_async(ipfs_hash)
+        # Try to retrieve as JSON first (may contain embedding)
+        try:
+            data = await self.storage_client.retrieve_json_async(object_key)
+            if isinstance(data, dict) and "content" in data:
+                # Reconstruct Document with all fields
+                return Document(
+                    content=data.get("content", ""),
+                    meta=data.get("meta", {}),
+                    id=data.get("id"),
+                    embedding=data.get("embedding"),  # May be None or list
+                )
+        except Exception:
+            pass
+
+        # Fallback: retrieve as text
+        text = await self.storage_client.retrieve_text_async(object_key)
         return Document(content=text)
 
     def _serialize_for_fingerprint(self, data: Any) -> str: