Merge pull request #137 from Hawksight-AI/utils

feat: Add pipeline_id support and fix parsing table display

semantica/conflicts/source_tracker.py

@@ -549,3 +549,122 @@ class SourceTracker:
             List of traceability records
         """
         return self.get_traceability_chain(entity_id, property_name)
+
+    def track_sources_batch(
+        self,
+        source_data: List[Dict[str, Any]],
+        pipeline_id: Optional[str] = None,
+    ) -> Dict[str, int]:
+        """
+        Track sources for multiple entities, properties, or relationships in batch.
+
+        Args:
+            source_data: List of source tracking dictionaries, each containing:
+                - type: "entity", "property", or "relationship"
+                - entity_id: Entity identifier (required for all types)
+                - property_name: Property name (required for "property" type)
+                - value: Property value (required for "property" type)
+                - relationship_id: Relationship identifier (required for "relationship" type)
+                - source: SourceReference object or dict to create SourceReference
+                - metadata: Optional metadata dictionary
+            pipeline_id: Optional pipeline ID for progress tracking
+
+        Returns:
+            dict: Statistics with keys:
+                - entities_tracked: Number of entities tracked
+                - properties_tracked: Number of properties tracked
+                - relationships_tracked: Number of relationships tracked
+                - total_tracked: Total number of items tracked
+        """
+        if not source_data:
+            return {
+                "entities_tracked": 0,
+                "properties_tracked": 0,
+                "relationships_tracked": 0,
+                "total_tracked": 0,
+            }
+
+        tracking_id = self.progress_tracker.start_tracking(
+            module="conflicts",
+            submodule="SourceTracker",
+            message=f"Tracking sources for {len(source_data)} items",
+            pipeline_id=pipeline_id,
+        )
+
+        stats = {
+            "entities_tracked": 0,
+            "properties_tracked": 0,
+            "relationships_tracked": 0,
+            "total_tracked": 0,
+        }
+
+        try:
+            for i, item in enumerate(source_data):
+                item_type = item.get("type", "property")
+                source_ref = item.get("source")
+                metadata = item.get("metadata", {})
+
+                # Convert source dict to SourceReference if needed
+                if isinstance(source_ref, dict):
+                    source_ref = SourceReference(**source_ref)
+                elif not isinstance(source_ref, SourceReference):
+                    self.logger.warning(f"Invalid source reference in item {i}: {item}")
+                    continue
+
+                if item_type == "entity":
+                    entity_id = item.get("entity_id")
+                    if entity_id:
+                        self.track_entity_source(entity_id, source_ref, **metadata)
+                        stats["entities_tracked"] += 1
+                        stats["total_tracked"] += 1
+
+                elif item_type == "property":
+                    entity_id = item.get("entity_id")
+                    property_name = item.get("property_name")
+                    value = item.get("value")
+                    if entity_id and property_name is not None:
+                        self.track_property_source(
+                            entity_id, property_name, value, source_ref, **metadata
+                        )
+                        stats["properties_tracked"] += 1
+                        stats["total_tracked"] += 1
+
+                elif item_type == "relationship":
+                    relationship_id = item.get("relationship_id")
+                    if relationship_id:
+                        self.track_relationship_source(
+                            relationship_id, source_ref, **metadata
+                        )
+                        stats["relationships_tracked"] += 1
+                        stats["total_tracked"] += 1
+
+                else:
+                    self.logger.warning(
+                        f"Unknown type '{item_type}' in item {i}, skipping"
+                    )
+
+                # Update progress
+                self.progress_tracker.update_progress(
+                    tracking_id,
+                    processed=i + 1,
+                    total=len(source_data),
+                    message=f"Tracking sources {i+1}/{len(source_data)}...",
+                )
+
+            message = (
+                f"Tracked {stats['total_tracked']} items: "
+                f"{stats['entities_tracked']} entities, "
+                f"{stats['properties_tracked']} properties, "
+                f"{stats['relationships_tracked']} relationships"
+            )
+
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="completed", message=message
+            )
+            return stats
+
+        except Exception as e:
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message=str(e)
+            )
+            raise

semantica/core/orchestrator.py

@@ -203,6 +203,7 @@ class Semantica:
             sources: List of data sources (files, URLs, streams)
             **kwargs: Additional processing options:
                 - pipeline: Custom pipeline configuration
+                - pipeline_id: Optional pipeline ID for progress tracking
                 - embeddings: Whether to generate embeddings
                 - graph: Whether to build knowledge graph
                 - normalize: Whether to normalize data
@@ -220,11 +221,19 @@ class Semantica:
         # Auto-initialize if not already initialized
         self._ensure_initialized()
 
+        # Extract or generate pipeline_id
+        pipeline_id = kwargs.get("pipeline_id")
+        if not pipeline_id:
+            # Generate a unique pipeline ID
+            import uuid
+            pipeline_id = f"kb_build_{uuid.uuid4().hex[:8]}"
+
         # Start overall progress tracking
         overall_tracking_id = self.progress_tracker.start_tracking(
             module="core",
             submodule="Semantica",
             message=f"Building knowledge base from {len(sources)} sources",
+            pipeline_id=pipeline_id,
         )
 
         try:
@@ -236,6 +245,23 @@ class Semantica:
             # Create processing pipeline
             pipeline_config = kwargs.get("pipeline", {})
             pipeline = self._create_pipeline(pipeline_config)
+            
+            # Register pipeline modules for progress tracking
+            if hasattr(pipeline, 'steps') and pipeline.steps:
+                module_list = []
+                module_order = {}
+                for idx, step in enumerate(pipeline.steps):
+                    module_name = getattr(step, 'module', None) or getattr(step, 'name', None) or str(step)
+                    if module_name and module_name not in module_list:
+                        module_list.append(module_name)
+                        module_order[module_name] = idx
+                
+                if module_list:
+                    self.progress_tracker.register_pipeline_modules(
+                        pipeline_id=pipeline_id,
+                        module_list=module_list,
+                        module_order=module_order
+                    )
 
             # Process sources
             results = []
@@ -251,6 +277,7 @@ class Semantica:
                         module="core",
                         submodule="build_knowledge_base",
                         message=f"Processing {Path(file_str).name if file_str else 'source'}",
+                        pipeline_id=pipeline_id,
                     )
                     try:
                         result = self.run_pipeline(pipeline, source)
@@ -308,6 +335,9 @@ class Semantica:
                 message=f"Processed {len(results)} sources",
             )
 
+            # Clear pipeline context when complete
+            self.progress_tracker.clear_pipeline_context(pipeline_id)
+
             # Show summary
             self.progress_tracker.show_summary()
 
@@ -319,6 +349,7 @@ class Semantica:
                 "metadata": {
                     "sources": validated_sources,
                     "pipeline": pipeline_config,
+                    "pipeline_id": pipeline_id,
                 },
             }
 
@@ -326,6 +357,8 @@ class Semantica:
             self.progress_tracker.stop_tracking(
                 overall_tracking_id, status="failed", message=str(e)
             )
+            # Clear pipeline context on failure
+            self.progress_tracker.clear_pipeline_context(pipeline_id)
             self.logger.error(f"Failed to build knowledge base: {e}")
             raise ProcessingError(f"Failed to build knowledge base: {e}")
 

semantica/embeddings/embedding_generator.py

@@ -258,13 +258,14 @@ class EmbeddingGenerator:
         return max(0.0, min(1.0, similarity))
 
     def process_batch(
-        self, data_items: List[Union[str, Path]], **options
+        self, data_items: List[Union[str, Path]], pipeline_id: Optional[str] = None, **options
     ) -> Dict[str, Any]:
         """
         Process multiple data items for embedding generation.
 
         Args:
             data_items: List of data items
