What it does
Multi-user access with auth
Admin uploads documents
Cited answers with source links
Department-based access control
Stack
FastAPILlamaIndexOpenAI / ClaudeSupabase pgvectorReact UI
Deploy on
✓ Vercel + Supabase✓ Railway✓ AWS
Full source code
Install commands are in the top comments. Copy and run.
# Company Docs RAG Agent — Production Ready
# Stack: FastAPI + LlamaIndex + Supabase pgvector + OpenAI
# pip install fastapi uvicorn llama-index llama-index-vector-stores-supabase
# python-multipart python-jose supabase openai
from fastapi import FastAPI, UploadFile, File, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordBearer
from llama_index.core import VectorStoreIndex, Document, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.supabase import SupabaseVectorStore
from supabase import create_client
import os, tempfile
from typing import Optional
from pydantic import BaseModel
app = FastAPI(title="Company Docs Agent API")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
# ── SETUP ─────────────────────────────────────────────────────────
SUPABASE_URL = os.environ["SUPABASE_URL"]
SUPABASE_KEY = os.environ["SUPABASE_SERVICE_KEY"]
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
Settings.llm = OpenAI(model="gpt-4o-mini", api_key=OPENAI_API_KEY) # Cheap + fast
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small", api_key=OPENAI_API_KEY)
# ── SUPABASE SETUP ─────────────────────────────────────────────────
# Run this SQL in Supabase first:
'''
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE documents (
id bigserial PRIMARY KEY,
content text,
metadata jsonb,
embedding vector(1536),
department text DEFAULT 'all',
created_at timestamptz DEFAULT now()
);
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops);
'''
# ── MODELS ────────────────────────────────────────────────────────
class QueryRequest(BaseModel):
question: str
department: Optional[str] = None
user_id: str
class QueryResponse(BaseModel):
answer: str
sources: list[dict]
confidence: float
# ── DOCUMENT UPLOAD ────────────────────────────────────────────────
@app.post("/upload")
async def upload_document(
file: UploadFile = File(...),
department: str = "all",
admin_key: str = ""
):
"""Admin endpoint: upload a PDF/DOCX/TXT to the knowledge base"""
if admin_key != os.environ.get("ADMIN_KEY", "changeme"):
raise HTTPException(403, "Invalid admin key")
# Save temp file
suffix = "." + file.filename.split(".")[-1]
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
content = await file.read()
tmp.write(content)
tmp_path = tmp.name
try:
from llama_index.core import SimpleDirectoryReader
import os as _os
# Load and chunk the document
reader = SimpleDirectoryReader(input_files=[tmp_path])
documents = reader.load_data()
# Add department metadata
for doc in documents:
doc.metadata["department"] = department
doc.metadata["filename"] = file.filename
# Index into Supabase
vector_store = SupabaseVectorStore(
postgres_connection_string=os.environ["SUPABASE_DB_URL"],
collection_name="documents"
)
index = VectorStoreIndex.from_documents(documents, vector_store=vector_store)
_os.unlink(tmp_path)
return {"status": "success", "chunks": len(documents), "file": file.filename}
except Exception as e:
_os.unlink(tmp_path)
raise HTTPException(500, str(e))
# ── QUERY ENDPOINT ────────────────────────────────────────────────
@app.post("/query", response_model=QueryResponse)
async def query_docs(request: QueryRequest):
"""Query the company knowledge base"""
try:
vector_store = SupabaseVectorStore(
postgres_connection_string=os.environ["SUPABASE_DB_URL"],
collection_name="documents"
)
index = VectorStoreIndex.from_vector_store(vector_store)
# System prompt for company assistant
system_prompt = """You are a helpful company assistant. Answer questions based ONLY on
the provided company documents. If the answer isn't in the documents, say so clearly.
Always cite which document your answer comes from."""
query_engine = index.as_query_engine(
similarity_top_k=4,
system_prompt=system_prompt
)
response = query_engine.query(request.question)
sources = []
for node in response.source_nodes:
sources.append({
"filename": node.metadata.get("filename", "Unknown"),
"department": node.metadata.get("department", "general"),
"relevance_score": round(node.score or 0, 3),
"excerpt": node.text[:200] + "..."
})
avg_score = sum(s["relevance_score"] for s in sources) / len(sources) if sources else 0
return QueryResponse(
answer=str(response),
sources=sources,
confidence=round(avg_score, 2)
)
except Exception as e:
raise HTTPException(500, f"Query failed: {str(e)}")
@app.get("/health")
def health():
return {"status": "ok", "service": "Company Docs Agent"}
# Run with: uvicorn main:app --reload --port 8000