社内文書回答のSlackアプリをLangChain(LCEL)とChatGPTで実装

ChatGPT/LangChainによるチャットシステム構築［実践］入門の「第8章 社内文書に答えるSlackアプリの実装」をLangChain0.2 のLCEL(LangChain Expression Language)で実装し直したのでその記録です。
GitHub yoshidashingo/langchain-bookにあるのが、元のソースコードでLangChain0.0.292のときなので、大きく変わっています。ある程度調べるのに時間使ったのでここに記録しておきます。

実装機能

以下の機能を実装しています。

• PDF文書の分散表現化: OpenAI/text-embedding-ada-002(text-embedding-3系でないのに深い理由なし)
• PDF文書の分散表現保存: Pinecone
• 会話履歴の保存: Momento Cache
• Chat画面: Slack

チャット画面とサンプル

こんなチャットサンプル

チャットの内部フロー

LangSmithで見ています。

1回目のチャット

大きくは、insert_history(会話履歴の挿入)とretrieval_chain(文書取得chain)で構成しています。

No.	Chain	内容	詳細
1	insert_history	履歴挿入
1.1	load_history	履歴ロード
2	retrieval_chain	RAGのChain
2.1	retrieve_documents	文書取得
2.1.1	Retriever	文書取得
2.2	stuff_documents_chain	文書結合のChain
2.2.1	format_inputs	入力のフォーマット
2.2.1.1	format_docs	文書のフォーマット
2.2.2	ChatOpenAI	OpenAI呼出

2回目のチャット

Chainそのものは1回目と同じです。会話履歴から最終質問生成のために1度OpenAIを呼んでいるのが特徴的。

No.	Chain	内容	詳細
1	insert_history	履歴挿入
1.1	load_history	履歴ロード
2	retrieval_chain	RAGのChain
2.1	retrieve_documents	文書取得
2.1.1	ChatOpenAI	OpenAI呼出
2.1.2	Retriever	文書取得
2.2	stuff_documents_chain	文書結合のChain
2.2.1	format_inputs	入力のフォーマット
2.2.1.1	format_docs	文書のフォーマット
2.2.2	ChatOpenAI	OpenAI呼出

プログラム

プログラムは以下の2つで構成。

1. ファイルをベクトルストアに挿入: add_document.py
2. RAG: app.py

前提

アーキテクチャー

ChatGPT/LangChainによるチャットシステム構築［実践］入門に準拠したアーキテクチャーにしています。基本的にすべて無料枠で使えています。
ここに書かれているサービスの使い方は、基本的に当記事に記載していません。私はChatGPT/LangChainによるチャットシステム構築［実践］入門を見ながらやりましたが、ググればすぐにできるものも多いです。経験ないとSlackとAWS Lambdaが少し苦戦するかも。

Type	環境	備考
IDE	AWS Cloud9
API	AWS Lambda	サーバレスいいですね
Chat画面	Slack	久々に使いました
ベクトルストア	Pinecone	初体験でしたが使いやすかったです
キャッシュサービス	Momento Cache	初体験でしたが使いやすかったです
Embeddingモデル	OpenAI/text-embedding-ada-002	text-embedding-3系でないのに深い理由なし
Chatモデル	OpenAI/GPT-3.5 Turbo	Chainの検証なので安いモデル

AWS Cloud9でハマったエラー
add_document.py を動かすときにTesseractを入れろとエラーが出て、pip installしようとすると重いのでメモリ不足で何度もエラー発生。古いUnstructuredPDFLoaderを使っていたのが原因だったようで、最新のlangchain_communityのUnstructuredPDFLoaderを呼び出せばTesseractも不要でした。

Pythonパッケージ

過不足あるかもしれませんが、だいたいあっているはず

Package	Version	備考
pillow_heif	0.16.0	確かPDF読み込む際の前提
opencv-python	4.9.0.80	確かPDF読み込む際の前提
unstructured	0.14.0
langchain	0.2.0
langchain-community	0.2.0
langchain-openai	0.1.7
langchain-pinecone	0.1.1
python-dotenv	1.0.1
pinecone-client	4.1.0
slack_bolt	1.18.1
momento	1.20.1
boto3	1.34.104	何かで必要と言われ・・・

.envファイル

dotenvパッケージで読み込む環境変数のファイル。Token系は記載消しています。

.env

SLACK_SIGNING_SECRET=
SLACK_BOT_TOKEN=
SLACK_APP_TOKEN=
OPENAI_API_KEY=
OPENAI_API_MODEL=gpt-3.5-turbo-16k-0613
OPENAI_API_TEMPERATURE=0.5
MOMENTO_AUTH_TOKEN=
MOMENTO_CACHE=langchain-book
MOMENTO_TTL=1
MOMENTO_API_KEY=
PINECONE_API_KEY=
PINECONE_INDEX=langchain-book
PINECONE_ENV=gcp-starter
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=

