wt002's picture
Update app.py
26f3037 verified
raw
history blame
33.2 kB
import os
import gradio as gr
import requests
import inspect
import pandas as pd
#from smolagents import tool, Tool, CodeAgent, DuckDuckGoSearchTool, HfApiModel, VisitWebpageTool, SpeechToTextTool, FinalAnswerTool
from dotenv import load_dotenv
import heapq
from collections import Counter
import re
from io import BytesIO
from youtube_transcript_api import YouTubeTranscriptApi
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_community.document_loaders import WikipediaLoader
from langchain_community.utilities import WikipediaAPIWrapper
from langchain_community.document_loaders import ArxivLoader
# (Keep Constants as is)
# --- Constants ---
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
#Load environment variables
load_dotenv()
import os
import time
import json
from typing import TypedDict, List, Union, Any, Dict, Optional
# LangChain and LangGraph imports
from langchain.schema import HumanMessage, AIMessage, SystemMessage
from langchain.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, END
from langchain_community.llms import HuggingFacePipeline
# Corrected Tool import: Use 'tool' (lowercase)
from langchain_core.tools import BaseTool, tool
# Hugging Face local model imports
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
import torch
# Tool-specific imports
from duckduckgo_search import DDGS
import wikipedia
import arxiv
from transformers import pipeline as hf_pipeline # Renamed to avoid clash with main pipeline
from youtube_transcript_api import YouTubeTranscriptApi
# --- Helper function for python_execution tool ---
def indent_code(code: str, indent: str = " ") -> str:
"""Indents multi-line code for execution within a function."""
return "\n".join(indent + line for line in code.splitlines())
# --- Tool Definitions ---
@tool
def duckduckgo_search(query: str) -> str:
"""Search web using DuckDuckGo. Returns top 3 results."""
print(f"DEBUG: duckduckgo_search called with: {query}")
try:
with DDGS() as ddgs:
return "\n\n".join(
f"Title: {res['title']}\nURL: {res['href']}\nSnippet: {res['body']}"
for res in ddgs.text(query, max_results=3)
)
except Exception as e:
return f"Error performing DuckDuckGo search: {str(e)}"
@tool
def wikipedia_search(query: str) -> str:
"""Get Wikipedia summaries. Returns first 3 sentences."""
print(f"DEBUG: wikipedia_search called with: {query}")
try:
return wikipedia.summary(query, sentences=3)
except wikipedia.DisambiguationError as e:
return f"Disambiguation options: {', '.join(e.options[:3])}"
except wikipedia.PageError:
return "Wikipedia page not found."
except Exception as e:
return f"Error performing Wikipedia search: {str(e)}"
@tool
def arxiv_search(query: str) -> str:
"""Search academic papers on arXiv. Returns top 3 results."""
print(f"DEBUG: arxiv_search called with: {query}")
try:
results = arxiv.Search(
query=query,
max_results=3,
sort_by=arxiv.SortCriterion.Relevance
).results()
return "\n\n".join(
f"Title: {r.title}\nAuthors: {', '.join(a.name for a in r.authors)}\n"
f"Published: {r.published.strftime('%Y-%m-%d')}\nSummary: {r.summary[:250]}..."
for r in results
)
except Exception as e:
return f"Error performing ArXiv search: {str(e)}"
@tool
def document_qa(input_str: str) -> str:
"""Answer questions from documents. Input format: 'document_text||question'"""
print(f"DEBUG: document_qa called with: {input_str}")
try:
if '||' not in input_str:
return "Invalid format. Input must be: 'document_text||question'"
context, question = input_str.split('||', 1)
# Load QA model on first call or ensure it's loaded once globally.
# It's better to load once in __init__ for BasicAgent if possible,
# but this lazy loading prevents initial heavy load if tool is not used.
qa_model = hf_pipeline('question-answering', model='deepset/roberta-base-squad2')
return qa_model(question=question, context=context)['answer']
except Exception as e:
return f"Error answering question from document: {str(e)}"
@tool
def python_execution(code: str) -> str:
"""Execute Python code and return output.
The code should assign its final result to a variable named '_result_value'.
Example: '_result_value = 1 + 1'
"""
print(f"DEBUG: python_execution called with: {code}")
try:
# Create isolated environment
env = {}
# Wrap code in a function to isolate scope and capture '_result_value'
# The exec function is used carefully here. In a production environment,
# consider a more robust and secure sandbox (e.g., Docker, dedicated service).
exec(f"def __exec_fn__():\n{indent_code(code)}\n_result_value = __exec_fn__()", globals(), env)
return str(env.get('_result_value', 'No explicit result assigned to "_result_value" variable.'))
except Exception as e:
return f"Python execution error: {str(e)}"
class VideoTranscriptionTool(BaseTool):
name: str = "transcript_video"
# CORRECTED LINE BELOW: Added '=' for assignment
description: str = "Fetch text transcript from YouTube videos using URL or ID. Use for any question involving video or audio. Input is the YouTube URL or ID."
def _run(self, url_or_id: str) -> str:
print(f"DEBUG: transcript_video called with: {url_or_id}")
video_id = None
# Basic parsing for common YouTube URL formats
if "youtube.com/watch?v=" in url_or_id:
video_id = url_or_id.split("v=")[1].split("&")[0]
elif "youtu.be/" in url_or_id:
video_id = url_or_id.split("youtu.be/")[1].split("?")[0]
elif len(url_or_id.strip()) == 11 and not ("http://" in url_or_id or "https://" in url_or_id):
video_id = url_or_id.strip() # Assume it's just the ID
if not video_id:
return f"Invalid or unsupported YouTube URL/ID: {url_or_id}. Please provide a valid YouTube URL or 11-character ID."
try:
transcription = YouTubeTranscriptApi.get_transcript(video_id)
return " ".join([part['text'] for part in transcription])
except Exception as e:
return f"Error fetching transcript for video ID '{video_id}': {str(e)}. It might not have an English transcript, or the video is unavailable."
def _arun(self, *args, **kwargs):
raise NotImplementedError("Async not supported for this tool.")
# --- Agent State Definition ---
class AgentState(TypedDict):
question: str
history: List[Union[HumanMessage, AIMessage]]
context: Dict[str, Any]
reasoning: str
iterations: int
final_answer: Union[str, float, int, None]
current_task: str
current_thoughts: str
tools: List[BaseTool] # Make sure tools are passed via state, using BaseTool type
# --- Utility Functions ---
def parse_agent_response(response_content: str) -> tuple[str, str, str]:
"""
Parses the LLM's JSON output for reasoning, action, and action input.
Returns (reasoning, action, action_input).
If JSON parsing fails, it attempts heuristic parsing.
"""
try:
response_json = json.loads(response_content)
reasoning = response_json.get("Reasoning", "").strip()
action = response_json.get("Action", "").strip()
action_input = response_json.get("Action Input", "").strip()
return reasoning, action, action_input
except json.JSONDecodeError:
print(f"WARNING: JSONDecodeError: LLM response was not valid JSON. Attempting heuristic parse: {response_content[:200]}...")
# Heuristic parsing for non-JSON or partial JSON responses
reasoning = ""
action = ""
action_input = ""
# Attempt to find Reasoning
reasoning_idx = response_content.find("Reasoning:")
action_idx = response_content.find("Action:")
if reasoning_idx != -1 and action_idx != -1 and reasoning_idx < action_idx:
reasoning = response_content[reasoning_idx + len("Reasoning:"):action_idx].strip()
if reasoning.startswith('"') and reasoning.endswith('"'):
reasoning = reasoning[1:-1]
elif reasoning_idx != -1:
reasoning = response_content[reasoning_idx + len("Reasoning:"):].strip()
if reasoning.startswith('"') and reasoning.endswith('"'):
reasoning = reasoning[1:-1]
# Attempt to find Action and Action Input
if action_idx != -1:
action_input_idx = response_content.find("Action Input:", action_idx)
if action_input_idx != -1:
action_part = response_content[action_idx + len("Action:"):action_input_idx].strip()
action = action_part
action_input = response_content[action_input_idx + len("Action Input:"):].strip()
else:
action = response_content[action_idx + len("Action:"):].strip()
if action.startswith('"') and action.endswith('"'):
action = action[1:-1]
if action_input.startswith('"') and action_input.endswith('"'):
action_input = action_input[1:-1]
# Final cleanup for any trailing JSON artifacts if heuristic grabs too much
action = action.split('"', 1)[0].strip()
action_input = action_input.split('"', 1)[0].strip()
return reasoning, action, action_input
# --- Graph Nodes ---
def should_continue(state: AgentState) -> str:
"""
Determines if the agent should continue reasoning, use a tool, or end.
"""
print(f"DEBUG: Entering should_continue. Current context: {state.get('context', {})}")
if state.get("final_answer") is not None:
print("DEBUG: should_continue -> END (Final Answer set in state)")
return "end"
if state.get("context", {}).get("pending_action"):
print("DEBUG: should_continue -> ACTION (Pending action in context)")
return "action"
print("DEBUG: should_continue -> REASON (Default to reasoning)")
return "reason"
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
# ====== DOCUMENT PROCESSING SETUP ======
def create_vector_store():
"""Create vector store with predefined documents"""
# Define the documents
documents = [
Document(page_content="The capital of France is Paris.", metadata={"source": "geography"}),
Document(page_content="Python is a popular programming language created by Guido van Rossum.", metadata={"source": "tech"}),
Document(page_content="The Eiffel Tower is located in Paris, France.", metadata={"source": "landmarks"}),
]
# Initialize embedding model
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
# Split documents into chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # Smaller chunks for better precision
chunk_overlap=100
)
chunks = text_splitter.split_documents(documents)
# Create in-memory vector store
return Chroma.from_documents(
documents=chunks,
embedding=embeddings
)
def reasoning_node(state: AgentState) -> AgentState:
"""
Node for the agent to analyze the question, determine next steps,
and select tools.
"""
print(f"DEBUG: Entering reasoning_node. Iteration: {state['iterations']}")
print(f"DEBUG: Current history length: {len(state.get('history', []))}")
state.setdefault("context", {})
state.setdefault("reasoning", "")
state.setdefault("iterations", 0)
state.setdefault("current_task", "Understand the question and plan the next step.")
state.setdefault("current_thoughts", "")
state["context"].pop("pending_action", None)
# --- Initialize local HuggingFacePipeline ---
# Using Mistral-7B-Instruct-v0.2 for better agent performance
model_name = "mistralai/Mistral-7B-Instruct-v0.2"
print(f"DEBUG: Loading local model: {model_name}...")
tokenizer = AutoTokenizer.from_pretrained(model_name)
# Load model with optimal settings for GPU if available, else CPU
# Use bfloat16 for GPUs that support it (NVIDIA Ampere architecture and newer)
# else float16 for older GPUs or float32 for CPU/fallback.
# device_map="auto" intelligently distributes the model across available devices.
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.bfloat16 if torch.cuda.is_available() else torch.float32,
device_map="auto"
)
# Create a transformers pipeline
pipe = pipeline(
"text-generation",
model=model,
tokenizer=tokenizer,
max_new_tokens=1024, # Increased max_new_tokens for potentially longer JSON
temperature=0.1, # Keep low for factual, tool-use tasks
do_sample=True, # Allow some sampling
top_p=0.9, # Nucleus sampling
repetition_penalty=1.1, # Avoid repetition
# device_map handled by model loading
)
llm = HuggingFacePipeline(pipeline=pipe)
# --- END LOCAL LLM INITIALIZATION ---
tool_descriptions = "\n".join([
f"- **{t.name}**: {t.description}" for t in state.get("tools", [])
])
# ====== RAG RETRIEVAL ======
# Initialize vector store if not present
if "vector_store" not in state["context"]:
state["context"]["vector_store"] = create_vector_store()
vector_store = state["context"]["vector_store"]
# Perform retrieval
relevant_docs = vector_store.similarity_search(
state["question"],
k=3 # Retrieve top 3 most relevant chunks
)
# Format context for LLM
rag_context = "\n\n[Relevant Knowledge]\n"
rag_context += "\n---\n".join([doc.page_content for doc in relevant_docs])
# ====== RAG RETRIEVAL ======
# Initialize vector store if not present
if "vector_store" not in state["context"]:
state["context"]["vector_store"] = create_vector_store()
vector_store = state["context"]["vector_store"]
# Perform retrieval
relevant_docs = vector_store.similarity_search(
state["question"],
k=3 # Retrieve top 3 most relevant chunks
)
# Format context for LLM
rag_context = "\n\n[Relevant Knowledge]\n"
rag_context += "\n---\n".join([doc.page_content for doc in relevant_docs])
# ====== MODIFIED PROMPT ======
# Add RAG context to system prompt
system_prompt = (
"You are an expert problem solver, designed to provide concise and accurate answers. "
"Your process involves analyzing the question, intelligently selecting and using tools, "
"and synthesizing information.\n\n"
"**Available Tools:**\n"
f"{tool_descriptions}\n\n"
"**Tool Usage Guidelines:**\n"
"- Use **duckduckgo_search** for current events, general facts, or quick lookups. Provide a concise search query.\n"
"- Use **wikipedia_search** for encyclopedic information, historical context, or detailed topics. Provide a concise search term.\n"
"- Use **arxiv_search** for scientific papers, research, or cutting-edge technical information. Provide a concise search query.\n"
"- Use **document_qa** when the question explicitly refers to a specific document or when you have content to query. Input format: 'document_text||question'.\n"
"- Use **python_execution** for complex calculations, data manipulation, or logical operations that cannot be done with simple reasoning. Always provide the full Python code, ensuring it's valid and executable, and assign the final result to a variable named '_result_value' (e.g., '_result_value = 1 + 1').\n"
"- Use **transcript_video** for any question involving video or audio content (e.g., YouTube). Provide the full YouTube URL or video ID.\n\n"
"**Retrieved Context:**\n{rag_context}\n\n" # ADDED RAG CONTEXT
"**Current Context:**\n{context}\n\n"
"**Previous Reasoning Steps:**\n{reasoning}\n\n"
"**Current Task:** {current_task}\n"
"**Current Thoughts:** {current_thoughts}\n\n"
# ... [rest of prompt remains same] ...
)
prompt = ChatPromptTemplate.from_messages([
SystemMessage(content=system_prompt),
*state["history"]
])
formatted_messages = prompt.format_messages(
rag_context=rag_context, # ADD THIS ARGUMENT
context=state["context"],
reasoning=state["reasoning"],
question=state["question"],
current_task=state["current_task"],
current_thoughts=state["current_thoughts"]
)
# Use tokenizer's chat template for optimal formatting with chat models
try:
full_input_string = tokenizer.apply_chat_template(
formatted_messages,
tokenize=False,
add_generation_prompt=True # Adds the assistant's turn start token
)
except Exception as e:
print(f"WARNING: Failed to apply chat template: {e}. Falling back to simple string join. Model performance may be affected.")
full_input_string = "\n".join([msg.content for msg in formatted_messages])
def call_with_retry_local(inputs, retries=3): # Reduced retries for local models as network isn't primary issue
for attempt in range(retries):
try:
response_text = llm.invoke(inputs)
# Strip the prompt from the generated text
if response_text.startswith(inputs):
content = response_text[len(inputs):].strip()
else:
content = response_text.strip()
print(f"DEBUG: RAW LOCAL LLM Response (Attempt {attempt+1}):\n---\n{content}\n---")
# Attempt to parse to validate structure
json.loads(content)
return AIMessage(content=content)
except json.JSONDecodeError as e:
print(f"[Retry {attempt+1}/{retries}] Local LLM returned invalid JSON. Error: {e}. Retrying...")
print(f"Invalid JSON content (partial): {content[:200]}...")
state["history"].append(AIMessage(content=f"[Parsing Error] The previous LLM output was not valid JSON. Expected format: ```json{{\"Reasoning\": \"...\", \"Action\": \"...\", \"Action Input\": \"...\"}}```. Please ensure your response is ONLY valid JSON and strictly follows the format. Error: {e}"))
time.sleep(5)
except Exception as e:
print(f"[Retry {attempt+1}/{retries}] An unexpected error occurred during local LLM call: {e}.")
state["history"].append(AIMessage(content=f"[Local LLM Error] Failed to get a response from the local LLM: {e}. Trying again."))
time.sleep(10)
raise RuntimeError("Failed after multiple retries due to local Hugging Face model issues or invalid JSON.")
response = call_with_retry_local(full_input_string)
content = response.content
reasoning, action, action_input = parse_agent_response(content)
print(f"DEBUG: Parsed Action: '{action}', Action Input: '{action_input[:100]}...'")
if isinstance(response, AIMessage) and content == response.content:
state["history"].append(AIMessage(content=content))
state["reasoning"] += f"\nStep {state['iterations'] + 1}: {reasoning}"
state["iterations"] += 1
state["current_thoughts"] = reasoning
if "final answer" in action.lower():
state["final_answer"] = action_input
else:
state["context"]["pending_action"] = {
"tool": action,
"input": action_input
}
state["history"].append(AIMessage(content=f"Agent decided to use tool: {action} with input: {action_input}"))
print(f"DEBUG: Exiting reasoning_node. New history length: {len(state['history'])}")
return state
def tool_node(state: AgentState) -> AgentState:
"""
Node for executing the chosen tool and returning its output.
"""
print(f"DEBUG: Entering tool_node. Iteration: {state['iterations']}")
tool_call_dict = state["context"].pop("pending_action", None)
if not tool_call_dict:
error_message = "[Tool Error] No pending_action found in context. This indicates an issue with graph flow."
print(f"ERROR: {error_message}")
state["history"].append(AIMessage(content=error_message))
return state
tool_name = tool_call_dict.get("tool")
tool_input = tool_call_dict.get("input")
if not tool_name or tool_input is None:
error_message = f"[Tool Error] Invalid action request from LLM: Tool name '{tool_name}' or input '{tool_input}' was empty or None. LLM needs to provide valid 'Action' and 'Action Input'."
print(f"ERROR: {error_message}")
state["history"].append(AIMessage(content=error_message))
state["context"].pop("pending_action", None)
return state
available_tools = state.get("tools", [])
tool_fn = next((t for t in available_tools if t.name == tool_name), None)
if tool_fn is None:
tool_output = f"[Tool Error] Tool '{tool_name}' not found or not available. Please choose from: {', '.join([t.name for t in available_tools])}"
print(f"ERROR: {tool_output}")
else:
try:
print(f"DEBUG: Invoking tool '{tool_name}' with input: '{tool_input[:100]}...'")
tool_output = tool_fn.run(tool_input)
if not tool_output and tool_output is not False:
tool_output = f"[{tool_name} output] No specific result found for '{tool_input}'. The tool might have returned an empty response."
except Exception as e:
tool_output = f"[Tool Error] An error occurred while running '{tool_name}': {str(e)}"
print(f"ERROR: {tool_output}")
state["history"].append(AIMessage(content=f"[{tool_name} output]\n{tool_output}"))
print(f"DEBUG: Exiting tool_node. Tool output added to history. New history length: {len(state['history'])}")
return state
# ====== Agent Graph ======
def create_agent_workflow(tools: List[BaseTool]): # Use BaseTool for consistency
workflow = StateGraph(AgentState)
workflow.add_node("reason", reasoning_node)
workflow.add_node("action", tool_node)
workflow.set_entry_point("reason")
workflow.add_conditional_edges(
"reason",
should_continue,
{
"action": "action",
"reason": "reason",
"end": END
}
)
workflow.add_edge("action", "reason")
app = workflow.compile()
return app
# ====== Agent Interface ======
class BasicAgent:
def __init__(self):
# Instantiate tools
self.tools = [
duckduckgo_search,
wikipedia_search,
arxiv_search,
document_qa,
python_execution,
VideoTranscriptionTool()
]
# Pre-initialize RAG vector store
self.vector_store = create_vector_store()
self.workflow = create_agent_workflow(self.tools)
def __call__(self, question: str) -> str:
print(f"\n--- Agent received question: {question[:80]}{'...' if len(question) > 80 else ''} ---")
state = {
"question": question,
"context": {
"vector_store": self.vector_store # Include vector store in context
},
"reasoning": "",
"iterations": 0,
"history": [HumanMessage(content=question)],
"final_answer": None,
"current_task": "Understand the question and plan the next step.",
"current_thoughts": "",
"tools": self.tools
}
final_state = self.workflow.invoke(state)
if final_state.get("final_answer") is not None:
answer = final_state["final_answer"]
print(f"--- Agent returning FINAL ANSWER: {answer} ---")
return answer
else:
print(f"--- ERROR: Agent finished without setting 'final_answer' for question: {question} ---")
if final_state["history"]:
last_message = final_state["history"][-1].content
print(f"Last message in history: {last_message}")
return f"Agent could not fully answer. Last message: {last_message}"
else:
raise ValueError("Agent finished without providing a final answer and no history messages.")
def run_and_submit_all( profile: gr.OAuthProfile | None):
"""
Fetches all questions, runs the BasicAgent on them, submits all answers,
and displays the results.
"""
# --- Determine HF Space Runtime URL and Repo URL ---
space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code
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"
# 1. Instantiate Agent ( modify this part to create your agent)
try:
agent = BasicAgent()
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)
# 2. Fetch Questions
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 requests.exceptions.RequestException as e:
print(f"Error fetching questions: {e}")
return f"Error fetching questions: {e}", None
except requests.exceptions.JSONDecodeError as e:
print(f"Error decoding JSON response from questions endpoint: {e}")
print(f"Response text: {response.text[:500]}")
return f"Error decoding server response for questions: {e}", None
except Exception as e:
print(f"An unexpected error occurred fetching questions: {e}")
return f"An unexpected error occurred fetching questions: {e}", None
# 3. Run your Agent
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)
# 4. Prepare Submission
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)
# 5. Submit
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 requests.exceptions.HTTPError as e:
error_detail = f"Server responded with status {e.response.status_code}."
try:
error_json = e.response.json()
error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
except requests.exceptions.JSONDecodeError:
error_detail += f" Response: {e.response.text[:500]}"
status_message = f"Submission Failed: {error_detail}"
print(status_message)
results_df = pd.DataFrame(results_log)
return status_message, results_df
except requests.exceptions.Timeout:
status_message = "Submission Failed: The request timed out."
print(status_message)
results_df = pd.DataFrame(results_log)
return status_message, results_df
except requests.exceptions.RequestException as e:
status_message = f"Submission Failed: Network error - {e}"
print(status_message)
results_df = pd.DataFrame(results_log)
return status_message, results_df
except Exception as e:
status_message = f"An unexpected error occurred during submission: {e}"
print(status_message)
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.
---
**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).
This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
"""
)
gr.LoginButton()
run_button = gr.Button("Run Evaluation & Submit All Answers")
status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
# Removed max_rows=10 from DataFrame constructor
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)
# 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
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 repo URLs if SPACE_ID is found
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)