|
|
import os |
|
|
import gradio as gr |
|
|
import requests |
|
|
import pandas as pd |
|
|
import json |
|
|
import re |
|
|
from langchain_core.messages import HumanMessage |
|
|
from langgraph_agent import build_graph |
|
|
|
|
|
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" |
|
|
|
|
|
class BasicAgent: |
|
|
def __init__(self): |
|
|
print("BasicAgent initialized.") |
|
|
self.graph = build_graph() |
|
|
|
|
|
def __call__(self, question: str, task_id: str = None) -> str: |
|
|
print(f"Agent received question (first 50 chars): {question[:50]}...") |
|
|
messages = [HumanMessage(content=question)] |
|
|
messages = self.graph.invoke({"messages": messages}) |
|
|
|
|
|
if not messages or not messages.get('messages') or messages['messages'][-1].content is None: |
|
|
return "I am unable to determine the information using the available tools." |
|
|
|
|
|
answer = messages['messages'][-1].content |
|
|
|
|
|
if isinstance(answer, list) and not answer: |
|
|
return "I am unable to determine the information using the available tools." |
|
|
|
|
|
if not isinstance(answer, str): |
|
|
answer = str(answer) |
|
|
|
|
|
answer = answer.strip() |
|
|
|
|
|
match = re.search(r'FINAL ANSWER:\s*(.*)', answer, re.IGNORECASE | re.DOTALL) |
|
|
if match: |
|
|
final_answer = match.group(1).strip() |
|
|
if (final_answer.startswith('"') and final_answer.endswith('"')) or \ |
|
|
(final_answer.startswith("'") and final_answer.endswith("'")): |
|
|
final_answer = final_answer[1:-1].strip() |
|
|
answer = final_answer |
|
|
else: |
|
|
print("Warning: 'FINAL ANSWER:' not found; submitting full answer.") |
|
|
|
|
|
if not answer: |
|
|
answer = "I am unable to determine the information using the available tools." |
|
|
|
|
|
return answer |
|
|
|
|
|
def run_and_submit_all(profile: gr.OAuthProfile | None): |
|
|
space_id = os.getenv("SPACE_ID") |
|
|
|
|
|
if profile: |
|
|
username = 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: |
|
|
agent = BasicAgent() |
|
|
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"Error fetching questions: {e}") |
|
|
return f"Error 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: |
|
|
model_answer = agent(question_text, task_id=task_id).strip() |
|
|
|
|
|
if (model_answer.startswith('"') and model_answer.endswith('"')) or \ |
|
|
(model_answer.startswith("'") and model_answer.endswith("'")): |
|
|
model_answer = model_answer[1:-1].strip() |
|
|
|
|
|
print(f"Answer for task {task_id}: '{model_answer}'") |
|
|
answers_payload.append({ |
|
|
"task_id": task_id, |
|
|
"submitted_answer": model_answer |
|
|
}) |
|
|
results_log.append({ |
|
|
"Task ID": task_id, |
|
|
"Question": question_text, |
|
|
"Submitted Answer": model_answer |
|
|
}) |
|
|
except Exception as e: |
|
|
print(f"Error running agent on task {task_id}: {e}") |
|
|
answers_payload.append({ |
|
|
"task_id": task_id, |
|
|
"submitted_answer": f"AGENT ERROR: {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) |
|
|
|
|
|
data = { |
|
|
"username": username.strip(), |
|
|
"agent_code": agent_code, |
|
|
"answers": answers_payload |
|
|
} |
|
|
|
|
|
print(f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'...") |
|
|
print("Submission payload preview (first 3 answers):") |
|
|
for ans in answers_payload[:3]: |
|
|
print(json.dumps(ans, ensure_ascii=False)) |
|
|
|
|
|
print(f"Submitting answers to: {submit_url}") |
|
|
try: |
|
|
response = requests.post(submit_url, json=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"Submission Failed: {e}" |
|
|
print(status_message) |
|
|
results_df = pd.DataFrame(results_log) |
|
|
return status_message, results_df |
|
|
|
|
|
|
|
|
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 separate action or even to answer the questions asynchronously. |
|
|
""" |
|
|
) |
|
|
|
|
|
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) |
|
|
|
