Spaces:
Sleeping
Sleeping
File size: 1,810 Bytes
8dddc6c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 |
import argparse
import textwrap
from typing import Any
import requests
from app import DEFAULT_API_URL, GAIAAgent
def fetch_question_row(task_id: str, api: str = DEFAULT_API_URL) -> dict[str, Any]:
"""Return the question dict associated with *task_id* (raises if not found)."""
resp = requests.get(f"{api}/questions", timeout=15)
resp.raise_for_status()
for row in resp.json():
if row["task_id"] == task_id:
return row
raise ValueError(f"task_id '{task_id}' not present in /questions.")
def run_one(task_id: str | None, question: str | None) -> None:
agent = GAIAAgent()
if task_id:
row = fetch_question_row(task_id)
question = row["question"]
print(f"\n{row}\n") # show full row incl. metadata
# --- show pretty question
print("=" * 90)
print(f"QUESTION ({task_id or 'adhoc'})")
print(textwrap.fill(question or "", width=90))
print("=" * 90)
assert question is not None, "Internal error: question was None"
answer = agent(question, task_id=task_id)
print(f"\nFINAL ANSWER --> {answer}")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run one GAIAAgent query locally.")
parser.add_argument("--task_id", help="GAIA task_id to fetch & run")
parser.add_argument("question", nargs="?", help="Ad-hoc question text (positional)")
ns = parser.parse_args()
# mutual-exclusion checks
if ns.task_id and ns.question:
parser.error("Provide either --task_id OR a question, not both.")
if ns.task_id is None and ns.question is None:
parser.error("You must supply a GAIA --task_id or a question.")
return ns
if __name__ == "__main__":
args = parse_args()
run_one(task_id=args.task_id, question=args.question)
|