目录

• 一、前言
• 二、架构解析
• 三、环境准备
• • 3.1、亚马逊云科技账户配置
  • 3.2、Slack App 创建与设置
  • 3.3、亚马逊云科技资源部署
• 四、实践流程
• • 4.1、Lambda 函数核心逻辑
  • 4.2、部署与验证
• 五、总结

一、前言

在数字化协作日益成为主流的商业环境中，Slack 等团队沟通工具已然成为企业运营不可或缺的组成部分。伴随生成式人工智能技术的迅猛发展，将AI能力深度融入日常工作流程，以期提升效率与激发创新，正成为当前技术演进的核心驱动力。

本文将深入探讨如何巧妙运用亚马逊云科技的 Amazon Bedrock 服务，并结合 Slack 平台，共同构建一个功能强大的图像生成助手应用。

Amazon Bedrock 作为一项全面托管的基础模型（FM）服务，为开发者提供了便捷的接口，其无服务器架构的特性，使得开发者无需耗费精力管理底层基础设施，得以将全部精力聚焦于如何安全、高效地利用亚马逊云科技的各项服务，将生成式AI能力融入各类应用场景。本文将以 Stability AI 的 SDXL 基础模型为例，详细阐述从系统架构设计到实际部署的全过程，力求为技术实践者提供一套清晰、可操作的解决方案。

二、架构解析

当用户在 Slack 的特定频道中输入文本提示后，这些提示信息会通过 Slack API 传输到后端。在后端，部署在亚马逊云科技 Lambda 服务中的 Slack 网关会对其进行初步处理。

Lambda 函数作为核心处理单元，负责接收、验证并解析 Slack 消息，随后触发 Amazon Bedrock 服务来执行图像生成任务。生成的图像不会直接回传至 Slack，而是先被安全地存储在亚马逊云科技 S3 存储桶中，再通过 Amazon CloudFront 内容分发网络进行全球分发，确保图像能够快速加载且易于访问。

此外，该解决方案还整合了 Secrets Manager 服务，用于安全管理敏感凭证，同时引入 DynamoDB 数据库以防止消息重复处理，共同构建起一个稳定可靠的图像生成与分发链路。

三、环境准备

在着手构建基于 Amazon Bedrock 的 Slack 图像生成助手 App 之前，一系列细致的环境准备工作是必不可少的。这涵盖了亚马逊云科技账户的全面配置、Slack App 的创建与权限设置，以及所有必要资源的部署。本章将提供详尽的指导，确保顺利完成这些准备步骤，为后续的实践流程奠定坚实的基础。

3.1、亚马逊云科技账户配置

首先，确保拥有一个功能完备的亚马逊云科技账户，首先进入亚马逊云科技官网，点击创建免费账户：

输入电子邮件地址和账户名称：

填写手机号，信用卡或者借记卡号进行绑定并进行验证。

3.2、Slack App 创建与设置

下面开始创建和配置 Slack App 的详细步骤，首先，访问 Slack API 官方网站（https://api.slack.com/apps/new），并选择“From a manifest”选项来创建新的 Slack 应用程序。此方式允许通过 YAML 格式的配置文件来定义 App 的基本信息、功能和权限，从而实现更精细化的控制与更便捷的批量配置。

接下来导入 App Manifest，将以下 YAML 格式的 App Manifest 内容完整复制并粘贴到文本框中。

display_information:
  name: SDXL
features:
  app_home:
    home_tab_enabled: true
    messages_tab_enabled: false
    messages_tab_read_only_enabled: false
  bot_user:
    display_name: SDXL
    always_online: false
oauth_config:
  scopes:
    bot:
      - im:read
      - im:write
      - chat:write
      - app_mentions:read
      - im:history
      - incoming-webhook
settings:
  event_subscriptions:
    request_url: https://tobemodified.com
    bot_events:
      - app_home_opened
      - app_mention
      - message.im
  org_deploy_enabled: false
  socket_mode_enabled: false
  token_rotation_enabled: false

