Slack上でLLMを活用する社内チャットボット「しぇるぱ」の実装事例を紹介

func parseSlackEvent(body string) (*slackevents.EventsAPIEvent, error) {
    event, err := slackevents.ParseEvent(json.RawMessage(body), slackevents.OptionNoVerifyToken())
    if err != nil {
        return nil, fmt.Errorf("Slackイベントの解析に失敗しました: %w", err)
    }
    return &event, nil
}

func handleURLVerification(body string) (events.APIGatewayProxyResponse, error) {
    var r struct {
        Challenge string `json:"challenge"`
    }
    if err := json.Unmarshal([]byte(body), &r); err != nil {
        return createErrorResponse(400, err)
    }
    return events.APIGatewayProxyResponse{
        StatusCode: 200,
        Body:       r.Challenge,
    }, nil
}

func verifySlackRequest(body string, headers http.Header) error {
    // 署名検証処理（省略）
    return nil
}

func isSlackRetry(headers http.Header) bool {
    return headers.Get("X-Slack-Retry-Num") != ""
}

func createIgnoredRetryResponse() (events.APIGatewayProxyResponse, error) {
    responseBody, _ := json.Marshal(map[string]string{"message": "Slackの再試行を無視します"})
    return events.APIGatewayProxyResponse{
        StatusCode: 200,
        Headers:    map[string]string{"Content-Type": "application/json"},
        Body:       string(responseBody),
    }, nil
}

// handleCallbackEvent は、コールバックイベントを処理します。（5.1での説明対象）
func handleCallbackEvent(ctx context.Context, isOrchestrator bool, event *slackevents.EventsAPIEvent) (events.APIGatewayProxyResponse, error) {
    innerEvent := event.InnerEvent

    switch innerEvent.Data.(type) {
    case *slackevents.AppMentionEvent:
        // AppMentionEventの場合の処理（5.2で詳細説明）
    case *slackevents.MessageEvent:
        // MessageEventの場合の処理（5.2で詳細説明）
    case *slackevents.ReactionAddedEvent:
        // ReactionAddedEventの場合の処理（5.2で詳細説明）
    }

    return events.APIGatewayProxyResponse{Body: "OK", StatusCode: http.StatusOK}, nil
}

func sanitizeContent(htmlContent string) string {
    // 基本的なサニタイズ
    ugcPolicy := bluemonday.UGCPolicy()
    sanitized := ugcPolicy.Sanitize(htmlContent)

    // カスタムポリシーで特定タグを許可
    customPolicy := bluemonday.NewPolicy()
    customPolicy.AllowLists()
    customPolicy.AllowTables()
    customPolicy.AllowAttrs("href").OnElements("a")

    // タグごとに改行を追加して可読性向上
    formattedContent := addNewlinesForTags(sanitized, "p")

    // カスタムポリシー適用後の最終サニタイズ
    finalContent := customPolicy.Sanitize(formattedContent)

    return finalContent
}

func addNewlinesForTags(htmlStr string, tags ...string) string {
    for _, tag := range tags {
        closeTag := fmt.Sprintf("</%s>", tag)
        htmlStr = strings.ReplaceAll(htmlStr, closeTag, closeTag+"\n")
    }
    return htmlStr
}

import openai
from shared.secrets import get_secret

class OpenAIClientFactory:
    @staticmethod
    def create_client(region="eastus2", model="gpt-4o") -> openai.OpenAI:
        secret = get_secret()
        api_type = secret.get("openai_api_type", "azure")

        if api_type == "azure":
            return openai.AzureOpenAI(
                api_key=secret.get(f"azure_openai_api_key_{region}"),
                azure_endpoint=secret.get(f"azure_openai_endpoint_{region}"),
                api_version=secret.get(
                    f"azure_openai_api_version_{region}", "2024-07-01-preview"
                ),
            )
        elif api_type == "openai":
            return openai.OpenAI(api_key=secret.get("openai_api_key"))

        raise ValueError(f"Invalid api_type: {api_type}")

import json
from typing import Any, Dict, List
import openai
from openai.types.chat import ChatCompletionMessageParam
from shared.openai_client import OpenAIClientFactory

