My model added

.gitignore

@@ -0,0 +1,4 @@
+.idea/
+__pycache__/
+dataset/
+.env

model.py

@@ -0,0 +1,180 @@
+import os
+import pandas as pd
+
+from PIL import Image
+import io
+from typing import TypedDict, Annotated
+
+from dotenv import load_dotenv
+from langgraph.graph import START, StateGraph
+from langchain_core.messages import AnyMessage
+from langgraph.graph.message import add_messages
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_openai import AzureChatOpenAI
+from langgraph.graph.state import CompiledStateGraph
+from langgraph.prebuilt import tools_condition
+from langgraph.prebuilt import ToolNode
+import matplotlib.pyplot as plt
+
+from typing import Optional
+
+from tools import get_all_tools
+
+
+load_dotenv(override=True)
+
+
+class AgentState(TypedDict):
+    # The input document
+    input_file:  Optional[str]
+    messages: Annotated[list[AnyMessage], add_messages]
+
+assistant_system = (
+    'You are a general AI assistant. I will ask you a question. Think step-by-step, 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."
+)
+
+class AssistantModel:
+    def __init__(self):
+        llm = AzureChatOpenAI(
+            openai_api_version="2024-02-01",
+            azure_deployment=os.getenv("AZURE_OPENAI_DEPLOYMENT"),
+            openai_api_key=os.getenv("AZURE_OPENAI_API_KEY"),
+            azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
+            temperature=0.0
+        )
+
+        self.llm_with_tools = llm.bind_tools(get_all_tools(), parallel_tool_calls=False)
+        self.graph = self._build_graph()
+        # self.show_graph()
+
+
+    def _assistant(self, state: AgentState):
+        sys_msg = SystemMessage(content=assistant_system)
+
+        return {"messages": [self.llm_with_tools.invoke([sys_msg] + state["messages"])]}
+
+    def show_graph(self):
+        # python -m pip install --config-settings="--global-option=build_ext" --config-settings="--global-option=-IC:\Program Files\Graphviz\include" --config-settings="--global-option=-LC:\Program Files\Graphviz\lib" pygraphviz
+        png = self.graph.get_graph(xray=True).draw_png()
+        image = Image.open(io.BytesIO(png))
+
+        plt.imshow(image)
+        plt.axis('off')  # Turn off axes for better visualization
+        plt.show(block=False)
+
+
+    def _build_graph(self) -> CompiledStateGraph:
+        # Graph
+        builder = StateGraph(AgentState)
+
+        # Define nodes: these do the work
+        builder.add_node("assistant", self._assistant)
+        builder.add_node("tools", ToolNode(get_all_tools()))
+
+        # Define edges: these determine how the control flow moves
+        builder.add_edge(START, "assistant")
+        builder.add_conditional_edges(
+            "assistant",
+            # If the latest message (result) from assistant is a tool call -> tools_condition routes to tools
+            # If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END
+            tools_condition,
+        )
+        builder.add_edge("tools", "assistant")
+        react_graph = builder.compile()
+
+        return react_graph
+
+    @staticmethod
+    def _get_final_answer(message: AnyMessage) -> str:
+        """Extract the final answer from the message content."""
+        # Assuming the final answer is always at the end of the message
+        return message.content.split("FINAL ANSWER:")[-1].strip()
+
+    def _get_file_content(self, file_name: str) -> str:
+        """Get the file content."""
+        if file_name is None or file_name == '':
+            return ''
+
+        header = '**Attached file content:**\n'
+
+        text_file = ['.py', '.txt', '.json']
+
+        full_file_name = os.path.join(r'.\dataset', file_name)
+
+        if any(file_name.endswith(ext) for ext in text_file):
+            with open(full_file_name, 'r', encoding='utf-8') as f:
+                return header + f.read()
+
+        elif file_name.endswith(".xlsx"):
+            df = pd.read_excel(full_file_name)
+            res = df.to_html(index=False)
+            return header + res if res else ''
+
+        else:
+            return ''
+
+    def _get_image_url(self, file_name: str) -> str:
+        exts = ['.png', '.jpg', '.jpeg', '.gif']
+
+        if any(file_name.endswith(ext) for ext in exts):
+            without_ext = file_name.split('.')[0]
+            return f'https://agents-course-unit4-scoring.hf.space/files/{without_ext}'
+        else:
+            return ''
+
+
+    def ask_question(self, question: str, file_name: str) -> str:
+        question_with_file = question + '\n' + self._get_file_content(file_name)
+        image_url = self._get_image_url(file_name)
+
+        if image_url != '':
+            content = [
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": image_url
+                    }
+                },
+                {
+                    "type": "text",
+                    "text": question_with_file
+                }
+            ]
+        else:
+            content = question_with_file
+
+        messages = [HumanMessage(content=content)]
+
+        messages = self.graph.invoke({"messages": messages})
+
+        for m in messages['messages']:
+            m.pretty_print()
+
+        print('@' * 50)
+        final_answer = AssistantModel._get_final_answer(messages['messages'][-1])
+        print('The final answer is:', final_answer)
+
+        return final_answer
+
+if __name__ == '__main__':
+    model = AssistantModel()
+
+    q = 'Divide 6790 by 5'
+    # q = 'How many studio albums were published by Mercedes Sosa between 2000 and 2009 (included)? You can use the latest 2022 version of english wikipedia.'
+    # q = '.rewsna eht sa "tfel" drow eht fo etisoppo eht etirw ,ecnetnes siht dnatsrednu uoy fI'
+    # q = 'Who nominated the only Featured Article on English Wikipedia about a dinosaur that was promoted in November 2016?'
+    # q = 'Who did the actor who played Ray in the Polish-language version of Everybody Loves Raymond play in Magda M.? Give only the first name.'
+    q = 'What is the final numeric output from the attached Python code?'
+    f = 'f918266a-b3e0-4914-865d-4faa564f1aef.py'
+    q = 'The attached Excel file contains the sales of menu items for a local fast-food chain. What were the total sales that the chain made from food (not including drinks)? Express your answer in USD with two decimal places.'
+    f = '7bd855d8-463d-4ed5-93ca-5fe35145f733.xlsx'
+    # q = "Review the chess position provided in the image. It is black's turn. Provide the correct next move for black which guarantees a win. Please provide your response in algebraic notation."
+    # f = 'cca530fc-4052-43b2-b130-b30968d8aa44.png'
+
+    answer = model.ask_question(q, f)

