import json import os import re import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from typing import Any, Optional from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse import requests # --------------------------------------------------------------------------- # Free, no-key, OpenAI-compatible providers (work from datacenter / HF Spaces). # # Primary : llm7.io -> https://api.llm7.io/v1 (large free model pool) # Fallback : pollinations.ai -> https://text.pollinations.ai/openai # Fallback : GAS Claude proxy -> Google Apps Script (claude45sonnet) # Fallback : HF Space proxy -> legacy gpt-4o-mini proxy # # llm7.io anonymous tier is rate limited (60/hour, 10/min, 1/sec). Add a FREE # token (https://token.llm7.io) as the Space secret LLM7_API_KEY to lift the # limits and make multi-step queries fast and reliable. # --------------------------------------------------------------------------- DEFAULT_LLM7_ENDPOINT = os.getenv("LLM7_ENDPOINT", "https://api.llm7.io/v1") DEFAULT_LLM7_MODELS = [ model.strip() for model in os.getenv( # Confirmed FREE on the anonymous tier (the deepseek-v3.1/kimi/minimax # models return HTTP 402 "pro — upgrade required"). mistral-small-3.2 is # the fastest reliable instruction-follower, so it leads the pool. "LLM7_MODELS", "mistral-small-3.2,codestral-latest,qwen3-235b", ).split(",") if model.strip() ] DEFAULT_POLLINATIONS_ENDPOINT = os.getenv( "POLLINATIONS_ENDPOINT", "https://text.pollinations.ai/openai" ) DEFAULT_POLLINATIONS_MODELS = [ model.strip() for model in os.getenv("POLLINATIONS_MODELS", "openai,mistral,openai-large").split(",") if model.strip() ] DEFAULT_HF_QWEN_ENDPOINT = os.getenv( "HF_QWEN_ENDPOINT", "https://cl4ude-asa.hf.space/v1/chat/completions", ) DEFAULT_GAS_CLAUDE_ENDPOINT = os.getenv( "GAS_CLAUDE_ENDPOINT", "https://script.google.com/macros/s/AKfycbx3gnzRKi7ExnNYa2yCvPxbz2JgAjB_sdWGhooyDztEeCCYxYR-GfU8g1OkafXfkg/exec", ) DEFAULT_HF_PROXY_MODELS = [ model.strip() for model in os.getenv("HF_PROXY_MODELS", "gpt-4o-mini").split(",") if model.strip() ] @dataclass class LLMEndpoint: name: str url: str models: list[str] headers: dict[str, str] = field(default_factory=dict) timeout: float = 60.0 path_mode: str = "direct" body_mode: str = "json" max_chars: int = 0 # 0 = no truncation min_interval: float = 0.0 # min seconds between requests to this endpoint enabled: bool = True @property def chat_url(self) -> str: if self.path_mode == "query": parsed = urlparse(self.url) query = dict(parse_qsl(parsed.query, keep_blank_values=True)) query["path"] = "v1/chat/completions" return urlunparse(parsed._replace(query=urlencode(query))) if "/chat/completions" in self.url: return self.url return self.url.rstrip("/") + "/chat/completions" @dataclass class Candidate: endpoint: LLMEndpoint model: str @property def key(self) -> str: return f"{self.endpoint.name}:{self.model}" class CandidateSkipped(RuntimeError): pass def _default_endpoints() -> list[dict[str, Any]]: # --- llm7.io (primary) ------------------------------------------------- llm7_headers: dict[str, str] = {} llm7_key = os.getenv("LLM7_API_KEY") or os.getenv("LLM7_TOKEN") if llm7_key: llm7_headers["Authorization"] = f"Bearer {llm7_key}" # With a token llm7 lifts the 1/sec cap; stay polite either way. llm7_interval = float(os.getenv("LLM7_MIN_INTERVAL", "0.4" if llm7_key else "1.2")) # Anonymous tier caps prompts at 8000 chars; a token lifts that. Truncate to # fit when anonymous so long planning prompts still go to the fast llm7 pool # instead of cascading through failed attempts to the slower fallback proxy. llm7_max_chars = int(os.getenv("LLM7_MAX_CHARS", "0" if llm7_key else "7600")) # --- pollinations.ai (fallback) --------------------------------------- poll_headers: dict[str, str] = {} poll_key = os.getenv("POLLINATIONS_API_KEY") or os.getenv("POLLINATIONS_TOKEN") if poll_key: poll_headers["Authorization"] = f"Bearer {poll_key}" # --- legacy HF gpt-4o-mini proxy (fallback) --------------------------- hf_headers: dict[str, str] = {} hf_key = os.getenv("HF_QWEN_PROXY_KEY") or os.getenv("PROXY_KEY") if hf_key: hf_headers["Authorization"] = f"Bearer {hf_key}" hf_headers["X-Proxy-Key"] = hf_key endpoints = [ { "name": "llm7", "url": DEFAULT_LLM7_ENDPOINT, "models": DEFAULT_LLM7_MODELS, "headers": llm7_headers, "timeout": 45, "path_mode": "direct", "body_mode": "json", "max_chars": llm7_max_chars, "min_interval": llm7_interval, }, { "name": "pollinations", "url": DEFAULT_POLLINATIONS_ENDPOINT, "models": DEFAULT_POLLINATIONS_MODELS, "headers": poll_headers, "timeout": 90, "path_mode": "direct", "body_mode": "json", "max_chars": 0, "min_interval": float(os.getenv("POLLINATIONS_MIN_INTERVAL", "1.0")), }, { "name": "gas-claude", "url": DEFAULT_GAS_CLAUDE_ENDPOINT, "models": ["claude45sonnet"], "headers": { "Content-Type": "text/plain;charset=utf-8", "Accept": "application/json", }, "timeout": 90, "path_mode": "query", "body_mode": "text", "max_chars": 0, "min_interval": 0.5, }, { "name": "hf-space-auto", "url": DEFAULT_HF_QWEN_ENDPOINT, "models": DEFAULT_HF_PROXY_MODELS, "headers": hf_headers, "timeout": 60, "path_mode": "direct", "body_mode": "json", "max_chars": int(os.getenv("HF_PROXY_MAX_CHARS", "120000" if hf_key else "7600")), "min_interval": 1.25, }, ] # Allow disabling fallbacks via env (comma-separated names). disabled = { name.strip() for name in os.getenv("LLM_DISABLED_ENDPOINTS", "").split(",") if name.strip() } return [e for e in endpoints if e["name"] not in disabled] def _load_endpoints() -> list[LLMEndpoint]: raw = os.getenv("LLM_ENDPOINTS_JSON") try: configs = json.loads(raw) if raw else _default_endpoints() except json.JSONDecodeError: configs = _default_endpoints() endpoints: list[LLMEndpoint] = [] for item in configs: if not isinstance(item, dict): continue url = str(item.get("url") or item.get("base_url") or "") models = item.get("models") or item.get("allowed_models") or [item.get("model") or "auto"] models = [str(model) for model in models if model] if not url or not models: continue timeout_ms = item.get("timeoutMs") timeout = float(item.get("timeout") or (float(timeout_ms) / 1000.0 if timeout_ms else 60.0)) endpoints.append( LLMEndpoint( name=str(item.get("name") or item.get("id") or url), url=url, models=models, headers={str(k): str(v) for k, v in dict(item.get("headers") or {}).items()}, timeout=timeout, path_mode=str(item.get("path_mode") or item.get("pathMode") or "direct"), body_mode=str(item.get("body_mode") or item.get("bodyMode") or "json"), max_chars=int(item.get("max_chars") or item.get("maxChars") or 0), min_interval=float(item.get("min_interval") or item.get("minInterval") or 0.0), enabled=bool(item.get("enabled", True)), ) ) return [endpoint for endpoint in endpoints if endpoint.enabled] class AutoFastLLM: """ OpenAI-compatible auto router across free providers. Keeps the same callable interface used by AgentFlow engines, but routes each request to the first healthy candidate (sticky on the last good one) and transparently falls back across providers/models on rate limits or errors. """ _lock = threading.Lock() _candidates: list[Candidate] = [] _latencies: dict[str, float] = {} _errors: dict[str, str] = {} _best_key: Optional[str] = None _last_benchmark_at: float = 0.0 _last_request_at: dict[str, float] = {} _cooldown_until: dict[str, float] = {} _rotate_idx: int = 0 def __init__(self, model_string: str = "auto", temperature: float = 0.0, **_: Any): self.model_string = model_string or "auto" self.temperature = temperature self.benchmark_ttl = int(os.getenv("AUTO_LLM_BENCHMARK_TTL_SECONDS", "900")) self.max_workers = int(os.getenv("AUTO_LLM_BENCHMARK_WORKERS", "4")) self.endpoints = _load_endpoints() if not AutoFastLLM._candidates: AutoFastLLM._candidates = [ Candidate(endpoint, model) for endpoint in self.endpoints for model in endpoint.models ] # -- public callable interface ----------------------------------------- def __call__(self, prompt: Any, response_format: Any = None, max_tokens: Optional[int] = None, **kwargs: Any) -> Any: messages = self._normalize_prompt(prompt) if response_format is not None: messages = self._with_json_instruction(messages, response_format) text = self.chat(messages, max_tokens=max_tokens or kwargs.get("max_completion_tokens") or 1200) if response_format is not None: return self._parse_response_format(text, response_format, prompt) return text def chat(self, messages: list[dict[str, str]], max_tokens: int = 1200) -> str: candidates = self._ordered_candidates() errors: list[str] = [] for candidate in candidates: if self._in_cooldown(candidate.key): errors.append(f"{candidate.key}: cooling down after rate limit") continue try: text, elapsed = self._call_candidate(candidate, messages, max_tokens=max_tokens) if self._looks_like_deflection(text): raise RuntimeError(f"deflection response: {text[:60]!r}") with AutoFastLLM._lock: AutoFastLLM._latencies[candidate.key] = elapsed AutoFastLLM._best_key = candidate.key AutoFastLLM._last_benchmark_at = time.time() AutoFastLLM._errors.pop(candidate.key, None) return text except CandidateSkipped as exc: errors.append(f"{candidate.key}: {exc}") except Exception as exc: errors.append(f"{candidate.key}: {exc}") with AutoFastLLM._lock: AutoFastLLM._errors[candidate.key] = str(exc) if AutoFastLLM._best_key == candidate.key: AutoFastLLM._best_key = None raise RuntimeError("All AutoFastLLM endpoints failed.\n" + "\n".join(errors)) @classmethod def status(cls) -> dict[str, Any]: return { "best": cls._best_key, "latencies": dict(sorted(cls._latencies.items(), key=lambda item: item[1])), "errors": dict(cls._errors), "last_benchmark_at": cls._last_benchmark_at, } def benchmark(self) -> dict[str, Any]: """Lightweight, rate-limit-friendly probe. Tries candidates SEQUENTIALLY (respecting per-endpoint spacing) and stops at the first one that responds, so we don't waste the free per-hour quota on a concurrent benchmark storm. """ candidates = AutoFastLLM._candidates or [ Candidate(endpoint, model) for endpoint in self.endpoints for model in endpoint.models ] test_messages = [{"role": "user", "content": "Reply with OK only."}] latencies: dict[str, float] = {} errors: dict[str, str] = {} best_key: Optional[str] = None for candidate in candidates: try: _text, elapsed = self._call_candidate(candidate, test_messages, 16) latencies[candidate.key] = elapsed best_key = candidate.key break except Exception as exc: errors[candidate.key] = str(exc) with AutoFastLLM._lock: AutoFastLLM._latencies.update(latencies) AutoFastLLM._errors.update(errors) if best_key: AutoFastLLM._best_key = best_key AutoFastLLM._last_benchmark_at = time.time() return self.status() # -- candidate ordering ------------------------------------------------- def _in_cooldown(self, key: str) -> bool: with AutoFastLLM._lock: return time.time() < AutoFastLLM._cooldown_until.get(key, 0.0) def _ordered_candidates(self) -> list[Candidate]: candidates = AutoFastLLM._candidates by_key = {candidate.key: candidate for candidate in candidates} ordered_keys: list[str] = [] # 1) Stick to the last known-good candidate. if AutoFastLLM._best_key and AutoFastLLM._best_key in by_key: ordered_keys.append(AutoFastLLM._best_key) # 2) Rotate the primary provider's models so we don't hammer a single # model (and to spread load across the pool). primary_name = candidates[0].endpoint.name if candidates else None primary = [c for c in candidates if c.endpoint.name == primary_name] if primary: with AutoFastLLM._lock: start = AutoFastLLM._rotate_idx % len(primary) AutoFastLLM._rotate_idx = (AutoFastLLM._rotate_idx + 1) % max(1, len(primary)) rotated = primary[start:] + primary[:start] ordered_keys.extend(c.key for c in rotated) # 3) Everything else (fallback providers), then any remaining. ordered_keys.extend(c.key for c in candidates if c.endpoint.name != primary_name) ordered_keys.extend(c.key for c in candidates) seen: set[str] = set() ordered: list[Candidate] = [] for key in ordered_keys: if key in seen or key not in by_key: continue seen.add(key) ordered.append(by_key[key]) return ordered # -- transport ---------------------------------------------------------- def _respect_interval(self, endpoint: LLMEndpoint) -> None: if endpoint.min_interval <= 0: return with AutoFastLLM._lock: last = AutoFastLLM._last_request_at.get(endpoint.name, 0.0) wait = endpoint.min_interval - (time.time() - last) if wait > 0: time.sleep(min(wait, endpoint.min_interval)) AutoFastLLM._last_request_at[endpoint.name] = time.time() def _call_candidate(self, candidate: Candidate, messages: list[dict[str, str]], max_tokens: int) -> tuple[str, float]: endpoint = candidate.endpoint message_chars = sum(len(str(message.get("content", ""))) for message in messages) if endpoint.max_chars and message_chars > endpoint.max_chars: messages = self._fit_messages_to_limit(messages, endpoint.max_chars) payload = { "model": candidate.model, "messages": messages, "temperature": self.temperature, "max_tokens": max_tokens, "stream": False, } headers = { "Accept": "application/json", "Content-Type": "application/json", **endpoint.headers, } if endpoint.body_mode == "text": headers["Content-Type"] = endpoint.headers.get("Content-Type", "text/plain;charset=utf-8") self._respect_interval(endpoint) started = time.perf_counter() body = json.dumps(payload, ensure_ascii=False) retry_count = int(os.getenv("AUTO_LLM_RETRIES", "2")) retry_sleep = float(os.getenv("AUTO_LLM_RETRY_SLEEP_SECONDS", "2.0")) max_retry_sleep = float(os.getenv("AUTO_LLM_MAX_RETRY_SLEEP_SECONDS", "12.0")) response = None for attempt in range(retry_count + 1): response = requests.post( endpoint.chat_url, headers=headers, data=body.encode("utf-8"), timeout=endpoint.timeout, ) if response.status_code not in {429, 500, 502, 503, 504} or attempt >= retry_count: break header_delay = response.headers.get("Retry-After") try: delay = float(header_delay) if header_delay else retry_sleep * (2 ** attempt) except ValueError: delay = retry_sleep * (2 ** attempt) if response.status_code == 429: delay = max(delay, retry_sleep * (2 ** attempt)) with AutoFastLLM._lock: AutoFastLLM._last_request_at[endpoint.name] = time.time() + delay time.sleep(max(0.25, min(delay, max_retry_sleep))) elapsed = time.perf_counter() - started if response.status_code == 429: # Put this candidate on a short cooldown so chat() rotates away. with AutoFastLLM._lock: AutoFastLLM._cooldown_until[candidate.key] = time.time() + float( os.getenv("AUTO_LLM_COOLDOWN_SECONDS", "30") ) raise RuntimeError(f"HTTP 429 (rate limited): {response.text[:200]}") if response.status_code >= 400: raise RuntimeError(f"HTTP {response.status_code}: {response.text[:300]}") try: data = response.json() text = self._extract_text(data) except Exception: text = response.text if not text or not text.strip(): raise RuntimeError("empty response") return text.strip(), elapsed # -- response parsing (unchanged behaviour) ----------------------------- @staticmethod def _extract_text(data: Any) -> str: if isinstance(data, str): return data if not isinstance(data, dict): return str(data) choices = data.get("choices") if isinstance(choices, list) and choices: first = choices[0] or {} message = first.get("message") or {} if isinstance(message, dict) and message.get("content") is not None: return str(message["content"]) if first.get("text") is not None: return str(first["text"]) for key in ("content", "text", "answer", "response"): if data.get(key) is not None: return str(data[key]) return json.dumps(data, ensure_ascii=False) @staticmethod def _looks_like_deflection(text: str) -> bool: snippet = str(text or "").strip().lower() if not snippet: return True deflections = ( "daftar dan ulangi permintaan anda", "sebutkan dan ulangi permintaan", "ulangi permintaan anda", "please repeat your request", "list and repeat your request", ) return any(snippet.startswith(d) or snippet == d for d in deflections) @staticmethod def _fit_messages_to_limit(messages: list[dict[str, str]], limit: int) -> list[dict[str, str]]: budget = max(2000, int(limit * 0.96)) fitted = [dict(m) for m in messages] def total() -> int: return sum(len(str(m.get("content", ""))) for m in fitted) if total() <= budget: return fitted marker = "\n\n...[context truncated to fit model context limit]...\n\n" guard = 0 while total() > budget and guard < 200: guard += 1 idx = max(range(len(fitted)), key=lambda i: len(str(fitted[i].get("content", "")))) content = str(fitted[idx].get("content", "")) if len(content) <= 800: break overflow = total() - budget cut = min(len(content) - 800, overflow + len(marker)) if cut <= len(marker): break keep = len(content) - cut head_len = max(300, int(keep * 0.55)) tail_len = max(0, keep - head_len) new_content = content[:head_len] + marker + (content[len(content) - tail_len:] if tail_len else "") fitted[idx]["content"] = new_content return fitted @staticmethod def _normalize_prompt(prompt: Any) -> list[dict[str, str]]: if isinstance(prompt, list): parts: list[str] = [] for item in prompt: if isinstance(item, bytes): parts.append("[Binary input omitted in CPU text mode]") else: parts.append(str(item)) content = "\n\n".join(parts) else: content = str(prompt) return [{"role": "user", "content": content}] @staticmethod def _fields_for_model(response_format: Any) -> list[str]: fields = getattr(response_format, "model_fields", None) if isinstance(fields, dict): return list(fields.keys()) fields = getattr(response_format, "__fields__", None) if isinstance(fields, dict): return list(fields.keys()) return [] def _with_json_instruction(self, messages: list[dict[str, str]], response_format: Any) -> list[dict[str, str]]: fields = self._fields_for_model(response_format) schema_hint = ", ".join(fields) if fields else "the requested fields" instruction = ( "Return only a valid JSON object. " f"The object must contain these fields: {schema_hint}. " "Do not wrap the JSON in markdown." ) next_messages = list(messages) next_messages[-1] = { **next_messages[-1], "content": next_messages[-1]["content"] + "\n\n" + instruction, } return next_messages def _parse_response_format(self, text: str, response_format: Any, prompt: Any = None) -> Any: fields = self._fields_for_model(response_format) try: parsed = self._extract_json_object(text) except Exception: parsed = self._fallback_structured(text, fields, prompt) if "tool_name" in fields and parsed.get("tool_name") is not None: parsed["tool_name"] = self._normalize_tool_alias(str(parsed["tool_name"])) if "command" in fields: parsed["command"] = self._normalize_tool_command(parsed.get("command"), prompt) if "stop_signal" in fields and isinstance(parsed.get("stop_signal"), str): parsed["stop_signal"] = parsed["stop_signal"].lower() in {"true", "stop", "yes", "1"} try: return response_format(**parsed) except Exception: return response_format(**self._fallback_structured(text, fields, prompt)) @staticmethod def _extract_json_object(text: str) -> dict[str, Any]: cleaned = str(text or "").strip() if cleaned.startswith("```"): cleaned = cleaned.strip("`") cleaned = cleaned.replace("json\n", "", 1).replace("JSON\n", "", 1) try: parsed = json.loads(cleaned) if isinstance(parsed, dict): return parsed except Exception: pass start = cleaned.find("{") end = cleaned.rfind("}") if start >= 0 and end > start: parsed = json.loads(cleaned[start : end + 1]) if isinstance(parsed, dict): return parsed raise ValueError("No JSON object found") def _fallback_structured(self, text: str, fields: list[str], prompt: Any = None) -> dict[str, Any]: text = str(text or "") lower = text.lower() data: dict[str, Any] = {} for field in fields: if field == "stop_signal": data[field] = "stop" in lower and "continue" not in lower elif field == "command": data[field] = self._command_from_prompt(prompt) elif field == "tool_name": data[field] = self._extract_loose_field(text, "tool_name") or "SearXNG_Search_Tool" elif field in {"context", "sub_goal", "justification", "analysis", "explanation"}: data[field] = self._extract_loose_field(text, field) or text else: data[field] = text return data def _normalize_tool_command(self, command: Any, prompt: Any = None) -> str: command_text = str(command or "").strip() if not command_text: return self._command_from_prompt(prompt) if command_text.startswith("```"): command_text = re.sub(r"^```(?:python)?\s*", "", command_text, flags=re.I) command_text = re.sub(r"\s*```$", "", command_text) single_line_match = re.search(r"execution\s*=\s*tool\.execute\([^\n]*\)", command_text) if single_line_match: candidate = single_line_match.group(0).strip() if self._is_valid_tool_execute_command(candidate): return candidate code_block_match = re.search(r"```python\s*(.*?)```", command_text, flags=re.I | re.S) if code_block_match: block = code_block_match.group(1).strip() single_line_match = re.search(r"execution\s*=\s*tool\.execute\([^\n]*\)", block) if single_line_match: candidate = single_line_match.group(0).strip() if self._is_valid_tool_execute_command(candidate): return candidate return self._command_from_prompt(prompt) @staticmethod def _is_valid_tool_execute_command(command: str) -> bool: if "tool.execute(" not in command: return False if "query=" not in command: return False if "queries=" in command: return False if re.search(r"=\s*[a-zA-Z_]\w*\s*(?:,|\))", command) and not re.search(r"=\s*(True|False|None)\s*(?:,|\))", command): return False return True def _command_from_prompt(self, prompt: Any) -> str: prompt_text = self._prompt_to_text(prompt) query = self._extract_prompt_field(prompt_text, "Sub-Goal") if query: query = self._extract_loose_field(query, "sub_goal") or self._extract_loose_field(query, "query") or query if not query: query = self._extract_prompt_field(prompt_text, "Query") if not query: query = "Search the web for information relevant to the user query." query = " ".join(query.split()) return f"execution = tool.execute(query={json.dumps(query, ensure_ascii=False)})" @staticmethod def _extract_loose_field(text: str, field_name: str) -> str: text = str(text or "") snake = re.escape(field_name) spaced = re.escape(field_name.replace("_", " ")) patterns = [ rf'"{snake}"\s*:\s*"([^"]+)"', rf"'{snake}'\s*:\s*'([^']+)'", rf"{snake}\s*:\s*(.*?)(?:\n\s*[A-Za-z_ ]+\s*:|\n\s*[,}}]|\Z)", rf"{spaced}\s*:\s*(.*?)(?:\n\s*[A-Za-z_ ]+\s*:|\n\s*[,}}]|\Z)", ] for pattern in patterns: match = re.search(pattern, text, flags=re.I | re.S) if match: return match.group(1).strip().strip(",").strip() return "" @staticmethod def _prompt_to_text(prompt: Any) -> str: if isinstance(prompt, list): return "\n\n".join(str(item) for item in prompt if not isinstance(item, bytes)) return str(prompt or "") @staticmethod def _extract_prompt_field(prompt_text: str, field_name: str) -> str: patterns = [ rf"-\s*\*\*{re.escape(field_name)}:\*\*\s*(.*?)(?:\n-\s*\*\*|\n\s*Instructions:|\n\s*Output Format:|\Z)", rf"{re.escape(field_name)}:\s*(.*?)(?:\n[A-Z][A-Za-z -]+:|\n\s*Instructions:|\n\s*Output Format:|\Z)", ] for pattern in patterns: match = re.search(pattern, prompt_text, flags=re.S) if match: return match.group(1).strip() return "" @staticmethod def _normalize_tool_alias(tool_name: str) -> str: normalized = " ".join(tool_name.replace("_", " ").lower().split()) search_aliases = { "search", "web search", "web search tool", "searxng", "searxng search", "searxng search tool", "google search", "google search tool", "ground google search tool", } if normalized in search_aliases: return "SearXNG_Search_Tool" generator_aliases = { "base generator", "base generator tool", "generalist", "generalist tool", "generalist solution generator", "generalist solution generator l", "generalist solution generator tool", } if normalized in generator_aliases or normalized.startswith("generalist solution generator"): return "Generalist_Solution_Generator_Tool" python_aliases = { "python", "python coder", "python coder tool", "python code generator", "python code generator tool", } if normalized in python_aliases or normalized.startswith("python"): return "Python_Code_Generator_Tool" return tool_name