Tutorial: Advanced Call Center Call Analysis

This tutorial demonstrates how to build a robust pipeline to process call center audio recordings. By the end, you will have an automated workflow that:

1. Transcribes call audio, identifying different speakers (diarization).
2. Anonymizes sensitive data (such as CPF, email, phone) directly in the transcript.
3. Performs a Structured Analysis to extract quality metrics, sentiment, and other KPIs.
4. Prepares the data to be sent to a database, BI tool, or spreadsheet.

We will combine two powerful SipPulse AI features: Advanced Audio Transcription and Structured Analyses.

Prerequisites

• An active account on the SipPulse AI platform.
• A Structured Analysis already created on the platform to evaluate the calls. (See Step 1).
• Call recordings in a compatible audio format (e.g., .wav, .mp3).
• Basic knowledge of Python or TypeScript (Node.js) to run the scripts.

Step 1: Create the Structured Analysis

Before processing the calls, we need to define which information we want to extract. We do this with a Structured Analysis. Think of it as a form that the AI will fill out for each call.

For a detailed guide, see the Structured Analyses documentation.

1. Go to Structured Analyses and click New Analysis.
2. Give it a clear name, such as call_quality_analysis.
3. In the description, explain the goal: "Extracts quality metrics, customer sentiment, and problem resolution from support call transcripts."
4. Add the fields you want to extract. For example:

• call_summary (Type: Text): "A concise summary of the entire dialogue."
• customer_sentiment (Type: Enum, Options: positive, negative, neutral): "The overall sentiment of the customer during the call."
• problem_resolved (Type: Boolean): "Was the customer's issue resolved? (true/false)"
• agent_performance_score (Type: Number): "Score from 1 to 5 for the agent's performance."

5. Save the analysis and copy its ID. You will need it in the next step.

Step 2: Build the Processing Pipeline

Now, let's create the script that orchestrates the entire process. The flow for each audio file will be:

1. Send for Transcription: Call the STT API with the diarization parameter and a configuration object for anonymize, specifying the entities to redact.
2. Receive Anonymized Transcript: Get the transcript text with sensitive data already masked.
3. Send for Analysis: Use the transcribed text as input for the Structured Analysis we created.
4. Combine Results: Merge the transcript and structured insights into a single object.

Below are complete code examples to automate this flow.

Python

import os
import requests
import json
from typing import List, Dict, Any

# --- Configuration ---
SIPPULSE_API_KEY = os.environ.get("SIPPULSE_API_KEY")
API_BASE_URL = "https://api.sippulse.ai"
# Paste the Structured Analysis ID created in Step 1
ANALYSIS_ID = "YOUR_STRUCTURED_ANALYSIS_ID"
# Define the entities you want to anonymize in the transcript
# See the Anonymization documentation for all available entities
ANONYMIZE_CONFIG = {
  "entities": ["CPF", "EMAIL", "PERSON", "LOCATION", "NUMBER"],
  "sequence": 4 # Anonymize numbers with 4 or more digits
}

def process_call_recording(file_path: str, analysis_id: str) -> Dict[str, Any]:
  """
  Transcribes, anonymizes, and analyzes a single call recording.
  """
  if not SIPPULSE_API_KEY:
    raise ValueError("The SIPPULSE_API_KEY environment variable is not set.")

  headers = {"api-key": SIPPULSE_API_KEY}

  # 1. Transcribe with diarization and anonymization
  print(f"Processing file: {file_path}")
  with open(file_path, "rb") as audio_file:
    files = {"file": (os.path.basename(file_path), audio_file)}
    payload = {
      "model": "pulse-precision",
      "format": "diarization", # Returns text with speaker identification
      "anonymize": json.dumps(ANONYMIZE_CONFIG) # Enables and configures anonymization
    }
    asr_response = requests.post(
      f"{API_BASE_URL}/v1/asr/transcribe",
      headers=headers,
      files=files,
      data=payload
    )
    asr_response.raise_for_status()
    transcription_data = asr_response.json()
    anonymized_text = transcription_data.get("text")

  if not anonymized_text:
    raise Exception("Failed to obtain anonymized transcript text.")

  # 2. Run Structured Analysis
  analysis_response = requests.post(
    f"{API_BASE_URL}/structured-analyses/{analysis_id}/execute",
    headers=headers,
    json={"content": anonymized_text}
  )
  analysis_response.raise_for_status()
  analysis_result = analysis_response.json().get("content")

  # 3. Combine results
  return {
    "filename": os.path.basename(file_path),
    "transcription": transcription_data,
    "quality_analysis": analysis_result
  }

def save_to_database(data: Dict[str, Any]):
  """
  Example function to save results.
  Replace with your integration logic (e.g., save to a database,
  send to a BI API, or add to a Google Sheets spreadsheet).
  """
  print("\n--- Final Result for Storage ---")
  print(json.dumps(data, indent=2, ensure_ascii=False))
  print("--------------------------------\n")


