Spaces:
Running
Running
| # app.py (New LangChain version) | |
| import os | |
| import gradio as gr | |
| import pandas as pd | |
| from bs4 import BeautifulSoup | |
| import datetime | |
| import pytz | |
| import math | |
| import re | |
| import requests | |
| import traceback | |
| import sys | |
| # --- LangChain and new Transformers imports --- | |
| from langchain.agents import AgentExecutor, create_react_agent | |
| from langchain_huggingface import HuggingFacePipeline | |
| from langchain_core.prompts import PromptTemplate | |
| from langchain.tools import Tool | |
| from langchain_community.tools import DuckDuckGoSearchRun | |
| # --- Other imports --- | |
| import transformers # Still useful for version checking | |
| print(f"--- Using transformers version: {transformers.__version__} ---") | |
| # --- Constants --- | |
| DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
| # --- Tool Definitions (LangChain Style) --- | |
| # For LangChain, we define the functions and then wrap them in LangChain's Tool class. | |
| def get_current_time_in_timezone_func(timezone: str) -> str: | |
| """A tool that fetches the current local time in a specified IANA timezone. Always use this tool for questions about the current time. Input should be a valid timezone string (e.g., 'America/New_York', 'Europe/London').""" | |
| print(f"--- Tool: Executing get_current_time_in_timezone for: {timezone} ---") | |
| try: | |
| tz = pytz.timezone(timezone) | |
| local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S %Z%z") | |
| return f"The current local time in {timezone} is: {local_time}" | |
| except pytz.exceptions.UnknownTimeZoneError: | |
| return f"Error: Unknown timezone '{timezone}'. Please use a valid IANA timezone name." | |
| except Exception as e: | |
| return f"Error fetching time for timezone '{timezone}': {str(e)}" | |
| # Using the DuckDuckGoSearchRun tool from LangChain for stability | |
| # The description is very important for the agent to know when to use it. | |
| search_tool = DuckDuckGoSearchRun( | |
| name="web_search", | |
| description="A tool that performs a web search using DuckDuckGo. Use this to find up-to-date information about events, facts, or topics when the answer isn't already known." | |
| ) | |
| def safe_calculator_func(expression: str) -> str: | |
| """A tool for evaluating simple mathematical expressions. Use this tool *only* for calculations involving numbers, +, -, *, /, %, parentheses, and the math functions: sqrt, pow. Do not use it to run other code.""" | |
| print(f"--- Tool: Executing safe_calculator with expression: {expression} ---") | |
| try: | |
| # Using a more restricted eval context for safety | |
| allowed_names = {"sqrt": math.sqrt, "pow": math.pow, "pi": math.pi} | |
| result = eval(expression, {"__builtins__": {}}, allowed_names) | |
| return str(result) | |
| except Exception as e: | |
| print(f"Error during calculation for '{expression}': {e}") | |
| return f"Error calculating '{expression}': Invalid expression or calculation error ({e})." | |
| # --- LangChain Agent Definition --- | |
| class LangChainAgentWrapper: | |
| def __init__(self): | |
| print("Initializing LangChainAgentWrapper...") | |
| # Using a newer, more capable instruction-tuned model. | |
| # This model is generally better at following the ReAct prompt format used by LangChain agents. | |
| model_id = "google/gemma-2b-it" | |
| # model_id = "bigcode/starcoderbase-1b" # You can still use starcoder if you prefer | |
| try: | |
| hf_auth_token = os.getenv("HF_TOKEN") | |
| if not hf_auth_token: | |
| raise ValueError("HF_TOKEN secret is missing. It is required for downloading models.") | |
| else: | |
| print("HF_TOKEN secret found.") | |
| # Create the Hugging Face pipeline | |
| print(f"Loading model pipeline for: {model_id}") | |
| llm_pipeline = transformers.pipeline( | |
| "text-generation", | |
| model=model_id, | |
| model_kwargs={"torch_dtype": "auto"}, # Use "auto" for dtype | |
| device_map="auto", # Requires accelerate | |
| token=hf_auth_token, | |
| ) | |
| print("Model pipeline loaded successfully.") | |
| # Wrap the pipeline in a LangChain LLM object | |
| self.llm = HuggingFacePipeline(pipeline=llm_pipeline) | |
| # Define the list of LangChain tools | |
| self.tools = [ | |
| Tool( | |
| name="get_current_time_in_timezone", | |
| func=get_current_time_in_timezone_func, | |
| description=get_current_time_in_timezone_func.__doc__ | |
| ), | |
| search_tool, # This is already a LangChain Tool instance | |
| Tool( | |
| name="safe_calculator", | |
| func=safe_calculator_func, | |
| description=safe_calculator_func.__doc__ | |
| ), | |
| ] | |
| print(f"Tools prepared for agent: {[tool.name for tool in self.tools]}") | |
| # Create the ReAct agent prompt from a template | |
| # The prompt is crucial for teaching the agent how to think and use tools. | |
| react_prompt = PromptTemplate.from_template( | |
| """ | |
| You are a helpful assistant. Answer the following questions as best you can. | |
| You have access to the following tools: | |
| {tools} | |
| Use the following format: | |
| Question: the input question you must answer | |
| Thought: you should always think about what to do | |
| Action: the action to take, should be one of [{tool_names}] | |
| Action Input: the input to the action | |
| Observation: the result of the action | |
| ... (this Thought/Action/Action Input/Observation can repeat N times) | |
| Thought: I now know the final answer | |
| Final Answer: the final answer to the original input question | |
| Begin! | |
| Question: {input} | |
| Thought:{agent_scratchpad} | |
| """ | |
| ) | |
| # Create the agent | |
| agent = create_react_agent(self.llm, self.tools, react_prompt) | |
| # Create the agent executor, which runs the agent loop | |
| self.agent_executor = AgentExecutor(agent=agent, tools=self.tools, verbose=True, handle_parsing_errors=True) | |
| print("LangChain agent created successfully.") | |
| except Exception as e: | |
| print(f"CRITICAL ERROR: Failed to initialize LangChain agent: {e}") | |
| traceback.print_exc() | |
| raise RuntimeError(f"LangChain agent initialization failed: {e}") from e | |
| def __call__(self, question: str) -> str: | |
| print(f"\n--- LangChainAgentWrapper received question: {question[:100]}... ---") | |
| try: | |
| # Invoke the agent executor | |
| response = self.agent_executor.invoke({"input": question}) | |
| # The answer is in the 'output' key of the response dictionary | |
| return response.get("output", "No output found.") | |
| except Exception as e: | |
| print(f"ERROR: LangChain agent execution failed: {e}") | |
| traceback.print_exc() | |
| return f"Agent Error: Failed to process the question. Details: {e}" | |
| # --- Main Evaluation Logic --- | |
| def run_and_submit_all(profile: gr.OAuthProfile | None): | |
| """ | |
| Fetches all questions, runs the agent on them, submits all answers, | |
| and displays the results. | |
| """ | |
| space_id = os.getenv("SPACE_ID") | |
| if profile: | |
| username= f"{profile.username}" | |
| print(f"User logged in: {username}") | |
| else: | |
| print("User not logged in.") | |
| return "Please Login to Hugging Face with the button.", None | |
| api_url = DEFAULT_API_URL | |
| questions_url = f"{api_url}/questions" | |
| submit_url = f"{api_url}/submit" | |
| try: | |
| # Now instantiate our new LangChain agent | |
| agent = LangChainAgentWrapper() | |
| except Exception as e: | |
| print(f"Error instantiating agent: {e}") | |
| return f"Error initializing agent: {e}", None | |
| agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" | |
| print(agent_code) | |
| print(f"Fetching questions from: {questions_url}") | |
| try: | |
| response = requests.get(questions_url, timeout=15) | |
| response.raise_for_status() | |
| questions_data = response.json() | |
| if not questions_data: | |
| print("Fetched questions list is empty.") | |
| return "Fetched questions list is empty or invalid format.", None | |
| print(f"Fetched {len(questions_data)} questions.") | |
| except Exception as e: | |
| print(f"An unexpected error occurred fetching questions: {e}") | |
| return f"An unexpected error occurred fetching questions: {e}", None | |
| results_log = [] | |
| answers_payload = [] | |
| print(f"Running agent on {len(questions_data)} questions...") | |
| for item in questions_data: | |
| task_id = item.get("task_id") | |
| question_text = item.get("question") | |
| if not task_id or question_text is None: | |
| print(f"Skipping item with missing task_id or question: {item}") | |
| 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: | |
| print(f"Error running agent on task {task_id}: {e}") | |
| results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"}) | |
| if not answers_payload: | |
| print("Agent did not produce any answers to submit.") | |
| return "Agent did not produce any answers to submit.", pd.DataFrame(results_log) | |
| submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload} | |
| status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..." | |
| print(status_update) | |
| print(f"Submitting {len(answers_payload)} answers to: {submit_url}") | |
| try: | |
| response = requests.post(submit_url, json=submission_data, timeout=60) | |
| response.raise_for_status() | |
| result_data = response.json() | |
| final_status = ( | |
| f"Submission Successful!\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.')}" | |
| ) | |
| print("Submission successful.") | |
| results_df = pd.DataFrame(results_log) | |
| return final_status, results_df | |
| except Exception as e: | |
| status_message = f"An unexpected error occurred during submission: {e}" | |
| print(status_message) | |
| traceback.print_exc() | |
| results_df = pd.DataFrame(results_log) | |
| return status_message, results_df | |
| # --- Build Gradio Interface using Blocks --- | |
| with gr.Blocks() as demo: | |
| gr.Markdown("# Basic Agent Evaluation Runner") | |
| 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. | |
| """ | |
| ) | |
| 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__": | |
| print("\n" + "-"*30 + " App Starting " + "-"*30) | |
| space_host_startup = os.getenv("SPACE_HOST") | |
| space_id_startup = os.getenv("SPACE_ID") | |
| if space_host_startup: | |
| print(f"✅ SPACE_HOST found: {space_host_startup}") | |
| print(f" Runtime URL should be: https://{space_host_startup}.hf.space") | |
| else: | |
| print("ℹ️ SPACE_HOST environment variable not found (running locally?).") | |
| if space_id_startup: | |
| print(f"✅ SPACE_ID found: {space_id_startup}") | |
| print(f" Repo URL: https://huggingface.co/spaces/{space_id_startup}") | |
| print(f" Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main") | |
| else: | |
| print("ℹ️ SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.") | |
| print("-"*(60 + len(" App Starting ")) + "\n") | |
| print("Launching Gradio Interface for Basic Agent Evaluation...") | |
| demo.launch(debug=True, share=False) |