Automate Customer Feedback Triage, Sentiment Analysis, and Escalation with Google Gemini and Slack

id: ai-customer-feedback-triage
namespace: company.team

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

  - id: hourly_polling
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"
    disabled: true

inputs:
  - id: customer_id
    type: STRING
    displayName: Customer ID
    description: Unique identifier of the customer submitting feedback
    defaults: "CUST-9842"

  - id: customer_tier
    type: STRING
    displayName: Customer Tier
    description: Subscription tier of the user (Enterprise, Business, or Free)
    defaults: "Enterprise"

  - id: feedback_text
    type: STRING
    displayName: Feedback Content
    description: Raw customer feedback or support ticket text
    defaults: "The dashboard is constantly throwing 504 gateway timeout errors when
      exporting large reports. Our entire data team is blocked, and we need this
      resolved immediately!"

tasks:
  - id: analyze_sentiment_and_topics
    type: io.kestra.plugin.gemini.StructuredOutputCompletion
    description: Classify customer feedback sentiment, score urgency on a scale of 1
      to 10, and extract actionable topics using Google Gemini structured
      output.
    apiKey: "{{ secret('GEMINI_API_KEY') }}"
    model: "gemini-3.5-flash-lite"
    prompt: |
      You are an expert customer success and incident triage AI.
      Analyze the following customer feedback:
      Customer Tier: {{ inputs.customer_tier }}
      Feedback: "{{ inputs.feedback_text }}"

      Evaluate sentiment, compute urgency score on a scale of 1 to 10, categorize the primary issue, and suggest an actionable response.
    jsonResponseSchema: |
      {
        "type": "object",
        "properties": {
          "sentiment": {
            "type": "string",
            "enum": ["POSITIVE", "NEUTRAL", "NEGATIVE"]
          },
          "urgency_score": {
            "type": "integer",
            "minimum": 1,
            "maximum": 10
          },
          "category": {
            "type": "string",
            "enum": ["PERFORMANCE", "BUG", "FEATURE_REQUEST", "BILLING", "GENERAL"]
          },
          "summary": {
            "type": "string"
          },
          "action_item": {
            "type": "string"
          }
        },
        "required": ["sentiment", "urgency_score", "category", "summary", "action_item"]
      }

  - id: process_and_persist
    type: io.kestra.plugin.core.flow.Parallel
    description: Concurrently persist the analyzed feedback into DuckDB and stream
      metrics to execution logs.
    tasks:
      - id: record_in_duckdb
        type: io.kestra.plugin.jdbc.duckdb.Queries
        description: Record customer attributes, AI sentiment, and urgency score into an
          embedded DuckDB analytics table.
        sql: |
          CREATE TABLE IF NOT EXISTS feedback_analytics (
            customer_id VARCHAR,
            tier VARCHAR,
            sentiment VARCHAR,
            urgency_score INTEGER,
            category VARCHAR,
            summary VARCHAR,
            recorded_at TIMESTAMP
          );

          INSERT INTO feedback_analytics VALUES (
            '{{ inputs.customer_id }}',
            '{{ inputs.customer_tier }}',
            '{{ outputs.analyze_sentiment_and_topics.predictions | first | jq(".sentiment") | first }}',
            {{ outputs.analyze_sentiment_and_topics.predictions | first | jq(".urgency_score") | first }},
            '{{ outputs.analyze_sentiment_and_topics.predictions | first | jq(".category") | first }}',
            '{{ outputs.analyze_sentiment_and_topics.predictions | first | jq(".summary") | first | replace("'", "''") }}',
            CURRENT_TIMESTAMP
          );

      - id: log_triage_metrics
        type: io.kestra.plugin.core.log.Log
        description: Log real-time incident triage metrics for operational observability.
        message: "Customer {{ inputs.customer_id }} ({{ inputs.customer_tier }})
          triaged: Category={{ outputs.analyze_sentiment_and_topics.predictions
          | first | jq('.category') | first }}, Urgency={{
          outputs.analyze_sentiment_and_topics.predictions | first |
          jq('.urgency_score') | first }}/10"

  - id: check_escalation
    type: io.kestra.plugin.core.flow.If
    description: Conditionally evaluate whether feedback urgency warrants immediate
      escalation to customer engineering.
    condition: "{{ (outputs.analyze_sentiment_and_topics.predictions | first |
      jq('.urgency_score') | first >= 7) or (inputs.customer_tier ==
      'Enterprise') }}"
    then:
      - id: escalate_to_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatch a rich priority alert to the on-call team via Slack
          incoming webhook.
        url: "{{ secret('SLACK_WEBHOOK') }}"
        messageText: |
          🚨 *Urgent Customer Feedback Alert* 🚨
          • *Customer ID:* `{{ inputs.customer_id }}` (Tier: *{{ inputs.customer_tier }}*)
          • *Category:* {{ outputs.analyze_sentiment_and_topics.predictions | first | jq('.category') | first }}
          • *Urgency Score:* {{ outputs.analyze_sentiment_and_topics.predictions | first | jq('.urgency_score') | first }}/10
          • *Sentiment:* {{ outputs.analyze_sentiment_and_topics.predictions | first | jq('.sentiment') | first }}
          • *Summary:* {{ outputs.analyze_sentiment_and_topics.predictions | first | jq('.summary') | first }}
          • *Recommended Action:* {{ outputs.analyze_sentiment_and_topics.predictions | first | jq('.action_item') | first }}