Matteo Benedetto
4776b4b496
- gateway: separate config, models, state, audit, guardrail, embeddings, store, metrics, cleanup and routes; keep main.py as FastAPI bootstrap - extension: split client/config, six tools, config command and rules; preserve jiti entrypoint and registrations - Dockerfile copies the complete gateway module set - tests: update monkeypatch boundaries for modular config/state
110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
"""Memory Gateway — bootstrap FastAPI, lifecycle e middleware.
|
|
|
|
Gli endpoint e la logica di dominio sono separati in moduli:
|
|
config, models, state, audit, guardrail, embed, store, metrics, cleanup,
|
|
routes. Il contratto HTTP resta invariato.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
import uuid
|
|
from contextlib import asynccontextmanager
|
|
|
|
import uvicorn
|
|
from fastapi import FastAPI, Request
|
|
from qdrant_client.http import models as qm
|
|
|
|
import cleanup
|
|
import embed as embedding
|
|
import metrics
|
|
import state
|
|
from config import (
|
|
COLLECTION,
|
|
EMBED_DIM,
|
|
METRICS_ENABLED,
|
|
SPARSE_VECTOR_NAME,
|
|
GATEWAY_VERSION,
|
|
log,
|
|
)
|
|
from routes import router
|
|
|
|
# Alias utili per compatibilità con import/debug locali; lo stato effettivo è in state.py.
|
|
qdrant = state.qdrant
|
|
embed = embedding.embed
|
|
state.embed = embedding.embed
|
|
state.sparse_encode = embedding.sparse_encode
|
|
|
|
|
|
async def _lifespan(_app: FastAPI):
|
|
"""Crea collection/indici e avvia i loop periodici."""
|
|
collections = state.qdrant.get_collections().collections
|
|
if not any(c.name == COLLECTION for c in collections):
|
|
state.qdrant.create_collection(
|
|
collection_name=COLLECTION,
|
|
vectors_config=qm.VectorParams(size=EMBED_DIM, distance=qm.Distance.COSINE),
|
|
sparse_vectors_config={SPARSE_VECTOR_NAME: qm.SparseVectorParams(modifier=qm.Modifier.IDF)},
|
|
)
|
|
for field in ("agent_id", "project_id", "scope", "kind", "supersedes_id", "superseded_by", "text_hash", "parent_id", "level", "topic"):
|
|
state.qdrant.create_payload_index(collection_name=COLLECTION, field_name=field, field_schema=qm.PayloadSchemaType.KEYWORD)
|
|
state.qdrant.create_payload_index(collection_name=COLLECTION, field_name="text", field_schema=qm.PayloadSchemaType.TEXT)
|
|
log.info("collection %s creata con indici (dense + sparse %s)", COLLECTION, SPARSE_VECTOR_NAME)
|
|
else:
|
|
log.info("collection %s già esistente", COLLECTION)
|
|
for field in ("parent_id", "level", "topic"):
|
|
try:
|
|
state.qdrant.create_payload_index(collection_name=COLLECTION, field_name=field, field_schema=qm.PayloadSchemaType.KEYWORD)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
info = state.qdrant.get_collection(COLLECTION)
|
|
sparse_vectors = (info.config.params.sparse_vectors or {}) if info.config and info.config.params else {}
|
|
if SPARSE_VECTOR_NAME not in sparse_vectors:
|
|
state.qdrant.create_vector_name(COLLECTION, SPARSE_VECTOR_NAME, qm.SparseVectorNameConfig(sparse=qm.SparseVectorConfig(modifier=qm.Modifier.IDF)))
|
|
log.info("sparse vector %s aggiunto alla collection esistente", SPARSE_VECTOR_NAME)
|
|
embedding.backfill_sparse(state.qdrant, COLLECTION)
|
|
|
|
cleanup_task = asyncio.create_task(cleanup.loop(state.qdrant, COLLECTION, state.invalidate_meta))
|
|
metrics_task = asyncio.create_task(metrics.push_loop(state.qdrant, COLLECTION, embedding.get_http)) if METRICS_ENABLED else None
|
|
try:
|
|
yield
|
|
finally:
|
|
cleanup_task.cancel()
|
|
try:
|
|
await cleanup_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
if metrics_task is not None:
|
|
metrics_task.cancel()
|
|
try:
|
|
await metrics_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
await embedding.close_http()
|
|
|
|
|
|
app = FastAPI(title="Memory Gateway", version=GATEWAY_VERSION, lifespan=_lifespan)
|
|
app.include_router(router)
|
|
|
|
|
|
@app.middleware("http")
|
|
async def request_id_middleware(request: Request, call_next):
|
|
rid = request.headers.get("X-Request-ID") or str(uuid.uuid4())
|
|
state.request_id.set(rid)
|
|
response = await call_next(request)
|
|
response.headers["X-Request-ID"] = rid
|
|
return response
|
|
|
|
|
|
@app.middleware("http")
|
|
async def metrics_middleware(request: Request, call_next):
|
|
start = time.monotonic()
|
|
response = await call_next(request)
|
|
route = request.scope.get("route")
|
|
endpoint = route.path if route else request.url.path
|
|
metrics.record_request(endpoint, time.monotonic() - start, response.status_code)
|
|
return response
|
|
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=8080)
|