if __name__ == "__main__":
  # Simulate a list of files from a recordings folder
  recordings_path = "./call_recordings" # Create this folder and add your audio files
  if not os.path.exists(recordings_path):
    os.makedirs(recordings_path)
    print(f"Folder '{recordings_path}' created. Add your .wav or .mp3 audio files to it.")
  
  audio_files = [os.path.join(recordings_path, f) for f in os.listdir(recordings_path) if f.endswith(('.wav', '.mp3'))]

  if not audio_files:
    print("No audio files found in the 'call_recordings' folder.")

  for audio_file in audio_files:
    try:
      final_result = process_call_recording(audio_file, ANALYSIS_ID)
      save_to_database(final_result)
    except Exception as e:
      print(f"Error processing file {audio_file}: {e}")

Node.js

import fs from 'fs/promises';
import path from 'path';
import FormData from 'form-data';
import fetch from 'node-fetch'; // `npm install node-fetch`

// --- Configuration ---
const SIPPULSE_API_KEY = process.env.SIPPULSE_API_KEY;
const API_BASE_URL = 'https://api.sippulse.ai';
// Paste the Structured Analysis ID created in Step 1
const ANALYSIS_ID = 'YOUR_STRUCTURED_ANALYSIS_ID';
// Define the entities you want to anonymize in the transcript
// See the Anonymization documentation for all available entities
const ANONYMIZE_CONFIG = {
  entities: ["CPF", "EMAIL", "PERSON", "LOCATION", "NUMBER"],
  sequence: 4, // Anonymize numbers with 4 or more digits
};

async function processCallRecording(filePath: string, analysisId: string): Promise<any> {
  if (!SIPPULSE_API_KEY) {
  throw new Error('The SIPPULSE_API_KEY environment variable is not set.');
  }

  const headers = { 'api-key': SIPPULSE_API_KEY };

  // 1. Transcribe with diarization and anonymization
  console.log(`Processing file: ${filePath}`);
  const fileBuffer = await fs.readFile(filePath);
  const form = new FormData();
  form.append('file', fileBuffer, path.basename(filePath));
  form.append('model', 'pulse-precision');
  form.append('format', 'diarization'); // Returns text with speaker identification
  form.append('anonymize', JSON.stringify(ANONYMIZE_CONFIG)); // Enables and configures anonymization

  const asrResponse = await fetch(`${API_BASE_URL}/v1/asr/transcribe`, {
  method: 'POST',
  headers: { ...headers, ...form.getHeaders() },
  body: form,
  });

  if (!asrResponse.ok) {
  throw new Error(`Transcription error: ${await asrResponse.text()}`);
  }
  const transcriptionData: any = await asrResponse.json();
  const anonymizedText = transcriptionData?.text;

  if (!anonymizedText) {
  throw new Error('Failed to obtain anonymized transcript text.');
  }

  // 2. Run Structured Analysis
  const analysisResponse = await fetch(`${API_BASE_URL}/structured-analyses/${analysisId}/execute`, {
    method: 'POST',
    headers: { ...headers, 'Content-Type': 'application/json' },
    body: JSON.stringify({ content: anonymizedText }),
  });

  if (!analysisResponse.ok) {
    throw new Error(`Structured analysis error: ${await analysisResponse.text()}`);
  }
  const analysisResult = (await analysisResponse.json() as any).content;

  // 3. Combine results
  return {
  filename: path.basename(filePath),
  transcription: transcriptionData,
  quality_analysis: analysisResult,
  };
}

function saveToDatabase(data: any) {
  /**
   * Example function to save results.
   * Replace with your integration logic (e.g., save to a database,
   * send to a BI API, or add to a Google Sheets spreadsheet).
   */
  console.log('\n--- Final Result for Storage ---');
  console.log(JSON.stringify(data, null, 2));
  console.log('--------------------------------\n');
}

(async () => {
  // Simulate a list of files from a recordings folder
  const recordingsPath = './call_recordings'; // Create this folder and add your audio files
  try {
  await fs.access(recordingsPath);
  } catch {
  await fs.mkdir(recordingsPath);
  console.log(`Folder '${recordingsPath}' created. Add your .wav or .mp3 audio files to it.`);
  }

  const files = await fs.readdir(recordingsPath);
  const audioFiles = files
  .filter(f => f.endsWith('.wav') || f.endsWith('.mp3'))
  .map(f => path.join(recordingsPath, f));

  if (audioFiles.length === 0) {
  console.log("No audio files found in the 'call_recordings' folder.");
  return;
  }

  for (const audioFile of audioFiles) {
  try {
    const finalResult = await processCallRecording(audioFile, ANALYSIS_ID);
    saveToDatabase(finalResult);
  } catch (error) {
    console.error(`Error processing file ${audioFile}:`, error);
  }
  }
})();

