FastAPI를 활용한 RAG API 구축 및 배포 가이드

def ingest_document(file_path: str, filename: str = "") -> int:
    # 1. Load
    loader = PyPDFLoader(file_path) if file_path.endswith(".pdf") else TextLoader(file_path)
    documents = loader.load()
    
    # 2. Chunk
    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
    chunks = splitter.split_documents(documents)
    
    # 3. Embed and Store
    global _vectorstore
    if _vectorstore is None:
        _vectorstore = FAISS.from_documents(chunks, embeddings)
    else:
        _vectorstore.add_documents(chunks)
    
    # 4. Persist to disk
    _vectorstore.save_local(FAISS_INDEX_PATH)
    return len(chunks)

def query_rag(question: str, top_k: int = 4) -> dict:
    vs = _load_vectorstore()
    retriever = vs.as_retriever(search_kwargs={"k": top_k})
    
    prompt = PromptTemplate.from_template("""Context: {context}
Question: {question}
Answer:""")
    llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
    
    # LCEL chain
    retrieve = RunnableParallel({"context": retriever | _format_docs, "question": RunnablePassthrough()})
    answer_chain = prompt | llm | StrOutputParser()
    
    retrieved = retrieve.invoke(question)
    answer = answer_chain.invoke(retrieved)
    return {"answer": answer, "sources": [doc.metadata.get("source") for doc in retrieved["context"]]}

@app.post("/query", response_model=QueryResponse)
def query(request: QueryRequest):
    if not request.question.strip():
        raise HTTPException(status_code=400, detail="Question cannot be empty.")
    try:
        result = query_rag(request.question, request.top_k)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    return QueryResponse(answer=result["answer"], sources=result["sources"])