langchain sends the reques in specific format and bedrock doesnt allow it, so the request doesnt reaches mistral and now manual calling

services/pipeline_executor.py

@@ -108,109 +108,148 @@ def execute_pipeline_bedrock_streaming(
     session_id: Optional[str] = None
 ) -> Generator[Dict[str, Any], None, None]:
     """
-    Execute pipeline using Bedrock + LangChain with ReAct agent (works with Mistral)
+    Execute pipeline using Bedrock with MANUAL tool calling loop (bypasses LangChain agents)
     """
     if not BEDROCK_AVAILABLE:
         raise RuntimeError("Bedrock LangChain not available")
     
     try:
-        # Use Mistral (the only model you have access to)
-        llm = ChatBedrock(
-            model_id="mistral.mistral-large-2402-v1:0",
-            region_name=os.getenv("AWS_REGION", "us-east-1"),
-            model_kwargs={
-                "temperature": 0.0,
-                "max_tokens": 4096
-            }
+        import re
+        import boto3
+        
+        # Get Bedrock client directly
+        bedrock_runtime = boto3.client(
+            service_name='bedrock-runtime',
+            region_name=os.getenv("AWS_REGION", "us-east-1")
         )
         
-        tools = get_langchain_tools()
+        tools_dict = {tool.name: tool for tool in get_langchain_tools()}
         
-        # ReAct prompt template - uses text-based reasoning
-        react_prompt = PromptTemplate.from_template("""You are MasterLLM, a document processing assistant that executes tools step-by-step.
+        # Build tool descriptions for prompt
+        tool_descriptions = []
+        for name, tool in tools_dict.items():
+            tool_descriptions.append(f"- {name}: {tool.description}")
+        
+        tools_text = "\n".join(tool_descriptions)
+        tool_names = ", ".join(tools_dict.keys())
+        
+        # Initial prompt
+        system_prompt = f"""You are MasterLLM, a document processing assistant.
 
 You have access to these tools:
-{tools}
-
-Tool names: {tool_names}
+{tools_text}
 
-Use the following format EXACTLY:
-
-Thought: Think about what you need to do
+To use a tool, you MUST write EXACTLY in this format:
 Action: tool_name
 Action Input: {{"param1": "value1", "param2": value2}}
-Observation: [result will appear here]
-... (repeat Thought/Action/Action Input/Observation as needed)
-Thought: I have completed all steps
-Final Answer: [summarize what was done]
 
-CRITICAL RULES:
-1. You MUST use the Action/Action Input format to call tools
-2. Action Input MUST be valid JSON
-3. After Observation, think again and take the next action
-4. Call tools for EACH pipeline component
-5. When done, provide Final Answer
+After you write Action and Action Input, I will execute the tool and give you the Observation.
+Then you can take another Action or provide your Final Answer.
+
+CRITICAL:
+- Write "Action:" followed by the tool name
+- Write "Action Input:" followed by valid JSON on the SAME line or next line
+- After seeing Observation, you can take another Action
+- When done, write "Final Answer:" followed by summary
 
 File path: {file_path}
-Pipeline to execute: {pipeline_json}
+Pipeline components to execute:
+{json.dumps(pipeline.get('components', []), indent=2)}
 
-Begin! Execute each component in the pipeline.
+Execute each component by calling the corresponding tool."""
 
-{agent_scratchpad}""")
+        user_message = f"Execute the pipeline: {pipeline['pipeline_name']}"
         
-        # Create ReAct agent
-        agent = create_react_agent(llm, tools, react_prompt)
-        executor = AgentExecutor(
-            agent=agent,
-            tools=tools,
-            verbose=True,
-            max_iterations=25,
-            handle_parsing_errors=True,
-            return_intermediate_steps=True
-        )
+        conversation_history = []
+        tool_results = {}
+        has_called_tools = False
+        step_count = 0
+        max_iterations = 10
         
-        # Yield initial status
         yield {
             "type": "status",
-            "message": "Initializing Bedrock ReAct executor...",
+            "message": "Initializing Bedrock manual executor...",
             "executor": "bedrock"
         }
         
-        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 running 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
-                    has_called_tools = True
-                    tool = getattr(action, "tool", "unknown")
-                    tool_input = getattr(action, "tool_input", {})
+        for iteration in range(max_iterations):
+            # Prepare messages
+            messages = [{"role": "user", "content": user_message}]
+            messages.extend(conversation_history)
+            
+            # Call Bedrock directly using converse API
+            response = bedrock_runtime.converse(
+                modelId="mistral.mistral-large-2402-v1:0",
+                messages=messages,
+                system=[{"text": system_prompt}],
+                inferenceConfig={
+                    "temperature": 0.0,
+                    "maxTokens": 2048
+                }
+            )
+            
+            # Get response text
+            assistant_message = response['output']['message']['content'][0]['text']
+            print(f"\n🤖 Mistral Response (Iteration {iteration + 1}):\n{assistant_message}\n")
+            
+            # Add to conversation
+            conversation_history.append({"role": "assistant", "content": assistant_message})
+            
+            # Check for Final Answer
+            if "Final Answer:" in assistant_message or "final answer" in assistant_message.lower():
+                # Done!
+                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": assistant_message
+                    }
                     
                     yield {
-                        "type": "step",
-                        "step": step_count,
-                        "tool": tool,
-                        "status": "executing",
-                        "executor": "bedrock",
-                        "input": str(tool_input)[:200]
+                        "type": "final",
+                        "data": structured_result,
+                        "executor": "bedrock"
                     }
+                else:
+                    yield {
+                        "type": "error",
+                        "error": "Bedrock completed but no tools were called",
+                        "executor": "bedrock"
+                    }
+                return
+            
+            # Parse for Action and Action Input
+            action_match = re.search(r'Action:\s*(\w+)', assistant_message)
+            action_input_match = re.search(r'Action Input:\s*(\{.*?\})', assistant_message, re.DOTALL)
             
-            # Track tool outputs
-            elif "steps" in event:
-                for step in event.get("steps", []):
-                    action = getattr(step, "action", None)
-                    observation = getattr(step, "observation", "")
+            if action_match and action_input_match:
+                tool_name = action_match.group(1)
+                action_input_str = action_input_match.group(1)
+                
+                try:
+                    # Parse JSON input
+                    tool_input = json.loads(action_input_str)
                     
-                    if action:
-                        tool_name = getattr(action, "tool", "unknown")
+                    if tool_name in tools_dict:
+                        step_count += 1
+                        has_called_tools = True
+                        
+                        yield {
+                            "type": "step",
+                            "step": step_count,
+                            "tool": tool_name,
+                            "status": "executing",
+                            "executor": "bedrock",
+                            "input": str(tool_input)[:200]
+                        }
+                        
+                        # Execute the tool!
+                        tool = tools_dict[tool_name]
+                        observation = tool.invoke(tool_input)
                         tool_results[tool_name] = observation
                         
                         yield {
@@ -221,23 +260,33 @@ Begin! Execute each component in the pipeline.
                             "observation": str(observation)[:500],
                             "executor": "bedrock"
                         }
-            
-            # Final output
-            elif "output" in event:
-                output = event.get("output")
-                
-                # Check if tools were actually called
-                if not has_called_tools:
+                        
+                        # Add observation to conversation
+                        observation_message = f"Observation: {observation}"
+                        conversation_history.append({"role": "user", "content": observation_message})
+                        
+                    else:
+                        # Unknown tool
+                        error_msg = f"Unknown tool: {tool_name}"
+                        conversation_history.append({"role": "user", "content": f"Error: {error_msg}"})
+                        
+                except json.JSONDecodeError as e:
+                    # Invalid JSON
+                    error_msg = f"Invalid JSON in Action Input: {e}"
+                    conversation_history.append({"role": "user", "content": f"Error: {error_msg}"})
+            else:
+                # No action found - agent might be confused or done
+                if iteration > 0 and not has_called_tools:
+                    # Agent isn't calling tools properly
                     yield {
                         "type": "error",
-                        "error": "Bedrock ReAct agent didn't call tools properly. Falling back to CrewAI.",
+                        "error": "Bedrock didn't call tools in correct format. Falling back to CrewAI.",
                         "executor": "bedrock",
-                        "debug_output": str(output)[:500]
+                        "debug_output": assistant_message[:500]
                     }
                     return
-                
-                # If we have tool results, structure them
-                if tool_results:
+                elif iteration > 0:
+                    # Has called some tools but stopped - might be done
                     structured_result = {
                         "status": "completed",
                         "components_executed": tool_results,
@@ -245,7 +294,7 @@ Begin! Execute each component in the pipeline.
                             "total_tools_called": len(tool_results),
                             "tools": list(tool_results.keys())
                         },
-                        "final_output": output if isinstance(output, str) else json.dumps(output)
+                        "final_output": assistant_message
                     }
                     
                     yield {
@@ -253,13 +302,31 @@ Begin! Execute each component in the pipeline.
                         "data": structured_result,
                         "executor": "bedrock"
                     }
-                else:
-                    yield {
-                        "type": "error",
-                        "error": "No tool results collected from Bedrock execution",
-                        "executor": "bedrock"
-                    }
-                return
+                    return
+        
+        # Max iterations reached
+        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": "Max iterations reached"
+            }
+            
+            yield {
+                "type": "final",
+                "data": structured_result,
+                "executor": "bedrock"
+            }
+        else:
+            yield {
+                "type": "error",
+                "error": "Max iterations reached without tool calls",
+                "executor": "bedrock"
+            }
         
     except Exception as e:
         yield {