Spaces:
Sleeping
Sleeping
| # src/agents/system_instructions_rag.py | |
| from typing import List, Dict, Optional | |
| from src.agents.rag_agent import RAGResponse | |
| from src.utils.logger import logger | |
| from src.agents.rag_agent import RAGAgent | |
| from src.utils.greeting_handler import GreetingHandler | |
| class SystemInstructionsRAGAgent(RAGAgent): | |
| """RAG Agent with enhanced system instructions for specific use cases""" | |
| async def generate_response( | |
| self, | |
| query: str, | |
| conversation_id: Optional[str] = None, | |
| temperature: float = 0.7, | |
| max_tokens: Optional[int] = None, | |
| context_docs: Optional[List[str]] = None | |
| ) -> RAGResponse: | |
| """Generate response with specific handling for introduction and no-context cases""" | |
| try: | |
| # First, check if this is an introduction/welcome message query | |
| is_introduction = ( | |
| "wants support" in query and | |
| "This is Introduction" in query and | |
| ("A new user with name:" in query or "An old user with name:" in query) | |
| ) | |
| if is_introduction: | |
| # Handle introduction message - no context needed | |
| welcome_message = self._handle_contact_query(query) | |
| return RAGResponse( | |
| response=welcome_message, | |
| context_docs=[], | |
| sources=[], | |
| scores=None | |
| ) | |
| # Check if this is a greeting (using the singleton GreetingHandler) | |
| greeting_handler = GreetingHandler() | |
| is_greeting, greeting_response = greeting_handler.is_greeting( | |
| query) | |
| if is_greeting: | |
| logger.info(f"Detected greeting in query: {query}") | |
| return RAGResponse( | |
| response=greeting_response, | |
| context_docs=[], | |
| sources=[], | |
| scores=None | |
| ) | |
| # Get conversation history if conversation_id exists | |
| conversation_history = [] | |
| if conversation_id: | |
| try: | |
| conversation_history = await self.mongodb.get_recent_messages( | |
| conversation_id, | |
| limit=self.conversation_manager.max_messages | |
| ) | |
| # Get relevant history within token limits | |
| conversation_history = self.conversation_manager.get_relevant_history( | |
| messages=conversation_history, | |
| current_query=query | |
| ) | |
| except Exception as e: | |
| logger.warning( | |
| f"Error fetching conversation history: {str(e)}") | |
| # For all other queries, proceed with context-based response | |
| if not context_docs: | |
| context_docs, sources, scores = await self.retrieve_context( | |
| query, | |
| conversation_history=conversation_history | |
| ) | |
| # Check if we have relevant context | |
| has_relevant_context = self._check_context_relevance( | |
| query, context_docs or [] | |
| ) | |
| # If no relevant context found, return the standard message | |
| if not has_relevant_context: | |
| return RAGResponse( | |
| response="Information about this is not available, do you want to inquire about something else?", | |
| context_docs=[], | |
| sources=[], | |
| scores=None | |
| ) | |
| # Generate response using context and conversation history | |
| prompt = self._create_response_prompt( | |
| query=query, | |
| context_docs=context_docs, | |
| conversation_history=conversation_history | |
| ) | |
| response_text = self.llm.generate( | |
| prompt, | |
| temperature=temperature, | |
| max_tokens=max_tokens | |
| ) | |
| # Check if the generated response indicates no information | |
| cleaned_response = self._clean_response(response_text) | |
| if self._is_no_info_response(cleaned_response): | |
| return RAGResponse( | |
| response="Information about this is not available, do you want to inquire about something else?", | |
| context_docs=[], | |
| sources=[], | |
| scores=None | |
| ) | |
| return RAGResponse( | |
| response=cleaned_response, | |
| context_docs=context_docs, | |
| sources=sources, | |
| scores=scores | |
| ) | |
| except Exception as e: | |
| logger.error(f"Error in SystemInstructionsRAGAgent: {str(e)}") | |
| raise | |
| def _create_response_prompt( | |
| self, | |
| query: str, | |
| context_docs: List[str], | |
| conversation_history: Optional[List[Dict]] = None | |
| ) -> str: | |
| """Create prompt for generating response from context and conversation history""" | |
| # Format context documents | |
| formatted_context = '\n\n'.join( | |
| f"Context {i+1}:\n{doc.strip()}" | |
| for i, doc in enumerate(context_docs) | |
| if doc and doc.strip() | |
| ) | |
| # Format conversation history if available | |
| history_context = "" | |
| if conversation_history: | |
| history_messages = [] | |
| # Use last 3 messages for context | |
| for msg in conversation_history[-3:]: | |
| role = msg.get('role', 'unknown') | |
| content = msg.get('content', '') | |
| history_messages.append(f"{role.capitalize()}: {content}") | |
| if history_messages: | |
| history_context = "\nPrevious Conversation:\n" + \ | |
| "\n".join(history_messages) | |
| return f""" | |
| Use the following context and conversation history to provide information about: {query} | |
| Context Information: | |
| {formatted_context} | |
| {history_context} | |
| Instructions: | |
| 1. Use information from both the context and conversation history | |
| 2. If the information is found, provide a direct and concise response | |
| 3. Do not make assumptions or add information not present in the context | |
| 4. Ensure the response is clear and complete based on available information | |
| 5. If you cannot find relevant information about the specific query, | |
| respond exactly with: "Information about this is not available, do you want to inquire about something else?" | |
| Query: {query} | |
| Response:""" | |
| def _is_no_info_response(self, response: str) -> bool: | |
| """Check if the response indicates no information available""" | |
| no_info_indicators = [ | |
| "i do not have", | |
| "i don't have", | |
| "no information", | |
| "not available", | |
| "could not find", | |
| "couldn't find", | |
| "cannot find" | |
| ] | |
| response_lower = response.lower() | |
| return any(indicator in response_lower for indicator in no_info_indicators) | |
| def _check_context_relevance(self, query: str, context_docs: List[str]) -> bool: | |
| """Check if context contains information relevant to the query""" | |
| if not context_docs: | |
| return False | |
| # Extract key terms from query (keeping important words) | |
| query_words = query.lower().split() | |
| stop_words = {'me', 'a', 'about', 'what', 'is', | |
| 'are', 'the', 'in', 'how', 'why', 'when', 'where'} | |
| # Remove only basic stop words, keep important terms like "report", "share", etc. | |
| query_terms = {word for word in query_words if word not in stop_words} | |
| # Add additional relevant terms that might appear in the content | |
| related_terms = { | |
| 'comprehensive', | |
| 'report', | |
| 'overview', | |
| 'summary', | |
| 'details', | |
| 'information' | |
| } | |
| query_terms.update( | |
| word for word in query_words if word in related_terms) | |
| # Check each context document for relevance | |
| for doc in context_docs: | |
| if not doc: | |
| continue | |
| doc_lower = doc.lower() | |
| # Consider document relevant if it contains any query terms | |
| # or if it starts with common report headers | |
| if any(term in doc_lower for term in query_terms) or \ | |
| any(header in doc_lower for header in ['overview', 'comprehensive report', 'summary']): | |
| return True | |
| return False | |
| def _handle_contact_query(self, query: str) -> str: | |
| """Handle queries from /user/contact endpoint""" | |
| try: | |
| name_start = query.find('name: "') + 7 | |
| name_end = query.find('"', name_start) | |
| name = query[name_start:name_end] if name_start > 6 and name_end != -1 else "there" | |
| is_returning = ( | |
| "An old user with name:" in query and | |
| "wants support again" in query | |
| ) | |
| if is_returning: | |
| return f"Welcome back {name}, How can I help you?" | |
| return f"Welcome {name}, How can I help you?" | |
| except Exception as e: | |
| logger.error(f"Error handling contact query: {str(e)}") | |
| return "Welcome, How can I help you?" | |
| def _clean_response(self, response: str) -> str: | |
| """Clean response by removing unwanted phrases""" | |
| if not response: | |
| return response | |
| phrases_to_remove = [ | |
| "Based on the context provided,", | |
| "According to the documents,", | |
| "From the information available,", | |
| "I can tell you that", | |
| "Let me help you with that", | |
| "I understand you're asking about", | |
| "To answer your question,", | |
| "The documents indicate that", | |
| "Based on the available information,", | |
| "As per the provided context,", | |
| "I would be happy to help you with that", | |
| "Let me provide you with information about", | |
| "Here's what I found:", | |
| "Here's the information you requested:", | |
| "According to the provided information,", | |
| "Based on the documents,", | |
| "The information suggests that", | |
| "From what I can see,", | |
| "Let me explain", | |
| "To clarify,", | |
| "It appears that", | |
| "I can see that", | |
| "Sure,", | |
| "Well,", | |
| "Based on the given context,", | |
| "The available information shows that", | |
| "From the context provided,", | |
| "The documentation mentions that", | |
| "According to the context,", | |
| "As shown in the context,", | |
| "I apologize," | |
| ] | |
| cleaned_response = response | |
| for phrase in phrases_to_remove: | |
| cleaned_response = cleaned_response.replace(phrase, "").strip() | |
| cleaned_response = " ".join(cleaned_response.split()) | |
| if not cleaned_response: | |
| return response | |
| if cleaned_response[0].islower(): | |
| cleaned_response = cleaned_response[0].upper( | |
| ) + cleaned_response[1:] | |
| return cleaned_response | |