Added a tool to start the ingestion pipeline

Author: vaaale

packages/fetchcraft-ingestion-admin/src/fetchcraft/ingestion/admin/ingestion.py

@@ -10,7 +10,7 @@
 from fetchcraft.embeddings import OpenAIEmbeddings
 from fetchcraft.index.vector_index import VectorIndex
 from fetchcraft.ingestion.base import ConnectorSource, IngestionPipeline, Record, Sink
-from fetchcraft.ingestion.pipeline.transformation import ChunkingTransformation
+from fetchcraft.ingestion.pipeline.transformation import ChunkingTransformation, DocumentSummarization
 from fetchcraft.ingestion.sqlite_backend import AsyncSQLiteQueue
 from fetchcraft.node import DocumentNode, Node
 from fetchcraft.node_parser import HierarchicalNodeParser
@@ -149,7 +149,7 @@ def filter_fn(file: LocalFile) -> bool:
                 "application/pdf": RemoteDoclingParser(docling_url=DOCLING_SERVER)
             }
         ))
-        # .add_transformation(DocumentSummarization(max_sentences=2, simulate_latency=0.2), deferred=True)
+        # .add_transformation(DocumentSummarization(max_sentences=2), deferred=True)
         # .add_transformation(ExtractKeywords())
         .add_transformation(ChunkingTransformation(chunker=chunker))
         .add_sink(DocumentStoreSink(doc_store=doc_store))

packages/fetchcraft-ingestion-admin/src/fetchcraft/ingestion/admin/server.py

@@ -583,6 +583,29 @@ async def list_queues() -> dict:
         raise RuntimeError(f"Error listing queues: {str(e)}")
 
 
+@mcp.tool()
+async def execute_ingestion() -> str:
+    """
+    Run the ingestion pipeline.
+
+    This tool will run the ingestion pipeline to refresh the knowledgebase.
+
+    Returns:
+        Dictionary confirming the retry operation
+    """
+    from multiprocessing import Process
+    
+    def _run_ingestion_wrapper():
+        """Wrapper to run the async ingestion function in a separate process."""
+        asyncio.run(run_ingestion())
+    
+    print("Starting ingestion in a separate process...")
+    p = Process(target=_run_ingestion_wrapper, name="ingestion-process")
+    p.start()
+    
+    return f"Ingestion started in process {p.pid}. The process will run independently."
+
+
 # ============================================================================
 # Main Entry Point
 # ============================================================================