Browser edge compute optimization

backend/app/config.py

@@ -15,10 +15,20 @@ class Settings(BaseSettings):
     PYTHON_ENV: str = "development"
     LOG_LEVEL: str = "INFO"
 
-    # Redis & Celery
-    REDIS_URL: str = "redis://localhost:6379/0"
-    CELERY_BROKER_URL: str = "redis://localhost:6379/0"
-    CELERY_RESULT_BACKEND: str = "redis://localhost:6379/0"
+    # Temporal (optional durable execution alongside Celery).
+    # When enabled, long-running workflows run on Temporal instead of Celery
+    # so they can pause, resume, and track deep historical states.
+    TEMPORAL_ENABLED: bool = False
+    TEMPORAL_HOST: str = "localhost:7233"
+    TEMPORAL_NAMESPACE: str = "gridify"
+    TEMPORAL_TASK_QUEUE: str = "gridify-tasks"
+
+    # Valkey / Redis (Valkey is a Redis-compatible drop-in; redis-py client
+    # works against Valkey because Valkey speaks the RESP protocol).
+    VALKEY_URL: str = "valkey://localhost:6379/0"
+    REDIS_URL: str = "valkey://localhost:6379/0"
+    CELERY_BROKER_URL: str = "valkey://localhost:6379/0"
+    CELERY_RESULT_BACKEND: str = "valkey://localhost:6379/0"
 
     # Database
     DATABASE_URL: str = "postgresql://gridify:gridify_password@localhost:5432/gridify"
@@ -31,6 +41,18 @@ class Settings(BaseSettings):
     CHROMA_PORT: int = 8000
     QDRANT_URL: Optional[str] = None
 
+    # LLM Gateway Proxies (Portkey / Langfuse).
+    # When configured, all LiteLLM calls are routed through the gateway so
+    # tracking, retries, budget limits, and fallbacks live outside the core
+    # FastAPI code.
+    LLM_GATEWAY_PROVIDER: str = "litellm"
+    LLM_GATEWAY_API_KEY: Optional[str] = None
+    LLM_GATEWAY_BASE_URL: Optional[str] = None
+    LLM_GATEWAY_PORTKEY_API_KEY: Optional[str] = None
+    LLM_GATEWAY_LANGFUSE_PUBLIC_KEY: Optional[str] = None
+    LLM_GATEWAY_LANGFUSE_SECRET_KEY: Optional[str] = None
+    LLM_GATEWAY_LANGFUSE_HOST: str = "https://cloud.langfuse.com"
+
     # LLM Configuration
     LITELLM_API_KEY: Optional[str] = None
     LLM_PROVIDER: str = "gemini"

backend/app/services/agent_service.py

@@ -25,6 +25,7 @@
     GENAI_AVAILABLE = False
 
 from app.config import settings
+from app.services.semantic_layer import get_semantic_layer, SemanticLayer, SemanticQueryError
 from app.utils.logger import setup_logger
 from app.services.duckdb_service import get_duckdb_service
 
@@ -81,6 +82,30 @@ def run_sql(sql: str) -> str:
             f"Executed SQL against DuckDB: {len(rows)} rows. "
             f"Sample: {preview}"
         )
+
+    @staticmethod
+    def semantic_query(query: Dict[str, Any]) -> str:
+        """Execute a structured semantic query instead of raw SQL.
+
+        The LLM should emit JSON with ``measures``, ``dimensions``, and
+        optional ``filters`` rather than free-form SQL. The semantic layer
+        validates the query against the catalog and compiles safe SQL.
+        """
+        try:
+            layer = get_semantic_layer()
+            validated = layer.validate(query)
+            sql = layer.to_sql(validated)
+            db = get_duckdb_service()
+            table = db.english_to_sql(sql)
+            rows = table.to_pylist()
+            return (
+                f"Semantic query compiled to: {sql}\n"
+                f"Returned {len(rows)} rows. Sample: {rows[:10]}"
+            )
+        except SemanticQueryError as exc:
+            return f"Semantic query rejected: {exc}"
+        except Exception as exc:
+            return f"Semantic query execution failed: {exc}"
 
     @staticmethod
     def list_devices() -> str:
@@ -109,7 +134,9 @@ class AIAgentService:
         "You are an expert IoT telemetry analyst. You help users understand "
         "their sensor data, identify patterns, and generate actionable "
         "insights. Use the available tools to query data and create "