此 Manifest 文件详细定义了 App 的显示信息（如名称）、各项功能（如 App Home 的启用状态）、OAuth 权限，以及事件订阅的配置。完成粘贴后，点击“下一步”继续。

在审查配置信息阶段，务必仔细核对所有细节，确认后，点击“Create”按钮即可完成 Slack App 的创建。

在 App 创建完成后，需要在 Slack App 的配置页面中找到并记录下此密钥。Signing Secret 的作用是用于验证所有来自 Slack 的请求是否真实有效，防止未经授权的请求。此密钥将在后续通过 CloudFormation 部署时，被安全地保存到 Secrets Manager 中。

然后选择左侧导航栏的“OAuth & Permissions”选项，然后点击“Install to Workspace”，接着点击“Allow”。这一系列操作将生成一个 OAuth Token。同样，此 Token 也需要被精确记录下来，因为它将是 Lambda 函数向 Slack 发送消息时进行身份认证的关键凭证，并且也将通过 CloudFormation 被安全地保存到 Secrets Manager 中。

3.3、亚马逊云科技资源部署

为简化亚马逊云科技资源的部署过程，并确保所有组件的配置一致性与可重复性，将采用 CloudFormation 进行自动化部署。CloudFormation 允许通过编写模板来定义所需的所有云资源，并由其自动完成资源的创建、更新和删除操作，从而大大减少手动配置可能带来的错误。

在准备好 CloudFormation 模板后，即可开始创建 CloudFormation 堆栈。首先，登录亚马逊云科技管理控制台，并导航至 CloudFormation 服务页面。在此，可以选择创建一个新的堆栈，并上传准备好的 CloudFormation 模板文件。

在堆栈创建向导中，系统将要求填写一些必要的参数。这些参数通常包括：用于保存 Bedrock SDXL 生成图片的私有 S3 存储桶的名称、用于分发图片的 CloudFront 分配的域名、之前从 Slack App 获取的 Signing Secret，以及同样从 Slack App 获取的 OAuth Token。在确认所有参数无误后，务必勾选相关的确认选项（例如 I acknowledge that AMazon CloudFormation might create IAM resources.），然后点击“Create stack”按钮，启动资源的自动化部署过程。

堆栈创建成功后，从 CloudFormation 堆栈的输出中复制 SlackAPIGatewayEndpoint。此端点是 Lambda 函数接收 Slack 事件的唯一 URL，它将作为 Slack App 与后端服务通信的桥梁。

最后，返回 Slack API 网站中 App 的配置页面，并导航至“Event Subscriptions”部分。在此，将之前从 CloudFormation 堆栈输出中复制的 SlackAPIGatewayEndpoint 配置为事件订阅的请求 URL，替换掉 Manifest 中默认的 https://tobemodified.com 占位符。完成配置后，若 Slack 成功验证了该 URL 的可达性，将看到“Verified”状态，这表明 Slack App 已成功与 Lambda 函数建立了连接。此时，点击“Save Changes”保存配置。

至此，环境准备工作已全面完成。

四、实践流程

4.1、Lambda 函数核心逻辑

Lambda 函数是整个应用的大脑，它负责协调 Slack、Amazon Bedrock、S3 和 DynamoDB 之间的复杂交互。其主要逻辑封装在一个 handler 函数中，该函数作为 Lambda 事件的入口点，承载着处理所有传入 Slack 事件的重任。在函数内部，首先会进行严格的 Slack 签名验证，以确保请求的真实性。

随后，它会智能地区分两种类型的事件：一种是 Slack 在配置事件订阅时发送的 URL 验证请求，另一种则是用户在 Slack 中发送的实际消息事件。对于用户消息事件，handler 函数会采取一项关键的防重复处理机制：它会尝试将消息的 client_msg_id 写入 DynamoDB 表。如果写入操作因 ConditionalCheckFailedException 异常而失败，则表明该消息已被处理过，函数会立即返回成功状态码，从而有效避免了重复执行图像生成逻辑，确保了系统的幂等性。经过验证和去重之后，实际的消息内容将被转发给 handle_message 函数进行进一步的业务逻辑处理。

