Spaces:
Running
Running
File size: 11,658 Bytes
5ea3240 f4b92b8 5ea3240 ef2c57c 5ea3240 1121d82 5ea3240 13784a6 bb5d2bb 27716f7 b0af996 03bed0b f79a242 f0307a2 13784a6 5ea3240 f4b92b8 f0307a2 f4b92b8 27716f7 b0af996 27716f7 5ea3240 ef2c57c 27716f7 b0af996 27716f7 bb5d2bb 27716f7 b0af996 27716f7 bb5d2bb ef2c57c 5ea3240 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 | from __future__ import annotations
import tempfile
import time
from pathlib import Path
from typing import Annotated
from fastapi import Depends, FastAPI, File, Form, Header, HTTPException, Request, UploadFile
from fastapi.responses import Response
from prometheus_client import CONTENT_TYPE_LATEST, Counter, Histogram, generate_latest
from .config import get_settings
from .evaluation import demo_benchmark_metadata, run_demo_eval
from .pipeline import RAGEngine
from .rate_limit import RateLimitExceeded, limiter
from .schemas import CorpusSummary, EvaluationRequest, QueryRequest, QueryResponse, SessionResponse
from .workspace import registry
REQUESTS = Counter("ragforge_requests_total", "Total API requests", ["endpoint", "status"])
LATENCY = Histogram("ragforge_request_latency_seconds", "API request latency", ["endpoint"])
def _auth(authorization: Annotated[str | None, Header()] = None) -> None:
token = get_settings().app_api_token
if not token:
return
if authorization != f"Bearer {token}":
raise HTTPException(status_code=401, detail="Invalid or missing Bearer token")
def create_api() -> FastAPI:
app = FastAPI(title="RAGForge API", version="2.0.3")
@app.get("/api/health")
def health():
return {"status": "ok", "service": "RAGForge", "model": get_settings().default_model}
@app.get("/api/v1/info")
def info():
s = get_settings()
return {
"app": s.app_name,
"default_model": s.default_model,
"embedding_model": s.embedding_model,
"reranker_model": s.reranker_model,
"native_search_model": s.native_search_model,
"features": [
"semantic-query-planning", "source-profile-index", "hierarchical-retrieval",
"source-balanced-global-retrieval", "hybrid-search", "reranking", "hyde",
"corrective-rag", "conditional-web", "self-rag", "text2sql", "ask-the-web",
"citations", "guardrails", "layered-evaluation", "retrieval-ablation", "quality-gated-evaluation",
"cache-bypassed-benchmarking", "quota-aware-evaluation", "retry-after-backoff",
"workspace-preflight", "browser-session-continuity", "lazy-demo-recovery", "explicit-abstention",
"saved-evaluation-history", "incremental-deep-evaluation", "typed-text2sql-evaluation",
"adaptive-reranking", "semantic-citation-attribution", "insight-synthesis",
"table-citations", "hard-mode-evaluation", "chunk-level-reranker-ablation",
"optional-profile-benchmark", "node-latency-observability", "evaluation-run-history",
"grounded-absence-handling", "markdown-aware-citation-coverage", "evaluation-run-provenance",
"profile-policy-diagnostics", "context-efficiency-diagnostics",
"focused-context-pruning", "adaptive-context-budgeting", "adaptive-retrieval-depth",
"focused-evidence-compression", "context-budget-ablation", "compression-ablation",
"scale-stress-evaluation", "release-readiness-checklist", "prompt-budget-observability",
"retrieval-confidence-telemetry", "workspace-health-diagnostics"
],
}
@app.post("/api/v1/session", response_model=SessionResponse, dependencies=[Depends(_auth)])
def new_session():
ws = registry.create()
return SessionResponse(session_id=ws.session_id)
@app.get("/api/v1/session/{session_id}", dependencies=[Depends(_auth)])
def session_status(session_id: str):
try:
return registry.require(session_id).stats()
except Exception as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.get("/api/v1/session/{session_id}/diagnostics", dependencies=[Depends(_auth)])
def session_diagnostics(session_id: str):
try:
return registry.require(session_id).health_snapshot()
except Exception as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.get("/api/v1/evaluation/benchmark")
def evaluation_benchmark():
return demo_benchmark_metadata()
@app.get("/api/v1/evaluation/saved/{session_id}", dependencies=[Depends(_auth)])
def saved_evaluations(session_id: str):
try:
return {"session_id": session_id, "runs": registry.require(session_id).evaluation_inventory()}
except Exception as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.get("/api/v1/evaluation/history/{session_id}", dependencies=[Depends(_auth)])
def evaluation_history(session_id: str):
try:
return {"session_id": session_id, "runs": registry.require(session_id).evaluation_history_inventory()}
except Exception as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.get("/api/v1/evaluation/saved/{session_id}/{level}", dependencies=[Depends(_auth)])
def saved_evaluation(session_id: str, level: str):
normalized = level.strip().title()
if normalized not in {"Quick", "Standard", "Deep"}:
raise HTTPException(status_code=400, detail="level must be Quick, Standard or Deep")
try:
ws = registry.require(session_id)
report = ws.get_evaluation(normalized, require_current_corpus=False)
if not report:
raise HTTPException(status_code=404, detail=f"No saved {normalized} evaluation")
return report
except HTTPException:
raise
except Exception as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.post("/api/v1/ingest", response_model=CorpusSummary, dependencies=[Depends(_auth)])
async def ingest(
request: Request,
session_id: Annotated[str, Form()],
files: Annotated[list[UploadFile], File()],
use_ocr: Annotated[bool, Form()] = False,
semantic_chunking: Annotated[bool, Form()] = False,
):
started = time.perf_counter()
try:
client = request.client.host if request.client else "unknown"
limiter.check(f"api-ingest:{client}")
ws = registry.require(session_id)
paths = []
max_bytes = get_settings().max_upload_mb * 1024 * 1024
with tempfile.TemporaryDirectory() as td:
for upload in files:
target = Path(td) / Path(upload.filename or "upload").name
written = 0
with target.open("wb") as fh:
while True:
block = await upload.read(1024 * 1024)
if not block:
break
written += len(block)
if written > max_bytes:
raise ValueError(f"{target.name} exceeds the configured upload limit")
fh.write(block)
paths.append(target)
result = ws.ingest(paths, ocr=use_ocr, semantic_chunking=semantic_chunking)
REQUESTS.labels("ingest", "ok").inc()
return result
except RateLimitExceeded as exc:
REQUESTS.labels("ingest", "rate_limited").inc()
raise HTTPException(status_code=429, detail=str(exc)) from exc
except Exception as exc:
REQUESTS.labels("ingest", "error").inc()
raise HTTPException(status_code=400, detail=str(exc)) from exc
finally:
LATENCY.labels("ingest").observe(time.perf_counter() - started)
@app.post("/api/v1/query", response_model=QueryResponse, dependencies=[Depends(_auth)])
def query(payload: QueryRequest, request: Request):
started = time.perf_counter()
try:
client = request.client.host if request.client else "unknown"
limiter.check(f"api:{client}")
ws = registry.require(payload.session_id)
result = RAGEngine(ws).ask(payload.query, payload.config)
REQUESTS.labels("query", "ok").inc()
return result
except RateLimitExceeded as exc:
REQUESTS.labels("query", "rate_limited").inc()
raise HTTPException(status_code=429, detail=str(exc)) from exc
except Exception as exc:
REQUESTS.labels("query", "error").inc()
raise HTTPException(status_code=400, detail=str(exc)) from exc
finally:
LATENCY.labels("query").observe(time.perf_counter() - started)
@app.post("/api/v1/evaluate/demo", dependencies=[Depends(_auth)])
def evaluate_demo(payload: EvaluationRequest, request: Request):
started = time.perf_counter()
try:
client = request.client.host if request.client else "unknown"
limiter.check(f"api-eval:{client}")
ws = registry.require(payload.session_id)
required_demo_sources = {
"acme_cloud_runbook.md",
"orbitpay_policy.txt",
"release_notes.html",
"support_matrix.csv",
"NIST_AI_RMF_1.0.pdf",
}
if not required_demo_sources.issubset(set(ws.sources)):
raise ValueError(
"The bundled demo benchmark requires the five bundled demo sources to be indexed in this session."
)
benchmark_version = str(demo_benchmark_metadata().get("version", ""))
if payload.reuse_saved:
cached = ws.get_evaluation(
payload.level,
model=payload.model,
benchmark_version=benchmark_version,
require_current_corpus=True,
)
if cached:
REQUESTS.labels("evaluate_demo", "ok").inc()
return cached
standard_base = None
if payload.level == "Deep" and payload.reuse_saved and not payload.include_profile_benchmark:
standard_base = ws.get_evaluation(
"Standard",
model=payload.model,
benchmark_version=benchmark_version,
require_current_corpus=True,
)
report = run_demo_eval(
ws,
api_key=None,
model=payload.model,
level=payload.level,
target_rpm=payload.target_rpm,
base_standard_report=standard_base,
include_profile_benchmark=payload.include_profile_benchmark,
)
report = ws.save_evaluation(
payload.level,
report,
model=payload.model,
benchmark_version=benchmark_version,
)
REQUESTS.labels("evaluate_demo", "ok").inc()
return report
except RateLimitExceeded as exc:
REQUESTS.labels("evaluate_demo", "rate_limited").inc()
raise HTTPException(status_code=429, detail=str(exc)) from exc
except Exception as exc:
REQUESTS.labels("evaluate_demo", "error").inc()
raise HTTPException(status_code=400, detail=str(exc)) from exc
finally:
LATENCY.labels("evaluate_demo").observe(time.perf_counter() - started)
@app.get("/metrics", include_in_schema=False)
def metrics():
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
return app
|