"""
GAIA Solver Pro — Autonomous Multi-Tool Agent
Engine: DeepSeek API (deepseek-chat)
Designed for the GAIA Benchmark with CSV output (Asynchronous/Concurrent Execution)
"""
# Import operating system utilities (for file paths and environment variables)
import os
# Import regular expressions library for text manipulation and string cleaning
import re
# Import JSON parsing library for handling structured data
import json
# Import time module to measure task execution duration
import time
# Import random module for jittered retry backoff delays
import random
# Import shutil for safe archive extraction (streaming member writes)
import shutil
# Import queue module for thread-safe worker pooling in parallel execution
import queue
# Import subprocess for running ffmpeg/ffprobe (video frame extraction)
import subprocess
# Import threading module for the shared vision-model lock
import threading
# Import HTTP requests library to make API calls and download files
import requests
# Import pandas for data processing, CSV exporting, and Excel reading
import pandas as pd
# Import PIL Image for handling image files when working with visual models
from PIL import Image
# Import thread pool executor tools for concurrent multithreaded processing
from concurrent.futures import ThreadPoolExecutor
# Import core agent tools and framework wrappers from the smolagents framework
from smolagents import CodeAgent, OpenAIServerModel, DuckDuckGoSearchTool, VisitWebpageTool, tool
# Import dotenv to load environment variables from a local .env file
from dotenv import load_dotenv
# Load secret environment variables (like API keys) into os.environ
load_dotenv()
# ─── DeepSeek API Configuration ──────────────────────────────────────
# Set default endpoint URL where the GAIA benchmark questions & files are hosted
DEFAULT_API_URL = os.getenv("GAIA_API_URL", "https://agents-course-unit4-scoring.hf.space")
# Retrieve the DeepSeek API key from environment variables
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
# Set base URL for the DeepSeek API (OpenAI-compatible endpoint)
DEEPSEEK_API_BASE = os.getenv("DEEPSEEK_API_BASE", "https://api.deepseek.com/v1")
# Specify the main LLM model name used for agent reasoning and execution
AGENT_MODEL = os.getenv("DEEPSEEK_AGENT_MODEL", "deepseek-chat")
# Specify the LLM model name used for formatting the final string answer
CLEANING_MODEL = os.getenv("DEEPSEEK_CLEANING_MODEL", "deepseek-chat")
# Number of parallel worker agents used by the pipeline
NUM_WORKERS = int(os.getenv("GAIA_NUM_WORKERS", "4"))
# Maximum tokens allowed for the answer-cleaning pass (raised so long
# comma-separated list Answers are not truncated)
CLEANING_MAX_TOKENS = int(os.getenv("CLEANING_MAX_TOKENS", "1024"))
# Two-stage per-Task timeout (Part B, decisions B1-B4): at the SOFT deadline
# the Agent is stopped and a best-effort Answer is synthesized from its partial
# Worklog (status "forced"); at the HARD deadline the Task is terminated and
# recorded as "timeout". Hard is enforced as a total-time ceiling: the synthesis
# call is clamped to the remaining hard-soft budget, so a Task ends no later
# than ~hard seconds after it starts (a hard < soft config yields a zero
# synthesis budget, i.e. a plain timeout).
TASK_SOFT_TIMEOUT_SEC = float(os.getenv("GAIA_TASK_SOFT_TIMEOUT", "300"))
TASK_HARD_TIMEOUT_SEC = float(os.getenv("GAIA_TASK_HARD_TIMEOUT", "360"))
# Bounds the forced-answer synthesis call so soft + synthesis stays under the
# hard cap (the worker also clamps it to the remaining hard-soft budget).
FORCE_ANSWER_TIMEOUT_SEC = float(os.getenv("GAIA_FORCE_ANSWER_TIMEOUT", "45"))
# Local vision model used by the 'inspect_image' tool for image attachments
VISION_MODEL_ID = os.getenv("VISION_MODEL_ID", "Qwen/Qwen2-VL-2B-Instruct")
# ─── Multimodal config (ADR-0002: local-only) ─────────────────────────
# Local compute device for the multimodal models (Qwen vision + faster-whisper).
# One of: auto | cpu | mps | cuda (default 'auto'). 'auto' probes torch for an
# accelerator — Apple MPS first, then CUDA — and falls back to 'cpu'. Note:
# faster-whisper clamps mps->cpu because CTranslate2 has no MPS backend (see
# _whisper_device_config).
LOCAL_DEVICE = os.getenv("LOCAL_DEVICE", "auto").strip().lower()
# Local audio model used by the 'transcribe_audio' tool (faster-whisper).
# Lightweight by default; use a larger model (e.g. 'small'/'medium') for
# better accuracy at the cost of speed and memory.
WHISPER_MODEL = os.getenv("WHISPER_MODEL", "tiny")
# Set USE_WHISPER_API=1 to prefer the hosted Whisper API (OpenAI or
# OpenRouter) over the local faster-whisper model. The API path is also used
# as a documented fallback when the local model fails and a key is configured.
USE_WHISPER_API = os.getenv("USE_WHISPER_API", "").strip().lower() in ("1", "true", "yes", "on")
# Number of evenly-spaced frames the 'extract_video_frames' tool samples from
# a video; each frame is fed to the local vision model via 'inspect_image'.
VIDEO_FRAMES_COUNT = int(os.getenv("VIDEO_FRAMES_COUNT", "4"))
# ─── Answer bundle & Run outputs (ticket 03) ────────────────────────────
# The committed artifact the Space submits from: a JSON file keyed by task_id,
# each entry holding {question, level, file_name, answer, worklog, timestamp}.
# Generated by local Runs and committed to the repo (per the spec's Answer
# bundle contract / ADR-0001). It is deliberately NOT gitignored.
ANSWER_BUNDLE_PATH = os.getenv(
"ANSWER_BUNDLE_PATH",
os.path.join(os.path.dirname(os.path.abspath(__file__)), "answer_bundle.json"),
)
# Local Run results CSV output path (configurable so tests and custom Runs can
# write elsewhere without touching the repo's gaia_results.csv)
RESULTS_CSV_PATH = os.getenv("GAIA_RESULTS_CSV", "gaia_results.csv")
# ─── Prompt for Final Answer Cleaning ──────────────────────────────────
# System prompt template that instructs LLM how to format outputs strictly for benchmark submission
FINAL_ANSWER_PROMPT = """You are an expert assistant in generating final answers with the highest possible precision.
Given a question and an initial answer, your task is to generate the clean final answer.
**Rules:**
- The final answer must be a number and/or text string OR as few words as possible OR a comma-separated list.
- If a number is requested, DO NOT use commas to separate thousands NOR units like USD, $, percentage, or % unless specified.
- If a string is requested, DO NOT use articles or abbreviations, write digits in plain text unless specified.
- If a comma-separated list is requested, apply the previous rules depending on whether it is a number or a string.
- If the final answer is a number, use a number, not a word.
- If the final answer is a string, start with a capital letter.
- If the final answer is a comma-separated list of numbers, put a space after each comma.
- If the final answer is a comma-separated list of strings, put a space after each comma and start with a lowercase letter.
- DO NOT add content that is not in the initial answer.
**Examples:**
Question: What is the largest city in California?
Initial Answer: The largest city in California is Los Angeles.
Final Answer: Los Angeles
Question: How many 'r's are in strawberry?
Initial Answer: There are 3 'r's in strawberry.
Final Answer: 3
Question: {question}
Initial Answer: {answer}
Final Answer: """
# System prompt template for the forced-answer path (Part B): when a Task hits
# the soft deadline, a best-effort final Answer is synthesized from the Agent's
# partial reasoning trace so something real is still submitted (status "forced").
FORCED_ANSWER_PROMPT = """You are an expert assistant that must produce a best-effort final answer from an INCOMPLETE reasoning trace, because the agent hit its time limit.
**Rules:**
- Use ONLY evidence present in the trace; do not invent facts.
- If the trace already contains a candidate answer or concrete data, prefer it.
- If there is not enough evidence, give the most reasonable concise answer you can.
- Apply the same formatting rules as a normal benchmark answer: a number and/or short text string, or a comma-separated list; no units/percentages unless specified; no extra commentary.
QUESTION:
{question}
PARTIAL REASONING TRACE:
{trace}
FINAL ANSWER: """
# Tuples of prefixes commonly outputted by reasoners to strip during cleanup
ANSWER_PREFIXES = (
"Final Answer:",
"FINAL ANSWER:",
"Answer:",
"Respuesta Final:",
)
def _extract_youtube_video_id(youtube_url: str) -> str | None:
"""Extracts a YouTube video ID from watch, short, embed, or live URLs."""
# Import URL parsing components inside function scope
from urllib.parse import parse_qs, urlparse
# Parse the input YouTube URL string into structured URL components
parsed = urlparse(youtube_url)
# Get domain host name in lowercase (e.g., youtube.com or youtu.be)
host = (parsed.netloc or "").lower()
# Get the URL path string (e.g., /watch or /shorts/xyz)
path = parsed.path or ""
# Check if the URL is a short link (youtu.be/ID)
if "youtu.be" in host:
# Extract ID from the path segment
video_id = path.strip("/").split("/")[0]
return video_id or None
# Check if URL belongs to standard youtube.com domains
if "youtube.com" in host:
# Try extracting query parameter 'v' from ?v=VIDEO_ID
query_id = parse_qs(parsed.query).get("v", [None])[0]
if query_id:
return query_id
# Check path-based formats like /shorts/, /embed/, /live/
for prefix in ("/shorts/", "/embed/", "/live/"):
if path.startswith(prefix):
video_id = path[len(prefix):].split("/")[0]
return video_id or None
# Fallback: regex search for standard 11-character video IDs
match = re.search(r"(?:v=|youtu\.be/)([a-zA-Z0-9_-]{11})", youtube_url)
# Return matched ID if found, otherwise None
return match.group(1) if match else None
def _get_transcript_text(transcript_data) -> str:
"""Normalizes different versions of youtube_transcript_api into a single string."""
# Convert transcript object to raw list of dictionaries if method exists
if hasattr(transcript_data, "to_raw_data"):
segments = transcript_data.to_raw_data()
else:
segments = transcript_data
# List to hold individual text lines
parts = []
# Loop over every subtitle segment retrieved
for segment in segments:
# Extract text property or dictionary key depending on API version return type
if hasattr(segment, "text"):
text = segment.text
elif isinstance(segment, dict):
text = segment.get("text", "")
else:
text = str(segment)
# Trim leading and trailing whitespace
text = str(text).strip()
# If segment contains text, append it to collector list
if text:
parts.append(text)
# Combine all segments into a single space-delimited text paragraph
return " ".join(parts).strip()
# Define local folder path where all attachment files are stored (relative to this script)
LOCAL_FILES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "files")
# Folder where the 'extract_video_frames' tool writes sampled frames during a
# Run (kept under files/ so it stays out of version control)
VIDEO_FRAMES_DIR = os.getenv(
"VIDEO_FRAMES_DIR",
os.path.join(LOCAL_FILES_DIR, "_video_frames"),
)
def _get_local_task_file(task_id: str, file_name: str) -> str:
"""Locates a pre-downloaded file in the local files directory.
Checks both original file_name and task_id-based filenames.
"""
if not os.path.exists(LOCAL_FILES_DIR):
return ""
# Strategy 1: Check by exact file_name if present
if file_name and str(file_name).strip():
safe_name = os.path.basename(file_name)
p1 = os.path.join(LOCAL_FILES_DIR, safe_name)
if os.path.exists(p1):
return p1
# Strategy 2: Check by task_id + extension from file_name (e.g., 2b3ef98c-....mp3)
if task_id:
if file_name and "." in file_name:
ext = file_name.lower().split(".")[-1]
p2 = os.path.join(LOCAL_FILES_DIR, f"{task_id}.{ext}")
if os.path.exists(p2):
return p2
# Strategy 3: Check by task_id alone
p3 = os.path.join(LOCAL_FILES_DIR, task_id)
if os.path.exists(p3):
return p3
# Strategy 4: Search for any file in the folder starting with task_id
for existing_file in os.listdir(LOCAL_FILES_DIR):
if existing_file.startswith(task_id):
return os.path.join(LOCAL_FILES_DIR, existing_file)
return ""
def _normalize_answer_for_submission(answer: str | None) -> str:
"""Ensures only the clean answer is processed by removing thinking tags."""
# Handle None value safely and strip whitespace from answer string
text = "" if answer is None else str(answer).strip()
# Strip ... reasoning blocks from DeepSeek R1 models using regex
text = re.sub(r'.*?', '', text, flags=re.DOTALL).strip()
# Clean orphaned closing tags if leftover
text = re.sub(r'', '', text).strip()
# Clean orphaned opening tags if leftover
text = re.sub(r'', '', text).strip()
# Iterate over standard answer prefixes and remove them if present
for prefix in ANSWER_PREFIXES:
if text.startswith(prefix):
text = text[len(prefix):].strip()
# Strip residual code ticks and bold asterisks surrounding answer
text = text.strip("`*")
# Return normalized clean text string
return text
# Sentinel placeholders used to interpolate values into FINAL_ANSWER_PROMPT.
# Swapping the template's {question}/{answer} markers for these control-char
# sentinels first means a Question or Answer that itself contains the literal
# substrings "{question}"/"{answer}" cannot be double-substituted.
_PROMPT_QUESTION_SENTINEL = "\x00GAIA_QUESTION\x00"
_PROMPT_ANSWER_SENTINEL = "\x00GAIA_ANSWER\x00"
def _build_cleaning_prompt(question: str | None, answer: str | None) -> str:
"""Builds the final-answer cleaning prompt.
Interpolates the Question and raw Answer via string replacement (never
``str.format``), so Answers containing ``{`` or ``}`` (e.g. JSON or sets)
do not crash the cleaning pass.
"""
prompt = (
FINAL_ANSWER_PROMPT
.replace("{question}", _PROMPT_QUESTION_SENTINEL)
.replace("{answer}", _PROMPT_ANSWER_SENTINEL)
.replace(_PROMPT_QUESTION_SENTINEL, "" if question is None else str(question))
.replace(_PROMPT_ANSWER_SENTINEL, "" if answer is None else str(answer))
)
return prompt
# Sentinels for the forced-answer prompt (Part B); same rationale as the
# cleaning prompt: a partial trace containing "{"/"}" (code, JSON) must not
# crash the synthesis call or double-substitute.
_FORCE_QUESTION_SENTINEL = "\x00GAIA_FORCE_QUESTION\x00"
_FORCE_TRACE_SENTINEL = "\x00GAIA_FORCE_TRACE\x00"
def _build_forced_answer_prompt(question: str | None, partial_trace: str | None) -> str:
"""Builds the forced-answer synthesis prompt.
Interpolates the Question and partial reasoning trace via string
replacement (never ``str.format``) so traces containing ``{``/``}`` do not
crash the synthesis call.
"""
return (
FORCED_ANSWER_PROMPT
.replace("{question}", _FORCE_QUESTION_SENTINEL)
.replace("{trace}", _FORCE_TRACE_SENTINEL)
.replace(_FORCE_QUESTION_SENTINEL, "" if question is None else str(question))
.replace(_FORCE_TRACE_SENTINEL, "" if partial_trace is None else str(partial_trace))
)
def _chat_answer(model_id: str, prompt: str, *, description: str,
max_tokens: int | None = None) -> str:
"""Runs a zero-temperature DeepSeek chat completion and strips it to a bare answer.
Shared by the answer-cleaning pass and the forced-answer synthesis (Part B):
both need the same retried, ``-stripped, prefix-stripped text. Raises
on failure — callers decide whether to fall back (cleaning) or treat it as
a timeout (forced synthesis).
"""
import openai
client = openai.OpenAI(base_url=DEEPSEEK_API_BASE, api_key=DEEPSEEK_API_KEY)
def _request():
return client.chat.completions.create(
model=model_id,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens if max_tokens is not None else CLEANING_MAX_TOKENS,
temperature=0.0, # Deterministic zero-temperature for consistent answers
)
completion = _call_with_retry(_request, description=description)
# Extract response text content from LLM choice completion
text = completion.choices[0].message.content
text = "" if text is None else str(text).strip()
# Remove internal reasoning blocks if present in output
text = re.sub(r'.*?', '', text, flags=re.DOTALL).strip()
# Strip prefixes, formatting marks, and code blocks from the output
for prefix in [*ANSWER_PREFIXES, "**", "```"]:
if text.startswith(prefix):
text = text[len(prefix):].strip()
# Return the string without trailing formatting chars
return text.rstrip("*`")
# ─── Retry-with-backoff ───────────────────────────────────────────────
# Transient model/network failures (429 rate limits, connection drops,
# timeouts, 5xx) must not fail a Task and cost a point. smolagents retries
# rate-limit errors internally, but not network errors; the answer-cleaning
# call uses the raw OpenAI client and has no built-in retry at all. These
# helpers give both paths a uniform retry-with-backoff.
_RETRYABLE_STATUS_CODES = {408, 429, 500, 502, 503, 504}
# Message markers that indicate a transient failure when the exception type
# cannot be classified directly (e.g. wrapped/unknown errors).
_RETRYABLE_MESSAGE_MARKERS = (
"429",
"rate limit",
"rate_limit",
"too many requests",
"500",
"502",
"503",
"504",
"internal server error",
"service unavailable",
"connection",
"timeout",
"timed out",
"temporarily unavailable",
)
def _is_retryable_error(exc: BaseException) -> bool:
"""Returns True when ``exc`` is a transient model/network failure worth retrying."""
try:
from openai import APIConnectionError, APITimeoutError, RateLimitError, APIStatusError
retryable_classes = (APIConnectionError, APITimeoutError, RateLimitError)
except Exception:
retryable_classes = ()
APIStatusError = None
# OpenAI SDK transient error classes
if retryable_classes and isinstance(exc, retryable_classes):
return True
if APIStatusError is not None and isinstance(exc, APIStatusError):
status = getattr(exc, "status_code", None)
if status is not None:
return status in _RETRYABLE_STATUS_CODES
# requests-level errors: retry connection drops and timeouts, and HTTP
# 5xx/429 surfaced by raise_for_status(), but not permanent 4xx client errors
if isinstance(exc, requests.exceptions.HTTPError):
status = getattr(exc.response, "status_code", None)
if status is not None:
return status in _RETRYABLE_STATUS_CODES
return False
if isinstance(exc, (requests.exceptions.ConnectionError, requests.exceptions.Timeout)):
return True
# Fallback heuristic: recognizable markers in the message text
message = str(exc).lower()
return any(marker in message for marker in _RETRYABLE_MESSAGE_MARKERS)
def _call_with_retry(fn, *, max_attempts: int = 4, base_delay: float = 1.0,
max_delay: float = 30.0, jitter: float = 0.5,
description: str = "call"):
"""Calls ``fn``, retrying transient errors with exponential backoff + jitter.
Args:
fn: The callable to invoke (must be re-invocable).
max_attempts: Total attempts before giving up (default 4).
base_delay: Base backoff seconds, doubled after each failure.
max_delay: Ceiling on the backoff delay.
jitter: +/- fraction of the delay applied as random jitter.
description: Human-readable label used in retry log messages.
Returns:
Whatever ``fn`` returns on success.
Raises:
The last exception raised by ``fn`` once retries are exhausted or the
error is not transient.
"""
attempt = 0
while True:
attempt += 1
try:
return fn()
except Exception as exc:
if attempt >= max_attempts or not _is_retryable_error(exc):
raise
delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
delay *= 1.0 + jitter * (2.0 * random.random() - 1.0)
print(f"⚠️ {description} failed (attempt {attempt}/{max_attempts}): {exc}; "
f"retrying in {max(0.0, delay):.1f}s")
time.sleep(max(0.0, delay))
def _run_with_timeout(fn, timeout: float):
"""Runs ``fn`` on a daemon helper thread, bounding its runtime to ``timeout``.
Python cannot forcibly kill a thread blocked in a call (e.g. a hung HTTP
request), so the call runs on a short-lived **daemon** thread that the
caller joins with a timeout. If the call is still alive when the timeout
elapses, ``TimeoutError`` is raised and the caller moves on; the wedged
helper thread cannot block the worker or the process at exit (daemon).
Returns:
Whatever ``fn`` returns on success.
Raises:
``TimeoutError`` if ``fn`` does not finish within ``timeout`` seconds.
Any exception raised by ``fn`` otherwise.
"""
result_box = {}
def _target():
try:
result_box["value"] = fn()
except Exception as exc:
result_box["error"] = exc
helper = threading.Thread(target=_target, daemon=True)
helper.start()
helper.join(timeout)
if helper.is_alive():
raise TimeoutError(f"Task exceeded {timeout:.0f}s")
if "error" in result_box:
raise result_box["error"]
return result_box["value"]
# ─── Worklog capture & Answer bundle (ticket 03) ────────────────────────
# A Worklog is the step-by-step trace of an Agent's Run — its reasoning steps
# (thought), tool calls, and observations. Each local Run serializes the full
# trace plus a per-tool summary with timing into a structured Worklog, and
# writes a committed Answer bundle keyed by task_id so the Space can submit
# without re-solving (ADR-0001 / spec's Answer bundle contract).
def _now_iso() -> str:
"""Returns the current UTC time in ISO-8601 (seconds) format."""
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
def _step_duration_sec(step) -> float:
"""Returns a step's duration in seconds from its smolagents Timing, or 0.0."""
timing = getattr(step, "timing", None)
if timing is None:
return 0.0
duration = getattr(timing, "duration", None)
if duration is None:
return 0.0
try:
return round(float(duration), 3)
except (TypeError, ValueError):
return 0.0
def _iter_action_steps(steps):
"""Yields only the Agent Action steps (skipping Task/Planning steps)."""
for step in steps:
# Action steps carry a step_number; Task/Planning steps do not.
if getattr(step, "step_number", None) is not None:
yield step
def _coerce_thought(model_output) -> str:
"""Coerces a step's model_output (str or list-of-dicts) into a single string."""
if model_output is None:
return ""
if isinstance(model_output, str):
return model_output
if isinstance(model_output, (list, tuple)):
parts = []
for item in model_output:
if isinstance(item, dict):
parts.append(str(item.get("text", item)))
else:
parts.append(str(item))
return "\n".join(parts)
return str(model_output)
def _coerce_observations(observations) -> str:
"""Coerces a step's observations (str or list) into a single string."""
if observations is None:
return ""
if isinstance(observations, (list, tuple)):
return "\n".join(str(o) for o in observations)
return str(observations)
def _coerce_tool_arguments(arguments):
"""Coerces a ToolCall's arguments into a JSON-serializable value.
smolagents ToolCall.arguments is usually already a dict, but older/newer
paths can carry a JSON string; parse it so the Worklog is stable JSON.
"""
if arguments is None:
return ""
if isinstance(arguments, str):
try:
return json.loads(arguments)
except (ValueError, TypeError):
return arguments
return arguments
def _serialize_action_step(step) -> dict:
"""Serializes one Agent Action step into a JSON-safe Worklog trace entry."""
tool_calls = []
for call in getattr(step, "tool_calls", None) or []:
tool_calls.append({
"name": getattr(call, "name", "") or "unknown",
"arguments": _coerce_tool_arguments(getattr(call, "arguments", None)),
})
entry = {
"step_number": getattr(step, "step_number", None),
"type": "final" if getattr(step, "is_final_answer", False) else "action",
"thought": _coerce_thought(getattr(step, "model_output", None)),
"tool_calls": tool_calls,
"observations": _coerce_observations(getattr(step, "observations", None)),
"code_action": getattr(step, "code_action", None) or "",
"duration_sec": _step_duration_sec(step),
}
error = getattr(step, "error", None)
if error is not None:
entry["error"] = str(error)
action_output = getattr(step, "action_output", None)
if action_output is not None:
entry["action_output"] = str(action_output)[:2000]
return entry
def _serialize_tool_summary(steps) -> list[dict]:
"""Computes a per-tool summary (call count + timing) from Agent steps.
Each step's duration is split evenly across the tools called in that step
(smolagents only exposes per-step timing), so per-tool totals do not
double-count time. Sorted by total time, then name, for a stable order.
"""
per_tool: dict[str, dict] = {}
for step in _iter_action_steps(steps):
tool_calls = getattr(step, "tool_calls", None) or []
if not tool_calls:
continue
share = _step_duration_sec(step) / len(tool_calls)
for call in tool_calls:
name = getattr(call, "name", "") or "unknown"
bucket = per_tool.setdefault(name, {"calls": 0, "total_sec": 0.0})
bucket["calls"] += 1
bucket["total_sec"] += share
summary = []
for name in sorted(per_tool, key=lambda n: (-per_tool[n]["total_sec"], n)):
bucket = per_tool[name]
total = round(bucket["total_sec"], 3)
calls = bucket["calls"]
summary.append({
"name": name,
"calls": calls,
"total_sec": total,
"avg_sec": round(total / calls, 3) if calls else 0.0,
})
return summary
def _serialize_worklog(steps, *, status: str = "completed") -> dict:
"""Serializes the Agent's recorded steps into a structured Worklog.
The Worklog holds the full trace (thought / tool call / observation per
step), a per-tool summary with timing, and the total Run duration for the
Task. ``status`` is "completed" on success, "forced" when the Answer was
synthesized from a partial trace at the soft deadline, or "error" on
failure.
"""
action_steps = list(_iter_action_steps(steps))
return {
"steps": [_serialize_action_step(s) for s in action_steps],
"tool_summary": _serialize_tool_summary(action_steps),
"total_duration_sec": round(sum(_step_duration_sec(s) for s in action_steps), 3),
"status": status,
}
def _make_bundle_entry(task_id: str, question: str, level, file_name: str,
answer: str, worklog: dict, timestamp: str | None = None) -> dict:
"""Builds one Answer bundle entry, keyed by ``task_id``.
The entry carries the Task's final Answer and its structured Worklog, plus
the Question/Level/file_name so bundled submissions are self-describing and
consistent with the results CSV (spec's Answer bundle contract).
"""
return {
"task_id": task_id,
"question": question,
"level": level if level is not None else "?",
"file_name": file_name if file_name else "None",
"answer": answer,
"worklog": worklog,
"timestamp": timestamp or _now_iso(),
}
def _load_answer_bundle(path: str | None = None) -> dict:
"""Loads the committed Answer bundle from ``path`` (default ANSWER_BUNDLE_PATH).
Returns ``{}`` when the bundle is absent or unreadable, so a Run degrades
to a full solve rather than crashing.
"""
path = path or ANSWER_BUNDLE_PATH
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, dict):
return data
except (OSError, ValueError, TypeError):
pass
return {}
def _save_answer_bundle(bundle: dict, path: str | None = None) -> str:
"""Atomically writes the Answer bundle to ``path`` (default ANSWER_BUNDLE_PATH).
Writes to a sibling temp file and renames over the target, so a crash
mid-write cannot corrupt the committed bundle.
"""
path = path or ANSWER_BUNDLE_PATH
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
tmp_path = f"{path}.tmp"
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(bundle, f, indent=2, ensure_ascii=False)
os.replace(tmp_path, path)
return path
# ─── Local device resolution (Part A: device auto-selection) ──────────
# A tiny lazy-import seam so the device resolver is unit-testable without
# torch installed (tests inject a fake module).
def _import_torch():
"""Lazily imports and returns the ``torch`` module (or raises ImportError)."""
import torch
return torch
_VALID_LOCAL_DEVICES = ("cpu", "mps", "cuda")
def _is_accelerator(local_device: str) -> bool:
"""True if ``local_device`` is a torch accelerator (mps or cuda)."""
return local_device in ("mps", "cuda")
def _resolve_local_device() -> str:
"""Resolves the local compute device for the multimodal models.
Reads the module-level ``LOCAL_DEVICE`` setting ('auto' | 'cpu' | 'mps' |
'cuda', default 'auto'):
- An explicit ``cpu``/``mps``/``cuda`` is returned as-is (no torch probe).
- ``auto`` (or any unrecognized value) probes torch for an accelerator,
preferring Apple MPS, then CUDA, and finally falls back to ``cpu``. If
torch is unavailable, ``cpu`` is returned.
"""
if LOCAL_DEVICE in _VALID_LOCAL_DEVICES:
return LOCAL_DEVICE
try:
torch = _import_torch()
if torch.backends.mps.is_available():
return "mps"
if torch.cuda.is_available():
return "cuda"
except Exception:
pass
return "cpu"
def _whisper_device_config(local_device: str):
"""Maps a resolved local device to faster-whisper's ``(device, compute_type)``.
CTranslate2 (faster-whisper's backend) accepts only ``"cpu"`` or ``"cuda"``
— there is no MPS backend — so ``mps`` is clamped to ``cpu``. ``compute_type``
is ``"int8"`` on cpu (lightweight) and ``"float16"`` on cuda.
"""
if local_device == "cuda":
return "cuda", "float16"
return "cpu", "int8"
def _vision_load_config(torch, local_device: str):
"""Returns ``(torch_dtype, device_map)`` for the Qwen vision model.
``torch_dtype`` is ``float16`` on accelerators (mps/cuda) to halve memory
and ``float32`` on cpu. ``device_map='auto'`` is used only for cuda, because
transformers' device_map has no MPS backend.
"""
dtype = torch.float16 if _is_accelerator(local_device) else torch.float32
device_map = "auto" if local_device == "cuda" else None
return dtype, device_map
# ─── Agent Tools ──────────────────────────────────────────────────
# Shared faster-whisper model singleton (loaded once, reused across workers)
_whisper_model = None
_whisper_lock = threading.Lock()
def _transcribe_audio_local(audio_path: str) -> str:
"""Transcribes ``audio_path`` with the local faster-whisper model.
The model (configurable via the WHISPER_MODEL setting, default 'tiny') is
loaded once and reused across worker threads. Returns the transcript text,
or an ``"Error: ..."`` string on failure so the Agent can react without
crashing the Run (matching the other tools).
"""
global _whisper_model
try:
# Import the lightweight local model only when the tool is first used
from faster_whisper import WhisperModel
# Load the shared model once (double-checked locking, like the vision model)
if _whisper_model is None:
with _whisper_lock:
if _whisper_model is None:
# Route through the shared local-device resolver. faster-whisper
# has no MPS backend, so mps is clamped to cpu (int8); cuda uses
# float16. See _whisper_device_config.
device, compute_type = _whisper_device_config(_resolve_local_device())
_whisper_model = WhisperModel(WHISPER_MODEL, device=device, compute_type=compute_type)
# Beam size 1 keeps latency low for the Agent loop
segments, _info = _whisper_model.transcribe(audio_path, beam_size=1)
parts = [
segment.text.strip()
for segment in segments
if segment.text and segment.text.strip()
]
transcript = " ".join(parts).strip()
if not transcript:
return f"Error: Local transcription returned no text for '{audio_path}'."
return transcript
except Exception as e:
return f"Error: Local transcription failed (model '{WHISPER_MODEL}'): {e}"
def _transcribe_audio_api(audio_path: str) -> str:
"""Transcribes ``audio_path`` with the hosted Whisper API (documented fallback).
Requires ``OPENAI_API_KEY`` (whisper-1) or ``OPENROUTER_API_KEY``
(openai/whisper-large-v3-turbo). Returns the transcript text, or an
``"Error: ..."`` string on failure.
"""
try:
# Import openai library inside tool execution logic
import openai
# Check for OpenAI or OpenRouter API keys (for Whisper transcription only;
# separate from the DeepSeek agent config since DeepSeek has no audio API)
api_key = os.getenv("OPENAI_API_KEY") or os.getenv("OPENROUTER_API_KEY")
# Select base API endpoint depending on which service key is set
base_url = "https://api.openai.com/v1" if os.getenv("OPENAI_API_KEY") else "https://openrouter.ai/api/v1"
# Choose Whisper model based on API provider
model = "whisper-1" if os.getenv("OPENAI_API_KEY") else "openai/whisper-large-v3-turbo"
# Return warning message if neither API key is present
if not api_key:
return "Error: Audio transcription fallback requires configuring OPENAI_API_KEY or OPENROUTER_API_KEY."
# Instantiate OpenAI client with configured API host & key
client = openai.OpenAI(base_url=base_url, api_key=api_key)
# Open target audio file in binary reading mode
with open(audio_path, "rb") as audio_file:
# Request audio transcription from Whisper model endpoint
transcription = client.audio.transcriptions.create(
model=model,
file=audio_file,
)
# Return transcript text content
return transcription.text
except Exception as e:
# Return error string if transcription fails
return f"Error transcribing audio via API: {e}"
@tool
def transcribe_audio(audio_path: str) -> str:
"""Transcribes an audio file (MP3, WAV, M4A, etc.).
Uses the lightweight local faster-whisper model (configurable via the
WHISPER_MODEL setting, default 'tiny'). If USE_WHISPER_API=1, or the local
model fails and an API key is configured, it falls back to the hosted
Whisper API (OpenAI or OpenRouter) as a documented fallback.
Args:
audio_path: The local path to the audio file (e.g., 'downloads/file.mp3').
"""
if not USE_WHISPER_API:
# Local-first per ADR-0002; fall back to the API only when the local
# model fails AND a hosted key is configured
local = _transcribe_audio_local(audio_path)
if not local.startswith("Error:"):
return local
if os.getenv("OPENAI_API_KEY") or os.getenv("OPENROUTER_API_KEY"):
return _transcribe_audio_api(audio_path)
return local
# Explicit opt-in to the hosted API
return _transcribe_audio_api(audio_path)
@tool
def get_youtube_transcript(youtube_url: str) -> str:
"""Gets the official transcript/subtitles of a YouTube video.
Args:
youtube_url: The full URL of the YouTube video (e.g., 'https://www.youtube.com/watch?v=...').
"""
try:
# Import YouTube transcript API library
from youtube_transcript_api import YouTubeTranscriptApi
# Extract video ID from input URL string using helper function
video_id = _extract_youtube_video_id(youtube_url)
if not video_id:
return f"Error: Could not extract video ID from URL: {youtube_url}"
# Preference list of subtitle languages to query
preferred_languages = ['en', 'es', 'fr', 'de']
try:
# Attempt direct fetching of preferred language subtitles
transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=preferred_languages)
except Exception:
# Fallback: list available transcripts and fetch best matching language
transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
transcript = transcript_list.find_transcript(preferred_languages).fetch()
# Normalize fetched transcript object into single text string
return _get_transcript_text(transcript)
except Exception as e:
# Return error description string if retrieval fails
return f"Error obtaining YouTube transcript: {e}."
def _video_frame_timestamps(duration: float, num_frames: int) -> list[float]:
"""Computes N evenly-spaced frame timestamps within a video's duration.
Timestamps are the centers of N equal slices of ``[0, duration]``, so the
first and last frames are never the exact first/last instant (often a
frozen title card or a black frame). Clamped to ``[0, duration]``.
Args:
duration: The video duration in seconds.
num_frames: The number of evenly-spaced frames to sample (values below
1 are clamped to 1).
Returns:
A list of timestamps in seconds, in ascending order.
"""
n = max(1, int(num_frames))
if duration <= 0:
return [0.0]
step = duration / n
return [round((i + 0.5) * step, 3) for i in range(n)]
def _probe_video_duration(video_path: str) -> float:
"""Returns the duration (seconds) of a video file via ffprobe."""
result = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1", video_path],
capture_output=True, text=True, timeout=60,
)
if result.returncode != 0:
raise RuntimeError(f"ffprobe failed: {result.stderr.strip()}")
return float(result.stdout.strip())
def _resolve_frame_count(num_frames) -> int:
"""Coerces an LLM-supplied frame count to a valid positive int.
The Agent calls the tool with whatever it inferred, so ``num_frames`` may
be None, a float-string ("5.5"), or a word. Anything that is not a
positive int falls back to the VIDEO_FRAMES_COUNT setting rather than
failing the whole tool call.
"""
try:
n = int(num_frames)
if n > 0:
return n
except (TypeError, ValueError):
pass
return VIDEO_FRAMES_COUNT
def _extract_frames_from_video(video_path: str, num_frames: int, job_id: str,
out_dir: str | None = None) -> list[str]:
"""Probes ``video_path`` and extracts N evenly-spaced JPEG frames.
Frames are written as ``frame_{job_id}_NN.jpg`` into ``out_dir`` (defaults
to the video-frames folder) and returned as absolute paths. Requires
ffmpeg/ffprobe on PATH. Raises on failure so the caller can clean up any
partial files.
"""
out_dir = out_dir or VIDEO_FRAMES_DIR
os.makedirs(out_dir, exist_ok=True)
duration = _probe_video_duration(video_path)
timestamps = _video_frame_timestamps(duration, num_frames)
frame_paths = []
for i, ts in enumerate(timestamps, start=1):
frame_path = os.path.join(out_dir, f"frame_{job_id}_{i:02d}.jpg")
subprocess.run(
["ffmpeg", "-y", "-ss", f"{ts:.3f}", "-i", video_path,
"-frames:v", "1", "-q:v", "2", frame_path],
capture_output=True, text=True, timeout=180, check=True,
)
frame_paths.append(frame_path)
return frame_paths
def _cleanup_job_artifacts(job_id: str) -> None:
"""Removes any partial files for ``job_id`` from the video-frames folder.
Covers the downloaded video (and its ``.part``/``.ytdl`` temp files, which
share the ``video_{job_id}`` prefix) plus any frames already written before
a failure, so a failed extraction never orphans files on disk.
"""
if not os.path.isdir(VIDEO_FRAMES_DIR):
return
for name in os.listdir(VIDEO_FRAMES_DIR):
if name.startswith(f"video_{job_id}") or name.startswith(f"frame_{job_id}"):
try:
os.remove(os.path.join(VIDEO_FRAMES_DIR, name))
except OSError:
pass
@tool
def extract_video_frames(video_url: str, num_frames: int | None = None) -> str:
"""Downloads a video and extracts N evenly-spaced frames to local files.
Use this for the VISUAL content of a video (charts, diagrams, on-screen
text, people/objects). For the SPOKEN content use 'get_youtube_transcript'
instead. The sampled frames are written to the video-frames folder and can
be read one-by-one with 'inspect_image'. Requires yt-dlp and
ffmpeg/ffprobe on PATH.
Args:
video_url: The full URL of the video (e.g., 'https://www.youtube.com/watch?v=...').
num_frames: Number of evenly-spaced frames to extract (defaults to the
VIDEO_FRAMES_COUNT setting, usually 4).
"""
job_id = None
try:
import yt_dlp
n = _resolve_frame_count(num_frames)
os.makedirs(VIDEO_FRAMES_DIR, exist_ok=True)
job_id = str(int(time.time() * 1000))
base_prefix = os.path.join(VIDEO_FRAMES_DIR, f"video_{job_id}")
# Download the video. Bounding the height keeps the download small and
# fast; noplaylist avoids pulling an entire playlist for one clip.
ydl_opts = {
"format": "bv*[height<=720]+ba/b[height<=720]/b",
"outtmpl": base_prefix + ".%(ext)s",
"noplaylist": True,
"quiet": True,
"no_warnings": True,
"merge_output_format": "mp4",
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.extract_info(video_url, download=True)
# Locate the actual downloaded file (the final extension depends on the
# stream/merge chosen by yt-dlp, so glob for our prefix)
downloaded = [
os.path.join(VIDEO_FRAMES_DIR, f)
for f in os.listdir(VIDEO_FRAMES_DIR)
if f.startswith(os.path.basename(base_prefix))
]
if not downloaded:
return f"Error: Download completed but no video file was found for {video_url}"
video_path = downloaded[0]
try:
# Probe the duration, then sample N evenly-spaced frame timestamps
frame_paths = _extract_frames_from_video(video_path, n, job_id)
finally:
# The source video is no longer needed once frames are extracted
try:
os.remove(video_path)
except OSError:
pass
lines = [f"Extracted {len(frame_paths)} evenly-spaced frames from the video:"]
lines += [f" - {p}" for p in frame_paths]
lines.append("Inspect each frame with 'inspect_image' to read its visual content.")
return "\n".join(lines)
except Exception as e:
# Clean up this job's partial artifacts so failures never orphan files
if job_id:
_cleanup_job_artifacts(job_id)
return f"Error extracting video frames: {e}"
@tool
def inspect_pdf(pdf_path: str) -> str:
"""Extracts all text from a PDF file in a structured format.
Args:
pdf_path: The local path to the PDF file (e.g., 'downloads/document.pdf').
"""
try:
# Import pdfplumber library for text extraction
import pdfplumber
# Container to accumulate text page by page
texts = []
# Open PDF file in pdfplumber handle context
with pdfplumber.open(pdf_path) as pdf:
# Iterate through each page of document with index
for i, page in enumerate(pdf.pages):
# Extract text content from current page
content = page.extract_text()
if content:
# Append formatted header with page content
texts.append(f"--- Page {i+1} ---\n{content}")
# Join extracted page texts and truncate to 60,000 chars to avoid prompt context blowup
return "\n".join(texts)[:60000]
except Exception as e:
# Return error message if reading PDF fails
return f"Error reading PDF: {e}"
@tool
def inspect_excel(file_path: str) -> str:
"""Reads an Excel file (.xlsx, .xls) and returns a summary of its sheets.
Args:
file_path: The local path to the Excel file (e.g., 'downloads/data.xlsx').
"""
try:
# Load Excel workbook structure into pandas ExcelFile reader
xl = pd.ExcelFile(file_path)
# Start summary list with sheet names found
summary = [f"Sheets found: {xl.sheet_names}"]
# Loop through each sheet name inside workbook
for sheet in xl.sheet_names:
# Read first 20 rows of current sheet into DataFrame
df = pd.read_excel(xl, sheet_name=sheet).head(20)
# Format sheet header and sample string table into summary
summary.append(f"\n--- Sheet: {sheet} (Sample) ---\n{df.to_string()}")
# Return joined summary string capped at 50,000 characters max
return "\n".join(summary)[:50000]
except Exception as e:
# Return error message string if Excel read fails
return f"Error reading Excel: {e}"
@tool
def read_file_as_text(file_path: str) -> str:
"""Reads a plain text file like CSV, JSON, TXT, etc.
Args:
file_path: The local path to the text file (e.g., 'downloads/report.csv').
"""
try:
# Open target text file with UTF-8 encoding, replacing invalid characters
with open(file_path, "r", encoding="utf-8", errors="replace") as f:
content = f.read()
# Truncate content to 50,000 characters if file is too large
if len(content) > 50000:
content = content[:50000] + "\n...[CONTENT TRUNCATED]..."
# Return file text contents
return content
except Exception as e:
# Return error message string if text read fails
return f"Error reading file: {e}"
# Shared vision model singletons (loaded once, reused across all worker threads)
_vision_model = None
_vision_processor = None
_vision_lock = threading.Lock()
@tool
def inspect_image(image_path: str, question: str = "") -> str:
"""Extracts the content of an image using a local vision model (Qwen2-VL-2B) plus OCR.
Use this tool for image attachments (.png, .jpg, .jpeg, etc.). Pass the task
question so the model focuses on the exact detail needed for the answer.
Args:
image_path: The local path to the image file (e.g., 'downloads/image.png').
question: The task question to focus the extraction on (optional).
"""
global _vision_model, _vision_processor
try:
# Import heavy libraries only when the tool is first invoked
torch = _import_torch()
from transformers import AutoProcessor
# Resolve the correct auto class (newer transformers name, with older fallback)
try:
from transformers import AutoModelForImageTextToText as _AutoVL
except ImportError:
from transformers import AutoModelForVision2Seq as _AutoVL
# Shared local-device resolver (LOCAL_DEVICE env: auto|cpu|mps|cuda)
device = _resolve_local_device()
# ── Load the shared model + processor once (double-checked locking) ──
if _vision_model is None:
with _vision_lock:
if _vision_model is None:
dtype, device_map = _vision_load_config(torch, device)
_vision_processor = AutoProcessor.from_pretrained(VISION_MODEL_ID)
_vision_model = _AutoVL.from_pretrained(
VISION_MODEL_ID,
torch_dtype=dtype,
device_map=device_map,
)
_vision_model.eval()
if device == "cpu":
_vision_model.to("cpu")
elif device == "mps":
_vision_model.to("mps")
# ── Open the image and normalize to RGB ──
image = Image.open(image_path).convert("RGB")
# ── Build a question-directed prompt for the VLM ──
if question and str(question).strip():
prompt_text = (
"Look carefully at this image and answer the following question. "
"Extract the exact text, numbers, names, or details shown in the image "
"that are needed to answer. Be precise and do not guess.\n"
f"Question: {question}"
)
else:
prompt_text = (
"Describe the contents of this image in detail. "
"Include all visible text, numbers, labels, and names exactly as shown."
)
# ── Run inference (serialized across worker threads) ──
with _vision_lock:
try:
# Preferred path: chat template (Qwen2-VL style)
messages = [{
"role": "user",
"content": [
{"type": "image"},
{"type": "text", "text": prompt_text},
],
}]
text = _vision_processor.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
inputs = _vision_processor(
text=[text], images=[image], return_tensors="pt"
)
except (AttributeError, NotImplementedError):
# Fallback path: plain image + text inputs
inputs = _vision_processor(
images=image, text=prompt_text, return_tensors="pt"
)
# Move inputs to the compute device (accelerators only)
if _is_accelerator(device):
inputs = {
k: v.to(device) for k, v in inputs.items()
if isinstance(v, torch.Tensor)
}
with torch.no_grad():
outputs = _vision_model.generate(
**inputs,
max_new_tokens=512,
do_sample=False,
)
# Decode only the newly generated tokens
input_len = inputs["input_ids"].shape[1]
generated = _vision_processor.batch_decode(
outputs[:, input_len:], skip_special_tokens=True
)
vlm_text = " ".join(generated).strip()
# ── Best-effort OCR pass for literal text ──
ocr_text = ""
try:
import pytesseract
ocr_text = pytesseract.image_to_string(image).strip()
if len(ocr_text) > 3000:
ocr_text = ocr_text[:3000] + "\n...[OCR TRUNCATED]..."
except Exception:
ocr_text = "" # OCR unavailable (e.g., Tesseract binary not installed)
# ── Combine results and cap the total length ──
result = f"VISION MODEL OUTPUT:\n{vlm_text}"
if ocr_text:
result += f"\n\nOCR TEXT:\n{ocr_text}"
return result[:8000]
except Exception as e:
return f"Error analyzing image: {e}"
@tool
def inspect_docx(file_path: str) -> str:
"""Extracts all text (paragraphs and tables) from a Word .docx file.
Args:
file_path: The local path to the .docx file (e.g., 'downloads/document.docx').
"""
try:
# Import python-docx library inside tool execution logic
import docx
# Open the Word document
document = docx.Document(file_path)
# Container for extracted text parts
parts = []
# Collect non-empty paragraphs
for para in document.paragraphs:
text = para.text.strip()
if text:
parts.append(text)
# Collect table contents row by row
for t_idx, table in enumerate(document.tables):
parts.append(f"\n--- Table {t_idx + 1} ---")
for row in table.rows:
cells = [cell.text.strip() for cell in row.cells]
parts.append(" | ".join(cells))
# Return joined text capped at 50,000 characters
return "\n".join(parts)[:50000]
except Exception as e:
# Return error message if .docx reading fails
return f"Error reading .docx: {e}"
@tool
def inspect_pptx(file_path: str) -> str:
"""Extracts all text from each slide of a PowerPoint .pptx file.
Args:
file_path: The local path to the .pptx file (e.g., 'downloads/presentation.pptx').
"""
try:
# Import python-pptx library inside tool execution logic
from pptx import Presentation
# Open the PowerPoint presentation
prs = Presentation(file_path)
# Container for extracted slide text
parts = []
# Loop over every slide with its index
for s_idx, slide in enumerate(prs.slides):
slide_texts = []
# Iterate shapes (text boxes, tables, etc.) in the slide
for shape in slide.shapes:
if hasattr(shape, "text") and shape.text.strip():
slide_texts.append(shape.text.strip())
if slide_texts:
# Append formatted slide header with its text lines
parts.append(f"--- Slide {s_idx + 1} ---\n" + "\n".join(slide_texts))
# Return joined slide text capped at 50,000 characters
return "\n\n".join(parts)[:50000]
except Exception as e:
# Return error message if .pptx reading fails
return f"Error reading .pptx: {e}"
def _safe_extract_all(zf, extract_dir: str) -> tuple[list[str], list[str]]:
"""Extracts every member of an open ``zipfile.ZipFile`` without path traversal.
Never uses ``extractall()``: each member's resolved destination is checked
to stay inside ``extract_dir``, absolute paths and ``..`` traversal are
rejected, and members are streamed out as plain files so symlink entries
cannot be followed outside the extraction folder.
Returns:
A ``(extracted, skipped)`` pair: the names of members written, and the
names of unsafe members that were rejected.
"""
base = os.path.realpath(extract_dir)
os.makedirs(base, exist_ok=True)
extracted: list[str] = []
skipped: list[str] = []
for member in zf.infolist():
# Normalize Windows separators and reject absolute/escaping paths
member_path = member.filename.replace("\\", "/")
if member_path.startswith("/") or ".." in member_path.split("/"):
skipped.append(member.filename)
continue
target = os.path.realpath(os.path.join(base, member_path))
# Guard: resolved target must stay inside the extraction folder
if not (target == base or target.startswith(base + os.sep)):
skipped.append(member.filename)
continue
if member.is_dir():
os.makedirs(target, exist_ok=True)
extracted.append(member.filename)
continue
# Stream the member out as a plain file (never follow symlinks)
os.makedirs(os.path.dirname(target) or base, exist_ok=True)
with zf.open(member, "r") as src, open(target, "wb") as dst:
shutil.copyfileobj(src, dst)
extracted.append(member.filename)
return extracted, skipped
@tool
def extract_zip(zip_path: str, extract_dir: str = "") -> str:
"""Lists and extracts the contents of a .zip archive into a folder.
Args:
zip_path: The local path to the .zip file (e.g., 'downloads/files.zip').
extract_dir: The folder to extract into (defaults to the files directory).
"""
try:
# Import zipfile library inside tool execution logic
import zipfile
# Default extraction folder next to the attachment files
if not extract_dir:
extract_dir = os.path.join(LOCAL_FILES_DIR, "_extracted")
# Open the archive and extract entries safely (no extractall / zip-slip)
with zipfile.ZipFile(zip_path, "r") as zf:
extracted, skipped = _safe_extract_all(zf, extract_dir)
# Build summary listing every extracted entry
summary = [f"Extracted {len(extracted)} entries to '{extract_dir}':"]
for name in extracted:
summary.append(f" - {name}")
# Surface any rejected unsafe entries so the Agent knows they were skipped
if skipped:
summary.append(f"\n⚠️ Skipped {len(skipped)} unsafe entries (path traversal):")
for name in skipped:
summary.append(f" - {name}")
# Return listing capped at 20,000 characters
return "\n".join(summary)[:20000]
except Exception as e:
# Return error message if .zip extraction fails
return f"Error extracting .zip: {e}"
# ─── Main Agent Class ──────────────────────────────────────────────
class _RetryingOpenAIServerModel(OpenAIServerModel):
"""OpenAIServerModel that retries transient model/network errors with backoff.
smolagents already retries rate-limit errors internally; this wrapper adds
retries for connection/timeout/5xx failures so a transient network blip
cannot fail a Task. The Agent's step memory is preserved across retries
(unlike retrying the whole ``agent.run``), so no tool side-effects repeat.
"""
def generate(self, messages, **kwargs):
# Zero-arg super() only works directly in a method body — it is NOT
# propagated into nested lambdas (no __class__ cell), so capture it
# here and let the retry lambda call the bound base method.
base = super()
return _call_with_retry(
lambda: base.generate(messages, **kwargs),
description="model call",
)
class GAIASolverAgent:
def __init__(self, *, multimodal: bool = True):
"""Builds the ReAct solver Agent.
Args:
multimodal: When True (default, local harness), the full toolset is
registered, including the local vision model (inspect_image),
local audio transcription (transcribe_audio), and video frame
sampling (extract_video_frames). When False (live-on-Space
fallback), only text/web + document tools are registered, so no
heavy local model is ever loaded — per ADR-0002 the Space
fallback is text/web-only (no GPU, no hosted key).
"""
# Validate that DEEPSEEK_API_KEY is available before proceeding
if not DEEPSEEK_API_KEY:
raise ValueError("⚠️ The environment variable 'DEEPSEEK_API_KEY' is not configured.")
# Initialize retrying model pointing to configured API base and model
self.main_model = _RetryingOpenAIServerModel(
model_id=AGENT_MODEL,
api_base=DEEPSEEK_API_BASE,
api_key=DEEPSEEK_API_KEY,
)
# Text/web-only base + document tools, present in both modes
tools = [
DuckDuckGoSearchTool(), # Web search tool
VisitWebpageTool(), # Web scraping/browsing tool
get_youtube_transcript, # YouTube transcript extraction tool (spoken content)
inspect_pdf, # PDF text extraction tool
inspect_excel, # Excel sheet inspector tool
read_file_as_text, # General text/CSV/JSON reader tool
inspect_docx, # Word .docx text extraction tool
inspect_pptx, # PowerPoint .pptx text extraction tool
extract_zip, # Zip archive extraction tool
]
if multimodal:
# Local-model tools (ADR-0002): vision, audio, and video frames.
# Excluded from the Space fallback so Qwen2-VL / faster-whisper are
# never loaded on the constrained free Space.
tools += [
transcribe_audio, # Audio transcription tool (local faster-whisper)
extract_video_frames, # Video frame-sampling tool (visual content)
inspect_image, # Image vision tool (Qwen2-VL-2B + OCR)
]
# Instantiate CodeAgent from smolagents framework
self.agent = CodeAgent(
tools=tools,
model=self.main_model,
# Explicit Python module imports authorized for code-execution agent actions
additional_authorized_imports=[
# ── Standard Python Libraries ──
"sys", "os", "pathlib", "glob", "shutil", "tempfile",
"hashlib", "base64", "math", "statistics", "random",
"re", "string", "datetime", "time", "json", "csv",
"xml", "html", "urllib", "urllib.parse", "io", "copy",
"collections", "itertools", "functools", "operator",
"typing", "dataclasses", "unicodedata", "wave", "zipfile",
"tarfile", "sqlite3", "uuid", "ast", "inspect",
# ── Data Science & Math ──
"numpy", "pandas", "scipy", "scipy.stats", "scipy.optimize",
"matplotlib", "matplotlib.pyplot", "seaborn", "sympy",
"sklearn", "stat",
# ── File & Image Processing ──
"PIL", "PIL.Image", "PIL.ImageDraw", "cv2",
"openpyxl", "xlrd", "pdfplumber", "PyPDF2", "pypdf",
"fitz", "docx", "pptx",
# ── Web, Scraping & Networking ──
"requests", "bs4", "lxml", "queue"
],
# Set upper limit on agent reasoning steps per task execution run
max_steps=25,
)
def _clean_answer(self, question: str, raw_answer: str) -> str:
"""Uses the DeepSeek model to structure the clean final answer synchronously per thread."""
try:
# Build the cleaning prompt via replacement interpolation (safe for
# Answers/Questions containing braces) and raise the token budget so
# long comma-separated list Answers are not truncated
prompt = _build_cleaning_prompt(question, raw_answer)
return _chat_answer(CLEANING_MODEL, prompt, description="answer cleaning")
except Exception:
# Fallback to standard normalization regex if API formatting call fails
return _normalize_answer_for_submission(raw_answer)
def __call__(self, task_id: str, question: str, file_name: str) -> str:
"""Solves a Task and returns just the final Answer (see ``solve``)."""
return self.solve(task_id, question, file_name)[0]
def solve(self, task_id: str, question: str, file_name: str,
file_path: str | None = None) -> tuple[str, dict]:
"""Solves a Task and returns ``(final_answer, worklog)``.
``worklog`` is the structured serialization of this Agent's recorded
ReAct steps (thought / tool call / observation per step) plus a per-tool
summary with timing, captured fresh from the Agent's memory after the
run. On failure it returns an ``Error: ...`` Answer and an empty Worklog
so the Run continues; the Task stays out of the Answer bundle and is
re-solved on the next Run.
``file_path`` is the local path to the Task's Attachment. When omitted,
the pre-downloaded file is located via ``_get_local_task_file`` (local
harness behavior); when provided (Space live fallback, where the
Attachment was just downloaded from the server), it is used directly.
"""
# Ensure the Worklog only reflects THIS Task. Agent instances are reused
# across Tasks in the worker pool, and run() resets memory only once it
# actually starts, so clear it before any work can fail (a failure before
# run() would otherwise serialize the previous Task's stale trace).
try:
self.agent.memory.reset()
except Exception:
pass # fresh agent: memory is already empty
try:
# Locate the Attachment: an explicitly-provided download path (Space
# live fallback) wins; otherwise look for the pre-downloaded file in
# the local files directory (local harness)
if file_path is None:
file_path = _get_local_task_file(task_id, file_name)
# Construct agent task prompt
prompt = f"TASK: {question}\n\n"
prompt += "### FORMATTING RULES:\n"
prompt += "- If you need to write or run Python code, you MUST wrap the code block strictly inside \n[your python code]\n tags.\n"
prompt += "- Ensure there is a newline after the opening tag and before the closing tag.\n\n"
if file_path:
ext = file_path.lower().split('.')[-1]
if ext in ('png', 'jpg', 'jpeg', 'gif', 'bmp', 'webp'):
prompt += f"🖼️ IMAGE saved at '{file_path}'. Use 'inspect_image' with the task question to extract its content.\n\n"
elif ext in ('mp3', 'wav', 'm4a', 'ogg', 'flac'):
prompt += f"🎙️ AUDIO saved at '{file_path}'. Use 'transcribe_audio' (local faster-whisper).\n\n"
elif ext in ('pdf',):
prompt += f"📄 PDF saved at '{file_path}'. Use 'inspect_pdf' or Python libraries like pdfplumber/fitz.\n\n"
elif ext in ('xlsx', 'xls'):
prompt += f"📊 EXCEL saved at '{file_path}'. Use 'inspect_excel' or Python pandas.\n\n"
elif ext in ('docx',):
prompt += f"📝 WORD saved at '{file_path}'. Use 'inspect_docx' to extract text and tables.\n\n"
elif ext in ('pptx',):
prompt += f"📽️ POWERPOINT saved at '{file_path}'. Use 'inspect_pptx' to extract slide text.\n\n"
elif ext in ('zip',):
prompt += f"🗜️ ZIP ARCHIVE saved at '{file_path}'. Use 'extract_zip' to list and extract its contents.\n\n"
else:
prompt += f"📎 FILE saved at '{file_path}'. Use 'read_file_as_text' or open with Python.\n\n"
elif file_name and str(file_name).strip():
prompt += f"⚠️ Note: Attachment '{file_name}' was specified but not found locally in files directory.\n\n"
if "youtube.com" in question.lower() or "youtu.be" in question.lower():
prompt += (
"📹 YOUTUBE: Use 'get_youtube_transcript' for the SPOKEN content, "
"and 'extract_video_frames' to sample frames if the answer depends "
"on VISUAL content (charts, diagrams, on-screen text). Use whichever "
"the Question needs — or both.\n\n"
)
prompt += "CRITICAL INSTRUCTIONS:\n1. Answer concisely.\n2. Provide the exact data required.\n"
# Run agent loop (text-only; image attachments are handled via the
# 'inspect_image' tool). Transient model/network failures are retried
# inside _RetryingOpenAIServerModel.generate, which preserves the
# Agent's step memory (retrying the whole loop would repeat tool
# side-effects)
raw_answer = self.agent.run(prompt)
raw_answer_str = str(raw_answer).strip()
raw_answer_str = re.sub(r'.*?', '', raw_answer_str, flags=re.DOTALL).strip()
final_answer = _normalize_answer_for_submission(
self._clean_answer(question, raw_answer_str)
)
# Capture the Agent's recorded steps (thought / tool call /
# observation) into the structured Worklog before anything else
# can mutate this Agent's memory
worklog = _serialize_worklog(self.agent.memory.steps, status="completed")
return final_answer, worklog
except Exception as e:
# Keep the partial trace gathered before the failure so the Worklog
# still shows how far the Agent got (status "error" + the reason)
memory = getattr(self.agent, "memory", None)
steps = memory.steps if memory is not None else []
worklog = _serialize_worklog(steps, status="error")
worklog["error"] = str(e)
return f"Error: {e}", worklog
# ─── Forced-answer synthesis helpers (Part B) ─────────────────────────
# When a Task hits the soft deadline, its partial Worklog is snapshotted and a
# best-effort Answer is synthesized from it (decisions B1-B4) so something real
# is still submitted with status "forced".
def _format_partial_trace(worklog: dict) -> str:
"""Formats a (partial) Worklog into a compact readable trace for synthesis."""
lines = []
for step in worklog.get("steps", []):
thought = (step.get("thought") or "").strip()
if thought:
lines.append(f"Thought: {thought}")
for call in step.get("tool_calls") or []:
name = call.get("name", "unknown")
args = _coerce_tool_arguments(call.get("arguments", ""))
lines.append(f"Tool: {name}({json.dumps(args, ensure_ascii=False)})")
obs = (step.get("observations") or "").strip()
if obs:
lines.append(f"Observation: {obs[:2000]}")
return "\n".join(lines)
def _synthesize_forced_answer(question: str, partial_trace: str) -> str:
"""Synthesizes a best-effort final Answer from a partial reasoning trace.
Uses the cleaning-model path (DeepSeek, retried via ``_call_with_retry``)
with the forced-answer prompt. The caller bounds this call (via
``_run_with_timeout``) so the total stays under the hard cap. Raises on
model failure — the worker then records the Task as ``timeout`` (decision
B2: a failed synthesis is NOT submitted).
"""
prompt = _build_forced_answer_prompt(question, partial_trace)
text = _chat_answer(CLEANING_MODEL, prompt, description="forced answer")
if not text:
raise ValueError("Forced-answer synthesis returned an empty answer")
return _normalize_answer_for_submission(text)
def _forced_answer_budget() -> float:
"""Seconds available for forced-answer synthesis before the hard cap.
Clamped to ``FORCE_ANSWER_TIMEOUT_SEC`` and to the remaining hard-soft
budget; 0 (or negative) when the hard cap leaves no room, which means the
Task must be recorded as a plain timeout instead.
"""
remaining = TASK_HARD_TIMEOUT_SEC - TASK_SOFT_TIMEOUT_SEC
return max(0.0, min(FORCE_ANSWER_TIMEOUT_SEC, remaining))
def _snapshot_partial_worklog(agent) -> dict:
"""Best-effort snapshot of an Agent's partial ReAct trace at the soft deadline.
The wedged ``solve`` daemon helper thread may still be mutating
``agent.agent.memory.steps`` while this runs, so the read is deliberately
racy: it serializes whatever steps are visible and, if the in-flight
mutation makes serialization fail, falls back to an empty trace (the
synthesis then produces a best-effort Answer from the Question alone).
"""
try:
memory = getattr(getattr(agent, "agent", None), "memory", None)
steps = memory.steps if memory is not None else []
return _serialize_worklog(steps, status="forced")
except Exception:
worklog = _serialize_worklog([], status="forced")
worklog["error"] = "Partial Worklog unavailable at the soft deadline"
return worklog
# ─── Asynchronous Threaded Console Script Execution ──────────────────────
def _make_result(idx, task, answer, elapsed, *, worklog=None, source="live",
status="completed", timestamp=None):
"""Builds a structured result record for a Task.
``source`` is "live" (solved by the Agent this Run) or "bundle" (submitted
from the committed Answer bundle). ``status`` is "completed", "forced"
(Answer synthesized from a partial trace at the soft deadline), "error",
or "timeout". The record stays consistent with the Answer bundle entry for
the same Task (same task_id / answer / worklog / timestamp).
"""
return {
"index": idx + 1,
"task_id": task.get("task_id", "???"),
"level": task.get("Level", "?"),
"file_name": task.get("file_name", "") or "None",
"question": task.get("question", ""),
"answer": answer,
"execution_time_sec": round(elapsed, 2),
"source": source,
"status": status,
"timestamp": timestamp or _now_iso(),
"worklog": worklog,
}
def _make_bundled_result(idx, task, entry):
"""Builds a result record for a Task submitted from the Answer bundle.
The bundle is authoritative for what was actually submitted (Answer,
Worklog, timestamp, and the bundled Question/Level/file_name); the server's
Task metadata is used only as a fallback for missing fields.
"""
result = _make_result(
idx, task, entry.get("answer", ""), 0.0,
worklog=entry.get("worklog"), source="bundle", status="completed",
timestamp=entry.get("timestamp", _now_iso()),
)
result["level"] = entry.get("level", result["level"])
result["file_name"] = entry.get("file_name", result["file_name"])
result["question"] = entry.get("question", result["question"])
return result
def _timeout_result(idx, task, t0) -> dict:
"""Builds the hard-cap timeout result record (not submitted, decision B2/B3)."""
return _make_result(
idx, task,
f"Error: Task timed out after {TASK_HARD_TIMEOUT_SEC:.0f}s",
time.time() - t0, worklog=None, source="live", status="timeout",
)
def _run_task(agent, idx, task) -> tuple[dict, bool]:
"""Solves one Task under the two-stage soft/hard timeout.
Returns ``(result, replace_agent)`` where ``result`` is the structured
result record (status ``completed``, ``forced``, ``timeout``, or ``error``)
and ``replace_agent`` is True only when the Agent's ``solve`` was
interrupted at the soft deadline — its daemon helper thread may still be
wedged, so it must not be reused (the caller swaps in a fresh Agent).
The two-stage budget (decisions B1-B4):
- Within ``TASK_SOFT_TIMEOUT_SEC`` the normal result is used.
- Otherwise the partial Worklog is snapshotted (best-effort — the wedged
thread may still be mutating ``agent.memory.steps``) and a best-effort
Answer is synthesized from it, bounded by ``_forced_answer_budget()`` so
the total stays under the hard cap. Synthesis success → status ``forced``
(submitted); failure or hard cap → status ``timeout`` (not submitted).
"""
t0 = time.time()
try:
answer, worklog = _run_with_timeout(
lambda: agent.solve(
task.get("task_id", "???"),
task.get("question", ""),
task.get("file_name", ""),
),
TASK_SOFT_TIMEOUT_SEC,
)
return (
_make_result(idx, task, answer, time.time() - t0,
worklog=worklog, source="live",
status=worklog.get("status", "completed")),
False,
)
except TimeoutError:
# Soft deadline reached. The wedged Agent must always be discarded
# (replace_agent=True), whether the synthesis succeeds (forced) or not
# (timeout).
budget = _forced_answer_budget()
if budget <= 0:
# No room left before the hard cap: record a plain timeout.
return _timeout_result(idx, task, t0), True
try:
partial_worklog = _snapshot_partial_worklog(agent)
answer = _run_with_timeout(
lambda: _synthesize_forced_answer(
task.get("question", ""),
_format_partial_trace(partial_worklog),
),
budget,
)
return (
_make_result(idx, task, answer, time.time() - t0,
worklog=partial_worklog, source="live",
status="forced"),
True,
)
except Exception:
# Forced-answer synthesis failed or hit the hard cap: record a
# plain timeout (not submitted, per decision B2/B3).
return _timeout_result(idx, task, t0), True
except Exception as e:
# A throwing Task must not lose the worker: record an error Answer so
# the Run still completes for every Task. The Agent is still reusable.
return (
_make_result(idx, task, f"Error: {e}", time.time() - t0,
worklog=None, source="live", status="error"),
False,
)
def run_pipeline_and_save_csv(force: bool = False):
"""Fetches questions, solves them, and writes the Answer bundle + results CSV.
Args:
force: When True, re-solve every Task from scratch, ignoring the
committed Answer bundle. When False (default), Tasks already solved
in the bundle are submitted from it without calling the Agent;
only Tasks missing from the bundle are solved live. Every Task that
solves successfully is recorded into the Answer bundle, so the next
Run solves strictly fewer Tasks.
"""
print("Base: Connecting to the GAIA server to fetch questions...")
try:
# Send HTTP GET request to download questions dataset JSON from server,
# retrying transient network failures with backoff
def _fetch_questions():
response = requests.get(f"{DEFAULT_API_URL}/questions", timeout=15)
response.raise_for_status()
return response.json()
questions_data = _call_with_retry(_fetch_questions, description="questions fetch")
# If question list is empty, exit pipeline
if not questions_data:
print("⚠️ No questions were retrieved from the server.")
return
except Exception as e:
# Print error message and cancel run if question retrieval fails
print(f"❌ Error downloading the questions dataset: {e}")
return
total = len(questions_data)
print(f"📋 {total} questions loaded.")
# Load the committed Answer bundle (empty on the first Run) and classify
# Tasks: bundled Tasks are submitted from the bundle; the rest are solved
# live below (re-Runs only solve Tasks missing from the bundle, per the
# spec / ADR-0001).
bundle = _load_answer_bundle()
bundled_ids = set(bundle)
completed_results = []
pending = []
for i, task in enumerate(questions_data):
task_id = task.get("task_id")
if not task_id:
continue
if not force and task_id in bundled_ids:
completed_results.append(_make_bundled_result(i, task, bundle[task_id]))
else:
pending.append((i, task))
bundled_count = len(completed_results)
solved_count = len(pending)
if bundled_count:
print(f"📦 {bundled_count}/{total} Tasks already in the Answer bundle — "
f"submitting from the bundle (Agent not called).")
if solved_count:
print(f"🚀 Solving {solved_count}/{total} Tasks live (missing from the bundle) "
f"with up to {NUM_WORKERS} parallel agents (DeepSeek)...")
# ── Solve the pending Tasks with the Agent pool (lazy: no Agent is built
# ── unless there is actually something to solve, so a fully-bundled Run
# ── does not need a DeepSeek key) ─────────────────────────────────────
if pending:
try:
agents = [GAIASolverAgent() for _ in range(min(NUM_WORKERS, len(pending)))]
except Exception as e:
print(f"❌ Error creating the agent pool: {e}")
# Record an error row per pending Task so the Run still completes
for i, task in pending:
completed_results.append(
_make_result(i, task, f"Error: {e}", 0.0,
worklog=None, source="live", status="error")
)
else:
# Create thread-safe queue to manage available GAIASolverAgent
# instances across threads
agent_queue = queue.Queue()
for a in agents:
agent_queue.put(a) # Put each worker agent into queue
def _resolve(idx, task):
"""Worker thread task execution wrapper."""
# Pop an available agent instance from queue (blocking if all are busy)
my_agent = agent_queue.get()
t0 = time.time()
replaced = False
try:
# Delegate the two-stage soft/hard timeout (and forced-answer
# synthesis at the soft deadline) to _run_task
result, must_replace = _run_task(my_agent, idx, task)
replaced = must_replace
if must_replace:
# The Agent may still be wedged in a hung call on a
# daemon helper thread. Never reuse a possibly-corrupted
# Agent: drop it and put a fresh one in its place so the
# pool stays at full strength
try:
agent_queue.put(GAIASolverAgent())
except Exception as e:
print(f"⚠️ Could not replace a timed-out Agent: {e}")
return result
except Exception as e:
# A throwing Task must not lose the worker: record an error
# Answer for this Task so the Run still completes for every
# Task
return _make_result(idx, task, f"Error: {e}", time.time() - t0,
worklog=None, source="live", status="error")
finally:
# Always return the Agent to the pool (unless it was
# replaced above) so no worker is ever lost and the queue
# can never deadlock, even when a Task throws
if not replaced:
agent_queue.put(my_agent)
print(f"\n🚀 Processing {len(pending)} Tasks asynchronously in parallel...")
# Initialize ThreadPoolExecutor with maximum configured thread workers
executor = ThreadPoolExecutor(max_workers=NUM_WORKERS)
futures = {
executor.submit(_resolve, i, task): (i, task)
for i, task in pending
}
total_tasks = len(futures)
completed = 0
try:
# Consume results in submission order. Each worker enforces its
# own soft/hard timeout internally, so every future resolves in
# bounded time
for future, (i, task) in futures.items():
try:
res = future.result()
except Exception as e:
# Unexpected executor-level failure (worker already
# handles Task-level errors); still record an Answer
res = _make_result(i, task, f"Error: {e}", 0.0,
worklog=None, source="live", status="error")
completed += 1
completed_results.append(res) # Store result record
# Fold a successfully solved Task into the Answer bundle so
# the next Run submits it without re-solving. Completed and
# forced Tasks (a synthesized soft-deadline Answer that IS
# submitted — decision B3) are folded; error/timeout Tasks
# stay out and are retried next Run.
if res["status"] in ("completed", "forced"):
bundle[res["task_id"]] = _make_bundle_entry(
task_id=res["task_id"],
question=res["question"],
level=res["level"],
file_name=res["file_name"],
answer=res["answer"],
worklog=res["worklog"],
timestamp=res["timestamp"],
)
# Print live task progress log to stdout
print(f"✅ [{completed}/{total_tasks}] Task #{res['index']} "
f"({res['source']}) {res['status']} in "
f"{res['execution_time_sec']}s | Answer: {res['answer']}")
finally:
# Worker threads finish their current Task within
# TASK_SOFT_TIMEOUT_SEC + FORCE_ANSWER_TIMEOUT_SEC, so
# shutdown(wait=True) completes promptly; wedged daemon helper
# threads cannot block the process at exit
executor.shutdown(wait=True, cancel_futures=True)
# Re-sort results list by original dataset question index ordering
completed_results = sorted(completed_results, key=lambda x: x["index"])
# Keep the bundle consistent with this Run and the served Question set:
# (1) a force re-solve that now FAILS a previously-bundled Task must not
# leave a stale entry behind (the next non-forced Run would resubmit the
# old Answer while this Run's CSV shows an error) — drop it so it is
# retried. A forced (soft-deadline) result is kept, not dropped: it was
# already folded above and IS submitted and scored (decision B3), so it
# must not be treated like an error/timeout; (2) Tasks the server no
# longer serves are pruned so the bundle stays consistent with the results
# CSV and never outgrows the served set.
if force:
for res in completed_results:
if res["source"] == "live" and res["status"] not in ("completed", "forced"):
bundle.pop(res["task_id"], None)
served_ids = {t.get("task_id") for t in questions_data if t.get("task_id")}
for task_id in list(bundle):
if task_id not in served_ids:
bundle.pop(task_id, None)
# Write the committed Answer bundle (JSON keyed by task_id) so the Space
# can submit without re-solving, then the results CSV
_save_answer_bundle(bundle, ANSWER_BUNDLE_PATH)
csv_columns = [
"index", "task_id", "level", "file_name", "question", "answer",
"execution_time_sec", "source", "status", "timestamp",
]
df_output = pd.DataFrame(completed_results, columns=csv_columns)
df_output.to_csv(RESULTS_CSV_PATH, index=False, encoding="utf-8")
# Print completion summary message and target output paths
print(f"\n{'═'*50}")
print(f"📊 PROCESS COMPLETED SUCCESSFULLY")
print(f"💾 Answer bundle ({len(bundle)} entries) saved to: {os.path.abspath(ANSWER_BUNDLE_PATH)}")
print(f"💾 Results saved to: {os.path.abspath(RESULTS_CSV_PATH)}")
print(f"{'═'*50}\n")
# Python script entry point check
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(
description="Run the local GAIA solver and write the committed Answer "
"bundle + results CSV."
)
parser.add_argument(
"--force", action="store_true",
help="Force a full re-solve, ignoring the committed Answer bundle.",
)
args = parser.parse_args()
# Execute full pipeline when script is run directly from command line
run_pipeline_and_save_csv(force=args.force)