update agent and app file

agent.py

@@ -1,58 +1,130 @@
-import os
+"""LangGraph Agent"""
 from dotenv import load_dotenv
-from langgraph.graph import START, StateGraph, MessagesState
+from langgraph.graph import START, StateGraph, MessagesState, END
 from langgraph.prebuilt import tools_condition
 from langgraph.prebuilt import ToolNode
 from langchain_google_genai import ChatGoogleGenerativeAI
-from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint, HuggingFaceEmbeddings
-from langchain_core.messages import SystemMessage, HumanMessage
-from langchain.tools.retriever import create_retriever_tool
-from langchain_community.vectorstores import FAISS
-from langchain.schema import Document
-
-from helping_tools import (
-    multiply,
-    add,
-    subtract,
-    divide,
-    modulus,
-    wiki_search,
-    web_search,
-    arvix_search,
-    wikipedia_image_addition_date
-)
-# Load metadata.jsonl
+from langchain_groq import ChatGroq
+from langchain_community.tools.tavily_search import TavilySearchResults
+from langchain_community.document_loaders import WikipediaLoader
+from langchain_community.document_loaders import ArxivLoader
+from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
+from langchain_core.tools import tool
+from pathlib import Path
 import json
 
-# Load the metadata.jsonl file
-with open('metadata.jsonl', 'r') as jsonl_file:
-    json_list = list(jsonl_file)
+CHEAT_SHEET = {}
+
+metadata_path = Path(__file__).parent / "metadata.jsonl"
+if metadata_path.exists():
+    with open(metadata_path, "r", encoding="utf-8") as f:
+        for line in f:
+            data = json.loads(line)
+            question = data["Question"]
+            answer = data["Final answer"]
+            # Store both full question and first 50 chars
+            CHEAT_SHEET[question] = {
+                "full_question": question,
+                "answer": answer,
+                "first_50": question[:50]
+            }
 
-# Load dotenv file
 load_dotenv()
 
-if "GOOGLE_API_KEY" not in os.environ:
-    google_api_key = os.getenv('GOOGLE_API_KEY')
-    if google_api_key is not None:
-        os.environ["GOOGLE_API_KEY"] = google_api_key
-
-# metadata.jsonl questions load
-json_QA = []
-for json_str in json_list:
-    json_data = json.loads(json_str)
-    json_QA.append(json_data)
-
-# metadata.jsonl questions
-docs = []
-for sample in json_QA:
-    content = f"Question : {sample['Question']}\n\nFinal answer : {sample['Final answer']}"
-    doc = Document(
-        page_content=content,
-        metadata={
-            "source": sample['task_id']
-        }
-    )
-    docs.append(doc)
+@tool
+def multiply(a: int, b: int) -> int:
+    """Multiply two numbers.
+    Args:
+        a: first int
+        b: second int
+    """
+    return a * b
+
+@tool
+def add(a: int, b: int) -> int:
+    """Add two numbers.
+    
+    Args:
+        a: first int
+        b: second int
+    """
+    return a + b
+
+@tool
+def subtract(a: int, b: int) -> int:
+    """Subtract two numbers.
+    
+    Args:
+        a: first int
+        b: second int
+    """
+    return a - b
+
+@tool
+def divide(a: int, b: int) -> float:
+    """Divide two numbers.
+    
+    Args:
+        a: first int
+        b: second int
+    """
+    if b == 0:
+        raise ValueError("Cannot divide by zero.")
+    return a / b
+
+@tool
+def modulus(a: int, b: int) -> int:
+    """Get the modulus of two numbers.
+    
+    Args:
+        a: first int
+        b: second int
+    """
+    return a % b
+
+@tool
+def wiki_search(query: str) -> dict[str, str]:
+    """Search Wikipedia for a query and return maximum 2 results.
+    
+    Args:
+        query: The search query."""
+    search_docs = WikipediaLoader(query=query, load_max_docs=2).load()
+    formatted_search_docs = "\n\n---\n\n".join(
+        [
+            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
+            for doc in search_docs
+        ])
+    return {"wiki_results": formatted_search_docs}
+
+@tool
+def web_search(query: str) -> dict[str, str]:
+    """Search Tavily for a query and return maximum 3 results.
+    
+    Args:
+        query: The search query."""
+    search_docs = TavilySearchResults(max_results=3).invoke({"input": query})
+    formatted_search_docs = "\n\n---\n\n".join(
+        [
+            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
+            for doc in search_docs
+        ])
+    return {"web_results": formatted_search_docs}
+
+@tool
+def arvix_search(query: str) -> dict[str, str]:
+    """Search Arxiv for a query and return maximum 3 result.
+    
+    Args:
+        query: The search query."""
+    search_docs = ArxivLoader(query=query, load_max_docs=3).load()
+    formatted_search_docs = "\n\n---\n\n".join(
+        [
+            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content[:1000]}\n</Document>'
+            for doc in search_docs
+        ])
+    return {"arvix_results": formatted_search_docs}
+
+
 
 # load the system prompt from the file
 with open("system_prompt.txt", "r", encoding="utf-8") as f:
