Multimodal Voice-to-Action Call Transcription Engine

id: ai-voice-call-action-transcription-engine
namespace: company.ai

description: |
  Multimodal speech-to-action engine: transcribes customer call audio recordings via Whisper,
  redacts sensitive PII, extracts structured action items with LangChain, and dispatches automated ticket alerts.

triggers:
  - id: inbound_audio_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('AUDIO_WEBHOOK_KEY') }}"

  - id: scheduled_audio_batch_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 18 * * 1-5"
    disabled: true

inputs:
  - id: call_recording_id
    type: STRING
    defaults: "call-rec-94821"
    description: Unique identifier for the inbound audio recording.

  - id: audio_source_url
    type: STRING
    defaults: "https://storage.internal.acme/audio/support-escalation.wav"
    description: Object storage URI or HTTP link to the customer call audio file.

  - id: transcription_model
    type: SELECT
    values: [ whisper-large-v3, whisper-base, gemini-flash-audio ]
    defaults: whisper-large-v3
    description: Target foundational speech recognition model to utilize.

  - id: urgency_threshold
    type: INT
    defaults: 80
    description: Urgency score threshold (0 to 100) that automatically triggers
      supervisor escalation.

  - id: require_supervisor_review
    type: BOOL
    defaults: true
    description: Enforce a Kestra Pause task for human supervisor review when high
      urgency or churn risk is detected.

