Spaces:
Sleeping
Sleeping
import os | |
import gradio as gr | |
import requests | |
import inspect | |
import pandas as pd | |
from langgraph.prebuilt import ToolNode | |
# from typing import Any, Dict | |
# from typing import TypedDict, Annotated | |
from langchain_openai import ChatOpenAI | |
from langgraph.graph import StateGraph, START, END | |
from langgraph.graph.message import add_messages | |
from langchain.schema import HumanMessage, SystemMessage, AIMessage | |
# Create a ToolNode that knows about your web_search function | |
import json | |
from old2state import AgentState | |
# --- Constants --- | |
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" | |
from old2tools import ocr_image_tool, parse_excel_tool, web_search_tool, run_tools, audio_transcriber_tool, wikipedia_search_tool | |
llm = ChatOpenAI(model_name="gpt-4.1") | |
# βββ 1) plan_node βββ | |
# βββ 1) plan_node βββ | |
tool_counter = 0 | |
# βββ 1) plan_node βββ | |
def plan_node(state: AgentState) -> AgentState: | |
""" | |
Step 1: Ask GPT to draft a concise direct answer (INTERIM_ANSWER), | |
then decide if it's confident enough to stop or if it needs one tool. | |
If confident: return {"final_answer":"<answer>"} | |
Otherwise: return exactly one of: | |
{"wiki_query":"..."}, | |
{"ocr_path":"..."}, | |
{"excel_path":"...","excel_sheet_name":"..."}, | |
{"audio_path":"..."} | |
""" | |
prior_msgs = state.get("messages", []) | |
user_input = "" | |
for msg in reversed(prior_msgs): | |
if isinstance(msg, HumanMessage): | |
user_input = msg.content | |
break | |
system_msg = SystemMessage( | |
content=( | |
"You are an agent that must do two things in one JSON output:\n\n" | |
" 1) Provide a concise, direct answer to the user's question (no explanation).\n" | |
" 2) Judge whether that answer is reliable:\n" | |
" β’ If you are fully confident, return exactly:\n" | |
" {\"final_answer\":\"<your concise answer>\"}\n" | |
" and nothing else.\n" | |
" β’ Otherwise, return exactly one of:\n" | |
" {\"wiki_query\":\"<Wikipedia search>\"}\n" | |
" {\"ocr_path\":\"<image path or task_id>\"}\n" | |
" {\"excel_path\":\"<xlsx path>\", \"excel_sheet_name\":\"<sheet name>\"}\n" | |
" {\"audio_path\":\"<audio path or task_id>\"}\n" | |
" and nothing else.\n" | |
"Do NOT wrap in markdownβoutput only a single JSON object.\n" | |
f"User's question: \"{user_input}\"\n" | |
) | |
) | |
human_msg = HumanMessage(content=user_input) | |
llm_response = llm([system_msg, human_msg]) | |
llm_out = llm_response.content.strip() | |
ai_msg = AIMessage(content=llm_out) | |
new_msgs = prior_msgs.copy() + [ai_msg] | |
try: | |
parsed = json.loads(llm_out) | |
if isinstance(parsed, dict): | |
partial: AgentState = {"messages": new_msgs} | |
allowed = { | |
"final_answer", | |
"wiki_query", | |
"ocr_path", | |
"excel_path", | |
"excel_sheet_name", | |
"audio_path", | |
} | |
for k, v in parsed.items(): | |
if k in allowed: | |
partial[k] = v | |
return partial | |
except json.JSONDecodeError: | |
pass | |
return { | |
"messages": new_msgs, | |
"final_answer": "Sorry, I could not parse your intent.", | |
} | |
# βββ 2) store_prev_state βββ | |
def store_prev_state(state: AgentState) -> AgentState: | |
return {**state, "prev_state": state.copy()} | |
# βββ 3) tools_node βββ | |
def tool_node(state: AgentState) -> AgentState: | |
""" | |
Dispatch exactly one tool based on which key was set: | |
- wiki_query β wikipedia_search_tool | |
- ocr_path β ocr_image_tool | |
- excel_path β parse_excel_tool | |
- audio_path β audio_transcriber_tool | |
""" | |
global tool_counter | |
if tool_counter >= 5: | |
# If we've already run 5 tools, do nothing | |
return { | |
"messages": state["messages"], | |
"final_answer": state.get("final_answer", "No interim answer available.") | |
} | |
tool_counter += 1 | |
if state.get("wiki_query"): | |
return wikipedia_search_tool(state) | |
if state.get("ocr_path"): | |
return ocr_image_tool(state) | |
if state.get("excel_path"): | |
return parse_excel_tool(state) | |
if state.get("audio_path"): | |
return audio_transcriber_tool(state) | |
return {} # no tool key present | |
# βββ 4) merge_tool_output βββ | |
def merge_tool_output(state: AgentState) -> AgentState: | |
""" | |
Combine previous state and tool output into one, but remove any stale request-keys. | |
""" | |
prev = state.get("prev_state", {}).copy() | |
# Drop stale request-keys in prev | |
for dead in ["wiki_query", "ocr_path", "excel_path", "excel_sheet_name", "audio_path"]: | |
prev.pop(dead, None) | |
merged = {**prev, **state} | |
# Drop them again from merged so they don't persist into the next cycle | |
for dead in ["wiki_query", "ocr_path", "excel_path", "excel_sheet_name", "audio_path"]: | |
merged.pop(dead, None) | |
merged.pop("prev_state", None) | |
return merged | |
# βββ 5) inspect_node βββ | |
def inspect_node(state: AgentState) -> AgentState: | |
""" | |
After running a tool, show GPT: | |
- ORIGINAL user question | |
- Any tool results (web_search_result, ocr_result, excel_result, transcript, wiki_result) | |
- The INTERIM_ANSWER (always present if plan_node ran correctly) | |
If tool_counter β₯ 5, use LLM once more (with full context) to craft a final answer. | |
Otherwise, ask GPT to either: | |
β’ Return {"final_answer":"<final>"} if done, OR | |
β’ Return exactly one tool key to run next (wiki_query / ocr_path / excel_path & excel_sheet_name / audio_path). | |
""" | |
global tool_counter | |
# If we've already run 5 tools, ask GPT for a strictlyβformatted JSON final_answer | |
if tool_counter >= 5: | |
messages_for_llm = [] | |
# Reβinsert the userβs question | |
question = "" | |
for msg in reversed(state.get("messages", [])): | |
if isinstance(msg, HumanMessage): | |
question = msg.content | |
break | |
messages_for_llm.append(SystemMessage(content=f"USER_QUESTION: {question}")) | |
# Add any tool results so far | |
if sr := state.get("web_search_result"): | |
messages_for_llm.append(SystemMessage(content=f"WEB_SEARCH_RESULT: {sr}")) | |
if orc := state.get("ocr_result"): | |
messages_for_llm.append(SystemMessage(content=f"OCR_RESULT: {orc}")) | |
if exr := state.get("excel_result"): | |
messages_for_llm.append(SystemMessage(content=f"EXCEL_RESULT: {exr}")) | |
if tr := state.get("transcript"): | |
messages_for_llm.append(SystemMessage(content=f"AUDIO_TRANSCRIPT: {tr}")) | |
if wr := state.get("wiki_result"): | |
messages_for_llm.append(SystemMessage(content=f"WIKIPEDIA_RESULT: {wr}")) | |
# Show the interim answer | |
interim = state.get("interim_answer", "") | |
messages_for_llm.append(SystemMessage(content=f"INTERIM_ANSWER: {interim}")) | |
# Now ask for JSON ONLY (no reasoning, no extra text) | |
final_prompt = ( | |
"Finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER]. YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string." | |
"Using only the information aboveβincluding the USER_QUESTION, " | |
"any TOOL_RESULT, and the INTERIM_ANSWERβproduce a concise final answer. " | |
"Return exactly one JSON object and nothing else, in this format:\n\n" | |
"{\"final_answer\":\"<your final answer>\"}\n" | |
"Do not include any other words or punctuation outside that JSON. if its numbers, dont show the units" | |
) | |
messages_for_llm.append(SystemMessage(content=final_prompt)) | |
llm_response = llm(messages_for_llm) | |
raw = llm_response.content.strip() | |
new_msgs = state["messages"] + [AIMessage(content=raw)] | |
# Try to parse exactly one JSON with "final_answer" | |
try: | |
parsed = json.loads(raw) | |
if isinstance(parsed, dict) and "final_answer" in parsed: | |
return {"messages": new_msgs, "final_answer": parsed["final_answer"]} | |
except json.JSONDecodeError: | |
pass | |
# Fallback to returning the interim in case JSON parse fails | |
return {"messages": new_msgs, "final_answer": interim} | |
# βββββββββββ If tool_counter < 5, proceed as before βββββββββββ | |
messages_for_llm = [] | |
# (1) Reβinsert original user question | |
question = "" | |
for msg in reversed(state.get("messages", [])): | |
if isinstance(msg, HumanMessage): | |
question = msg.content | |
break | |
messages_for_llm.append(SystemMessage(content=f"USER_QUESTION: {question}")) | |
# (2) Add any tool results | |
if sr := state.get("web_search_result"): | |
messages_for_llm.append(SystemMessage(content=f"WEB_SEARCH_RESULT: {sr}")) | |
if orc := state.get("ocr_result"): | |
messages_for_llm.append(SystemMessage(content=f"OCR_RESULT: {orc}")) | |
if exr := state.get("excel_result"): | |
messages_for_llm.append(SystemMessage(content=f"EXCEL_RESULT: {exr}")) | |
if tr := state.get("transcript"): | |
messages_for_llm.append(SystemMessage(content=f"AUDIO_TRANSCRIPT: {tr}")) | |
if wr := state.get("wiki_result"): | |
messages_for_llm.append(SystemMessage(content=f"WIKIPEDIA_RESULT: {wr}")) | |
# (3) Always show the interim answer | |
interim = state.get("interim_answer", "") | |
messages_for_llm.append(SystemMessage(content=f"INTERIM_ANSWER: {interim}")) | |
# (4) Prompt GPT to decide final or another tool | |
prompt = ( | |
"You have a current draft answer (INTERIM_ANSWER) and possibly some tool results above.\n" | |
"If you are confident itβs correct, return exactly:\n" | |
" {\"final_answer\":\"<your final answer>\"}\n" | |
"and nothing else.\n" | |
"Otherwise, return exactly one of these JSON literals to fetch another tool:\n" | |
" {\"wiki_query\":\"<query for Wikipedia>\"}\n" | |
" {\"ocr_path\":\"<image path or task_id>\"}\n" | |
" {\"excel_path\":\"<xls path>\", \"excel_sheet_name\":\"<sheet name>\"}\n" | |
" {\"audio_path\":\"<audio path or task_id>\"}\n" | |
"Do NOT wrap in markdownβreturn only the JSON object.\n" | |
) | |
messages_for_llm.append(SystemMessage(content=prompt)) | |
llm_response = llm(messages_for_llm) | |
raw = llm_response.content.strip() | |
new_msgs = state["messages"] + [AIMessage(content=raw)] | |
# Try to parse the LLMβs JSON | |
try: | |
parsed = json.loads(raw) | |
if isinstance(parsed, dict): | |
# (a) If GPT gave a final_answer, return immediately | |
if "final_answer" in parsed: | |
return {"messages": new_msgs, "final_answer": parsed["final_answer"]} | |
# (b) If GPT requested exactly one valid tool, return only that key | |
valid_keys = {"wiki_query", "ocr_path", "excel_path", "excel_sheet_name", "audio_path"} | |
requested_keys = set(parsed.keys()) & valid_keys | |
if len(requested_keys) == 1: | |
clean: AgentState = {"messages": new_msgs} | |
for k in requested_keys: | |
clean[k] = parsed[k] | |
return clean | |
except json.JSONDecodeError: | |
pass | |
# (c) Fallback: if GPT never returned a valid tool key or a final_answer, | |
# just finalize with the existing interim_answer | |
return {"messages": new_msgs, "final_answer": interim} | |
# βββ 6) finalize_node βββ | |
def finalize_node(state: AgentState) -> AgentState: | |
""" | |
If state already has "final_answer", return it. Otherwise, it's an error. | |
""" | |
if fa := state.get("final_answer"): | |
return {"final_answer": fa} | |
return {"final_answer": "ERROR: finalize called without a final_answer."} | |
# βββ 7) Build the graph and wire edges βββ | |
graph = StateGraph(AgentState) | |
# Register nodes | |
graph.add_node("plan", plan_node) | |
graph.add_node("store_prev_state", store_prev_state) | |
graph.add_node("tools", tool_node) | |
graph.add_node("merge_tool_output", merge_tool_output) | |
graph.add_node("inspect", inspect_node) | |
graph.add_node("finalize", finalize_node) | |
# START β plan | |
graph.add_edge(START, "plan") | |
# plan β either finalize (if plan set final_answer) or store_prev_state (if plan wants a tool) | |
def route_plan(plan_out: AgentState) -> str: | |
if plan_out.get("final_answer") is not None: | |
return "finalize" | |
return "store_prev_state" | |
graph.add_conditional_edges( | |
"plan", | |
route_plan, | |
{"store_prev_state": "store_prev_state", "finalize": "finalize"}, | |
) | |
# store_prev_state β tools | |
graph.add_edge("store_prev_state", "tools") | |
# tools β merge_tool_output | |
graph.add_edge("tools", "merge_tool_output") | |
# merge_tool_output β inspect | |
graph.add_edge("merge_tool_output", "inspect") | |
# inspect β either finalize (if inspect set final_answer) or store_prev_state (if inspect wants another tool) | |
def route_inspect(inspect_out: AgentState) -> str: | |
if inspect_out.get("final_answer") is not None: | |
return "finalize" | |
return "store_prev_state" | |
graph.add_conditional_edges( | |
"inspect", | |
route_inspect, | |
{"store_prev_state": "store_prev_state", "finalize": "finalize"}, | |
) | |
# finalize β END | |
graph.add_edge("finalize", END) | |
compiled_graph = graph.compile() | |
# βββ 8) respond_to_input βββ | |
def respond_to_input(user_input: str, task_id) -> str: | |
""" | |
Reset the global tool_counter, seed state['messages'], invoke the graph, | |
and return the final_answer. | |
""" | |
global tool_counter | |
tool_counter = 0 # Reset on every new user query | |
system_msg = SystemMessage( | |
content=( | |
"You are an agent orchestrator. Decide whether to use a tool or answer directly.\n" | |
"Try not to use tools so many times. If you think you can answer the question without using a tool, do it Please.\n" | |
"Tools available:\n" | |
" β’ Wikipedia: set {\"wiki_query\":\"<search terms>\"}\n" | |
" β’ OCR: set {\"ocr_path\":\"<image path or task_id>\"}\n" | |
" β’ Excel: set {\"excel_path\":\"<xlsx path>\", \"excel_sheet_name\":\"<sheet>\"}\n" | |
" β’ Audio transcription: set {\"audio_path\":\"<audio path or task_id>\"}\n" | |
"If you can answer immediately, set {\"final_answer\":\"<answer>\"}. " | |
"Respond with only one JSON object and no extra formatting." | |
) | |
) | |
human_msg = HumanMessage(content=user_input) | |
initial_state: AgentState = {"messages": [system_msg, human_msg], "task_id": task_id} | |
final_state = compiled_graph.invoke(initial_state) | |
return final_state.get("final_answer", "Error: No final answer generated.") | |
class BasicAgent: | |
def __init__(self): | |
print("BasicAgent initialized.") | |
def __call__(self, question: str, task_id) -> str: | |
# print(f"Agent received question (first 50 chars): {question[:50]}...") | |
# fixed_answer = "This is a default answer." | |
# print(f"Agent returning fixed answer: {fixed_answer}") | |
print() | |
print() | |
print() | |
print() | |
print(f"Agent received question: {question}") | |
print() | |
return respond_to_input(question, task_id) | |
# return fixed_answer | |
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, task_id) | |
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("LangGraph version:", langgraph.__version__) | |
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 | |
# import langgraph | |
# print("βΆοΈ LangGraph version:", langgraph.__version__) | |
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) |