def chatCompletionHandler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    request_body = json.loads(event["body"])
    messages: List[ChatCompletionMessageParam] = request_body.get("messages", [])
    model = request_body.get("model", "gpt-4o")

    client = OpenAIClientFactory.create_client(model=model)

    temperature = request_body.get("temperature", 0.7)
    max_tokens = request_body.get("max_tokens", 4000)
    response_format = request_body.get("response_format", None)

    completion = client.chat.completions.create(
        model=model,
        stream=False,
        messages=messages,
        max_tokens=max_tokens,
        frequency_penalty=0,
        presence_penalty=0,
        temperature=temperature,
        response_format=response_format,
    )

    return {
        "statusCode": 200,
        "body": json.dumps(completion.to_dict()),
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Methods": "OPTIONS,POST",
            "Access-Control-Allow-Headers": "Content-Type,X-Amz-Date,Authorization,X-Api-Key,X-Amz-Security-Token",
        },
    }

name: Build and Upload FAISS
on:
  workflow_dispatch:
    inputs:
      environment:
        description: デプロイ環境
        type: environment
        default: dev

jobs:
  build-and-upload-faiss:
    environment: ${{ inputs.environment }}
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      # 必要なパッケージのインストール（faiss-cpu）
      - name: Install faiss-cpu
        run: |
          set -e
          echo "Installing faiss-cpu..."
          pip install faiss-cpu --no-deps

      # faiss のバイナリをアーカイブ
      - name: Archive faiss binaries
        run: |
          mkdir -p faiss_package
          pip install --target=faiss_package faiss-cpu
          tar -czvf faiss_package.tar.gz faiss_package

      # AWS認証情報の設定（環境に合わせてSecretsやRoleを指定）
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v3
        with:
          aws-access-key-id: ${{ secrets.CICD_AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.CICD_AWS_SECRET_ACCESS_KEY }}
          aws-region: ap-northeast-1

      # S3 にアップロード
      - name: Upload faiss binaries to S3
        run: |
          echo "Uploading faiss_package.tar.gz to S3..."
          aws s3 cp faiss_package.tar.gz s3://${{ secrets.AWS_S3_BUCKET }}/lambda/faiss_package.tar.gz
          echo "Upload complete."

# setup_faiss例：S3上のfaissパッケージをダウンロードし、sys.pathに追加
import os
import sys
import tarfile
from shared.logger import getLogger
from shared.s3_client import S3Client

logger = getLogger(__name__)

def setup_faiss(s3_client: S3Client, s3_bucket: str) -> None:
    try:
        import faiss
        logger.info("faiss が既にインポートされています。")
    except ImportError:
        logger.info("faiss が見つかりません。S3からダウンロードします。")
        faiss_package_key = "lambda/faiss_package.tar.gz"
        faiss_package_path = "/tmp/faiss_package.tar.gz"
        faiss_extract_path = "/tmp/faiss_package"

        # S3からパッケージをダウンロードして展開
        s3_client.download_file(bucket_name=s3_bucket, key=faiss_package_key, file_path=faiss_package_path)
        with tarfile.open(faiss_package_path, "r:gz") as tar:
            for member in tar.getmembers():
                member.name = os.path.relpath(member.name, start=member.name.split("/")[0])
                tar.extract(member, faiss_extract_path)

        sys.path.insert(0, faiss_extract_path)
        import faiss
        logger.info("faiss のインポートに成功しました。")

from typing import Any, Dict, List, Optional
from langchain_community.vectorstores import FAISS
from langchain_core.documents.base import Document
from langchain_core.vectorstores.base import VectorStoreRetriever
from shared.secrets import get_secret
from shared.logger import getLogger
from langchain_openai import AzureOpenAIEmbeddings, OpenAIEmbeddings

logger = getLogger(__name__)

