File size: 5,068 Bytes
9663fd1 | 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 140 141 142 | # Agente para la evaluacion final del curso de agentes de Hugging Face.
# Dependencias: pip install requests pandas ddgs openai openpyxl
# La clave de Groq NUNCA se escribe aqui: se lee de la variable de entorno
# GROQ_API_KEY o se pide por teclado con getpass.
import os
import re
import io
import time
import getpass
import requests
import pandas as pd
from openai import OpenAI
from ddgs import DDGS
# ================== CONFIGURACION ==================
USERNAME = "imn0tjuaness"
SPACE_CODE_URL = "https://huggingface.co/imn0tjuaness/agent-code/tree/main"
API = "https://agents-course-unit4-scoring.hf.space"
MODEL = "openai/gpt-oss-120b" # alternativa si hay 429: "openai/gpt-oss-20b"
PAUSA = 5 # segundos entre preguntas
ENVIAR = True # False para ver respuestas sin enviar
# ===================================================
client = OpenAI(
api_key=os.environ.get("GROQ_API_KEY") or getpass.getpass("Groq API key: "),
base_url="https://api.groq.com/openai/v1",
)
def llm(system, user, max_tokens=300, retries=4):
for attempt in range(retries):
try:
r = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
max_tokens=max_tokens + 1500, # margen para el razonamiento interno
temperature=0.0,
)
out = (r.choices[0].message.content or "").strip()
if not out:
raise ValueError("respuesta vacia")
return out
except Exception as e:
msg = str(e)
print(f" [llm error intento {attempt + 1}]", msg[:200])
if "model_not_found" in msg or "401" in msg:
raise
time.sleep(10 * (attempt + 1))
raise RuntimeError("El LLM fallo tras varios intentos")
def search_web(query, n=4):
try:
res = DDGS().text(query, max_results=n)
return "\n".join(f"- {r['title']}: {r['body'][:300]}" for r in res)
except Exception as e:
print(" [search error]", e)
return ""
def get_file_context(task_id):
try:
r = requests.get(f"{API}/files/{task_id}", timeout=30)
if r.status_code != 200 or not r.content:
return ""
ctype = r.headers.get("content-type", "")
cd = r.headers.get("content-disposition", "")
if "xlsx" in cd or "spreadsheet" in ctype:
df = pd.read_excel(io.BytesIO(r.content))
return f"\nArchivo adjunto (Excel):\n{df.to_string()[:3000]}\n"
if "csv" in cd or "csv" in ctype:
return f"\nArchivo adjunto (CSV):\n{r.text[:3000]}\n"
if "text" in ctype or ".py" in cd or ".txt" in cd:
return f"\nArchivo adjunto:\n{r.text[:3000]}\n"
return "\n(Hay un archivo adjunto binario, imagen o audio, que no se puede leer.)\n"
except Exception as e:
print(" [file error]", e)
return ""
SYSTEM = """You answer exam questions graded by EXACT MATCH.
Think briefly if needed, then finish with a line in this format:
FINAL ANSWER: <answer>
Rules for <answer>: just a number, a few words, or a comma-separated list.
No units unless asked, no thousands separators, no articles (a/the) for strings,
no trailing period. Lists are comma-separated in the order requested."""
def solve_task(task_id, question):
file_ctx = get_file_context(task_id)
query = llm(
"Write ONE short web search query (max 8 words) that would help answer the question. Output only the query.",
question[:1000],
max_tokens=30,
).strip('"')
web = search_web(query)
prompt = f"Question: {question}\n{file_ctx}\nWeb results for '{query}':\n{web}"
out = llm(SYSTEM, prompt, max_tokens=400)
m = re.search(r"FINAL ANSWER:\s*(.+)", out, re.S | re.I)
ans = (m.group(1) if m else out).strip().split("\n")[0]
return ans.strip('"\'`').rstrip(".")
def main():
questions = requests.get(f"{API}/questions", timeout=15).json()
print("Modelo en uso:", MODEL)
cache = {}
for i, q in enumerate(questions, 1):
tid = q["task_id"]
print(f"[{i}/{len(questions)}] {q['question'][:80]}...")
try:
cache[tid] = solve_task(tid, q["question"])
except Exception as e:
print(" [ERROR]", e)
cache[tid] = "unknown"
print(" ->", cache[tid])
time.sleep(PAUSA)
ok = sum(v != "unknown" for v in cache.values())
print(f"\nResueltas: {ok}/{len(questions)}")
if ENVIAR:
answers = [
{"task_id": q["task_id"], "submitted_answer": cache.get(q["task_id"], "unknown")}
for q in questions
]
r = requests.post(
f"{API}/submit",
timeout=120,
json={"username": USERNAME, "agent_code": SPACE_CODE_URL, "answers": answers},
)
print(r.json())
if __name__ == "__main__":
main()
|