feat: Python LLM/RAG query service (Milestone 3)

.github/workflows/ci.yml

@@ -86,6 +86,9 @@ jobs:
     lint-python:
         name: Lint (Python)
         runs-on: ubuntu-latest
+        strategy:
+            matrix:
+                service: [log-producer, llm-query-python]
         steps:
             - uses: actions/checkout@v4
 
@@ -98,12 +101,31 @@ jobs:
               run: pip install ruff==0.7.4
 
             - name: Run ruff check
-              working-directory: services/log-producer
+              working-directory: services/${{ matrix.service }}
               run: ruff check .
 
             - name: Run ruff format check
-              working-directory: services/log-producer
+              working-directory: services/${{ matrix.service }}
               run: ruff format --check .
+    test-python:
+        name: Test (Python - llm-query)
+        runs-on: ubuntu-latest
+        needs: lint-python
+        steps:
+            - uses: actions/checkout@v4
+
+            - name: Set up Python
+              uses: actions/setup-python@v5
+              with:
+                  python-version: "3.12"
+
+            - name: Install dependencies
+              working-directory: services/llm-query-python
+              run: pip install -r requirements-dev.txt
+
+            - name: Run tests
+              working-directory: services/llm-query-python
+              run: python -m pytest tests/ -v
     test-go-integration:
         name: Integration Test (Go + TimescaleDB)
         runs-on: ubuntu-latest
@@ -140,7 +162,7 @@ jobs:
     docker-build:
         name: Docker Build Validation
         runs-on: ubuntu-latest
-        needs: [test-go, lint-python]
+        needs: [test-go, lint-test-rust, test-python]
         steps:
             - uses: actions/checkout@v4
 
@@ -166,11 +188,17 @@ jobs:
                   context: services/anomaly-rust
                   push: false
                   tags: sentinelops/anomaly-rust:ci
+            - name: Build llm-query image
+              uses: docker/build-push-action@v6
+              with:
+                  context: services/llm-query-python
+                  push: false
+                  tags: sentinelops/llm-query-python:ci
     ci-success:
         name: CI Success
         runs-on: ubuntu-latest
         needs:
-            [lint-go, test-go, test-go-integration, lint-test-rust, lint-python, docker-build]
+            [lint-go, test-go, test-go-integration, lint-test-rust, lint-python, test-python, docker-build]
         if: always()
         steps:
             - name: Check all jobs succeeded

infra/docker-compose.yml

@@ -60,6 +60,7 @@ services:
     ports:
       - "6379:6379"
     networks: [sentinelops-net]
+
   log-producer:
     build:
       context: ../services/log-producer
@@ -75,6 +76,7 @@ services:
       ANOMALY_PROBABILITY: "0.02"
     restart: unless-stopped
     networks: [sentinelops-net]
+
   ingestion-go:
     build:
       context: ../services/ingestion-go
@@ -97,6 +99,7 @@ services:
       - "9100:9100"
     restart: unless-stopped
     networks: [sentinelops-net]
+
   anomaly-rust:
     build:
       context: ../services/anomaly-rust
@@ -119,6 +122,29 @@ services:
       - "9200:9200"
     restart: unless-stopped
     networks: [sentinelops-net]
+
+  llm-query:
+    build:
+      context: ../services/llm-query-python
+      dockerfile: Dockerfile
+    container_name: sentinelops-llm-query
+    depends_on:
+      timescaledb:
+        condition: service_healthy
+    environment:
+      DATABASE_URL: postgres://sentinelops:${TIMESCALE_PASSWORD:-devpassword}@timescaledb:5432/sentinelops
+      OLLAMA_BASE_URL: http://host.docker.internal:11434
+      OLLAMA_MODEL: llama3.2:3b
+      OLLAMA_KEEP_ALIVE: 10m
+      OLLAMA_NUM_PREDICT: "300"
+      OLLAMA_TIMEOUT_SECONDS: "180"
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    ports:
+      - "9300:9300"
+    restart: unless-stopped
+    networks: [sentinelops-net]
+
 networks:
   sentinelops-net:
     driver: bridge