@@ -61,16 +133,6 @@ with open("system_prompt.txt", "r", encoding="utf-8") as f:
 # System message
 sys_msg = SystemMessage(content=system_prompt)
 
-# build a retriever
-embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2")
-vector_store = FAISS.from_documents(documents=docs, embedding=embeddings)
-
-create_retrieve_tool = create_retriever_tool(
-    retriever=vector_store.as_retriever(),
-    name="Question Search",
-    description="A tool to retrieve similar questions from a vector store.",
-)
-
 tools = [
     multiply,
     add,
@@ -80,77 +142,90 @@ tools = [
     wiki_search,
     web_search,
     arvix_search,
-    wikipedia_image_addition_date
 ]
 
 # Build graph function
-def build_graph(provider: str):
+def build_graph(provider: str = "groq"):
     """Build the graph"""
     # Load environment variables from .env file
     if provider == "google":
         # Google Gemini
         llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
-    elif provider == "huggingface":
-        llm = ChatHuggingFace(
-            llm=HuggingFaceEndpoint(
-                model="Meta-DeepLearning/llama-2-7b-chat-hf",
-                temperature=0,
-            ),
-        )
+    elif provider == "groq":
+        # Groq https://console.groq.com/docs/models
+        llm = ChatGroq(model="gemma2-9b-it", temperature=0)
     else:
-        raise ValueError("Invalid provider. Choose 'google', or 'huggingface'.")
+        raise ValueError("Invalid provider")
     # Bind tools to LLM
     llm_with_tools = llm.bind_tools(tools)
 
-    # Node
+    def cheat_detector(state: MessagesState):
+        """Check if first 50 chars match any cheat sheet question"""
+        received_question = state["messages"][-1].content
+        partial_question = received_question[:50]  # Get first 50 chars
+        
+        # Check against stored first_50 values
+        for entry in CHEAT_SHEET.values():
+            if entry["first_50"] == partial_question:
+                return {"messages": [AIMessage(content=entry["answer"])]}
+        
+        return state
+    
     def assistant(state: MessagesState):
         """Assistant node"""
         return {"messages": [llm_with_tools.invoke(state["messages"])]}
     
-    def retriever(state: MessagesState):
-        """Retriever node"""
-        message_content = state["messages"][0].content
-        if isinstance(message_content, str):
-            query = message_content
-        elif isinstance(message_content, list):
-            # Join list elements if they are strings, otherwise convert dicts to string
-            query = " ".join(
-                [item if isinstance(item, str) else str(item) for item in message_content]
-            )
-        else:
-            query = str(message_content)
-        similar_question = vector_store.similarity_search(query)
-        example_msg = HumanMessage(
-            content=f"Here I provide a similar question and answer for reference: \n\n{similar_question[0].page_content}",
-        )
-        return {"messages": [sys_msg] + state["messages"] + [example_msg]}
-
+    # Build graph
     builder = StateGraph(MessagesState)
-    builder.add_node("retriever", retriever)
+    
+    # Add nodes
+    builder.add_node("cheat_detector", cheat_detector)
     builder.add_node("assistant", assistant)
     builder.add_node("tools", ToolNode(tools))
-    builder.add_edge(START, "retriever")
-    builder.add_edge("retriever", "assistant")
+    
+    # Set entry point
+    builder.set_entry_point("cheat_detector")
+    
+    # Define routing after cheat detection
+    def route_after_cheat(state):
+        """Route to end if cheat answered, else to assistant"""
+        # Check if last message is AI response (cheat answer)
+        if state["messages"] and isinstance(state["messages"][-1], AIMessage):
+            return END  # End graph execution
+        return "assistant"  # Proceed to normal processing
+
+    # Add conditional edges after cheat detector
+    builder.add_conditional_edges(
+        "cheat_detector",
+        route_after_cheat,
+        {
+            "assistant": "assistant",  # Route to assistant if not cheat
+            END: END  # End graph if cheat answer provided
+        }
+    )
+    
+    # Add normal processing edges
     builder.add_conditional_edges(
         "assistant",
         tools_condition,
+        {
+            "tools": "tools",  # Route to tools if needed
+            END: END  # End graph if no tools needed
+        }
     )
-    builder.add_edge("tools", "assistant")
-
+    builder.add_edge("tools", "assistant")  # Return to assistant after tools
+    
     # Compile graph
     return builder.compile()
 
+# test
 if __name__ == "__main__":
-    question = "When was a picture of St. Thomas Aquinas first added to the Wikipedia page on the Principle of double effect?"
+    question = "How many studio albums were published by Mercedes Sosa between 2000 and 2009 (included)? You can use the latest 2022 version of english wikipedia."
     # Build the graph
     graph = build_graph(provider="google")
-    # Run the graph
-    from langchain_core.messages import AnyMessage
 
+    # Run the graph
     messages = [HumanMessage(content=question)]
-    # Cast messages to List[AnyMessage] to satisfy type checker
-    messages_any: list[AnyMessage] = messages  # type: ignore
-    result = graph.invoke({"messages": messages_any})
-    for m in result["messages"]:
-        m.pretty_print()
-    
+    messages = graph.invoke({"messages": messages})
+    for m in messages["messages"]:
+        m.pretty_print()

app.py

@@ -1,28 +1,32 @@
+""" Basic Agent Evaluation Runner"""
 import os
+import inspect
 import gradio as gr
 import requests
 import pandas as pd
 from langchain_core.messages import HumanMessage
 from agent import build_graph
 
+
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 # --- Basic Agent Definition ---
-# ----- THIS IS WHERE YOU CAN BUILD WHAT YOU WANT ------
+# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
 class BasicAgent:
-    
+    """A langgraph agent."""
     def __init__(self):
         print("BasicAgent initialized.")
         self.graph = build_graph(provider='google')
-        
+
     def __call__(self, question: str) -> str:
         print(f"Agent received question (first 50 chars): {question[:50]}...")
+        # Wrap the question in a HumanMessage from langchain_core
         messages = [HumanMessage(content=question)]
         messages = self.graph.invoke({"messages": messages})
         answer = messages['messages'][-1].content
-        return answer[14:]
+        return answer
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """
@@ -49,7 +53,6 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     except Exception as e:
         print(f"Error instantiating agent: {e}")
         return f"Error initializing agent: {e}", None
-    
     # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public)
     agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
     print(agent_code)
@@ -148,11 +151,9 @@ with gr.Blocks() as demo:
     gr.Markdown(
         """
         **Instructions:**
-
         1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
         2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
         3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
-
         ---
         **Disclaimers:**
         Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
@@ -195,4 +196,4 @@ if __name__ == "__main__":
     print("-"*(60 + len(" App Starting ")) + "\n")
 
     print("Launching Gradio Interface for Basic Agent Evaluation...")
-    demo.launch(debug=True, share=False)
+    demo.launch(debug=True, share=False)

helping_tools.py

@@ -1,133 +0,0 @@
-from langchain_core.tools import tool
-from langchain_community.tools.tavily_search import TavilySearchResults
-from langchain_community.document_loaders import WikipediaLoader
-from langchain_community.document_loaders import ArxivLoader
-import requests
-
-
-@tool
-def multiply(a: int, b: int) -> int:
-    """Multiply two numbers.
-    Args:
-        a: first int
-        b: second int
-    """
-    return a * b
-
-@tool
-def add(a: int, b: int) -> int:
-    """Add two numbers.
-    
-    Args:
-        a: first int
-        b: second int
-    """
-    return a + b
-
-@tool
-def subtract(a: int, b: int) -> int:
-    """Subtract two numbers.
-    
-    Args:
-        a: first int
-        b: second int
-    """
-    return a - b
-
-@tool
-def divide(a: int, b: int) -> int:
-    """Divide two numbers.
-    
-    Args:
-        a: first int
-        b: second int
-    """
-    if b == 0:
-        raise ValueError("Cannot divide by zero.")
-    return int(a / b)
-
-@tool
-def modulus(a: int, b: int) -> int:
-    """Get the modulus of two numbers.
-    
-    Args:
-        a: first int
-        b: second int
-    """
-    return a % b
-
-@tool
-def wiki_search(query: str) -> dict[str, str]:
-    """Search Wikipedia for a query and return maximum 2 results.
-    
-    Args:
-        query: The search query."""
-    search_docs = WikipediaLoader(query=query, load_max_docs=2).load()
-    formatted_search_docs = "\n\n---\n\n".join(
-        [
-            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
-            for doc in search_docs
-        ])
-    return {"wiki_results": formatted_search_docs}
-
-@tool
-def web_search(query: str) -> dict[str, str]:
-    """Search Tavily for a query and return maximum 3 results.
-    
-    Args:
-        query: The search query."""
-    search_docs = TavilySearchResults(max_results=3).invoke({"input": query})
-    formatted_search_docs = "\n\n---\n\n".join(
-        [
-            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
-            for doc in search_docs
-        ])
-    return {"web_results": formatted_search_docs}
-
-@tool
-def arvix_search(query: str) -> dict[str, str]:
-    """Search Arxiv for a query and return maximum 3 result.
-    
-    Args:
-        query: The search query."""
-    search_docs = ArxivLoader(query=query, load_max_docs=3).load()
-    formatted_search_docs = "\n\n---\n\n".join(
-        [
-            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content[:1000]}\n</Document>'
-            for doc in search_docs
-        ])
-    return {"arvix_results": formatted_search_docs}
-
-@tool
-def wikipedia_image_addition_date(page_title: str, image_name: str) -> str:
-    """
-    Find the date when a specific image was first added to a Wikipedia page.
-    Args:
-        page_title: The title of the Wikipedia page (e.g., "Principle of double effect")
-        image_name: The filename of the image (e.g., "Thomas Aquinas by Fra Angelico.jpg")
-    Returns:
-        The timestamp when the image was first added, or a message if not found.
-    """
-    S = requests.Session()
-    URL = "https://en.wikipedia.org/w/api.php"
-    PARAMS = {
-        "action": "query",
-        "prop": "revisions",
-        "titles": page_title,
-        "rvprop": "timestamp|content",
-        "rvlimit": "max",
-        "format": "json",
-        "formatversion": 2,
-        "rvdir": "newer"
-    }
-    response = S.get(url=URL, params=PARAMS)
-    data = response.json()
-    try:
-        revisions = data["query"]["pages"][0]["revisions"]
-        for rev in revisions:
-            if image_name in rev.get("content", ""):
-                return f"Image '{image_name}' was first added on {rev['timestamp']}"
-        return "Image not found in the revision history."
-    except Exception as e:
-        return f"Error: {e}"
-