Spaces:
Sleeping
Sleeping
| """Tools for the GAIA agent: web search, arithmetic, and image recognition.""" | |
| import ast | |
| import base64 | |
| import math | |
| import operator | |
| import re | |
| from typing import Any | |
| import requests | |
| from ddgs import DDGS | |
| from langchain_core.messages import HumanMessage | |
| from prompts import IMAGE_ANALYSIS_PROMPT | |
| MAX_IMAGES_PER_QUESTION = 3 | |
| _IMAGE_EXTENSIONS = r"(?:jpg|jpeg|png|gif|webp|bmp|svg)" | |
| _URL_TAIL = r"(?:\?[^\s\)\]\"\'<>]*)?" | |
| # Safe operators for GAIA-style numeric questions | |
| _SAFE_OPERATORS = { | |
| ast.Add: operator.add, | |
| ast.Sub: operator.sub, | |
| ast.Mult: operator.mul, | |
| ast.Div: operator.truediv, | |
| ast.FloorDiv: operator.floordiv, | |
| ast.Pow: operator.pow, | |
| ast.Mod: operator.mod, | |
| ast.USub: operator.neg, | |
| ast.UAdd: operator.pos, | |
| } | |
| def _eval_ast(node: ast.AST) -> float: | |
| if isinstance(node, ast.Constant): | |
| if isinstance(node.value, (int, float)): | |
| return float(node.value) | |
| raise ValueError(f"Unsupported constant: {node.value}") | |
| if isinstance(node, ast.Num): # pragma: no cover - py<3.8 | |
| return float(node.n) | |
| if isinstance(node, ast.BinOp): | |
| op = _SAFE_OPERATORS.get(type(node.op)) | |
| if not op: | |
| raise ValueError(f"Unsupported operator: {type(node.op).__name__}") | |
| return op(_eval_ast(node.left), _eval_ast(node.right)) | |
| if isinstance(node, ast.UnaryOp): | |
| op = _SAFE_OPERATORS.get(type(node.op)) | |
| if not op: | |
| raise ValueError(f"Unsupported unary op: {type(node.op).__name__}") | |
| return op(_eval_ast(node.operand)) | |
| if isinstance(node, ast.Call): | |
| if isinstance(node.func, ast.Name): | |
| name = node.func.id | |
| args = [_eval_ast(a) for a in node.args] | |
| if name == "sqrt" and len(args) == 1: | |
| return math.sqrt(args[0]) | |
| if name == "round" and len(args) in (1, 2): | |
| return float(round(args[0], int(args[1]) if len(args) == 2 else 0)) | |
| if name == "abs" and len(args) == 1: | |
| return abs(args[0]) | |
| raise ValueError("Unsupported function call") | |
| raise ValueError(f"Unsupported expression node: {type(node).__name__}") | |
| def _normalize_expression(expr: str) -> str: | |
| expr = expr.strip() | |
| expr = expr.replace("×", "*").replace("÷", "/").replace("^", "**") | |
| expr = re.sub(r"\s+", "", expr) | |
| return expr | |
| def calculate(expression: str) -> str: | |
| """ | |
| Evaluate a numeric expression safely (numbers, + - * / // ** %, sqrt, abs, round). | |
| Returns a string result for the agent to use as evidence. | |
| """ | |
| expr = _normalize_expression(expression) | |
| if not expr: | |
| return "Error: empty expression" | |
| try: | |
| tree = ast.parse(expr, mode="eval") | |
| result = _eval_ast(tree.body) | |
| if result == int(result) and "." not in str(expr).split("/")[-1]: | |
| return str(int(result)) | |
| return str(round(result, 10)).rstrip("0").rstrip(".") | |
| except Exception as e: | |
| return f"Error: {e}" | |
| def run_calculations(expressions: list[str]) -> list[str]: | |
| """Run multiple calculations; skip empty or duplicate expressions.""" | |
| seen: set[str] = set() | |
| outputs: list[str] = [] | |
| for raw in expressions: | |
| expr = str(raw).strip() | |
| if not expr or expr in seen: | |
| continue | |
| seen.add(expr) | |
| result = calculate(expr) | |
| outputs.append(f"{expr} = {result}") | |
| return outputs | |
| def _message_content(response: Any) -> str: | |
| content = response.content | |
| if isinstance(content, str): | |
| return content.strip() | |
| if isinstance(content, list): | |
| parts = [] | |
| for block in content: | |
| if isinstance(block, str): | |
| parts.append(block) | |
| elif isinstance(block, dict) and block.get("text"): | |
| parts.append(str(block["text"])) | |
| return "\n".join(parts).strip() | |
| return str(content).strip() | |
| def extract_image_urls(text: str) -> list[str]: | |
| """Find image URLs in question text (markdown, HTML, plain links).""" | |
| candidates: list[str] = [] | |
| patterns = [ | |
| rf"https?://[^\s\)\]\"\'<>]+\.{_IMAGE_EXTENSIONS}{_URL_TAIL}", | |
| r"!\[[^\]]*\]\((https?://[^\)]+)\)", | |
| r"https?://huggingface\.co/datasets/[^\s\)\]\"\'<>]+", | |
| r"https?://[^\s\)\]\"\'<>]*(?:/image|/images|/file|/files)[^\s\)\]\"\'<>]*", | |
| ] | |
| for pattern in patterns: | |
| for match in re.finditer(pattern, text, re.IGNORECASE): | |
| url = match.group(1) if match.lastindex else match.group(0) | |
| url = url.rstrip(".,;:") | |
| if url.startswith("http"): | |
| candidates.append(url) | |
| seen: set[str] = set() | |
| deduped: list[str] = [] | |
| for url in candidates: | |
| key = url.lower() | |
| if key not in seen: | |
| seen.add(key) | |
| deduped.append(url) | |
| if len(deduped) >= MAX_IMAGES_PER_QUESTION: | |
| break | |
| return deduped | |
| def _fetch_image_base64(url: str) -> tuple[str, str]: | |
| """Download image and return (media_type, base64_data).""" | |
| headers = {"User-Agent": "Mozilla/5.0 (compatible; GAIA-Agent/1.0)"} | |
| resp = requests.get(url, timeout=45, headers=headers) | |
| resp.raise_for_status() | |
| content_type = (resp.headers.get("content-type") or "image/jpeg").split(";")[0].strip() | |
| if not content_type.startswith("image/"): | |
| content_type = "image/jpeg" | |
| return content_type, base64.standard_b64encode(resp.content).decode("ascii") | |
| def describe_image(url: str, question: str, vision_model: Any) -> str: | |
| """ | |
| Analyze an image URL with Claude vision and return a text description. | |
| """ | |
| try: | |
| media_type, data = _fetch_image_base64(url) | |
| prompt = IMAGE_ANALYSIS_PROMPT.format(question=question, url=url) | |
| message = HumanMessage( | |
| content=[ | |
| {"type": "text", "text": prompt}, | |
| { | |
| "type": "image", | |
| "source": { | |
| "type": "base64", | |
| "media_type": media_type, | |
| "data": data, | |
| }, | |
| }, | |
| ] | |
| ) | |
| return _message_content(vision_model.invoke([message])) | |
| except Exception as e: | |
| return f"Vision error: {e}" | |
| def analyze_images(urls: list[str], question: str, vision_model: Any) -> list[str]: | |
| """Run vision on up to MAX_IMAGES_PER_QUESTION URLs.""" | |
| results: list[str] = [] | |
| for url in urls[:MAX_IMAGES_PER_QUESTION]: | |
| description = describe_image(url, question, vision_model) | |
| results.append(f"[Image: {url}]\n{description}") | |
| return results | |
| def search_web(query: str) -> str: | |
| """Search the web for information related to the query.""" | |
| results = DDGS().text(query, max_results=8) | |
| if not results: | |
| return "No results found." | |
| return "\n\n".join( | |
| f"{r.get('title', '')}: {r.get('body', r.get('href', ''))}" | |
| for r in results | |
| ) | |