Spaces:
Build error
Build error
| # app.py (Final Version) | |
| import os | |
| import re | |
| import gradio as gr | |
| import requests | |
| import pandas as pd | |
| import logging | |
| import numexpr | |
| from typing import TypedDict, Annotated | |
| # --- Langchain & HF Imports --- | |
| from langchain_huggingface import HuggingFaceEndpoint | |
| from langchain_community.tools import DuckDuckGoSearchRun | |
| from langchain_core.prompts import PromptTemplate | |
| from langchain_core.output_parsers import StrOutputParser | |
| from langchain_core.tools import tool | |
| from langgraph.graph import StateGraph, END | |
| from langchain_community.document_loaders.youtube import YoutubeLoader | |
| from transformers.pipelines import pipeline as hf_pipeline # Renamed to avoid conflict | |
| # --- Constants --- | |
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
| SYSTEM_PROMPT = """You are a helpful and expert assistant named GAIA, designed to answer questions accurately. To do this, you have access to a set of tools. Based on the user's question, you must decide which tool to use, if any. Your process is: | |
| 1. **Analyze the Question**: Understand what is being asked. | |
| 2. **Select a Tool**: If necessary, choose the best tool. Your available tools are: `web_search`, `math_calculator`, `image_analyzer`, `youtube_transcript_reader`. | |
| 3. **Call the Tool**: Output a tool call in the format `tool_name("argument")`. For example: `web_search("what is the weather in Paris?")`. | |
| 4. **Analyze the Result**: Look at the tool's output. | |
| 5. **Final Answer**: If you have enough information, provide the final answer. If not, you can use another tool. | |
| When you have the final answer, you **must** output it in the following format, and nothing else: | |
| FINAL ANSWER: [YOUR FINAL ANSWER]""" | |
| # --- Tool Definitions --- | |
| image_to_text_pipeline = None | |
| def web_search(query: str) -> str: | |
| """Searches the web using DuckDuckGo for up-to-date information.""" | |
| logging.info(f"--- Calling Web Search Tool with query: {query} ---") | |
| search = DuckDuckGoSearchRun() | |
| return search.run(query) | |
| def math_calculator(expression: str) -> str: | |
| """Calculates the result of a mathematical expression.""" | |
| logging.info(f"--- Calling Math Calculator Tool with expression: {expression} ---") | |
| try: | |
| if not re.match(r"^[0-9\.\+\-\*\/\(\)\s]+$", expression): | |
| return "Error: Invalid characters in expression." | |
| result = numexpr.evaluate(expression).item() | |
| return str(result) | |
| except Exception as e: | |
| logging.error(f"Calculator error: {e}") | |
| return f"Error: {e}" | |
| def image_analyzer(image_url: str) -> str: | |
| """Analyzes an image from a URL and returns a text description.""" | |
| global image_to_text_pipeline | |
| logging.info(f"--- Calling Image Analyzer Tool with URL: {image_url} ---") | |
| try: | |
| if image_to_text_pipeline is None: | |
| logging.info( | |
| "--- Initializing Image Analyzer pipeline (lazy loading)... ---" | |
| ) | |
| image_to_text_pipeline = hf_pipeline( | |
| "image-to-text", model="Salesforce/blip-image-captioning-base" | |
| ) | |
| logging.info("--- Image Analyzer pipeline initialized. ---") | |
| pipeline_output = image_to_text_pipeline(image_url) | |
| if ( | |
| pipeline_output | |
| and isinstance(pipeline_output, list) | |
| and len(pipeline_output) > 0 | |
| ): | |
| description = pipeline_output[0].get( | |
| "generated_text", "Error: Could not generate text." | |
| ) | |
| else: | |
| description = "Error: Could not analyze image." | |
| return description | |
| except Exception as e: | |
| logging.error(f"Error analyzing image: {e}") | |
| return f"Error analyzing image: {e}" | |
| def youtube_transcript_reader(youtube_url: str) -> str: | |
| """Reads the transcript of a YouTube video from its URL.""" | |
| logging.info( | |
| f"--- Calling YouTube Transcript Reader Tool with URL: {youtube_url} ---" | |
| ) | |
| try: | |
| loader = YoutubeLoader.from_youtube_url(youtube_url, add_video_info=False) | |
| docs = loader.load() | |
| transcript = " ".join([doc.page_content for doc in docs]) | |
| return transcript[:4000] | |
| except Exception as e: | |
| logging.error(f"Error reading YouTube transcript: {e}") | |
| return f"Error: {e}" | |
| class AgentState(TypedDict): | |
| question: str | |
| messages: Annotated[list, lambda x, y: x + y] | |
| sender: str | |
| class GaiaAgent: | |
| def __init__(self): | |
| logging.info("Initializing GaiaAgent...") | |
| self.tools = [ | |
| web_search, | |
| math_calculator, | |
| image_analyzer, | |
| youtube_transcript_reader, | |
| ] | |
| # --- THIS IS THE CORRECTED LLM INITIALIZATION --- | |
| logging.info("Initializing LLM via modern HuggingFaceEndpoint...") | |
| llm = HuggingFaceEndpoint( | |
| repo_id="HuggingFaceH4/zephyr-7b-beta", | |
| temperature=0.1, | |
| max_new_tokens=1024, | |
| huggingfacehub_api_token=os.getenv("HUGGINGFACEHUB_API_TOKEN"), | |
| ) | |
| logging.info("LLM initialized successfully.") | |
| # The rest of the class remains the same | |
| prompt = PromptTemplate( | |
| template=SYSTEM_PROMPT | |
| + "\nHere is the current conversation:\n{messages}\n\nQuestion: {question}", | |
| input_variables=["messages", "question"], | |
| ) | |
| self.agent = prompt | llm | StrOutputParser() | |
| self.graph = self._create_graph() | |
| logging.info("GaiaAgent initialized successfully.") | |
| def _create_graph(self): | |
| graph = StateGraph(AgentState) | |
| graph.add_node("agent", self._call_agent) | |
| graph.add_node("tools", self._call_tools) | |
| graph.add_conditional_edges( | |
| "agent", self._decide_action, {END: END, "tools": "tools"} | |
| ) | |
| graph.add_edge("tools", "agent") | |
| graph.set_entry_point("agent") | |
| return graph.compile() | |
| def _call_agent(self, state: AgentState): | |
| logging.info("--- Calling Agent ---") | |
| message_history = "\n".join(state["messages"]) | |
| response = self.agent.invoke( | |
| {"messages": message_history, "question": state["question"]} | |
| ) | |
| return {"messages": [response], "sender": "agent"} | |
| def _decide_action(self, state: AgentState): | |
| logging.info("--- Deciding Action ---") | |
| response = state["messages"][-1] | |
| if "FINAL ANSWER:" in response: | |
| return END | |
| else: | |
| return "tools" | |
| def _call_tools(self, state: AgentState): | |
| logging.info("--- Calling Tools ---") | |
| raw_tool_call = state["messages"][-1] | |
| tool_call_match = re.search(r"(\w+)\s*\((.*?)\)", raw_tool_call, re.DOTALL) | |
| if not tool_call_match: | |
| logging.warning("No valid tool call found in agent response.") | |
| return { | |
| "messages": [ | |
| 'No valid tool call found. Please format your response as `tool_name("argument")` or provide a `FINAL ANSWER:`.' | |
| ], | |
| "sender": "tools", | |
| } | |
| tool_name = tool_call_match.group(1).strip() | |
| tool_input_str = tool_call_match.group(2).strip() | |
| if (tool_input_str.startswith('"') and tool_input_str.endswith('"')) or ( | |
| tool_input_str.startswith("'") and tool_input_str.endswith("'") | |
| ): | |
| tool_input = tool_input_str[1:-1] | |
| else: | |
| tool_input = tool_input_str | |
| tool_to_call = next((t for t in self.tools if t.name == tool_name), None) | |
| if tool_to_call: | |
| try: | |
| result = tool_to_call.run(tool_input) | |
| return {"messages": [str(result)], "sender": "tools"} | |
| except Exception as e: | |
| logging.error(f"Error executing tool {tool_name}: {e}") | |
| return { | |
| "messages": [f"Error executing tool {tool_name}: {e}"], | |
| "sender": "tools", | |
| } | |
| else: | |
| logging.warning(f"Tool '{tool_name}' not found.") | |
| return {"messages": [f"Tool '{tool_name}' not found."], "sender": "tools"} | |
| def __call__(self, question: str) -> str: | |
| logging.info(f"Agent received question: {question[:100]}...") | |
| try: | |
| initial_state = {"question": question, "messages": [], "sender": "user"} | |
| final_state = self.graph.invoke(initial_state, {"recursion_limit": 15}) | |
| final_response = final_state["messages"][-1] | |
| match = re.search( | |
| r"FINAL ANSWER:\s*(.*)", final_response, re.IGNORECASE | re.DOTALL | |
| ) | |
| if match: | |
| extracted_answer = match.group(1).strip() | |
| logging.info(f"Agent returning final answer: {extracted_answer}") | |
| return extracted_answer | |
| else: | |
| logging.warning( | |
| "Agent could not find a final answer. Returning the last message." | |
| ) | |
| return final_response | |
| except Exception as e: | |
| logging.error(f"Error during agent invocation: {e}", exc_info=True) | |
| return f"Error during agent invocation: {e}" | |
| # In app.py | |
| # ... (keep all the code above this function) | |
| def run_and_submit_all(profile: gr.OAuthProfile | None): | |
| if not profile: | |
| return "Please Login to Hugging Face with the button.", None | |
| username = profile.username | |
| logging.info(f"User logged in: {username}") | |
| space_id = os.getenv("SPACE_ID") | |
| if not space_id: | |
| space_id = "leofltt/HF_Agents_Final_Assignment" | |
| logging.warning(f"SPACE_ID not found, using fallback for local run: {space_id}") | |
| if not space_id: | |
| return "CRITICAL ERROR: SPACE_ID environment variable is not set.", None | |
| api_url = DEFAULT_API_URL | |
| questions_url = f"{api_url}/questions" | |
| submit_url = f"{api_url}/submit" | |
| try: | |
| agent = GaiaAgent() | |
| except Exception as e: | |
| logging.critical(f"Fatal error instantiating agent: {e}", exc_info=True) | |
| return f"Fatal error initializing agent: {e}", None | |
| agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" | |
| logging.info(f"Fetching questions from: {questions_url}") | |
| try: | |
| response = requests.get(questions_url, timeout=20) | |
| response.raise_for_status() | |
| questions_data = response.json() | |
| if not questions_data: | |
| return "Fetched questions list is empty.", None | |
| logging.info(f"Successfully fetched {len(questions_data)} questions.") | |
| except Exception as e: | |
| return f"Error fetching questions: {e}", None | |
| # --- MODIFICATION FOR DEBUGGING --- | |
| # We will only process the first question from the list. | |
| questions_to_process = [questions_data[0]] | |
| logging.info( | |
| f"DEBUG MODE: Processing only the first question out of {len(questions_data)}." | |
| ) | |
| # --- END OF MODIFICATION --- | |
| results_log = [] | |
| answers_payload = [] | |
| # The loop now runs only once. | |
| for i, item in enumerate(questions_to_process): | |
| task_id = item.get("task_id") | |
| question_text = item.get("question") | |
| logging.info(f"--- Processing question (Task ID: {task_id}) ---") | |
| if not task_id or question_text is None: | |
| continue | |
| try: | |
| 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: | |
| logging.error(f"Error running agent on task {task_id}: {e}", exc_info=True) | |
| results_log.append( | |
| { | |
| "Task ID": task_id, | |
| "Question": question_text, | |
| "Submitted Answer": f"AGENT ERROR: {e}", | |
| } | |
| ) | |
| # Also return the error in the status for immediate feedback | |
| return f"Agent failed on the first question with error: {e}", pd.DataFrame( | |
| results_log | |
| ) | |
| if not answers_payload: | |
| return "Agent did not produce an answer for the first question.", pd.DataFrame( | |
| results_log | |
| ) | |
| submission_data = { | |
| "username": username.strip(), | |
| "agent_code": agent_code, | |
| "answers": answers_payload, | |
| } | |
| logging.info(f"Submitting {len(answers_payload)} answer for user '{username}'...") | |
| try: | |
| response = requests.post(submit_url, json=submission_data, timeout=60) | |
| response.raise_for_status() | |
| result_data = response.json() | |
| final_status = ( | |
| f"Submission Successful (for one question)!\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.')}" | |
| ) | |
| return final_status, pd.DataFrame(results_log) | |
| except requests.exceptions.HTTPError as e: | |
| error_detail = f"Server responded with status {e.response.status_code}. Detail: {e.response.text}" | |
| return f"Submission Failed: {error_detail}", pd.DataFrame(results_log) | |
| except Exception as e: | |
| return f"An unexpected error occurred during submission: {e}", pd.DataFrame( | |
| results_log | |
| ) | |
| with gr.Blocks() as demo: | |
| gr.Markdown("# GAIA Agent Evaluation Runner") | |
| gr.Markdown( | |
| "This agent uses LangGraph and Mistral-7B to answer questions from the GAIA benchmark." | |
| ) | |
| gr.LoginButton() | |
| run_button = gr.Button("Run Evaluation & Submit All Answers") | |
| status_output = gr.Textbox( | |
| label="Run Status / Submission Result", lines=5, interactive=False | |
| ) | |
| results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True) | |
| run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table]) | |
| if __name__ == "__main__": | |
| logging.basicConfig( | |
| level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" | |
| ) | |
| logging.info("App Starting (Final Version)...") | |
| demo.launch() | |