mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
Standardize notebooks to handler-based pipeline API: add explicit step dependencies, use data injection for inputs, remove legacy func/args usage; update supply chain, intelligence, forensics, healthcare examples; refresh pipeline docs.
This commit is contained in:
KaifAhmad1
@@ -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"
|
||||
]
|
||||
},
|
||||
{
|
||||
|
||||
@@ -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"
|
||||
]
|
||||
},
|
||||
{
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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"
|
||||
]
|
||||
},
|
||||
{
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
+197
-99
@@ -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"})
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user