プログラム

ファイルをベクトルストアに挿入

Pythonプログラム

add_document.py

import logging
import os
import sys

from pinecone import Pinecone, ServerlessSpec
from dotenv import load_dotenv
from langchain_community.document_loaders import UnstructuredPDFLoader
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain_pinecone import PineconeVectorStore

load_dotenv()

logging.basicConfig(
    format="%(asctime)s [%(levelname)s] %(message)s", level=logging.INFO
)

logger = logging.getLogger(__name__)

def initialize_vectorstore():
    Pinecone(api_key=os.environ.get("PINECONE_API_KEY"))
    index_name = os.environ["PINECONE_INDEX"]
    embeddings = OpenAIEmbeddings()
    return PineconeVectorStore.from_existing_index(index_name, embeddings)
    

if __name__ == "__main__":
    file_path = sys.argv[1] # 引数を取得
    loader = UnstructuredPDFLoader(file_path)
    raw_docs = loader.load()
    logger.info("Loaded %d documents", len(raw_docs))
    
    text_splitter = CharacterTextSplitter(chunk_size=300, chunk_overlap=30)
    docs = text_splitter.split_documents(raw_docs)
    logger.info("Split %d documents", len(docs))
    
    vectorestore = initialize_vectorstore()
    vectorestore.add_documents(docs)

プログラム実行

元ファイルは生成AIの利用ガイドラインから「生成AIの利用ガイドライン【条項のみ】(第1.1版, 2023年10月公開)」をダウンロードして、PDFファイルai-guideline.pdfで保存。
で、Pythonプログラムを実行。引数にファイル名を指定。

プログラム実行

python add_document.py ai-guideline.pdf
2024-05-26 08:12:45,587 [INFO] pikepdf C++ to Python logger bridge initialized
2024-05-26 08:12:47,422 [INFO] Loaded 1 documents
2024-05-26 08:12:47,424 [INFO] Split 16 documents
2024-05-26 08:12:49,277 [INFO] HTTP Request: POST https://api.openai.com/v1/embeddings "HTTP/1.1 200 OK"

Pineconeに16エントリが挿入。

RAG

Pythonプログラム

app.py

# slack_bolt python-dotenv langchain-openai langchain(with langchain-core) momento boto3
from add_document import initialize_vectorstore

from datetime import timedelta
import json
import logging
import os
import re
import time
from typing import Any

from dotenv import load_dotenv
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.schema import HumanMessage, LLMResult, SystemMessage
from langchain.globals import set_debug
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import MomentoChatMessageHistory
from langchain_core.callbacks.base import BaseCallbackHandler
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from slack_bolt.adapter.aws_lambda import SlackRequestHandler

CHAT_UPDATE_INTERVAL_SEC = 1

load_dotenv()
set_debug(True)

