AI call monitoring transcription for insurance contact centers — a step-by-step build guide AI call monitoring transcription for insurance contact centers — a step-by-step build guide

I’ve reviewed a dozen “drop-in” AI monitoring tools that claim to transcribe and score calls. In every case, the real bottleneck isn’t the API—it’s the data pipeline, the compliance model, and the reconciliation step where finance kills the budget, and this guide is the one i give my own engineering team when they’re asked to. It’s opinionated, it’s brutal on scope, and it ends with a 45-second compliance checklist you can hand to Legal on Friday.

Who this is for VP of Engineering who needs to ship a working prototype before the next board meeting.

CTO of an MGA who wants to stop paying $1.2 million a year in third-party call auditors. Claims Ops lead who has to prove call quality to the state DOI in an upcoming exam.

• If you’re a data scientist who just wants to train the next BERT model, go elsewhere. We’re building a system that answers: “Did Agent Smith mention the deductible? Did they log it in Salesforce? If not, how much did that cost us?”
• What you’ll spend Category
• Low end High end

Notes Cloud compute (30 day pilot)

≈ $8 k ≈ $25 k

Use DynamoDB for fast lookup by call_id. Table schema: PK: call_id (S)

SK: transcript (S) GSI1: agent_id (S) → sort by timestamp

1. GSI2: policy_number (S) Item size: ~2 KB per call.

2. Write capacity: 5 RCU/WCU per call → 25 RCU/WCU total. With on-demand DynamoDB it’s ~$15/month. Phase 2 — compliance and privacy gating

3. Step 5: Apply redaction rules Before any downstream processing, run a regex and PII detector:

SSN: \b\d{3}-\d{2}-\d{4}\b

Credit card: Luhn-valid 16-digit Policy number: your specific pattern (e.g., [A-Z]{2}\d{8})

Agent PII: names, phone, email Use Presidio from Microsoft. It’s open-source, MIT license, and you can self-host on ECS.

<call_id>_20240514T162231Z.wav
<call_id>_20240514T162231Z.json  // call metadata

Store the anonymized text back in DynamoDB under a new attribute text_anonymized. Keep the original in a separate S3 bucket with bucket policy that denies all reads except to Legal in case of litigation hold. Step 6: Legal hold and retention policy

