Spaces:
Sleeping
Sleeping
| """Bounded DeepSeek fallback for publication-grounded research answers. | |
| Called when the Llama Space is explicitly paused, after eligible Llama | |
| infrastructure failures, or while a short circuit breaker is active. | |
| Never changes retrieval or media analysis. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import json | |
| import logging | |
| import threading | |
| import time | |
| import httpx | |
| _LOCK = threading.Lock() | |
| _UNAVAILABLE_UNTIL = 0.0 | |
| def llama_space_paused() -> bool: | |
| """Check HF runtime without waking the GPU Space. | |
| A confirmed PAUSED state skips Gradio entirely. If the Hugging Face | |
| status API cannot be reached, retain the existing Llama request and | |
| error-triggered fallback behavior rather than misclassifying it. | |
| """ | |
| space_id = os.getenv("WATERSCOPE_SPACE_ID", "EnvironmentalAI/WaterScopeAI") | |
| token = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_HUB_TOKEN") | |
| headers = {"Authorization": f"Bearer {token}"} if token else {} | |
| try: | |
| response = httpx.get( | |
| f"https://huggingface.co/api/spaces/{space_id}/runtime", | |
| headers=headers, | |
| timeout=3.0, | |
| ) | |
| response.raise_for_status() | |
| payload = response.json() | |
| stage = str(payload.get("stage", "")).upper() if isinstance(payload, dict) else "" | |
| return stage == "PAUSED" | |
| except (httpx.HTTPError, ValueError, TypeError) as exc: | |
| logging.getLogger(__name__).debug( | |
| "Could not check Llama Space runtime status; proceeding with normal request: %s", | |
| type(exc).__name__, | |
| ) | |
| return False | |
| def is_infrastructure_error(exc: Exception) -> bool: | |
| text = f"{type(exc).__name__}: {exc}".lower() | |
| signals = ( | |
| "paused", "queue is full", "queueerror", "scheduling failure", | |
| "unable to schedule", "space is unavailable", "invalid state", | |
| "connection error", "connecterror", "connection refused", "timed out", | |
| "timeout", "readtimeout", "service unavailable", "502 bad gateway", | |
| "503 service", "504 gateway", "gradio", "network error", | |
| ) | |
| return any(signal in text for signal in signals) | |
| def is_gradio_transport_json_error(exc: Exception) -> bool: | |
| """Classify only malformed JSON returned while contacting the Llama Space. | |
| Never treat JSON parsing failures in publication retrieval or our own | |
| application data as an infrastructure outage. | |
| """ | |
| if not isinstance(exc, json.JSONDecodeError): | |
| return False | |
| tb = exc.__traceback__ | |
| while tb is not None: | |
| module = str(tb.tb_frame.f_globals.get("__name__", "")) | |
| if module == "gradio_client" or module.startswith("gradio_client."): | |
| return True | |
| tb = tb.tb_next | |
| return False | |
| def llama_circuit_open() -> bool: | |
| with _LOCK: | |
| return time.monotonic() < _UNAVAILABLE_UNTIL | |
| def mark_llama_unavailable() -> None: | |
| global _UNAVAILABLE_UNTIL | |
| cooldown = max(0, min(3600, int(os.getenv("WATERSCOPE_FALLBACK_COOLDOWN_SECONDS", "180")))) | |
| with _LOCK: | |
| _UNAVAILABLE_UNTIL = time.monotonic() + cooldown | |
| def clear_llama_circuit() -> None: | |
| global _UNAVAILABLE_UNTIL | |
| with _LOCK: | |
| _UNAVAILABLE_UNTIL = 0.0 | |
| def deepseek_research_answer( | |
| context: str, question: str, application: str, *, | |
| general_chat: bool = False, max_output_tokens: int | None = None, | |
| ) -> tuple[str, str]: | |
| """Answer from supplied retrieved evidence, without independently browsing papers.""" | |
| from openai import OpenAI | |
| key = os.getenv("DEEPSEEK_API_KEY", "").strip() | |
| if not key: | |
| raise RuntimeError("DeepSeek research fallback is not configured (missing DEEPSEEK_API_KEY)") | |
| model = os.getenv("WATERSCOPE_RESEARCH_FALLBACK_MODEL", "deepseek-flash").strip() | |
| if model not in {"deepseek-flash", "deepseek-v4-pro"}: | |
| raise ValueError("Unsupported research fallback model") | |
| max_context_chars = max(3000, min(150000, int(os.getenv("WATERSCOPE_FALLBACK_MAX_CONTEXT_CHARS", "60000")))) | |
| max_tokens = max(256, min(8192, int(os.getenv("WATERSCOPE_FALLBACK_MAX_OUTPUT_TOKENS", "1600")))) | |
| if max_output_tokens is not None: | |
| max_tokens = max(32, min(max_tokens, int(max_output_tokens))) | |
| client = OpenAI(base_url="https://api.deepseek.com", api_key=key, timeout=90.0, max_retries=0) | |
| instructions = ( | |
| "You are WaterScope-AI, a helpful conversational assistant focused on " | |
| "environmental science and engineering. Answer the user's message naturally " | |
| "and concisely. Do not invent citations or claim to have searched papers when " | |
| "none were provided. For evidence-based discussion, invite the user to " | |
| "select publications." | |
| if general_chat else | |
| "You are WaterScope-AI, a research assistant for environmental science and engineering. " | |
| "Answer the user using ONLY the retrieved publication evidence provided. " | |
| "Respect the task type. For summaries, synthesize supported findings; for research gaps, " | |
| "distinguish gaps explicitly reported in the sources from possible directions inferred from them. " | |
| "Do not invent observations, references, numerical values or citations. " | |
| "Clearly say when evidence is insufficient. Refer to titles/authors present in the context. " | |
| "Use concise Markdown formatting. Do not expose internal retrieval identifiers." | |
| ) | |
| response = client.responses.create( | |
| model=model, | |
| instructions=instructions, | |
| input=[{"role":"user","content":( | |
| question if general_chat else | |
| f"Research task: {application}\n\nRetrieved publication evidence:\n" | |
| f"{context[:max_context_chars]}\n\nUser question:\n{question}" | |
| )}], | |
| max_output_tokens=max_tokens, | |
| reasoning={"effort":"none"}, | |
| ) | |
| answer = (getattr(response, "output_text", None) or "").strip() | |
| if not answer: | |
| pieces = [] | |
| for item in getattr(response, "output", []) or []: | |
| for segment in getattr(item, "content", []) or []: | |
| if getattr(segment, "type", None) == "output_text": | |
| pieces.append(getattr(segment, "text", "")) | |
| answer = "\n".join(pieces).strip() | |
| if not answer: | |
| raise RuntimeError("DeepSeek research fallback returned an empty response") | |
| return answer, model | |