Update app.py

app.py

@@ -74,114 +74,91 @@ def indent_code(code: str, indent: str = "    ") -> str:
     return "\n".join(indent + line for line in code.splitlines())
 
 # --- Tool Definitions ---
-@tool 
-def duckduckgo_search(query: str) -> str:
-    """Search web using DuckDuckGo. Returns top 3 results."""
-    print(f"DEBUG: duckduckgo_search called with: {query}")
-    try:
-        with DDGS() as ddgs:
-            return "\n\n".join(
-                f"Title: {res['title']}\nURL: {res['href']}\nSnippet: {res['body']}"
-                for res in ddgs.text(query, max_results=3)
-            )
-    except Exception as e:
-        return f"Error performing DuckDuckGo search: {str(e)}"
-
-@tool 
-def wikipedia_search(query: str) -> str:
-    """Get Wikipedia summaries. Returns first 3 sentences."""
-    print(f"DEBUG: wikipedia_search called with: {query}")
-    try:
-        return wikipedia.summary(query, sentences=3)
-    except wikipedia.DisambiguationError as e:
-        return f"Disambiguation options: {', '.join(e.options[:3])}"
-    except wikipedia.PageError:
-        return "Wikipedia page not found."
-    except Exception as e:
-        return f"Error performing Wikipedia search: {str(e)}"
-
-@tool 
-def arxiv_search(query: str) -> str:
-    """Search academic papers on arXiv. Returns top 3 results."""
-    print(f"DEBUG: arxiv_search called with: {query}")
-    try:
-        results = arxiv.Search(
-            query=query, 
-            max_results=3,
-            sort_by=arxiv.SortCriterion.Relevance
-        ).results()
-        
-        return "\n\n".join(
-            f"Title: {r.title}\nAuthors: {', '.join(a.name for a in r.authors)}\n"
-            f"Published: {r.published.strftime('%Y-%m-%d')}\nSummary: {r.summary[:250]}..."
-            for r in results
-        )
-    except Exception as e:
-        return f"Error performing ArXiv search: {str(e)}"
-
-@tool 
-def document_qa(input_str: str) -> str:
-    """Answer questions from documents. Input format: 'document_text||question'"""
-    print(f"DEBUG: document_qa called with: {input_str}")
-    try:
-        if '||' not in input_str:
-            return "Invalid format. Input must be: 'document_text||question'"
-        
-        context, question = input_str.split('||', 1)
-        # Load QA model on first call or ensure it's loaded once globally.
-        # It's better to load once in __init__ for BasicAgent if possible,
-        # but this lazy loading prevents initial heavy load if tool is not used.
-        qa_model = hf_pipeline('question-answering', model='deepset/roberta-base-squad2')
-        return qa_model(question=question, context=context)['answer']
-    except Exception as e:
-        return f"Error answering question from document: {str(e)}"
-
-@tool 
-def python_execution(code: str) -> str:
-    """Execute Python code and return output.
-    The code should assign its final result to a variable named '_result_value'.
-    Example: '_result_value = 1 + 1'
-    """
-    print(f"DEBUG: python_execution called with: {code}")
-    try:
-        # Create isolated environment
-        env = {}
-        # Wrap code in a function to isolate scope and capture '_result_value'
-        # The exec function is used carefully here. In a production environment,
-        # consider a more robust and secure sandbox (e.g., Docker, dedicated service).
-        exec(f"def __exec_fn__():\n{indent_code(code)}\n_result_value = __exec_fn__()", globals(), env)
-        return str(env.get('_result_value', 'No explicit result assigned to "_result_value" variable.'))
-    except Exception as e:
-        return f"Python execution error: {str(e)}"
+import wikipedia # <--- Make sure you have this installed: pip install wikipedia
+
+# --- Dummy Tools (replace with actual, robust implementations for full functionality) ---
+class DuckDuckGoSearchTool(BaseTool):
+    name: str = "duckduckgo_search"
+    description: str = "Performs a DuckDuckGo web search for current events, general facts, or quick lookups."
+    def _run(self, query: str) -> str:
+        print(f"DEBUG: Executing duckduckgo_search with query: {query}")
+        if "current year" in query.lower():
+            # Current time is Saturday, June 7, 2025 at 12:21:08 PM NZST.
+            return "The current year is 2025."
+        if "capital of france" in query.lower():
+            return "The capital of France is Paris."
+        if "python creator" in query.lower():
+            return "Python was created by Guido van Rossum."
+        return f"Search result for '{query}': Information about {query}."
+    async def _arun(self, query: str) -> str:
+        raise NotImplementedError("Asynchronous execution not supported for now.")
+
+class WikipediaSearchTool(BaseTool):
+    name: str = "wikipedia_search"
+    description: str = "Performs a Wikipedia search for encyclopedic information, historical context, or detailed topics. Returns the first 3 sentences of the summary."
+    def _run(self, query: str) -> str:
+        print(f"DEBUG: wikipedia_search called with: {query}")
+        try:
+            return wikipedia.summary(query, sentences=3)
+        except wikipedia.DisambiguationError as e:
+            return f"Disambiguation options: {', '.join(e.options[:3])}. Please refine your query."
+        except wikipedia.PageError:
+            return "Wikipedia page not found for your query."
+        except Exception as e:
+            return f"Error performing Wikipedia search: {str(e)}"
+    async def _arun(self, query: str) -> str:
+        raise NotImplementedError("Asynchronous execution not supported for now.")
+
+class ArxivSearchTool(BaseTool):
+    name: str = "arxiv_search"
+    description: str = "Searches ArXiv for scientific papers, research, or cutting-edge technical information."
+    def _run(self, query: str) -> str:
+        print(f"DEBUG: Executing arxiv_search with query: {query}")
+        return f"ArXiv result for '{query}': Scientific papers related to {query}."
+    async def _arun(self, query: str) -> str:
+        raise NotImplementedError("Asynchronous execution not supported for now.")
+
+class DocumentQATool(BaseTool):
+    name: str = "document_qa"
+    description: str = "Answers questions based on provided document text. Input format: 'document_text||question'."
+    def _run(self, input_str: str) -> str:
+        print(f"DEBUG: Executing document_qa with input: {input_str}")
+        if "||" not in input_str:
+            return "[Error] Invalid input for document_qa. Expected 'document_text||question'."
+        doc_text, question = input_str.split("||", 1)
+        if "Paris" in doc_text and "capital" in question:
+            return "The capital of France is Paris."
+        return f"Answer to '{question}' from document: '{doc_text[:50]}...' is not directly found."
+    async def _arun(self, query: str) -> str:
+        raise NotImplementedError("Asynchronous execution not supported for now.")
+
+class PythonExecutionTool(BaseTool):
+    name: str = "python_execution"
+    description: "Executes Python code for complex calculations, data manipulation, or logical operations. Always assign the final result to a variable named '_result_value'."
+    def _run(self, code: str) -> str:
+        print(f"DEBUG: Executing python_execution with code: {code}")
+        try:
+            local_vars = {}
+            exec(code, globals(), local_vars)
+            if '_result_value' in local_vars:
+                return str(local_vars['_result_value'])
+            return "Python code executed successfully, but no _result_value was assigned."
+        except Exception as e:
+            return f"[Python Error] {str(e)}"
+    async def _arun(self, query: str) -> str:
+        raise NotImplementedError("Asynchronous execution not supported for now.")
 
 class VideoTranscriptionTool(BaseTool):
     name: str = "transcript_video"
