| |
| |
| |
| |
|
|
| import os |
| import re |
| import io |
| import time |
| import getpass |
| import requests |
| import pandas as pd |
| from openai import OpenAI |
| from ddgs import DDGS |
|
|
| |
| 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" |
| PAUSA = 5 |
| ENVIAR = True |
| |
|
|
| 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, |
| 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() |
|
|
