Spaces:
Running
Running
File size: 7,876 Bytes
1f74aa2 5a0e691 336a23e 1f74aa2 5a0e691 1f74aa2 5a0e691 336a23e 5bb077e 5a0e691 1f74aa2 0e66d74 1f74aa2 0e66d74 1f74aa2 5a0e691 5bb077e 1f74aa2 336a23e 5a0e691 336a23e 1f74aa2 0e66d74 1f74aa2 5bb077e 1f74aa2 0e66d74 1f74aa2 5a0e691 1f74aa2 | 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 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 | import asyncio
import json
import logging
import re
from typing import AsyncGenerator
from app.config import settings
from app.ai.llm_client import llm_client
from app.ai.prompts import get_prompt
logger = logging.getLogger(__name__)
_litellm = None
_litellm_tried = False
def _get_acompletion():
"""Lazily import litellm so a missing/broken install never breaks import.
Returns the ``acompletion`` callable or None (callers turn that into a
friendly error instead of a stack trace)."""
global _litellm, _litellm_tried
if _litellm_tried:
return getattr(_litellm, "acompletion", None) if _litellm else None
_litellm_tried = True
try:
import litellm as _m # noqa: PLC0415 - deliberate lazy import
_litellm = _m
return _m.acompletion
except Exception as e: # pragma: no cover - env dependent
logger.warning("litellm not importable, AI interpretation disabled: %s", e)
_litellm = None
return None
def _retry_delay_seconds(error: BaseException) -> float | None:
"""Parse litellm/rate-limit errors of the form 'retry in 9.7s' or 'retry in 30 seconds'."""
text = str(error)
for pattern in (
r"retry[^\d]{0,20}(\d+(?:\.\d+)?)\s*s\b",
r"retry[^\d]{0,20}(\d+(?:\.\d+)?)\s+seconds?\b",
):
m = re.search(pattern, text, re.IGNORECASE)
if m:
try:
delay = float(m.group(1))
return min(max(delay, 1.0), 30.0)
except ValueError:
return None
return None
def _is_model_missing(error: BaseException) -> bool:
"""True when the provider reports the model doesn't exist / isn't enabled —
retrying it is pointless, so move on to the next provider immediately."""
text = str(error).lower()
return any(
needle in text
for needle in (
"not found",
"not support",
"notfounderror",
"modelnotfound",
"models/",
"is not found",
"not accessible",
"does not exist",
)
)
def _friendly_error(error: BaseException) -> str:
text = str(error)
if "organization_restricted" in text or "Organization has been restricted" in text:
return "AI interpretation is temporarily unavailable due to a provider restriction. Please try again later."
if _is_model_missing(error):
return "AI interpretation unavailable: the configured AI model is not available on its provider. Check the PRO_MODEL / API key settings."
if "QUOTA_EXCEEDED" in text or "429" in text or "rate limit" in text.lower() or "too many requests" in text.lower():
delay = _retry_delay_seconds(error)
if delay:
return f"The AI provider is rate-limited (try again in ~{delay:.0f}s). Retrying with backups…"
return "The AI provider is rate-limited. Retrying with backups…"
if "permission_denied" in text.lower() or "401" in text:
return "AI interpretation unavailable: provider authentication failed."
if "timeout" in text.lower() or "timed out" in text.lower():
return "AI interpretation timed out. Please try again."
if len(text) > 180:
return f"AI interpretation failed: {text[:180]}…"
return f"AI interpretation failed: {text}"
async def interpret_stream(pipeline_type: str, context: dict) -> AsyncGenerator[str, None]:
acompletion = _get_acompletion()
if acompletion is None:
yield _error_event("AI interpretation unavailable: LLM provider library is not installed.")
return
candidates = llm_client.get_all_candidates()
if not candidates:
yield _error_event("No LLM API keys configured. AI interpretation unavailable.")
return
prompt = llm_client.build_prompt(pipeline_type, context)
last_error = None
for candidate in candidates:
# Per-provider retries with exponential backoff, honoring any
# provider-returned retry-after delay (e.g. Gemini free-tier 429s).
for attempt in range(3):
try:
response = await acompletion(
model=candidate["model"],
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=2000,
stream=True,
timeout=25,
api_key=candidate["api_key"],
)
async for chunk in response:
if chunk.choices and chunk.choices[0].delta.content:
yield _chunk_event(chunk.choices[0].delta.content)
yield _done_event({"model": candidate["model"], "pipeline_type": pipeline_type})
return
except Exception as e:
last_error = e
logger.warning(
"LLM provider %s attempt %d failed: %s",
candidate["name"],
attempt + 1,
e,
)
if _is_model_missing(e):
break
delay = _retry_delay_seconds(e) or (2 ** attempt)
if attempt < 2:
yield _retry_event(candidate["name"], attempt + 1, delay)
await asyncio.sleep(delay)
yield _notice_event(f"Provider {candidate['name']} unavailable, trying next…")
yield _error_event(_friendly_error(last_error) if last_error else "All AI providers failed.")
async def interpret_text(pipeline_type: str, context: dict) -> dict:
"""Non-streaming interpretation for pipeline runs — same retry/fallback logic."""
acompletion = _get_acompletion()
if acompletion is None:
return {"interpretation": "AI interpretation unavailable: LLM provider library is not installed."}
candidates = llm_client.get_all_candidates()
if not candidates:
return {"interpretation": "AI interpretation unavailable: no LLM API keys configured"}
prompt = llm_client.build_prompt(pipeline_type, context)
last_error = None
for candidate in candidates:
for attempt in range(3):
try:
response = await asyncio.wait_for(
acompletion(
model=candidate["model"],
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=2000,
timeout=25,
api_key=candidate["api_key"],
),
timeout=30,
)
text = response.choices[0].message.content if response.choices else ""
if text:
return {"interpretation": text}
except Exception as e:
last_error = e
logger.warning("LLM provider %s attempt %d failed: %s", candidate["name"], attempt + 1, e)
if _is_model_missing(e):
break
delay = _retry_delay_seconds(e) or (2 ** attempt)
if attempt < 2:
await asyncio.sleep(delay)
return {"interpretation": _friendly_error(last_error) if last_error else "All AI providers failed."}
def _chunk_event(text: str) -> str:
return f"data: {json.dumps({'chunk': text})}\n\n"
def _done_event(meta: dict) -> str:
return f"data: {json.dumps({'done': True, 'meta': meta})}\n\n"
def _error_event(msg: str) -> str:
return f"data: {json.dumps({'error': msg})}\n\n"
def _retry_event(provider_name: str, attempt: int, delay: float) -> str:
return f"data: {json.dumps({'notice': f'Retrying {provider_name} (attempt {attempt + 1}) in ~{delay:.0f}s…'})}\n\n"
def _notice_event(msg: str) -> str:
return f"data: {json.dumps({'notice': msg})}\n\n"
|