-    # CORRECTED LINE BELOW: Added '=' for assignment
-    description: str = "Fetch text transcript from YouTube videos using URL or ID. Use for any question involving video or audio. Input is the YouTube URL or ID." 
-
-    def _run(self, url_or_id: str) -> str:
-        print(f"DEBUG: transcript_video called with: {url_or_id}")
-        video_id = None
-        # Basic parsing for common YouTube URL formats
-        if "youtube.com/watch?v=" in url_or_id:
-            video_id = url_or_id.split("v=")[1].split("&")[0]
-        elif "youtu.be/" in url_or_id:
-            video_id = url_or_id.split("youtu.be/")[1].split("?")[0]
-        elif len(url_or_id.strip()) == 11 and not ("http://" in url_or_id or "https://" in url_or_id):
-            video_id = url_or_id.strip() # Assume it's just the ID
-
-        if not video_id:
-            return f"Invalid or unsupported YouTube URL/ID: {url_or_id}. Please provide a valid YouTube URL or 11-character ID."
-
-        try:
-            transcription = YouTubeTranscriptApi.get_transcript(video_id)
-            return " ".join([part['text'] for part in transcription])
-
-        except Exception as e:
-            return f"Error fetching transcript for video ID '{video_id}': {str(e)}. It might not have an English transcript, or the video is unavailable."
+    description: str = "Transcribes video content from a given YouTube URL or video ID."
+    def _run(self, query: str) -> str:
+        print(f"DEBUG: Executing transcript_video with query: {query}")
+        if "youtube.com" in query or "youtu.be" in query:
+            return f"Transcription of YouTube video '{query}': This is a sample transcription of the video content."
+        return "[Error] Invalid input for transcript_video. Please provide a valid YouTube URL or video ID."
+    async def _arun(self, query: str) -> str:
+        raise NotImplementedError("Asynchronous execution not supported for now.")
 
-    def _arun(self, *args, **kwargs):
-        raise NotImplementedError("Async not supported for this tool.")
 
 # --- Agent State Definition ---
 # --- Agent State ---