code

app.py

@@ -16,6 +16,8 @@ from huggingface_hub import InferenceClient
 import io
 import mimetypes
 import base64
+from code_interpreter import CodeInterpreter
+from decision_maker import DecisionMaker, ToolType
 
 # -------------------------
 # Environment & constants
@@ -94,6 +96,7 @@ class AgentState(TypedDict):
     search_query: Annotated[str, override]
     task_id: Annotated[str, override]
     logs: Annotated[Dict[str, Any], merge_dicts]
+    code_blocks: Annotated[List[Dict[str, str]], list.__add__]
 
 # -------------------------
 # BasicAgent implementation
@@ -104,6 +107,8 @@ class BasicAgent:
         if not OPENAI_API_KEY:
             raise EnvironmentError("OPENAI_API_KEY not set")
         self.llm = OpenAI(api_key=OPENAI_API_KEY)
+        self.code_interpreter = CodeInterpreter()
+        self.decision_maker = DecisionMaker()
         self.workflow = self._build_workflow()
 
     # ---- Low‑level LLM call
@@ -121,6 +126,19 @@ class BasicAgent:
 
     # ---- Workflow nodes
     def _analyze_question(self, state: AgentState) -> AgentState:
+        # First, analyze the request using the decision maker
+        request_analysis = self.decision_maker.analyze_request(state["question"])
+        state["logs"]["request_analysis"] = request_analysis
+        
+        # Check for code-related content
+        if "code" in request_analysis["intent"]:
+            # Extract code blocks from the question
+            code_blocks = self._extract_code_blocks(state["question"])
+            if code_blocks:
+                state["current_step"] = "code_analysis"
+                state["code_blocks"] = code_blocks
+                return state
+        
         # Check for multimodal content
         q = state["question"].lower()
         if "video" in q or q.endswith(".mp4"):
@@ -153,6 +171,53 @@ class BasicAgent:
             state["history"].append({"step": "analyze", "output": decision})
         return state
 
+    def _extract_code_blocks(self, text: str) -> List[Dict[str, str]]:
+        """Extract code blocks from text using markdown-style code blocks."""
+        code_blocks = []
+        pattern = r"```(\w+)?\n(.*?)```"
+        matches = re.finditer(pattern, text, re.DOTALL)
+        
+        for match in matches:
+            language = match.group(1) or "python"
+            code = match.group(2).strip()
+            code_blocks.append({
+                "language": language,
+                "code": code
+            })
+            
+        return code_blocks
+
+    def _code_analysis_node(self, state: AgentState) -> AgentState:
+        """Handle code analysis requests."""
+        try:
+            results = []
+            for block in state["code_blocks"]:
+                # Analyze code using the code interpreter
+                analysis = self.code_interpreter.analyze_code(
+                    block["code"],
+                    language=block["language"]
+                )
+                
+                # Get improvement suggestions
+                suggestions = self.code_interpreter.suggest_improvements(analysis)
+                
+                # Format the results
+                result = {
+                    "language": block["language"],
+                    "analysis": analysis,
+                    "suggestions": suggestions
+                }
+                results.append(result)
+            
+            state["history"].append({"step": "code_analysis", "output": results})
+            state["current_step"] = "answer"
+            
+        except Exception as e:
+            state["logs"]["code_analysis_error"] = str(e)
+            state["current_step"] = "answer"
+            
+        return state
+
     def _image_node(self, state: AgentState) -> AgentState:
         """Handle image-based questions."""
         try:
@@ -284,6 +349,7 @@ Think step-by-step. Write ANSWER: <answer> on its own line.
         sg.add_node("image", self._image_node)
         sg.add_node("video", self._video_node)
         sg.add_node("sheet", self._sheet_node)
+        sg.add_node("code_analysis", self._code_analysis_node)
 
         # Add edges
         sg.add_edge("analyze", "search")
@@ -292,6 +358,7 @@ Think step-by-step. Write ANSWER: <answer> on its own line.
         sg.add_edge("image", "answer")
         sg.add_edge("video", "answer")
         sg.add_edge("sheet", "answer")
+        sg.add_edge("code_analysis", "answer")
 
         def router(state: AgentState):
             return state["current_step"]
@@ -301,7 +368,8 @@ Think step-by-step. Write ANSWER: <answer> on its own line.
             "answer": "answer",
             "image": "image",
             "video": "video",
-            "sheet": "sheet"
+            "sheet": "sheet",
+            "code_analysis": "code_analysis"
         })
         sg.add_conditional_edges("recheck", router, {
             "search": "search", 
@@ -323,6 +391,7 @@ Think step-by-step. Write ANSWER: <answer> on its own line.
             "search_query": "",
             "task_id": task_id,
             "logs": {},
+            "code_blocks": [],
         }
         final_state = self.workflow.invoke(state)
         return final_state["final_answer"]
@@ -409,6 +478,7 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
                 "search_query": "",
                 "task_id": task_id,
                 "logs": {},
+                "code_blocks": [],
             }
             
             # Run the workflow

code_interpreter.py

