feat: adaptive embedding rate limiting

README.md

@@ -357,7 +357,7 @@ Bootstrap settings use the `BIGRAG_` prefix as environment variables, or configu
 | Variable | Description | Default |
 |----------|-------------|---------|
 | `BIGRAG_CHAT_PROVIDER` | Chat provider | `openai` |
-| `BIGRAG_CHAT_MODEL` | Default chat model | `gpt-4o-mini` |
+| `BIGRAG_CHAT_MODEL` | Default chat model | `gpt-4.1` |
 | `BIGRAG_CHAT_BASE_URL` | Base URL for OpenAI-compatible chat endpoints | — |
 | `BIGRAG_CHAT_TEMPERATURE` | Default chat temperature | `0.2` |
 | `BIGRAG_CHAT_MAX_CONTEXT_CHARS` | Max retrieved-context characters per chat call | `120000` |

api/bigrag/app_factory/routers.py

@@ -27,7 +27,7 @@ def include_all_routers(app: FastAPI) -> None:
     from bigrag.routers.mcp_servers import router as mcp_servers_router
     from bigrag.routers.preferences import router as preferences_router
     from bigrag.routers.query import router as query_router
-    from bigrag.routers.realtime import router as realtime_router
+    from bigrag.routers.status import router as status_router
     from bigrag.routers.upload_sessions import router as upload_sessions_router
     from bigrag.routers.usage import router as usage_router
     from bigrag.routers.vectors import router as vectors_router
@@ -43,7 +43,7 @@ def include_all_routers(app: FastAPI) -> None:
     app.include_router(admin_vector_storage_router)
     app.include_router(mcp_servers_router)
     app.include_router(admin_audit_router)
-    app.include_router(realtime_router)
+    app.include_router(status_router)
     app.include_router(embedding_presets_router)
     app.include_router(collections_router)
     app.include_router(connectors_router)

api/bigrag/config.py

@@ -68,7 +68,7 @@ class Settings(BaseSettings):
     allow_local_webhooks: bool = False
 
     chat_provider: str = "openai"
-    chat_model: str = "gpt-4o-mini"
+    chat_model: str = "gpt-4.1"
     chat_base_url: str | None = None
     chat_temperature: float = 0.2
     chat_max_context_chars: int = 120_000

api/bigrag/models/status.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class OverviewStatusResponse(BaseModel):
+    platform: dict[str, object]
+    readiness: dict[str, object]
+
+
+class CollectionsStatusResponse(BaseModel):
+    collections_total: int
+    documents_total: int
+    documents_ready: int
+    documents_pending: int
+    documents_processing: int
+    documents_failed: int
+    total_chunks: int
+    total_tokens: int
+    total_size_bytes: int

