Auto-initialize pipeline status in LightRAG.initialize_storages()

• Remove manual initialize_pipeline_status calls
• Auto-init in initialize_storages method
• Update error messages for clarity
• Warn on workspace conflicts

Author: danielaskdd

lightrag/api/lightrag_server.py

@@ -58,7 +58,6 @@
     get_namespace_data,
     get_default_workspace,
     # set_default_workspace,
-    initialize_pipeline_status,
     cleanup_keyed_lock,
     finalize_share_data,
 )
@@ -352,9 +351,8 @@ async def lifespan(app: FastAPI):
 
         try:
             # Initialize database connections
-            # set_default_workspace(rag.workspace)  # comment this line to test auto default workspace setting in initialize_storages
+            # Note: initialize_storages() now auto-initializes pipeline_status for rag.workspace
             await rag.initialize_storages()
-            await initialize_pipeline_status()  # with default workspace
 
             # Data migration regardless of storage implementation
             await rag.check_and_migrate_data()

lightrag/exceptions.py

@@ -68,10 +68,7 @@ def __init__(self, storage_type: str = "Storage"):
             f"{storage_type} not initialized. Please ensure proper initialization:\n"
             f"\n"
             f"  rag = LightRAG(...)\n"
-            f"  await rag.initialize_storages()  # Required\n"
-            f"  \n"
-            f"  from lightrag.kg.shared_storage import initialize_pipeline_status\n"
-            f"  await initialize_pipeline_status()  # Required for pipeline operations\n"
+            f"  await rag.initialize_storages()  # Required - auto-initializes pipeline_status\n"
             f"\n"
             f"See: https://github.com/HKUDS/LightRAG#important-initialization-requirements"
         )
@@ -82,18 +79,21 @@ class PipelineNotInitializedError(KeyError):
 
     def __init__(self, namespace: str = ""):
         msg = (
-            f"Pipeline namespace '{namespace}' not found. "
-            f"This usually means pipeline status was not initialized.\n"
+            f"Pipeline namespace '{namespace}' not found.\n"
             f"\n"
-            f"Please call 'await initialize_pipeline_status()' after initializing storages:\n"
+            f"Pipeline status should be auto-initialized by initialize_storages().\n"
+            f"If you see this error, please ensure:\n"
             f"\n"
-            f"  from lightrag.kg.shared_storage import initialize_pipeline_status\n"
-            f"  await initialize_pipeline_status()\n"
+            f"  1. You called await rag.initialize_storages()\n"
+            f"  2. For multi-workspace setups, each LightRAG instance was properly initialized\n"
             f"\n"
-            f"Full initialization sequence:\n"
-            f"  rag = LightRAG(...)\n"
-            f"  await rag.initialize_storages()\n"
-            f"  await initialize_pipeline_status()"
+            f"Standard initialization:\n"
+            f"  rag = LightRAG(workspace='your_workspace')\n"
+            f"  await rag.initialize_storages()  # Auto-initializes pipeline_status\n"
+            f"\n"
+            f"If you need manual control (advanced):\n"
+            f"  from lightrag.kg.shared_storage import initialize_pipeline_status\n"
+            f"  await initialize_pipeline_status(workspace='your_workspace')"
         )
         super().__init__(msg)
 

lightrag/lightrag.py

@@ -664,6 +664,16 @@ async def initialize_storages(self):
             default_workspace = get_default_workspace()
             if default_workspace is None:
                 set_default_workspace(self.workspace)
+            elif default_workspace != self.workspace:
+                logger.warning(
+                    f"Creating LightRAG instance with workspace='{self.workspace}' "
+                    f"but default workspace is already set to '{default_workspace}'."
+                )
+
+            # Auto-initialize pipeline_status for this workspace
+            from lightrag.kg.shared_storage import initialize_pipeline_status
+
+            await initialize_pipeline_status(workspace=self.workspace)
 
             for storage in (
                 self.full_docs,