+            pipeline_id: Optional pipeline ID for progress tracking
             **options: Processing options
 
         Returns:
@@ -275,6 +276,7 @@ class EmbeddingGenerator:
             module="embeddings",
             submodule="EmbeddingGenerator",
             message=f"Batch of {len(data_items)} items",
+            pipeline_id=pipeline_id,
         )
 
         try:

semantica/kg/graph_builder.py

@@ -237,6 +237,7 @@ class GraphBuilder:
         self,
         sources: Union[List[Any], Any],
         second_arg: Optional[Any] = None,
+        pipeline_id: Optional[str] = None,
         **options,
     ) -> Dict[str, Any]:
         """
@@ -245,6 +246,7 @@ class GraphBuilder:
         Args:
             sources: Entities or sources list
             second_arg: Optional relationships list or entity_resolver (for backward compatibility)
+            pipeline_id: Optional pipeline ID for progress tracking
             **options: Additional build options
                 - extract: Whether to extract entities from text (default: True)
                 - extract_relations: Whether to extract relations from text (default: False)
@@ -307,6 +309,7 @@ class GraphBuilder:
             module="kg",
             submodule="GraphBuilder",
             message=f"Knowledge graph from {len(sources)} source(s)",
+            pipeline_id=pipeline_id,
         )
 
         try:
@@ -340,6 +343,7 @@ class GraphBuilder:
                         module="kg",
                         submodule="GraphBuilder",
                         message=f"Processing {len(entities_list)} entities",
+                        pipeline_id=pipeline_id,
                     )
                     
                     # Check if entities are already in dictionary format
@@ -409,6 +413,7 @@ class GraphBuilder:
                         module="kg",
                         submodule="GraphBuilder",
                         message=f"Processing {len(relationships_list)} relationships",
+                        pipeline_id=pipeline_id,
                     )
                     
                     # Check if relationships are already in dictionary format

semantica/kg/provenance_tracker.py

@@ -217,3 +217,131 @@ class ProvenanceTracker:
                   or None if entity is not tracked
         """
         return self.provenance_data.get(entity_id)