import json
import boto3
import os
import time
import hmac
import hashlib
import base64
import uuid
from botocore.exceptions import ClientError
from urllib import request as http_request

# 从环境变量或Secrets Manager加载配置
SLACK_SIGNING_SECRET = os.environ.get("SLACK_SIGNING_SECRET") # 实际应从Secrets Manager获取
SLACK_OAUTH_TOKEN = os.environ.get("SLACK_OAUTH_TOKEN") # 实际应从Secrets Manager获取
SLACK_API_URL = "https://slack.com/api/chat.postMessage"
S3_BUCKET_NAME = os.environ.get("S3_BUCKET_NAME")
CLOUDFRONT_NAME = os.environ.get("CLOUDFRONT_NAME")
S3_PREFIX = "generated-images/"

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ.get("DYNAMODB_TABLE_NAME"))
s3_client = boto3.client("s3")

def verify_slack_signature(headers, body):
    timestamp = headers.get("X-Slack-Request-Timestamp")
    if not timestamp:
        return False

    req_signature = headers.get("X-Slack-Signature")
    if not req_signature:
        return False

    basestring = f"v0:{timestamp}:{body}".encode("utf-8")
    my_signature = "v0=" + hmac.new(SLACK_SIGNING_SECRET.encode("utf-8"), basestring, hashlib.sha256).hexdigest()
    return hmac.compare_digest(my_signature, req_signature)

def handler(event, context):
    event_body = json.loads(event.get("body"))
    response = None

    # 1. 验证 Slack 签名
    if not verify_slack_signature(event["headers"], event["body"]):
        print("Invalid Slack signature")
        return {
            "statusCode": 401,
            "body": json.dumps({"error": "Invalid Slack signature"})
        }

    # 2. 处理 URL 验证或实际消息事件
    if event_body.get("type") == "url_verification":
        print("URL verification request received")
        response = {
            "statusCode": 200,
            "body": event_body["challenge"]
        }
    else:
        client_msg_id = event_body["event"]["client_msg_id"]
        try:
            # 3. 防止重复处理：写入 DynamoDB
            table.put_item(
                Item={
                    "client_msg_id": client_msg_id,
                    "timestamp": int(time.time())  # 添加时间戳以防止重放攻击
                },
                ConditionExpression="attribute_not_exists(client_msg_id)"
            )
            print(f"Processing new message with client_msg_id: {client_msg_id}")
            response = handle_message(event_body)

        except ClientError as e:
            if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
                print(f"Event already processed: {client_msg_id}")
                return {
                    "statusCode": 200,
                    "body": json.dumps("Event already processed")
                }
            else:
                print(f"DynamoDB error: {e}")
                raise # 抛出其他异常
        except Exception as e:
            print(f"Error in handler: {e}")
            raise # 抛出其他异常

    return response

def handle_message(slack_body):
    import re
    slack_text = slack_body.get("event").get("text")
    slack_user = slack_body.get("event").get("user")
    channel = slack_body.get("event").get("channel")

    # 提取实际的提示词，去除 @App名称
    pattern = r"<@\w+>\s*(.+)"
    match = re.search(pattern, slack_text)
    if match:
        prompt_text = match.group(1).strip()
    else:
        prompt_text = slack_text.strip()

    print(f"Received prompt: {prompt_text} from user: {slack_user} in channel: {channel}")

    # 4. 调用 Bedrock 生成图像
    image_url = call_bedrock(prompt_text)

    # 5. 构造 Slack 消息并发送
    data = {
        "channel": channel,
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"<@{slack_user}> 生成的图像:"
                }
            },
            {
                "type": "image",
                "image_url": image_url,
                "alt_text": "Generated Image"
            }
        ]
    }

    headers = {"Authorization": f"Bearer {SLACK_OAUTH_TOKEN}", "Content-Type": "application/json"}
    req = http_request.Request(SLACK_API_URL, data=json.dumps(data).encode("utf-8"), headers=headers)
    try:
        with http_request.urlopen(req) as res:
            slack_response = json.loads(res.read().decode("utf-8"))
            if not slack_response.get("ok"):
                print(f"Error sending message to Slack: {slack_response.get("error")}")
    except Exception as e:
        print(f"HTTP request error to Slack: {e}")

    return {"statusCode": 200, "body": json.dumps({"msg": "message received and processed"})}

def call_bedrock(question):
    bedrock_runtime_client = boto3.client("bedrock-runtime")
    model_id = "stability.stable-diffusion-xl-v1"

    native_request = {
        "text_prompts": [{"text": question}],
        "style_preset": "photographic",
        "seed": os.urandom(4).hex(), # 使用更安全的随机数生成方式
        "cfg_scale": 10,
        "steps": 30,
    }

    request_body = json.dumps(native_request)

    try:
        response = bedrock_runtime_client.invoke_model(modelId=model_id, body=request_body)
    except Exception as e:
        print(f"调用 Amazon Bedrock AI 模型出错: {e}")
        return "抱歉，当前无法处理请求。"

    model_response = json.loads(response["body"].read())
    base64_image_data = model_response["artifacts"][0]["base64"]

    # 6. 上传图像到 S3 并获取 CloudFront URL
    image_key = f"{S3_PREFIX}{str(uuid.uuid4())}.png"
    s3_client.put_object(
        Bucket=S3_BUCKET_NAME,
        Key=image_key,
        Body=base64.b64decode(base64_image_data),
        ContentType="image/png"
    )

    image_url = f"https://{CLOUDFRONT_NAME}/{image_key}"
    print(f"Generated image URL: {image_url}")
    return image_url

4.2、部署与验证

完成了 Amazon Lambda 函数的代码编写和所有相关资源的部署之后，接下来便是对整个图像生成助手 App 进行实际的部署与验证。

首先，需要将上述 Amazon Lambda 函数的代码进行打包，并将其部署到在 CloudFormation 中创建的 Amazon Lambda 函数实例中。在部署过程中，务必确保所有必要的环境变量都已得到正确配置。这些环境变量包括 SLACK_SIGNING_SECRET、SLACK_OAUTH_TOKEN、S3_BUCKET_NAME、CLOUDFRONT_NAME 以及 DYNAMODB_TABLE_NAME。

完成部署后，便可以进入实际的 Slack 环境进行验证。打开 Slack 工作区，并导航至任意一个频道。在频道中，可以通过 @App名称 加上提示词 的方式向图像生成助手 App 发送消息。例如，如果之前将 App 命名为 SDXL，那么可以尝试输入 @SDXL 一只在太空漫步的宇航员狗，以此来测试其图像生成能力。值得注意的是，在首次使用该 App 时，Slack 可能会弹出一个提示，要求邀请该 App 加入当前频道。此时，只需点击“Invite Them”即可完成操作，允许 App 在该频道中进行交互。

发送消息后，耐心等待片刻，在频道中将看到由图像生成助手 App 发送的包含精美生成图像的消息，这标志着整个系统已成功地端到端运行。

通过以上详尽的实践流程，就成功地构建并运行了一个基于 Amazon Bedrock 的 Slack 图像生成助手 App。

这个应用不仅直观地展示了生成式 AI 在实际场景中的强大潜力和无限可能，更为团队协作和内容创作带来了前所未有的便捷与效率。

五、总结

展望未来，随着生成式 AI 技术的持续演进和 Amazon Bedrock 服务能力的不断增强，可以预见此类智能助手将在更广泛的业务场景中发挥其关键作用，从而有力推动企业的数字化转型迈向新的高度。