LangChainによるChatAPI

複数のチャットサービスに対応しつつ、複数ユーザとの会話記憶を共有するAPI。
LangChainはとにかく書き方が多くて迷うので、備忘録的に。

前提

• Python 3.10
• FastAPI 0.95.0
• LangChain 0.0.132

コード

import os
import json
import datetime

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel, Field

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage

from dotenv import load_dotenv
load_dotenv()
print(os.environ["OPENAI_API_KEY"])

class Payload(BaseModel):
    service_name: str = Field(..., example="slack", description="service name")
    timestamp: str = Field(..., example="2023-04-20 16:47:22", description="timestamp")
    sender: str = Field(..., example="MyUser", description="sender name")
    reciever: str = Field(..., example="MyBot", description="reciever name")
    content: str = Field(..., example="Hello, World!", description="message content")

class ChatApi:
    def __init__(self) -> None:
        self.app = FastAPI()
        self.ai = LangchainBot()

        @self.app.post("/question", response_model=Payload)
        async def question(payload: Payload):
            response = self.ai.chat(payload)  
            return vars(response)

class LangchainBot:
    def __init__(self):
        self.log_dir = "logs"
        self.llm = ChatOpenAI(model_name="gpt-4", temperature=0.9)
        self.chat_history = []

    def chat(self, payload: Payload):
        self.load_history(payload)
        bot_name = payload.reciever
        local_chat_history = ""
        for item in self.chat_history[-100:]:
            local_chat_history += f"{item.service_name} | {item.timestamp} | {item.sender} to {item.reciever} | {item.content}\n"
        local_chat_history = local_chat_history[-8000:]

        system_template = f"""
        現在の日時を認識してください: {payload.timestamp}

        あなたの名前は{bot_name}です。
        {bot_name}は日本語を話す親切なアシスタントです。
        {bot_name}は同時に複数の相手と会話することができます。
        今回の発言者は{payload.sender}です、
        {bot_name}は過去の記憶を以下のフォーマットで保持しています。
        "[Conversation Interface] | [Timestamp] | [Sender] to [Reciever] | [Content]"

        これまでの会話の履歴:
        {local_chat_history}

        今回の{payload.sender}の発言:
        {{user_input}}

        では、返答を考えてください。日本語で返答するのを忘れずに。
        """

        self.save_history(payload)

        chat = self.llm([SystemMessage(content=system_template), HumanMessage(content=payload.content)])
       
        now = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))).strftime("%Y-%m-%d %H:%M:%S")
        response = Payload(service_name=payload.service_name, timestamp=now, sender=bot_name, reciever=payload.sender, content=chat.content)
        self.save_history(response)

        return response

    def load_history(self, payload: Payload):
        chat_log_file_name = f"{self.log_dir}/chat_history_{payload.reciever}.json"
        if os.path.exists(chat_log_file_name):
            with open(chat_log_file_name, "r", encoding="utf-8") as f:
                json_payload = json.load(f)
            for item in json_payload["history"]:
                self.chat_history.append(Payload(**item))

    def save_history(self, payload: Payload):
        chat_log_file_name = f"{self.log_dir}/chat_history_{payload.sender}.json"
        self.chat_history.append(payload)
        _json_payload = []
        for item in self.chat_history:
            _json_payload.append(item.dict())
        json_payload = {"history": _json_payload}
        with open(chat_log_file_name, "w", encoding="utf-8") as f:
            json.dump(json_payload, f, ensure_ascii=False, indent=4)

if __name__ == "__main__":
    api = ChatApi()
    app = api.app
    uvicorn.run("__main__:app", host="0.0.0.0", port=9301, reload=False)

# 以下に差し替えると、CUIでテストできる。
'''
if __name__ == "__main__":
    bot = LangchainBot()
    service_name = "CUI
    now = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))).strftime("%Y-%m-%d %H:%M:%S")
    user_name = "MyUser"
    bot_name = "MyBot"

    while True:
        user_input = input(f"{user_name}: ")
        payload = Payload(service_name=service_name, timestamp=now, sender=user_name, reciever=bot_name, content=user_input)
        result = bot.chat(payload)
        print(f"bot: {result}")
'''

補足

LangChainのmemoryを使わない理由

どうしてもタイムスタンプが欲しかったのだが、LangChain側で保持させる手段が見つからず...
力づくの対応をしてしまった。長期記憶の保存もjsonファイルの読み書きだけの手抜き仕様。
長期記憶に関しては、SQLやPineConeに差し替え可能と思う。

タイムスタンプの効果

• 現在時刻と、過去の会話の時刻を認識するようになった！
• 昨日の何時ごろのアレ、で話が通じるようになった！
• 複数の話者の区別がつきやすくなった（たぶん）

Payloadについて

ChatBot側でPayloadのスキーマに沿ったjsonをPOSTする想定。レスポンスも同一スキーマで返す。
長期記憶も同じスキーマのjsonなので、後々流用しやすいかな、と。

chat_historyのファイル名

ボット名別に記憶を振り分けている。ChatBot側でBot名をpayload(通常recieverになるはず)に入れてくれれば、複数のボット（ペルソナ）に対応した共通APIになれるかも。

レスポンス返す時は、senderがボット名になる。loadとsaveで、recieverとsenderが入れ替わってるのはこの為。

今気づいたけど、複数のボットを運用するならsystem_templateも分けなきゃダメだわ。system_templeteまでpayloadに含めてもいいけど...どうしようかな。