Spaces:
Runtime error
Runtime error
| from fastapi import FastAPI, HTTPException | |
| from pydantic import BaseModel | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.responses import FileResponse | |
| from ingest import ingest_repository | |
| from query import ( | |
| VECTORSTORE_CACHE, | |
| MEMORY_CACHE, | |
| initialize_repo_caches, | |
| ask_question, | |
| ) | |
| app = FastAPI(title="RAG Backend", version="1.0.0") | |
| app.mount("/static", StaticFiles(directory="static"), name="static") | |
| def index(): | |
| return FileResponse("static/index.html") | |
| class LoadRepoRequest(BaseModel): | |
| repo_url: str | |
| class AskRequest(BaseModel): | |
| repo_name: str | |
| question: str | |
| def load_repo(payload: LoadRepoRequest): | |
| repo_url = payload.repo_url.strip() | |
| if not repo_url: | |
| raise HTTPException(status_code=400, detail="repo_url is required") | |
| repo_name = ingest_repository(repo_url) | |
| initialize_repo_caches(repo_name) | |
| print("AFTER LOAD:", VECTORSTORE_CACHE.keys(), MEMORY_CACHE.keys()) | |
| return { | |
| "status": "success", | |
| "repo": repo_name, | |
| } | |
| def ask(payload: AskRequest): | |
| repo_name = payload.repo_name.strip() | |
| question = payload.question.strip() | |
| if not repo_name: | |
| raise HTTPException(status_code=400, detail="repo_name is required") | |
| if not question: | |
| raise HTTPException(status_code=400, detail="question is required") | |
| if repo_name not in VECTORSTORE_CACHE or repo_name not in MEMORY_CACHE: | |
| raise HTTPException(status_code=400, detail="repo not loaded") | |
| answer, docs = ask_question(question, repo_name) | |
| sources = [] | |
| seen = set() | |
| for doc in docs: | |
| path = doc.metadata.get("path") | |
| if path and path not in seen: | |
| seen.add(path) | |
| sources.append(path) | |
| return { | |
| "answer": answer, | |
| "sources": sources, | |
| } | |