services/llm-query-python/.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+.venv/
+*.pyc
+.pytest_cache/

services/llm-query-python/Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app/ ./app/
+
+ENV DATABASE_URL=postgresql://sentinelops:devpassword@timescaledb:5432/sentinelops \
+    OLLAMA_BASE_URL=http://host.docker.internal:11434 \
+    OLLAMA_MODEL=llama3.2:3b \
+    METRICS_PORT=9300
+
+EXPOSE 9300
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "9300"]

services/llm-query-python/app/config.py

@@ -0,0 +1,22 @@
+"""Configuration loaded from environment variables."""
+
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    database_url: str = (
+        "postgresql://sentinelops:devpassword@localhost:5432/sentinelops"
+    )
+    ollama_base_url: str = "http://host.docker.internal:11434"
+    ollama_model: str = "llama3.2:3b"
+    ollama_keep_alive: str = "10m"
+    ollama_num_predict: int = 300
+    ollama_timeout_seconds: float = 180.0
+    default_lookback_hours: int = 6
+    metrics_port: int = 9300
+
+    class Config:
+        env_prefix = ""
+
+
+settings = Settings()

services/llm-query-python/app/jobs.py

@@ -0,0 +1,82 @@
+"""In-memory async job queue for LLM query requests.
+
+Ollama inference on constrained hardware can take tens of seconds to
+minutes, which makes a blocking HTTP request/response the wrong shape for
+this API. Instead, POST /query enqueues a job and returns immediately;
+GET /query/{job_id} polls for status and result.
+
+This is intentionally in-memory (a dict), not Redis-backed, for now: it's
+simple, correct for a single-process deployment, and the documented
+upgrade path (Redis is already provisioned in the stack) is to swap the
+JobStore implementation without changing the API contract, once multi-
+worker or multi-process deployment is actually needed.
+"""
+
+import asyncio
+import uuid
+from datetime import datetime, timezone
+from enum import Enum
+
+from pydantic import BaseModel
+
+
+class JobStatus(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETE = "complete"
+    FAILED = "failed"
+
+
+class Job(BaseModel):
+    job_id: str
+    status: JobStatus
+    query: str
+    created_at: datetime
+    completed_at: datetime | None = None
+    result: str | None = None
+    error: str | None = None
+
+
+class JobStore:
+    def __init__(self) -> None:
+        self._jobs: dict[str, Job] = {}
+        self._lock = asyncio.Lock()
+
+    async def create(self, query: str) -> Job:
+        job = Job(
+            job_id=str(uuid.uuid4()),
+            status=JobStatus.PENDING,
+            query=query,
+            created_at=datetime.now(timezone.utc),
+        )
+        async with self._lock:
+            self._jobs[job.job_id] = job
+        return job
+
+    async def get(self, job_id: str) -> Job | None:
+        async with self._lock:
+            return self._jobs.get(job_id)
+
+    async def mark_running(self, job_id: str) -> None:
+        async with self._lock:
+            if job_id in self._jobs:
+                self._jobs[job_id].status = JobStatus.RUNNING
+
+    async def mark_complete(self, job_id: str, result: str) -> None:
+        async with self._lock:
+            if job_id in self._jobs:
+                job = self._jobs[job_id]
+                job.status = JobStatus.COMPLETE
+                job.result = result
+                job.completed_at = datetime.now(timezone.utc)
+
+    async def mark_failed(self, job_id: str, error: str) -> None:
+        async with self._lock:
+            if job_id in self._jobs:
+                job = self._jobs[job_id]
+                job.status = JobStatus.FAILED
+                job.error = error
+                job.completed_at = datetime.now(timezone.utc)
+
+
+job_store = JobStore()

services/llm-query-python/app/main.py

@@ -0,0 +1,128 @@
+"""SentinelOps LLM Query Service.
+
+Exposes a natural-language query interface over detected anomalies and
+service stats, grounded via retrieval against TimescaleDB (RAG), with
+generation via a locally-hosted Ollama model.
+"""
+
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+
+import asyncpg
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import PlainTextResponse
+from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
+from pydantic import BaseModel
+
+from app.config import settings
+from app.jobs import Job, JobStatus, job_store
+from app.ollama_client import OllamaError, generate
+from app.retrieval import format_context_for_prompt, retrieve_context
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+)
+logger = logging.getLogger("llm-query-service")
+
+QUERIES_TOTAL = Counter(
+    "sentinelops_llm_queries_total", "Total number of queries submitted."
+)
+QUERY_FAILURES_TOTAL = Counter(
+    "sentinelops_llm_query_failures_total", "Total number of failed queries."
+)
+QUERY_DURATION = Histogram(
+    "sentinelops_llm_query_duration_seconds",
+    "Time from job creation to completion, including LLM generation.",
+    buckets=[1, 5, 15, 30, 60, 120, 180, 300],
+)
+
+SYSTEM_PROMPT = (
+    "You are an SRE assistant analyzing distributed system telemetry. "
+    "Answer questions using ONLY the context data provided below. "
+    "If the context doesn't contain enough information to answer, say so "
+    "explicitly rather than guessing. Be concise and specific — cite actual "
+    "numbers from the context. Do not invent services, timestamps, or metrics "
+    "that are not present in the context."
+)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    app.state.pool = await asyncpg.create_pool(
+        settings.database_url, min_size=1, max_size=5
+    )
+    logger.info("connected to TimescaleDB")
+    yield
+    await app.state.pool.close()
+
+
+app = FastAPI(title="SentinelOps LLM Query Service", lifespan=lifespan)
+
+
+class QueryRequest(BaseModel):
+    query: str
+    service: str | None = None
+    lookback_hours: int | None = None
+
+
+class QueryAccepted(BaseModel):
+    job_id: str
+    status: JobStatus
+
+
+async def _run_query_job(
+    job_id: str, request: QueryRequest, pool: asyncpg.Pool
+) -> None:
+    await job_store.mark_running(job_id)
+    try:
+        context = await retrieve_context(
+            pool,
+            lookback_hours=request.lookback_hours or settings.default_lookback_hours,
+            service_filter=request.service,
+        )
+        context_text = format_context_for_prompt(context)
+
+        prompt = f"{context_text}\n\nQuestion: {request.query}\n\nAnswer:"
+        answer = await generate(prompt, system=SYSTEM_PROMPT)
+
+        await job_store.mark_complete(job_id, answer)
+    except OllamaError as e:
+        logger.error("ollama generation failed", exc_info=e)
+        QUERY_FAILURES_TOTAL.inc()
+        await job_store.mark_failed(job_id, str(e))
+    except Exception as e:  # noqa: BLE001 - job failures must never crash the worker
+        logger.error("unexpected error processing query job", exc_info=e)
+        QUERY_FAILURES_TOTAL.inc()
+        await job_store.mark_failed(job_id, "internal error processing query")
+
+
+@app.post("/query", response_model=QueryAccepted, status_code=202)
+async def submit_query(request: QueryRequest) -> QueryAccepted:
+    QUERIES_TOTAL.inc()
+    job = await job_store.create(request.query)
+
+    async def _tracked_run():
+        with QUERY_DURATION.time():
+            await _run_query_job(job.job_id, request, app.state.pool)
+
+    asyncio.create_task(_tracked_run())
+    return QueryAccepted(job_id=job.job_id, status=job.status)
+
+
+@app.get("/query/{job_id}", response_model=Job)
+async def get_query_result(job_id: str) -> Job:
+    job = await job_store.get(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="job not found")
+    return job
+
+
+@app.get("/healthz")
+async def healthz() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.get("/metrics")
+async def metrics() -> PlainTextResponse:
+    return PlainTextResponse(generate_latest(), media_type=CONTENT_TYPE_LATEST)