| from __future__ import annotations |
|
|
| from typing import Any |
| from fastapi import FastAPI, HTTPException |
| from pydantic import BaseModel, Field |
|
|
| from .core import DocumentRecord, FuzzyIndex, FuzzyRetrievalConfig |
| from .graph import FuzzyRAGGraph |
|
|
|
|
| class IngestDocument(BaseModel): |
| id: str |
| text: str |
| title: str = "" |
| source: str = "" |
| authority: float = Field(0.5, ge=0.0, le=1.0) |
| freshness: float = Field(0.5, ge=0.0, le=1.0) |
| reliability: float = Field(0.5, ge=0.0, le=1.0) |
| metadata: dict[str, Any] = Field(default_factory=dict) |
|
|
|
|
| class IngestRequest(BaseModel): |
| documents: list[IngestDocument] |
| replace: bool = True |
|
|
|
|
| class QueryRequest(BaseModel): |
| question: str |
| k: int = Field(4, ge=1, le=20) |
|
|
|
|
| def create_app() -> FastAPI: |
| app = FastAPI(title="Fuzzy LangGraph RAG API", version="1.0.0") |
| holder: dict[str, Any] = {"index": FuzzyIndex(config=FuzzyRetrievalConfig()), "graph": None, "k": 4} |
|
|
| @app.get("/health") |
| def health(): |
| return { |
| "status": "ok", |
| "documents": len(holder["index"].documents), |
| "graph_ready": holder["graph"] is not None, |
| } |
|
|
| @app.post("/ingest") |
| def ingest(req: IngestRequest): |
| docs: list[DocumentRecord] = [] |
| for d in req.documents: |
| md = dict(d.metadata) |
| md.update({"authority": d.authority, "freshness": d.freshness, "reliability": d.reliability}) |
| docs.append(DocumentRecord(d.id, d.text, d.title, d.source, md)) |
| if req.replace: |
| holder["index"].fit(docs) |
| else: |
| holder["index"].add(docs) |
| holder["graph"] = FuzzyRAGGraph(holder["index"], initial_k=holder["k"]) |
| return {"documents": len(holder["index"].documents)} |
|
|
| @app.post("/query") |
| def query(req: QueryRequest): |
| if not holder["index"].documents: |
| raise HTTPException(status_code=400, detail="No documents ingested") |
| if holder["graph"] is None or req.k != holder["k"]: |
| holder["k"] = req.k |
| holder["graph"] = FuzzyRAGGraph(holder["index"], initial_k=req.k) |
| return holder["graph"].invoke(req.question) |
|
|
| return app |
|
|
|
|
| app = create_app() |
|
|
