Call Analytics Pipeline

1
import os
2
import json
3
import hashlib
4
from pathlib import Path
5
from datetime import datetime
6
from typing import List, Dict, Optional
7
from pydub import AudioSegment
8
from sarvamai import SarvamAI
9
import textwrap
10
11
OUTPUT_DIR = "outputs"
12
Path(OUTPUT_DIR).mkdir(exist_ok=True)

1
def split_audio(audio_path: str, chunk_duration_ms: int = 60 * 60 * 1000) -> List[AudioSegment]:
2
    audio = AudioSegment.from_file(audio_path)
3
    return [audio[i:i + chunk_duration_ms] for i in range(0, len(audio), chunk_duration_ms)] if len(audio) > chunk_duration_ms else [audio]

1
class CallAnalytics:
2
    def __init__(self, client):
3
        self.client = client
4
        self.transcriptions = {}  

1
def process_audio_files(self, audio_paths: List[str]) -> Dict[str, str]:
2
    if not audio_paths:
3
        print("No audio files provided")
4
        return {}
5
6
    print(f"Processing {len(audio_paths)} audio files...")
7
8
    try:
9
        job = self.client.speech_to_text_translate_job.create_job(
10
            model="saaras:v3",
11
            mode="translate",
12
            with_diarization=True,
13
        )
14
15
        job.upload_files(file_paths=audio_paths, timeout=300)
16
        job.start()
17
18
        print("Waiting for transcription to complete...")
19
        job.wait_until_complete()
20
21
        if job.is_failed():
22
            print("Transcription failed!")
23
            return {}
24
25
        output_dir = Path(f"{OUTPUT_DIR}/transcriptions_{job.job_id}")
26
        output_dir.mkdir(parents=True, exist_ok=True)
27
        job.download_outputs(output_dir=str(output_dir))
28
        json_files = list(output_dir.glob("*.json"))
29
        if not json_files:
30
          raise FileNotFoundError(f"No .json transcription files found in {output_dir}.")
31
32
        transcriptions = self._parse_transcriptions(output_dir)
33
        self.transcriptions.update(transcriptions)
34
35
        print(f"Successfully transcribed {len(transcriptions)} files!")
36
37
        for file_name, data in transcriptions.items():
38
            self.analyze_transcription(data["conversation_path"], output_dir, file_name)
39
40
        return transcriptions
41
42
    except Exception as e:
43
        print(f"Error processing audio files: {e}")
44
        return {}

1
def _parse_transcriptions(self, output_dir: Path) -> Dict[str, dict]:
2
    transcriptions = {}
3
    for json_file in output_dir.glob("*.json"):
4
        try:
5
            with open(json_file, "r", encoding="utf-8") as f:
6
                data = json.load(f)
7
            diarized = data.get("diarized_transcript", {}).get("entries")
8
            speaker_times = {}
9
            lines = []
10
            if diarized:
11
                for entry in diarized:
12
                    speaker = entry["speaker_id"]
13
                    text = entry["transcript"]
14
                    lines.append(f"{speaker}: {text}")
15
                    start = entry.get("start_time_seconds")
16
                    end = entry.get("end_time_seconds")
17
                    if start is not None and end is not None:
18
                        duration = end - start
19
                        speaker_times[speaker] = speaker_times.get(speaker, 0.0) + duration
20
            else:
21
                lines = [f"UNKNOWN: {data.get('transcript', '')}"]
22
            
23
            conversation_text = "\n".join(lines)
24
            txt_path = output_dir / f"{json_file.stem}_conversation.txt"
25
            with open(txt_path, "w", encoding="utf-8") as f:
26
                f.write(conversation_text)
27
            
28
            timing_path = None
29
            if speaker_times:
30
                timing_path = output_dir / f"{json_file.stem}_timing.json"
31
                with open(timing_path, "w", encoding="utf-8") as f:
32
                    json.dump(speaker_times, f, indent=2)
33
            transcriptions[json_file.stem] = {
34
                "entries": diarized or [],
35
                "conversation_path": str(txt_path),
36
                "timing_path": str(timing_path) if timing_path else None
37
            }
38
        except Exception as e:
39
            print(f"Error parsing {json_file}: {e}")
40
    return transcriptions

1
ANALYSIS_PROMPT_TEMPLATE = """
2
Analyze this call transcription thoroughly from start to finish.
3
4
TRANSCRIPTION:
5
{transcription}
6
7
Please answer the following:
8
9
1. Identify which speaker is the **customer** and which one is the **agent**.
10
2. Determine if the customer is a **new/potential customer** or an **existing customer**.
11
3. What **problem, query, or doubt** did the customer raise at the beginning?
12
4. What **services/products** was the customer inquiring about or facing issues with?
13
5. How did the agent respond to and resolve the issue throughout the call?
14
6. Was the **customer satisfied** at the end of the call?
15
7. Did the customer express any **emotions or sentiments** (positive, negative, or neutral)?
16
8. Were there any mentions of **competitors**, or any opportunities for **upselling or cross-selling**?
17
9. Summarize the **resolution** and whether it was successful.
18
19
Provide your answer in a clear, structured format with section headings and bullet points.
20
"""