SlackRequestHandler.clear_all_log_handlers()
logging.basicConfig(
    format="%(asctime)s [%(levelname)s] %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)


# ボットトークンを使ってアプリを初期化します
app = App(
    signing_secret=os.environ["SLACK_SIGNING_SECRET"],
    token=os.environ.get("SLACK_BOT_TOKEN"),
    process_before_response=True,
    )


class SlackStreamingCallbackHandler(BaseCallbackHandler):
    last_send_time = time.time()
    message = ""
    
    def __init__(self, channel, ts):
        self.channel = channel
        self.ts = ts
        self.interval = CHAT_UPDATE_INTERVAL_SEC
        
        # 投稿を更新した累計回数カウンタ
        self.update_count = 0


    def on_llm_new_token(self, token: str, **kwargs) -> None:
        self.message += token
        
        now = time.time()
        if now - self.last_send_time > CHAT_UPDATE_INTERVAL_SEC:
            app.client.chat_update(
                channel=self.channel,
                ts=self.ts,
                text=f"{self.message}..."
                )
            self.last_send_time = now
            self.update_count += 1

            # update_countが現在の更新間隔X10より多くなるたびに更新間隔を2倍にする
            if self.update_count / 10 > self.interval:
                self.interval = self.interval * 2
                

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
        message_context = "OpenAI APIで生成される情報は不正確または不適切な場合がありますが、当社の見解を述べるものではありません。"
        message_blocks = [
            {"type": "section", "text": {"type": "mrkdwn", "text": self.message}},
            {"type": "divider"},
            {
                "type": "context",
                "elements": [{"type": "mrkdwn", "text": message_context}],
            },
        ]
        
        # print(f"{self.message=}")
        app.client.chat_update(
            channel=self.channel,
            ts=self.ts, 
            text=self.message,
            blocks=message_blocks,
            )


#@app.event("app_mention")
def handle_mention(event, say):
    channel=event["channel"]
    thread_ts = event["ts"]
    message = re.sub("<@.*>", "", event["text"])
    
    # 投稿のキー(=Momentoキー)：初回=event["ts"],2回目以降=event["thread_ts"]
    id_ts = event["ts"]
    if "thread_ts" in event:
        id_ts = event["thread_ts"]

    result = say("\n\nTyping...", thread_ts=thread_ts)
    ts = result["ts"]
    
    # ここで渡しているid_tsに接頭辞「message_store:」を付加することでMomentoのキーとなる
    print(f"{id_ts=}")

    vectorstore = initialize_vectorstore()

    callback = SlackStreamingCallbackHandler(channel=channel, ts=ts)
    
    llm = ChatOpenAI(
        model_name=os.environ["OPENAI_API_MODEL"],
        temperature=os.environ["OPENAI_API_TEMPERATURE"],
        streaming=True,
        callbacks=[callback],
        )

    ### Contextualize question ###
    contextualize_q_system_prompt = (
        "Given a chat history and the latest user question "
        "which might reference context in the chat history, "
        "formulate a standalone question which can be understood "
        "without the chat history. Do NOT answer the question, "
        "just reformulate it if needed and otherwise return it as is."
    )
    contextualize_q_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", contextualize_q_system_prompt),
            MessagesPlaceholder("chat_history"),
            ("human", "{input}"),
        ]
    )
    
    history_aware_retriever = create_history_aware_retriever(
        llm, vectorstore.as_retriever(), contextualize_q_prompt
    )

    system_prompt = (
    "You are an assistant for question-answering tasks. "
    "Use the following pieces of retrieved context to answer "
    "the question. If you don't know the answer, say that you "
    "don't know. Use three sentences maximum and keep the "
    "answer concise."
    "\n\n"
    "{context}"
    )
    qa_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", system_prompt),
            MessagesPlaceholder("chat_history"),
            ("human", "{input}"),
        ]
    )
    question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)

    rag_chain = create_retrieval_chain(
        history_aware_retriever, 
        question_answer_chain
        )

    history = MomentoChatMessageHistory.from_client_params(
        id_ts,
        os.environ["MOMENTO_CACHE"],
        timedelta(hours=int(os.environ["MOMENTO_TTL"])),  # 1時間でキャッシュ失効
        )

    conversational_rag_chain = RunnableWithMessageHistory(
        rag_chain,
        lambda session_id: history,
        input_messages_key="input",
        history_messages_key="chat_history",
        output_messages_key="answer",
    )
    response = conversational_rag_chain.invoke(
        {"input": message},
        config={
            "configurable": {"session_id": ts}
        },
    )
    print(response['answer'])


def just_ack(ack):
    ack()

    
app.event("app_mention")(ack=just_ack, lazy=[handle_mention])


def handler(event, context):
    logger.info("handler called")
    header = event["headers"]
    logger.info(json.dumps(header))
    
    if "x-slack-retry-num" in header:
        logger.info("SKIP > x-slack-retry-num: %s", header["x-slack-retry-num"])
        return 200
        
    # AWS Lambda 環境のリクエスト情報を app が処理できるよう変換してくれるアダプター
    slack_handler = SlackRequestHandler(app=app)
    # 応答はそのまま AWS Lambda の戻り値として返せます
    return slack_handler.handle(event, context)


# ソケットモードハンドラーを使ってアプリを起動します
if __name__ == "__main__":
    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()

プログラム実行

プログラム実行することで、待機状態になります。今回はAWS Lambdaから実行していません。

プログラム実行

python app.py
2024-05-26 09:11:45,291 [INFO] A new session has been established (session id: dd509c49-f308-4d8b-b804-3b7cdf11da55)
2024-05-26 09:11:45,292 [INFO] ⚡ Bolt app is running!

SlackからChatをすると以下のようにターミナルにLogが出力されます。set_debug(True)にしているのでLogの量が多いです。

2024-05-26 09:11:45,379 [INFO] Starting to receive messages from a new connection (session id: dd509c49-f308-4d8b-b804-3b7cdf11da55)
id_ts='1715405674.956279'
2024-05-26 09:12:10,879 [INFO] Creating cache with name: langchain-book
[chain/start] [chain:RunnableWithMessageHistory] Entering Chain run with input:
{
  "input": " 生成AIに著作物の内容を入力すること自体は問題ありませんか？一言で回答してください。"
}

後略

Momento Cacheに会話履歴が追加されます。Keyはmessage_store:を接頭辞にして、MomentoChatMessageHistory.from_client_paramsに渡すsession_idの値(このプログラムでは変数id_tsの値)を付加してConsoleからデータを確認できます。有効期限が1時間後というこも確認できます。