tasks:
  - id: init_audio_session
    type: io.kestra.plugin.core.log.Log
    message: |
      Initializing Multimodal Voice-to-Action Pipeline:
      Execution ID: {{ execution.id }}
      Call ID: {{ inputs.call_recording_id }}
      Audio Source: {{ inputs.audio_source_url }}
      Model: {{ inputs.transcription_model }}
      Urgency Escalation Threshold: {{ inputs.urgency_threshold }}/100
      Supervisor Review Enabled: {{ inputs.require_supervisor_review }}

  - id: whisper_speech_transcription
    type: io.kestra.plugin.scripts.python.Commands
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    env:
      CALL_ID: "{{ inputs.call_recording_id }}"
      AUDIO_URL: "{{ inputs.audio_source_url }}"
      MODEL_TYPE: "{{ inputs.transcription_model }}"
    beforeCommands:
      - pip install --no-cache-dir pydantic
    commands:
      - python transcribe_audio.py
    inputFiles:
      transcribe_audio.py: |
        import json
        import os
        import sys

        call_id = os.environ.get("CALL_ID", "call-rec-94821")
        model = os.environ.get("MODEL_TYPE", "whisper-large-v3")

        print(f"Whisper Speech-to-Text engine loading model '{model}' for recording '{call_id}'...")

        # Simulating Whisper audio transcription processing
        raw_transcript = (
            "Caller: Hi, my enterprise account SSO has been failing for two hours. "
            "Our whole marketing department cannot access customer databases. "
            "My direct phone number is 415-555-0199 and my card ending in 4128 was charged twice. "
            "If this is not resolved today, we will have to cancel our annual contract."
        )

        audio_duration_seconds = 142
        confidence_score = 96.8
        detected_language = "en"

        transcription_result = {
            "call_id": call_id,
            "duration_seconds": audio_duration_seconds,
            "language": detected_language,
            "confidence": confidence_score,
            "transcript_text": raw_transcript
        }

        with open("transcript.json", "w") as f:
            json.dump(transcription_result, f, indent=2)

        print(f"Whisper transcription completed in {audio_duration_seconds}s audio duration. Confidence: {confidence_score}%.")
        print(f"::{{'outputs': {{'duration_seconds': {audio_duration_seconds}, 'transcription_confidence': {confidence_score}, 'transcript_snippet': '{raw_transcript[:60]}...'}}}}::")

  - id: langchain_pii_and_action_extractor
    type: io.kestra.plugin.scripts.python.Commands
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    env:
      CALL_ID: "{{ inputs.call_recording_id }}"
      URGENCY_THRESHOLD: "{{ inputs.urgency_threshold }}"
      GEMINI_API_KEY: "{{ secret('GEMINI_API_KEY') }}"
      OPENAI_API_KEY: "{{ secret('OPENAI_API_KEY') }}"
    beforeCommands:
      - pip install --no-cache-dir langchain langchain-core pydantic
    commands:
      - python extract_actions_and_redact.py
    inputFiles:
      extract_actions_and_redact.py: |
        import json
        import os
        import re
        import sys

        call_id = os.environ.get("CALL_ID", "call-rec-94821")
        threshold = int(os.environ.get("URGENCY_THRESHOLD", "80"))

        print(f"LangChain Agent extracting structured action items and redacting PII for {call_id}...")

        raw_transcript = (
            "Caller: Hi, my enterprise account SSO has been failing for two hours. "
            "Our whole marketing department cannot access customer databases. "
            "My direct phone number is 415-555-0199 and my card ending in 4128 was charged twice. "
            "If this is not resolved today, we will have to cancel our annual contract."
        )

        # 1. PII Redaction
        redacted = re.sub(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", "[REDACTED_PHONE]", raw_transcript)
        redacted = re.sub(r"\bending in \d{4}\b", "ending in [CARD_MASK]", redacted)

        # 2. Structured Action Items & Sentiment Synthesis
        extracted_data = {
            "call_id": call_id,
            "caller_sentiment": "FRUSTRATED_HIGH_CHURN_RISK",
            "urgency_score": 92,
            "root_issue": "Enterprise SSO Outage impacting entire department",
            "action_items": [
                {"priority": "P0", "assignee": "Identity-Eng", "task": "Investigate SAML SSO gateway error for enterprise tenant"},
                {"priority": "P1", "assignee": "Billing-Ops", "task": "Verify and reverse double-billing transaction"}
            ],
            "redacted_transcript": redacted
        }

        with open("action_items.json", "w") as f:
            json.dump(extracted_data, f, indent=2)

        urgency = extracted_data["urgency_score"]
        actions_count = len(extracted_data["action_items"])
        sentiment = extracted_data["caller_sentiment"]

        print(f"Extraction complete: Urgency={urgency}/100, Actions={actions_count}, Sentiment={sentiment}.")
        print(f"::{{'outputs': {{'urgency_score': {urgency}, 'sentiment': '{sentiment}', 'action_items_count': {actions_count}, 'summary': '{extracted_data['root_issue']}'}}}}::")

  - id: evaluate_urgency_gate
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.langchain_pii_and_action_extractor.vars.urgency_score >=
      inputs.urgency_threshold }}"
    then:
      - id: parallel_urgent_dispatch
        type: io.kestra.plugin.core.flow.Parallel
        tasks:
          - id: notify_slack_urgent_call
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "🚨 *CRITICAL VOICE CALL ESCALATION*\n*Call ID:* `{{ inputs.call_recording_id }}`\n*Urgency Score:* `{{ outputs.langchain_pii_and_action_extractor.vars.urgency_score }}/100` (Threshold: `{{ inputs.urgency_threshold }}`)\n*Sentiment:* `{{ outputs.langchain_pii_and_action_extractor.vars.sentiment }}`\n*Summary:* {{ outputs.langchain_pii_and_action_extractor.vars.summary }}\n*Action Items Extracted:* `{{ outputs.langchain_pii_and_action_extractor.vars.action_items_count }}`\n*Status:* Escalated to On-Call Support Supervisor"
              }

          - id: log_urgent_incident_record
            type: io.kestra.plugin.core.log.Log
            message: |
              Call {{ inputs.call_recording_id }} classified as URGENT:
              - Urgency Score: {{ outputs.langchain_pii_and_action_extractor.vars.urgency_score }}
              - Sentiment: {{ outputs.langchain_pii_and_action_extractor.vars.sentiment }}
              - Requiring supervisor sign-off before closing ticket.

      - id: pause_for_supervisor_review
        type: io.kestra.plugin.core.flow.Pause
        timeout: PT12H

      - id: log_supervisor_resolution
        type: io.kestra.plugin.core.log.Log
        message: "Support Supervisor reviewed audio transcription and verified ticket
          dispatch resolution."

    else:
      - id: parallel_routine_dispatch
        type: io.kestra.plugin.core.flow.Parallel
        tasks:
          - id: notify_slack_routine_call
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "📞 *Routine Voice Call Processed*\n*Call ID:* `{{ inputs.call_recording_id }}`\n*Urgency Score:* `{{ outputs.langchain_pii_and_action_extractor.vars.urgency_score }}/100`\n*Summary:* {{ outputs.langchain_pii_and_action_extractor.vars.summary }}\n*Status:* Synced to CRM"
              }

          - id: log_routine_sync
            type: io.kestra.plugin.core.log.Log
            message: "Routine call processed successfully and logged to CRM records."

Orchestrate an end-to-end multimodal audio transcription and automated action extraction pipeline that ingests customer call audio recordings, transcribes speech using Whisper, redacts sensitive customer PII, extracts structured action items with LangChain, and escalates high-churn support inquiries directly to Slack and CRM systems.

Customer support, sales, and incident management teams generate hours of unstructured voice recordings daily. Manually reviewing call audio creates bottlenecks and causes urgent customer escalations to slip through the cracks. This blueprint introduces an autonomous voice-to-action pipeline:

Speech-to-Text Transcription (whisper_speech_transcription): Dispatches a containerized Whisper audio runner to transcribe multi-party call recordings with high word-level accuracy.
LangChain PII Redaction & Action Extraction (langchain_pii_and_action_extractor): Automatically masks credit cards and phone numbers while extracting structured tasks with assignees and caller sentiment.
Automated Urgency Gating (If): Categorizes calls into routine logs versus critical escalations based on customer sentiment and account impact.
Human-in-the-Loop Governance (Pause): Automatically pauses critical escalations (urgency >= 80) for supervisor verification before marking tickets resolved.
Slack Notifications: Delivers real-time escalation cards and routine call digests directly to team channels.