def get_embeddings(secrets: Dict[str, str]):
    api_type: str = secrets.get("openai_api_type", "azure")

    if api_type == "azure":
        return AzureOpenAIEmbeddings(
            openai_api_key=secrets.get("azure_openai_api_key_eastus2"),
            azure_endpoint=secrets.get("azure_openai_endpoint_eastus2"),
            model="text-embedding-3-large",
            api_version=secrets.get("azure_openai_api_version_eastus2", "2023-07-01-preview"),
        )
    elif api_type == "openai":
        return OpenAIEmbeddings(
            openai_api_key=secrets.get("openai_api_key"),
            model="text-embedding-3-large",
        )
    else:
        logger.error("無効なAPIタイプが指定されています。")
        raise ValueError("Invalid api_type")

def search_data(
    query: str,
    index_folder_path: str,
    search_type: str = "similarity",
    score_threshold: Optional[float] = None,
    k: Optional[int] = None,
    fetch_k: Optional[int] = None,
    lambda_mult: Optional[float] = None,
) -> List[Dict]:
    secrets: Dict[str, str] = get_secret()
    embeddings = get_embeddings(secrets)
    db: FAISS = FAISS.load_local(
        folder_path=index_folder_path,
        embeddings=embeddings,
        allow_dangerous_deserialization=True,
    )

    search_kwargs = {"k": k}
    if search_type == "similarity_score_threshold" and score_threshold is not None:
        search_kwargs["score_threshold"] = score_threshold
    elif search_type == "mmr":
        search_kwargs["fetch_k"] = fetch_k or k * 4
        if lambda_mult is not None:
            search_kwargs["lambda_mult"] = lambda_mult

    retriever: VectorStoreRetriever = db.as_retriever(
        search_type=search_type,
        search_kwargs=search_kwargs,
    )
    results: List[Document] = retriever.invoke(input=query)

    return [{"content": doc.page_content, "metadata": doc.metadata} for doc in results]

import asyncio
import json
import os
from shared.s3_client import S3Client
from shared.logger import getLogger
from shared.token_verifier import with_token_verification

logger = getLogger(__name__)
RESULT_NUM = 5

@with_token_verification
async def async_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    env = os.getenv("ENV")
    s3_client = S3Client()
    s3_bucket = "bucket-name"

    setup_faiss(s3_client, s3_bucket)

    request_body_str = event.get("body", "{}")
    request_body = json.loads(request_body_str)
    query = request_body.get("query")
    index_path = request_body.get("index_path")

    local_index_dir = "/tmp/index_faiss"
    await download_files(s3_client, s3_bucket, index_path, local_index_dir)

    results = search_data(
        query,
        local_index_dir,
        search_type=request_body.get("search_type", "similarity"),
        score_threshold=request_body.get("score_threshold"),
        k=request_body.get("k", RESULT_NUM),
        fetch_k=request_body.get("fetch_k"),
        lambda_mult=request_body.get("lambda_mult"),
    )

    return create_response(200, results)

def retrieverHandler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    return asyncio.run(async_handler(event, context))

def create_response(status_code: int, body: Any) -> Dict[str, Any]:
    return {
        "statusCode": status_code,
        "body": json.dumps(body, ensure_ascii=False),
        "headers": {
            "Content-Type": "application/json",
        },
    }

async def download_files(s3_client: S3Client, bucket: str, key: str, file_path: str) -> None:
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, download_files_from_s3, s3_client, bucket, key, file_path)

def download_files_from_s3(s3_client: S3Client, s3_bucket: str, prefix: str, local_dir: str) -> None:
    keys = s3_client.list_objects(bucket_name=s3_bucket, prefix=prefix)
    if not keys:
        logger.info(f"'{prefix}'内にファイルが存在しません。")
        return
    for key in keys:
        relative_path = os.path.relpath(key, prefix)
        local_file_path = os.path.join(local_dir, relative_path)
        os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
        s3_client.download_file(bucket_name=s3_bucket, key=key, file_path=local_file_path)

import json
from typing import Any, Dict, List
from langchain_core.documents.base import Document
from shared.logger import getLogger

logger = getLogger(__name__)

