Delete nodes.py

nodes.py

@@ -1,322 +0,0 @@
-from state import QuestionState
-from agent_model import model
-from utilities import parse_response
-
-from langchain_core.messages import HumanMessage
-
-import mistune
-from bs4 import BeautifulSoup
-from youtube_transcript_api import YouTubeTranscriptApi
-# from googlesearch import search, SearchResult
-import re
-import os
-from dotenv import load_dotenv
-import base64
-import whisper
-import pandas as pd
-import requests
-import json
-
-model_whisper = whisper.load_model("tiny")
-load_dotenv()
-
-
-def read_question(state: QuestionState):
-    """Agent reads and logs incoming question"""
-    question = state["question"]
-
-    print(f"Here is the current question:\n{question}")
-
-    return {}
-
-
-def get_file(state: QuestionState):
-    """Agent downloads a file through an API call"""
-    # Constants
-    DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-    task_id = state["task_id"]
-    file_name = state["file_name"]
-
-    # Construct the endpoint URL
-    endpoint = f"{DEFAULT_API_URL}/files/{task_id}"
-
-    # Make the GET request to download the file
-    response = requests.get(endpoint)
-
-    # Check if the request was successful
-    if response.status_code == 200:
-        # Write the content to a file
-        with open(file_name, 'wb') as file:
-            file.write(response.content)
-        print(f"File downloaded successfully and saved as '{file_name}'")
-    else:
-        print(f"Failed to download file. Status code: {response.status_code}")
-
-    return {}
-
-
-def classify_question(state: QuestionState):
-    """Classifies the question and determines its type or category."""
-    question = state["question"]
-
-    # Prepare our prompt for the LLM
-    prompt = f"""
-    As a knowledgeable assistant, analyze the following question and classify it into a specific category.
-    Question:
-    {question}
-    These are the pairs of classification and context you can use, in the form of <classification>+<context>: 
-    - web_search+<query_to_use_for_web_search> (use direct language in query, as you would in a google search)
-    - url_crawl
-    - text_reasoning+<the_question_or_statement_for_reasoning>
-    - excel_file
-    - code+<code_snippet_to_execute>
-    - audio
-    - youtube+<youtube_link_to_get_transcription>
-    - other+<any_other_relevant_context>
-    It is MANDATORY to provide your classification and context in the following format:
-    classification: <category>
-    context: <context>
-    """
-
-    # Call the LLM
-    messages = [HumanMessage(content=prompt)]
-    response = model.invoke(messages)
-
-    # Parse the response
-    classification, context = parse_response(response)  # Implement this function to extract classification and context
-
-    # Define context generation based on classification
-    context_mapping = {
-        "web_search": f"context: {question}",
-        "url_crawl": f"context: {question}",
-        "text_reasoning": f"context: {question}",
-        "mathematics": f"context: {question}",  # Implement this function
-        "code": f"context: {question}",
-        "other": f"context: {question}",
-    }
-
-    # Update messages for tracking
-    new_messages = state.get("messages", []) + [
-        {"role": "user", "content": prompt},
-        {"role": "assistant", "content": response.content}
-    ]
-
-    # Return state updates
-    return {
-        "question_category": classification,
-        "context": context,
-        "messages": new_messages
-    }
-
-
-def handle_other(state: QuestionState):
-    """The assistant gives a response through the llm model"""
-    question = state["question"]
-
-    prompt = f"""
-    {question}
-    """
-
-    # Call the LLM
-    messages = [HumanMessage(content=prompt)]
-    response = model.invoke(messages)
-
-    return {
-        "expected_answer": response.content
-    }
-
-
-def handle_reasoning(state: QuestionState):
-    """The assistant gives a response through the llm model"""
-    question = state["question"]
-    context = state["context"]
-
-    system_prompt = f"""
-    You are a helpful assistant tasked with answering questions using a set of tools. 
-    Now, I will ask you a question. Report your thoughts, and finish your answer with the following template: 
-    FINAL ANSWER: [YOUR FINAL ANSWER]. 
-    YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string.
-    Your answer should only start with "FINAL ANSWER: ", then follows with the answer. 
-    """
-
-    prompt = f"""
-    {question}
-    {context}
-    """
-
-    message_content = [
-        {
-            "type": "text",
-            "text": system_prompt,
-        },
-        {
-            "type": "text",
-            "text": prompt,
-        }
-    ]
-
-    if state["file_name"] and "png" in state["file_name"]:
-        image_path = state["file_name"]
-        with open(image_path, "rb") as image_file:
-            # Encode the image data to base64
-            image_data = base64.b64encode(image_file.read()).decode("utf-8")
-
-        message_content.append(
-            {
-                "type": "image",
-                "source_type": "base64",
-                "data": image_data,
-                "mime_type": "image/jpeg",
-            })
-
-    # Call the LLM
-    messages = [HumanMessage(content=message_content)]
-    response = model.invoke(messages)
-
-    return {
-        "expected_answer": response.content
-    }
-
-
-def excel_to_markdown(state: QuestionState):
-    file_name = state["file_name"]
-    # Read the Excel file
-    df = pd.read_excel(file_name)
-
-    # Convert the DataFrame to Markdown format
-    markdown_table = df.to_markdown(index=False)
-
-    return {
-        "context": markdown_table
-    }
-
-
-def create_json_for_math(state: QuestionState):
-    question = state["question"]
-    context = state["context"]
-
-    prompt = f"""
-    considering this table:
-    {context}
-    and this question:
-    {question}
-    create a json file in this format: [1, 2, 3, 4, 5]
-    Where each number is going to be added together later
-    in the json, put all the numbers that need to be added together, following the request of the question.
-    finally, only give this as a response:
-    json_numbers: <json_object_you_created>
-    """
-    message_content = [
-        {
-            "type": "text",
-            "text": prompt,
-        }
-    ]
-
-    # Call the LLM
-    messages = [HumanMessage(content=message_content)]
-    response = model.invoke(messages)
-
-    json_to_add = response.content.replace("json_numbers: ", "").strip()
-
-    parsed_data = json.loads(json_to_add)
-
-    return {
-        "context": str(sum(parsed_data))
-    }
-
-
-def transcribe_audio(state: QuestionState):
-    file_name = state["file_name"]
-    audio_transcription = model_whisper.transcribe(file_name)
-    text = audio_transcription['text']
-
-    return {
-        "context": text
-    }
-
-
-def get_youtube_transcript(state: QuestionState):
-    """Tool to get the transcript of a YouTube video"""
-    question = state["question"]
-
-    # Use a regular expression to find the video ID
-    match = re.search(r'https://www\.youtube\.com/watch\?v=([a-zA-Z0-9_-]+)', question)
-
-    if match:
-        video_id = match.group(1)
-        print(video_id)
-    else:
-        print("No video ID found.")
-
-    # video_id = question.split("https://www.youtube.com/watch?v=")
-    ytt_api = YouTubeTranscriptApi()
-    fetched_transcript = ytt_api.fetch(video_id)
-
-    full_transcript = ''
-    for snippet in fetched_transcript:
-        full_transcript += snippet.text + '\n'
-
-    return {"context": full_transcript}
-
-
-def get_final_answer(state: QuestionState):
-    expected_answer = state["expected_answer"]
-
-    return {
-        "expected_answer": expected_answer.split("FINAL ANSWER: ")[1].strip()
-    }
-
-
-def handle_web_search(state: QuestionState):
-    """The assistant runs a web search tool to search for information on the Web"""
-    base_url = "https://www.googleapis.com/customsearch/v1"
-
-    search_input = state["context"].replace("\"", "")
-    print(f"Search query: {search_input}")
-    params = {
-        'key': os.getenv("CUSTOM_SEARCH_API"),
-        'cx': os.getenv("SEARCH_ID"),
-        'q': search_input,
-        'num': 5,
-        'start': 1
-    }
-    # search_results = search(search_input, num_results=10, advanced=True)
-    response = requests.get(base_url, params=params)
-    response_json = response.json()
-    print('---RESPONSE JSON---')
-    print(response_json)
-
-    chosen_link = "" # i don't want pdf files, they're too long
-    # Extract links from the items
-    links = [item['link'] for item in response_json['items']]
-    for l in links:
-        if ".pdf" not in l:
-            chosen_link = l
-            break
-    print('---LINKS---')
-    print(links)
-
-    return {
-        "context": chosen_link
-    }
-
-
-def enter_url(state: QuestionState):
-    url = state["context"]
-    response = requests.get(url)
-    html_content = response.text
-
-    # Step 2: Parse the HTML content
-    soup = BeautifulSoup(html_content, 'html.parser')
-
-    # Step 3: Convert to Markdown
-    markdown_converter = mistune.Markdown()
-    markdown_content = markdown_converter(soup.get_text())
-
-    # Print the Markdown content
-    print(markdown_content)
-
-    return {
-        "context": markdown_content
-    }