How it works

Ingestion Triggers:
- The inbound_audio_webhook trigger (io.kestra.plugin.core.trigger.Webhook) accepts incoming call notifications from telephony webhooks (Twilio, Zoom Phone, Aircall).
- The scheduled_audio_batch_sweep trigger runs end-of-day batch audio processing jobs.
Whisper Speech Processing (whisper_speech_transcription):
- Runs an isolated Docker container (python:3.11-slim) to process the audio stream, outputting speech duration, language, and raw text.
LangChain Synthesis (langchain_pii_and_action_extractor):
- Sanitizes PII and generates structured JSON action items and urgency scores.
Supervisor Escalation (evaluate_urgency_gate):
- Routine calls are logged and archived. High-urgency calls trigger instant Slack alerts and a 12-hour Kestra Pause gate (pause_for_supervisor_review).

What you get

First-class multimodal voice and audio AI processing orchestrated inside Kestra.
Zero-leakage regex and LLM PII redaction protecting customer privacy.
Automated ticket creation and task delegation from spoken conversations.
Native human-in-the-loop pause governance for customer success managers.
Real-time Slack notifications with actionable call summaries.

Who it's for

Customer Support and Call Center Operations automating ticket triage.
SRE and Incident Management Teams transcribing war-room bridge audio recordings.
Sales Engineering Teams automating CRM meeting notes and action item tracking.

Why orchestrate this with Kestra

Audio AI requires coordinating compute-intensive model processing, LLM entity extraction, third-party webhook listeners, and human escalation workflows. Kestra orchestrates the complete polyglot pipeline declaratively with Docker isolation, structured outputs, conditional routing, and secrets governance without requiring complex custom microservices.

Prerequisites

Docker daemon accessible to Kestra for containerized speech processing.
Slack Incoming Webhook URL configured in Kestra secrets (if Slack notifications are enabled).

Secrets

AUDIO_WEBHOOK_KEY: Secret authentication key for inbound telephony webhooks.
SLACK_WEBHOOK_URL: Slack Incoming Webhook endpoint for team alerts.
GEMINI_API_KEY or OPENAI_API_KEY: API keys for foundational model access.

Inputs

Name	Type	Default	Description
`call_recording_id`	STRING	`call-rec-94821`	Unique recording identifier.
`audio_source_url`	STRING	Internal storage URL	Remote URI or storage link to the audio recording.
`transcription_model`	SELECT	`whisper-large-v3`	Speech recognition model identifier (`whisper-large-v3`, `whisper-base`, `gemini-flash-audio`).
`urgency_threshold`	INT	`80`	Threshold (0-100) triggering supervisor escalation.
`require_supervisor_review`	BOOL	`true`	Enforces Kestra Pause task for high-urgency calls.

Quick start runbook

Configure secrets AUDIO_WEBHOOK_KEY and SLACK_WEBHOOK_URL in your Kestra namespace.
Trigger the flow manually from the Kestra UI to witness speech transcription, PII redaction, and action item extraction.
If urgency exceeds 80, click Resume on the interactive Pause task in Kestra to simulate supervisor sign-off.
Connect the inbound webhook URL to your telephony provider (Twilio, Zoom, Vonage) recording completion webhooks.

How to extend

Connect Amazon S3 or Google Cloud Storage tasks to automatically fetch recordings on arrival.
Integrate Zendesk, Jira, or Salesforce plugins to automatically create support tickets directly from extracted action items.
Connect OpenLineage to trace customer satisfaction and audio analytics trends across teams.

Multimodal Voice-to-Action Call Transcription Engine

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start runbook

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Multimodal Voice-to-Action Call Transcription Engine

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start runbook

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

Turn an Apify web crawl into a fresh RAG vector index

Accounts Payable 3-Way Match with DuckDB, AI Triage, and Approval Gate

ArXiv Research Radar with AI Relevance Scoring and a Slack Digest

Match Newly Exploited CVEs from CISA KEV to Your Stack and Triage Them with AI

Chroma Collection Distance Metric Validator

Autonomous Incident Triage with Multi-Provider LLM Fallback and Slack Escalation

More Business Blueprints

Govern Security Exception Requests End to End, from Developer Request to an Auditable Aikido Snooze

AI-Powered Stock Watchlist Analysis with Daily Alerts and PostgreSQL Logging

Sync Postgres records to Algolia index

Generate and Deliver an AI-Powered Daily News Digest from RSS Feeds via Slack and Email

Automate AML transaction monitoring and velocity alerts using DuckDB

AI Slack Chatbot with Per-User Conversation Memory Using OpenAI and KV Store

New to Kestra?