File size: 2,197 Bytes
7b4d34e | 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 | 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()
|