""" Call Analysis with Quality Parameters Uses AWS Nova AI to analyze calls against business-specific quality parameters """ import boto3 import json import logging from typing import Dict, List, Optional from datetime import datetime import os from quality_parameters_handler import QualityParametersHandler from talk_listen_calculator import calculate_talk_listen_ratio from db_handler import DatabaseHandler from summary_config import load_summary_pipeline_config, resolve_summary_prompt_parts logger = logging.getLogger(__name__) # AWS Bedrock configuration AWS_REGION = os.getenv('AWS_REGION', 'us-east-1') AWS_ACCESS_KEY_ID = os.getenv('AWS_ACCESS_KEY_ID') AWS_SECRET_ACCESS_KEY = os.getenv('AWS_SECRET_ACCESS_KEY') AWS_NOVA_MODEL = os.getenv('AWS_NOVA_MODEL', 'amazon.nova-lite-v1:0') MIN_DURATION_FOR_SCORING_SECONDS = 15 class CallAnalyzer: """Analyze calls using AWS Nova with business-specific quality parameters""" def __init__(self, config): self.config = config self.quality_params_handler = QualityParametersHandler(config) self.db_handler = DatabaseHandler(config) # Initialize AWS Bedrock client self.bedrock_runtime = boto3.client( service_name='bedrock-runtime', region_name=AWS_REGION, aws_access_key_id=AWS_ACCESS_KEY_ID, aws_secret_access_key=AWS_SECRET_ACCESS_KEY ) def format_parameters_for_prompt(self, parameters: List[Dict]) -> str: """Format quality parameters for AI prompt""" formatted = [] for param in parameters: param_str = f""" **Parameter:** {param['parameter_name']} - **Category:** {param['parameter_group']} - **Max Score:** {param['max_score']} - **What to check:** {param['check_description']} - **Description:** {param['detailed_description']} - **Sample utterances:** {param['sample_utterances'] or 'N/A'} - **Auto-detect N/A:** {param['auto_detect_na']} - **Type:** {param['parameter_type']} """ formatted.append(param_str) return '\n'.join(formatted) def _summary_parts_for_bid( self, bid: Optional[str], *, has_quality_parameters: bool = True ) -> tuple: pipeline_cfg = load_summary_pipeline_config(self.db_handler, bid) if bid else None return resolve_summary_prompt_parts( pipeline_cfg, has_quality_parameters=has_quality_parameters ) def build_analysis_prompt(self, transcript: str, parameters: List[Dict], bid: Optional[str] = None) -> str: """Build comprehensive AI prompt for call analysis""" params_formatted = self.format_parameters_for_prompt(parameters) summary_config_block, overall_summary_hint = self._summary_parts_for_bid(bid) if summary_config_block: summary_section = f"\n{summary_config_block}\n\n" else: summary_section = "\n\n" prompt = f"""You are an expert call quality analyst. Analyze the following call transcript against specific quality parameters. CALL TRANSCRIPT: {transcript} QUALITY PARAMETERS TO EVALUATE: {params_formatted} INSTRUCTIONS: For EACH parameter above: 1. **Determine Applicability:** Decide if this parameter applies to this specific call. Be generous - only mark as "not applicable" if the parameter truly cannot be evaluated for this call type. Not-applicable parameters count as 0 points toward the overall quality score (their max_score still counts in the denominator). 2. **Score (if applicable):** If the parameter is applicable, score it as an integer from 0 up to max_score (inclusive). - The score must be on the same scale as max_score (e.g. if max_score is 10, valid scores are 0-10, NOT 0-100). - **DEFAULT: Give FULL max_score unless there are clear problems** - If the agent made partial effort, deduct proportionally on the 0..max_score scale (e.g. 7/10, not 70/10). - Only deduct points for serious, obvious failures or complete absence of the behavior - When in doubt, give the higher score - be generous and supportive 3. **Provide Evidence:** - Exact transcript segment where you detected this (quote the relevant lines) - Approximate speaker turn number (counting from the start of the conversation) - Approximate timestamp in format "MM:SS" 4. **Reasoning:** Brief explanation of why you gave this score or why it's not applicable. {summary_section}CUSTOMER PROFILING (BANT): Also extract sales qualification details for Budget, Authority, Need, and Timeline. - Provide a clear value if mentioned (amount, range, or statement). - If not mentioned, set value to "Not mentioned" and evidence to "N/A". - Include evidence and reasoning for each field. OUTPUT FORMAT (JSON only, no additional text): {{ "parameters": {{ "Parameter Name": {{ "applicable": true/false, "score": , "max_score": , "detected_in_segment": "", "speaker_turn": , "timestamp": "MM:SS", "reasoning": "" }} }}, "customer_profile": {{ "budget": {{ "value": "", "evidence": "", "reasoning": "", "confidence": "high/medium/low" }}, "authority": {{ "value": "", "evidence": "", "reasoning": "", "confidence": "high/medium/low" }}, "need": {{ "value": "", "evidence": "", "reasoning": "", "confidence": "high/medium/low" }}, "timeline": {{ "value": "", "evidence": "", "reasoning": "", "confidence": "high/medium/low" }} }}, "customer_profile_summary": "<1-2 sentence summary of BANT status>", "overall_summary": "{overall_summary_hint}", "call_purpose": "
", "objections_concerns": "", "objection_type": "<2-3 word classification of the objection/concern type, e.g., 'Delivery Issue', 'Food Quality', 'Payment Problem', 'Order Delay', 'Wrong Items', 'Bad Service', 'None' if no objection>", "sentiment": "" }} IMPORTANT: - Return ONLY valid JSON, no markdown formatting - Use exact parameter names as keys in the "parameters" object - For N/A parameters, set "applicable": false and provide a reason - Provide actual transcript quotes for "detected_in_segment" - Be precise with timestamps and speaker turns - For customer_profile fields, always include value, evidence, reasoning, and confidence """ return prompt def build_short_call_prompt(self, transcript: str, bid: Optional[str] = None) -> str: """Build a prompt for short calls where quality scoring is not applicable""" summary_config_block, overall_summary_hint = self._summary_parts_for_bid( bid, has_quality_parameters=False ) summary_section = ( f"\n{summary_config_block}\n\n" if summary_config_block else "\n" ) return f"""Analyze the following short call transcript for sales qualification and summary. CALL TRANSCRIPT: {transcript} {summary_section}OUTPUT FORMAT (JSON only, no additional text): {{ "overall_summary": "{overall_summary_hint}", "call_purpose": "
", "sentiment": "", "customer_profile": {{ "budget": {{ "value": "", "evidence": "", "reasoning": "", "confidence": "high/medium/low" }}, "authority": {{ "value": "", "evidence": "", "reasoning": "", "confidence": "high/medium/low" }}, "need": {{ "value": "", "evidence": "", "reasoning": "", "confidence": "high/medium/low" }}, "timeline": {{ "value": "", "evidence": "", "reasoning": "", "confidence": "high/medium/low" }} }}, "customer_profile_summary": "<1-2 sentence summary of BANT status>" }} IMPORTANT: - Return ONLY valid JSON, no markdown formatting """ def _get_duration_seconds(self, speaker_segments: List[Dict], actual_duration: Optional[float]) -> Optional[float]: """Get duration in seconds from actual duration or speaker segments.""" if actual_duration is not None: try: return float(actual_duration) except (TypeError, ValueError): return None if not speaker_segments: return None try: end_times = [] for seg in speaker_segments: end_val = seg.get('end_time') if end_val is None: end_val = seg.get('end') if end_val is not None: end_times.append(float(end_val)) return max(end_times) if end_times else None except (TypeError, ValueError): return None def _normalize_sentiment(self, sentiment_value: str) -> str: """Normalize sentiment to one of: positive, negative, neutral""" if sentiment_value is None: return 'neutral' if not isinstance(sentiment_value, str): return 'neutral' if not sentiment_value.strip(): return 'neutral' sentiment_lower = sentiment_value.lower().strip() # Extract first word if it's a sentence if ' ' in sentiment_lower: sentiment_lower = sentiment_lower.split()[0] # Map to standard values if 'positive' in sentiment_lower or 'good' in sentiment_lower: return 'positive' elif 'negative' in sentiment_lower or 'bad' in sentiment_lower or 'poor' in sentiment_lower: return 'negative' else: return 'neutral' def call_aws_nova(self, prompt: str) -> Dict: """Call AWS Nova Bedrock API""" try: # Prepare the request body request_body = { "messages": [ { "role": "user", "content": [{"text": prompt}] } ], "inferenceConfig": { "temperature": 0.3, "max_new_tokens": 3000 } } # Call Bedrock using invoke_model (compatible with boto3 1.28.85) logger.info(f"Calling AWS Nova model: {AWS_NOVA_MODEL}") response = self.bedrock_runtime.invoke_model( modelId=AWS_NOVA_MODEL, contentType="application/json", accept="application/json", body=json.dumps(request_body) ) # Parse the response raw_response = response['body'].read().decode('utf-8') response_body = json.loads(raw_response) # Extract response text from the response structure content_list = response_body.get('output', {}).get('message', {}).get('content', []) if content_list and isinstance(content_list, list): response_text = content_list[0].get('text', '').strip() else: raise ValueError("Empty content received from Bedrock response") logger.info(f"Received response from Nova: {response_text[:200]}...") # Parse JSON response import re def extract_json_from_text(text): # Try finding markdown JSON blocks match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', text, re.DOTALL) if match: json_str = match.group(1) else: # Look for first { and last } start_idx = text.find('{') end_idx = text.rfind('}') if start_idx != -1 and end_idx != -1 and end_idx > start_idx: json_str = text[start_idx:end_idx+1] else: json_str = text # Simple fix for common trailing comma issues in LLM outputs json_str = re.sub(r',\s*([\]}])', r'\1', json_str) # Nova sometimes returns speaker_turn as: "speaker_turn": 1, 2, 3 # which is not valid JSON. Treat that as an array of turns. json_str = re.sub( r'("speaker_turn"\s*:\s*)(\d+(?:\s*,\s*\d+)+)', r'\1[\2]', json_str ) json_str = re.sub( r'("speaker_turn"\s*:\s*)(\d+\s*-\s*\d+)', r'\1"\2"', json_str ) return json_str clean_json_str = extract_json_from_text(response_text) try: return json.loads(clean_json_str) except json.JSONDecodeError as nested_e: logger.error(f"Failed to parse cleaned JSON: {nested_e}") logger.error(f"Cleaned string attempts: {clean_json_str}") raise except json.JSONDecodeError as e: logger.error(f"Failed to parse AWS Nova response as JSON: {e}") logger.error(f"Response text: {response_text}") raise except Exception as e: logger.error(f"Error calling AWS Nova: {e}") raise def _classify_objections(self, bid: str, objections_text: str, transcript: str) -> List[Dict]: """Classify objections into predefined categories using AI""" if not objections_text or objections_text.strip() == '': return [] try: # Get objection classifications from database with self.db_handler.get_connection() as conn: cursor = conn.cursor() cursor.execute(''' SELECT id, description FROM objection_classifications WHERE bid = %s ORDER BY id ''', (bid,)) classifications = cursor.fetchall() if not classifications: logger.warning(f"No objection classifications found for BID {bid}") return [] # Build classification prompt classifications_text = "\n".join([ f"{c['id']}. {c['description']}" for c in classifications ]) prompt = f"""You are analyzing customer objections from a call transcript. OBJECTIONS/CONCERNS IDENTIFIED: {objections_text} CALL TRANSCRIPT (for context): {transcript} AVAILABLE OBJECTION CATEGORIES: {classifications_text} TASK: Match each objection/concern to the most appropriate category ID from the list above. An objection can match multiple categories if applicable. OUTPUT FORMAT (JSON only, no additional text): {{ "classified_objections": [ {{ "classification_id": , "objection_text": "", "confidence": "" }} ] }} IMPORTANT: - Return ONLY valid JSON - Only include objections that clearly match a category - If no objections match any category, return empty array - Multiple objections can have the same classification_id """ ai_response = self.call_aws_nova(prompt) return ai_response.get('classified_objections', []) except Exception as e: logger.error(f"Error classifying objections: {e}") return [] def analyze_call(self, bid: str, callid: str, transcript: str, speaker_segments: List[Dict], actual_duration: float = None) -> Dict: """ Analyze a call with quality parameters and talk-to-listen ratio Args: bid: Business ID callid: Call ID transcript: Full call transcript speaker_segments: List of speaker segments with timing info actual_duration: Actual call duration in seconds Returns: Dict containing all analysis results """ try: duration_seconds = self._get_duration_seconds(speaker_segments, actual_duration) is_short_call = duration_seconds is not None and duration_seconds < MIN_DURATION_FOR_SCORING_SECONDS if is_short_call: logger.info( "Call %s duration %.2fs is below %ss; skipping quality scoring.", callid, duration_seconds, MIN_DURATION_FOR_SCORING_SECONDS ) ai_response = self.call_aws_nova(self.build_short_call_prompt(transcript, bid=bid)) ai_response['quality_score_note'] = 'Call duration too low to assign score' ratio_data = calculate_talk_listen_ratio( speaker_segments, actual_duration=duration_seconds ) analytics_data = { 'callid': callid, 'bid': bid, 'summary': ai_response.get('overall_summary', ''), 'call_purpose': ai_response.get('call_purpose', ''), 'objections_concerns': '', 'objection_type': 'None', 'sentiment': self._normalize_sentiment(ai_response.get('sentiment') or 'neutral'), 'quality_score': None, 'parameter_scores': None, 'parameter_detections': None, 'total_possible_score': None, 'parameters_not_applicable': json.dumps(['Call duration too low to assign score']), 'talk_listen_ratio': ratio_data['talk_listen_ratio'], 'agent_talk_time': ratio_data['agent_talk_time'], 'customer_talk_time': ratio_data['customer_talk_time'], 'dead_air_percentage': ratio_data['dead_air_percentage'], 'agent_speak_percentage': ratio_data['agent_speak_percentage'], 'customer_speak_percentage': ratio_data['customer_speak_percentage'], 'talk_listen_assessment': ratio_data['assessment'], 'analysis_model': AWS_NOVA_MODEL, 'raw_response': ai_response } self.db_handler.save_call_analytics(bid, callid, analytics_data) self.db_handler.save_bant_analysis( bid, callid, ai_response.get('customer_profile'), ai_response.get('customer_profile_summary', '') ) return analytics_data # Step 1: Get quality parameters for this business parameters = self.quality_params_handler.get_parameters(bid) if not parameters: logger.warning(f"No quality parameters found for BID {bid}. Using fallback analysis.") return self._fallback_analysis(bid, callid, transcript, speaker_segments, actual_duration) # Step 2: Build AI prompt prompt = self.build_analysis_prompt(transcript, parameters, bid=bid) # Step 3: Call AWS Nova logger.info(f"Analyzing call {callid} for BID {bid} with {len(parameters)} parameters") ai_response = self.call_aws_nova(prompt) # Step 4: Process AI response parameter_scores = {} parameter_detections = {} parameters_not_applicable = [] total_score = 0 total_possible_score = 0 for param in parameters: param_name = param['parameter_name'] param_result = ai_response['parameters'].get(param_name, {}) if param_result.get('applicable', True): # Parameter is applicable score = param_result.get('score', 0) if score is None: score = 0 max_score = param['max_score'] or 0 score = self.db_handler.normalize_parameter_score(score, max_score) parameter_scores[param_name] = { 'score': score, 'max_score': max_score, 'applicable': True } parameter_detections[param_name] = { 'detected': True, 'transcript_segment': param_result.get('detected_in_segment', ''), 'speaker_turn': param_result.get('speaker_turn', 0), 'timestamp': param_result.get('timestamp', '00:00'), 'reasoning': param_result.get('reasoning', '') } total_score += score total_possible_score += max_score else: # N/A counts as 0 points; max_score still counts toward total possible max_score = param['max_score'] or 0 parameter_scores[param_name] = { 'score': 0, 'max_score': max_score, 'applicable': False, 'reason': param_result.get('reasoning', 'Not applicable to this call') } parameters_not_applicable.append(param_name) total_possible_score += max_score # Step 5: Calculate talk-to-listen ratio ratio_data = calculate_talk_listen_ratio( speaker_segments, actual_duration=actual_duration ) # Step 6: Calculate overall quality score (as percentage) computed_qs, computed_tps = self.db_handler.compute_quality_score_from_parameter_scores( parameter_scores ) if computed_qs is not None: quality_score = computed_qs total_possible_score = computed_tps elif total_possible_score > 0: quality_score = round((total_score / total_possible_score) * 100, 2) else: quality_score = 0.0 # Step 7: Classify objections if present objections_text = ai_response.get('objections_concerns', '') classified_objections = [] if objections_text and objections_text.strip(): logger.info(f"Classifying objections for call {callid}") classified_objections = self._classify_objections(bid, objections_text, transcript) logger.info(f"Found {len(classified_objections)} classified objections") # Step 8: Prepare analytics data for database analytics_data = { 'callid': callid, 'bid': bid, 'summary': ai_response.get('overall_summary', ''), 'call_purpose': ai_response.get('call_purpose', ''), 'objections_concerns': objections_text, 'objection_type': ai_response.get('objection_type', 'None'), 'sentiment': self._normalize_sentiment(ai_response.get('sentiment') or 'neutral'), 'quality_score': quality_score, 'parameter_scores': json.dumps(parameter_scores), 'parameter_detections': json.dumps(parameter_detections), 'total_possible_score': total_possible_score, 'parameters_not_applicable': json.dumps(parameters_not_applicable), 'talk_listen_ratio': ratio_data['talk_listen_ratio'], 'agent_talk_time': ratio_data['agent_talk_time'], 'customer_talk_time': ratio_data['customer_talk_time'], 'dead_air_percentage': ratio_data['dead_air_percentage'], 'agent_speak_percentage': ratio_data['agent_speak_percentage'], 'customer_speak_percentage': ratio_data['customer_speak_percentage'], 'talk_listen_assessment': ratio_data['assessment'], 'analysis_model': AWS_NOVA_MODEL, 'raw_response': ai_response, 'classified_objections': classified_objections # Add classified objections } # Step 9: Save to database self.db_handler.save_call_analytics(bid, callid, analytics_data) self.db_handler.save_bant_analysis( bid, callid, ai_response.get('customer_profile'), ai_response.get('customer_profile_summary', '') ) logger.info(f"✓ Call {callid} analyzed: Score {quality_score}/{total_possible_score} (Total: {total_possible_score})") return analytics_data except Exception as e: logger.error(f"Error analyzing call {callid} for BID {bid}: {e}") raise def _fallback_analysis(self, bid: str, callid: str, transcript: str, speaker_segments: List[Dict], actual_duration: float = None) -> Dict: """Fallback analysis when no quality parameters are configured""" logger.info(f"Using fallback analysis for call {callid}") # Just calculate talk-to-listen ratio and basic summary ratio_data = calculate_talk_listen_ratio(speaker_segments, actual_duration=actual_duration) summary_config_block, overall_summary_hint = self._summary_parts_for_bid( bid, has_quality_parameters=False ) summary_section = ( f"\n{summary_config_block}\n\n" if summary_config_block else "" ) summary_line = ( "- overall_summary: (follow SUMMARY CONFIG above)" if summary_config_block else "- overall_summary: 5-6 sentence summary" ) # Use basic Nova analysis without parameters basic_prompt = f"""Analyze this call transcript and provide: TRANSCRIPT: {transcript} {summary_section}Provide a JSON response with: {summary_line} - call_purpose: main reason for call - objections_concerns: any issues raised - objection_type: 2-3 word classification (e.g., 'Delivery Issue', 'Food Quality', 'None') - sentiment: positive/negative/neutral - customer_profile: object with budget, authority, need, timeline (value/evidence/reasoning/confidence) - customer_profile_summary: 1-2 sentence summary of BANT status Return ONLY valid JSON.""" try: ai_response = self.call_aws_nova(basic_prompt) except: ai_response = { 'overall_summary': 'Call analysis unavailable', 'call_purpose': 'Unknown', 'objections_concerns': '', 'sentiment': 'neutral' } analytics_data = { 'callid': callid, 'bid': bid, 'summary': ai_response.get('overall_summary', ''), 'call_purpose': ai_response.get('call_purpose', ''), 'objections_concerns': ai_response.get('objections_concerns', ''), 'objection_type': ai_response.get('objection_type', 'None'), 'sentiment': self._normalize_sentiment(ai_response.get('sentiment', 'neutral')), 'quality_score': 50, # Default score 'parameter_scores': None, 'parameter_detections': None, 'total_possible_score': None, 'parameters_not_applicable': None, 'talk_listen_ratio': ratio_data['talk_listen_ratio'], 'agent_talk_time': ratio_data['agent_talk_time'], 'customer_talk_time': ratio_data['customer_talk_time'], 'dead_air_percentage': ratio_data['dead_air_percentage'], 'agent_speak_percentage': ratio_data['agent_speak_percentage'], 'customer_speak_percentage': ratio_data['customer_speak_percentage'], 'talk_listen_assessment': ratio_data['assessment'], 'analysis_model': AWS_NOVA_MODEL, 'raw_response': ai_response } self.db_handler.save_call_analytics(bid, callid, analytics_data) self.db_handler.save_bant_analysis( bid, callid, ai_response.get('customer_profile'), ai_response.get('customer_profile_summary', '') ) return analytics_data def batch_analyze_calls(bid: str, config: Dict, limit: int = 10): """Batch analyze calls that need processing""" analyzer = CallAnalyzer(config) db_handler = DatabaseHandler(config) # Get calls that need analysis calls = db_handler.get_calls_for_analysis(bid, limit) logger.info(f"Found {len(calls)} calls to analyze for BID {bid}") results = { 'success': [], 'failed': [] } for call in calls: try: # Get transcript and speaker segments call_details = db_handler.get_raw_call_details(bid, call['callid']) if not call_details or not call_details.get('transcripts'): logger.warning(f"No transcript found for call {call['callid']}, skipping") continue transcript = call_details['transcripts'] speaker_segments = call_details.get('speaker_segments', []) if speaker_segments and isinstance(speaker_segments, str): try: speaker_segments = json.loads(speaker_segments) except (json.JSONDecodeError, TypeError): speaker_segments = [] actual_duration = call_details.get('duration') # Analyze the call result = analyzer.analyze_call(bid, call['callid'], transcript, speaker_segments, actual_duration) results['success'].append(call['callid']) except Exception as e: logger.error(f"Failed to analyze call {call['callid']}: {e}") results['failed'].append({'callid': call['callid'], 'error': str(e)}) return results