Spaces:
Sleeping
Sleeping
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 --- | |
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)}" | |
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)}" | |
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)}" | |
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)}" | |
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) |