File size: 2,759 Bytes
c641d5f | 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 | """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)
|