+
+    def track_entities_batch(
+        self,
+        entities: List[Dict[str, Any]],
+        source: str,
+        pipeline_id: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> int:
+        """
+        Track provenance for multiple entities in batch.
+
+        Args:
+            entities: List of entity dictionaries, each containing at least 'id' key
+            source: Source identifier
+            pipeline_id: Optional pipeline ID for progress tracking
+            metadata: Optional metadata dictionary to apply to all entities
+
+        Returns:
+            int: Number of entities tracked
+        """
+        if not entities:
+            return 0
+
+        tracking_id = self.progress_tracker.start_tracking(
+            module="kg",
+            submodule="ProvenanceTracker",
+            message=f"Tracking provenance for {len(entities)} entities",
+            pipeline_id=pipeline_id,
+        )
+
+        try:
+            tracked_count = 0
+            for i, entity in enumerate(entities):
+                entity_id = entity.get("id") or entity.get("entity_id")
+                if not entity_id:
+                    self.logger.warning(f"Skipping entity without ID: {entity}")
+                    continue
+
+                # Merge entity-specific metadata with batch metadata
+                entity_metadata = {**(metadata or {}), **(entity.get("metadata", {}))}
+                self.track_entity(entity_id, source, entity_metadata)
+
+                tracked_count += 1
+
+                # Update progress
+                self.progress_tracker.update_progress(
+                    tracking_id,
+                    processed=i + 1,
+                    total=len(entities),
+                    message=f"Tracking entity {i+1}/{len(entities)}...",
+                )
+
+            self.progress_tracker.stop_tracking(
+                tracking_id,
+                status="completed",
+                message=f"Tracked {tracked_count} entities",
+            )
+            return tracked_count
+
+        except Exception as e:
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message=str(e)
+            )
+            raise
+
+    def track_relationships_batch(
+        self,
+        relationships: List[Dict[str, Any]],
+        source: str,
+        pipeline_id: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> int:
+        """
+        Track provenance for multiple relationships in batch.
+
+        Args:
+            relationships: List of relationship dictionaries, each containing at least 'id' key
+            source: Source identifier
+            pipeline_id: Optional pipeline ID for progress tracking
+            metadata: Optional metadata dictionary to apply to all relationships
+
+        Returns:
+            int: Number of relationships tracked
+        """
+        if not relationships:
+            return 0
+
+        tracking_id = self.progress_tracker.start_tracking(
+            module="kg",
+            submodule="ProvenanceTracker",
+            message=f"Tracking provenance for {len(relationships)} relationships",
+            pipeline_id=pipeline_id,
+        )
+
+        try:
+            tracked_count = 0
+            for i, relationship in enumerate(relationships):
+                relationship_id = relationship.get("id") or relationship.get("relationship_id")
+                if not relationship_id:
+                    self.logger.warning(f"Skipping relationship without ID: {relationship}")
+                    continue
+
+                # Merge relationship-specific metadata with batch metadata
+                rel_metadata = {**(metadata or {}), **(relationship.get("metadata", {}))}
+                self.track_relationship(relationship_id, source, rel_metadata)
+
+                tracked_count += 1
+
+                # Update progress
+                self.progress_tracker.update_progress(
+                    tracking_id,
+                    processed=i + 1,
+                    total=len(relationships),
+                    message=f"Tracking relationship {i+1}/{len(relationships)}...",
+                )
+
+            self.progress_tracker.stop_tracking(
+                tracking_id,
+                status="completed",
+                message=f"Tracked {tracked_count} relationships",
+            )
+            return tracked_count
+
+        except Exception as e:
+            self.progress_tracker.stop_tracking(
+                tracking_id, status="failed", message=str(e)
+            )
+            raise

semantica/parse/docling_parser.py

@@ -119,12 +119,16 @@ class DoclingParser:
         """
         file_path = Path(file_path)
 
+        # Get pipeline_id from options if provided
+        pipeline_id = options.get("pipeline_id", None)
+        
         # Track document parsing
         tracking_id = self.progress_tracker.start_tracking(
             file=str(file_path),
             module="parse",
             submodule="DoclingParser",
             message=f"Docling: {file_path.name}",
+            pipeline_id=pipeline_id,
         )
 
         try:
@@ -159,6 +163,7 @@ class DoclingParser:
 
             # Stage 2: Document conversion (10-70%) - This is the longest step
             # Note: This is a blocking operation, but we'll update progress after it completes
+            # Emphasize Docling as core dependency in message
             self.progress_tracker.update_progress(
                 tracking_id,
                 processed=2,
@@ -247,10 +252,38 @@ class DoclingParser:
                 )
                 images = self._extract_images(result)
 
+            # Prepare extraction counts and metadata
+            extraction_counts = {
+                "tables": len(tables),
+                "images": len(images),
+                "pages": len(pages) or metadata.page_count or 0,
+            }
+            
+            # Build completion message with Docling emphasis
+            count_parts = []
+            if extraction_counts["tables"] > 0:
+                count_parts.append(f"{extraction_counts['tables']} tables")
+            if extraction_counts["images"] > 0:
+                count_parts.append(f"{extraction_counts['images']} images")
+            if extraction_counts["pages"] > 0:
+                count_parts.append(f"{extraction_counts['pages']} pages")
+            
+            if count_parts:
+                completion_message = f"Parsed document (Docling): {', '.join(count_parts)} extracted"
+            else:
+                completion_message = f"Parsed document (Docling): 0 tables, 0 images, {extraction_counts['pages']} pages extracted"
+            
+            # Store metadata with extraction counts and core dependency
+            metadata_dict = {
+                "extraction_counts": extraction_counts,
+                "core_dependency": "docling",
+            }
+            
             self.progress_tracker.stop_tracking(
                 tracking_id,
                 status="completed",
-                message=f"Parsed document: {len(tables)} tables extracted",
+                message=completion_message,
+                metadata=metadata_dict,
             )
 
             return {

semantica/parse/docx_parser.py

@@ -84,12 +84,13 @@ class DOCXParser:
         self.config = config
         self.progress_tracker = get_progress_tracker()
 
-    def parse(self, file_path: Union[str, Path], **options) -> Dict[str, Any]:
+    def parse(self, file_path: Union[str, Path], pipeline_id: Optional[str] = None, **options) -> Dict[str, Any]:
         """
         Parse DOCX document.
 
         Args:
             file_path: Path to DOCX file
+            pipeline_id: Optional pipeline ID for progress tracking
             **options: Parsing options:
                 - extract_formatting: Whether to extract formatting (default: False)
                 - extract_tables: Whether to extract tables (default: True)
@@ -106,6 +107,7 @@ class DOCXParser:
             module="parse",
             submodule="DOCXParser",
             message=f"DOCX: {file_path.name}",
+            pipeline_id=pipeline_id,
         )
 
         try:

semantica/parse/pdf_parser.py

@@ -84,12 +84,13 @@ class PDFParser:
         if not self.progress_tracker.enabled:
             self.progress_tracker.enabled = True
 
