Update services/pipeline_executor.py

services/pipeline_executor.py

@@ -50,21 +50,32 @@ def execute_pipeline_bedrock(
         tools = get_langchain_tools()
         
         system_instructions = """You are MasterLLM, a precise document processing agent.
+
 Execute the provided pipeline components in ORDER. For each component:
 1. Call the corresponding tool with exact parameters
 2. Wait for the result
 3. Move to next component
+
 IMPORTANT:
 - Follow the pipeline order strictly
 - Use the file_path provided for all file-based operations
 - For text-processing tools (summarize, classify, NER, translate), use extracted text from previous steps
-- At the end, call 'finalize' tool with complete results"""
+- At the end, call 'finalize' tool with complete results
+
+Pipeline components will be in format:
+{
+  "tool_name": "extract_text",
+  "start_page": 1,
+  "end_page": 5,
+  "params": {}
+}"""
         
         prompt = ChatPromptTemplate.from_messages([
             ("system", system_instructions),
-            ("human", "Execute the pipeline for document: {file_path}"),
-            ("human", "Pipeline to execute: {pipeline_json}"),
-            MessagesPlaceholder(variable_name="agent_scratchpad"),
+            ("system", "File path: {file_path}"),
+            ("system", "Pipeline to execute: {pipeline_json}"),
+            ("system", "Session ID: {session_id}"),
+            ("human", "Execute the pipeline. Process each component in order and finalize with complete JSON results.")
         ])
         
         agent = create_tool_calling_agent(llm, tools, prompt)
@@ -80,7 +91,7 @@ IMPORTANT:
             "input": f"Execute pipeline: {pipeline['pipeline_name']}",
             "file_path": file_path,
             "pipeline_json": json.dumps(pipeline, indent=2),
-            "agent_scratchpad": []  # Add this
+            "session_id": session_id or "unknown"
         })
         
         return result
@@ -108,23 +119,35 @@ def execute_pipeline_bedrock_streaming(
         
         tools = get_langchain_tools()
         
-        # Fixed prompt template with required placeholders
-        system_instructions = """You are MasterLLM, a precise document processing agent.
-Execute the provided pipeline components in ORDER. For each component:
-1. Call the corresponding tool with exact parameters
-2. Wait for the result
-3. Move to next component
-IMPORTANT:
-- Follow the pipeline order strictly
-- Use the file_path provided for all file-based operations
-- For text-processing tools (summarize, classify, NER, translate), use extracted text from previous steps
-- At the end, call 'finalize' tool with complete results"""
+        # STRONGER prompt that FORCES tool usage
+        system_instructions = """You are MasterLLM, an AI agent that MUST execute document processing tools.
+
+CRITICAL INSTRUCTIONS:
+1. You MUST call the tools provided to you - do NOT just describe what you would do
+2. Execute the pipeline components IN ORDER
+3. For each component in the pipeline, you MUST:
+   - Call the corresponding tool with the specified parameters
+   - Wait for the actual result from the tool
+   - Use the file_path provided for file operations
+   - Store the results to pass to the next component
+4. After ALL components are executed, call the 'finalize' tool with the collected results
+5. DO NOT generate placeholder text like "TEXT_EXTRACTION_RESULT" - call the actual tools!
+
+The pipeline components are structured as:
+{
+  "tool_name": "extract_text",
+  "start_page": 1,
+  "end_page": -1,
+  "params": {}
+}
+
+You must call tools, not generate descriptions. This is mandatory."""
         
         prompt = ChatPromptTemplate.from_messages([
             ("system", system_instructions),
-            ("human", "Execute the pipeline for document: {file_path}"),
-            ("human", "Pipeline to execute: {pipeline_json}"),
-            MessagesPlaceholder(variable_name="agent_scratchpad"),
+            ("system", "File to process: {file_path}"),
+            ("system", "Pipeline configuration: {pipeline_json}"),
+            ("human", "Execute ALL the tools in the pipeline. Call each tool and get real results. Do not describe, actually execute!")
         ])
         
         agent = create_tool_calling_agent(llm, tools, prompt)
@@ -132,9 +155,9 @@ IMPORTANT:
             agent=agent,
             tools=tools,
             verbose=True,
-            max_iterations=15,
+            max_iterations=25,  # Increased for multi-step pipelines
             handle_parsing_errors=True,
-            return_intermediate_steps=True,
+            return_intermediate_steps=True  # Important: get intermediate results
         )
         
         # Yield initial status
@@ -144,103 +167,92 @@ IMPORTANT:
             "executor": "bedrock"
         }
         
-        # Stream execution with proper inputs including agent_scratchpad
-        try:
-            # Initialize inputs
-            inputs = {
-                "input": f"Execute pipeline: {pipeline['pipeline_name']} for file: {file_path}",
-                "file_path": file_path,
-                "pipeline_json": json.dumps(pipeline, indent=2),
-                "agent_scratchpad": [],  # Initialize empty scratchpad
-            }
-            
-            step_count = 0
-            
-            # Stream execution
-            for event in executor.stream(inputs):
-                # Handle different event types
-                if "agent" in event:
-                    # Agent is thinking/acting
+        step_count = 0
+        tool_results = {}
+        has_called_tools = False
+        
+        # Stream execution
+        for event in executor.stream({
+            "input": f"Execute the pipeline '{pipeline['pipeline_name']}' by calling each tool in the components list",
+            "file_path": file_path,
+            "pipeline_json": json.dumps(pipeline, indent=2)
+        }):
+            # Track tool actions
+            if "actions" in event:
+                for action in event.get("actions", []):
                     step_count += 1
-                    action = event.get("agent", {})
-                    tool = action.get("tool", "thinking")
+                    has_called_tools = True
+                    tool = getattr(action, "tool", "unknown")
+                    tool_input = getattr(action, "tool_input", {})
                     
                     yield {
                         "type": "step",
                         "step": step_count,
                         "tool": tool,
                         "status": "executing",
-                        "executor": "bedrock"
+                        "executor": "bedrock",
+                        "input": str(tool_input)[:200]
                     }
-                
-                elif "actions" in event:
-                    # Multiple actions
-                    for action in event.get("actions", []):
-                        step_count += 1
-                        tool = getattr(action, "tool", "unknown")
-                        yield {
-                            "type": "step",
-                            "step": step_count,
-                            "tool": tool,
-                            "status": "executing",
-                            "executor": "bedrock"
-                        }
-                
-                elif "steps" in event:
-                    # Steps completed
-                    for step in event.get("steps", []):
-                        observation = str(getattr(step, "observation", ""))[:500]
+            
+            # Track tool outputs
+            elif "steps" in event:
+                for step in event.get("steps", []):
+                    action = getattr(step, "action", None)
+                    observation = getattr(step, "observation", "")
+                    
+                    if action:
+                        tool_name = getattr(action, "tool", "unknown")
+                        tool_results[tool_name] = observation
+                        
                         yield {
                             "type": "step",
                             "step": step_count,
+                            "tool": tool_name,
                             "status": "completed",
-                            "observation": observation,
+                            "observation": str(observation)[:500],
                             "executor": "bedrock"
                         }
+            
+            # Final output
+            elif "output" in event:
+                output = event.get("output")
                 
-                elif "output" in event:
-                    # Final output
+                # Check if tools were actually called
+                if not has_called_tools:
+                    # Agent didn't call tools, just generated text - this is a failure
                     yield {
-                        "type": "final",
-                        "data": event.get("output"),
-                        "executor": "bedrock"
+                        "type": "error",
+                        "error": "Bedrock agent generated text instead of calling tools. Falling back to CrewAI.",
+                        "executor": "bedrock",
+                        "debug_output": str(output)[:500]
                     }
                     return
                 
-                elif "intermediate_steps" in event:
-                    # Intermediate steps (if enabled)
-                    steps = event.get("intermediate_steps", [])
-                    for i, (action, observation) in enumerate(steps):
-                        yield {
-                            "type": "step",
-                            "step": i + 1,
-                            "tool": getattr(action, "tool", str(action)),
-                            "status": "completed",
-                            "observation": str(observation)[:500],
-                            "executor": "bedrock"
-                        }
-            
-        except Exception as stream_error:
-            # If streaming fails, try non-streaming execution
-            yield {
-                "type": "warning",
-                "message": f"Streaming failed, trying non-streaming: {str(stream_error)}",
-                "executor": "bedrock"
-            }
-            
-            # Fallback to non-streaming execution
-            result = executor.invoke({
-                "input": f"Execute pipeline: {pipeline['pipeline_name']}",
-                "file_path": file_path,
-                "pipeline_json": json.dumps(pipeline, indent=2),
-                "agent_scratchpad": []
-            })
-            
-            yield {
-                "type": "final",
-                "data": result.get("output"),
-                "executor": "bedrock"
-            }
+                # If we have tool results, structure them properly
+                if tool_results:
+                    structured_result = {
+                        "status": "completed",
+                        "components_executed": tool_results,
+                        "summary": {
+                            "total_tools_called": len(tool_results),
+                            "tools": list(tool_results.keys())
+                        },
+                        "final_output": output if isinstance(output, str) else json.dumps(output)
+                    }
+                    
+                    yield {
+                        "type": "final",
+                        "data": structured_result,
+                        "executor": "bedrock"
+                    }
+                else:
+                    # No tool results collected, likely a problem
+                    yield {
+                        "type": "error",
+                        "error": "No tool results collected from Bedrock execution",
+                        "executor": "bedrock"
+                    }
+                return
         
     except Exception as e:
         yield {
@@ -249,6 +261,7 @@ IMPORTANT:
             "executor": "bedrock"
         }
 
+
 # ========================
 # CREWAI EXECUTOR (FALLBACK)
 # ========================