agentflow/engine/auto_fast.py

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