Spaces:
Running
Running
File size: 3,346 Bytes
16ab8a2 b2320c3 16ab8a2 | 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 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | from __future__ import annotations
import json
import time
from pathlib import Path
from typing import Any
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
API_URL = "https://agents-course-unit4-scoring.hf.space/submit"
ANSWERS_FILE = Path("answers.json")
USERNAME = "nq001"
AGENT_CODE = (
"https://huggingface.co/spaces/"
"nq001/Vertex-agent/tree/main"
)
def load_answers() -> list[dict[str, str]]:
with ANSWERS_FILE.open("r", encoding="utf-8") as file:
answers: Any = json.load(file)
if not isinstance(answers, list):
raise ValueError("answers.json must contain a JSON list.")
if len(answers) != 20:
raise ValueError(
f"Expected 20 answers, but found {len(answers)}."
)
validated: list[dict[str, str]] = []
task_ids: set[str] = set()
for index, answer in enumerate(answers, start=1):
if not isinstance(answer, dict):
raise ValueError(f"Answer {index} is not an object.")
task_id = str(answer.get("task_id", "")).strip()
submitted_answer = str(
answer.get("submitted_answer", "")
).strip()
if not task_id:
raise ValueError(f"Answer {index} has no task_id.")
if task_id in task_ids:
raise ValueError(f"Duplicate task_id: {task_id}")
task_ids.add(task_id)
validated.append(
{
"task_id": task_id,
"submitted_answer": submitted_answer,
}
)
return validated
def create_session() -> requests.Session:
retry = Retry(
total=4,
connect=4,
read=4,
backoff_factor=3,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=frozenset({"GET", "POST"}),
raise_on_status=False,
)
adapter = HTTPAdapter(
max_retries=retry,
pool_connections=5,
pool_maxsize=5,
)
session = requests.Session()
session.mount("https://", adapter)
session.headers.update(
{
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "Vertex-Agent/1.0",
}
)
return session
def main() -> None:
answers = load_answers()
payload = {
"username": USERNAME,
"agent_code": AGENT_CODE,
"answers": answers,
}
print(f"Submitting {len(answers)} existing answers...")
print(f"Username: {USERNAME}")
print(f"Code URL: {AGENT_CODE}")
session = create_session()
try:
response = session.post(
API_URL,
json=payload,
timeout=(30, 300),
)
except requests.RequestException as exc:
raise SystemExit(
f"Submission connection failed after retries:\n{exc}"
) from exc
print("HTTP status:", response.status_code)
if not response.ok:
print("Response body:")
print(response.text[:3000])
response.raise_for_status()
try:
result = response.json()
except ValueError:
print("Non-JSON response:")
print(response.text[:3000])
raise SystemExit(1)
print("\nSubmission result:")
print(json.dumps(result, indent=2, ensure_ascii=False))
if __name__ == "__main__":
main() |