Update services/pipeline_executor.py

services/pipeline_executor.py

@@ -536,6 +536,8 @@ def get_tool_executor(tool_name: str) -> Optional[Any]:
 # UNIFIED EXECUTOR WITH FALLBACK (UPDATED)
 # ========================
 
+# Update the execute_pipeline_streaming function:
+
 def execute_pipeline_streaming(
     pipeline: Dict[str, Any],
     file_path: str,
@@ -544,6 +546,7 @@ def execute_pipeline_streaming(
 ) -> Generator[Dict[str, Any], None, None]:
     """
     Execute pipeline with fallback mechanism using master_tools.
+    FIXED: Handle both 'components' and 'pipeline_steps' formats
     """
     components_executed = []
     final_output = None
@@ -554,13 +557,30 @@ def execute_pipeline_streaming(
     # Initialize pipeline info
     pipeline_id = pipeline.get("pipeline_id")
     pipeline_name = pipeline.get("pipeline_name", "Unnamed Pipeline")
-    steps = pipeline.get("pipeline_steps", [])
     
-    yield {
-        "type": "info",
-        "message": f"Starting pipeline: {pipeline_name}",
-        "executor": "initializing"
-    }
+    # FIX: Get steps from either 'components' or 'pipeline_steps'
+    steps = []
+    if "pipeline_steps" in pipeline:
+        steps = pipeline.get("pipeline_steps", [])
+    elif "components" in pipeline:
+        steps = pipeline.get("components", [])
+        # Also update the pipeline to have both for consistency
+        pipeline["pipeline_steps"] = steps
+    
+    if not steps:
+        error_msg = f"No steps/components found in pipeline: {pipeline_name}"
+        yield {
+            "type": "error",
+            "error": error_msg,
+            "data": {
+                "pipeline_id": pipeline_id,
+                "pipeline_name": pipeline_name,
+                "status": "failed",
+                "components_executed": [],
+                "error": error_msg
+            }
+        }
+        return
     
     # Check if tools are available
     if not TOOL_REGISTRY:
@@ -578,6 +598,15 @@ def execute_pipeline_streaming(
         }
         return
     
+    print(f"🏆 Executing pipeline '{pipeline_name}' with {len(steps)} steps")
+    print(f"   Steps format: {[s.get('tool_name', s.get('tool', 'unknown')) for s in steps]}")
+    
+    yield {
+        "type": "info",
+        "message": f"Starting pipeline: {pipeline_name} with {len(steps)} steps",
+        "executor": "initializing"
+    }
+    
     # Try Bedrock first (priority)
     if prefer_bedrock and BEDROCK_AVAILABLE:
         try:
@@ -588,9 +617,13 @@ def execute_pipeline_streaming(
                 "executor": "bedrock"
             }
             
+            # Global components_executed list for step-by-step execution
+            # NOTE: This needs to be declared as nonlocal or passed to helper functions
+            components_executed = []  # Reset for Bedrock execution
+            
             # Execute step by step with Bedrock
             for step_num, step_def in enumerate(steps, 1):
-                tool_name = step_def.get("tool_name", "unknown")
+                tool_name = step_def.get("tool_name", step_def.get("tool", "unknown"))
                 
                 yield {
                     "type": "step",
@@ -608,14 +641,17 @@ def execute_pipeline_streaming(
                         step_num=step_num,
                         total_steps=len(steps),
                         session_id=session_id,
-                        prefer_bedrock=True
+                        prefer_bedrock=True,
+                        previous_results=components_executed  # Pass previous results
                     )
                     
                     executor_used = "bedrock"
                     
                     # Create component result object
                     component_result = {
-                        **step_def,
+                        "tool_name": tool_name,
+                        "tool": tool_name,  # For compatibility
+                        **step_def,  # Include all step definition fields
                         "result": result.get("output"),
                         "status": "completed",
                         "executor": executor_used,
@@ -641,8 +677,12 @@ def execute_pipeline_streaming(
                     file_path = _update_file_path(file_path, result)
                     
                 except Exception as step_error:
+                    print(f"❌ Step {step_num} failed with Bedrock: {str(step_error)}")
+                    
                     # Create failed component result
                     component_result = {
+                        "tool_name": tool_name,
+                        "tool": tool_name,
                         **step_def,
                         "result": {"error": str(step_error)},
                         "status": "failed",
@@ -704,7 +744,7 @@ def execute_pipeline_streaming(
         
         for step_num in range(start_step, len(steps) + 1):
             step_def = steps[step_num - 1]
-            tool_name = step_def.get("tool_name", "unknown")
+            tool_name = step_def.get("tool_name", step_def.get("tool", "unknown"))
             
             yield {
                 "type": "step",
@@ -722,13 +762,16 @@ def execute_pipeline_streaming(
                     step_num=step_num,
                     total_steps=len(steps),
                     session_id=session_id,
-                    prefer_bedrock=False
+                    prefer_bedrock=False,
+                    previous_results=components_executed
                 )
                 
                 executor_used = "crewai"
                 
                 # Create component result object
                 component_result = {
+                    "tool_name": tool_name,
+                    "tool": tool_name,
                     **step_def,
                     "result": result.get("output"),
                     "status": "completed",
@@ -759,8 +802,12 @@ def execute_pipeline_streaming(
                 file_path = _update_file_path(file_path, result)
                     
             except Exception as step_error:
+                print(f"❌ Step {step_num} failed with CrewAI: {str(step_error)}")
+                
                 # Create failed component result
                 component_result = {
+                    "tool_name": tool_name,
+                    "tool": tool_name,
                     **step_def,
                     "result": {"error": str(step_error)},
                     "status": "failed",
@@ -820,22 +867,30 @@ def _execute_step_with_master_tool(
     step_num: int,
     total_steps: int,
     session_id: Optional[str] = None,
-    prefer_bedrock: bool = True
+    prefer_bedrock: bool = True,
+    previous_results: List[Dict[str, Any]] = None
 ) -> Dict[str, Any]:
     """
     Execute a pipeline step using master_tools.
+    FIXED: Handle step_def with either 'tool_name' or 'tool' field
     """
     import time
     import inspect
     
-    tool_name = step_def.get("tool_name", "unknown")
+    # FIX: Get tool name from either 'tool_name' or 'tool' field
+    tool_name = step_def.get("tool_name", step_def.get("tool", "unknown"))
     start_time = time.time()
     
+    print(f"   🔨 Executing step {step_num}/{total_steps}: {tool_name}")
+    print(f"   Step definition: {step_def}")
+    
     # Get tool from registry
     tool = get_tool_executor(tool_name)
     
     if not tool:
-        raise ValueError(f"Tool '{tool_name}' not found in registry. Available tools: {list(TOOL_REGISTRY.keys())}")
+        error_msg = f"Tool '{tool_name}' not found in registry. Available tools: {list(TOOL_REGISTRY.keys())}"
+        print(f"   ❌ {error_msg}")
+        raise ValueError(error_msg)
     
     # Prepare arguments
     args = {}
@@ -857,18 +912,25 @@ def _execute_step_with_master_tool(
             elif field_name == "session_id" and session_id:
                 args[field_name] = session_id
             # Handle text parameter if not provided but we have previous output
-            elif field_name == "text" and field_name not in step_def and step_num > 1:
+            elif field_name == "text" and field_name not in step_def and previous_results:
                 # Try to get text from previous step's output
-                if components_executed and len(components_executed) >= step_num - 1:
-                    prev_result = components_executed[step_num - 2].get("result")
+                if step_num > 1 and len(previous_results) >= step_num - 1:
+                    prev_result = previous_results[step_num - 2].get("result")
                     if isinstance(prev_result, dict) and "text" in prev_result:
                         args["text"] = prev_result["text"]
+                        print(f"   📝 Using text from previous step: {args['text'][:100]}...")
+                    elif isinstance(prev_result, str):
+                        args["text"] = prev_result
+                        print(f"   📝 Using text from previous step: {args['text'][:100]}...")
         
         try:
             # Execute the tool
+            print(f"   🚀 Invoking tool {tool_name} with args: {args}")
             output = tool.invoke(args)
             execution_time = time.time() - start_time
             
+            print(f"   ✅ Step {step_num} completed in {execution_time:.2f}s")
+            
             return {
                 "output": output,
                 "executor": "bedrock" if prefer_bedrock else "crewai",
@@ -918,17 +980,22 @@ def _execute_step_with_master_tool(
                 elif param_name == "session_id" and session_id:
                     call_args[param_name] = session_id
                 # Handle text parameter
-                elif param_name == "text" and param_name not in step_def and step_num > 1:
+                elif param_name == "text" and param_name not in step_def and previous_results:
                     # Try to get text from previous step
-                    if components_executed and len(components_executed) >= step_num - 1:
-                        prev_result = components_executed[step_num - 2].get("result")
+                    if step_num > 1 and len(previous_results) >= step_num - 1:
+                        prev_result = previous_results[step_num - 2].get("result")
                         if isinstance(prev_result, dict) and "text" in prev_result:
                             call_args["text"] = prev_result["text"]
+                        elif isinstance(prev_result, str):
+                            call_args["text"] = prev_result
             
             # Execute the function
+            print(f"   🚀 Calling function {tool_name} with args: {call_args}")
             output = tool(**call_args)
             execution_time = time.time() - start_time
             
+            print(f"   ✅ Step {step_num} completed in {execution_time:.2f}s")
+            
             return {
                 "output": output,
                 "executor": "bedrock" if prefer_bedrock else "crewai",
@@ -972,14 +1039,21 @@ def _build_final_output(
 ) -> Dict[str, Any]:
     """
     Build final output with components_executed array.
+    FIXED: Handle both component formats
     """
+    # Get steps count from pipeline
+    steps = pipeline.get("pipeline_steps", pipeline.get("components", []))
+    
     # Find the finalize step result if present
     final_result = None
     for component in components_executed:
-        if component.get("tool_name") == "finalize":
+        if component.get("tool_name") == "finalize" or component.get("tool") == "finalize":
             final_result = component.get("result")
             break
     
+    # Count completed steps
+    completed_steps = len([c for c in components_executed if c.get("status") == "completed"])
+    
     final_output = {
         "pipeline_id": pipeline.get("pipeline_id"),
         "pipeline_name": pipeline.get("pipeline_name"),
@@ -987,8 +1061,8 @@ def _build_final_output(
         "components_executed": components_executed,
         "executor": executor_used,
         "summary": f"Pipeline execution {status} with {executor_used}",
-        "total_steps": len(pipeline.get("pipeline_steps", [])),
-        "completed_steps": len([c for c in components_executed if c.get("status") == "completed"]),
+        "total_steps": len(steps),
+        "completed_steps": completed_steps,
         "final_result": final_result
     }
     
@@ -1005,10 +1079,16 @@ def _build_final_output(
                     final_output["text"] = result
                     break
                 elif isinstance(result, dict):
-                    for field in ["text", "summary", "content", "translation"]:
+                    for field in ["text", "summary", "content", "translation", "output"]:
                         if field in result and isinstance(result[field], str):
                             final_output["text"] = result[field]
                             break
+                    # If no text field found but dict has string values
+                    if "text" not in final_output:
+                        for key, value in result.items():
+                            if isinstance(value, str) and len(value) > 10:
+                                final_output["text"] = value
+                                break
     
     return final_output