1
def analyze_transcription(self, conversation_path: str, output_dir: Path, file_name: str) -> Dict:
2
    try:
3
        with open(conversation_path, "r", encoding="utf-8") as f:
4
            transcription = f.read()
5
        analysis_prompt = textwrap.dedent(ANALYSIS_PROMPT_TEMPLATE.format(transcription=transcription))
6
        messages = [
7
            {"role": "system", "content": "You are a call analytics expert working for a company's support operations team. Your job is to understand customer calls end-to-end and provide structured insights to improve customer experience and agent effectiveness."},
8
            {"role": "user", "content": analysis_prompt},
9
        ]
10
        response = self.client.chat.completions(model="sarvam-105b", messages=messages)
11
        analysis = response.choices[0].message.content
12
        analysis_path = output_dir / f"{file_name}_analysis.txt"
13
        with open(analysis_path, "w", encoding="utf-8") as f:
14
            f.write(analysis.strip())
15
        print(f"Analysis saved to {analysis_path}")
16
        return {"file_name": file_name, "analysis_path": str(analysis_path)}
17
    except Exception as e:
18
        error_msg = f"Error analyzing transcription: {str(e)}"
19
        print(error_msg)
20
        return {"file_name": file_name, "error": error_msg, "timestamp": datetime.now().isoformat()}

1
def answer_question(self, question: str, output_dir: Optional[Path] = None) -> None:
2
    for file_name, data in self.transcriptions.items():
3
        try:
4
            with open(data["conversation_path"], "r", encoding="utf-8") as f:
5
                transcription = f.read()
6
            prompt = f"Based on this call transcription, answer the question below:\n\nTRANSCRIPTION:\n{transcription}\n\nQUESTION: {question}"
7
            messages = [
8
                {"role": "system", "content": ""},
9
                {"role": "user", "content": prompt},
10
            ]
11
            response = self.client.chat.completions(model="sarvam-105b", messages=messages)
12
            answer = response.choices[0].message.content
13
            q_hash = hashlib.sha1(question.encode()).hexdigest()[:6]
14
            path = Path(data["conversation_path"]).parent / f"{file_name}_question_{q_hash}.txt"
15
            with open(path, "w", encoding="utf-8") as f:
16
                f.write(f"Question: {question}\n\nAnswer:\n{answer}")
17
            print(f"Answer saved to {path}")
18
        except Exception as e:
19
            print(f"Error answering question for {file_name}: {e}")

1
SUMMARY_PROMPT_TEMPLATE = """
2
Based on this call analysis, summarize each of the following in 2–3 words:
3
4
{analysis_text}
5
6
1. Customer & Agent
7
2. Customer Type
8
3. Main Issue
9
4. Service Discussed
10
5. Agent's Response
11
6. Customer Satisfaction
12
7. Sentiment
13
8. Competitor or Upsell
14
9. Resolution
15
"""

1
def get_summary(self, output_dir: Optional[Path] = None) -> None:
2
    output_dir = output_dir or Path(OUTPUT_DIR)
3
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
4
    summary_path = output_dir / f"summary_{timestamp}.txt"
5
    try:
6
        with open(summary_path, "w", encoding="utf-8") as f:
7
            f.write("CALL ANALYTICS SUMMARY REPORT\n")
8
            f.write("=" * 60 + "\n")
9
            f.write(f"Generated: {datetime.now()}\n")
10
            f.write(f"Total Calls: {len(self.transcriptions)}\n")
11
            f.write("=" * 60 + "\n\n")
12
            for file_name, data in self.transcriptions.items():
13
                analysis_file = Path(data["conversation_path"]).parent / f"{file_name}_analysis.txt"
14
                if not analysis_file.exists():
15
                    print(f"Analysis file not found for {file_name}, skipping.")
16
                    continue
17
                with open(analysis_file, "r", encoding="utf-8") as af:
18
                    analysis_text = af.read()
19
                summary_prompt = textwrap.dedent(SUMMARY_PROMPT_TEMPLATE.format(analysis_text=analysis_text))
20
                messages = [
21
                    {"role": "system", "content": "You are a call analytics summarizing expert. Provide concise and clear answers to each point "},
22
                    {"role": "user", "content": summary_prompt},
23
                ]
24
                response = self.client.chat.completions(model="sarvam-105b", messages=messages)
25
                concise_summary = response.choices[0].message.content.strip()
26
                f.write(f"Call File: {file_name}\n")
27
                f.write("-" * 30 + "\n")
28
                f.write(f"{concise_summary}\n\n")
29
        print(f"Summary saved to {summary_path}")
30
    except Exception as e:
31
        print(f"Error writing summary: {e}")

1
client = SarvamAI(api_subscription_key="YOUR_SARVAM_API_KEY")
2
analytics = CallAnalytics(client=client)
3
4
audio_path = "/path/to/your/audio/file.mp3"  
5
analytics.process_audio_files([audio_path]) 
6
analytics.answer_question("Add your question here")
7
analytics.get_summary()