diff --git a/cookbook/advanced/07_Pipeline_Orchestration.ipynb b/cookbook/advanced/07_Pipeline_Orchestration.ipynb
index 2ad77850..6f124724 100644
--- a/cookbook/advanced/07_Pipeline_Orchestration.ipynb
+++ b/cookbook/advanced/07_Pipeline_Orchestration.ipynb
@@ -38,7 +38,9 @@
     "    PipelineBuilder,\n",
     "    ExecutionEngine,\n",
     "    FailureHandler,\n",
-    "    ParallelismManager\n",
+    "    ParallelismManager,\n",
+    "    RetryPolicy,\n",
+    "    RetryStrategy\n",
     ")\n",
     "from semantica.ingest import FileIngestor\n",
     "from semantica.parse import DocumentParser\n",
@@ -67,11 +69,46 @@
     "ner_extractor = NERExtractor()\n",
     "graph_builder = GraphBuilder()\n",
     "\n",
-    "pipeline = builder.add_step(\"ingest\", file_ingestor) \\\n",
-    "                  .add_step(\"parse\", document_parser) \\\n",
-    "                  .add_step(\"extract\", ner_extractor) \\\n",
-    "                  .add_step(\"build_graph\", graph_builder) \\\n",
-    "                  .build()\n"
+    "# Define handlers for each pipeline step\n",
+    "def ingest_handler(data, **config):\n",
+    "    files = data.get(\"files\", [])\n",
+    "    if files:\n",
+    "        # Ingest first file as example\n",
+    "        file_obj = file_ingestor.ingest_file(files[0], read_content=True)\n",
+    "        return {**data, \"file\": file_obj}\n",
+    "    return data\n",
+    "\n",
+    "def parse_handler(data, **config):\n",
+    "    # If a file was ingested, try parsing; otherwise pass text through\n",
+    "    file_obj = data.get(\"file\")\n",
+    "    if file_obj and getattr(file_obj, \"path\", None):\n",
+    "        parsed = document_parser.parse_document(file_obj.path)\n",
+    "        text = parsed.get(\"text\") if isinstance(parsed, dict) else None\n",
+    "        return {**data, \"text\": text or data.get(\"text\")}\n",
+    "    return data\n",
+    "\n",
+    "def extract_handler(data, **config):\n",
+    "    text = data.get(\"text\", \"\")\n",
+    "    entities = ner_extractor.extract_entities(text)\n",
+    "    # Normalize to dict list for graph builder\n",
+    "    entity_dicts = [\n",
+    "        {\"id\": f\"e{i}\", \"name\": e.text, \"type\": e.label} for i, e in enumerate(entities)\n",
+    "    ]\n",
+    "    return {**data, \"entities\": entity_dicts}\n",
+    "\n",
+    "def build_graph_handler(data, **config):\n",
+    "    entities = data.get(\"entities\", [])\n",
+    "    graph = graph_builder.build({\"entities\": entities})\n",
+    "    return {**data, \"graph\": graph}\n",
+    "\n",
+    "# Build pipeline with proper handlers and dependencies\n",
+    "pipeline = (\n",
+    "    builder\n",
+    "    .add_step(\"ingest\", \"ingest\", handler=ingest_handler)\n",
+    "    .add_step(\"parse\", \"parse\", dependencies=[\"ingest\"], handler=parse_handler)\n",
+    "    .add_step(\"extract\", \"extract\", dependencies=[\"parse\"], handler=extract_handler)\n",
+    "    .add_step(\"build_graph\", \"build_graph\", dependencies=[\"extract\"], handler=build_graph_handler)\n",
+    ").build()\n"
    ]
   },
   {
@@ -95,8 +132,8 @@
     "}\n",
     "\n",
     "start_time = time.time()\n",
-    "results = engine.execute(pipeline, input_data)\n",
-    "execution_time = time.time() - start_time\n"
+    "result = engine.execute_pipeline(pipeline, input_data)\n",
+    "execution_time = result.metrics.get(\"execution_time\", time.time() - start_time)\n"
    ]
   },
   {
@@ -112,17 +149,14 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "failure_handler = FailureHandler()\n",
-    "\n",
-    "pipeline_with_retry = failure_handler.configure_retry(pipeline, max_retries=3)\n",
-    "\n",
-    "pipeline_with_error_handling = failure_handler.configure_error_handling(\n",
-    "    pipeline_with_retry, \n",
-    "    on_error=\"skip\"\n",
+    "# Configure retry policy for the 'extract' step type\n",
+    "engine.failure_handler.set_retry_policy(\n",
+    "    \"extract\",\n",
+    "    RetryPolicy(max_retries=3, backoff_factor=2.0, strategy=RetryStrategy.EXPONENTIAL)\n",
     ")\n",
     "\n",
-    "results = engine.execute(pipeline_with_error_handling, input_data)\n",
-    "print(\"Pipeline executed successfully with error handling configured\")\n"
+    "result = engine.execute_pipeline(pipeline, input_data)\n",
+    "print(\"Pipeline executed with retry policy configured\")\n"
    ]
   },
   {
@@ -138,12 +172,16 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "parallelism = ParallelismManager()\n",
+    "parallelism = ParallelismManager(max_workers=4)\n",
     "\n",
-    "parallel_pipeline = parallelism.enable_parallel(pipeline, max_workers=4)\n",
+    "# Identify groups of steps that can run in parallel\n",
+    "groups = parallelism.identify_parallelizable_steps(pipeline)\n",
     "\n",
+    "# Execute first parallelizable group as a demonstration\n",
     "start_time = time.time()\n",
-    "results_parallel = engine.execute(parallel_pipeline, input_data)\n",
+    "parallel_results = []\n",
+    "for group in groups:\n",
+    "    parallel_results.extend(parallelism.execute_pipeline_steps_parallel(group, input_data, max_workers=4))\n",
     "parallel_time = time.time() - start_time\n"
    ]
   },
@@ -160,18 +198,14 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "metrics = engine.get_metrics() if hasattr(engine, 'get_metrics') else {\n",
-    "    'duration': execution_time,\n",
-    "    'items_processed': 1,\n",
-    "    'steps_completed': 4,\n",
-    "    'errors': 0\n",
-    "}\n",
+    "# Metrics from execution engine\n",
+    "metrics = result.metrics\n",
+    "progress = engine.get_progress(pipeline.name)\n",
     "\n",