Step 3: Run and Store the Results

1. Set Up the Environment:

• Create a folder named call_recordings in the same directory as your script.
• Place some call audio files inside it.
• Set the SIPPULSE_API_KEY environment variable with your API key.
• Replace YOUR_STRUCTURED_ANALYSIS_ID in the script with your actual analysis ID.

2. Run the Script:

• For Python: python script_name.py
• For Node.js: node script_name.js

3. Check the Output: The script will iterate over each audio file, process it, and print a final JSON object to the console. This object contains the file name, the full transcript with diarization, and the result of your quality analysis.

The save_to_database function is a placeholder. This is where you would add logic to send the final JSON to your destination, such as:

• Insert into a PostgreSQL or MongoDB table.
• Add a new row to a Google Sheets spreadsheet.
• Send to a BI platform like Power BI or Looker.

Step 4 (Optional): Creating an Endpoint for Continuous Automation

For full automation, instead of running the script manually on a folder, you can wrap the logic in an API endpoint. Your telephony system (or any other system generating the recordings) can then simply send the audio file to this endpoint whenever a new call is finished.

This creates a decoupled and scalable workflow. Below are examples of how to create a simple web server for this purpose using FastAPI (Python) and Express (Node.js).

Production Tip

For production environments, it is recommended that file processing (which may take several seconds) be performed asynchronously in a task queue (such as Celery for Python or BullMQ for Node.js). The endpoint would return an immediate response (202 Accepted) and processing would occur in the background.

Python: FastAPI

# Save as `server.py`.
import os
import shutil
from fastapi import FastAPI, UploadFile, File, HTTPException

# Assuming the Step 2 script was saved as `pipeline.py`
from pipeline import process_call_recording, save_to_database, ANALYSIS_ID

app = FastAPI()

@app.post("/process-recording/")
async def handle_recording_upload(file: UploadFile = File(...)):
  temp_dir = "./temp_recordings"
  os.makedirs(temp_dir, exist_ok=True)
  temp_file_path = os.path.join(temp_dir, file.filename)

  try:
    with open(temp_file_path, "wb") as buffer:
      shutil.copyfileobj(file.file, buffer)

    print(f"Received new file to process: {file.filename}")
    # FastAPI runs sync calls in a thread pool to avoid blocking.
    final_result = process_call_recording(temp_file_path, ANALYSIS_ID)
    save_to_database(final_result)

    return {"filename": file.filename, "status": "processed"}
  except Exception as e:
    print(f"Error processing {file.filename}: {e}")
    raise HTTPException(status_code=500, detail=f"Failed to process file: {str(e)}")
  finally:
    if os.path.exists(temp_file_path):
      os.remove(temp_file_path)

# To run:
# 1. Save the Python code from Step 2 in a file called `pipeline.py`.
# 2. Install FastAPI and Uvicorn: pip install "fastapi[all]"
# 3. Run the server: uvicorn server:app --reload

Node.js: Express

// Save as `server.js`.
import express from 'express';
import multer from 'multer';
import fs from 'fs/promises';
import path from 'path';

// Assuming the Step 2 code was saved as `pipeline.js` and the necessary functions were exported.
import { processCallRecording, saveToDatabase, ANALYSIS_ID } from './pipeline.js';

const app = express();
const port = 3000;

const tempDir = './temp_recordings';
const upload = multer({ dest: tempDir });

app.post('/process-recording', upload.single('file'), async (req, res) => {
  if (!req.file) {
  return res.status(400).send({ message: 'No file uploaded.' });
  }

  const tempFilePath = req.file.path;
  console.log(`Received new file to process: ${req.file.originalname}`);

  try {
  const finalResult = await processCallRecording(tempFilePath, ANALYSIS_ID);
  saveToDatabase(finalResult);

  res.status(200).send({ filename: req.file.originalname, status: 'processed' });
  } catch (error) {
  console.error(`Error processing ${req.file.originalname}:`, error);
  res.status(500).send({ message: 'Failed to process file.', error: error.message });
  } finally {
  // Clean up the temporary file
  await fs.unlink(tempFilePath);
  }
});

app.listen(port, async () => {
  await fs.mkdir(tempDir, { recursive: true });
  console.log(`Processing server listening on port ${port}`);
});

// To run:
// 1. Save the Node.js code from Step 2 in `pipeline.js` and export the necessary functions.
// 2. Install Express and Multer: npm install express multer
// 3. Run the server: node server.js

Conclusion

Congratulations! You have created an end-to-end pipeline that transforms raw call recordings into actionable, structured insights, ensuring privacy with automatic anonymization.

This process, which previously required hours of manual work, can now be executed in minutes, allowing your quality team to focus on using the data to train agents and improve the customer experience.