-    def parse(self, file_path: Union[str, Path], **options) -> Dict[str, Any]:
+    def parse(self, file_path: Union[str, Path], pipeline_id: Optional[str] = None, **options) -> Dict[str, Any]:
         """
         Parse PDF document.
 
         Args:
             file_path: Path to PDF file
+            pipeline_id: Optional pipeline ID for progress tracking
             **options: Parsing options:
                 - extract_text: Whether to extract text (default: True)
                 - extract_tables: Whether to extract tables (default: True)
@@ -107,6 +108,7 @@ class PDFParser:
             module="parse",
             submodule="PDFParser",
             message=f"PDF: {file_path.name}",
+            pipeline_id=pipeline_id,
         )
 
         try:

semantica/pipeline/execution_engine.py

@@ -132,11 +132,37 @@ class ExecutionEngine:
             module="pipeline",
             submodule="ExecutionEngine",
             message=f"Executing pipeline: {pipeline_id}",
+            pipeline_id=pipeline_id,
         )
 
         try:
             self.logger.info(f"Executing pipeline: {pipeline_id}")
 
+            # Register pipeline modules for progress tracking
+            module_list = []
+            module_order = {}
+            if hasattr(pipeline, 'steps') and pipeline.steps:
+                for idx, step in enumerate(pipeline.steps):
+                    # Extract module name from step
+                    module_name = getattr(step, 'module', None) or getattr(step, 'name', None) or str(step)
+                    if module_name and module_name not in module_list:
+                        module_list.append(module_name)
+                        module_order[module_name] = idx
+            
+            # If no steps found, try to infer from pipeline structure
+            if not module_list:
+                # Common pipeline modules
+                module_list = ["ingest", "parse", "normalize", "semantic_extract", "kg", "embeddings"]
+                module_order = {module: idx for idx, module in enumerate(module_list)}
+            
+            # Register pipeline modules
+            if module_list:
+                self.progress_tracker.register_pipeline_modules(
+                    pipeline_id=pipeline_id,
+                    module_list=module_list,
+                    module_order=module_order
+                )
+
             # Set status
             with self.pipeline_lock:
                 self.pipeline_status[pipeline_id] = PipelineStatus.RUNNING
@@ -174,6 +200,9 @@ class ExecutionEngine:
                     status="completed" if metrics["steps_failed"] == 0 else "failed",
                     message=f"Executed {metrics['steps_executed']} steps in {execution_time:.2f}s",
                 )
+                
+                # Clear pipeline context when pipeline completes
+                self.progress_tracker.clear_pipeline_context(pipeline_id)
 
                 return ExecutionResult(
                     success=metrics["steps_failed"] == 0,
@@ -193,6 +222,8 @@ class ExecutionEngine:
             self.progress_tracker.stop_tracking(
                 pipeline_tracking_id, status="failed", message=str(e)
             )
+            # Clear pipeline context on failure
+            self.progress_tracker.clear_pipeline_context(pipeline_id)
             self.logger.error(f"Pipeline execution failed: {e}")
             with self.pipeline_lock:
                 self.pipeline_status[pipeline_id] = PipelineStatus.FAILED

semantica/pipeline/parallelism_manager.py

@@ -104,13 +104,14 @@ class ParallelismManager:
         self.lock = threading.Lock()
 
     def execute_parallel(
-        self, tasks: List[Task], **options
+        self, tasks: List[Task], pipeline_id: Optional[str] = None, **options
     ) -> List[ParallelExecutionResult]:
         """
         Execute tasks in parallel.
 
         Args:
             tasks: List of tasks to execute
+            pipeline_id: Optional pipeline ID for progress tracking
             **options: Additional options
 
         Returns:
@@ -120,6 +121,7 @@ class ParallelismManager:
             module="pipeline",
             submodule="ParallelismManager",
             message=f"Executing {len(tasks)} tasks in parallel",
+            pipeline_id=pipeline_id,
         )
 
         try:

semantica/semantic_extract/ner_extractor.py

@@ -141,13 +141,14 @@ class NERExtractor:
                     f"spaCy model {self.model_name} not found. ML method will fallback."
                 )
 
-    def extract(self, text: Union[str, List[Dict[str, Any]], List[str]], **kwargs) -> Union[List[Entity], List[List[Entity]]]:
+    def extract(self, text: Union[str, List[Dict[str, Any]], List[str]], pipeline_id: Optional[str] = None, **kwargs) -> Union[List[Entity], List[List[Entity]]]:
         """
         Alias for extract_entities.
         Handles both single string and list of documents.
         
         Args:
             text: Input text or list of documents
+            pipeline_id: Optional pipeline ID for progress tracking
             **kwargs: Extraction options
             
         Returns:
@@ -159,6 +160,7 @@ class NERExtractor:
                 module="semantic_extract",
                 submodule="NERExtractor",
                 message=f"Batch extracting entities from {len(text)} documents",
+                pipeline_id=pipeline_id,
             )
             
             try:

semantica/semantic_extract/relation_extractor.py

@@ -168,6 +168,7 @@ class RelationExtractor:
         self,
         text: Union[str, List[Dict[str, Any]], List[str]],
         entities: Union[List[Entity], List[List[Entity]]],
+        pipeline_id: Optional[str] = None,
         **kwargs
     ) -> Union[List[Relation], List[List[Relation]]]:
         """
@@ -177,6 +178,7 @@ class RelationExtractor:
         Args:
             text: Input text or list of documents
             entities: List of entities or list of list of entities
+            pipeline_id: Optional pipeline ID for progress tracking
             **kwargs: Extraction options
             
         Returns:
@@ -188,6 +190,7 @@ class RelationExtractor:
                 module="semantic_extract",
                 submodule="RelationExtractor",
                 message=f"Batch extracting relations from {len(text)} documents",
+                pipeline_id=pipeline_id,
             )
             
             try:

semantica/split/provenance_tracker.py

@@ -127,16 +127,22 @@ class ProvenanceTracker:
 
         return provenance_id
 
-    def track(self, **kwargs) -> Union[str, List[str]]:
+    def track(self, pipeline_id: Optional[str] = None, **kwargs) -> Union[str, List[str]]:
         """
         Track provenance (alias for track_chunk/track_chunks).
         Delegates based on input arguments.
+
+        Args:
+            pipeline_id: Optional pipeline ID for progress tracking
+            **kwargs: Arguments passed to track_chunk or track_chunks
         """
         if "chunks" in kwargs:
+            kwargs["pipeline_id"] = pipeline_id
             return self.track_chunks(**kwargs)
         if "chunk" in kwargs and isinstance(kwargs["chunk"], list):
              # Handle list passed to chunk arg
              kwargs["chunks"] = kwargs.pop("chunk")
+             kwargs["pipeline_id"] = pipeline_id
              return self.track_chunks(**kwargs)
         return self.track_chunk(**kwargs)
 
@@ -145,6 +151,7 @@ class ProvenanceTracker:
         chunks: List[Chunk],
         source_document: str,
         source_path: Optional[str] = None,
+        pipeline_id: Optional[str] = None,
         **metadata,
     ) -> List[str]:
         """
@@ -154,6 +161,7 @@ class ProvenanceTracker:
             chunks: List of chunks to track
             source_document: Source document identifier
             source_path: Path to source document
+            pipeline_id: Optional pipeline ID for progress tracking
             **metadata: Additional metadata
 
         Returns:
@@ -163,6 +171,7 @@ class ProvenanceTracker:
             module="split",
             submodule="ProvenanceTracker",
             message=f"Tracking provenance for {len(chunks)} chunks",
+            pipeline_id=pipeline_id,
         )
 
         try:
@@ -170,8 +179,11 @@ class ProvenanceTracker:
             parent_chunk_id = None
 
             for i, chunk in enumerate(chunks):
-                self.progress_tracker.update_tracking(
-                    tracking_id, message=f"Tracking chunk {i+1}/{len(chunks)}..."
+                self.progress_tracker.update_progress(
+                    tracking_id,
+                    processed=i + 1,
+                    total=len(chunks),
+                    message=f"Tracking chunk {i+1}/{len(chunks)}...",
                 )
                 provenance_id = self.track_chunk(
                     chunk, source_document, source_path, parent_chunk_id, **metadata

semantica/utils/progress_tracker.py

@@ -77,6 +77,8 @@ class ProgressItem:
     total_items: Optional[int] = None  # Total items to process
     processed_items: Optional[int] = None  # Items processed so far
     estimated_remaining: Optional[float] = None  # Estimated remaining time in seconds
+    pipeline_id: Optional[str] = None  # Pipeline ID this item belongs to
+    pipeline_order: Optional[int] = None  # Order of this module in the pipeline
 
 
 class ProgressDisplay(ABC):
@@ -232,78 +234,130 @@ class ConsoleProgressDisplay(ProgressDisplay):
             return
 
         with self.lock:
-            # Create unique key for this item
-            key = f"{item.module}:{item.submodule}"
-            if item.file:
-                key = f"{item.file}:{key}"
-
-            # Build progress line
-            parts = []
-
-            # Semantica branding with action
-            if self.use_emoji:
-                parts.append("🧠")
-
-            # Create action message based on module (now includes custom message)
-            action_msg = self._get_action_message(item.module, item.message)
-            parts.append(action_msg)
-
-            # Progress bar and percentage
-            if item.progress_percentage is not None:
-                pct = item.progress_percentage
-                bar_width = 15
-                filled = int(bar_width * pct / 100)
-                bar = "█" * filled + "░" * (bar_width - filled)
-                parts.append(f"|{bar}| {pct:.1f}%")
-
-            # Count information [done/total]
-            if item.processed_items is not None and item.total_items is not None:
-                parts.append(f"[{item.processed_items}/{item.total_items}]")
-
-            # Status and Module emojis
-            if self.use_emoji:
-                status_emoji = self._get_status_emoji(item.status)
-                module_emoji = self._get_emoji_for_module(item.module or "")
-                parts.append(f"{status_emoji}{module_emoji}")
-
-            # ETA and Rate
-            metrics = []
-            if item.estimated_remaining is not None and item.estimated_remaining > 0:
-                if item.estimated_remaining < 60:
-                    metrics.append(f"ETA: {item.estimated_remaining:.1f}s")
-                elif item.estimated_remaining < 3600:
-                    metrics.append(f"ETA: {item.estimated_remaining/60:.1f}m")
-                else:
-                    metrics.append(f"ETA: {item.estimated_remaining/3600:.1f}h")
+            # If item is part of a pipeline, get all pipeline items for display
+            pipeline_items = []
+            if item.pipeline_id:
+                # Get tracker instance to access pipeline items
+                # Use the singleton instance directly
+                tracker = ProgressTracker.get_instance()
+                pipeline_items = tracker.get_pipeline_items(item.pipeline_id)
             
-            if item.start_time and item.processed_items is not None and item.processed_items > 0:
-                elapsed = time.time() - item.start_time
-                if elapsed > 0:
-                    rate = item.processed_items / elapsed
-                    metrics.append(f"{rate:.1f}/s")
-            
-            if metrics:
-                parts.append(f"({' | '.join(metrics)})")
-            
-            # Elapsed time if no progress info
-            if item.start_time and item.progress_percentage is None:
-                elapsed = time.time() - item.start_time
-                parts.append(f"({elapsed:.1f}s)")
-
-            line = " ".join(parts)
-            self.current_lines[key] = line
-
-            # Print all current lines
-            # For multiple lines, we combine them to avoid console scrolling issues
-            sys.stdout.write("\r" + " " * 120 + "\r")
-            if len(self.current_lines) == 1:
-                sys.stdout.write(line)
+            # If we have pipeline items, show all of them
+            if pipeline_items:
+                # Clear and show all pipeline items
+                sys.stdout.write("\r" + " " * 150 + "\r")
+                
+                # Show header if first time
+                if not hasattr(self, '_pipeline_header_shown'):
+                    if self.use_emoji:
+                        sys.stdout.write("🧠 Semantica - 📊 Current Progress\n")
+                    else:
+                        sys.stdout.write("Semantica - Current Progress\n")
+                    sys.stdout.write("=" * 150 + "\n")
+                    self._pipeline_header_shown = True
+                
+                # Display all pipeline items
+                for pipeline_item in pipeline_items:
+                    self._display_item_line(pipeline_item)
+                    sys.stdout.write("\n")
+                
+                sys.stdout.flush()
             else:
-                # Combine multiple active lines onto one or two lines to be safe
-                lines_list = list(self.current_lines.values())
-                sys.stdout.write(" | ".join(lines_list[-2:]))
-            
-            sys.stdout.flush()
+                # Original single-item display
+                self._display_item_line(item)
+                sys.stdout.flush()
+    
+    def _display_item_line(self, item: ProgressItem) -> None:
+        """Display a single progress item line."""
+        # Create unique key for this item
+        key = f"{item.module}:{item.submodule}"
+        if item.file:
+            key = f"{item.file}:{key}"
+
+        # Build progress line
+        parts = []
+
+        # Status emoji
+        if self.use_emoji:
+            status_emoji = self._get_status_emoji(item.status)
+            parts.append(status_emoji)
+
+        # Create action message based on module
+        action_msg = self._get_action_message(item.module, item.message)
+        parts.append(action_msg)
+
+        # Module and Submodule
+        if self.use_emoji:
+            module_emoji = self._get_emoji_for_module(item.module or "")
+            parts.append(f"{module_emoji} {item.module or 'N/A'}")
+        else:
+            parts.append(f"{item.module or 'N/A'}")
+        parts.append(f"{item.submodule or 'N/A'}")
+
+        # Progress bar and percentage
+        if item.progress_percentage is not None:
+            pct = item.progress_percentage
+            bar_width = 15
+            filled = int(bar_width * pct / 100)
+            bar = "█" * filled + "░" * (bar_width - filled)
+            parts.append(f"|{bar}| {pct:.1f}%")
+        else:
+            parts.append("|" + "░" * 15 + "| 0.0%")
+
+        # ETA
+        if item.estimated_remaining is not None and item.estimated_remaining > 0:
+            if item.estimated_remaining < 60:
+                parts.append(f"ETA: {item.estimated_remaining:.1f}s")
+            elif item.estimated_remaining < 3600:
+                parts.append(f"ETA: {item.estimated_remaining/60:.1f}m")
+            else:
+                parts.append(f"ETA: {item.estimated_remaining/3600:.1f}h")
+        else:
+            parts.append("ETA: -")
+
+        # Rate
+        if item.start_time and item.processed_items is not None and item.processed_items > 0:
+            elapsed = time.time() - item.start_time
+            if elapsed > 0:
+                rate = item.processed_items / elapsed
+                parts.append(f"Rate: {rate:.1f}/s")
+        else:
+            parts.append("Rate: -")
+
+        # Time
+        if item.start_time:
+            if item.end_time:
+                elapsed = item.end_time - item.start_time
+                parts.append(f"Time: {elapsed:.2f}s")
+            else:
+                elapsed = time.time() - item.start_time
+                parts.append(f"Time: {elapsed:.2f}s")
+        else:
+            parts.append("Time: -")
+
+        # Extraction counts (if available)
+        if item.metadata.get('extraction_counts'):
+            counts = item.metadata['extraction_counts']
+            count_parts = []
+            if 'tables' in counts:
+                count_parts.append(f"{counts['tables']} tables")
+            if 'images' in counts:
+                count_parts.append(f"{counts['images']} images")
+            if 'pages' in counts:
+                count_parts.append(f"{counts['pages']} pages")
+            if count_parts:
+                extracted = ", ".join(count_parts)
+                # Add Docling indicator if core dependency is docling
+                if item.metadata.get('core_dependency') == 'docling':
+                    parts.append(f"Extracted (Docling): {extracted}")
+                else:
+                    parts.append(f"Extracted: {extracted}")
+        else:
+            parts.append("Extracted: -")
+
+        line = " ".join(parts)
+        self.current_lines[key] = line
+        sys.stdout.write(line)
 
     def show_summary(self, items: List[ProgressItem]) -> None:
         """Show final summary."""
@@ -482,12 +536,52 @@ class JupyterProgressDisplay(ProgressDisplay):
         # Current status
         html_parts.append("<h4>🧠 Semantica - 📊 Current Progress</h4>")
         html_parts.append("<table style='width: 100%; border-collapse: collapse;'>")
+        
+        # Check if any item is part of a pipeline
+        pipeline_id = None
+        pipeline_items = []
+        for item in items:
+            if item.pipeline_id:
+                pipeline_id = item.pipeline_id
+                break
+        
+        # If pipeline context exists, get all pipeline items
+        if pipeline_id:
+            tracker = ProgressTracker.get_instance()
+            pipeline_items = tracker.get_pipeline_items(pipeline_id)
+            # Also include expected modules that haven't started yet
+            if pipeline_id in tracker.pipeline_contexts:
+                expected_modules = tracker.pipeline_contexts[pipeline_id]
+                module_order = tracker.pipeline_module_order.get(pipeline_id, {})
+                # Create pending items for modules not yet started
+                existing_modules = {item.module for item in pipeline_items if item.module}
+                for module in expected_modules:
+                    if module not in existing_modules:
+                        pending_item = ProgressItem(
+                            module=module,
+                            submodule="Pending",
+                            status="pending",
+                            progress_percentage=0.0,
+                            pipeline_id=pipeline_id,
+                            pipeline_order=module_order.get(module, 999)
+                        )
+                        pipeline_items.append(pending_item)
+                # Sort by pipeline order
+                pipeline_items.sort(key=lambda x: (x.pipeline_order if x.pipeline_order is not None else 999, x.module or ""))
+        
+        # Use pipeline items if available, otherwise use regular items
+        display_items = pipeline_items if pipeline_items else items[-10:]
+        
+        # Determine column header based on whether we have Docling items
+        has_docling = any(item.metadata.get('core_dependency') == 'docling' for item in display_items)
+        extracted_header = "Extracted (Docling)" if has_docling else "Extracted"
+        
         html_parts.append(
-            "<tr><th>Status</th><th>Action</th><th>Module</th><th>Submodule</th><th>Progress</th><th>ETA</th><th>Rate</th><th>Time</th></tr>"
+            f"<tr><th>Status</th><th>Action</th><th>Module</th><th>Submodule</th><th>Progress</th><th>ETA</th><th>Rate</th><th>Time</th><th>{extracted_header}</th></tr>"
         )
 
-        # Show last 10 items
-        for item in items[-10:]:
+        # Show pipeline items or last 10 items
+        for item in display_items:
             status_emoji = self._get_status_emoji(item.status)
             module_emoji = self._get_emoji_for_module(item.module or "")
             action_msg = self._get_action_message(item.module, item.message)
@@ -523,8 +617,32 @@ class JupyterProgressDisplay(ProgressDisplay):
                     elapsed = f"{(item.end_time - item.start_time):.2f}s"
                 else:
                     elapsed = f"{(time.time() - item.start_time):.2f}s"
+            else:
+                elapsed = "-"
 
             file_name = Path(item.file).name if item.file else "-"
+            
+            # Extraction counts
+            extracted_str = "-"
+            if item.metadata.get('extraction_counts'):
+                counts = item.metadata['extraction_counts']
+                count_parts = []
+                if 'tables' in counts:
+                    count_parts.append(f"{counts['tables']} tables")
+                if 'images' in counts:
+                    count_parts.append(f"{counts['images']} images")
+                if 'pages' in counts:
+                    count_parts.append(f"{counts['pages']} pages")
+                if count_parts:
+                    extracted_str = ", ".join(count_parts)
+            elif item.status == "pending":
+                extracted_str = "-"
+            elif item.status == "running" and item.module == "parse":
+                # Show progress message for running parse operations
+                if "Docling" in item.message or item.metadata.get('core_dependency') == 'docling':
+                    extracted_str = "Converting with Docling..."
+                else:
+                    extracted_str = "-"
 
             html_parts.append(
                 f"<tr>"
@@ -536,6 +654,7 @@ class JupyterProgressDisplay(ProgressDisplay):
                 f"<td>{eta_str}</td>"
                 f"<td>{rate_str}</td>"
                 f"<td>{elapsed}</td>"
+                f"<td>{extracted_str}</td>"
                 f"</tr>"
             )
 
@@ -562,9 +681,35 @@ class JupyterProgressDisplay(ProgressDisplay):
         else:
             self.items.append(item)
 
+        # If item is part of a pipeline, get all pipeline items for display
+        display_items = self.items
+        if item.pipeline_id:
+            tracker = ProgressTracker.get_instance()
+            pipeline_items = tracker.get_pipeline_items(item.pipeline_id)
+            # Also include expected modules that haven't started yet
+            if item.pipeline_id in tracker.pipeline_contexts:
+                expected_modules = tracker.pipeline_contexts[item.pipeline_id]
+                module_order = tracker.pipeline_module_order.get(item.pipeline_id, {})
+                # Create pending items for modules not yet started
+                existing_modules = {item.module for item in pipeline_items if item.module}
+                for module in expected_modules:
+                    if module not in existing_modules:
+                        pending_item = ProgressItem(
+                            module=module,
+                            submodule="Pending",
+                            status="pending",
+                            progress_percentage=0.0,
+                            pipeline_id=item.pipeline_id,
+                            pipeline_order=module_order.get(module, 999)
+                        )
+                        pipeline_items.append(pending_item)
+                # Sort by pipeline order
+                pipeline_items.sort(key=lambda x: (x.pipeline_order if x.pipeline_order is not None else 999, x.module or ""))
+                display_items = pipeline_items
+
         # Update display - always update immediately in Jupyter/Colab
         if IPYTHON_AVAILABLE:
-            html = self._build_html(self.items)
+            html = self._build_html(display_items)
             try:
                 # Check if we're in Google Colab (Colab sometimes needs fresh displays)
                 is_colab = False
@@ -901,6 +1046,11 @@ class ProgressTracker:
         self.items: List[ProgressItem] = []
         self.active_items: Dict[str, ProgressItem] = {}
         self.lock = threading.Lock()
+        
+        # Pipeline context tracking
+        self.pipeline_contexts: Dict[str, List[str]] = {}  # pipeline_id -> list of module names
+        self.pipeline_items: Dict[str, Dict[str, ProgressItem]] = {}  # pipeline_id -> {tracking_id: item}
+        self.pipeline_module_order: Dict[str, Dict[str, int]] = {}  # pipeline_id -> {module: order}
 
     def _detect_jupyter(self) -> bool:
         """Detect if running in Jupyter notebook or Google Colab."""
@@ -962,12 +1112,86 @@ class ProgressTracker:
             cls._instance.enabled = True
         return cls._instance
 
+    def register_pipeline_modules(
+        self, pipeline_id: str, module_list: List[str], module_order: Optional[Dict[str, int]] = None
+    ) -> None:
+        """
+        Register modules that belong to a pipeline.
+
+        Args:
+            pipeline_id: Unique pipeline identifier
+            module_list: List of module names in the pipeline
+            module_order: Optional dict mapping module names to their order in pipeline
+        """
+        if not self.enabled:
+            return
+
+        with self.lock:
+            self.pipeline_contexts[pipeline_id] = module_list
+            if module_order:
+                self.pipeline_module_order[pipeline_id] = module_order
+            else:
+                # Auto-generate order if not provided
+                self.pipeline_module_order[pipeline_id] = {
+                    module: idx for idx, module in enumerate(module_list)
+                }
+            # Initialize pipeline items dict
+            if pipeline_id not in self.pipeline_items:
+                self.pipeline_items[pipeline_id] = {}
+
+    def get_pipeline_items(self, pipeline_id: str) -> List[ProgressItem]:
+        """
+        Get all items for a pipeline (completed + active).
+
+        Args:
+            pipeline_id: Pipeline identifier
+
+        Returns:
+            List of ProgressItem objects for the pipeline, ordered by pipeline_order
+        """
+        if not self.enabled or pipeline_id not in self.pipeline_contexts:
+            return []
+
+        with self.lock:
+            items = []
+            # Get items from pipeline_items (completed)
+            if pipeline_id in self.pipeline_items:
+                items.extend(self.pipeline_items[pipeline_id].values())
+            
+            # Get active items that belong to this pipeline
+            for tracking_id, item in self.active_items.items():
+                if item.pipeline_id == pipeline_id:
+                    items.append(item)
+            
+            # Sort by pipeline_order
+            items.sort(key=lambda x: (x.pipeline_order if x.pipeline_order is not None else 999, x.module or ""))
+            return items
+
+    def clear_pipeline_context(self, pipeline_id: str) -> None:
+        """
+        Clear pipeline context when pipeline completes.
+
+        Args:
+            pipeline_id: Pipeline identifier to clear
+        """
+        if not self.enabled:
+            return
+
+        with self.lock:
+            if pipeline_id in self.pipeline_contexts:
+                del self.pipeline_contexts[pipeline_id]
+            if pipeline_id in self.pipeline_items:
+                del self.pipeline_items[pipeline_id]
+            if pipeline_id in self.pipeline_module_order:
+                del self.pipeline_module_order[pipeline_id]
+
     def start_tracking(
         self,
         file: Optional[str] = None,
         module: Optional[str] = None,
         submodule: Optional[str] = None,
         message: str = "",
+        pipeline_id: Optional[str] = None,
     ) -> str:
         """
         Start tracking a progress item.
