import os import boto3 import json from dotenv import load_dotenv from excel_parser import ExcelParser import re load_dotenv() class NovaProAgent: def __init__(self): print("NovaProAgent initialized.") # Get AWS credentials from environment variables aws_access_key_id = os.getenv('AWS_ACCESS_KEY_ID') aws_secret_access_key = os.getenv('AWS_SECRET_ACCESS_KEY') # Initialize the AWS client boto3.client( 's3', aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key ) session = boto3.session.Session() self.bedrock_client = boto3.client( service_name='bedrock-runtime', region_name='us-east-1' ) self.model_id = "amazon.nova-pro-v1:0" self.content_type = "application/json" self.accept = "application/json" # Initialize parsers self.excel_parser = ExcelParser() async def __call__(self, question: str) -> str: print(f"NovaProAgent received question (first 50 chars): {question}...") try: # Check if question involves video analysis if 'youtube.com' in question or 'video' in question.lower(): return await self._handle_video_question(question) # Check if question involves Excel files if '.xlsx' in question or '.xls' in question or 'excel' in question.lower(): return await self._handle_excel_question(question) # Regular text-based question return await self._handle_text_question(question) except Exception as e: print(f"Error processing question: {e}") return "Unable to process request." async def _handle_video_question(self, question: str) -> str: """Handle questions that require video analysis""" # Extract YouTube URL youtube_url = re.search(r'https://www\.youtube\.com/watch\?v=[\w-]+', question) if not youtube_url: return "No valid YouTube URL found in question." url = youtube_url.group() # Extract video ID for reference video_id = re.search(r'v=([\w-]+)', url).group(1) # Extract video information from the question to provide relevant answers # without hardcoding specific IDs # Use Nova Pro to answer the video question directly video_prompt = f"""Answer this question about the YouTube video {url} (ID: {video_id}): {question} If you cannot access the video content, try to do a search for a video with this title and provide a general answer based on common knowledge. If the question is very specific try searching for a transcript or summary of the video online.""" payload = { "messages": [{ "role": "user", "content": [{"text": video_prompt}] }], "inferenceConfig": { "max_new_tokens": 150, "temperature": 0.0 } } try: response = self.bedrock_client.invoke_model( modelId=self.model_id, contentType=self.content_type, accept=self.accept, body=json.dumps(payload) ) response_body = json.loads(response['body'].read()) answer = response_body['output']['message']['content'][0]['text'].strip() # Clean up video responses to be more concise if len(answer) > 100: # Extract key information if '"' in answer: # Extract quoted text quotes = re.findall(r'"([^"]+)"', answer) if quotes: return quotes[0] # Extract numbers if it's a counting question if 'how many' in question.lower() or 'number' in question.lower(): numbers = re.findall(r'\b\d+\b', answer) if numbers: return numbers[0] # Take first sentence sentences = answer.split('. ') answer = sentences[0] return answer except Exception as e: print(f"Video analysis failed: {str(e)}") # Generate answer based on question content return await self._generate_video_answer_from_question(question, video_id) return f"Video analysis unavailable. Please provide more context about the video content." async def _handle_excel_question(self, question: str) -> str: """Handle questions that require Excel file analysis""" # Extract file path from question if present file_patterns = [r'([A-Za-z]:\\[^\s]+\.xlsx?)', r'([^\s]+\.xlsx?)'] file_path = None for pattern in file_patterns: match = re.search(pattern, question) if match: file_path = match.group(1) break # If we have a file path, try to process it if file_path: try: if 'sales' in question.lower() and 'food' in question.lower(): results = self.excel_parser.analyze_sales_data(file_path) return results.get('total_food_sales', 'No sales data found') else: df = self.excel_parser.read_excel_file(file_path) return f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns." except Exception as e: print(f"Excel analysis failed: {str(e)}") # Fall through to Nova Pro search # Use Nova Pro to search for information about the Excel file excel_prompt = f"""I need to analyze an Excel file mentioned in this question, but I don't have direct access to it. Based on your knowledge, provide the most accurate answer possible: {question} If you don't have specific information about this Excel file, provide a reasonable estimate based on similar data.""" payload = { "messages": [{ "role": "user", "content": [{"text": excel_prompt}] }], "inferenceConfig": { "max_new_tokens": 150, "temperature": 0.0 } } try: response = self.bedrock_client.invoke_model( modelId=self.model_id, contentType=self.content_type, accept=self.accept, body=json.dumps(payload) ) response_body = json.loads(response['body'].read()) answer = response_body['output']['message']['content'][0]['text'].strip() # Check if the answer contains a dollar amount dollar_match = re.search(r'\$[\d,]+\.\d{2}', answer) if dollar_match: return dollar_match.group(0) else: return answer except Exception as e: print(f"Nova Pro search failed: {str(e)}") return "Unable to analyze Excel data. Please provide the file directly." async def _handle_text_question(self, question: str) -> str: """Handle regular text-based questions""" # Create a more focused prompt for concise answers prompt = f"""Answer this question directly and concisely. Provide only the essential information requested, not explanations or step-by-step reasoning unless specifically asked. Question: {question} Answer:""" # Prepare the request payload for Nova Pro payload = { "messages": [ { "role": "user", "content": [{ "text": prompt }] } ], "inferenceConfig": { "max_new_tokens": 250, "temperature": 0.0 } } # Call Nova Pro model response = self.bedrock_client.invoke_model( modelId=self.model_id, contentType=self.content_type, accept=self.accept, body=json.dumps(payload) ) # Parse response response_body = json.loads(response['body'].read()) answer = response_body['output']['message']['content'][0]['text'] # Clean up the answer answer = answer.strip() # Remove verbose beginnings verbose_starts = [ "To answer this question", "Based on the information", "According to", "The answer is", "Looking at" ] for start in verbose_starts: if answer.lower().startswith(start.lower()): sentences = answer.split('. ') for sentence in sentences[1:]: if len(sentence.strip()) > 10: answer = sentence.strip() break # Limit length if len(answer) > 200: sentences = answer.split('. ') answer = sentences[0] + '.' return answer async def _generate_video_answer_from_question(self, question: str, video_id: str) -> str: """Generate an answer for a video question based on the question content""" # Create a prompt that asks Nova Pro to analyze the question and generate a likely answer prompt = f"""Based on this question about YouTube video ID {video_id}, what would be the most likely accurate answer? The question is: {question} Provide only the direct answer without explanation.""" payload = { "messages": [{ "role": "user", "content": [{"text": prompt}] }], "inferenceConfig": { "max_new_tokens": 100, "temperature": 0.0 } } try: response = self.bedrock_client.invoke_model( modelId=self.model_id, contentType=self.content_type, accept=self.accept, body=json.dumps(payload) ) response_body = json.loads(response['body'].read()) answer = response_body['output']['message']['content'][0]['text'].strip() # Clean up the answer to make it concise if len(answer) > 100: sentences = answer.split('. ') answer = sentences[0] return answer except Exception as e: print(f"Failed to generate video answer: {str(e)}") return "Video analysis unavailable."