• Define a lifecycle rule on the raw bucket:
• • Day 0–7: Standard IA Day 8–90: Glacier Instant Retrieval
  • Day 91+: Glacier Deep Archive Set a bucket policy that denies deletion for legal-hold tag. Use AWS S3 Object Lock in compliance mode. Cost: $0.023/GB/month in Glacier Deep Archive.

  Step 7: PCI and PII attestation Create a CloudWatch metric filter that counts every text_anonymized record that still contains a credit card pattern. If the count ever goes above zero, trigger an SNS alert to #ai-call-audit and auto-delete the record.

  Write the CloudFormation snippet: You now have a system that cannot legally leak card numbers.

  Phase 3 — scoring engine: from transcript to KPI Step 8: Define the scoring rubric

  Pick 5–7 rules that actually map to dollars: Compliance: Agent must say “deductible” in the first 30 seconds of a P&C claim call.

  • Upsell: Agent must offer umbrella coverage to homeowners with assets >$1 M. Accuracy: Policy number spoken by agent must match the one in Salesforce within 3 edit distance.
  • Empathy: Use a sentiment classifier fine-tuned on insurance calls (score >0.7). Resolution: Call must end with an open task in Salesforce.
  • Each rule is a Python function that takes the transcript and returns True/False. Store the rubric in a DynamoDB table scoring_rules so non-engineers can tweak it. Step 9: Run scoring in batch or streaming

  Option A — batch: run every night at 02:00 UTC on yesterday’s calls. Step Functions workflow:

  import boto3, os, uuid
  
  transcribe = boto3.client('transcribe')
  s3 = boto3.client('s3')
  
  def lambda_handler(event, context):
      bucket = event['Records'][0]['s3']['bucket']['name']
      key = event['Records'][0]['s3']['object']['key']
      call_id = key.split('_')[0]
  
      job_name = f'transcribe-{call_id}-{uuid.uuid4().hex[:8]}'
      output_uri = f's3://transcript-bucket/{call_id}/'
  
      transcribe.start_transcription_job(
          TranscriptionJobName=job_name,
          Media={'MediaFileUri': f's3://{bucket}/{key}'},
          MediaFormat='wav',
          LanguageCode='en-US',
          OutputBucketName='transcript-bucket',
          OutputKey=call_id + '/',
          Settings={
              'ShowSpeakerLabels': True,
              'MaxSpeakerLabels': 2,
              'ChannelIdentification': False
          }
      )
      return {'statusCode': 202}
  

  List all call_id from DynamoDB with status=transcribed. Run parallel Lambda workers (300 MB, 1024 MB max).

  Write results to call_scores table. Option B — streaming: use Kinesis Data Streams + Lambda.

  def lambda_handler(event, context):
      job_name = event['detail']['TranscriptionJobName']
      call_id = job_name.replace('transcribe-','').split('-')[0]
      bucket = 'transcript-bucket'
      key = f'{call_id}/transcript.json'
      s3.download_file(bucket, key, '/tmp/transcript.json')
      # ---> next step
  

  Resource estimate: 5 k calls/day → 58 calls/sec peak. Kinesis shard limit is 1 MB/sec → 1 shard is enough. Lambda concurrency: 100 → cost ~$30/month. Step 10: Expose the KPI in a dashboard

  Use Amazon QuickSight. Connect to call_scores table. Build a single bar chart: X-axis: hour of day

  Y-axis: % of calls that violated at least one rule Color: red for >15%, green for ≤15%

  pip install pyannote.audio transformers
  

  Set a CloudWatch alarm that pages the VP of Ops when the 24-hour average crosses 15%. Phase 4 — integration with CRM and financials

  from pyannote.audio import Pipeline
  
  pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization-3.1")
  diarization = pipeline("/tmp/audio.wav")
  
  with open('/tmp/transcript.json') as f:
      transcript = json.load(f)
  
  for turn, _, speaker in diarization.itertracks(yield_label=True):
      for seg in transcript['results']['segments']:
          if seg['start'] >= turn.start and seg['end'] <= turn.end:
              seg['speaker'] = speaker
  

  Step 11: Salesforce write-back Use the Salesforce REST API. Create a custom object Call_Compliance__c with fields:

  Call_ID__c (text) Agent_ID__c (text)

  Score__c (number, 0–100) Rule_Violations__c (text, comma-separated)

  • CreatedDate (auto) Use Named Credential + JWT bearer token. Lambda code:
  • Resource estimate: 5 k calls/day → 100 writes/sec peak. Salesforce bulk API limit is 200 rows/batch → 25 batches/day. Cost: $0.0005/batch → $12.50/month. Step 12: Tie to financial impact
  • Create a lookup table in Redshift (or PostgreSQL) called call_impact: violation_type (text)
  • avg_fine_usd (decimal) avg_customer_churn_rate (decimal)
  • avg_upsell_lost_usd (decimal) Join call_scores with call_impact to produce a daily report:

  CFO now has a dollar figure they can defend in the next board meeting. Phase 5 — governance, audit, and sunset

  Step 13: Build the compliance artifact Create a single JSON file that Legal can hand to the DOI examiner:

  system_diagram.pdf (draw.io exported) data_flow.md (mermaid sequence)

  pii_redaction_rules.json (Presidio configuration) retention_policy.yaml (CloudFormation lifecycle)

• scoring_rubric.json (rules + weights) Store it in a separate S3 bucket with bucket policy that denies deletion for 7 years.
• Step 14: Kill switch Write a Lambda that can disable the entire pipeline in <2 minutes. Role needs lambda:InvokeFunction on every micro-service. Tag every resource with Environment=prod and Owner=ai-call-audit. Use AWS Systems Manager Automation to run the kill switch playbook.
• Step 15: Measure ROI Run a 30-day A/B: 10% of calls go through human auditors, the rest through AI. Track:

Compliance fines per 1 k calls Customer NPS delta

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

