Upload reasoning_system.py

reasoning_system.py

@@ -0,0 +1,668 @@
+"""
+Reasoning System for GAIA-Ready AI Agent
+
+This module provides advanced reasoning capabilities for the AI agent,
+implementing the ReAct approach (Reasoning + Acting) and supporting
+the Think-Act-Observe workflow.
+"""
+
+import os
+import json
+from typing import List, Dict, Any, Optional, Union, Tuple
+from datetime import datetime
+import traceback
+import re
+
+try:
+    from smolagents import Agent, InferenceClientModel, Tool
+except ImportError:
+    import subprocess
+    subprocess.check_call(["pip", "install", "smolagents"])
+    from smolagents import Agent, InferenceClientModel, Tool
+
+
+class ReasoningSystem:
+    """
+    Advanced reasoning system implementing the ReAct approach
+    and supporting the Think-Act-Observe workflow.
+    """
+    def __init__(self, agent, memory_manager):
+        self.agent = agent
+        self.memory_manager = memory_manager
+        self.max_reasoning_depth = 5
+        self.reasoning_templates = self._load_reasoning_templates()
+    
+    def _load_reasoning_templates(self) -> Dict[str, str]:
+        """Load reasoning templates for different stages of the workflow"""
+        return {
+            "think": """
+# Task Analysis and Planning
+
+## Task
+{query}
+
+## Relevant Context
+{context}
+
+## Analysis
+Let me analyze this task step by step:
+1. What is being asked?
+2. What information do I need?
+3. What challenges might I encounter?
+
+## Plan
+Based on my analysis, here's my plan:
+1. [First step]
+2. [Second step]
+3. [Third step]
+...
+
+## Tools Needed
+To accomplish this task, I'll need:
+- [Tool 1]: For [purpose]
+- [Tool 2]: For [purpose]
+...
+
+## Expected Outcome
+If successful, I expect to:
+[Description of expected outcome]
+""",
+            "act": """
+# Action Execution
+
+## Current Task
+{query}
+
+## Current Plan
+{plan}
+
+## Previous Results
+{previous_results}
+
+## Next Action
+Based on my plan and previous results, I'll now:
+1. Use the [tool name] tool
+2. With parameters: [parameters]
+3. Purpose: [why this action is needed]
+
+## Execution
+[Detailed description of how I'll execute this action]
+""",
+            "observe": """
+# Result Analysis
+
+## Current Task
+{query}
+
+## Action Taken
+{action}
+
+## Results Obtained
+{results}
+
+## Analysis
+Let me analyze these results:
+1. What did I learn?
+2. Does this answer the original question?
+3. Are there any inconsistencies or gaps?
+
+## Next Steps
+Based on my analysis:
+- [Next step recommendation]
+- [Alternative approach if needed]
+
+## Progress Assessment
+Task completion status: [percentage]%
+[Explanation of current progress]
+"""
+        }
+    
+    def think(self, query: str) -> Dict[str, Any]:
+        """
+        Analyze the task and plan an approach (Think phase)
+        
+        Args:
+            query: The user's query or task
+            
+        Returns:
+            Dictionary containing analysis and plan
+        """
+        # Retrieve relevant memories
+        relevant_memories = self.memory_manager.get_relevant_memories(query)
+        
+        # Format context from relevant memories
+        context = ""
+        if relevant_memories:
+            context_items = []
+            for memory in relevant_memories:
+                memory_type = memory.get("type", "general")
+                content = memory.get("content", "")
+                relevance = memory.get("relevance_score", 0)
+                context_items.append(f"- [{memory_type.upper()}] (Relevance: {relevance:.2f}): {content}")
+            context = "\n".join(context_items)
+        else:
+            context = "No relevant prior knowledge found."
+        
+        # Apply the thinking template
+        thinking_template = self.reasoning_templates["think"]
+        thinking_prompt = thinking_template.format(
+            query=query,
+            context=context
+        )
+        
+        # Use the agent to generate a plan
+        try:
+            response = self.agent.chat(thinking_prompt)
+            
+            # Store the thinking in memory
+            self.memory_manager.add_to_short_term({
+                "type": "thinking",
+                "content": response,
+                "timestamp": datetime.now().isoformat()
+            })
+            
+            # Parse the response to extract structured information
+            analysis = self._extract_section(response, "Analysis")
+            plan = self._extract_section(response, "Plan")
+            tools_needed = self._extract_section(response, "Tools Needed")
+            expected_outcome = self._extract_section(response, "Expected Outcome")
+            
+            return {
+                "raw_response": response,
+                "analysis": analysis,
+                "plan": plan,
+                "tools_needed": tools_needed,
+                "expected_outcome": expected_outcome
+            }
+        except Exception as e:
+            error_msg = f"Error during thinking phase: {str(e)}\n{traceback.format_exc()}"
+            print(error_msg)
+            
+            # Store the error in memory
+            self.memory_manager.add_to_short_term({
+                "type": "error",
+                "content": error_msg,
+                "timestamp": datetime.now().isoformat()
+            })
+            
+            # Return a basic plan despite the error
+            return {
+                "raw_response": "Error occurred during thinking phase.",
+                "analysis": "Could not analyze the task due to an error.",
+                "plan": "1. Try a simpler approach\n2. Break down the task into smaller steps",
+                "tools_needed": "web_search: To find basic information",
+                "expected_outcome": "Partial answer to the query"
+            }
+    
+    def act(self, plan: Dict[str, Any], query: str, previous_results: str = "") -> Dict[str, Any]:
+        """
+        Execute actions based on the plan (Act phase)
+        
+        Args:
+            plan: The plan generated by the think step
+            query: The original query
+            previous_results: Results from previous actions
+            
+        Returns:
+            Dictionary containing action details and results
+        """
+        # Apply the action template
+        action_template = self.reasoning_templates["act"]
+        action_prompt = action_template.format(
+            query=query,
+            plan=plan.get("plan", "No plan available."),
+            previous_results=previous_results if previous_results else "No previous results."
+        )
+        
+        try:
+            # Use the agent to determine the next action
+            action_response = self.agent.chat(action_prompt)
+            
+            # Store the action planning in memory
+            self.memory_manager.add_to_short_term({
+                "type": "action_planning",
+                "content": action_response,
+                "timestamp": datetime.now().isoformat()
+            })
+            
+            # Parse the action response to extract tool and parameters
+            tool_info = self._extract_tool_info(action_response)
+            
+            if not tool_info:
+                # If no tool was identified, try a more direct approach
+                direct_prompt = f"""
+Based on the task "{query}" and the plan:
+{plan.get('plan', 'No plan available.')}
+
+Which specific tool should I use next and with what parameters?
+Respond in this format:
+TOOL: [tool name]
+PARAMETERS: [parameter1=value1, parameter2=value2, ...]
+"""
+                direct_response = self.agent.chat(direct_prompt)
+                tool_info = self._extract_tool_info(direct_response)
+            
+            if tool_info:
+                tool_name = tool_info["tool"]
+                tool_params = tool_info["parameters"]
+                
+                # Find the matching tool
+                matching_tool = None
+                for tool in self.agent.tools:
+                    if tool.name == tool_name:
+                        matching_tool = tool
+                        break
+                
+                if matching_tool:
+                    # Execute the tool
+                    try:
+                        if isinstance(tool_params, dict):
+                            result = matching_tool.function(**tool_params)
+                        else:
+                            result = matching_tool.function(tool_params)
+                        
+                        # Store the successful action result in memory
+                        self.memory_manager.add_to_short_term({
+                            "type": "action_result",
+                            "content": f"Tool: {tool_name}\nParameters: {tool_params}\nResult: {result}",
+                            "timestamp": datetime.now().isoformat()
+                        })
+                        
+                        return {
+                            "tool": tool_name,
+                            "parameters": tool_params,
+                            "result": result,
+                            "success": True,
+                            "error": None
+                        }
+                    except Exception as e:
+                        error_msg = f"Error executing tool {tool_name}: {str(e)}\n{traceback.format_exc()}"
+                        print(error_msg)
+                        
+                        # Store the error in memory
+                        self.memory_manager.add_to_short_term({
+                            "type": "error",
+                            "content": error_msg,
+                            "timestamp": datetime.now().isoformat()
+                        })
+                        
+                        return {
+                            "tool": tool_name,
+                            "parameters": tool_params,
+                            "result": f"Error: {str(e)}",
+                            "success": False,
+                            "error": str(e)
+                        }
+                else:
+                    error_msg = f"Tool '{tool_name}' not found."
+                    print(error_msg)
+                    
+                    # Store the error in memory
+                    self.memory_manager.add_to_short_term({
+                        "type": "error",
+                        "content": error_msg,
+                        "timestamp": datetime.now().isoformat()
+                    })
+                    
+                    return {
+                        "tool": tool_name,
+                        "parameters": tool_params,
+                        "result": f"Error: Tool '{tool_name}' not found.",
+                        "success": False,
+                        "error": "Tool not found"
+                    }
+            else:
+                error_msg = "Could not determine which tool to use."
+                print(error_msg)
+                
+                # Store the error in memory
+                self.memory_manager.add_to_short_term({
+                    "type": "error",
+                    "content": error_msg,
+                    "timestamp": datetime.now().isoformat()
+                })
+                
+                # Default to web search as a fallback
+                try:
+                    web_search_tool = None
+                    for tool in self.agent.tools:
+                        if tool.name == "web_search":
+                            web_search_tool = tool
+                            break
+                    
+                    if web_search_tool:
+                        result = web_search_tool.function(query)
+                        return {
+                            "tool": "web_search",
+                            "parameters": query,
+                            "result": result,
+                            "success": True,
+                            "error": None,
+                            "fallback": True
+                        }
+                    else:
+                        return {
+                            "tool": "none",
+                            "parameters": "none",
+                            "result": "Could not determine which tool to use and web_search fallback not available.",
+                            "success": False,
+                            "error": "No tool selected"
+                        }
+                except Exception as e:
+                    return {
+                        "tool": "web_search",
+                        "parameters": query,
+                        "result": f"Error in fallback web search: {str(e)}",
+                        "success": False,
+                        "error": str(e),
+                        "fallback": True
+                    }
+        except Exception as e:
+            error_msg = f"Error during action phase: {str(e)}\n{traceback.format_exc()}"
+            print(error_msg)
+            
+            # Store the error in memory
+            self.memory_manager.add_to_short_term({
+                "type": "error",
+                "content": error_msg,
+                "timestamp": datetime.now().isoformat()
+            })
+            
+            return {
+                "tool": "none",
+                "parameters": "none",
+                "result": f"Error during action planning: {str(e)}",
+                "success": False,
+                "error": str(e)
+            }
+    
+    def observe(self, action_result: Dict[str, Any], plan: Dict[str, Any], query: str) -> Dict[str, Any]:
+        """
+        Analyze the results of actions and determine next steps (Observe phase)
+        
+        Args:
+            action_result: Results from the act step
+            plan: The original plan
+            query: The original query
+            
+        Returns:
+            Dictionary containing observation and next steps
+        """
+        # Apply the observation template
+        observation_template = self.reasoning_templates["observe"]
+        observation_prompt = observation_template.format(
+            query=query,
+            action=f"Tool: {action_result.get('tool', 'none')}\nParameters: {action_result.get('parameters', 'none')}",
+            results=action_result.get('result', 'No results.')
+        )
+        
+        try:
+            # Use the agent to analyze the results
+            observation_response = self.agent.chat(observation_prompt)
+            
+            # Store the observation in memory
+            self.memory_manager.add_to_short_term({
+                "type": "observation",
+                "content": observation_response,
+                "timestamp": datetime.now().isoformat()
+            })
+            
+            # Parse the observation to extract structured information
+            analysis = self._extract_section(observation_response, "Analysis")
+            next_steps = self._extract_section(observation_response, "Next Steps")
+            progress = self._extract_section(observation_response, "Progress Assessment")
+            
+            # Determine if we need to continue with more actions
+            continue_execution = True
+            
+            # Check for completion indicators
+            completion_phrases = [
+                "task complete", "question answered", "fully answered",
+                "100%", "task is complete", "fully resolved"
+            ]
+            
+            if any(phrase in observation_response.lower() for phrase in completion_phrases):
+                continue_execution = False
+                
+                # Store the final answer in long-term memory
+                self.memory_manager.add_to_long_term({
+                    "type": "final_answer",
+                    "query": query,
+                    "content": observation_response,
+                    "timestamp": datetime.now().isoformat(),
+                    "importance": 0.8  # High importance for final answers
+                })
+            
+            return {
+                "raw_response": observation_response,
+                "analysis": analysis,
+                "next_steps": next_steps,
+                "progress": progress,
+                "continue": continue_execution
+            }
+        except Exception as e:
+            error_msg = f"Error during observation phase: {str(e)}\n{traceback.format_exc()}"
+            print(error_msg)
+            
+            # Store the error in memory
+            self.memory_manager.add_to_short_term({
+                "type": "error",
+                "content": error_msg,
+                "timestamp": datetime.now().isoformat()
+            })
+            
+            # Default observation with continuation
+            return {
+                "raw_response": f"Error occurred during observation phase: {str(e)}",
+                "analysis": "Could not analyze the results due to an error.",
+                "next_steps": "Try a different approach or tool.",
+                "progress": "Unknown due to error.",
+                "continue": True  # Continue by default on error
+            }
+    
+    def _extract_section(self, text: str, section_name: str) -> str:
+        """Extract a section from the response text"""
+        pattern = rf"(?:^|\n)(?:#+\s*{re.escape(section_name)}:?|\*\*{re.escape(section_name)}:?\*\*|{re.escape(section_name)}:?)\s*(.*?)(?:\n(?:#+\s*|$)|\Z)"
+        match = re.search(pattern, text, re.DOTALL | re.IGNORECASE)
+        
+        if match:
+            content = match.group(1).strip()
+            return content
+        
+        # Try a more lenient approach if the first one fails
+        pattern = rf"{re.escape(section_name)}:?\s*(.*?)(?:\n\n|\n[A-Z]|\Z)"
+        match = re.search(pattern, text, re.DOTALL | re.IGNORECASE)
+        
+        if match:
+            content = match.group(1).strip()
+            return content
+        
+        return f"No {section_name.lower()} found."
+    
+    def _extract_tool_info(self, text: str) -> Optional[Dict[str, Any]]:
+        """Extract tool name and parameters from the response text"""
+        # Try to find tool name
+        tool_pattern = r"(?:TOOL|Tool|tool):\s*(\w+)"
+        tool_match = re.search(tool_pattern, text)
+        
+        if not tool_match:
+            return None
+        
+        tool_name = tool_match.group(1).strip()
+        
+        # Try to find parameters
+        params_pattern = r"(?:PARAMETERS|Parameters|parameters):\s*(.*?)(?:\n\n|\n[A-Z]|\Z)"
+        params_match = re.search(params_pattern, text, re.DOTALL)
+        
+        if params_match:
+            params_text = params_match.group(1).strip()
+            
+            # Check if parameters are in key=value format
+            if "=" in params_text:
+                # Parse as dictionary
+                params_dict = {}
+                param_pairs = re.findall(r"(\w+)\s*=\s*([^,\n]+)", params_text)
+                
+                for key, value in param_pairs:
+                    params_dict[key.strip()] = value.strip()
+                
+                return {
+                    "tool": tool_name,
+                    "parameters": params_dict
+                }
+            else:
+                # Treat as a single string parameter
+                return {
+                    "tool": tool_name,
+                    "parameters": params_text
+                }
+        else:
+            # No parameters found, use empty dict
+            return {
+                "tool": tool_name,
+                "parameters": {}
+            }
+    
+    def execute_reasoning_cycle(self, query: str, max_iterations: int = 5) -> str:
+        """
+        Execute a complete Think-Act-Observe reasoning cycle
+        
+        Args:
+            query: The user's query or task
+            max_iterations: Maximum number of iterations
+            
+        Returns:
+            Final answer to the query
+        """
+        # Store the query in memory
+        self.memory_manager.add_to_short_term({
+            "type": "query",
+            "content": query,
+            "timestamp": datetime.now().isoformat()
+        })
+        
+        # Initialize the workflow
+        iteration = 0
+        final_answer = None
+        all_results = []
+        
+        while iteration < max_iterations:
+            print(f"Iteration {iteration + 1}/{max_iterations}")
+            
+            # Think
+            print("Thinking...")
+            plan = self.think(query)
+            
+            # Act
+            print("Acting...")
+            previous_results = "\n".join([r.get("result", "") for r in all_results])
+            action_result = self.act(plan, query, previous_results)
+            all_results.append(action_result)
+            
+            # Observe
+            print("Observing...")
+            observation = self.observe(action_result, plan, query)
+            
+            # Check if we have a final answer
+            if not observation["continue"]:
+                # Generate final answer
+                final_answer_prompt = f"""
+TASK: {query}
+
+REASONING PROCESS:
+{plan.get('raw_response', 'No thinking process available.')}
+
+ACTIONS TAKEN:
+{', '.join([f"{r.get('tool', 'unknown')}({r.get('parameters', '')})" for r in all_results])}
+
+RESULTS:
+{previous_results}
+{action_result.get('result', '')}
+
+OBSERVATION:
+{observation.get('raw_response', 'No observation available.')}
+
+Based on all the above, provide a comprehensive final answer to the original task.
+"""
+                final_answer = self.agent.chat(final_answer_prompt)
+                
+                # Store the final answer in long-term memory
+                self.memory_manager.add_to_long_term({
+                    "type": "final_answer",
+                    "query": query,
+                    "content": final_answer,
+                    "timestamp": datetime.now().isoformat(),
+                    "importance": 0.9  # Very high importance
+                })
+                
+                break
+            
+            # Update the query with the observation for the next iteration
+            query = f"""
+Original task: {query}
+
+Progress so far:
+{observation.get('raw_response', 'No observation available.')}
+
+Please continue solving this task.
+"""
+            
+            iteration += 1
+        
+        # If we reached max iterations without a final answer
+        if final_answer is None:
+            final_answer = f"""
+I've spent {max_iterations} iterations trying to solve this task.
+Here's my best answer based on what I've learned:
+
+{observation.get('raw_response', 'No final observation available.')}
+
+Note: This answer may be incomplete as I reached the maximum number of iterations.
+"""
+            
+            # Store the partial answer in long-term memory
+            self.memory_manager.add_to_long_term({
+                "type": "partial_answer",
+                "query": query,
+                "content": final_answer,
+                "timestamp": datetime.now().isoformat(),
+                "importance": 0.6  # Medium importance for partial answers
+            })
+        
+        return final_answer
+
+
+# Example usage
+if __name__ == "__main__":
+    # This would be imported from your agent.py
+    from smolagents import Agent, InferenceClientModel, Tool
+    
+    # Mock agent for testing
+    class MockAgent:
+        def __init__(self):
+            self.tools = [
+                Tool(name="web_search", description="Search the web", function=lambda x: f"Search results for: {x}"),
+                Tool(name="calculator", description="Calculate", function=lambda x: f"Result: {eval(x)}")
+            ]
+        
+        def chat(self, message):
+            return f"Response to: {message[:50]}..."
+    
+    # Mock memory manager
+    class MockMemoryManager:
+        def add_to_short_term(self, item):
+            print(f"Added to short-term: {item['type']}")
+        
+        def add_to_long_term(self, item):
+            print(f"Added to long-term: {item['type']}")
+        
+        def get_relevant_memories(self, query):
+            return []
+    
+    # Test the reasoning system
+    agent = MockAgent()
+    memory_manager = MockMemoryManager()
+    reasoning = ReasoningSystem(agent, memory_manager)
+    
+    result = reasoning.execute_reasoning_cycle("What is 2+2?")
+    print(f"\nFinal result: {result}")