Multimedika/Bot_Development

1import json2import logging3from typing import Any4 5from dotenv import load_dotenv6from fastapi import HTTPException, UploadFile7from fastapi.responses import JSONResponse8 9from core.chat.engine import Engine10from core.parser import clean_text11from langfuse.llama_index import LlamaIndexCallbackHandler12from script.document_uploader import Uploader13from script.vector_db import IndexManager14from service.aws_loader import Loader15from service.dto import BotResponseStreaming16from utils.error_handlers import handle_exception17 18load_dotenv()19 20# Configure logging21logging.basicConfig(level=logging.INFO)22 23 24async def data_ingestion(reference, file: UploadFile, lang: str = "en") -> Any:25    try:26        # Assuming you have a Langfuse callback handler27        langfuse_callback_handler = LlamaIndexCallbackHandler()28        langfuse_callback_handler.set_trace_params(29            user_id="admin_book_uploaded",30        )31        32        uploader = Uploader(reference, file, lang)33        nodes_with_metadata, file_stream = await uploader.process_documents()34        35        if isinstance(nodes_with_metadata, JSONResponse):36            return nodes_with_metadata  # Return the error response directly37 38        # Build indexes using IndexManager39        index = IndexManager()40        index.build_indexes(nodes_with_metadata)41        42        # Upload AWS43        file_name = f"{reference['title']}"44        aws_loader = Loader()45 46        aws_loader.upload_to_s3(file_stream, file_name)47 48        return json.dumps(49            {"status": "success", "message": "Vector Index loaded successfully."}50        )51 52    except Exception as e:53        # Log the error54        logging.error("An error occurred in data ingestion: %s", e)55        # Use handle_exception for structured error handling56        return handle_exception(e)57 58async def generate_streaming_completion(user_request, session_id):59    try:60        engine = Engine()61        index_manager = IndexManager()62 63        # Load existing indexes64        index = index_manager.load_existing_indexes()65        # Retrieve the chat engine with the loaded index66        chat_engine = engine.get_chat_engine(index, session_id)67        # Generate completion response68        response = chat_engine.stream_chat(user_request)69 70        completed_response = ""71 72        for gen in response.response_gen:73            completed_response += gen  # Concatenate the new string74            yield BotResponseStreaming(75                content=gen, completed_content=completed_response76            )77 78        nodes = response.source_nodes79        for node in nodes:80            reference = str(clean_text(node.node.get_text()))81            metadata = dict(node.node.metadata)82            score = float(node.score)83            yield BotResponseStreaming(84                completed_content=completed_response,85                reference=reference,86                metadata=metadata,87                score=score,88            )89    except Exception as e:90        yield {"error": str(e)}91 92    except Exception as e:93        # Log the error and raise HTTPException for FastAPI94        logging.error(f"An error occurred in generate text: {e}")95        raise HTTPException(96            status_code=500,97            detail="An internal server error occurred in generate text.",98        ) from e99