results = analyzer.analyze(text=transcript_text, language='en')
anonymized = anonymizer.anonymize(text=transcript_text, analyzer_results=results)

Agent coaching hours saved Typical result after 60 days:

Compliance fines ↓ 38% NPS ↑ 7 points

Agent coaching hours ↓ 19% If the fines saved pay for the system inside 6 months, green-light the full rollout. If not, delete the entire stack and move on.

• Week 10 checklist — hand to Legal on Friday Bucket policy for raw audio: deny delete, enable object lock.
• Presidio PII detector: tested on 100 sample calls, no false negatives. CloudWatch metric filter PIILeakCount alarms to #ai-call-audit.

Scoring rubric: published in Confluence, version-controlled in Git. Salesforce custom object Call_Compliance__c has field-level security set to Read-Only for all profiles except Compliance.

Redshift view call_impact_daily is shared with Finance, refresh 06:00 UTC daily. Kill switch Lambda tested: disable pipeline, verify all writes stop within 120 seconds.

If any box isn’t ticked, you don’t ship to production. Common failure modes and how to avoid them

Speaker diarization accuracy below 85% Problem: Agents talk over customers, background noise, whispering.

Resources:
  AnomalyAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      MetricName: PIILeakCount
      Namespace: AI/CallAudit
      Statistic: Sum
      Period: 300
      EvaluationPeriods: 1
      Threshold: 1
      ComparisonOperator: GreaterThanOrEqualToThreshold
      Dimensions:
        - Name: Pipeline
          Value: prod
      AlarmActions:
        - !Ref AnomalyTopic
      TreatMissingData: notBreaching

Fix: Tune pyannote threshold to 0.75. Add a post-processing step: if two adjacent segments have the same speaker label, merge them. If the call is <2 minutes, skip diarization and assign everything to “agent”. Whisper hallucinates policy numbers

Problem: It invents a policy number that doesn’t exist. Fix: Run a Levenshtein distance check against the CRM token. If distance >3, mark the call as “policy mismatch” and route to human auditor.

Compliance team rejects the transcript Problem: Legal says the transcript is discoverable in litigation.

Fix: Before any scoring, run a second PII pass with strict mode. Any transcript that still contains a credit card, SSN, or policy number is immediately moved to a separate legal-hold bucket and excluded from scoring. When to walk away

1. If you tick any of these boxes, stop the project: Your call center platform cannot forward audio to S3 in real time.
2. Your CRM has no REST API or you cannot add a custom object. Legal will not sign off on a 90-day retention policy.
3. You cannot allocate 3 engineering weeks + 2 legal weeks. The sunk cost fallacy is the single biggest killer of insurtech pilots. If the prerequisites aren’t met, cancel the budget and tell the CFO you’ve saved $100 k.
4. Was this article helpful? Comments.

import json, os
from scoring_engine import score_call

def lambda_handler(event, context):
    for record in event['Records']:
        call_id = record['dynamodb']['NewImage']['call_id']['S']
        transcript = dynamodb.get_item(TableName='calls', Key={'call_id': {'S': call_id}})
        result = score_call(transcript)
        dynamodb.put_item(TableName='call_scores', Item=result)

import simple_salesforce as sf

sf_client = sf.Salesforce(
    instance_url=os.getenv('SF_INSTANCE'),
    session_id=os.getenv('SF_TOKEN')
)

def write_score(call_id, score_data):
    sf_client.Call_Compliance__c.create({
        'Call_ID__c': call_id,
        'Agent_ID__c': score_data['agent_id'],
        'Score__c': score_data['overall_score'],
        'Rule_Violations__c': ','.join(score_data['violations'])
    })

SELECT
    DATE_TRUNC('day', cs.created_at) AS day,
    SUM(ci.avg_fine_usd * CASE WHEN cs.violation = ci.violation_type THEN 1 ELSE 0 END) AS estimated_fines,
    SUM(ci.avg_upsell_lost_usd * ...) AS estimated_upsell_lost
FROM call_scores cs
JOIN call_impact ci ON cs.violation = ci.violation_type
GROUP BY 1