| """Deterministic task routing for GAIA Level 1.""" |
|
|
| from __future__ import annotations |
|
|
| import re |
| from dataclasses import dataclass |
| from enum import Enum |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| class Route(str, Enum): |
| WEB = "web_research" |
| YOUTUBE = "youtube_transcript" |
| AUDIO = "audio_transcription" |
| IMAGE_CHESS = "image_chess" |
| PYTHON = "python_execution" |
| SPREADSHEET = "spreadsheet" |
| MARKDOWN_TABLE = "markdown_table" |
| TEXT_TRANSFORM = "text_transformation" |
| GENERAL = "general_reasoning" |
|
|
|
|
| @dataclass(frozen=True) |
| class RouteDecision: |
| route: Route |
| reason: str |
| requires_llm: bool |
|
|
|
|
| def _looks_reversed(text: str) -> bool: |
| reversed_text = text[::-1].lower() |
| clues = ("write ", "answer", "sentence", "opposite", "reverse") |
| return sum(clue in reversed_text for clue in clues) >= 2 |
|
|
|
|
| class TaskRouter: |
| def route( |
| self, task: dict[str, Any], file_path: Path | None = None |
| ) -> RouteDecision: |
| question = str(task.get("question", "")) |
| suffix = ( |
| file_path.suffix |
| if file_path |
| else Path(str(task.get("file_name", ""))).suffix |
| ).lower() |
| if suffix in {".mp3", ".wav", ".m4a", ".ogg", ".flac"}: |
| return RouteDecision(Route.AUDIO, f"audio attachment ({suffix})", True) |
| if suffix in {".png", ".jpg", ".jpeg", ".webp"}: |
| return RouteDecision( |
| Route.IMAGE_CHESS, f"image attachment ({suffix})", True |
| ) |
| if suffix == ".py": |
| return RouteDecision(Route.PYTHON, "Python attachment", False) |
| if suffix in {".xlsx", ".xls"}: |
| return RouteDecision(Route.SPREADSHEET, "spreadsheet attachment", True) |
| if re.search(r"(?:youtube\.com/watch|youtu\.be/)", question, re.IGNORECASE): |
| return RouteDecision(Route.YOUTUBE, "YouTube URL", True) |
| if _looks_reversed(question) or re.search( |
| r"\b(reverse|alphabeti[sz]e|uppercase|lowercase|replace every|rot-?13|numeric(?:ally)? sort|calculate|compute|evaluate|extract the \d+)\b", |
| question, |
| re.IGNORECASE, |
| ): |
| return RouteDecision( |
| Route.TEXT_TRANSFORM, "deterministic text operation", False |
| ) |
| if re.search(r"^\s*\|.+\|\s*$", question, re.MULTILINE): |
| return RouteDecision(Route.MARKDOWN_TABLE, "Markdown table", True) |
| if re.search( |
| r"https?://|\b(what|who|when|where|which|how many|latest|wikipedia|article|published)\b", |
| question, |
| re.IGNORECASE, |
| ): |
| return RouteDecision(Route.WEB, "requires external facts", True) |
| return RouteDecision(Route.GENERAL, "no deterministic specialist matched", True) |
|
|