tools.py

@@ -0,0 +1,104 @@
+from typing import Type, Optional
+
+from langchain_community.document_loaders import AsyncChromiumLoader
+from langchain_community.document_transformers import BeautifulSoupTransformer
+from langchain_community.tools.wikipedia.tool import WikipediaQueryInput
+from langchain_community.tools import WikipediaQueryRun
+from langchain_community.utilities import WikipediaAPIWrapper
+from langchain_core.callbacks import CallbackManagerForToolRun
+from langchain_core.tools import BaseTool
+from langchain_tavily import TavilySearch
+from pydantic import BaseModel, Field
+
+
+def _get_web_page(url: str) -> str:
+    """Fetches the content of a web page and transforms (beautify) it into a string."""
+
+    loader = AsyncChromiumLoader([url])
+    docs = loader.load()
+    bs_transformer = BeautifulSoupTransformer()
+    docs = bs_transformer.transform_documents(docs)
+    return '\n'.join(['=' * 30 + '\n' + doc.page_content for doc in docs])
+
+class WikipediaQueryLoad(BaseTool):
+    """Tool that searches the Wikipedia API."""
+
+    name: str = "wikipedia"
+    description: str = (
+        "A wrapper around Wikipedia. "
+        "Useful for when you need to answer general questions about "
+        "people, places, companies, facts, historical events, or other subjects. "
+        "Input should be a search query."
+    )
+    api_wrapper: WikipediaAPIWrapper = WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=20000)
+
+    args_schema: Type[BaseModel] = WikipediaQueryInput
+
+    def _run(
+        self,
+        query: str,
+        run_manager: Optional[CallbackManagerForToolRun] = None,
+    ) -> str:
+        """Use the Wikipedia tool."""
+        page_titles = self.api_wrapper.wiki_client.search(
+            query[:300], results=1
+        )
+        summary = self.api_wrapper.run(query)
+
+        # Wikipedia python package doesn't properly support some wiki syntax (i.e. tables), so
+        # the full wiki page is read separately
+        full_page = _get_web_page(f"https://en.wikipedia.org/wiki/{page_titles[0]}")
+
+        res = [
+            '**Wiki page url**:',
+            page_titles[0],
+            '**Wiki page summary:**',
+            summary,
+            '**Full page content:**',
+            full_page
+        ]
+        return '\n'.join(res)
+
+
+class WebScrapTool(BaseTool):
+    name: str = "webscraper"
+
+    description: str = "Loads full content of the web page."
+
+    # Load HTML
+    def _run(
+        self,
+        url: str,
+        run_manager: Optional[CallbackManagerForToolRun] = None,
+    ) -> str:
+        return _get_web_page(url)
+
+
+class CalculatorTool(BaseTool):
+    """Tool that performs basic calculations."""
+
+    name: str = "calculator"
+    description: str = (
+        "A calculator. "
+        "Useful for when you need to perform basic calculations."
+    )
+
+    def _run(
+        self,
+        expression: str,
+        run_manager: Optional[CallbackManagerForToolRun] = None,
+    ) -> float:
+        """Use the calculator tool."""
+        return eval(expression)
+
+
+wiki = WikipediaQueryLoad(api_wrapper=WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=20000))
+
+def get_all_tools() -> list[BaseTool]:
+    """Get all tools."""
+    return [
+        wiki,
+        WebScrapTool(),
+        TavilySearch(max_results=5, topic="general"),
+        CalculatorTool()
+    ]