@@ -1004,6 +1228,17 @@ class ProgressTracker:
         # Create tracking ID
         tracking_id = f"{module}:{submodule}:{file or ''}"
 
+        # Determine pipeline_id and pipeline_order if module is part of a pipeline
+        pipeline_order = None
+        if pipeline_id is None and module:
+            # Try to find pipeline_id from existing contexts
+            for pid, modules in self.pipeline_contexts.items():
+                if module in modules:
+                    pipeline_id = pid
+                    if pid in self.pipeline_module_order:
+                        pipeline_order = self.pipeline_module_order[pid].get(module)
+                    break
+
         with self.lock:
             item = ProgressItem(
                 file=file,
@@ -1013,9 +1248,17 @@ class ProgressTracker:
                 start_time=time.time(),
                 message=message,
                 emoji=self._get_emoji_for_module(module or ""),
+                pipeline_id=pipeline_id,
+                pipeline_order=pipeline_order,
             )
 
             self.active_items[tracking_id] = item
+            
+            # If part of pipeline, also store in pipeline_items
+            if pipeline_id:
+                if pipeline_id not in self.pipeline_items:
+                    self.pipeline_items[pipeline_id] = {}
+                self.pipeline_items[pipeline_id][tracking_id] = item
 
             # Update displays
             for display in self.displays:
@@ -1057,9 +1300,19 @@ class ProgressTracker:
                     # Reset progress fields on completion
                     item.progress_percentage = 100.0 if status == "completed" else None
                     item.estimated_remaining = 0.0 if status == "completed" else None
-                    # Move to completed items
-                    self.items.append(item)
-                    del self.active_items[tracking_id]
+                    
+                    # If part of an active pipeline, keep it in pipeline_items instead of removing
+                    if item.pipeline_id and item.pipeline_id in self.pipeline_contexts:
+                        # Keep in pipeline_items for visibility
+                        if item.pipeline_id not in self.pipeline_items:
+                            self.pipeline_items[item.pipeline_id] = {}
+                        self.pipeline_items[item.pipeline_id][tracking_id] = item
+                        # Remove from active_items but keep in pipeline_items
+                        del self.active_items[tracking_id]
+                    else:
+                        # Not part of pipeline, move to completed items as before
+                        self.items.append(item)
+                        del self.active_items[tracking_id]
 
                 # Update displays
                 for display in self.displays:
@@ -1157,7 +1410,7 @@ class ProgressTracker:
         return max(0.0, eta_seconds)
 
     def stop_tracking(
-        self, tracking_id: str, status: str = "completed", message: str = ""
+        self, tracking_id: str, status: str = "completed", message: str = "", metadata: Optional[Dict[str, Any]] = None
     ) -> None:
         """
         Stop tracking an item.
@@ -1166,7 +1419,22 @@ class ProgressTracker:
             tracking_id: Tracking ID from start_tracking
             status: Final status (completed, failed)
             message: Final message
+            metadata: Optional metadata to store (e.g., extraction_counts, core_dependency)
         """
+        with self.lock:
+            # Update metadata if provided
+            if tracking_id in self.active_items:
+                item = self.active_items[tracking_id]
+                if metadata:
+                    item.metadata.update(metadata)
+            # Also check pipeline_items
+            for pipeline_id, items in self.pipeline_items.items():
+                if tracking_id in items:
+                    item = items[tracking_id]
+                    if metadata:
+                        item.metadata.update(metadata)
+                    break
+        
         self.update_tracking(tracking_id, status=status, message=message)
 
     def _get_emoji_for_module(self, module: str) -> str: