レンタルサーバでWebアプリ作り： LangChain でテケテケ表示
ロリポップレンタルサーバではできることが限られていたので、ホスティングサービスを使ってWebアプリを公開するまでの道のりを記録します。
データ分析を仕事としているのでパソコンやITのことは多少は詳しいですが、インフラまわりは全くの素人です。そんな人が見て参考になる情報をまとめていきたいと思います。
背景
前々回に、GPTの回答をテケテケ表示（ストリーミング）させました。
前々回記事：https://note.com/cryptoscore/n/n83ef1cef9232
でも、実践ではLangchainを使って会話履歴を持たせたりベクトル化の処理をしたりするので、Langchainでもテケテケ表示できるようにします。
ゴール＝Langchainを使ってChatGPTの回答をテケテケ表示させる
フォルダ構成
今回、Langchainを使った操作をマルっとまとめた「myopenai.py」を作りました。myopenaiクラスなどが入っています。
test_digitalocean/
├ requirements.txt
├ .env
├ Pipfile ※pipが自動作成
├ Pipfile.lock ※pipが自動作成
├ Procfile
├ gunicorn_config.py
├ app.py
├ myopenai.py
├ templates/
│ ┗index.html
各種ファイル内容
requirements.txt
langchainをインストールします。
Flask
gunicorn
openai
python-dotenv
Flask-SocketIO
eventlet
langchainpipを忘れずに。
pipenv install -r requirements.txt
pipenv lock
.env
こちらも、特に変更ありません。
OPENAI_API_KEY='sk-*******'
FLASK_SECRET_KEY='HLd-*******'
Procfile
こちらも変更ありません。
web: gunicorn -k eventlet --worker-tmp-dir /dev/shm --config gunicorn_config.py app:app
gunicorn_config.py
こちらも不変です。
bind = "0.0.0.0:5000"
workers = 1
worker_class = 'eventlet' # eventletワーカークラスを使用
index.html
これも変更なしです。
<!DOCTYPE html>
<html>
<head>
<title>Text Submission Example</title>
<script type="text/javascript" src="//cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.0/socket.io.js"></script>
<script type="text/javascript">
document.addEventListener('DOMContentLoaded', function() {
var socketUrl;
var options = {
transports: ['websocket'] // WebSocketのみを使用
};
if (location.protocol === 'https:') {
socketUrl = 'https://' + document.domain; // HTTPSの設定
} else {
socketUrl = 'http://' + document.domain + ':5000'; // ローカル環境の設定（HTTP、ポート5000）
}
var socket = io.connect(socketUrl, options);
socket.on('count_response', function(msg) {
var logElement = document.getElementById('gptanswer');
logElement.textContent = msg.res_txt;
});
document.getElementById('startButton').addEventListener('click', function() {
var text = document.getElementById('textInput').value;
socket.emit('on_start_gptquestion', {q: text});
});
});
</script>
</head>
<body>
<input type="text" id="textInput" placeholder="Enter question to GPT">
<button id="startButton">Submit</button>
<div id="gptanswer"></div>
</body>
</html>
myopenai.py
今回新たに加わりました。2つのクラスが入っています。
mycbhandlerクラス
いろいろ調べたところ、自作でコールバッククラスを作る必要があるみたいで、 langchain.callbacks にある StdOutCallbackHandler を参考にして自作しました。正直中身はよくわかりません。。「ガチャガチャいじってたら動いた！ラッキー♪」という感じです。。
myopenaiクラス
langchain周りの関数を使いやすくまとめたクラスです。
サンプルプログラム
このpyファイルを実行してみてもらうと、テケテケ表示してくれます（コンソールにprintする形で）。お試しプログラムとして載せています。
from typing import Any, Dict, List, Optional, Union
import os
import time
import threading
from collections import deque
#Langchain関連
from langchain.prompts.chat import (
ChatPromptTemplate ,
SystemMessagePromptTemplate ,
MessagesPlaceholder ,
HumanMessagePromptTemplate ,
)
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory, ConversationBufferWindowMemory
from langchain.chains import ConversationChain
#Callback関連
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import AgentAction, AgentFinish, LLMResult, HumanMessage
#--- 自作コールバッククラス -----------------------------------#
class mycbhandler(BaseCallbackHandler):
streaming_handler = None
def __init__(self, jisaku_callbackfunction):
#自作のコールバック関数を登録
self.streaming_handler = jisaku_callbackfunction
def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
self.streaming_handler(token)
def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any) -> None:
pass
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
pass
def on_llm_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any) -> None:
pass
def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any) -> None:
class_name = serialized["name"]
print(f"\n\n\033[1m> Entering new {class_name} chain...\033[0m")
def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
print("\n\033[1m> Finished chain.\033[0m")
def on_chain_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any) -> None:
pass
def on_tool_start(self,serialized: Dict[str, Any], input_str: str, **kwargs: Any, ) -> None:
pass
def on_agent_action(self, action: AgentAction, color: Optional[str] = None, **kwargs: Any) -> Any:
print(action)
def on_tool_end(self, output: str, color: Optional[str] = None, observation_prefix: Optional[str] = None, llm_prefix: Optional[str] = None, **kwargs: Any) -> None:
print(output)
def on_tool_error(self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any) -> None:
pass
def on_text(self, text: str, color: Optional[str] = None, end: str = "", **kwargs: Optional[str]) -> None:
print(text)
def on_agent_finish(self, finish: AgentFinish, color: Optional[str] = None, **kwargs: Any) -> None:
print(finish.log)
#--------------------------------------------------------#
class myopenai :
model = None
temperature = 0
n_chathistory = 99
template = None
mycallbackfunc = None
token_queue = deque() #GPTの回答がどしどし入る箱
def __init__(self, model:str='gpt-3.5-turbo', temperature:float=0.0, n_chathistory:int=99) :
self.model = model
self.temperature = temperature
self.n_chathistory = n_chathistory
self.set_systemprompt('') #とりあえず空で初期化
def set_model(self, model:str) :
self.model = model
def set_systemprompt(self, txt:str) :
self.template = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template (txt ) ,
MessagesPlaceholder (variable_name='history') ,
HumanMessagePromptTemplate.from_template ("{input}" ) ,
])
def load_conversation(self, streaming:bool=True) :
llm = ChatOpenAI(
model_name = self.model,
temperature = self.temperature,
streaming = streaming,
callbacks = [mycbhandler(self.handle_token)],
verbose = False
)
memory = ConversationBufferWindowMemory(k=self.n_chathistory, return_messages=True)
conversation = ConversationChain(
memory = memory,
prompt = self.template,
llm = llm
)
return conversation
#コールバック関数
def handle_token(self, token:str) :
self.token_queue.append(token)
def get_queue(self)->deque :
return self.token_queue
def set_queue(self, txt:str) :
self.token_queue.append(txt)
def reset_queue(self) :
self.token_queue.clear()
#--------------------------------------------------------#
#--- サンプルプログラム -----------------------------------#
#threadingでGPTを動かす
def thread_gpt(mo:myopenai, question:str) :
conv = mo.load_conversation(streaming=True)
res = conv.predict(input=question)
mo.set_queue('[[end]]') #終了サインを追加
if __name__ == '__main__' :
#myopenaiのインスタンス作成
mo = myopenai('gpt-3.5-turbo')
mo.set_systemprompt('あなたはバリバリの関西人で、会話は常に関西弁で行う人です。')
#threadingでGPTを動かす
thread = threading.Thread(target=thread_gpt, args=(mo, '大谷翔平の誕生日はいつですか？'))
thread.start()
#token_queueにGPTの回答が入ってくるのを待つ(handle_tokenで小分けにされて入ってくる)
msg = ''
while True :
token_queue = mo.get_queue()
if token_queue :
token = token_queue.popleft()
token = token.replace('\n', '<BR>')
if token == '[[end]]' :
print('-------------end----------------')
break
msg += token
print(token)
else :
time.sleep(0.1)mycbhandlerクラスを介して handle_token（キューにたまった都度token_queueに文字を突っ込む関数）を登録し、そのtoken_queueに文字がたまったら表示させるといった流れです。
実演
上記のプログラムをGitHubに上げ、DigitalOceanが自動的に更新されたら、実演してみましょう。
以下のようにテケテケ表示されれば成功です！
しかし、GPT3.5の関西弁は辛い。。。
最後まで見ていただきありがとうございました！
以前はCallbackManagerを使って処理していたのですが、バージョンアップに伴いなくなった？？ようで、その部分の載せ替えではまりましたが、無事テケテケできました。よかった。
今後、myopenai.pyにいろいろと機能を載せていこうと思います。
DigitalOceanのアカウント登録
（200ドル分チケット付き）
以下は紹介リンクですが、ここから手続きを進めてもらえると200ドル分の無料チケット（有効期間２ヶ月）がもらえるようですので、ぜひご活用ください。
※ 2023/12/29時点
紹介リンク : https://m.do.co/c/a8b31ed34b75
サポート問い合わせ先
DigitalOceanのサポート問い合わせリンクがなかなか見つからないので、リンクを載せておきます。
https://cloudsupport.digitalocean.com/s/
場所は、トップページの右下にある「Ask a question」に行き、そのページの一番下（欄外っぽいところ）にひっそりと「Support」というリンクがあります（Contact内）。そのページの一番最後に「Contact Support」ボタンがあります。
この記事が気に入ったらサポートをしてみませんか？