def load_json(file_path: str) -> List[Document]:
    """
    JSONファイルを読み込み、Documentオブジェクトのリストを返します。
    JSONは [{"title": "...", "content": "...", "id": "...", "url": "..."}] のような形式を想定。
    """
    with open(file_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    if not isinstance(data, list):
        raise ValueError("JSONトップレベルがリストではありません。")

    documents = []
    for record in data:
        if not isinstance(record, dict):
            logger.warning(f"スキップされたレコード（辞書でない）: {record}")
            continue
        title = record.get("title", "")
        content = record.get("content", "")
        metadata = {
            "id": record.get("id"),
            "title": title,
            "url": record.get("url"),
        }
        # タイトルとコンテンツをまとめたテキストとしてDocument化
        doc = Document(page_content=f"Title: {title}\nContent: {content}", metadata=metadata)
        documents.append(doc)

    logger.info(f"{len(documents)} 件のドキュメントをロードしました。")
    return documents

import os
from langchain_community.vectorstores import FAISS
from langchain_core.text_splitter import RecursiveCharacterTextSplitter
from shared.logger import getLogger

logger = getLogger(__name__)

def vectorize_and_save(documents: List[Document], output_dir: str, embeddings) -> None:
    """
    ドキュメントをEmbeddingし、FAISSインデックスを作成してローカルに保存します。
    """
    # テキスト分割器を使用してドキュメントを小さなチャンクに分割
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=128)
    split_docs = text_splitter.split_documents(documents)
    logger.info(f"{len(split_docs)} 件の分割済みドキュメント")

    # Embeddingsでベクトル化し、FAISSインデックス構築
    db: FAISS = FAISS.from_documents(split_docs, embeddings)
    logger.info("ベクトルDBの構築が完了しました。")

    os.makedirs(output_dir, exist_ok=True)
    db.save_local(output_dir)
    logger.info(f"ベクトルDBを {output_dir} に保存しました。")

from shared.s3_client import S3Client
from shared.logger import getLogger

logger = getLogger(__name__)

def upload_faiss_to_s3(s3_client: S3Client, s3_bucket: str, local_index_dir: str, index_s3_path: str) -> None:
    """
    FAISSインデックスをS3にアップロードします。
    """
    index_files = ["index.faiss", "index.pkl"]
    for file_name in index_files:
        local_file_path = os.path.join(local_index_dir, file_name)
        s3_index_key = os.path.join(index_s3_path, file_name)
        s3_client.upload_file(local_file_path, s3_bucket, s3_index_key)
        logger.info(f"FAISSインデックスファイルを s3://{s3_bucket}/{s3_index_key} にアップロードしました。")

import os
from shared.faiss import setup_faiss
from shared.logger import getLogger
from shared.s3_client import S3Client
from shared.secrets import get_secret

logger = getLogger(__name__)

def index_to_s3(json_s3_key: str, index_s3_path: str) -> Dict[str, Any]:
    """
    S3からJSONファイルをダウンロードし、EmbeddingとFAISSインデックス作成を行い、インデックスをS3に保存します。
    """
    env = os.getenv("ENV")
    if env is None:
        error_msg = "ENV 環境変数が設定されていません。"
        logger.error(error_msg)
        return {"status": "error", "message": error_msg}

    try:
        s3_client = S3Client()
        s3_bucket = "bucket-name"
        local_json_path = "/tmp/json_file.json"
        local_index_dir = "/tmp/index"

        # 必要なら faiss のセットアップ（S3からダウンロード）
        setup_faiss(s3_client, s3_bucket)

        # JSONファイルをS3からダウンロード
        s3_client.download_file(s3_bucket, json_s3_key, local_json_path)
        documents = load_json(local_json_path)

        # Embeddingsクライアント取得
        secrets = get_secret()
        embeddings = get_embeddings(secrets)

        # ベクトル化とFAISSインデックス作成
        vectorize_and_save(documents, local_index_dir, embeddings)

        # インデックスファイルをS3へアップロード
        upload_faiss_to_s3(s3_client, s3_bucket, local_index_dir, index_s3_path)

        return {
            "status": "success",
            "message": "FAISSインデックスを作成し、S3にアップロードしました。",
            "output": {
                "bucket": s3_bucket,
                "index_key": index_s3_path,
            },
        }

    except Exception as e:
        logger.error(f"インデックス作成処理中にエラー発生: {e}")
        return {"status": "error", "message": str(e)}