170 lines
7.1 KiB
Python
170 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import nullcontext
|
|
from hmac import compare_digest
|
|
from time import perf_counter
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, Request, Response
|
|
|
|
from adapters.langfuse_scores import LangfuseScoreClient
|
|
from adapters.postgres import PostgresTraceRepository
|
|
from bootstrap import build_runtime
|
|
from config import Settings, get_settings
|
|
from rag.answer import GroundedAnswerService
|
|
from rag.metrics import REQUEST_DURATION, REQUESTS, NullMetrics
|
|
from rag.telemetry import (
|
|
annotate_current_span,
|
|
configure_telemetry,
|
|
correlation_context,
|
|
current_trace_id,
|
|
request_span,
|
|
)
|
|
from routers.rag import router as rag_router
|
|
|
|
|
|
def create_app(
|
|
*,
|
|
settings: Settings | None = None,
|
|
answer_service: GroundedAnswerService | None = None,
|
|
conversational: Any | None = None,
|
|
trace_writer: PostgresTraceRepository | None = None,
|
|
metrics: Any | None = None,
|
|
section_retriever: Any | None = None,
|
|
) -> FastAPI:
|
|
configured = settings or get_settings()
|
|
effective_metrics = metrics or NullMetrics()
|
|
configure_telemetry(configured, effective_metrics)
|
|
app = FastAPI(title=configured.app_name, version="0.1.0")
|
|
app.state.answer_service = answer_service
|
|
app.state.conversational = conversational
|
|
app.state.trace_writer = trace_writer
|
|
app.state.metrics = metrics
|
|
# None unless Langfuse is fully configured; `/v1/rag/feedback` then skips
|
|
# mirroring the user's verdict there. Built from the same settings the
|
|
# span exporter reads, so the two can never point at different projects.
|
|
app.state.langfuse_scores = LangfuseScoreClient.from_settings(configured)
|
|
# The raw Qdrant retriever, not routed through RagAgent — Feature-List
|
|
# #4/#23's section-list and verbatim-section-text endpoints are plain
|
|
# payload-filtered reads with no LLM/generation step, so nothing about
|
|
# them belongs on the answer/agent path. None whenever embedding is
|
|
# disabled (no Qdrant client exists at all in that mode).
|
|
app.state.section_retriever = section_retriever
|
|
|
|
@app.middleware("http")
|
|
async def correlate_and_trace(request: Request, call_next):
|
|
route = _route_label(request.url.path)
|
|
method = request.method.upper()
|
|
started = perf_counter()
|
|
status_code = 500
|
|
# Kubernetes probes and the Prometheus scrape run every few seconds
|
|
# forever and carry no information a trace could show. Left traced they
|
|
# were ~90% of all spans within a day, burying the RAG turns anyone
|
|
# actually opens a trace viewer to read, and costing storage and
|
|
# ingestion for nothing. Their Prometheus counters below are untouched,
|
|
# which is the right place to watch probe health anyway.
|
|
span_context = (
|
|
nullcontext(None)
|
|
if route in _UNTRACED_ROUTES
|
|
else request_span(method, route, request.headers)
|
|
)
|
|
with correlation_context(request.headers.get("x-correlation-id")) as correlation_id:
|
|
with span_context as span:
|
|
try:
|
|
response = await call_next(request)
|
|
status_code = response.status_code
|
|
response.headers["X-Correlation-ID"] = correlation_id
|
|
trace_id = current_trace_id()
|
|
if trace_id:
|
|
response.headers["X-Trace-ID"] = trace_id
|
|
return response
|
|
finally:
|
|
status = f"{status_code // 100}xx"
|
|
elapsed = perf_counter() - started
|
|
effective_metrics.increment(
|
|
REQUESTS, method=method, route=route, status=status
|
|
)
|
|
effective_metrics.observe(
|
|
REQUEST_DURATION,
|
|
elapsed,
|
|
method=method,
|
|
route=route,
|
|
status=status,
|
|
)
|
|
if span is not None:
|
|
span.set_attribute("http.response.status_code", status_code)
|
|
annotate_current_span(
|
|
**{
|
|
"duocthu.http.status_class": status,
|
|
"duocthu.duration_ms": elapsed * 1000,
|
|
}
|
|
)
|
|
|
|
@app.get("/health")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
@app.get("/ready")
|
|
def ready() -> Response:
|
|
# Runtime construction already verifies the Qdrant corpus manifest and
|
|
# fails startup on mismatch. PostgreSQL trace/history are intentionally
|
|
# fail-open and therefore must not make readiness flap.
|
|
if app.state.answer_service is None and configured.embedding_provider != "disabled":
|
|
return Response(status_code=503)
|
|
return Response(content='{"status":"ready"}', media_type="application/json")
|
|
|
|
@app.get("/metrics")
|
|
def prometheus_metrics(request: Request) -> Response:
|
|
# Optional bearer token. Today this endpoint is unreachable from the
|
|
# internet — Caddy proxies only `web`, and ai-service publishes no
|
|
# host port — so an unset token keeps local development and the
|
|
# current compose scrape working unchanged. It stops being safe the
|
|
# moment the service is exposed through an Ingress, which the Helm
|
|
# chart now makes possible, so the guard lives here rather than in
|
|
# whichever deployment happens to expose it first.
|
|
expected = configured.metrics_token
|
|
if expected:
|
|
supplied = request.headers.get("authorization", "")
|
|
prefix = "Bearer "
|
|
token = supplied[len(prefix):] if supplied.startswith(prefix) else ""
|
|
# Constant-time compare: a scrape token is a shared secret, and
|
|
# `==` on a secret leaks its prefix through timing.
|
|
if not compare_digest(token, expected):
|
|
return Response(status_code=401)
|
|
exporter = getattr(app.state, "metrics", None)
|
|
if exporter is None or not hasattr(exporter, "render"):
|
|
# 404 rather than an empty 200: a scrape that silently succeeds
|
|
# with no samples looks identical to a service answering nothing.
|
|
return Response(status_code=404)
|
|
body, content_type = exporter.render()
|
|
return Response(content=body, media_type=content_type)
|
|
|
|
app.include_router(rag_router)
|
|
return app
|
|
|
|
|
|
_UNTRACED_ROUTES = frozenset({"/health", "/ready", "/metrics"})
|
|
|
|
|
|
def _route_label(path: str) -> str:
|
|
known = {
|
|
"/health", "/ready", "/metrics", "/v1/rag/query", "/v1/rag/suggest",
|
|
"/v1/rag/feedback", "/v1/rag/history", "/v1/rag/sections",
|
|
"/v1/rag/section-text",
|
|
}
|
|
return path if path in known else "other"
|
|
|
|
|
|
_settings = get_settings()
|
|
_answer_service, _conversational, _trace_writer, _metrics, _section_retriever = (
|
|
build_runtime(_settings)
|
|
)
|
|
app = create_app(
|
|
settings=_settings,
|
|
answer_service=_answer_service,
|
|
conversational=_conversational,
|
|
trace_writer=_trace_writer,
|
|
metrics=_metrics,
|
|
section_retriever=_section_retriever,
|
|
)
|