import chainlit as cl import os import logging from dotenv import load_dotenv import time import asyncio import re import logging # Import llama-index and related libraries from llama_index.core import VectorStoreIndex, StorageContext from llama_index.vector_stores.milvus import MilvusVectorStore from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.core import Settings from llama_index.llms.litellm import LiteLLM from my_config import MY_CONFIG import query_utils # Global variables for LLM and index vector_index = None initialization_complete = False # Create logs directory if it doesn't exist os.makedirs('logs/chainlit', exist_ok=True) logging.basicConfig(level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s', handlers=[ logging.FileHandler('logs/chainlit/chainlit_vector.log', mode='a'), logging.StreamHandler() ], force=True) logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) os.environ['HF_ENDPOINT'] = 'https://hf-mirror.com' # Load environment variables from .env file load_dotenv() def initialize(): """ Initialize LLM and Milvus vector database using llama-index. This function sets up the necessary components for the chat application. """ global vector_index, initialization_complete if initialization_complete: return logger.info("Initializing LLM and vector database...") # raise Exception ("init exception test") # debug try: ## embedding model Settings.embed_model = HuggingFaceEmbedding( model_name = MY_CONFIG.EMBEDDING_MODEL ) logger.info(f"āœ… Using embedding model: {MY_CONFIG.EMBEDDING_MODEL}") # Setup LLM logger.info(f"āœ… Using LLM model : {MY_CONFIG.LLM_MODEL}") Settings.llm = LiteLLM( model=MY_CONFIG.LLM_MODEL, ) # Initialize Milvus vector store vector_store = MilvusVectorStore( uri = MY_CONFIG.MILVUS_URI_VECTOR, dim = MY_CONFIG.EMBEDDING_LENGTH , collection_name = MY_CONFIG.COLLECTION_NAME, overwrite=False # so we load the index from db ) storage_context = StorageContext.from_defaults(vector_store=vector_store) logger.info (f"āœ… Connected to Milvus instance: {MY_CONFIG.MILVUS_URI_VECTOR}") vector_index = VectorStoreIndex.from_vector_store( vector_store=vector_store, storage_context=storage_context) logger.info (f"āœ… Loaded index from vector db: {MY_CONFIG.MILVUS_URI_VECTOR}") logger.info("Successfully initialized LLM and vector database") initialization_complete = True except Exception as e: initialization_complete = False logger.error(f"Error initializing LLM and vector database: {str(e)}") raise (e) # return False ## ------------- def extract_thinking_section(response_text): """ Extract thinking section from LLM response if present. Args: response_text (str): The full response from the LLM Returns: tuple: (thinking_content, cleaned_response) - thinking_content: Content within tags or None if not found - cleaned_response: Response with thinking section removed """ thinking_pattern = r'(.*?)' match = re.search(thinking_pattern, response_text, re.DOTALL) if match: thinking_content = match.group(1).strip() cleaned_response = re.sub(thinking_pattern, '', response_text, flags=re.DOTALL).strip() return thinking_content, cleaned_response else: return None, response_text async def get_llm_response(message): """ Process the user message and get a response from the LLM using Vector RAG with structured prompt """ global vector_index, initialization_complete # Check if LLM and index are initialized if vector_index is None or initialization_complete is None: return "System did not initialize. Please try again later.", 0 start_time = time.time() response_text = '' try: # Step 1: Query preprocessing async with cl.Step(name="Query Preprocessing", type="tool") as step: logger.info("Start query preprocessing step...") step.input = message # Create a query engine from the index query_engine = vector_index.as_query_engine() # Preprocess the query original_message = message message = query_utils.tweak_query(message, MY_CONFIG.LLM_MODEL) step.output = f"Optimized query: {message}" ## --- end: Step 1 --- # Query the index with structured prompting logger.info("Calling LLM with structured prompting...") t1 = time.time() # Get initial vector response vector_response = query_engine.query(message) vector_text = str(vector_response).strip() # Structured prompt structured_prompt = f"""Please provide a comprehensive, well-structured answer using the provided document information. Question: {message} Document Information: {vector_text} Instructions: 1. Provide accurate, factual information based on the documents 2. Structure your response clearly with proper formatting 3. Be comprehensive yet concise 4. Highlight key relationships and important details when relevant 5. Use bullet points or sections when appropriate for clarity Please provide your answer:""" # Use structured prompt for final synthesis response = query_engine.query(structured_prompt) t2 = time.time() if response: response_text = str(response).strip() else: response_text = "No response from LLM." logger.info(f"LLM response received in {(t2 - t1):.2f} seconds:\n{response_text[:200]}") # Step 2: Vector search and retrieval async with cl.Step(name="Document Retrieval", type="retrieval") as step: step.input = message # Show retrieved documents if hasattr(response, 'source_nodes') and response.source_nodes: sources_output = [] for i, node in enumerate(response.source_nodes[:3]): # Show top 3 sources score = node.score if hasattr(node, 'score') else 'N/A' text_preview = node.text[:200] + "..." if len(node.text) > 200 else node.text sources_output.append(f"Source {i+1} (Score: {score}): {text_preview}") step.output = "\n\n".join(sources_output) else: step.output = "No relevant documents found." # Extract thinking section if present thinking_content, cleaned_response = extract_thinking_section(response_text) # print (f"------ Thinking Content:-----\n{thinking_content}\n------") # Debug print # print (f"------ Cleaned Response:-----\n{cleaned_response}\n------") # Debug print # Step 3: Optional Thinking Process if thinking_content: async with cl.Step(name="šŸ’­ Thinking Process", type="run") as step: step.input = "" step.output = thinking_content logger.info(f"Thinking:\n{thinking_content[:200]}...") # Step 4: LLM Answer async with cl.Step(name="Response", type="llm") as step: step.input = f"Query: {message}\nContext: Retrieved from vector database" if cleaned_response: step.output = cleaned_response logger.info(f"Response:\n{cleaned_response[:200]}...") else: step.output = "No response from LLM." logger.info(f"Response:\nNo response from LLM.") except Exception as e: logger.error(f"Error getting LLM response: {str(e)}") response_text = f"Sorry, I encountered an error while processing your request:\n{str(e)}" end_time = time.time() elapsed_time = end_time - start_time return response_text, elapsed_time ## --- end: def get_llm_response(): # ====== CHAINLIT SPECIFIC CODE ====== @cl.set_starters async def set_starters(): starters = [] for prompt in MY_CONFIG.STARTER_PROMPTS: starters.append( cl.Starter( label=prompt.strip(), message=prompt.strip(), ) ) return starters ## --- end: def set_starters(): --- @cl.on_chat_start async def start(): """Initialize the chat session""" # Store initialization state in user session cl.user_session.set("chat_started", True) logger.info("User chat session started") init_error = None try: initialize() # await cl.Message(content="How can I assist you today?").send() except Exception as e: init_error = str(e) error_msg = f"""System Initialization Error The system failed to initialize with the following error: ``` {init_error} ``` Please check your configuration and environment variables.""" await cl.Message(content=error_msg).send() @cl.on_message async def main(message: cl.Message): """Handle incoming messages""" user_message = message.content # Get response from LLM with RAG steps shown FIRST response_text, elapsed_time = await get_llm_response(user_message) # logger.info(f"LLM Response:\n{response_text[:200]}...") # Log first 200 chars thinking_content, cleaned_response = extract_thinking_section(response_text) # Add timing stat to response full_response = cleaned_response + f"\n\nā±ļø *Total time: {elapsed_time:.1f} seconds*" # THEN create a new message for streaming msg = cl.Message(content="") await msg.send() # Stream the response character by character for better UX # This simulates streaming - in a real implementation you'd stream from the LLM for i in range(0, len(full_response), 5): # Stream in chunks of 5 characters await msg.stream_token(full_response[i:i+5]) await asyncio.sleep(0.01) # Small delay for visual effect # Update the final message msg.content = full_response await msg.update() ## ------- if __name__ == '__main__': logger.info("App starting up...") print(f"{'='*60}\n")