-        "visualizations."
+        "visualizations. Prefer the 'semantic_query' tool over 'run_sql' "
+        "whenever possible: emit structured JSON with 'measures', 'dimensions', "
+        "and optional 'filters' instead of raw SQL to avoid injection risks."
     )
 
     def __init__(self, model: Optional[str] = None):
@@ -129,6 +156,7 @@ def _setup_tools(self) -> Dict[str, Callable[..., str]]:
         return {
             "query_telemetry": tools.query_telemetry,
             "run_sql": tools.run_sql,
+            "semantic_query": tools.semantic_query,
             "list_devices": tools.list_devices,
             "generate_summary": tools.generate_summary,
             "create_visualization": tools.create_visualization,

backend/app/services/arrow_service.py

@@ -129,6 +129,14 @@ class GridifyFlightServer:
     Each dataset is published under a Flight ``descriptor`` whose command is the
     dataset name (e.g. ``b"devices"``). Consumers call ``get_flight_info`` then
     ``do_get`` to stream the columnar data with zero JSON serialization.
+
+    Flight SQL support
+    ------------------
+    When a client sends a SQL statement as the ticket (or as a Flight
+    descriptor command starting with ``b"SQL "``), the server executes the
+    query against DuckDB and streams the resulting record batch back as an
+    Arrow stream. This lets any Arrow Flight SQL client query the telemetry
+    catalog directly without hitting the HTTP REST layer.
     """
 
     def __init__(
@@ -156,6 +164,22 @@ def list_flights(self, context, criteria):  # noqa: D401, ANN001
                     )
 
             def get_flight_info(self, context, descriptor):  # noqa: ANN001
+                command = descriptor.command
+                if command and command.startswith(b"SQL "):
+                    sql = command[4:].decode("utf-8", errors="replace").strip()
+                    if not sql:
+                        raise ValueError("Empty SQL command")
+                    from app.services.duckdb_service import get_duckdb_service
+                    db = get_duckdb_service()
+                    table = db.query_to_arrow(sql)
+                    endpoint = _flight.FlightEndpoint(
+                        command,
+                        [_flight.Location.for_grpc_tcp("0.0.0.0", 8815)],
+                    )
+                    return _flight.FlightInfo(
+                        table.schema, descriptor, [endpoint], table.num_rows, -1
+                    )
+
                 ds = descriptor.command.decode() if descriptor.command else "devices"
                 table = telemetry_to_table(provider(), ds)
                 endpoint = _flight.FlightEndpoint(
@@ -167,12 +191,26 @@ def get_flight_info(self, context, descriptor):  # noqa: ANN001
                 )
 
             def do_get(self, context, ticket):  # noqa: ANN001
-                ds = ticket.ticket.decode()
+                ticket_str = ticket.ticket.decode("utf-8", errors="replace")
+
+                if ticket_str.startswith("SQL "):
+                    sql = ticket_str[4:].strip()
+                    if not sql:
+                        raise ValueError("Empty SQL ticket")
+                    from app.services.duckdb_service import get_duckdb_service
+                    db = get_duckdb_service()
+                    table = db.query_to_arrow(sql)
+                    return _flight.RecordBatchStream(table)
+
+                ds = ticket_str
+                if ds not in DATASETS:
+                    raise ValueError(
+                        f"Unknown dataset {ds!r}; expected one of {DATASETS!r}"
+                    )
                 table = telemetry_to_table(provider(), ds)
                 return _flight.RecordBatchStream(table)
 
             def do_put(self, context, descriptor, reader):  # noqa: ANN001
-                # Read-and-discard: this server is a read-only publisher.
                 for _ in reader:
                     pass
 

backend/app/services/llm_gateway.py

@@ -0,0 +1,104 @@
+"""LLM gateway abstraction for Portkey / Langfuse.
+
+Instead of managing fallback logic directly inside the core application code,
+all LiteLLM calls can be routed through a standalone gateway. This cleanly
+offloads tracking, automatic retries, budget tracking, and fallbacks away from
+the central FastAPI code.
+
+Supported providers:
+- ``litellm`` (default): direct LiteLLM calls, no gateway overhead.
+- ``portkey``: routes through Portkey's virtual keys and gateway.
+- ``langfuse``: routes through Langfuse's proxy for observability.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Dict, List, Optional
+
+from app.config import settings
+from app.utils.logger import setup_logger
+
+logger = setup_logger(__name__)
+
+try:  # LiteLLM is optional at import time so tests run without it.
+    import litellm  # type: ignore
+
+    LITELLM_AVAILABLE = True
+except Exception:  # pragma: no cover
+    LITELLM_AVAILABLE = False
+
+
+class LLMGateway:
+    """Routes LLM completions through an optional gateway provider.
+
+    When no gateway is configured, calls fall back to plain LiteLLM so the
+    rest of the code path is unchanged.
+    """
+
+    def __init__(self) -> None:
+        self.provider = settings.LLM_GATEWAY_PROVIDER.lower()
+        self._completion: Callable[..., Any] = self._build_completion_fn()
+
+    def _build_completion_fn(self) -> Callable[..., Any]:
+        if not LITELLM_AVAILABLE:
+            logger.warning("LiteLLM not installed; gateway unavailable")
+            return lambda **kwargs: (_ for _ in ()).throw(
+                RuntimeError("LiteLLM is not installed")
+            )
+
+        if self.provider == "portkey" and settings.LLM_GATEWAY_PORTKEY_API_KEY:
+            logger.info("Routing LLM calls through Portkey gateway")
+            try:
+                import portkey_ai  # type: ignore
+
+                portkey_ai.api_key = settings.LLM_GATEWAY_PORTKEY_API_KEY
+                if settings.LLM_GATEWAY_BASE_URL:
+                    portkey_ai.api_base = settings.LLM_GATEWAY_BASE_URL
+                return portkey_ai.Completions.acompletions
+            except Exception as exc:  # pragma: no cover - optional dep
+                logger.warning("Portkey gateway unavailable (%s); using LiteLLM", exc)
+
+        if self.provider == "langfuse" and (
+            settings.LLM_GATEWAY_LANGFUSE_PUBLIC_KEY
+            and settings.LLM_GATEWAY_LANGFUSE_SECRET_KEY
+        ):
+            logger.info("Routing LLM calls through Langfuse gateway")
+            try:
+                from langfuse.openai import OpenAI  # type: ignore
+
+                client = OpenAI(
+                    base_url=settings.LLM_GATEWAY_BASE_URL or settings.LLM_GATEWAY_LANGFUSE_HOST,
+                    api_key=settings.LLM_GATEWAY_LANGFUSE_SECRET_KEY,
+                )
+                return lambda **kwargs: client.chat.completions.create(**kwargs).model_dump()
+            except Exception as exc:  # pragma: no cover - optional dep
+                logger.warning("Langfuse gateway unavailable (%s); using LiteLLM", exc)
+
+        logger.debug("No LLM gateway configured; using direct LiteLLM")
+        return litellm.completion
+
+    def complete(self, kwargs: Dict[str, Any]) -> Dict[str, Any]:
+        """Run a completion through the configured gateway.
+
+        Args:
+            kwargs: Keyword arguments forwarded to the underlying client
+                (model, messages, temperature, etc.).
+
+        Returns:
+            OpenAI-style response dict.
+        """
+        return self._completion(**kwargs)
+
+    def is_available(self) -> bool:
+        return LITELLM_AVAILABLE
+
+
+_gateway: Optional[LLMGateway] = None
+
+
+def get_llm_gateway() -> LLMGateway:
+    """Return the shared :class:`LLMGateway` singleton."""
+    global _gateway
+    if _gateway is None:
+        _gateway = LLMGateway()
+    return _gateway

backend/app/services/llm_service.py

@@ -19,6 +19,7 @@
 from typing import Any, Callable, Dict, List, Optional
 
 from app.config import settings
+from app.services.llm_gateway import get_llm_gateway
 from app.utils.logger import setup_logger
 
 logger = setup_logger(__name__)
@@ -120,9 +121,10 @@ def complete(
 
             try:
                 attempts.append(label)
-                response = self._completion(**kwargs)
+                gateway = get_llm_gateway()
+                response = gateway.complete(kwargs)
                 content = _extract_content(response)
-                logger.info("LLM served by %s", label)
+                logger.info("LLM served by %s via %s", label, settings.LLM_GATEWAY_PROVIDER)
                 return LLMResult(content=content, model=label, attempts=attempts)
             except Exception as exc:  # noqa: BLE001 - we intentionally try next provider
                 last_error = exc