LangChain × SlackAPI × Lambda でWebページを要約してくれるSlackChatGPTサーバーレスアプリ作った

## 作ったもの
SlackでWebページを渡すと要約してくれて、その後はURLを元に回答を生成してくれるアプリを作りました🤗
(難解なWebページとか読むのめんどくさい時用です)
[![Image from Gyazo](https://i.gyazo.com/ca209fcc3eac1d1f97f93bf06ffb53b1.png)](https://gyazo.com/ca209fcc3eac1d1f97f93bf06ffb53b1)

## インフラ構成
[![Image from Gyazo](https://i.gyazo.com/5b92423f911713ad3f94d49c10043438.png)](https://gyazo.com/5b92423f911713ad3f94d49c10043438)

### それぞれの役割
[![Image from Gyazo](https://i.gyazo.com/9cdbbdaeab202e6ebeb550a715fca5a4.png)](https://gyazo.com/9cdbbdaeab202e6ebeb550a715fca5a4)

こんな感じの構成になっています。
LangChain周りのパッケージが重たすぎて、LambdaLayerのデプロイパッケージの制限(解凍後サイズ 250MB)を超えてしまうので、ECRにDockerイメージを保管しそこからLambdaを展開しています。

## 資料
イベント登壇時の資料です
<iframe src="https://docs.google.com/presentation/d/e/2PACX-1vQdFmuibIpWg7HB1zAXHJHkxmlwWxkFebKxax-CMhJRakUCLkfAFycK7hv10Ri7Fw/embed?start=false&loop=false&delayms=3000" frameborder="0" width="960" height="749" allowfullscreen="true" mozallowfullscreen="true" webkitallowfullscreen="true"></iframe>

## コード

超特急で作ったので細かい所や汚いところは目を瞑ってください🥲

```Dockerfile:Dockerfile
FROM public.ecr.aws/lambda/python:3.8

# Upgrade pip
RUN python3 -m pip install --upgrade pip

COPY requirements.txt ./
RUN pip install -r requirements.txt

COPY handler.py ./

CMD ["handler.lambda_handler"]
```

```text:requirements.tx
aiofiles==23.1.0
aiohttp==3.8.4
aiosignal==1.3.1
altair==5.0.0
anyio==3.6.2
async-timeout==4.0.2
attrs==23.1.0
beautifulsoup4==4.12.2
boto3==1.26.96
botocore==1.29.136
certifi==2023.5.7
cffi==1.15.1
charset-normalizer==3.1.0
click==8.1.3
contourpy==1.0.7
cryptography==40.0.2
cycler==0.11.0
dataclasses-json==0.5.7
faiss-cpu==1.7.4
fastapi==0.95.2
ffmpy==0.3.0
filelock==3.12.0
fonttools==4.39.4
frozenlist==1.3.3
fsspec==2023.5.0
geojson==2.5.0
gradio==3.23.0
greenlet==2.0.2
h11==0.14.0
httpcore==0.17.1
httpx==0.24.1
huggingface-hub==0.14.1
idna==3.4
Jinja2==3.1.2
jmespath==1.0.1
jsonschema==4.17.3
kiwisolver==1.4.4
langchain==0.0.126
linkify-it-py==2.0.2
markdown-it-py==2.2.0
MarkupSafe==2.1.2
marshmallow==3.19.0
marshmallow-enum==1.5.1
matplotlib==3.7.1
mdit-py-plugins==0.3.3
mdurl==0.1.2
multidict==6.0.4
mypy-extensions==1.0.0
numexpr==2.8.4
numpy==1.24.3
openai==0.27.2
openapi-schema-pydantic==1.2.4
orjson==3.8.12
packaging==23.1
pandas==2.0.1
pdfminer.six==20221105
pdfplumber==0.9.0
Pillow==9.5.0
pycparser==2.21
pydantic==1.10.7
pydub==0.25.1
pyowm==3.3.0
pyparsing==3.0.9
pyrsistent==0.19.3
PySocks==1.7.1
python-dateutil==2.8.2
python-multipart==0.0.6
pytz==2023.3
PyYAML==6.0
regex==2023.5.5
requests==2.30.0
s3transfer==0.6.1
semantic-version==2.10.0
six==1.16.0
sniffio==1.3.0
soupsieve==2.4.1
SQLAlchemy==1.4.48
starlette==0.27.0
tenacity==8.2.2
tiktoken==0.3.3
tokenizers==0.13.3
toolz==0.12.0
tqdm==4.65.0
transformers==4.29.2
typing-inspect==0.8.0
typing_extensions==4.5.0
tzdata==2023.3
uc-micro-py==1.0.2
urllib3==1.26.15
uvicorn==0.22.0
Wand==0.6.11
websockets==11.0.3
yarl==1.9.2
```

```python:handler.py
import os
import json
import tempfile
import boto3
import openai
import langchain
import requests
import pdfplumber
import re

from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chat_models import ChatOpenAI
from langchain.docstore.document import Document
from langchain.chains.summarize import load_summarize_chain
from langchain.prompts import PromptTemplate
from langchain.vectorstores import FAISS
from langchain.chains import ConversationalRetrievalChain
from langchain.memory.chat_message_histories import DynamoDBChatMessageHistory
from langchain.chains import LLMChain
from langchain.chains.question_answering import load_qa_chain
from langchain.chains.conversational_retrieval.prompts import CONDENSE_QUESTION_PROMPT
from urllib.parse import urljoin, urlparse
from bs4 import BeautifulSoup
from io import BytesIO
from datetime import datetime

URL_SUMMARIZER_ID = os.environ["URL_SUMMARIZER_ID"]
BUCKET_NAME = os.environ["BUCKET_NAME"]
SESSION_TABLE_NAME = os.environ["SESSION_TABLE_NAME"]

# S3の準備
s3 = boto3.resource('s3')
s3_bucket = s3.Bucket(BUCKET_NAME)

# ChatGPTの準備
openai.api_key = os.environ["OPENAI_API_KEY"]
llm = ChatOpenAI(temperature=0, model_name="gpt-3.5-turbo")
embeddings = OpenAIEmbeddings()

url = ""

summarize_prompt_template = """以下の文章を簡潔に要約してください。:

{text}

要約:"""

qa_prompt_template = """

{context}

これを踏まえて、次の質問に日本語で答えてください。

質問: {question}
回答:"""

# Webページからテキストを抽出
def get_webpage_texts(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'html.parser')
    text_list = []

    main_div = soup.find('div', class_='main')
    if main_div:
        for string in main_div.stripped_strings:
            text_list.append(string)
    else:
        body = soup.find('body')
        if body:
            for string in body.stripped_strings:
                text_list.append(string)

    return text_list

# PDFページからテキストを抽出
def get_pdf_text(url):
    response = requests.get(url)
    pdf_file = BytesIO(response.content)
    text_list = []
    with pdfplumber.open(pdf_file) as pdf:
        for page in pdf.pages:
            text_list.append(page.extract_text())
    return text_list

# URLをS3フォルダ名に変更 domain-yyyymmddhhmmssの形にする
def modify_url_to_s3_path(url):
    now = datetime.now()
    timestamp = now.strftime("%Y%m%d%H%M%S")
    parsed_url = url.split("/")
    domain = parsed_url[2].replace(".", "-")
    new_url = f"{domain}-{timestamp}"
    return new_url

# S3にディレクトリを格納
def upload_dir_s3(dirpath, s3bucket, s3_path):
    for root,dirs,files in os.walk(dirpath):
        for file in files:
            s3_key = os.path.join(root, file).replace(dirpath, s3_path, 1)
            s3bucket.upload_file(os.path.join(root, file), s3_key)

# S3からディレクトリを取得
def download_dir_s3(dirpath, s3bucket):
    for obj in s3bucket.objects.filter(Prefix = dirpath):
        if not os.path.exists(os.path.join('/tmp/', os.path.dirname(obj.key))):
            os.makedirs(os.path.join('/tmp/', os.path.dirname(obj.key)))
        s3bucket.download_file(obj.key, os.path.join('/tmp/', obj.key))
    return os.path.join('/tmp/', dirpath)

# DynamoDBへ保存処理
def write_to_dynamo(url, s3_path, summarize_result, channel_name, user_name):
    dynamodb = boto3.resource('dynamodb')
    table_name = os.environ["TABLE_NAME"]
    table = dynamodb.Table(table_name)
    now = datetime.now()

    # データを書き込む
    table.put_item(
        Item={
            'Url': url,
            'AttributeType': "S3Key",
            'AttributeValue': s3_path,
            'ChannelName': channel_name,
            'UserName': user_name,
            'SummarizeResult': summarize_result,
            'created_at': now.isoformat()
        }
    )

# DynamoDBからの取得処理
def get_from_dynamo(url):
    if url == "":
        return None

    dynamodb = boto3.resource('dynamodb')
    table_name = os.environ["TABLE_NAME"]
    table = dynamodb.Table(table_name)

    # データを取得する
    response = table.get_item(
        Key={
            'Url': url,
            'AttributeType': "S3Key"
        }
    )

    # 取得した項目が存在するかを確認
    if 'Item' in response:
        return response['Item']
    else:
        return None

# DynamoDBから会話履歴処理
def get_history(session_id):
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(SESSION_TABLE_NAME)

    # データを取得する
    response = table.get_item(
        Key={
            'SessionId': session_id
        }
    )

    chat_history = []
    # 取得した項目が存在するかを確認
    if 'Item' in response and response['Item'].get("History") is not None:
        # "History"の下のリストをループ処理してchat_historyを生成
        human_content = ''
        for history_item in response['Item']["History"]:
            if history_item["type"] == "human":
                human_content = history_item["data"]["content"]
            elif history_item["type"] == "ai" and human_content:
                ai_content = history_item["data"]["content"]
                chat_history.append((human_content, ai_content))
                human_content = ''
        print(chat_history)
        return chat_history
    else:
        print(chat_history)
        return chat_history

# Slackへ返信する
def send_slack_message(channel, text, thread_ts):
    SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
    headers = {
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {SLACK_BOT_TOKEN}"
    }

    data = {
        "token": SLACK_BOT_TOKEN,
        "channel": channel,
        "text": text,
        "thread_ts": thread_ts
    }

    response = requests.post("https://slack.com/api/chat.postMessage", json=data, headers=headers)
    return response

# VectorIndexを元にQA処理を行う
def exec_qa(item, query, thread_ts, channel, thread_head_ts):
    print("DynamoDB item: " + json.dumps(item))
    s3_download_path = item["AttributeValue"]
    print("s3_download_path: " + s3_download_path)

    # S3から保存済みのVectorIndexを取得
    index_dir = download_dir_s3(s3_download_path, s3_bucket)
    dbDownload = FAISS.load_local(index_dir, embeddings)

    # QA処理
    qa_prompt = PromptTemplate(
        template=qa_prompt_template, input_variables=["context", "question"]
    )

    question_generator = LLMChain(llm=llm, prompt=CONDENSE_QUESTION_PROMPT)
    doc_chain = load_qa_chain(llm, chain_type="stuff", prompt=qa_prompt)

    # スレッドの過去発言を取得
    message_history = DynamoDBChatMessageHistory(table_name=SESSION_TABLE_NAME, session_id=thread_head_ts)
    chat_history = get_history(thread_head_ts)

    # 無視という単語が入っていたらベクトルデータを参考にせずに質問させる
    if "無視" in query:
        dummy_index = download_dir_s3("dummy-empty-vectorindex", s3_bucket)
        dummy_db = FAISS.load_local(dummy_index, embeddings)
        qa = ConversationalRetrievalChain.from_llm(llm, dummy_db.as_retriever())
        qa_result = qa({"question": query, "chat_history": []})
    else:
        qa = ConversationalRetrievalChain(retriever=dbDownload.as_retriever(), question_generator=question_generator, combine_docs_chain=doc_chain)
        qa_result = qa({"question": query, "chat_history": chat_history})

    print("qa_result: " + qa_result["answer"])

    # DynamoDBにChatGPTの回答を保存
    message_history.add_ai_message(qa_result["answer"])
    send_slack_message(channel, qa_result["answer"], thread_ts)

# Slackのスレッドから元となる質問元となるURLを取得する
def get_thread_url(thread_ts, channel):
    headers = {
        'Authorization': 'Bearer ' + os.environ['SLACK_BOT_TOKEN'],
        'Content-type': 'application/json; charset=utf-8'
    }

    # メッセージからURLを取得するために正規表現を用意
    user_id_pattern = re.compile(re.escape(URL_SUMMARIZER_ID))
    url_pattern = r'(https?://[^|>]+)'

    next_cursor = None
    while True:
        params = {
            'channel': channel,
            'ts': thread_ts,
            'cursor': next_cursor  # ページネーションのためのcursorパラメータ
        }
        response = requests.get(
            'https://slack.com/api/conversations.replies',
            params=params,
            headers=headers
        )
        response.raise_for_status()  # 応答のステータスコードが200以外の場合にエラーを発生させる
        data = response.json()
        if not data.get('ok'):
            raise RuntimeError(f"Slack API request failed: {data.get('error')}")

        # 該当スレッドのメッセージ
        messages = data['messages']

        # メッセージを一つずつチェックしていく
        for message in messages:
            if re.search(user_id_pattern, message['text']) and re.search(url_pattern, message['text']):
                url = re.search(url_pattern, message['text']).group(0)
                print(f"Found URL: {url}")
                return url

        # 次のページが存在する場合、cursorを更新して再度リクエストを行う
        next_cursor = data.get('response_metadata', {}).get('next_cursor')
        if not next_cursor:
            break

    # URLが見つからなかった場合は空文字列を返却
    return ""

# Slackチャンネル名を取得
def get_channel_name(channel_id):
    headers = {
        'Authorization': 'Bearer ' + os.environ['SLACK_BOT_TOKEN'],
        'Content-type': 'application/json; charset=utf-8'
    }
    params = {
        'channel': channel_id
    }
    response = requests.get(
        'https://slack.com/api/conversations.info',
        params=params,
        headers=headers
    )
    response.raise_for_status()  # 応答のステータスコードが200以外の場合にエラーを発生させる
    data = response.json()
    if data.get('ok'):
        return data['channel'].get('name', 'Unknown channel')
    else:
        return 'Unknown channel'

# Slackユーザー名を取得
def get_user_name(user_id):
    headers = {
        'Authorization': 'Bearer ' + os.environ['SLACK_BOT_TOKEN'],
        'Content-type': 'application/json; charset=utf-8'
    }
    params = {
        'user': user_id
    }
    response = requests.get(
        'https://slack.com/api/users.info',
        params=params,
        headers=headers
    )
    response.raise_for_status()  # 応答のステータスコードが200以外の場合にエラーを発生させる
    data = response.json()
    if data.get('ok'):
        user = data['user']
        return user.get('real_name', user.get('name', 'Unknown user'))
    else:
        return 'Unknown user'

def lambda_handler(event, context):
    print(f"Event: {json.dumps(event)}")
    body = json.loads(event['body'])

    # Slackの疎通確認用
    if body['type'] == 'url_verification':
        print("Slack challenge.")
        return {
            'statusCode': 200,
            'body': body['challenge']
        }

    headers = event['headers']
    # 処理に時間がかかりSlackに3秒以内にリクエストを返せないとリトライされる。Slackからのリトライは無視。
    if 'x-slack-retry-reason' in headers or 'x-slack-retry-num' in headers:
        print("Slack retry.")
        return {
            'statusCode': 200,
            'body': "Slack retry. ok."
        }

    TEAM_ID = os.environ["TEAM_ID"]
    API_APP_ID = os.environ["API_APP_ID"]

    team_id = body['team_id']
    api_app_id = body['api_app_id']

    # 変なリクエストは弾く
    if team_id != TEAM_ID or api_app_id != API_APP_ID:
        print("Not authorized.")
        print("team_id: " + team_id)
        print("api_app_id: " + api_app_id)
        return {
            'statusCode': 400,
            'body': json.dumps({
                'error': {
                    'message': 'Invalid team ID or API app ID.',
                    'code': 'INVALID_REQUEST'
                }
            })
        }

    # メンション時
    if body['event']['type'] == 'app_mention':
        try:
            # 呼び出しもとSlack情報
            channel = body['event']['channel']
            thread_ts = body['event']['ts']
            text = body['event']['text']
            channel_id = body["event"]["channel"]
            user_id = body["event"]["user"]
            print("text: " + text)

            # Slack本文から正規表現を使ってURLを切り出し
            url_pattern = r'(https?://[^|>]+)'
            urls = re.findall(url_pattern, text)
            if urls:
                url = urls[0]
                url_contain = True
                print("url: " + url)
            else:
                url_contain = False

            if url_contain:

                # 対話を記憶させるためのテーブル準備 IDはSlackのスレッド頭のthread_tsにする
                message_history = DynamoDBChatMessageHistory(table_name=SESSION_TABLE_NAME, session_id=thread_ts)
                # 質問からメンションを削除してDynamoDBにユーザーの発言を保存
                user_message = re.sub(re.escape(URL_SUMMARIZER_ID), '', text)
                message_history.add_user_message(user_message)

                item = get_from_dynamo(url)
                if item is not None:
                    print("Already summarized.")
                    message = item["SummarizeResult"]
                    message_history.add_ai_message(message)
                    send_slack_message(channel, message, thread_ts)
                else:
                    # 投稿にurlが含まれておりので要約を行う
                    print("Summarize start.")

                    processing_message = "データを学習して要約作成中です。"
                    send_slack_message(channel, processing_message, thread_ts)

                    # テキストを取得してtext_listに格納
                    text_list = []
                    if urlparse(url).path.endswith('.pdf'):
                        text_list.extend(get_pdf_text(url))
                    else:
                        text_list.extend(get_webpage_texts(url))
                    print("text_list: " + text_list[0])

                    # テキストを学習用に分割する
                    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
                    texts = text_splitter.split_text(text_list)
                    docs = [Document(page_content=t) for t in texts[:3]]

                    # 要約処理
                    summarize_template = PromptTemplate(
                                template=summarize_prompt_template, input_variables=["text"])
                    chain = load_summarize_chain(llm, chain_type="map_reduce", map_prompt=summarize_template, combine_prompt=summarize_template)
                    summarize_result = chain.run(docs)
                    print("summarize_result: " + summarize_result)

                    # ベクトルデータ化してindexを作成する
                    dbCreate = FAISS.from_documents(docs, embeddings)
                    dbCreate.save_local("/tmp/tmp_index")
                    print("Vectorize done!")

                    # S3へアップロードする
                    s3_upload_path = modify_url_to_s3_path(url)
                    upload_dir_s3("/tmp/tmp_index", s3_bucket, s3_upload_path)
                    print("S3 uploaded: " + s3_upload_path)

                    # チャンネル名とユーザー名を取得
                    channel_name = get_channel_name(channel_id)
                    user_name = get_user_name(user_id)

                    # DynamoDBへアップロードする
                    write_to_dynamo(url, s3_upload_path, summarize_result, channel_name, user_name)
                    print("DynamoDB write done!")

                    # DynamoDBにChatGPTの回答を保存
                    message_history.add_ai_message(summarize_result)
                    send_slack_message(channel, summarize_result, thread_ts)
            else:
                # 投稿自体にurlが含まれていないので質問と判断
                print("QA start.")
                # 質問からメンションを削除する
                query = re.sub(re.escape(URL_SUMMARIZER_ID), '', text)
                print("query: " + query)

                # 後から使うかもなので空VectorIndexでConversationalRetrievalChainを用意しておく
                dummy_index = download_dir_s3("dummy-empty-vectorindex", s3_bucket)
                dummy_db = FAISS.load_local(dummy_index, embeddings)
                qa = ConversationalRetrievalChain.from_llm(llm, dummy_db.as_retriever())

                # スレッドから対象となるURLを取得する
                if 'thread_ts' in body['event']:
                    thread_head_ts = body['event']['thread_ts']
                    url = get_thread_url(thread_head_ts, channel)
                    print("QA url: " + url)

                    # 対話を記憶させるためのテーブル準備 IDはSlackのスレッド頭のthread_tsにする
                    message_history = DynamoDBChatMessageHistory(table_name=SESSION_TABLE_NAME, session_id=thread_head_ts)
                    # DynamoDBにユーザーの発言を保存
                    message_history.add_user_message(query)

                else:
                    # スレッド内ではないメンション時にURLが入っていないことを想定
                    # DynamoDBにユーザーの発言を保存
                    message_history = DynamoDBChatMessageHistory(table_name=SESSION_TABLE_NAME, session_id=thread_ts)
                    message_history.add_user_message(query)

                    qa_result = qa({"question": query, "chat_history": []})

                    print("qa_result: " + qa_result["answer"])

                    # DynamoDBにChatGPTの回答を保存
                    message_history.add_ai_message(qa_result["answer"])
                    send_slack_message(channel, qa_result["answer"], thread_ts)
                    return

                # DynamoDBから該当urlのベクトルデータが保存されているS3の情報を取得する
                item = get_from_dynamo(url)
                if item is not None:
                    # ChatGPTへQA処理
                    exec_qa(item, query, thread_ts, channel, thread_head_ts)

                else:
                    # スレッド内でURL要約が行われていないのにメンションされたことを想定
                    print("DynamoDB item not found.")
                    chat_history = get_history(thread_head_ts)
                    qa_result = qa({"question": query, "chat_history": chat_history})
                    print("qa_result: " + qa_result["answer"])
                    
                    send_slack_message(channel, qa_result["answer"], thread_ts)
        
        except Exception as e:
            # めんどくさいので一旦全てのエラーを拾う
            print(e)
            message = "予期せぬエラーが発生しました。エンジ森にお知らせください。"
            send_slack_message(channel, message, thread_ts)
```

## お世話になった参考資料
まだ記事が少ない中大変参考になりました！大感謝🤗

https://qiita.com/hiroki_okuhata_int/items/7102bab7d96eb2574e7d

https://qiita.com/hideki/items/6206b4f02593f7b812a5

https://zenn.dev/kun432/scraps/c4763f616f0e65

https://dev.classmethod.jp/articles/search-with-openai-embeddings/

https://ict-worker.com/ai/conversationalretrievalchain.html

https://aws.amazon.com/jp/blogs/news/new-for-aws-lambda-container-image-support/

https://zenn.dev/umi_mori/books/prompt-engineer/viewer/langchain_overview

https://python.langchain.com/en/latest/