Spaces:
Runtime error
Runtime error
File size: 2,299 Bytes
abb6f94 |
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 |
from fastapi import APIRouter, Request,Body, HTTPException
import logging
from typing import Dict
from app.rag_pipeline.model_initializer import initialize_models
from app.rag_pipeline.retriever_chain import RetrieverChain
from app.settings import Config
answer_router = APIRouter()
logger = logging.getLogger(__name__)
import warnings
warnings.filterwarnings("ignore")
conf = Config()
OPENAI_API_KEY = conf.API_KEY
MODEL_ID = conf.MODEL_ID
MODEL_BASENAME = conf.MODEL_BASENAME
COLLECTION_NAME = conf.COLLECTION_NAME
PERSIST_DIRECTORY = conf.PERSIST_DIRECTORY
# print(OPENAI_API_KEY, MODEL_ID, MODEL_BASENAME, PERSIST_DIRECTORY, COLLECTION_NAME)
embedding_model, llm_model = initialize_models(OPENAI_API_KEY,model_id=MODEL_ID, model_basename=MODEL_BASENAME)
def validate_question(data: Dict) -> str:
"""Extract and validate the 'question' field from the incoming data."""
question = data.get("question")
if not question or not isinstance(question, str) or not question.strip():
logger.warning("Received invalid question input.")
raise HTTPException(status_code=400, detail="Question must be a non-empty string.")
return question
@answer_router.post('/answer', response_model=dict)
async def generate_answer(data: Dict = Body(...)) -> Dict:
try:
# Validate and extract the question
question = validate_question(data)
# Log incoming question
logger.info(f"Received question: {question}")
# Generate the answer
retriever_qa = RetrieverChain(
collection_name=COLLECTION_NAME, embedding_function=embedding_model, persist_directory=PERSIST_DIRECTORY)
answer = retriever_qa.get_response(user_input = question, llm= llm_model)
# answer = f"Generated answer for: {question}"
# Log generated answer
logger.info(f"Generated answer: {answer}")
return {"answer": answer}
except HTTPException as http_exc:
logger.error(f"HTTP error: {http_exc.detail}")
raise http_exc # Re-raise the HTTPException to return the error response
except Exception as e:
logger.error(f"Unexpected error: {str(e)}")
raise HTTPException(status_code=500, detail="An internal server error occurred.")
|