Spaces:
Sleeping
Sleeping
File size: 3,667 Bytes
ba5110e |
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 |
"""
Wolfram Alpha tool for algebraic calculations.
"""
import os
import httpx
from typing import Optional
from backend.utils.rate_limit import wolfram_limiter, query_cache
WOLFRAM_BASE_URL = "https://api.wolframalpha.com/v2/query"
async def query_wolfram_alpha(
query: str,
max_retries: int = 3
) -> tuple[bool, str]:
"""
Query Wolfram Alpha for algebraic calculations.
Includes rate limiting (2000/month) and caching.
Returns:
tuple[bool, str]: (success, result_or_error_message)
"""
# Check cache first to save API calls
cached = query_cache.get(query, context="wolfram")
if cached:
return True, f"(Cached) {cached}"
# Check monthly rate limit
can_proceed, limit_msg, remaining = wolfram_limiter.can_make_request()
if not can_proceed:
return False, limit_msg
app_id = os.getenv("WOLFRAM_ALPHA_APP_ID")
if not app_id:
return False, "Wolfram Alpha APP_ID not configured"
params = {
"appid": app_id,
"input": query,
"format": "plaintext",
"output": "json",
}
for attempt in range(max_retries):
try:
async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
response = await client.get(WOLFRAM_BASE_URL, params=params)
response.raise_for_status()
# Record usage only on successful API call
wolfram_limiter.record_usage()
data = response.json()
if data.get("queryresult", {}).get("success"):
pods = data["queryresult"].get("pods", [])
results = []
for pod in pods:
title = pod.get("title", "")
subpods = pod.get("subpods", [])
for subpod in subpods:
plaintext = subpod.get("plaintext", "")
if plaintext:
results.append(f"**{title}**: {plaintext}")
if results:
result_text = "\n\n".join(results)
# Cache successful result
query_cache.set(query, result_text, context="wolfram")
# Add warning if running low on quota
if remaining <= 100:
result_text += f"\n\n⚠️ {limit_msg}"
return True, result_text
else:
return False, "No results found from Wolfram Alpha"
else:
# Don't retry if query was understood but no answer
return False, "Wolfram Alpha could not interpret the query"
except httpx.TimeoutException:
if attempt == max_retries - 1:
return False, "Wolfram Alpha request timed out after 3 attempts"
continue
except httpx.HTTPStatusError as e:
if attempt == max_retries - 1:
return False, f"Wolfram Alpha HTTP error: {e.response.status_code}"
continue
except Exception as e:
if attempt == max_retries - 1:
return False, f"Wolfram Alpha error: {str(e)}"
continue
return False, "Wolfram Alpha failed after maximum retries"
def get_wolfram_status() -> dict:
"""Get Wolfram API usage status."""
return wolfram_limiter.get_status()
|