Update app.py

app.py

@@ -16,7 +16,6 @@ from huggingface_hub import InferenceClient
 import io
 import mimetypes
 import base64
-from code_interpreter import CodeInterpreter
 from decision_maker import DecisionMaker, ToolType
 
 # -------------------------
@@ -56,18 +55,18 @@ def tighten(q: str) -> str:
 # -------------------------
 
 def image_qa(image_path: str, prompt: str) -> str:
-    """Query LLaVA model for image-based QA."""
+    """Query MiniGPT-4-V for image-based QA."""
     with open(image_path, "rb") as f:
         data = {"prompt": prompt, "image": f.read()}
         headers = {"Content-Type": "application/octet-stream"}
-    return client.post("llava-hf/llava-v1.6-mistral-7b-hf", data=data, headers=headers)
+    return client.post("Vision-CAIR/MiniGPT4-V", data=data, headers=headers)
 
 def video_label(video_path: str, topk: int = 1) -> str:
-    """Get video classification using VideoMAE."""
+    """Get video classification using VideoMAE-Base-Short."""
     with open(video_path, "rb") as f:
         headers = {"Content-Type": "application/octet-stream"}
         preds = client.post(
-            "MCG-NJU/videomae-base-finetuned-ucf101", 
+            "MCG-NJU/videomae-base-short-finetuned-ucf101", 
             data=f.read(),
             headers=headers
         )
@@ -107,12 +106,36 @@ class AgentState(TypedDict):
 # BasicAgent implementation
 # -------------------------
 
+def quick_code_stats(src: str) -> Dict[str, Any]:
+    """Lightweight code analysis using AST."""
+    try:
+        tree = ast.parse(src)
+        funcs = [n.name for n in tree.body if isinstance(n, ast.FunctionDef)]
+        classes = [n.name for n in tree.body if isinstance(n, ast.ClassDef)]
+        imports = []
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                imports.extend(n.name for n in node.names)
+            elif isinstance(node, ast.ImportFrom):
+                imports.append(f"{node.module}.{node.names[0].name}")
+        
+        return {
+            "functions": funcs,
+            "classes": classes,
+            "imports": imports,
+            "lines": len(src.splitlines())
+        }
+    except Exception as e:
+        return {
+            "error": str(e),
+            "lines": len(src.splitlines())
+        }
+
 class BasicAgent:
     def __init__(self):
         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()
 
@@ -145,31 +168,31 @@ class BasicAgent:
                 return state
         
         # Check for file attachments in the question
-        if "file_url" in state["question"]:
-            try:
-                # Parse the question to get file URL
-                question_data = json.loads(state["question"])
-                file_url = question_data.get("file_url")
-                if file_url:
-                    # Download the file
-                    file_data = self._download_file(file_url)
-                    # Store in state
-                    state["attachment_data"] = {
-                        "content": file_data,
-                        "type": self._detect_file_type(file_data, file_url)
-                    }
-                    # Set appropriate step based on file type
-                    if state["attachment_data"]["type"] == "video":
-                        state["current_step"] = "video"
-                    elif state["attachment_data"]["type"] == "image":
-                        state["current_step"] = "image"
-                    elif state["attachment_data"]["type"] in ["excel", "csv"]:
-                        state["current_step"] = "sheet"
-                    return state
-            except Exception as e:
-                state["logs"]["file_download_error"] = str(e)
-                state["current_step"] = "answer"
+        try:
+            question_data = json.loads(state["question"])
+            if "file_url" in question_data:
+                file_url = question_data["file_url"]
+                # Download the file
+                file_data = self._download_file(file_url)
+                # Store in state
+                state["attachment_data"] = {
+                    "content": file_data,
+                    "type": self._detect_file_type(file_data, file_url)
+                }
+                # Set appropriate step based on file type
+                if state["attachment_data"]["type"] == "video":
+                    state["current_step"] = "video"
+                elif state["attachment_data"]["type"] == "image":
+                    state["current_step"] = "image"
+                elif state["attachment_data"]["type"] in ["excel", "csv"]:
+                    state["current_step"] = "sheet"
                 return state