api/bigrag/routers/status.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, Query, Request
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from bigrag.db.session import get_session
+from bigrag.middleware.auth import get_current_user, require_admin_session
+from bigrag.models.access import AccessLogOverviewResponse
+from bigrag.models.status import CollectionsStatusResponse, OverviewStatusResponse
+from bigrag.services.access_log.queries import access_overview_payload
+from bigrag.services.health import readiness_payload
+from bigrag.services.platform_stats import platform_stats_payload
+from bigrag.services.status import collections_status_payload
+from bigrag.services.usage import UsageResponse, usage_payload
+
+router = APIRouter(tags=["status"])
+
+
+@router.get("/v1/status/overview", response_model=OverviewStatusResponse)
+async def overview_status(
+    request: Request,
+    _: dict = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> OverviewStatusResponse:
+    return OverviewStatusResponse(
+        platform=await platform_stats_payload(request.app.state.queue, session),
+        readiness=await readiness_payload(request.app.state.vector_store, request.app.state.queue),
+    )
+
+
+@router.get("/v1/status/collections", response_model=CollectionsStatusResponse)
+async def collections_status(
+    _: dict = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> CollectionsStatusResponse:
+    return await collections_status_payload(session)
+
+
+@router.get("/v1/status/usage", response_model=UsageResponse)
+async def usage_status(
+    window_days: int = Query(default=30, ge=1, le=365),
+    _: dict = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> UsageResponse:
+    return await usage_payload(session, window_days=window_days)
+
+
+@router.get("/v1/admin/status/access", response_model=AccessLogOverviewResponse)
+async def access_status(
+    window_days: int = Query(default=7, ge=1, le=90),
+    _: dict = Depends(require_admin_session),
+    session: AsyncSession = Depends(get_session),
+) -> AccessLogOverviewResponse:
+    return await access_overview_payload(session, window_days=window_days)

api/bigrag/services/access_log/queries.py

@@ -5,7 +5,6 @@
 
 import sqlalchemy as sa
 from fastapi import HTTPException
-from fastapi.encoders import jsonable_encoder
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from bigrag.db.models import AccessLog
@@ -16,12 +15,10 @@
     AccessLogOverviewResponse,
     AccessLogTimelinePoint,
 )
-from bigrag.services import redis_cache
 from bigrag.services.access_log.middleware import RAG_ACCESS_ACTIONS
 from bigrag.services.pagination import paginate
 
 _RAG_ACTION_FILTER = AccessLog.action.in_(tuple(sorted(RAG_ACCESS_ACTIONS)))
-_ACCESS_OVERVIEW_TTL = 15
 
 
 def access_log_entry(row: AccessLog) -> AccessLogEntry:
@@ -155,11 +152,6 @@ async def access_overview_payload(
     *,
     window_days: int,
 ) -> AccessLogOverviewResponse:
-    cache_key = f"access:overview:{window_days}"
-    cached = await redis_cache.get(cache_key)
-    if cached:
-        return AccessLogOverviewResponse.model_validate(cached)
-
     filters = [_RAG_ACTION_FILTER, _window_filter(window_days)]
 
     summary = (
@@ -204,7 +196,7 @@ async def access_overview_payload(
     total = int(summary.total or 0)
     errors = int(summary.errors or 0)
     successes = int(summary.successes or 0)
-    response = AccessLogOverviewResponse(
+    return AccessLogOverviewResponse(
         window_days=window_days,
         total_events=total,
         success_rate=round((successes / total) * 100, 2) if total else 0,
@@ -241,5 +233,3 @@ async def access_overview_payload(
         ],
         recent=[access_log_entry(row) for row in recent],
     )
-    await redis_cache.set(cache_key, jsonable_encoder(response), ttl=_ACCESS_OVERVIEW_TTL)
-    return response

api/bigrag/services/collection_scope.py

@@ -13,6 +13,9 @@
     ("GET", "/v1/collections"),
     ("GET", "/v1/usage"),
     ("GET", "/v1/stats"),
+    ("GET", "/v1/status/overview"),
+    ("GET", "/v1/status/collections"),
+    ("GET", "/v1/status/usage"),
     ("GET", "/v1/embeddings/models"),
 )
 

api/bigrag/services/connectors/progress.py

@@ -3,7 +3,6 @@
 from typing import Any
 
 from bigrag.db.models import ConnectorDocument, ConnectorSource, ConnectorSyncJob
-from bigrag.services.connectors.realtime import notify_connector_sync_jobs
 from bigrag.services.connectors.types import ConnectorSyncCounters, RemoteConnectorFile
 
 SYNC_PROGRESS_FIXED_PERCENT = {
@@ -98,5 +97,3 @@ async def update_sync_progress(
         ),
     }
     await session.commit()
-    if source is not None:
-        notify_connector_sync_jobs(job.provider, source.collection_name, str(source.id))

api/bigrag/services/connectors/scheduler.py

@@ -7,7 +7,6 @@
 
 from bigrag.db.engine import session_factory
 from bigrag.db.models import ConnectorSource, ConnectorSyncJob
-from bigrag.services.connectors.realtime import notify_connector_state
 from bigrag.services.connectors.sources import create_sync_job
 from bigrag.services.connectors.time import next_sync_at, utcnow
 
@@ -58,7 +57,6 @@ async def run_due_syncs(
     async with session_factory()() as session:
         await reap_stale_syncs(session, provider=provider)
     job_ids: list[str] = []
-    notifications: list[tuple[str, str, str]] = []
     async with session_factory()() as session:
         rows = (
             await session.scalars(
@@ -84,12 +82,9 @@ async def run_due_syncs(
                 commit=False,
             )
             await session.flush()
-            notifications.append((provider, source.collection_name, str(source.id)))
             if job.status == "pending" and job.started_at is None:
                 job_ids.append(str(job.id))
         await session.commit()
-    for event in notifications:
-        notify_connector_state(*event)
     for job_id in job_ids:
         start_sync_job(job_id)
     return len(job_ids)

api/bigrag/services/connectors/sources/sources_mutations.py

@@ -15,7 +15,6 @@
 )
 from bigrag.services import collection_cache
 from bigrag.services.connectors.progress import sync_progress_details
-from bigrag.services.connectors.realtime import notify_connector_sources, notify_connector_state
 from bigrag.services.connectors.sources.sources_credentials import upsert_source_credential
 from bigrag.services.connectors.sources.sources_queries import source_by_id
 from bigrag.services.connectors.time import next_sync_at, utcnow
@@ -146,7 +145,6 @@ async def create_source(
         )
         await session.commit()
         await session.refresh(existing)
-        notify_connector_state(provider, existing.collection_name, str(existing.id))
         if job.status == "pending" and job.started_at is None:
             start_sync_job(str(job.id))
         return existing, job
@@ -164,7 +162,6 @@ async def create_source(
     await session.commit()
     await session.refresh(source)
     await session.refresh(job)
-    notify_connector_state(provider, source.collection_name, str(source.id))
     if job.status == "pending" and job.started_at is None:
         start_sync_job(str(job.id))
     return source, job
@@ -197,7 +194,6 @@ async def trigger_sync(
     )
     if job.status == "pending" and job.started_at is None:
         start_sync_job(str(job.id))
-    notify_connector_state(provider, source.collection_name, str(source.id))
     return job
 
 
@@ -235,7 +231,6 @@ async def update_source(
     source.next_sync_at = next_sync_at(source)
     await session.commit()
     await session.refresh(source)
-    notify_connector_sources(provider, source.collection_name)
     return source
 
 
@@ -253,7 +248,6 @@ async def delete_source(
         not_found_message=not_found_message,
     )
     collection_name = source.collection_name
-    source_uuid = str(source.id)
     manifests = (
         await session.scalars(
             sa.select(ConnectorDocument).where(ConnectorDocument.source_id == source.id)
@@ -277,7 +271,6 @@ async def delete_source(
     await session.delete(source)
     await session.commit()
     await collection_cache.invalidate(collection_name)
-    notify_connector_state(provider, collection_name, source_uuid)
 
 
 def _tenant_id(collection: Collection, metadata: dict[str, Any]) -> str | None:

api/bigrag/services/connectors/status.py

@@ -5,7 +5,6 @@
 from bigrag.db.models import ConnectorSource, ConnectorSyncJob
 from bigrag.services.connectors.manifest import apply_counters
 from bigrag.services.connectors.progress import sync_counter_details, update_sync_progress
-from bigrag.services.connectors.realtime import notify_connector_sources
 from bigrag.services.connectors.time import next_sync_at, utcnow
 from bigrag.services.connectors.types import ConnectorSyncCounters
 from bigrag.services.webhook import enqueue_webhook_event
@@ -37,7 +36,6 @@ async def fail_sync(
         phase="failed",
         message=message,
     )
-    notify_connector_sources(job.provider, source.collection_name)
     data = {
         "provider": job.provider,
         "source_id": str(source.id),

api/bigrag/services/connectors/sync/finalize.py

@@ -5,7 +5,6 @@
 from bigrag.services import collection_cache
 from bigrag.services.connectors.manifest import apply_counters
 from bigrag.services.connectors.progress import sync_counter_details, update_sync_progress
-from bigrag.services.connectors.realtime import notify_connector_sources
 from bigrag.services.connectors.time import next_sync_at, utcnow
 from bigrag.services.connectors.types import ConnectorSyncAdapter, ConnectorSyncCounters
 from bigrag.services.documents import recount_collection_documents
@@ -58,7 +57,6 @@ async def finalize_sync(
         processed_items=counters.found + counters.deleted,
         total_items=counters.found + missing_count,
     )
-    notify_connector_sources(adapter.provider, source.collection_name)
     await collection_cache.invalidate(source.collection_name)
     await invalidate_collection_query_cache(source.collection_name)
     webhook_event = (