local app test

app_local_test.py

@@ -0,0 +1,470 @@
+from typing import List, Dict, Tuple
+import requests
+import os
+import json
+
+import ollama
+from smolagents import CodeAgent, DuckDuckGoSearchTool, VisitWebpageTool, LiteLLMModel, Tool
+from youtube_transcript_api import YouTubeTranscriptApi
+import whisper
+import pandas as pd
+from pytubefix import YouTube
+from pytubefix.cli import on_progress
+from bs4 import BeautifulSoup
+import wikipediaapi
+import cv2
+import numpy as np
+
+
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+CACHE_FILE = "answers_cache.json"
+
+
+class ImageLoaderTool(Tool):
+    name = "image_loader"
+    description = (
+        "Loads an image from a given URL using cv2 and returns it as a numpy array. "
+        "Input: URL of the image."
+        "Output: Image as a numpy array."
+        "Note: This tool requires the 'cv2' library to be installed."
+    )
+    inputs = {
+        "image_url": {"type": "string", "description": "URL of the image."},
+    }
+    output_type = "numpy.ndarray"
+    def forward(self, image_url: str) -> str:
+        if not image_url.startswith("http"):
+            raise ValueError(f"Invalid URL: {image_url}")
+        try:
+            response = requests.get(image_url)
+            image = cv2.imdecode(np.frombuffer(response.content, np.uint8), cv2.IMREAD_COLOR)
+            return image
+        except Exception as e:
+            raise ValueError(f"Error loading image: {e}")
+
+
+class SpeechToTextTool(Tool):
+    name = "speech_to_text"
+    description = (
+        "Converts an audio file to text. "
+    )
+    inputs = {
+        "audio_file_path": {"type": "string", "description": "Path to the audio file."},
+    }
+    output_type = "string"
+    
+    def __init__(self):
+        super().__init__()
+        self.model = whisper.load_model("base")
+
+    def forward(self, audio_file_path: str) -> str:
+        if not os.path.exists(audio_file_path):
+            raise ValueError(f"Audio file not found: {audio_file_path}")
+        result = self.model.transcribe(audio_file_path)
+        return result.get("text", "")
+
+
+class YoutubeSubtitlesTranscriptTool(Tool):
+    name = "youtube_subtitles_transcript"
+    description = (
+        "Fetches the transcript of a YouTube video. "
+        "Input: YouTube video URL."
+        "Output: Transcript text."
+    )
+    inputs = {
+        "video_url": {"type": "string", "description": "YouTube video URL."},
+    }
+    output_type = "string"
+
+    def forward(self, video_url: str) -> str:
+        if not video_url.startswith("https://www.youtube.com/watch?v="):
+            raise ValueError(f"Invalid YouTube URL: {video_url}")
+        video_id = video_url.split("v=")[-1]
+        try:
+            transcript = YouTubeTranscriptApi.get_transcript(video_id)
+            transcript_text = " ".join([entry["text"] for entry in transcript])
+            return transcript_text
+        except Exception as transcript_error:
+            print(f"Transcript not available: {transcript_error}")
+            try:
+                # Fallback: Download audio for processing
+                youtube_audio_transcript_tool = YoutubeAudioTranscriptTool()
+                transcript_text = youtube_audio_transcript_tool.forward(video_url)
+                print("Audio downloaded successfully.")
+                return transcript_text  # Assuming the tool returns some text representation
+            except Exception as e:
+                raise ValueError(f"Error downloading audio or converting to text: {e}")
+
+
+class YoutubeAudioTranscriptTool(Tool):
+    name = "youtube_audio_transcript"
+    description = (
+        "Downloads the audio from a YouTube video and converts it to text. "
+        "Input: YouTube video URL."
+    )
+    inputs = {
+        "video_url": {"type": "string", "description": "YouTube video URL."},
+    }
+    output_type = "string"
+
+    def forward(self, video_url: str) -> str:
+        if not video_url.startswith("https://www.youtube.com/watch?v="):
+            raise ValueError(f"Invalid YouTube URL: {video_url}")
+        try:
+            yt = YouTube(video_url, on_progress_callback=on_progress)
+            audio_stream = yt.streams.filter(progressive=True, file_extension='mp4').first()
+            audio_file_path = audio_stream.download(filename_prefix="audio_")
+            speech_to_text_tool = SpeechToTextTool()
+            transcript = speech_to_text_tool.forward(audio_file_path)
+            os.remove(audio_file_path)  # Clean up the downloaded file
+            return transcript
+        except Exception as e:
+            raise ValueError(f"Error downloading audio or converting to text: {e}")
+
+
+class WikipediaSearchTool(Tool):
+    name = "wikipedia_search"
+    description = (
+        "Searches Wikipedia for a given query and returns the summary of the first result."
+        "Input: Search query."
+        "Output: Wikipedia article."
+    )
+    inputs = {
+        "query": {"type": "string", "description": "Search query."},
+    }
+    output_type = "string"
+
+    def forward(self, query: str) -> str:
+        wiki_wiki = wikipediaapi.Wikipedia(
+            user_agent='wikipedia_agent',
+            language='en',
+            extract_format=wikipediaapi.ExtractFormat.WIKI
+        )
+        p_wiki = wiki_wiki.page(query)
+        if not p_wiki.exists():
+            raise ValueError(f"No Wikipedia page found for query: {query}")
+        print(p_wiki.text)
+        return p_wiki.text
+
+
+class ParseURLTool(Tool):
+    name = "parse_url"
+    description = (
+        "Parses a URL and returns the text content of the webpage."
+        "Input: URL."
+        "Output: Text content of the webpage."
+    )
+    inputs = {
+        "url": {"type": "string", "description": "URL to parse."},
+    }
+    output_type = "string"
+
+    def forward(self, url: str) -> str:
+        if not url:
+            raise ValueError("URL cannot be empty.")
+        # Fetch the HTML content
+        response = requests.get(url)
+        # Retrieve the HTML content
+        html = response.text
+        # Create a BesutifulSoup Object
+        soup = BeautifulSoup(html, 'html.parser')
+        # Select all <p> tags
+        paragraphs = soup.select("p")
+        webpage_text_list = []
+        for para in paragraphs:
+            # Get the text content of each <p> tag
+            text = para.text
+            webpage_text_list.append(text)
+
+        webpage_text = ",".join(webpage_text_list)
+        print(f"Webpage text:\n {webpage_text}")
+        return webpage_text
+        
+        
+class OllamaAgent:
+    def __init__(self, model_id: str = "llama3"):
+    
+        model = LiteLLMModel(
+            model_id=f"ollama/{model_id}", # Ollama model ID
+            api_base="http://127.0.0.1:11434", # Ollama API base URL
+            # num_ctx=8096,  # Increased context
+            # timeout=300,  # 5-minute timeout
+        )
+
+        self.agent = CodeAgent(
+            model=model,
+            tools=[
+                DuckDuckGoSearchTool(),
+                VisitWebpageTool(),
+                WikipediaSearchTool(),
+                YoutubeSubtitlesTranscriptTool(),
+                YoutubeAudioTranscriptTool(),
+                SpeechToTextTool(),
+                ParseURLTool(),
+                ],
+            verbosity_level=2,
+            # planning_interval=10,
+            add_base_tools=True,
+            additional_authorized_imports=[
+                "re",
+                "requests",
+                "bs4",
+                "urllib",
+                "pytubefix",
+                "pytubefix.cli",
+                "youtube_transcript_api",
+                "wikipediaapi",
+                "whisper",
+                "pandas",
+                "cv2",
+                "numpy",
+            ],
+            max_steps=5,
+        )
+
+        print("OllamaAgent initialized.")
+
+    def __call__(self, question: str) -> str:
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        answer = self.agent.run(question)
+        print(f"Agent returning answer: {answer}")
+        return answer
+
+
+def cache_answers(answers_payload, results_log):
+    """
+    Cache answers and results log to a local file.
+    """
+    cache_data = {
+        "answers_payload": answers_payload,
+        "results_log": results_log,
+    }
+    with open(CACHE_FILE, "w") as f:
+        json.dump(cache_data, f)
+    print(f"Cached {len(answers_payload)} answers to {CACHE_FILE}.")
+
+
+def load_cached_answers():
+    """
+    Load cached answers from the local file.
+    """
+    if os.path.exists(CACHE_FILE):
+        with open(CACHE_FILE, "r") as f:
+            cache_data = json.load(f)
+        print(f"Loaded {len(cache_data['answers_payload'])} cached answers from {CACHE_FILE}.")
+        return cache_data["answers_payload"], cache_data["results_log"]
+    return [], []
+
+
+def ollama_pull_model(model_name: str) -> bool | tuple[str, None]:
+    """
+    Check if the model is available locally and pull it if not.
+    
+    model_name: str
+        The name of the model to check.
+    
+    Returns True if the model is available, False otherwise.
+    """
+    try:
+        # Try to pull the model (this will check availability)
+        ollama.pull(model_name)
+        print(f"Model {model_name} is available.")
+        return True
+    except Exception as e:
+        # If the model doesn't exist, it will raise an error
+        print(f"Error pulling model: {e}")
+        return f"Error pulling model: {e}", None
+
+
+def fetch_questions(api_url: str) -> tuple[str, None] | List[Dict[str, str]]:
+    """
+    Fetch questions from the API.
+    
+    api_url: str
+        The base URL of the API.
+    
+    Returns a list of questions.
+    """
+    api_url = DEFAULT_API_URL
+    questions_url = f"{api_url}/questions"
+
+    print(f"Fetching questions from: {questions_url}")
+
+    try:
+        response = requests.get(questions_url, timeout=15)
+        response.raise_for_status()
+        questions_data = response.json()
+        if not questions_data:
+             print("Fetched questions list is empty.")
+             return "Fetched questions list is empty or invalid format.", None
+        print(f"Fetched {len(questions_data)} questions.")
+        return questions_data
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching questions: {e}")
+        return f"Error fetching questions: {e}", None
+    except requests.exceptions.JSONDecodeError as e:
+         print(f"Error decoding JSON response from questions endpoint: {e}")
+         print(f"Response text: {response.text[:500]}")
+         return f"Error decoding server response for questions: {e}", None
+    except Exception as e:
+        print(f"An unexpected error occurred fetching questions: {e}")
+        return f"An unexpected error occurred fetching questions: {e}", None
+
+
+def improve_prompt(prompt: str) -> str:
+    """
+    Improve the prompt by adding specific instructions for the agent.
+    
+    prompt: str
+        The original prompt.
+    
+    Returns the improved prompt.
+    """
+
+    prompt = f"Question: {prompt}\n" \
+        "Additional Instructions:\n" \
+        "Put your Thoughts (Thought) with a '#' at the beggining of their lines to avoid Error: invalid syntax and Code parsing fails." \
+
+    return prompt
+
+
+def run_agent(agent, questions_data) -> Tuple[List[Dict[str, str]], List[Dict[str, str]]]:
+    """
+    Run the agent on a list of questions and return the results.
+    
+    Args: 
+        agent: The agent to run.
+        questions_data: A list of dictionaries containing the questions and task IDs.
+        
+    Returns:
+        results_log: A list of dictionaries containing the task ID, question, and submitted answer.
+        answers_payload: A list of dictionaries containing the task ID and submitted answer.
+    """
+    results_log = []
+    answers_payload = []
+    print(f"Running agent on {len(questions_data)} questions...")
+    for item in questions_data:
+        task_id = item.get("task_id")
+        question_text = item.get("question")
+        if not task_id or question_text is None:
+            print(f"Skipping item with missing task_id or question: {item}")
+            continue
+        try:
+            # question_text = improve_prompt(question_text)
+            submitted_answer = agent(question_text)
+            answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
+            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
+        except Exception as e:
+                print(f"Error running agent on task {task_id}: {e}")
+                results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})
+
+    if not answers_payload:
+        print("Agent did not produce any answers to submit.")
+    
+    return results_log, answers_payload
+
+
+def submit_answers(
+    username: str,
+    agent_code: str,
+    answers_payload: List[Dict[str, str]],
+    results_log: List[Dict[str, str]]
+) -> Tuple[str, pd.DataFrame]:
+    """
+    Submit the answers to the API and return the status message and results DataFrame.
+    
+    Args:
+        username: The username of the person submitting the answers.
+        agent_code: The code of the agent used.
+        answers_payload: A list of dictionaries containing the task ID and submitted answer.
+        results_log: A list of dictionaries containing the task ID, question, and submitted answer.
+    
+    Returns:
+        status_message: A message indicating the status of the submission.
+        results_df: A DataFrame containing the results log.
+    """
+    submit_url = f"{DEFAULT_API_URL}/submit"
+    submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
+    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
+    print(status_update)
+    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
+    try:
+        response = requests.post(submit_url, json=submission_data, timeout=60)
+        response.raise_for_status()
+        result_data = response.json()
+        final_status = (
+            f"Submission Successful!\n"
+            f"User: {result_data.get('username')}\n"
+            f"Overall Score: {result_data.get('score', 'N/A')}% "
+            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
+            f"Message: {result_data.get('message', 'No message received.')}"
+        )
+        print("Submission successful.")
+        results_df = pd.DataFrame(results_log)
+        return final_status, results_df
+    except requests.exceptions.HTTPError as e:
+        error_detail = f"Server responded with status {e.response.status_code}."
+        try:
+            error_json = e.response.json()
+            error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
+        except requests.exceptions.JSONDecodeError:
+            error_detail += f" Response: {e.response.text[:500]}"
+        status_message = f"Submission Failed: {error_detail}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.Timeout:
+        status_message = "Submission Failed: The request timed out."
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.RequestException as e:
+        status_message = f"Submission Failed: Network error - {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except Exception as e:
+        status_message = f"An unexpected error occurred during submission: {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+
+def main():
+    model_id = 'qwen2.5:7b'
+    ollama_pull_model(model_id)
+    
+    # Initialize the agent
+    try:
+        agent = OllamaAgent(model_id=model_id)
+    except Exception as e:
+        print(f"Error instantiating agent: {e}")
+        return f"Error initializing agent: {e}", None
+    
+    # Fetch questions
+    questions_data = fetch_questions(DEFAULT_API_URL)[:3]
+    
+    # Run the agent
+    if isinstance(questions_data, list):
+        results_log, answers_payload = run_agent(agent, questions_data)
+        
+    # Cache answers
+    cache_answers(answers_payload, results_log)
+    
+    # Load cached answers
+    answers_payload, results_log = load_cached_answers()
+    
+    # Submit answers
+    status_message, results_df = submit_answers(
+        username="test_user",
+        agent_code="test_code_filler",
+        answers_payload=answers_payload,
+        results_log=results_log
+    )
+
+    print("Final status message:", status_message)
+    for TaskID, Question, SubmittedAnswer in zip(results_df["Task ID"], results_df["Question"], results_df["Submitted Answer"]):
+        print(f"Task ID: {TaskID}, Question: {Question}, Submitted Answer: {SubmittedAnswer}")
+
+
+if __name__ == "__main__":
+    main()