+        except (json.JSONDecodeError, KeyError):
+            pass  # Not a JSON question or no file_url
+        except Exception as e:
+            state["logs"]["file_download_error"] = str(e)
+            state["current_step"] = "answer"
+            return state
         
         # Regular text question analysis
         prompt = (
@@ -215,20 +238,13 @@ class BasicAgent:
         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)
+                # Analyze code using the lightweight analyzer
+                analysis = quick_code_stats(block["code"])
                 
                 # Format the results
                 result = {
                     "language": block["language"],
-                    "analysis": analysis,
-                    "suggestions": suggestions
+                    "analysis": analysis
                 }
                 results.append(result)
             
@@ -242,41 +258,12 @@ class BasicAgent:
         return state
 
     def _detect_file_type(self, data: bytes, url: str) -> str:
-        """Detect file type from content and URL."""
-        # Check URL extension first
-        url_lower = url.lower()
-        if url_lower.endswith((".mp4", ".avi", ".mov")):
-            return "video"
-        elif url_lower.endswith((".jpg", ".jpeg", ".png", ".gif")):
-            return "image"
-        elif url_lower.endswith(".xlsx"):
-            return "excel"
-        elif url_lower.endswith(".csv"):
-            return "csv"
-        
-        # If URL check fails, try content-based detection
-        try:
-            # Try to detect image
-            from PIL import Image
-            Image.open(io.BytesIO(data))
-            return "image"
-        except:
-            pass
-        
-        try:
-            # Try to detect Excel
-            pd.read_excel(io.BytesIO(data))
-            return "excel"
-        except:
-            pass
-        
-        try:
-            # Try to detect CSV
-            pd.read_csv(io.BytesIO(data))
-            return "csv"
-        except:
-            pass
-        
+        """Detect file type from URL extension."""
+        ext = url.split(".")[-1].lower()
+        if ext in {"mp4"}: return "video"
+        if ext in {"jpg", "jpeg", "png"}: return "image"
+        if ext in {"xlsx"}: return "excel"
+        if ext in {"csv"}: return "csv"
         return "unknown"
 
     def _image_node(self, state: AgentState) -> AgentState:
@@ -329,7 +316,7 @@ class BasicAgent:
 
     def _perform_search(self, state: AgentState) -> AgentState:
         try:
-            results = simple_search(state["search_query"], max_results=5)
+            results = simple_search(state["search_query"], max_results=10)
             print("\nSearch Results:")
             for i, s in enumerate(results, 1):
                 print(f"[{i}] {s[:120]}…")
@@ -376,16 +363,17 @@ class BasicAgent:
     def _generate_answer(self, state: AgentState) -> AgentState:
         # Collect all relevant tool outputs
         materials = []
-        for item in state["history"]:
-            if item["step"] in ("search", "image", "video", "sheet", "code_analysis"):
+        for step in state["history"]:
+            if step["step"] in {"search", "image", "video", "sheet", "code_analysis"}:
                 # Handle different output formats
-                if item["step"] == "search":
-                    output = item.get("results", [])
+                if step["step"] == "search":
+                    output = step.get("results", [])
                     if isinstance(output, list):
                         output = "\n".join(output)
                 else:
-                    output = item.get("output", "")
-                materials.append(str(output))
+                    output = step.get("output", "")
+                # Format the output as JSON for better readability
+                materials.append(json.dumps(output, indent=2))
         
         # Join all materials with proper formatting
         search_block = "\n".join(materials) if materials else "No artefacts available."
@@ -479,13 +467,9 @@ Think step-by-step. Write ANSWER: <answer> on its own line.
 
     def _download_file(self, url: str) -> bytes:
         """Download a file from a URL."""
-        try:
-            response = requests.get(url, timeout=20)
-            response.raise_for_status()
-            return response.content
-        except Exception as e:
-            print(f"Error downloading file from {url}: {e}")
-            raise
+        r = requests.get(url, timeout=30)
+        r.raise_for_status()
+        return r.content
 
 # ----------------------------------------------------------------------------------
 # Gradio Interface & Submission Routines