-    "print(f\"Duration: {metrics.get('duration', 0):.2f} seconds\")\n",
-    "print(f\"Items Processed: {metrics.get('items_processed', 0)}\")\n",
-    "print(f\"Steps Completed: {metrics.get('steps_completed', 0)}\")\n",
-    "print(f\"Errors: {metrics.get('errors', 0)}\")\n",
-    "print(f\"Success Rate: {(1 - metrics.get('errors', 0) / max(metrics.get('items_processed', 1), 1)) * 100:.1f}%\")\n"
+    "print(f\"Duration: {metrics.get('execution_time', 0):.2f} seconds\")\n",
+    "print(f\"Steps Executed: {metrics.get('steps_executed', 0)}\")\n",
+    "print(f\"Steps Failed: {metrics.get('steps_failed', 0)}\")\n",
+    "print(f\"Progress: {progress.get('progress_percentage', 0):.1f}% (status: {progress.get('status')})\")\n"
    ]
   },
   {
diff --git a/cookbook/use_cases/healthcare/04_Healthcare_GraphRAG_Hybrid.ipynb b/cookbook/use_cases/healthcare/04_Healthcare_GraphRAG_Hybrid.ipynb
index 692f70ab..fa9e028d 100644
--- a/cookbook/use_cases/healthcare/04_Healthcare_GraphRAG_Hybrid.ipynb
+++ b/cookbook/use_cases/healthcare/04_Healthcare_GraphRAG_Hybrid.ipynb
@@ -710,24 +710,51 @@
     "# Build complete pipeline using Semantica PipelineBuilder\n",
     "pipeline_builder = PipelineBuilder()\n",
     "\n",
-    "healthcare_graphrag_pipeline = pipeline_builder \\\n",
-    "    .add_step(\"ingest\", \"file_ingest\", source=temp_dir) \\\n",
-    "    .add_step(\"parse\", \"structured_parse\", formats=[\"json\"]) \\\n",
-    "    .add_step(\"normalize\", \"text_normalize\") \\\n",
-    "    .add_step(\"extract\", \"semantic_extract\", entities=True, relations=True) \\\n",
-    "    .add_step(\"build_kg\", \"kg_build\") \\\n",
-    "    .add_step(\"generate_embeddings\", \"embedding_generate\") \\\n",
-    "    .add_step(\"setup_vector_store\", \"vector_store_setup\") \\\n",
-    "    .add_step(\"setup_triple_store\", \"triple_store_setup\") \\\n",
-    "    .add_step(\"orchestrate_query\", \"graphrag_query\") \\\n",
+    "# Define handlers using existing Semantica modules initialized earlier\n",
+    "def ingest_handler(data, **config):\n",
+    "    source_dir = config.get(\"source\", temp_dir)\n",
+    "    files = []\n",
+    "    try:\n",
+    "        files = [os.path.join(source_dir, f) for f in os.listdir(source_dir)]\n",
+    "    except Exception:\n",
+    "        pass\n",
+    "    if files:\n",
+    "        file_obj = file_ingestor.ingest_file(files[0], read_content=True)\n",
+    "        return {**data, \"file\": file_obj}\n",
+    "    return data\n",
+    "\n",
+    "def parse_handler(data, **config):\n",
+    "    file_obj = data.get(\"file\")\n",
+    "    if file_obj and getattr(file_obj, \"path\", None):\n",
+    "        parsed = document_parser.parse_document(file_obj.path)\n",
+    "        text = parsed.get(\"text\") if isinstance(parsed, dict) else None\n",
+    "        return {**data, \"text\": text or data.get(\"text\")}\n",
+    "    return data\n",
+    "\n",
+    "def extract_handler(data, **config):\n",
+    "    text = data.get(\"text\", \"\")\n",
+    "    entities = ner_extractor.extract_entities(text) if hasattr(ner_extractor, 'extract_entities') else ner_extractor.extract(text)\n",
+    "    entity_dicts = [{\"id\": f\"e{i}\", \"name\": getattr(e, 'text', str(e)), \"type\": getattr(e, 'label', getattr(e, 'type', 'Entity'))} for i, e in enumerate(entities)]\n",
+    "    return {**data, \"entities\": entity_dicts}\n",
+    "\n",
+    "def build_kg_handler(data, **config):\n",
+    "    entities = data.get(\"entities\", [])\n",
+    "    kg = graph_builder.build({\"entities\": entities})\n",
+    "    return {**data, \"knowledge_graph\": kg}\n",
+    "healthcare_graphrag_pipeline = (\n",
+    "    pipeline_builder\n",
+    "    .add_step(\"ingest\", \"ingest\", handler=ingest_handler, source=temp_dir)\n",
+    "    .add_step(\"parse\", \"parse\", dependencies=[\"ingest\"], handler=parse_handler)\n",
+    "    .add_step(\"extract\", \"extract\", dependencies=[\"parse\"], handler=extract_handler)\n",
+    "    .add_step(\"build_kg\", \"build_kg\", dependencies=[\"extract\"], handler=build_kg_handler)\n",
     "    .build()\n",
     "\n",
     "# Execute pipeline using Semantica ExecutionEngine\n",
     "execution_engine = ExecutionEngine()\n",
-    "pipeline_result = execution_engine.execute_pipeline(healthcare_graphrag_pipeline)\n",
+    "pipeline_result = execution_engine.execute_pipeline(healthcare_graphrag_pipeline, {\"files\": []})\n",
     "\n",
     "print(f\"  - Pipeline steps: {len(healthcare_graphrag_pipeline.steps)}\")\n",
-    "print(f\"  - Execution status: {pipeline_result.success if hasattr(pipeline_result, 'success') else 'Completed'}\")\n"
+    "print(f\"  - Execution status: {getattr(pipeline_result, 'success', True)}\")\n"
    ]
   },
   {
diff --git a/cookbook/use_cases/intelligence/01_Criminal_Network_Analysis.ipynb b/cookbook/use_cases/intelligence/01_Criminal_Network_Analysis.ipynb
index c9fe15e2..6d354424 100644
--- a/cookbook/use_cases/intelligence/01_Criminal_Network_Analysis.ipynb
+++ b/cookbook/use_cases/intelligence/01_Criminal_Network_Analysis.ipynb
@@ -703,16 +703,41 @@
     "    )\n",
     "    return report_data\n",
     "\n",
-    "# Build parallel agent pipeline\n",
-    "criminal_network_pipeline = pipeline_builder \\\n",
-    "    .add_step(\"data_gathering\", \"custom\", func=agent_data_gathering, args=([police_reports_file, court_records_file, surveillance_file], agent_memory)) \\\n",
-    "    .add_step(\"network_analysis\", \"custom\", func=agent_network_analysis, args=(criminal_kg, graph_analyzer, agent_memory)) \\\n",
-    "    .add_step(\"pattern_detection\", \"custom\", func=agent_pattern_detection, args=(criminal_kg, inference_engine, agent_memory)) \\\n",
-    "    .add_step(\"report_generation\", \"custom\", func=agent_report_generation, args=({\"key_players\": key_players_pagerank, \"communities\": communities}, agent_memory)) \\\n",
-    "    .build()\n",
-    "\n",
-    "# Execute pipeline with parallel execution\n",
-    "pipeline_result = execution_engine.execute_pipeline(criminal_network_pipeline, parallel=True)\n",
+    "def data_gathering_handler(data, **config):\n",
+    "    sources = data.get(\"sources\", [])\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = agent_data_gathering(sources, memory)\n",
+    "    return {**data, \"gathered\": r}\n",
+    "def network_analysis_handler(data, **config):\n",
+    "    graph = data.get(\"criminal_kg\")\n",
+    "    analyzer = data.get(\"graph_analyzer\")\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = agent_network_analysis(graph, analyzer, memory)\n",
+    "    return {**data, \"network_analysis\": r}\n",
+    "def pattern_detection_handler(data, **config):\n",
+    "    graph = data.get(\"criminal_kg\")\n",
+    "    inf = data.get(\"inference_engine\")\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = agent_pattern_detection(graph, inf, memory)\n",
+    "    return {**data, \"patterns\": r}\n",
+    "def report_generation_handler(data, **config):\n",
+    "    memory = data.get(\"memory\")\n",
+    "    analysis_results = {\n",
+    "        \"network\": data.get(\"network_analysis\"),\n",
+    "        \"patterns\": data.get(\"patterns\")\n",
+    "    }\n",
+    "    r = agent_report_generation(analysis_results, memory)\n",
+    "    return {**data, \"report\": r}\n",
+    "criminal_network_pipeline = (\n",
+    "    pipeline_builder\n",
+    "    .add_step(\"data_gathering\", \"ingest\", handler=data_gathering_handler)\n",
+    "    .add_step(\"network_analysis\", \"analyze_graph\", dependencies=[\"data_gathering\"], handler=network_analysis_handler)\n",
+    "    .add_step(\"pattern_detection\", \"analysis\", dependencies=[\"network_analysis\"], handler=pattern_detection_handler)\n",
+    "    .add_step(\"report_generation\", \"report\", dependencies=[\"pattern_detection\"], handler=report_generation_handler)\n",
+    ")\n",
+    ".build()\n",
+    "input_data = {\"sources\": [police_reports_file, court_records_file, surveillance_file], \"criminal_kg\": criminal_kg, \"graph_analyzer\": graph_analyzer, \"inference_engine\": inference_engine, \"memory\": agent_memory}\n",
+    "pipeline_result = execution_engine.execute_pipeline(criminal_network_pipeline, data=input_data, parallel=True)\n",
     "\n",
     "print(f\"  - Pipeline steps: {len(criminal_network_pipeline.steps)}\")\n",
     "print(f\"  - Parallel execution: Enabled\")\n",
diff --git a/cookbook/use_cases/intelligence/02_Intelligence_Analysis.ipynb b/cookbook/use_cases/intelligence/02_Intelligence_Analysis.ipynb
index 8ac95f01..9b10c347 100644
--- a/cookbook/use_cases/intelligence/02_Intelligence_Analysis.ipynb
+++ b/cookbook/use_cases/intelligence/02_Intelligence_Analysis.ipynb
@@ -626,23 +626,63 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "# Build complete pipeline with all workers\n",
-    "intelligence_pipeline = pipeline_builder \\\n",
-    "    .add_step(\"data_ingestion\", \"custom\", func=data_ingestion_worker, args=([], agent_memory)) \\\n",
-    "    .add_step(\"ontology_building\", \"custom\", func=ontology_building_worker, args=(sample_entities, sample_relationships, agent_memory)) \\\n",
-    "    .add_step(\"graph_construction\", \"custom\", func=graph_construction_worker, args=(sample_entities, sample_relationships, agent_memory)) \\\n",
-    "    .add_step(\"graph_analytics\", \"custom\", func=graph_analytics_worker, args=(graph_result.get('knowledge_graph'), agent_memory)) \\\n",
-    "    .add_step(\"hybrid_rag\", \"custom\", func=hybrid_rag_worker, args=(graph_result.get('knowledge_graph'), vector_store, sample_entities, agent_memory)) \\\n",
-    "    .add_step(\"intelligence_analysis\", \"custom\", func=intelligence_analysis_worker, args=(graph_result.get('knowledge_graph'), analytics_result, agent_memory)) \\\n",
-    "    .add_step(\"report_generation\", \"custom\", func=report_generation_worker, args=(all_worker_results, agent_memory)) \\\n",
-    "    .build()\n",
+    "def data_ingestion_handler(data, **config):\n",
+    "    sources = data.get(\"sources\", [])\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = data_ingestion_worker(sources, memory)\n",
+    "    return {**data, \"osint\": r}\n",
+    "def ontology_handler(data, **config):\n",
+    "    entities = data.get(\"entities\", [])\n",
+    "    relationships = data.get(\"relationships\", [])\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = ontology_building_worker(entities, relationships, memory)\n",
+    "    return {**data, \"ontology_result\": r}\n",
+    "def graph_handler(data, **config):\n",
+    "    entities = data.get(\"entities\", [])\n",
+    "    relationships = data.get(\"relationships\", [])\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = graph_construction_worker(entities, relationships, memory)\n",
+    "    return {**data, \"graph_result\": r}\n",
+    "def analytics_handler(data, **config):\n",
+    "    kg = data.get(\"graph_result\", {}).get(\"knowledge_graph\")\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = graph_analytics_worker(kg, memory)\n",
+    "    return {**data, \"analytics_result\": r}\n",
+    "def rag_handler(data, **config):\n",
+    "    kg = data.get(\"graph_result\", {}).get(\"knowledge_graph\")\n",
+    "    vector_store = data.get(\"vector_store\")\n",
+    "    entities = data.get(\"entities\", [])\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = hybrid_rag_worker(kg, vector_store, entities, memory)\n",
+    "    return {**data, \"rag_result\": r}\n",
+    "def intel_handler(data, **config):\n",
+    "    kg = data.get(\"graph_result\", {}).get(\"knowledge_graph\")\n",
+    "    analytics_results = data.get(\"analytics_result\")\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = intelligence_analysis_worker(kg, analytics_results, memory)\n",
+    "    return {**data, \"intelligence_result\": r}\n",
+    "def report_handler(data, **config):\n",
+    "    memory = data.get(\"memory\")\n",
+    "    all_results = {\"ontology\": data.get(\"ontology_result\"), \"graph\": data.get(\"graph_result\"), \"analytics\": data.get(\"analytics_result\"), \"rag\": data.get(\"rag_result\"), \"intelligence\": data.get(\"intelligence_result\")}\n",
+    "    r = report_generation_worker(all_results, memory)\n",
+    "    return {**data, \"report_result\": r}\n",
+    "intelligence_pipeline = (\n",
+    "    pipeline_builder\n",
+    "    .add_step(\"data_ingestion\", \"ingest\", handler=data_ingestion_handler)\n",
+    "    .add_step(\"ontology_building\", \"ontology\", dependencies=[\"data_ingestion\"], handler=ontology_handler)\n",
+    "    .add_step(\"graph_construction\", \"build_graph\", dependencies=[\"ontology_building\"], handler=graph_handler)\n",
+    "    .add_step(\"graph_analytics\", \"analyze_graph\", dependencies=[\"graph_construction\"], handler=analytics_handler)\n",
+    "    .add_step(\"hybrid_rag\", \"rag\", dependencies=[\"graph_construction\"], handler=rag_handler)\n",
+    "    .add_step(\"intelligence_analysis\", \"analysis\", dependencies=[\"graph_analytics\", \"hybrid_rag\"], handler=intel_handler)\n",
+    "    .add_step(\"report_generation\", \"report\", dependencies=[\"intelligence_analysis\"], handler=report_handler)\n",
+    ")\n",
+    ".build()\n",
     "\n",
-    "# Execute pipeline with parallel workers\n",
-    "pipeline_result = orchestrator.execute_pipeline(intelligence_pipeline, parallel=True, max_workers=7)\n",
+    "input_data = {\"sources\": [], \"entities\": sample_entities, \"relationships\": sample_relationships, \"vector_store\": vector_store, \"memory\": agent_memory}\n",
+    "pipeline_result = orchestrator.execute_pipeline(intelligence_pipeline, data=input_data, parallel=True, max_workers=7)\n",
     "\n",
     "print(f\"  - Pipeline steps: {len(intelligence_pipeline.steps)}\")\n",
-    "print(f\"  - Parallel execution: Enabled (7 workers)\")\n",
-    "print(f\"  - Execution status: {pipeline_result.success if hasattr(pipeline_result, 'success') else 'Completed'}\")\n"
+    "print(f\"  - Execution status: {getattr(pipeline_result, 'success', True)}\")\n"
    ]
   },
   {
diff --git a/cookbook/use_cases/intelligence/03_Law_Enforcement_Forensics.ipynb b/cookbook/use_cases/intelligence/03_Law_Enforcement_Forensics.ipynb
index 5d88dd26..cf68e9c0 100644
--- a/cookbook/use_cases/intelligence/03_Law_Enforcement_Forensics.ipynb
+++ b/cookbook/use_cases/intelligence/03_Law_Enforcement_Forensics.ipynb
@@ -435,15 +435,42 @@
     "    )\n",
     "    return {\"report_data\": analysis_results, \"context_items\": len(context)}\n",
     "\n",
-    "# Build and execute agent pipeline\n",
-    "forensic_pipeline = pipeline_builder \\\n",
-    "    .add_step(\"evidence_collection\", \"custom\", func=agent_evidence_collection, args=(evidence_logs_data, agent_memory)) \\\n",
-    "    .add_step(\"timeline_analysis\", \"custom\", func=agent_timeline_analysis, args=(forensic_kg, temporal_query, agent_memory)) \\\n",
-    "    .add_step(\"cross_case_correlation\", \"custom\", func=agent_cross_case_correlation, args=(forensic_kg, graph_analyzer, agent_memory)) \\\n",
-    "    .add_step(\"forensic_report\", \"custom\", func=agent_forensic_report, args=({\"cases\": len(cases_data.get('cases', [])) if isinstance(cases_data, dict) else 0}, agent_memory)) \\\n",
-    "    .build()\n",
-    "\n",
-    "pipeline_result = execution_engine.execute_pipeline(forensic_pipeline, parallel=True)\n",
+    "def evidence_collection_handler(data, **config):\n",
+    "    evidence_data = data.get(\"evidence_logs_data\")\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = agent_evidence_collection(evidence_data, memory)\n",
+    "    return {**data, \"evidence_collection_result\": r}\n",
+    "def timeline_handler(data, **config):\n",
+    "    kg = data.get(\"forensic_kg\")\n",
+    "    temporal = data.get(\"temporal_query\")\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = agent_timeline_analysis(kg, temporal, memory)\n",
+    "    return {**data, \"timeline_result\": r}\n",
+    "def correlation_handler(data, **config):\n",
+    "    kg = data.get(\"forensic_kg\")\n",
+    "    analyzer = data.get(\"graph_analyzer\")\n",
+    "    memory = data.get(\"memory\")\n",
+    "    r = agent_cross_case_correlation(kg, analyzer, memory)\n",
+    "    return {**data, \"correlation_result\": r}\n",
+    "def report_handler(data, **config):\n",
+    "    memory = data.get(\"memory\")\n",
+    "    analysis = {\n",
+    "        \"evidence\": data.get(\"evidence_collection_result\"),\n",
+    "        \"timeline\": data.get(\"timeline_result\"),\n",
+    "        \"correlation\": data.get(\"correlation_result\")\n",
+    "    }\n",
+    "    r = agent_forensic_report(analysis, memory)\n",
+    "    return {**data, \"forensic_report_result\": r}\n",
+    "forensic_pipeline = (\n",
+    "    pipeline_builder\n",
+    "    .add_step(\"evidence_collection\", \"ingest\", handler=evidence_collection_handler)\n",
+    "    .add_step(\"timeline_analysis\", \"analyze_graph\", dependencies=[\"evidence_collection\"], handler=timeline_handler)\n",
+    "    .add_step(\"cross_case_correlation\", \"analyze_graph\", dependencies=[\"timeline_analysis\"], handler=correlation_handler)\n",
+    "    .add_step(\"forensic_report\", \"report\", dependencies=[\"cross_case_correlation\"], handler=report_handler)\n",
+    ")\n",
+    ".build()\n",
+    "input_data = {\"evidence_logs_data\": evidence_logs_data, \"forensic_kg\": forensic_kg, \"temporal_query\": temporal_query, \"graph_analyzer\": graph_analyzer, \"memory\": agent_memory, \"cases_data\": cases_data}\n",
+    "pipeline_result = execution_engine.execute_pipeline(forensic_pipeline, data=input_data, parallel=True)\n",
     "\n",
     "print(f\"  - Pipeline steps: {len(forensic_pipeline.steps)}\")\n",
     "print(f\"  - Parallel execution: Enabled\")\n"
diff --git a/cookbook/use_cases/supply_chain/02_Supply_Chain_Risk_Management.ipynb b/cookbook/use_cases/supply_chain/02_Supply_Chain_Risk_Management.ipynb
index 73f869af..86275d39 100644
--- a/cookbook/use_cases/supply_chain/02_Supply_Chain_Risk_Management.ipynb
+++ b/cookbook/use_cases/supply_chain/02_Supply_Chain_Risk_Management.ipynb
@@ -609,21 +609,70 @@
     "# Build complete pipeline using Semantica PipelineBuilder\n",
     "pipeline_builder = PipelineBuilder()\n",
     "\n",
-    "supply_chain_pipeline = pipeline_builder \\\n",
-    "    .add_step(\"ingest\", \"file_ingest\", source=temp_dir) \\\n",
-    "    .add_step(\"parse\", \"structured_parse\", formats=[\"json\"]) \\\n",
-    "    .add_step(\"normalize\", \"data_normalize\") \\\n",
-    "    .add_step(\"extract\", \"relation_extract\") \\\n",
-    "    .add_step(\"build_kg\", \"kg_build\") \\\n",
-    "    .add_step(\"analyze_risks\", \"graph_analyze\") \\\n",
-    "    .add_step(\"propagate_risks\", \"reasoning_infer\") \\\n",
-    "    .add_step(\"visualize\", \"visualize_network\") \\\n",
-    "    .add_step(\"generate_report\", \"export_report\") \\\n",
+    "def ingest_handler(data, **config):\n",
+    "    source_dir = config.get(\"source\", temp_dir)\n",
+    "    files = []\n",
+    "    try:\n",
+    "        files = [os.path.join(source_dir, f) for f in os.listdir(source_dir)]\n",
+    "    except Exception:\n",
+    "        pass\n",
+    "    return {**data, \"files\": files}\n",
+    "def parse_handler(data, **config):\n",
+    "    records = []\n",
+    "    files = data.get(\"files\", [])\n",
+    "    for fp in files:\n",
+    "        try:\n",
+    "            with open(fp, \"r\", encoding=\"utf-8\") as f:\n",
+    "                records.append(json.load(f))\n",
+    "        except Exception:\n",
+    "            pass\n",
+    "    return {**data, \"records\": records}\n",
+    "def normalize_handler(data, **config):\n",
+    "    records = data.get(\"records\", [])\n",
+    "    normalized = records\n",
+    "    return {**data, \"normalized\": normalized}\n",
+    "def extract_handler(data, **config):\n",
+    "    normalized = data.get(\"normalized\", [])\n",
+    "    entities = data.get(\"entities\") or []\n",
+    "    relationships = data.get(\"relationships\") or []\n",
+    "    return {**data, \"entities\": entities, \"relationships\": relationships}\n",
+    "def build_kg_handler(data, **config):\n",
+    "    ents = data.get(\"entities\", [])\n",
+    "    rels = data.get(\"relationships\", [])\n",
+    "    kg = graph_builder.build({\"entities\": ents, \"relationships\": rels})\n",
+    "    return {**data, \"supply_chain_kg\": kg}\n",
+    "def analyze_risks_handler(data, **config):\n",
+    "    kg = data.get(\"supply_chain_kg\")\n",
+    "    risks = graph_analyzer.compute_metrics(kg) if kg is not None else {}\n",
+    "    return {**data, \"risk_analysis\": risks}\n",
+    "def propagate_risks_handler(data, **config):\n",
+    "    kg = data.get(\"supply_chain_kg\")\n",
+    "    propagation = reasoning_engine.infer(knowledge_graph=kg, rules=[]) if kg is not None else {}\n",
+    "    return {**data, \"risk_propagation\": propagation}\n",
+    "def visualize_handler(data, **config):\n",
+    "    kg = data.get(\"supply_chain_kg\")\n",
+    "    fig = KGVisualizer().visualize_network(kg) if kg is not None else None\n",
+    "    return {**data, \"visualization\": fig}\n",
+    "def report_handler(data, **config):\n",
+    "    report = ReportGenerator().generate({\"risks\": data.get(\"risk_analysis\"), \"propagation\": data.get(\"risk_propagation\")})\n",
+    "    return {**data, \"report\": report}\n",
+    "supply_chain_pipeline = (\n",
+    "    pipeline_builder\n",
+    "    .add_step(\"ingest\", \"ingest\", handler=ingest_handler, source=temp_dir)\n",
+    "    .add_step(\"parse\", \"parse\", dependencies=[\"ingest\"], handler=parse_handler, formats=[\"json\"])\n",
+    "    .add_step(\"normalize\", \"normalize\", dependencies=[\"parse\"], handler=normalize_handler)\n",
+    "    .add_step(\"extract\", \"extract\", dependencies=[\"normalize\"], handler=extract_handler)\n",
+    "    .add_step(\"build_kg\", \"build_kg\", dependencies=[\"extract\"], handler=build_kg_handler)\n",
+    "    .add_step(\"analyze_risks\", \"analyze_graph\", dependencies=[\"build_kg\"], handler=analyze_risks_handler)\n",
+    "    .add_step(\"propagate_risks\", \"reasoning\", dependencies=[\"build_kg\"], handler=propagate_risks_handler)\n",
+    "    .add_step(\"visualize\", \"visualize\", dependencies=[\"analyze_risks\"], handler=visualize_handler)\n",
+    "    .add_step(\"generate_report\", \"report\", dependencies=[\"visualize\"], handler=report_handler)\n",
     "    .build()\n",
+    ")\n",
     "\n",
-    "# Execute pipeline using Semantica ExecutionEngine\n",
     "execution_engine = ExecutionEngine()\n",
-    "pipeline_result = execution_engine.execute_pipeline(supply_chain_pipeline)\n",
+    "input_data = {\"entities\": supply_chain_entities if 'supply_chain_entities' in globals() else [], \"relationships\": supply_chain_relationships if 'supply_chain_relationships' in globals() else []}\n",
+    "pipeline_result = execution_engine.execute_pipeline(supply_chain_pipeline, data=input_data)\n",
     "\n",
     "print(f\"  - Pipeline steps: {len(supply_chain_pipeline.steps)}\")\n",
     "print(f\"  - Execution status: {pipeline_result.success if hasattr(pipeline_result, 'success') else 'Completed'}\")\n"
diff --git a/docs/reference/pipeline.md b/docs/reference/pipeline.md
index b9432fb2..a4cf341f 100644
--- a/docs/reference/pipeline.md
+++ b/docs/reference/pipeline.md
@@ -77,7 +77,25 @@
 
 ---
 
-## Main Classes
+## API Reference
+
+### Types
+
+- `Pipeline` — Pipeline definition dataclass
+- `PipelineStep` — Pipeline step definition dataclass
+- `StepStatus` — Enum: `pending`, `running`, `completed`, `failed`, `skipped`
+- `ExecutionResult` — Execution result dataclass
+- `PipelineStatus` — Enum: `pending`, `running`, `paused`, `completed`, `failed`, `stopped`
+- `ValidationResult` — Validation result dataclass
+- `RetryPolicy` — Retry policy dataclass
+- `RetryStrategy` — Enum: `linear`, `exponential`, `fixed`
+- `ErrorSeverity` — Enum: `low`, `medium`, `high`, `critical`
+- `FailureRecovery` — Failure recovery dataclass
+- `Task` — Parallel task dataclass
+- `ParallelExecutionResult` — Parallel execution result dataclass
+- `ResourceType` — Enum: `cpu`, `gpu`, `memory`, `disk`, `network`
+- `Resource` — Resource definition dataclass
+- `ResourceAllocation` — Resource allocation record dataclass
 
 ### PipelineBuilder
 
@@ -85,27 +103,57 @@ Fluent interface for constructing pipelines.
 
 **Methods:**
 
-| Method | Description |
-|--------|-------------|
-| `add_step(name, func)` | Add a processing step |
-| `add_dependency(step, dep)` | Define execution order |
-| `set_error_handler(handler)` | Configure error handling |
-| `build()` | Create immutable Pipeline object |
+- `add_step(step_name, step_type, **config)` — Add a step
+- `connect_steps(from_step, to_step, **options)` — Add dependency
+- `set_parallelism(level)` — Configure parallelism
+- `build(name="default_pipeline")` — Build pipeline
+- `build_pipeline(pipeline_config, **options)` — Build from dict
+- `register_step_handler(step_type, handler)` — Register handler
+- `get_step(step_name)` — Get step by name
+- `serialize(format="json")` — Serialize builder state
+- `validate_pipeline()` — Validate pipeline
 
 **Example:**
 
 ```python
 from semantica.pipeline import PipelineBuilder
 
-pipeline = (
+builder = (
     PipelineBuilder()
-    .add_step("ingest", ingest_func)
-    .add_step("parse", parse_func)
-    .add_step("embed", embed_func)
-    .add_dependency("parse", "ingest")  # parse depends on ingest
-    .add_dependency("embed", "parse")   # embed depends on parse
-    .build()
+    .add_step("ingest", "ingest", handler=ingest_handler)
+    .add_step("parse", "parse", dependencies=["ingest"], handler=parse_handler)
+    .add_step("embed", "embed", dependencies=["parse"], model="text-embedding-3-large")
+    .set_parallelism(2)
 )
+pipeline = builder.build(name="MyPipeline")
+
+step = builder.get_step("parse")
+serialized = builder.serialize(format="json")
+validation = builder.validate_pipeline()
+```
+
+### PipelineSerializer
+
+Serialization utilities for pipelines.
+
+**Methods:**
+
+- `serialize_pipeline(pipeline, format="json")`
+- `deserialize_pipeline(serialized_pipeline, **options)`
+- `version_pipeline(pipeline, version_info)`
+
+**Example:**
+
+```python
+from semantica.pipeline import PipelineBuilder, PipelineSerializer
+
+builder = PipelineBuilder()
+pipeline = builder.add_step("step1", "type1").build()
+
+serializer = PipelineSerializer()
+serialized = serializer.serialize_pipeline(pipeline, format="json")
+restored = serializer.deserialize_pipeline(serialized)
+versioned = serializer.version_pipeline(restored, {"version": "1.1"})
 ```
 
 ### ExecutionEngine
@@ -114,74 +162,158 @@ Executes pipelines and manages lifecycle.
 
 **Methods:**
 
-| Method | Description |
-|--------|-------------|
-| `execute(pipeline, input)` | Run pipeline synchronously |
-| `execute_async(pipeline)` | Run in background |
-| `resume(execution_id)` | Resume failed execution |
-| `get_status(execution_id)` | Check progress |
+- `execute_pipeline(pipeline, data=None, **options)` — Run pipeline
+- `pause_pipeline(pipeline_id)` — Pause execution
+- `resume_pipeline(pipeline_id)` — Resume execution
+- `stop_pipeline(pipeline_id)` — Stop execution
+- `get_pipeline_status(pipeline_id)` — Get status
+- `get_progress(pipeline_id)` — Get progress
 
 **Example:**
 
 ```python
 from semantica.pipeline import ExecutionEngine
 
-engine = ExecutionEngine()
-result = engine.execute(pipeline, input_data={"files": ["doc.pdf"]})
+engine = ExecutionEngine(max_workers=4)
+result = engine.execute_pipeline(pipeline, data={"path": "document.pdf"})
 
-if result.status == "COMPLETED":
-    print("Success:", result.output)
-else:
-    print("Failed:", result.error)
+status = engine.get_pipeline_status(pipeline.name)
+progress = engine.get_progress(pipeline.name)
+
+engine.pause_pipeline(pipeline.name)
+engine.resume_pipeline(pipeline.name)
+engine.stop_pipeline(pipeline.name)
 ```
 
-### FailureHandler
+### Failure Handling
 
-Manages retries and error recovery.
+**Classes:** `FailureHandler`, `RetryHandler`, `FallbackHandler`, `ErrorRecovery`
+
+**FailureHandler Methods:**
+
+- `handle_step_failure(step, error, **options)`
+- `classify_error(error)`
+- `set_retry_policy(step_type, policy)`
+- `get_retry_policy(step_type)`
+- `retry_failed_step(step, error, **options)`
+- `get_error_history(step_name=None)`
+- `clear_error_history()`
+
+**Example:**
+
+```python
+from semantica.pipeline import FailureHandler, RetryPolicy, RetryStrategy
+
+handler = FailureHandler(default_max_retries=3, default_backoff_factor=2.0)
+policy = RetryPolicy(max_retries=5, strategy=RetryStrategy.EXPONENTIAL, initial_delay=1.0)
+handler.set_retry_policy("network", policy)
+
+classification = handler.classify_error(RuntimeError("timeout"))
+history_before = handler.get_error_history()
+handler.clear_error_history()
+```
+
+### Parallelism
+
+**Classes:** `ParallelismManager`, `ParallelExecutor`
+
+**ParallelismManager Methods:**
+
+- `execute_parallel(tasks, **options)`
+- `execute_pipeline_steps_parallel(steps, data, **options)`
+- `identify_parallelizable_steps(pipeline)`
+- `optimize_parallel_execution(pipeline, available_workers)`
+
+**Example:**
+
+```python
+from semantica.pipeline import ParallelismManager, Task, ParallelExecutor
+
+def work(x):
+    return x * 2
+
+manager = ParallelismManager(max_workers=4)
+tasks = [Task(task_id=f"t{i}", handler=work, args=(i,)) for i in range(4)]
+results = manager.execute_parallel(tasks)
+
+executor = ParallelExecutor(max_workers=2)
+exec_results = executor.execute_parallel(tasks)
+```
+
+### Resources
+
+**Class:** `ResourceScheduler`
 
 **Methods:**
 
-| Method | Description |
-|--------|-------------|
-| `handle_error(error, context)` | Process error |
-| `should_retry(attempt)` | Check retry policy |
+- `allocate_resources(pipeline, **options)`
+- `allocate_cpu(cores, pipeline_id, step_name=None)`
+- `allocate_memory(memory_gb, pipeline_id, step_name=None)`
+- `allocate_gpu(device_id, pipeline_id, step_name=None)`
+- `release_resources(allocations)`
+- `get_resource_usage()`
+- `optimize_resource_allocation(pipeline, **options)`
 
-**Configuration:**
+**Example:**
 
 ```python
-from semantica.pipeline import RetryPolicy
+from semantica.pipeline import ResourceScheduler, ResourceType
 
-policy = RetryPolicy(
-    max_retries=3,
-    backoff_factor=2.0,
-    exceptions=[NetworkError, TimeoutError]
-)
+scheduler = ResourceScheduler()
+cpu = scheduler.allocate_cpu(cores=2, pipeline_id="p1")
+mem = scheduler.allocate_memory(memory_gb=1.0, pipeline_id="p1")
+usage = scheduler.get_resource_usage()
+scheduler.release_resources({cpu.allocation_id: cpu, mem.allocation_id: mem})
 ```
 
-### ParallelismManager
+### Validation
 
-Manages concurrent execution.
+**Class:** `PipelineValidator`
 
 **Methods:**
 
-| Method | Description |
-|--------|-------------|
-| `execute_parallel(tasks)` | Run tasks concurrently |
-| `map(func, items)` | Parallel map operation |
+- `validate_pipeline(pipeline_or_builder, **options)`
+- `validate_step(step, **constraints)`
+- `check_dependencies(pipeline_or_builder)`
+- `validate_performance(pipeline, **options)`
 
----
-
-## Convenience Functions
+**Example:**
 
 ```python
-from semantica.pipeline import build_linear_pipeline
+from semantica.pipeline import PipelineValidator, PipelineBuilder
 
-# Quick linear pipeline
-pipeline = build_linear_pipeline([
-    step1_func,
-    step2_func,
-    step3_func
-])
+builder = PipelineBuilder()
+builder.add_step("a", "type")
+builder.add_step("b", "type", dependencies=["a"])
+
+validator = PipelineValidator()
+result = validator.validate_pipeline(builder)
+deps = validator.check_dependencies(builder)
+perf = validator.validate_performance(builder.build())
+```
+
+### Templates
+
+**Classes:** `PipelineTemplateManager`, `PipelineTemplate`
+
+**PipelineTemplateManager Methods:**
+
+- `get_template(template_name)`
+- `create_pipeline_from_template(template_name, **overrides)`
+- `register_template(template)`
+- `list_templates(category=None)`
+- `get_template_info(template_name)`
+
+**Example:**
+
+```python
+from semantica.pipeline import PipelineTemplateManager
+
+tm = PipelineTemplateManager()
+names = tm.list_templates()
+info = tm.get_template_info(names[0])
+builder = tm.create_pipeline_from_template(names[0])
+pipeline = builder.build()
 ```
 
 ---
@@ -198,62 +330,28 @@ export PIPELINE_CHECKPOINT_DIR=./checkpoints
 
 ### YAML Configuration
 
-```yaml
-pipeline:
-  execution:
-    max_workers: 4
-    timeout_seconds: 300
-    
-  retry:
-    default_retries: 3
-    backoff_factor: 1.5
-    
-  resources:
-    max_memory_mb: 4096
-```
+This module does not include built-in YAML loaders. Use your own configuration system to populate arguments for `PipelineBuilder`, `ExecutionEngine`, and related classes.
 
 ---
 
 ## Integration Examples
 
-### Complete RAG Ingestion Pipeline
+### RAG-Style Pipeline
 
 ```python
 from semantica.pipeline import PipelineBuilder, ExecutionEngine
-from semantica.ingest import Ingestor
-from semantica.split import TextSplitter
-from semantica.embeddings import EmbeddingGenerator
-from semantica.vector_store import VectorStore
 
-# 1. Define Steps
-def ingest(data):
-    return Ingestor().ingest(data['path'])
-
-def split(data):
-    return TextSplitter().split(data['text'])
-
-def embed(data):
-    return EmbeddingGenerator().generate(data['chunks'])
-
-def store(data):
-    return VectorStore().store(data['embeddings'])
-
-# 2. Build Pipeline
-pipeline = (
+builder = (
     PipelineBuilder()
-    .add_step("ingest", ingest)
-    .add_step("split", split)
-    .add_step("embed", embed)
-    .add_step("store", store)
-    .add_dependency("split", "ingest")
-    .add_dependency("embed", "split")
-    .add_dependency("store", "embed")
-    .build()
+    .add_step("ingest", "ingest")
+    .add_step("chunk", "chunk", dependencies=["ingest"])
+    .add_step("embed", "embed", dependencies=["chunk"]) 
+    .add_step("store_vectors", "store_vectors", dependencies=["embed"]) 
 )
 
-# 3. Execute
-engine = ExecutionEngine()
-result = engine.execute(pipeline, {"path": "document.pdf"})
+pipeline = builder.build(name="RAGPipeline")
+engine = ExecutionEngine(max_workers=4)
+result = engine.execute_pipeline(pipeline, data={"path": "document.pdf"})
 ```
 
 ---
diff --git a/semantica/pipeline/pipeline_usage.md b/semantica/pipeline/pipeline_usage.md
index 8ec2934c..37cc3651 100644
--- a/semantica/pipeline/pipeline_usage.md
+++ b/semantica/pipeline/pipeline_usage.md
@@ -140,17 +140,14 @@ pipeline = builder.build()
 ### Pipeline Serialization
 
 ```python
-from semantica.pipeline import PipelineBuilder
+from semantica.pipeline import PipelineBuilder, PipelineSerializer
 
 builder = PipelineBuilder()
 pipeline = builder.add_step("step1", "type1").build()
 
-# Serialize pipeline to JSON
-serialized = builder.serialize(pipeline, format="json")
-print(serialized)
-
-# Deserialize pipeline
-deserialized = builder.deserialize(serialized, format="json")
+serializer = PipelineSerializer()
+serialized = serializer.serialize_pipeline(pipeline, format="json")
+restored = serializer.deserialize_pipeline(serialized)
 ```
 
 ### Pipeline Metadata
@@ -210,17 +207,10 @@ print(f"Metrics: {result.metrics}")
 from semantica.pipeline import ExecutionEngine, PipelineStatus
 
 engine = ExecutionEngine()
-
-# Execute pipeline
 result = engine.execute_pipeline(pipeline)
 
-# Get pipeline status
-status = engine.get_status(pipeline.name)
-print(f"Status: {status}")
-
-# Check if running
-if status == PipelineStatus.RUNNING:
-    print("Pipeline is currently running")
+status = engine.get_pipeline_status(pipeline.name)
+print(status.value)
 ```
 
 ### Progress Monitoring
@@ -229,15 +219,13 @@ if status == PipelineStatus.RUNNING:
 from semantica.pipeline import ExecutionEngine
 
 engine = ExecutionEngine()
-
-# Execute pipeline
 result = engine.execute_pipeline(pipeline)
 
-# Get progress
 progress = engine.get_progress(pipeline.name)
-print(f"Progress: {progress.get('percentage', 0)}%")
-print(f"Completed steps: {progress.get('completed_steps', 0)}")
-print(f"Total steps: {progress.get('total_steps', 0)}")
+print(progress["progress_percentage"])  # float 0..100
+print(progress["completed_steps"])      # int
+print(progress["total_steps"])          # int
+print(progress["status"])               # status string
 ```
 
 ### Pause and Resume
@@ -246,17 +234,10 @@ print(f"Total steps: {progress.get('total_steps', 0)}")
 from semantica.pipeline import ExecutionEngine
 
 engine = ExecutionEngine()
+result = engine.execute_pipeline(pipeline)
 
-# Start execution
-result_future = engine.execute_pipeline_async(pipeline)
-
-# Pause execution
 engine.pause_pipeline(pipeline.name)
-
-# Resume execution
 engine.resume_pipeline(pipeline.name)
-
-# Stop execution
 engine.stop_pipeline(pipeline.name)
 ```
 
@@ -266,16 +247,12 @@ engine.stop_pipeline(pipeline.name)
 from semantica.pipeline import ExecutionEngine
 
 engine = ExecutionEngine()
-
-# Execute pipeline
 result = engine.execute_pipeline(pipeline)
 
-# Access execution metrics
 metrics = result.metrics
-print(f"Total execution time: {metrics.get('total_time', 0)}s")
-print(f"Steps executed: {metrics.get('steps_executed', 0)}")
-print(f"Steps failed: {metrics.get('steps_failed', 0)}")
-print(f"Memory used: {metrics.get('memory_used', 0)}MB")
+print(metrics.get("execution_time", 0))
+print(metrics.get("steps_executed", 0))
+print(metrics.get("steps_failed", 0))
 ```
 
 ## Error Handling
@@ -285,26 +262,15 @@ print(f"Memory used: {metrics.get('memory_used', 0)}MB")
 ```python
 from semantica.pipeline import FailureHandler, RetryPolicy, RetryStrategy
 
-# Create failure handler
 handler = FailureHandler()
+policy = RetryPolicy(max_retries=3, strategy=RetryStrategy.EXPONENTIAL, initial_delay=1.0)
 
-# Configure retry policy
-policy = RetryPolicy(
-    max_retries=3,
-    strategy=RetryStrategy.EXPONENTIAL,
-    initial_delay=1.0,
-    backoff_factor=2.0,
-    max_delay=60.0
-)
-
-# Handle step failure
 try:
-    # Execute step
-    result = execute_step(step)
+    result = step.handler({})
 except Exception as e:
-    recovery = handler.handle_step_failure(step, e, retry_policy=policy)
-    if recovery.should_retry:
-        print(f"Retrying after {recovery.retry_delay}s")
+    recovery = handler.handle_step_failure(step, e)
+    if recovery["retry"]:
+        print(recovery["retry_delay"])  # seconds
 ```
 
 ### Retry Strategies
@@ -345,80 +311,41 @@ fixed_policy = RetryPolicy(
 from semantica.pipeline import FailureHandler, ErrorSeverity
 
 handler = FailureHandler()
+classification = handler.classify_error(Exception("Connection timeout"))
 
-# Classify error
-error = Exception("Connection timeout")
-classification = handler.classify_error(error)
-
-print(f"Severity: {classification['severity']}")
-print(f"Category: {classification['category']}")
-print(f"Retryable: {classification['retryable']}")
-
-# Check severity
-if classification['severity'] == ErrorSeverity.CRITICAL:
-    print("Critical error - immediate attention required")
+print(classification["severity"])   # ErrorSeverity
+print(classification["error_type"]) # str
+print(classification["message"])    # str
 ```
 
 ### Fallback Handlers
 
 ```python
-from semantica.pipeline import FailureHandler, FallbackHandler
+from semantica.pipeline import FallbackHandler
 
-handler = FailureHandler()
-
-# Define fallback function
-def fallback_function(step, error):
-    print(f"Fallback for step {step.name}: {error}")
-    return {"status": "fallback_executed"}
-
-# Register fallback handler
-fallback = FallbackHandler(fallback_function)
-handler.register_fallback("step_type", fallback)
-
-# Handle failure with fallback
-recovery = handler.handle_step_failure(step, error)
-if recovery.recovery_action == "fallback":
-    print("Fallback handler executed")
+fallback = FallbackHandler()
+fallback.set_fallback_strategy("retry")
+strategy = fallback.handle_service_failure("vector_store")
 ```
 
 ### Error Recovery
 
 ```python
-from semantica.pipeline import FailureHandler, ErrorRecovery
+from semantica.pipeline import ErrorRecovery
 
-handler = FailureHandler()
-
-# Handle error with recovery
-error = Exception("Temporary failure")
-recovery = handler.handle_step_failure(step, error)
-
-if recovery.should_retry:
-    print(f"Retrying in {recovery.retry_delay} seconds")
-    time.sleep(recovery.retry_delay)
-    # Retry step...
-else:
-    print(f"Recovery action: {recovery.recovery_action}")
+recovery = ErrorRecovery()
+result = recovery.recover_from_error(Exception("Temporary failure"), {"step": "s1"})
+print(result["recovery_action"])  
 ```
 
 ### Custom Retry Policies
 
 ```python
-from semantica.pipeline import FailureHandler, RetryPolicy
+from semantica.pipeline import FailureHandler, RetryPolicy, RetryStrategy
 
 handler = FailureHandler()
-
-# Register custom retry policy for specific step type
-custom_policy = RetryPolicy(
-    max_retries=5,
-    strategy=RetryStrategy.EXPONENTIAL,
-    initial_delay=0.5,
-    backoff_factor=1.5,
-    retryable_errors=[ConnectionError, TimeoutError]
-)
-
-handler.register_retry_policy("network_step", custom_policy)
-
-# Policy will be used automatically for network_step failures
+custom_policy = RetryPolicy(max_retries=5, strategy=RetryStrategy.EXPONENTIAL, initial_delay=0.5)
+handler.set_retry_policy("network_step", custom_policy)
 ```
 
 ## Parallel Execution
@@ -530,25 +457,13 @@ results = manager.execute_parallel(tasks)
 ### Basic Resource Allocation
 
 ```python
-from semantica.pipeline import ResourceScheduler, ResourceType
+from semantica.pipeline import ResourceScheduler
 
-# Create resource scheduler
 scheduler = ResourceScheduler()
-
-# Register resources
-scheduler.register_resource("cpu1", ResourceType.CPU, capacity=100.0)
-scheduler.register_resource("gpu1", ResourceType.GPU, capacity=1.0)
-scheduler.register_resource("memory1", ResourceType.MEMORY, capacity=16.0)
-
-# Allocate resources
-allocation = scheduler.allocate_resource(
-    "cpu1",
-    ResourceType.CPU,
-    amount=50.0,
-    pipeline_id="pipeline1"
-)
-
-print(f"Allocated: {allocation.amount} CPU units")
+cpu = scheduler.allocate_cpu(cores=2, pipeline_id="p1")
+mem = scheduler.allocate_memory(memory_gb=1.0, pipeline_id="p1")
+usage = scheduler.get_resource_usage()
+scheduler.release_resources({cpu.allocation_id: cpu, mem.allocation_id: mem})
 ```
 
 ### Resource Types
@@ -557,21 +472,9 @@ print(f"Allocated: {allocation.amount} CPU units")
 from semantica.pipeline import ResourceScheduler, ResourceType
 
 scheduler = ResourceScheduler()
-
-# CPU resources
-cpu_allocation = scheduler.allocate_resource("cpu", ResourceType.CPU, 4.0, "pipeline1")
-
-# GPU resources
-gpu_allocation = scheduler.allocate_resource("gpu", ResourceType.GPU, 1.0, "pipeline1")
-
-# Memory resources
-memory_allocation = scheduler.allocate_resource("memory", ResourceType.MEMORY, 8.0, "pipeline1")
-
-# Disk resources
-disk_allocation = scheduler.allocate_resource("disk", ResourceType.DISK, 100.0, "pipeline1")
-
-# Network resources
-network_allocation = scheduler.allocate_resource("network", ResourceType.NETWORK, 1000.0, "pipeline1")
+cpu = scheduler.allocate_cpu(2, "p1")
+gpu = scheduler.allocate_gpu(0, "p1")
+mem = scheduler.allocate_memory(8.0, "p1")
 ```
 
 ### Resource Monitoring
@@ -580,19 +483,10 @@ network_allocation = scheduler.allocate_resource("network", ResourceType.NETWORK
 from semantica.pipeline import ResourceScheduler
 
 scheduler = ResourceScheduler()
-
-# Allocate resource
-allocation = scheduler.allocate_resource("cpu", ResourceType.CPU, 50.0, "pipeline1")
-
-# Check resource status
-status = scheduler.get_resource_status("cpu")
-print(f"Capacity: {status['capacity']}")
-print(f"Allocated: {status['allocated']}")
-print(f"Available: {status['available']}")
-
-# Get all allocations for pipeline
-allocations = scheduler.get_pipeline_allocations("pipeline1")
-print(f"Pipeline allocations: {len(allocations)}")
+usage = scheduler.get_resource_usage()
+print(usage["cpu"]["capacity"])
+print(usage["cpu"]["allocated"])
+print(usage["cpu"]["available"])
 ```
 
 ### Resource Deallocation
@@ -601,30 +495,17 @@ print(f"Pipeline allocations: {len(allocations)}")
 from semantica.pipeline import ResourceScheduler
 
 scheduler = ResourceScheduler()
-
-# Allocate resource
-allocation = scheduler.allocate_resource("cpu", ResourceType.CPU, 50.0, "pipeline1")
-
-# Deallocate resource
-scheduler.deallocate_resource(allocation.allocation_id)
-
-# Verify deallocation
-status = scheduler.get_resource_status("cpu")
-print(f"Available after deallocation: {status['available']}")
+cpu = scheduler.allocate_cpu(2, "p1")
+scheduler.release_resources({cpu.allocation_id: cpu})
 ```
 
 ### Automatic Resource Management
 
 ```python
-from semantica.pipeline import ExecutionEngine, ResourceScheduler
+from semantica.pipeline import ExecutionEngine
 
-# Execution engine automatically manages resources
 engine = ExecutionEngine()
-
-# Resources are allocated before execution and deallocated after
-result = engine.execute_pipeline(pipeline)
-
-# Resources are automatically cleaned up
+result = engine.execute_pipeline(pipeline, cpu_cores=2, memory_gb=1.0)
 ```
 
 ## Pipeline Validation
@@ -693,15 +574,8 @@ else:
 from semantica.pipeline import PipelineValidator
 
 validator = PipelineValidator()
-
-# Validate with performance checks
-result = validator.validate_pipeline(pipeline, check_performance=True)
-
-# Access performance metrics
-if "performance" in result.metadata:
-    perf = result.metadata["performance"]
-    print(f"Estimated execution time: {perf.get('estimated_time', 0)}s")
-    print(f"Resource requirements: {perf.get('resources', {})}")
+perf = validator.validate_performance(pipeline)
+print(perf["warnings"])  
 ```
 
 ## Pipeline Templates
@@ -977,58 +851,58 @@ Error classification uses pattern matching and exception type analysis.
 
 #### PipelineBuilder Methods
 
-- `add_step(name, step_type, **config)`: Add step to pipeline
+- `add_step(step_name, step_type, **config)`: Add step to pipeline
 - `connect_steps(from_step, to_step, **options)`: Connect two steps
-- `build(name, **metadata)`: Build pipeline from steps
-- `serialize(pipeline, format)`: Serialize pipeline to JSON/YAML
-- `deserialize(data, format)`: Deserialize pipeline from JSON/YAML
 - `set_parallelism(level)`: Set parallelism level
-- `validate()`: Validate pipeline structure
+- `build(name="default_pipeline")`: Build pipeline from steps
+- `build_pipeline(pipeline_config, **options)`: Build pipeline from dict
+- `register_step_handler(step_type, handler)`: Register handler
+- `get_step(step_name)`: Get step by name
+- `serialize(format="json")`: Serialize builder state
+- `validate_pipeline()`: Validate pipeline structure
 
 #### ExecutionEngine Methods
 
-- `execute_pipeline(pipeline, data, **options)`: Execute pipeline
-- `execute_pipeline_async(pipeline, data, **options)`: Execute asynchronously
-- `get_status(pipeline_id)`: Get pipeline execution status
+- `execute_pipeline(pipeline, data=None, **options)`: Execute pipeline
+- `get_pipeline_status(pipeline_id)`: Get pipeline execution status
 - `get_progress(pipeline_id)`: Get execution progress
 - `pause_pipeline(pipeline_id)`: Pause pipeline execution
 - `resume_pipeline(pipeline_id)`: Resume pipeline execution
 - `stop_pipeline(pipeline_id)`: Stop pipeline execution
-- `cancel_pipeline(pipeline_id)`: Cancel pipeline execution
 
 #### FailureHandler Methods
 
 - `handle_step_failure(step, error, **options)`: Handle step failure
 - `classify_error(error)`: Classify error severity and type
+- `set_retry_policy(step_type, policy)`: Set retry policy for step type
 - `get_retry_policy(step_type)`: Get retry policy for step type
-- `register_retry_policy(step_type, policy)`: Register custom retry policy
-- `register_fallback(step_type, fallback_handler)`: Register fallback handler
-- `should_retry(error, attempt, policy)`: Determine if should retry
+- `retry_failed_step(step, error, **options)`: Retry failed step
+- `get_error_history(step_name=None)`: Get error history
+- `clear_error_history()`: Clear error history
 
 #### ParallelismManager Methods
 
 - `execute_parallel(tasks, **options)`: Execute tasks in parallel
-- `execute_with_threads(tasks, **options)`: Execute using threads
-- `execute_with_processes(tasks, **options)`: Execute using processes
-- `get_worker_count()`: Get current worker count
-- `set_max_workers(count)`: Set maximum worker count
+- `execute_pipeline_steps_parallel(steps, data, **options)`: Execute pipeline steps in parallel
+- `identify_parallelizable_steps(pipeline)`: Identify parallelizable groups
+- `optimize_parallel_execution(pipeline, available_workers)`: Optimize plan
 
 #### ResourceScheduler Methods
 
-- `register_resource(resource_id, resource_type, capacity)`: Register resource
-- `allocate_resource(resource_id, resource_type, amount, pipeline_id)`: Allocate resource
-- `deallocate_resource(allocation_id)`: Deallocate resource
-- `get_resource_status(resource_id)`: Get resource status
-- `get_pipeline_allocations(pipeline_id)`: Get all allocations for pipeline
-- `get_available_capacity(resource_id)`: Get available capacity
+- `allocate_resources(pipeline, **options)`: Allocate CPU/memory/GPU
+- `allocate_cpu(cores, pipeline_id, step_name=None)`: Allocate CPU cores
+- `allocate_memory(memory_gb, pipeline_id, step_name=None)`: Allocate memory
+- `allocate_gpu(device_id, pipeline_id, step_name=None)`: Allocate GPU device
+- `release_resources(allocations)`: Release allocations
+- `get_resource_usage()`: Current resource usage
+- `optimize_resource_allocation(pipeline, **options)`: Recommendations
 
 #### PipelineValidator Methods
 
-- `validate_pipeline(pipeline, **options)`: Validate entire pipeline
-- `validate_dependencies(pipeline)`: Validate step dependencies
-- `detect_cycles(pipeline)`: Detect circular dependencies
-- `validate_structure(pipeline)`: Validate pipeline structure
-- `estimate_performance(pipeline)`: Estimate execution performance
+- `validate_pipeline(pipeline_or_builder, **options)`: Validate entire pipeline
+- `check_dependencies(pipeline_or_builder)`: Validate dependencies
+- `validate_step(step, **constraints)`: Validate a step
+- `validate_performance(pipeline, **options)`: Estimate performance
 
 #### PipelineTemplateManager Methods
 
@@ -1066,25 +940,9 @@ export PIPELINE_PARALLELISM_LEVEL=2
 ```python
 from semantica.pipeline import ExecutionEngine, FailureHandler, ParallelismManager
 
-# Configure execution engine
-engine = ExecutionEngine(
-    max_workers=4,
-    retry_on_failure=True,
-    default_max_retries=3
-)
-
-# Configure failure handler
-failure_handler = FailureHandler(
-    default_max_retries=3,
-    default_backoff_factor=2.0,
-    default_initial_delay=1.0
-)
-
-# Configure parallelism manager
-parallelism_manager = ParallelismManager(
-    max_workers=4,
-    use_processes=False
-)
+engine = ExecutionEngine(max_workers=4)
+failure_handler = FailureHandler(default_max_retries=3, default_backoff_factor=2.0)
+parallelism_manager = ParallelismManager(max_workers=4, use_processes=False)
 ```
 
 ### Configuration File (YAML)
@@ -1115,28 +973,18 @@ pipeline_templates:
     parallelism: 4
 ```
 
-```python
-from semantica.pipeline.config import PipelineConfig
 
-# Load from config file
-config = PipelineConfig(config_file="config.yaml")
-
-# Access configuration
-max_workers = config.get("max_workers", default=4)
-retry_on_failure = config.get("retry_on_failure", default=True)
-```
 
 ## Advanced Examples
 
 ### Complete Document Processing Pipeline
 
 ```python
-from semantica.pipeline import PipelineBuilder, ExecutionEngine, FailureHandler
+from semantica.pipeline import PipelineBuilder, ExecutionEngine
 
 # Create components
 builder = PipelineBuilder()
 engine = ExecutionEngine(max_workers=4)
-failure_handler = FailureHandler()
 
 # Build complete pipeline
 pipeline = builder.add_step(
@@ -1208,9 +1056,8 @@ retry_policy = RetryPolicy(
     retryable_errors=[ConnectionError, TimeoutError]
 )
 
-# Create failure handler with retry policy
 failure_handler = FailureHandler()
-failure_handler.register_retry_policy("network_step", retry_policy)
+failure_handler.set_retry_policy("network_step", retry_policy)
 
 # Build pipeline
 builder = PipelineBuilder()
@@ -1224,8 +1071,7 @@ pipeline = builder.add_step(
     dependencies=["fetch_data"]
 ).build()
 
-# Execute with error handling
-engine = ExecutionEngine(failure_handler=failure_handler)
+engine = ExecutionEngine()
 result = engine.execute_pipeline(pipeline)
 ```
 
@@ -1253,31 +1099,15 @@ result = engine.execute_pipeline(pipeline)
 ### Resource-Aware Pipeline Execution
 
 ```python
-from semantica.pipeline import (
-    PipelineBuilder,
-    ExecutionEngine,
-    ResourceScheduler,
-    ResourceType
-)
+from semantica.pipeline import PipelineBuilder, ExecutionEngine
 
-# Create resource scheduler
-scheduler = ResourceScheduler()
-
-# Register resources
-scheduler.register_resource("cpu", ResourceType.CPU, capacity=8.0)
-scheduler.register_resource("memory", ResourceType.MEMORY, capacity=16.0)
-
-# Build pipeline
 builder = PipelineBuilder()
-pipeline = builder.add_step("step1", "cpu_intensive", cpu_cores=4) \
-                  .add_step("step2", "memory_intensive", memory_gb=8) \
+pipeline = builder.add_step("step1", "cpu_intensive") \
+                  .add_step("step2", "memory_intensive", dependencies=["step1"]) \
                   .build()
 
-# Execute with resource management
-engine = ExecutionEngine(resource_scheduler=scheduler)
-result = engine.execute_pipeline(pipeline)
-
-# Resources are automatically allocated and deallocated
+engine = ExecutionEngine()
+result = engine.execute_pipeline(pipeline, cpu_cores=4, memory_gb=8)
 ```
 
 ### Template-Based Pipeline Creation
@@ -1316,39 +1146,23 @@ pipeline = builder.add_step("step1", "type1") \
                   .add_step("step3", "type3") \
                   .build()
 
-# Execute with progress monitoring
 engine = ExecutionEngine()
-
-def progress_callback(progress):
-    print(f"Progress: {progress['percentage']:.1f}%")
-    print(f"Completed: {progress['completed_steps']}/{progress['total_steps']}")
-
-result = engine.execute_pipeline(pipeline, progress_callback=progress_callback)
+result = engine.execute_pipeline(pipeline)
+status = engine.get_pipeline_status(pipeline.name)
+progress = engine.get_progress(pipeline.name)
 ```
 
 ### Pipeline Serialization and Persistence
 
 ```python
-from semantica.pipeline import PipelineBuilder
-import json
+from semantica.pipeline import PipelineBuilder, PipelineSerializer
 
-# Build pipeline
 builder = PipelineBuilder()
 pipeline = builder.add_step("step1", "type1").build()
 
-# Serialize pipeline
-serialized = builder.serialize(pipeline, format="json")
-
-# Save to file
-with open("pipeline.json", "w") as f:
-    json.dump(serialized, f, indent=2)
-
-# Load from file
-with open("pipeline.json", "r") as f:
-    serialized = json.load(f)
-
-# Deserialize pipeline
-pipeline = builder.deserialize(serialized, format="json")
+serializer = PipelineSerializer()
+serialized = serializer.serialize_pipeline(pipeline, format="json")
+restored = serializer.deserialize_pipeline(serialized)
 ```
 
 ### Custom Step Handlers