@@ -0,0 +1,135 @@
+import ast
+from typing import Dict, List, Any, Optional
+import re
+
+class CodeInterpreter:
+    def __init__(self):
+        self.language = 'python'  # Only support Python
+        
+    def analyze_code(self, code: str) -> Dict[str, Any]:
+        """
+        Analyze Python code and extract key information about its structure and functionality.
+        """
+        try:
+            return self._analyze_python_code(code)
+        except Exception as e:
+            return {"error": f"Python code analysis failed: {str(e)}"}
+    
+    def _analyze_python_code(self, code: str) -> Dict[str, Any]:
+        """
+        Analyze Python code using AST.
+        """
+        try:
+            tree = ast.parse(code)
+            analysis = {
+                "imports": [],
+                "functions": [],
+                "classes": [],
+                "variables": [],
+                "complexity": 0,
+                "docstrings": [],
+                "decorators": []
+            }
+            
+            for node in ast.walk(tree):
+                if isinstance(node, ast.Import):
+                    for name in node.names:
+                        analysis["imports"].append(name.name)
+                elif isinstance(node, ast.ImportFrom):
+                    analysis["imports"].append(f"{node.module}.{node.names[0].name}")
+                elif isinstance(node, ast.FunctionDef):
+                    func_info = {
+                        "name": node.name,
+                        "args": [arg.arg for arg in node.args.args],
+                        "returns": self._get_return_type(node),
+                        "complexity": self._calculate_complexity(node),
+                        "docstring": ast.get_docstring(node),
+                        "decorators": [d.id for d in node.decorator_list if isinstance(d, ast.Name)]
+                    }
+                    analysis["functions"].append(func_info)
+                elif isinstance(node, ast.ClassDef):
+                    class_info = {
+                        "name": node.name,
+                        "methods": [],
+                        "bases": [base.id for base in node.bases if isinstance(base, ast.Name)],
+                        "docstring": ast.get_docstring(node),
+                        "decorators": [d.id for d in node.decorator_list if isinstance(d, ast.Name)]
+                    }
+                    for item in node.body:
+                        if isinstance(item, ast.FunctionDef):
+                            class_info["methods"].append(item.name)
+                    analysis["classes"].append(class_info)
+                elif isinstance(node, ast.Assign):
+                    for target in node.targets:
+                        if isinstance(target, ast.Name):
+                            analysis["variables"].append(target.id)
+                elif isinstance(node, ast.Expr) and isinstance(node.value, ast.Str):
+                    analysis["docstrings"].append(node.value.s)
+            
+            analysis["complexity"] = sum(func["complexity"] for func in analysis["functions"])
+            return analysis
+            
+        except Exception as e:
+            return {"error": f"Python code analysis failed: {str(e)}"}
+    
+    def _get_return_type(self, node: ast.FunctionDef) -> Optional[str]:
+        """Extract return type annotation if present."""
+        if node.returns:
+            if isinstance(node.returns, ast.Name):
+                return node.returns.id
+            elif isinstance(node.returns, ast.Subscript):
+                return f"{node.returns.value.id}[{node.returns.slice.value.id}]"
+        return None
+    
+    def _calculate_complexity(self, node: ast.AST) -> int:
+        """Calculate cyclomatic complexity of a function."""
+        complexity = 1
+        for child in ast.walk(node):
+            if isinstance(child, (ast.If, ast.While, ast.For, ast.Try, ast.ExceptHandler)):
+                complexity += 1
+        return complexity
+    
+    def suggest_improvements(self, analysis: Dict[str, Any]) -> List[str]:
+        """
+        Suggest code improvements based on analysis.
+        """
+        suggestions = []
+        
+        # Check function complexity
+        for func in analysis.get("functions", []):
+            if func["complexity"] > 10:
+                suggestions.append(f"Function '{func['name']}' is too complex (complexity: {func['complexity']}). Consider breaking it down into smaller functions.")
+        
+        # Check for missing type hints
+        for func in analysis.get("functions", []):
+            if not func["returns"]:
+                suggestions.append(f"Function '{func['name']}' is missing return type annotation.")
+        
+        # Check for missing docstrings
+        for func in analysis.get("functions", []):
+            if not func["docstring"]:
+                suggestions.append(f"Function '{func['name']}' is missing a docstring.")
+        
+        # Check for unused imports
+        if len(analysis.get("imports", [])) > 10:
+            suggestions.append("Consider removing unused imports to improve code clarity.")
+        
+        # Check for long functions
+        for func in analysis.get("functions", []):
+            if len(func["args"]) > 5:
+                suggestions.append(f"Function '{func['name']}' has too many parameters ({len(func['args'])}). Consider using a data class or dictionary.")
+        
+        return suggestions
+    
+    def extract_code_context(self, code: str, line_number: int) -> Dict[str, Any]:
+        """
+        Extract context around a specific line of code.
+        """
+        lines = code.split('\n')
+        context = {
+            "line": lines[line_number - 1] if 0 <= line_number - 1 < len(lines) else "",
+            "before": lines[max(0, line_number - 3):line_number - 1],
+            "after": lines[line_number:min(len(lines), line_number + 3)],
+            "indentation": len(re.match(r'^\s*', lines[line_number - 1]).group()) if 0 <= line_number - 1 < len(lines) else 0
+        }
+        return context 

