Update services/pipeline_executor.py

services/pipeline_executor.py

@@ -50,32 +50,21 @@ 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
-
-Pipeline components will be in format:
-{
-  "tool_name": "extract_text",
-  "start_page": 1,
-  "end_page": 5,
-  "params": {}
-}"""
+- At the end, call 'finalize' tool with complete results"""
         
         prompt = ChatPromptTemplate.from_messages([
             ("system", system_instructions),
-            ("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.")
+            ("human", "Execute the pipeline for document: {file_path}"),
+            ("human", "Pipeline to execute: {pipeline_json}"),
+            MessagesPlaceholder(variable_name="agent_scratchpad"),
         ])
         
         agent = create_tool_calling_agent(llm, tools, prompt)
@@ -91,7 +80,7 @@ Pipeline components will be in format:
             "input": f"Execute pipeline: {pipeline['pipeline_name']}",
             "file_path": file_path,
             "pipeline_json": json.dumps(pipeline, indent=2),
-            "session_id": session_id or "unknown"
+            "agent_scratchpad": []  # Add this
         })
         
         return result
@@ -119,15 +108,23 @@ def execute_pipeline_bedrock_streaming(
         
         tools = get_langchain_tools()
         
-        system_instructions = """You are MasterLLM. Execute the pipeline components in ORDER.
-
-For each component, call the tool and wait for results."""
+        # 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"""
         
         prompt = ChatPromptTemplate.from_messages([
             ("system", system_instructions),
-            ("system", "File: {file_path}"),
-            ("system", "Pipeline: {pipeline_json}"),
-            ("human", "Execute the pipeline")
+            ("human", "Execute the pipeline for document: {file_path}"),
+            ("human", "Pipeline to execute: {pipeline_json}"),
+            MessagesPlaceholder(variable_name="agent_scratchpad"),
         ])
         
         agent = create_tool_calling_agent(llm, tools, prompt)
@@ -137,6 +134,7 @@ For each component, call the tool and wait for results."""
             verbose=True,
             max_iterations=15,
             handle_parsing_errors=True,
+            return_intermediate_steps=True,
         )
         
         # Yield initial status
@@ -146,18 +144,27 @@ For each component, call the tool and wait for results."""
             "executor": "bedrock"
         }
         
-        step_count = 0
-        
-        # Stream execution
-        for event in executor.stream({
-            "input": f"Execute: {pipeline['pipeline_name']}",
-            "file_path": file_path,
-            "pipeline_json": json.dumps(pipeline, indent=2)
-        }):
-            if "actions" in event:
-                for action in event.get("actions", []):
+        # 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 += 1
-                    tool = getattr(action, "tool", "unknown")
+                    action = event.get("agent", {})
+                    tool = action.get("tool", "thinking")
+                    
                     yield {
                         "type": "step",
                         "step": step_count,
@@ -165,25 +172,75 @@ For each component, call the tool and wait for results."""
                         "status": "executing",
                         "executor": "bedrock"
                     }
-            
-            elif "steps" in event:
-                for step in event.get("steps", []):
-                    observation = str(getattr(step, "observation", ""))[:500]
+                
+                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]
+                        yield {
+                            "type": "step",
+                            "step": step_count,
+                            "status": "completed",
+                            "observation": observation,
+                            "executor": "bedrock"
+                        }
+                
+                elif "output" in event:
+                    # Final output
                     yield {
-                        "type": "step",
-                        "step": step_count,
-                        "status": "completed",
-                        "observation": observation,
+                        "type": "final",
+                        "data": event.get("output"),
                         "executor": "bedrock"
                     }
+                    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"
+                        }
             
-            elif "output" in event:
-                yield {
-                    "type": "final",
-                    "data": event.get("output"),
-                    "executor": "bedrock"
-                }
-                return
+        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"
+            }
         
     except Exception as e:
         yield {
@@ -192,7 +249,6 @@ For each component, call the tool and wait for results."""
             "executor": "bedrock"
         }
 
-
 # ========================
 # CREWAI EXECUTOR (FALLBACK)
 # ========================