get 30% score

app.py

@@ -10,46 +10,200 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 # --- Basic Agent Definition ---
 # ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
-from smolagents import CodeAgent, OpenAIModel
-from tool import web_search_tool, web_fetch_tool
+import os
+from typing import TypedDict, List, Optional, Literal
+
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, BaseMessage
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.tools import Tool
+from langgraph.graph import StateGraph, END
+from langgraph.prebuilt import ToolNode
+from typing_extensions import Annotated
+
+from langchain_core.messages import BaseMessage
+from langgraph.graph.message import add_messages
+
+from tool import web_search, web_fetch, _extract_video_id, youtube_transcript  # wrappers from step 2
+
+
+# -----------------------------
+# State
+# -----------------------------
+class AgentState(TypedDict):
+    question: str
+    messages: Annotated[list[BaseMessage], add_messages]
+    final: Optional[str]
+    steps: int
+    last_error: Optional[str]
+
 
-#model used for CodeAgent, need to be multi-modal
-llm= OpenAIModel(
-    model="gpt-4o",  # or gpt-4.1, gpt-4-turbo
-    api_key="YOUR_API_KEY",
+MAX_STEPS = 10
+
+HELP_PROMPT = (
+    "You are a general AI assistant. 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."
 )
 
-class BasicAgent:
-    def __init__(self):
-        
-        print("BasicAgent initialized.")
-        self.smart_agent = CodeAgent(
-            tools=[web_search_tool, web_fetch_tool],
-            model=llm,
-            max_steps = 5,
-        )
 
-    def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        
-        help_prompt = (
-            "You are a general AI assistant. "
-            "Answer the following question and respond ONLY using this template:\n"
-            "FINAL ANSWER: [ANSWER]\n\n"
-            "Rules:\n"
-            "- The answer should be a number, a few words, or a comma-separated list.\n"
-            "- Do NOT include explanations or reasoning.\n"
-            "- Do NOT use units unless explicitly requested.\n"
-            "- Do NOT use articles (a, an, the).\n"
-            "- Write digits in plain text unless specified otherwise.\n\n"
-            "Question:\n"
+# -----------------------------
+# LLM + Tools
+# -----------------------------
+llm = ChatOpenAI(
+    model="gpt-4o",
+    api_key=os.environ["OPENAI_API_KEY"],
+    temperature=0,
+)
+
+tools = [web_search, web_fetch, youtube_transcript]
+tool_node = ToolNode(tools)
+
+
+# -----------------------------
+# Helper: check final format
+# -----------------------------
+import re
+
+def extract_final(text: str) -> Optional[str]:
+    """
+    Robustly extracts the final answer, handling case sensitivity and bold formatting.
+    """
+    # Use regex to find "FINAL ANSWER:" case-insensitive, potentially with ** or ##
+    match = re.search(r"(?i)(\*\*|##)?\s*FINAL ANSWER\s*(\*\*|##)?\s*:\s*(.*)", text, re.DOTALL)
+    
+    if match:
+        # Return the captured content (group 3)
+        return match.group(3).strip()
+    return None
+
+
+# -----------------------------
+# Nodes
+# -----------------------------
+def start(state: AgentState) -> AgentState:
+    state["messages"] = [
+        SystemMessage(content=HELP_PROMPT),
+        HumanMessage(content=state["question"]),
+    ]
+    state["steps"] = 0
+    state["final"] = None
+    state["last_error"] = None
+    return state
+
+
+def call_model(state: AgentState) -> AgentState:
+    state["steps"] += 1
+    resp = llm.bind_tools(tools).invoke(state["messages"])
+    state["messages"].append(resp)
+    return state
+
+
+def maybe_finalize(state: AgentState) -> AgentState:
+    """If the model produced FINAL ANSWER, store it. Otherwise keep going."""
+    last = state["messages"][-1]
+    if isinstance(last, AIMessage):
+        final_line = extract_final(last.content if isinstance(last.content, str) else str(last.content))
+        if final_line:
+            state["final"] = final_line
+    return state
+
+
+def format_guard(state: AgentState) -> AgentState:
+    """If we hit step limit and still no FINAL ANSWER, force one."""
+    if state["final"] is None:
+        # Ask model to rewrite into the required format only
+        state["messages"].append(
+            HumanMessage(
+                content="Rewrite your response to follow the required format exactly. "
+                        "Return only one line: FINAL ANSWER: ...")
         )
-        fixed_answer = self.smart_agent.run(help_prompt + question)
+    return state
 
+
+# -----------------------------
+# Router: decide next step
+# -----------------------------
+def route(state: AgentState) -> Literal["tools", "finalize", "guard", "end"]:
+    # 1. First, check if the model wants to call tools.
+    # We MUST execute tools if requested, otherwise we break the conversation chain.
+    last = state["messages"][-1]
+    if isinstance(last, AIMessage) and getattr(last, "tool_calls", None):
+        return "tools"
+
+    # 2. If no tools, check if we are done.
+    if state["final"] is not None:
+        return "end"
+
+    # 3. TIME LIMIT CHECK
+    if state["steps"] >= MAX_STEPS:
+        # CHECK FOR DEATH LOOP:
+        # Look at the message before the last one. Was it our "Rewrite" prompt?
+        # If yes, we already tried to guard and it failed. Don't try again.
+        messages = state["messages"]
+        if len(messages) >= 2:
+            second_to_last = messages[-2]
+            if isinstance(second_to_last, HumanMessage) and "Rewrite your response" in str(second_to_last.content):
+                # We tried, we failed. Just give up to save the recursion limit.
+                return "end"
         
-        print(f"Agent returning fixed answer: {fixed_answer}")
-        
-        return fixed_answer
+        # Otherwise, try the guard rail once.
+        return "guard"
+    
+    # 4. Default loop
+    return "finalize"
+
+
+# -----------------------------
+# Build graph
+# -----------------------------
+graph = StateGraph(AgentState)
+
+graph.add_node("start", start)
+graph.add_node("model", call_model)
+graph.add_node("tools", tool_node)
+graph.add_node("finalize", maybe_finalize)
+graph.add_node("guard", format_guard)
+
+graph.set_entry_point("start")
+graph.add_edge("start", "model")
+graph.add_edge("model", "finalize")
+
+graph.add_conditional_edges(
+    "finalize",
+    route,
+    {
+        "tools": "tools",
+        "finalize": "model",
+        "guard": "guard",
+        "end": END,
+    },
+)
+
+graph.add_edge("tools", "model")
+graph.add_edge("guard", "model")
+
+app = graph.compile()
+
+
+
+# -----------------------------
+# Public callable (like your BasicAgent)
+# -----------------------------
+class BasicAgentLangGraph:
+    def __init__(self):
+        print("BasicAgentLangGraph initialized.")
+
+    def __call__(self, question: str) -> str:
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        state: AgentState = {
+            "question": question,
+            "messages": [],
+            "final": None,
+            "steps": 0,
+            "last_error": None,
+        }
+        out = app.invoke(state)
+        # If still none, fallback
+        return out["final"] or "FINAL ANSWER: not available"
 
 
 
@@ -75,7 +229,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
 
     # 1. Instantiate Agent ( modify this part to create your agent)
     try:
-        agent = BasicAgent()
+        agent = BasicAgentLangGraph()
     except Exception as e:
         print(f"Error instantiating agent: {e}")
         return f"Error initializing agent: {e}", None
@@ -208,6 +362,9 @@ with gr.Blocks() as demo:
 
 if __name__ == "__main__":
     print("\n" + "-"*30 + " App Starting " + "-"*30)
+
+
+
     # Check for SPACE_HOST and SPACE_ID at startup for information
     space_host_startup = os.getenv("SPACE_HOST")
     space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup

tool.py

@@ -1,123 +1,155 @@
-from smolagents import DuckDuckGoSearchTool
 
-# Initialize the DuckDuckGo search tool
-web_search_tool = DuckDuckGoSearchTool()
+from __future__ import annotations
 
-from smolagents import Tool
-import json
 import os
-
-# PDF support
-import pdfplumber
-
-# CSV support
+import json
 import csv
+from typing import Optional
 
-
-class FileReader(Tool):
-    name = "file_reader"
-    description = (
-        "Read local files and return extracted text. "
-        "Supports PDF, JSON, TXT, and CSV."
-    )
-
-    inputs = {
-        "path": {
-            "type": "string",
-            "description": "Path to the file on disk"
-        }
-    }
-
-    output_type = "string"
-
-    def forward(self, path: str) -> str:
-        if not os.path.exists(path):
-            return f"Error: file not found at {path}"
-
-        ext = os.path.splitext(path)[1].lower()
-
-        try:
-            if ext == ".pdf":
-                return self._read_pdf(path)
-
-            elif ext == ".json":
-                return self._read_json(path)
-
-            elif ext == ".txt":
-                return self._read_txt(path)
-
-            elif ext == ".csv":
-                return self._read_csv(path)
-
-            else:
-                return f"Unsupported file type: {ext}"
-
-        except Exception as e:
-            return f"Error reading file: {str(e)}"
-
-    def _read_pdf(self, path: str) -> str:
-        text = []
-        with pdfplumber.open(path) as pdf:
-            for page in pdf.pages:
-                page_text = page.extract_text()
-                if page_text:
-                    text.append(page_text)
-        return "\n".join(text)
-
-    def _read_json(self, path: str) -> str:
-        with open(path, "r", encoding="utf-8") as f:
-            data = json.load(f)
-        return json.dumps(data, indent=2)
-
-    def _read_txt(self, path: str) -> str:
-        with open(path, "r", encoding="utf-8") as f:
-            return f.read()
-
-    def _read_csv(self, path: str) -> str:
-        rows = []
-        with open(path, newline="", encoding="utf-8") as f:
-            reader = csv.reader(f)
-            for row in reader:
-                rows.append(", ".join(row))
-        return "\n".join(rows)
-
-
-file_reader_tool = FileReader()
-
-
-from smolagents import Tool
+import pdfplumber
 import httpx
 from bs4 import BeautifulSoup
 
-class WebFetch(Tool):
-    name = "web_fetch"
-    description = "Fetch and read webpage content from a URL."
-    inputs = {
-        "url": {
-            "type": "string",
-            "description": "URL of the webpage to read"
-        }
-    }
-    output_type = "string"
-
-    def forward(self, url: str) -> str:
-        try:
-            with httpx.Client(follow_redirects=True, timeout=20) as client:
-                r = client.get(url)
-                r.raise_for_status()
-
-            soup = BeautifulSoup(r.text, "html.parser")
-
-            # Remove scripts, styles, nav, footer
-            for tag in soup(["script", "style", "noscript", "nav", "footer", "header", "aside"]):
-                tag.decompose()
-
-            text = soup.get_text(separator="\n")
-            lines = [line.strip() for line in text.splitlines() if line.strip()]
-
-            return "\n".join(lines[:5000])  # cap length for LLM
-
-        except Exception as e:
-            return f"Error fetching page: {str(e)}"
-
-web_fetch_tool = WebFetch()
+from langchain_core.tools import tool
+from langchain_community.tools import DuckDuckGoSearchRun
+
+
+# -------------------------
+# 1) DuckDuckGo search tool
+# -------------------------
+_ddg = DuckDuckGoSearchRun()
+
+@tool("web_search")
+def web_search(query: str) -> str:
+    """Search the web (DuckDuckGo) and return text results."""
+    # DuckDuckGoSearchRun returns a string summary of results
+    return _ddg.run(query)
+
+
+# # -------------------------
+# # 2) Local file reader tool
+# # -------------------------
+# def _read_pdf(path: str) -> str:
+#     text = []
+#     with pdfplumber.open(path) as pdf:
+#         for page in pdf.pages:
+#             page_text = page.extract_text()
+#             if page_text:
+#                 text.append(page_text)
+#     return "\n".join(text)
+
+# def _read_json(path: str) -> str:
+#     with open(path, "r", encoding="utf-8") as f:
+#         data = json.load(f)
+#     return json.dumps(data, indent=2, ensure_ascii=False)
+
+# def _read_txt(path: str) -> str:
+#     with open(path, "r", encoding="utf-8") as f:
+#         return f.read()
+
+# def _read_csv(path: str) -> str:
+#     rows = []
+#     with open(path, newline="", encoding="utf-8") as f:
+#         reader = csv.reader(f)
+#         for row in reader:
+#             rows.append(", ".join(row))
+#     return "\n".join(rows)
+
+# @tool("file_reader")
+# def file_reader(path: str) -> str:
+#     """
+#     Read local files and return extracted text.
+#     Supports PDF, JSON, TXT, and CSV.
+#     """
+#     if not os.path.exists(path):
+#         return f"Error: file not found at {path}"
+
+#     ext = os.path.splitext(path)[1].lower()
+
+#     try:
+#         if ext == ".pdf":
+#             return _read_pdf(path)
+#         if ext == ".json":
+#             return _read_json(path)
+#         if ext == ".txt":
+#             return _read_txt(path)
+#         if ext == ".csv":
+#             return _read_csv(path)
+#         return f"Unsupported file type: {ext}"
+#     except Exception as e:
+#         return f"Error reading file: {e}"
+
+
+# -------------------------
+# 3) Web fetch tool
+# -------------------------
+def _clean_html_to_text(html: str, max_lines: int = 5000) -> str:
+    soup = BeautifulSoup(html, "html.parser")
+
+    # Remove noisy tags
+    for tag in soup(["script", "style", "noscript", "nav", "footer", "header", "aside"]):
+        tag.decompose()
+
+    text = soup.get_text(separator="\n")
+    lines = [line.strip() for line in text.splitlines() if line.strip()]
+    return "\n".join(lines[:max_lines])
+
+@tool("web_fetch")
+def web_fetch(url: str) -> str:
+    """
+    Retrieves and reads the text content of a specific URL. 
+    
+    Use this to read articles, documentation, or static webpages.
+    
+    Do NOT use this tool for YouTube URLs (use 'youtube_transcript' instead).
+    Limitations:
+    - Returns cleaned plain text, not raw HTML.
+    - Cannot execute JavaScript (may fail on heavy SPAs or dynamic sites).
+    - Content is truncated at 5000 lines.
+    """
+    try:
+        with httpx.Client(follow_redirects=True, timeout=20) as client:
+            r = client.get(
+                url,
+                headers={
+                    # Some sites block empty UA; this helps
+                    "User-Agent": "Mozilla/5.0 (compatible; LangChainTool/1.0)"
+                },
+            )
+            r.raise_for_status()
+
+        return _clean_html_to_text(r.text, max_lines=5000)
+    except Exception as e:
+        return f"Error fetching page: {e}"
+
+from langchain_core.tools import tool
+from youtube_transcript_api import YouTubeTranscriptApi
+
+def _extract_video_id(url: str) -> str:
+    # handles https://www.youtube.com/watch?v=VIDEOID
+    import urllib.parse as up
+    q = up.urlparse(url)
+    if q.hostname in ("www.youtube.com", "youtube.com"):
+        return up.parse_qs(q.query).get("v", [""])[0]
+    if q.hostname == "youtu.be":
+        return q.path.lstrip("/")
+    return ""
+
+@tool("youtube_transcript")
+def youtube_transcript(url: str) -> str:
+    """
+    Retrieves the full English transcript text from a YouTube video URL.
+    
+    Use this tool when a user asks questions about a video's content, wants a summary, 
+    or needs specific quotes. 
+    
+    Note: This tool only supports videos with English captions/subtitles.
+    """
+    vid = _extract_video_id(url)
+    if not vid:
+        return "Error: could not parse video id"
+    try:
+        chunks = YouTubeTranscriptApi.get_transcript(vid, languages=["en"])
+        return "\n".join([c["text"] for c in chunks])
+    except Exception as e:
+        return f"Error fetching transcript: {e}"