decision_maker.py

@@ -0,0 +1,162 @@
+from typing import Dict, List, Any, Optional
+from dataclasses import dataclass
+from enum import Enum
+import json
+import time
+
+class ToolType(Enum):
+    SEARCH = "search"
+    CODE_ANALYSIS = "code_analysis"
+    FILE_OPERATION = "file_operation"
+    UNKNOWN = "unknown"
+
+@dataclass
+class Tool:
+    name: str
+    type: ToolType
+    description: str
+    required_params: List[str]
+    optional_params: List[str]
+    confidence_threshold: float = 0.7
+
+class DecisionMaker:
+    def __init__(self):
+        self.tools = self._initialize_tools()
+        self.decision_history = []
+        
+    def _initialize_tools(self) -> Dict[str, Tool]:
+        """Initialize available tools with their metadata."""
+        return {
+            "simple_search": Tool(
+                name="simple_search",
+                type=ToolType.SEARCH,
+                description="Perform web search using DuckDuckGo",
+                required_params=["query"],
+                optional_params=["max_results"],
+                confidence_threshold=0.6
+            ),
+            "code_analysis": Tool(
+                name="code_analysis",
+                type=ToolType.CODE_ANALYSIS,
+                description="Analyze Python code structure and provide insights",
+                required_params=["code"],
+                optional_params=[],
+                confidence_threshold=0.8
+            ),
+            "file_operation": Tool(
+                name="file_operation",
+                type=ToolType.FILE_OPERATION,
+                description="Perform file operations like read/write",
+                required_params=["path"],
+                optional_params=["mode"],
+                confidence_threshold=0.9
+            )
+        }
+    
+    def analyze_request(self, request: str) -> Dict[str, Any]:
+        """
+        Analyze the user request to determine the best course of action.
+        """
+        analysis = {
+            "intent": self._detect_intent(request),
+            "required_tools": [],
+            "confidence": 0.0,
+            "suggested_actions": []
+        }
+        
+        # Determine required tools based on intent
+        if "search" in analysis["intent"]:
+            analysis["required_tools"].append(self.tools["simple_search"])
+        if "code" in analysis["intent"]:
+            analysis["required_tools"].append(self.tools["code_analysis"])
+        if "file" in analysis["intent"]:
+            analysis["required_tools"].append(self.tools["file_operation"])
+            
+        # Calculate confidence based on tool requirements
+        if analysis["required_tools"]:
+            analysis["confidence"] = min(tool.confidence_threshold for tool in analysis["required_tools"])
+            
+        # Generate suggested actions
+        analysis["suggested_actions"] = self._generate_actions(analysis)
+        
+        return analysis
+    
+    def _detect_intent(self, request: str) -> List[str]:
+        """Detect the intent(s) from the user request."""
+        intents = []
+        
+        # Python-specific keyword-based intent detection
+        keywords = {
+            "search": ["search", "find", "look up", "query"],
+            "code": ["python", "code", "function", "class", "analyze", "def", "import", "from"],
+            "file": ["file", "read", "write", "save", "load", ".py"]
+        }
+        
+        request_lower = request.lower()
+        for intent, words in keywords.items():
+            if any(word in request_lower for word in words):
+                intents.append(intent)
+                
+        return intents if intents else ["unknown"]
+    
+    def _generate_actions(self, analysis: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """Generate suggested actions based on the analysis."""
+        actions = []
+        
+        for tool in analysis["required_tools"]:
+            action = {
+                "tool": tool.name,
+                "type": tool.type.value,
+                "confidence": tool.confidence_threshold,
+                "required_params": tool.required_params,
+                "optional_params": tool.optional_params
+            }
+            actions.append(action)
+            
+        return actions
+    
+    def validate_tool_usage(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Validate if a tool can be used with the given parameters.
+        """
+        if tool_name not in self.tools:
+            return {
+                "valid": False,
+                "error": f"Unknown tool: {tool_name}"
+            }
+            
+        tool = self.tools[tool_name]
+        validation = {
+            "valid": True,
+            "missing_params": [],
+            "extra_params": []
+        }
+        
+        # Check required parameters
+        for param in tool.required_params:
+            if param not in params:
+                validation["valid"] = False
+                validation["missing_params"].append(param)
+                
+        # Check for extra parameters
+        for param in params:
+            if param not in tool.required_params and param not in tool.optional_params:
+                validation["extra_params"].append(param)
+                
+        return validation
+    
+    def log_decision(self, request: str, analysis: Dict[str, Any], outcome: Dict[str, Any]):
+        """
+        Log a decision made by the system for future reference.
+        """
+        decision = {
+            "timestamp": time.time(),
+            "request": request,
+            "analysis": analysis,
+            "outcome": outcome
+        }
+        self.decision_history.append(decision)
+        
+    def get_decision_history(self) -> List[Dict[str, Any]]:
+        """Get the history of decisions made."""
+        return self.decision_history