"""Groq (OpenAI-compatible) GAIA agent - reuses tools/file handling from gaia_agent.""" import base64 import json import pathlib import re import sys import time from groq import Groq from gaia_agent import ( SYSTEM_PROMPT, MAX_STEPS, TOOL_FUNCS, TOOLS, web_search, visit_webpage, run_python, download_task_file, ) GROQ_TEXT_MODEL = "openai/gpt-oss-120b" GROQ_VISION_MODEL = "qwen/qwen3.8-27b" GROQ_WHISPER_MODEL = "whisper-large-v3-turbo" MODEL_MAX_TOKENS = {GROQ_TEXT_MODEL: 16000, GROQ_VISION_MODEL: 900} TOOLS_OPENAI = [{"type": "function", "function": t} for t in TOOLS] GROQ_SYSTEM = SYSTEM_PROMPT + """ Additional tool-call rules (CRITICAL): - Tool call arguments MUST be valid JSON. The run_python "code" parameter must be ONE JSON-escaped string: newlines written as \\n, quotes escaped. NEVER emit raw Python as the arguments value. - If a tool call fails or arguments are rejected, re-issue the call with strictly valid JSON. Answer format (CRITICAL): - Follow the question's own format instructions EXACTLY: if it says alphabetize, sort alphabetically; if it says comma-separated, use exactly that. - For ingredient/name lists from audio: drop measurements but KEEP descriptive adjectives exactly as spoken (e.g. "ripe strawberries", "pure vanilla extract"). """ class GroqBrain: def __init__(self, api_key: str): self.client = Groq(api_key=api_key, timeout=180) self.calls = 0 def generate(self, messages, tools=None, model=GROQ_TEXT_MODEL, retries=10): last_err = None max_tokens = MODEL_MAX_TOKENS.get(model, 16000) for attempt in range(retries): try: kwargs = dict(model=model, messages=messages, temperature=0.1, max_tokens=max_tokens) if tools: kwargs["tools"] = tools resp = self.client.chat.completions.create(**kwargs) self.calls += 1 return resp.choices[0].message except Exception as e: last_err = e msg = str(e) if "tool_use_failed" in msg or "Failed to parse tool call arguments" in msg: if model != GROQ_VISION_MODEL: print(f" [toolargs {model}] switching to {GROQ_VISION_MODEL}", flush=True) model = GROQ_VISION_MODEL max_tokens = MODEL_MAX_TOKENS[model] time.sleep(2) continue if "output tokens per minute" in msg or "OTPM" in msg: m = re.search(r"Limit (\d+)", msg) if m: max_tokens = min(max_tokens, int(m.group(1)) - 100) print(f" [OTPM {model}] capping max_tokens={max_tokens}", flush=True) continue if "429" in msg or "Rate limit" in msg or "rate_limit" in msg: m = re.search(r"try again in ([\d.]+)s", msg) wait = float(m.group(1)) + 3 if m else 62 print(f" [429 {model}] waiting {wait:.0f}s", flush=True) time.sleep(wait) continue if "503" in msg or "model is currently" in msg or "decommissioned" in msg: print(f" [503 {model}] attempt {attempt}", flush=True) time.sleep(min(2 ** attempt + 1, 30)) continue print(f" [error {model}] {msg[:150]}", flush=True) time.sleep(min(2 ** attempt + 1, 30)) raise RuntimeError(f"Groq all attempts failed: {last_err}") def transcribe(self, audio_path: pathlib.Path) -> str: with open(audio_path, "rb") as f: resp = self.client.audio.transcriptions.create(model=GROQ_WHISPER_MODEL, file=f) return resp.text def build_user_content(question: dict, brain: GroqBrain): """Returns (messages, model) - multimodal content in OpenAI format.""" qtext = question["question"] task_id = question["task_id"] file_text = None image_data_url = None if question.get("file_name"): p = download_task_file(task_id, question["file_name"]) if p: ext = p.suffix.lower() if ext in (".png", ".jpg", ".jpeg"): b64 = base64.b64encode(p.read_bytes()).decode() image_data_url = f"data:image/{'png' if ext == '.png' else 'jpeg'};base64,{b64}" elif ext in (".mp3", ".wav", ".m4a"): transcript = brain.transcribe(p) file_text = ( "Attached audio file transcript (Whisper verbatim):\n" f'"""\n{transcript}\n"""\n' "IMPORTANT: when listing items from the transcript, use the EXACT wording as spoken " "including ALL descriptive words (e.g. 'freshly squeezed lemon juice' stays 'freshly squeezed lemon juice'), " "removing ONLY quantities/measurements (e.g. 'two cups of X' -> 'X'). " "Then apply the question's requested formatting (alphabetize / comma-separated if asked)." ) print(f" [whisper] {transcript[:120]}", flush=True) elif ext in (".xlsx", ".xls"): df_text = run_python(f""" import pandas as pd xl = pd.ExcelFile({str(p)!r}) for sheet in xl.sheet_names: df = xl.parse(sheet) print(f"=== SHEET: {{sheet}} ({{len(df)}} rows) ===") print(df.to_string(index=False)) """) file_text = f"The task file {p.name} is an Excel file. Full contents via pandas:\n{df_text}" else: content = p.read_text(errors="replace")[:30000] file_text = f"The task file {p.name} contains:\n```\n{content}\n```" if re.search(r'https://(?:www\.)?youtube\.com/watch\?v=[\w-]+', qtext): file_text = (file_text or "") + "\n[Note: you cannot watch YouTube videos directly. Use web_search to find information about this specific video.]" user_text = f"Question: {qtext}" if file_text: user_text += f"\n\n{file_text}" parts = [{"type": "text", "text": user_text}] if image_data_url: parts.append({"type": "image_url", "image_url": {"url": image_data_url}}) return parts, GROQ_VISION_MODEL return parts, GROQ_TEXT_MODEL def answer_question_groq(brain: GroqBrain, question: dict, log=print) -> str: content, model = build_user_content(question, brain) messages = [ {"role": "system", "content": GROQ_SYSTEM}, {"role": "user", "content": content}, ] for step in range(MAX_STEPS): msg = brain.generate(messages, tools=TOOLS_OPENAI, model=model) if msg.tool_calls: messages.append({ "role": "assistant", "content": msg.content or "", "tool_calls": [ {"id": tc.id, "type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments}} for tc in msg.tool_calls ], }) for tc in msg.tool_calls: args = json.loads(tc.function.arguments or "{}") log(f" step {step+1}: {tc.function.name}({str(args)[:140]})") result = TOOL_FUNCS[tc.function.name](**args) messages.append({"role": "tool", "tool_call_id": tc.id, "content": result[:6000]}) continue text = msg.content or "" m = re.search(r"FINAL:\s*(.+)", text, re.S) if m and m.group(1).strip(): return m.group(1).strip()[:200] messages.append({"role": "assistant", "content": text}) messages.append({"role": "user", "content": "Continue. When confident, output the final line EXACTLY as 'FINAL: ' with ONLY the concise answer value."}) for demand in [ "You have no steps left. Based on everything gathered, output your best answer NOW as 'FINAL: ' with ONLY the concise answer value.", "You MUST give a concrete answer now. Output 'FINAL: ' where value is a number, name, or short phrase - never empty. Make your best educated guess.", ]: messages.append({"role": "user", "content": demand}) msg = brain.generate(messages, tools=None, model=model) text = msg.content or "" m2 = re.search(r"FINAL:\s*(.+)", text, re.S) ans = (m2.group(1) if m2 else text).strip() if ans and ans.lower() not in ("none", "n/a", "unknown", "i don't know"): return ans[:200] return "No answer produced."