validation.py

@@ -0,0 +1,63 @@
+import json
+
+import requests
+
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+def _get_response(url: str):
+    try:
+        response = requests.get(url, timeout=15)
+        response.raise_for_status()
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching questions: {e}")
+        return None
+    except Exception as e:
+        print(f"An unexpected error occurred fetching questions: {e}")
+        return None
+    return response
+
+def _get_response_json(url: str):
+    try:
+        response = _get_response(url)
+        questions_data = response.json()
+        if not questions_data:
+            print("Fetched questions list is empty.")
+            return {}, None
+        print(f"Fetched {len(questions_data)} questions.")
+    except requests.exceptions.JSONDecodeError as e:
+         print(f"Error decoding JSON response from questions endpoint: {e}")
+         print(f"Response text: {response.text[:500]}")
+         return {}, None
+
+    return questions_data
+
+
+def load_questions() -> None:
+    questions_url = f"{DEFAULT_API_URL}/questions"
+
+    questions_data = _get_response_json(questions_url)
+
+    with open(r'./dataset/questions.json', 'w') as f:
+        json.dump(questions_data, f, indent=2)
+
+
+def load_files() -> None:
+    with open(r'./dataset/questions.json', 'r') as f:
+        questions_data = json.load(f)
+
+    for q in questions_data:
+        if q["file_name"] != '':
+            files_url = f'{DEFAULT_API_URL}/files/{q["task_id"]}'
+
+            print(f"Fetching file from: {files_url}")
+
+            file_data = _get_response(files_url)
+
+            with open(f'./dataset/{q["file_name"]}', 'wb') as f:
+                f.write(file_data.content)
+                print(f"File {q['file_name']} downloaded successfully.")
+
+
+if __name__ == '__main__':
+    # load_questions()
+    load_files()