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

id: ai-slack-chatbot-with-memory
namespace: company.team

inputs:
  - id: user_message
    type: STRING
    displayName: User message
    description: The message sent by the user (passed from Slack webhook payload)
  - id: user_id
    type: STRING
    displayName: Slack user ID
    description: Slack user ID for per-user conversation history (e.g., U12345678)
  - id: channel_id
    type: STRING
    displayName: Slack channel or DM ID
    description: Slack channel or DM channel to post the response to (e.g.,
      C12345678 or D12345678)
  - id: system_prompt
    type: STRING
    displayName: Bot system prompt
    description: System instruction defining the chatbot's persona and capabilities
    defaults: "You are a helpful assistant. Answer questions clearly and concisely.
      If asked about something outside your knowledge, say so honestly."
  - id: max_history_turns
    type: INT
    displayName: Max conversation turns to retain
    description: Number of previous message pairs (user+assistant) to include as
      context (max 10)
    defaults: 5

tasks:
  - id: load_conversation_history
    type: io.kestra.plugin.core.kv.Get
    key: "chat_history_{{ inputs.user_id }}"
    errorOnMissing: false

  - id: build_and_respond
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      history_raw.json: "{{ outputs.load_conversation_history.value | default('[]') }}"
    script: |
      import json
      from kestra import Kestra

      # Load existing history
      try:
          with open("history_raw.json") as f:
              content = f.read().strip()
              history = json.loads(content) if content and content != 'null' else []
      except Exception:
          history = []

      if not isinstance(history, list):
          history = []

      max_turns = {{ inputs.max_history_turns }}

      # Trim to max_turns pairs (each turn = user + assistant message)
      if len(history) > max_turns * 2:
          history = history[-(max_turns * 2):]

      # Build messages for OpenAI: system + history + new user message
      messages = [{"role": "system", "content": """{{ inputs.system_prompt | replace('"', '\\"') }}"""}]
      messages.extend(history)
      messages.append({"role": "user", "content": """{{ inputs.user_message | replace('"', '\\"') }}"""})

      Kestra.outputs({"messages": messages, "history_length": len(history)})

  - id: generate_response
    type: io.kestra.plugin.openai.ChatCompletion
    apiKey: "{{ secret('OPENAI_API_KEY') }}"
    model: gpt-4o-mini
    maxTokens: 800
    messages: "{{ outputs.build_and_respond.vars.messages }}"

  - id: update_conversation_history
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      history_raw.json: "{{ outputs.load_conversation_history.value | default('[]') }}"
    script: |
      import json
      from kestra import Kestra

      try:
          with open("history_raw.json") as f:
              content = f.read().strip()
              history = json.loads(content) if content and content != 'null' else []
      except Exception:
          history = []

      if not isinstance(history, list):
          history = []

      max_turns = {{ inputs.max_history_turns }}
      ai_reply = """{{ outputs.generate_response.choices[0].message.content | replace('"', '\\"') | replace('\n', '\\n') }}"""

      # Append new exchange
      history.append({"role": "user", "content": """{{ inputs.user_message | replace('"', '\\"') }}"""})
      history.append({"role": "assistant", "content": ai_reply})

      # Keep only the last max_turns * 2 messages
      if len(history) > max_turns * 2:
          history = history[-(max_turns * 2):]

      Kestra.outputs({"updated_history": json.dumps(history)})

  - id: save_conversation_history
    type: io.kestra.plugin.core.kv.Set
    key: "chat_history_{{ inputs.user_id }}"
    value: "{{ outputs.update_conversation_history.vars.updated_history }}"
    kvType: STRING

  - id: post_reply_to_slack
    type: io.kestra.plugin.core.http.Request
    uri: "https://slack.com/api/chat.postMessage"
    method: POST
    contentType: "application/json"
    headers:
      Authorization: "Bearer {{ secret('SLACK_BOT_TOKEN') }}"
    body: |
      {
        "channel": "{{ inputs.channel_id }}",
        "text": "{{ outputs.generate_response.choices[0].message.content | replace('"', '\\"') | replace('\n', '\\n') }}",
        "mrkdwn": true
      }

triggers:
  - id: slack_message_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('SLACK_WEBHOOK_KEY') }}"
    description: |
      Receives POST requests from the Slack Events API (event_type: message).
      Configure your Slack app's Event Subscriptions URL to:
      https://<your-kestra-host>/api/v1/executions/webhook/company.team/ai-slack-chatbot-with-memory/<SLACK_WEBHOOK_KEY>
      Map trigger.body fields to flow inputs in the Slack app configuration or use a pre-processing flow.