Spaces:
Sleeping
Sleeping
File size: 3,904 Bytes
60b97da | 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 | import os
from pathlib import Path
from typing import List, Literal, Optional
from fastapi import BackgroundTasks, Depends, FastAPI, Header, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field, HttpUrl
from dotenv import load_dotenv
from src.rag_system import CodebaseRAGSystem
load_dotenv(Path(__file__).with_name(".env"))
app = FastAPI(
title="Codebase RAG API",
description="Index GitHub repositories and answer natural-language questions with grounded citations.",
version="2.0.0",
)
cors_origins = [
origin.strip()
for origin in os.getenv("CORS_ORIGINS", "http://localhost:3000").split(",")
if origin.strip()
]
app.add_middleware(
CORSMiddleware,
allow_origins=cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
rag_system: Optional[CodebaseRAGSystem] = None
class RepoIndexRequest(BaseModel):
github_url: HttpUrl
class QueryRequest(BaseModel):
repo_id: int = Field(..., ge=1)
question: str = Field(..., min_length=3)
top_k: int = Field(8, ge=3, le=12)
history: List["MessageTurn"] = Field(default_factory=list, max_length=8)
class MessageTurn(BaseModel):
role: Literal["user", "assistant"]
content: str = Field(..., min_length=1, max_length=4000)
def require_session_id(x_session_id: Optional[str] = Header(None, alias="X-Session-Id")) -> str:
if not x_session_id or not x_session_id.strip():
raise HTTPException(status_code=400, detail="Missing session id")
return x_session_id.strip()
@app.on_event("startup")
def startup():
global rag_system
Path("./data").mkdir(exist_ok=True)
rag_system = CodebaseRAGSystem()
@app.get("/")
async def root():
return {
"status": "online",
"message": "Codebase RAG API is running",
}
@app.get("/api/health")
async def health():
return {
"status": "ok",
}
@app.get("/api/repos")
async def list_repositories(session_id: str = Depends(require_session_id)):
return rag_system.list_repositories_for_session(session_id)
@app.get("/api/repos/{repo_id}")
async def get_repository(repo_id: int, session_id: str = Depends(require_session_id)):
repo = rag_system.get_repository_for_session(repo_id, session_id)
if not repo:
raise HTTPException(status_code=404, detail="Repository not found")
return repo
@app.post("/api/repos/index")
async def queue_repository_index(
request: RepoIndexRequest,
background_tasks: BackgroundTasks,
session_id: str = Depends(require_session_id),
):
try:
repo = rag_system.create_or_reset_repository(str(request.github_url), session_id)
background_tasks.add_task(rag_system.index_repository, repo.id)
return {
"success": True,
"message": "Repository indexing started",
"repo": rag_system.get_repository_for_session(repo.id, session_id),
}
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc))
@app.post("/api/query")
async def query_repository(request: QueryRequest, session_id: str = Depends(require_session_id)):
try:
return rag_system.answer_question(
repo_id=request.repo_id,
session_key=session_id,
question=request.question.strip(),
top_k=request.top_k,
history=request.history,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
except Exception as exc:
raise HTTPException(status_code=500, detail=str(exc))
@app.post("/api/session/end")
async def end_session(session_id: str = Query(..., min_length=8)):
rag_system.end_session(session_id)
return {"success": True}
if __name__ == "__main__":
import uvicorn
uvicorn.run("server_app:app", host="0.0.0.0", port=8000, reload=True)
|