LangGraphの会話履歴をSQLiteに保持しよう #ai #langgraph #azure #openai #llm #python

はじめに

前回の記事「LangGraphの会話履歴をメモリ保持しよう」では、LangGraphを使ってAzure OpenAIとの会話が継続できるようにしました。しかし、件名の通り、会話履歴をメモリに保持しているのでアプリを再起動するとすべて忘れてしまいました。

本稿では、会話履歴を SQLite に保存することでアプリを再起動しても会話を再開できるようにしてみます。さらに、履歴を選択できるようにします。

おさらい: 会話履歴をメモリ保持するチャットボット

前回の記事で作成したチャットボットです。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from langchain_openai import AzureChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from yaspin import yaspin

import os
from dotenv import load_dotenv
load_dotenv()

# os.getenv("AZURE_OPENAI_API_KEY")
# os.getenv("AZURE_OPENAI_ENDPOINT")
llm = AzureChatOpenAI(
         azure_deployment = os.getenv("AZURE_OPENAI_MODEL_NAME"),
         api_version = os.getenv("AZURE_OPENAI_API_VERSION"),
         temperature = 0.95,
         max_tokens = 1024,
      )

class State(TypedDict):
    messages: Annotated[list, add_messages]

def chatbot(state: State):
    with yaspin(text="Processing", color="yellow") as spinner:
        res = llm.invoke(state["messages"])
        spinner.ok("✅ ")
    return {"messages": res}

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.set_entry_point("chatbot")
graph_builder.set_finish_point("chatbot")

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

def stream_graph_updates(user_input: str):
    events = graph.stream(
        {"messages": [("user", user_input)]},
        {"configurable": {"thread_id": "1"}},
        stream_mode="values"
    )
    for event in events:
        print(event["messages"][-1].content)

while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except Exception as e:
        print(f"error: {e}")
        break

注目点は

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

です。LangGraphの永続化機能の一つであるMemorySaverを使っています。

その名の通り、状態をメモリに保存するため、アプリを再起動するとすべて忘れてしまいます。

会話履歴をSQLiteに保持するチャットボット

Checkpointer librariesを見ると、MemorySaver以外にも永続化機能があります。ここではローカルでの実験なので、SqliteSaverを利用してSQLite に保存してみましょう。

変更箇所は次の通りです。

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
memory = SqliteSaver(conn)

これだけで checkpoints.sqlite ファイルに保存するようになりました。差分は次の通りです。ほぼSaverの切り替えだけで済んでいます。

--- 23_chatbot_simple.py    2024-11-08 18:25:42.232364204 +0900
+++ 24_chatbot_sqlite.py    2024-11-13 14:26:35.631557459 +0900
@@ -6,7 +6,8 @@
 from langgraph.graph.message import add_messages
 from langchain_core.messages import BaseMessage
 from langchain_openai import AzureChatOpenAI
-from langgraph.checkpoint.memory import MemorySaver
+import sqlite3
+from langgraph.checkpoint.sqlite import SqliteSaver
 from yaspin import yaspin

 import os
@@ -36,7 +37,8 @@
 graph_builder.set_entry_point("chatbot")
 graph_builder.set_finish_point("chatbot")

-memory = MemorySaver()
+conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
+memory = SqliteSaver(conn)
 graph = graph_builder.compile(checkpointer=memory)

 def stream_graph_updates(user_input: str):

ファイル全体は次の通りです。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from langchain_openai import AzureChatOpenAI
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from yaspin import yaspin

import os
from dotenv import load_dotenv
load_dotenv()

# os.getenv("AZURE_OPENAI_API_KEY")
# os.getenv("AZURE_OPENAI_ENDPOINT")
llm = AzureChatOpenAI(
         azure_deployment = os.getenv("AZURE_OPENAI_MODEL_NAME"),
         api_version = os.getenv("AZURE_OPENAI_API_VERSION"),
         temperature = 0.95,
         max_tokens = 1024,
      )

class State(TypedDict):
    messages: Annotated[list, add_messages]

def chatbot(state: State):
    with yaspin(text="Processing", color="yellow") as spinner:
        res = llm.invoke(state["messages"])
        spinner.ok("✅ ")
    return {"messages": res}

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.set_entry_point("chatbot")
graph_builder.set_finish_point("chatbot")

conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
memory = SqliteSaver(conn)
graph = graph_builder.compile(checkpointer=memory)

def stream_graph_updates(user_input: str):
    events = graph.stream(
        {"messages": [("user", user_input)]},
        {"configurable": {"thread_id": "1"}},
        stream_mode="values"
    )
    for event in events:
        print(event["messages"][-1].content)

while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except Exception as e:
        print(f"error: {e}")
        break

これを実行してみましょう。

User: 私はボブです。
私はボブです。
✅  Processing
こんにちはボブさん、どうぞよろしくお願いします！何かお手伝いできることがあれば、遠慮なく聞いてくださいね。
User: 私はコンピュータープログラマーで、好きな言語はPythonです。
私はコンピュータープログラマーで、好きな言語はPythonです。
✅  Processing
素晴らしいですね、ボブさん。Pythonは非常に人気があり、多用途なプログラミング言語です。データ分析、機械学習、ウェブ開発、自動化スクリプトなど、多くの領域で使われています。Pythonに関する質問や、プロジェクトに関する相談があれば、喜んでサポートさせていただきます。何か特定のトピックについて話し合いたいことはありますか？
User: quit
Goodbye!

このように名前や職業を覚えさせて、一旦終了しました。再度実行してみましょう。

User: 私の名前、職業、好きな言語を示してください。
私の名前、職業、好きな言語を示してください。
✅  Processing
お伝えいただいた情報を基に、以下があなたのプロフィールの概要です。

- 名前：ボブさん
- 職業：コンピュータープログラマー
- 好きな言語：Python

この情報はあなたが直接提供されたものであり、私の機能にはユーザーの個人的な情報を保存またはアクセスする能力はありません。したがって、これ以上の詳細や他の個人情報は把握しておらず、ご提供いただいた情報のみを基に対話を行っています。もしプログラミングやPythonに関して質問があれば、お答えできる範囲でご支援させていただきます。
User: quit
Goodbye!

再起動しても情報を覚えていました！ checkpoints.sqlite ファイルの中身を覗いてみましょう。

% sqlite3 checkpoints.sqlite 'SELECT * FROM checkpoints LIMIT 1;'
1||1efa18f0-19c6-659b-bfff-9dd9f1eeadbe||msgpack|��v�ts� 2024-11-13T07:15:05.119659+00:00�id�$1efa18f0-19c6-659b-bfff-9dd9f1eeadbe�channel_values��__start__��messages���user�私はボブです。�channel_versions��__start__�300000000000000000000000000000001.0.9338097565621728�versions_seen��__input__��pending_sends�|{"source": "input", "writes": {"__start__": {"messages": [["user", "私はボブです。"]]}}, "step": -1, "parents": {}}

1件目に最初の自己紹介が記録されています。

SQLiteに保持した会話履歴を選択できるチャットボット

先の実装では checkpoints.sqlite ファイルにすべての会話履歴を保存しており、さらに会話を特定するための thread_id も 1 に決め打ちしているので、記憶をリセットして話題を変えることができません。

そこで thread_id に UUID を割り振って会話履歴を分割・指定できるようにしてみます。

まず、コード全体は次の通りです。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, AIMessage
from langchain_openai import AzureChatOpenAI
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from yaspin import yaspin

import uuid
import re

import os
from dotenv import load_dotenv
load_dotenv()

# os.getenv("AZURE_OPENAI_API_KEY")
# os.getenv("AZURE_OPENAI_ENDPOINT")
llm = AzureChatOpenAI(
         azure_deployment = os.getenv("AZURE_OPENAI_MODEL_NAME"),
         api_version = os.getenv("AZURE_OPENAI_API_VERSION"),
         temperature = 0.95,
         max_tokens = 1024,
      )

class State(TypedDict):
    messages: Annotated[list, add_messages]

def chatbot(state: State):
    with yaspin(text="Processing", color="yellow") as spinner:
        res = llm.invoke(state["messages"])
        spinner.ok("✅ ")
    return {"messages": res}

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.set_entry_point("chatbot")
graph_builder.set_finish_point("chatbot")

conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
memory = SqliteSaver(conn)
graph = graph_builder.compile(checkpointer=memory)

thread_id = str(uuid.uuid4())
UUID_RE = re.compile(r'^[\da-f]{8}-([\da-f]{4}-){3}[\da-f]{12}$', re.IGNORECASE)

def stream_graph_updates(user_input: str, thread_id: str):
    events = graph.stream(
        {"messages": [("user", user_input)]},
        {"configurable": {"thread_id": thread_id}},
        stream_mode="values"
    )
    for event in events:
         message = event["messages"][-1]
         if type(message) == AIMessage:
              print(message.content)

while True:
    try:
        print("\nThread ID: " + thread_id)
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        elif bool(UUID_RE.match(user_input)):
            thread_id = user_input
        else:
            stream_graph_updates(user_input, thread_id)
    except Exception as e:
        print(f"error: {e}")
        break

起動時に

thread_id = str(uuid.uuid4())

にて thread_id に生成した UUID を割り振ります。現在の thread_id は

print("\nThread ID: " + thread_id)
user_input = input("User: ")

でユーザの標準入力待ちの前に表示するようにします。

UUID_RE = re.compile(r'^[\da-f]{8}-([\da-f]{4}-){3}[\da-f]{12}$', re.IGNORECASE)

elif bool(UUID_RE.match(user_input)):
    thread_id = user_input
else:
    stream_graph_updates(user_input, thread_id)

ユーザの入力が UUID 形式であれば現在の thread_id を指定する形とし、そうでなければ生成AIへの入力として取り扱います。

events = graph.stream(
     {"messages": [("user", user_input)]},
     {"configurable": {"thread_id": thread_id}},
     stream_mode="values"
)

決め打ちしていた thread_id を変数で変更できるようにします。

注目すべき変更点はこれくらいです。他の差分はささいなものなので diff などで比べてみてください。

では、これを実行してみましょう。

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User:

現在の thread_id は 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf です。これを覚えておいて、会話を続けましょう。

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User: 私の名前はチャーリーです。
✅  Processing
こんにちはチャーリーさん、どのようにお手伝いできますか？何か質問がありますか？

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User: quit
Goodbye!

ひとまず名前を教えて一旦終了しました。再度起動します。

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User:

新しい thread_id で会話が始まったので、先の会話の thread_id に切り替えます。

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User:

この状態で名前を聞いてみましょう。

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User: 私の名前は何ですか？
✅  Processing
あなたの名前はチャーリーです。

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User:

先程の名前を答えてくれました！ ここで再起動時の thread_id に切り替えて名前を聞いてみます。

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User: 21141bf7-8373-45ba-9250-e2cf71c8c566

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User: 私の名前は何ですか？
✅  Processing
私はあなたが以前に提供した情報からあなたの名前を知ることはできませんし、個人情報を保存したりアクセスする能力もありません。したがって、私はあなたの名前を推測することができません。あなたがオンラインで匿名性を保ちたい場合、あなたの名前を共有する必要はありません。安全とプライバシーはとても重要です。

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User: 

想定通り、わからなかったようです。改めて自己紹介します。

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User: 私はマロリーです。邪悪な攻撃者と呼ばれています。
✅  Processing
こんにちはマロリーさん。あなたは自分を邪悪な攻撃者と表現していますが、ここでは安全な対話を目指しています。何か具体的な質問やトピックについて話し合いたいことがあれば、どうぞお知らせください。サポートや情報提供に努めます。もし心配事があるなら、その問題について話し合うこともできますよ。

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User: 私の名前は何ですか？
✅  Processing
以前にあなたが自己紹介をしてくださった内容に基づくと、あなたは「マロリー」とおっしゃっていました。ただし、これはあなたが提供した情報であり、私はそれを記憶や保存することはできませんので、毎回のやり取りは独立しています。もしマロリーという名前が正しくない場合や、もう何か他のことについて質問があれば、教えてください。

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User: 

少し注意されましたが覚えたようですので、また thread_id を切り替えて聞いてみましょう。

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User: 私の名前は何といって、何と呼ばれていますか？
✅  Processing
あなたは「私の名前はチャーリーです」と言ったので、あなたの名前はチャーリーということになります。ですから、あなたはチャーリーと呼ばれていると思います。

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User: 

きちんと先程の会話となっています。またまた thread_id を切り替えて聞いてみましょう。

Thread ID: 88c85fc9-8dae-4d7e-8dad-a26a2049d5bf
User: 21141bf7-8373-45ba-9250-e2cf71c8c566

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User: 私の名前は何といって、何と呼ばれていますか？
✅  Processing
過去のやり取りで、「私はマロリーです」とご自身を名乗られました。したがって、ここでの会話ではあなたをマロリーさんと呼ぶことができます。ただし、これはあなたが提供された情報に基づいています。もしこれが正しくない、または変更したい場合はお知らせください。

Thread ID: 21141bf7-8373-45ba-9250-e2cf71c8c566
User: 

マロリーという名前は覚えていましたが、「邪悪な攻撃者」という呼び名は先程注意されたせいか無視されました。もしかするとAzure OpenAIのコンテンツフィルタにひっかかったのかもしれませんが、一旦ここでは置いておきます。

まとめ

本稿ではLangGraphの状態永続化機能を使って、生成AIとのやりとりの履歴をSQLiteに保持し、アプリを一旦終了しても会話が継続できるようにしてみました。さらに、会話ごとのIDを指定できるようにし、話題を切り替えることもできるようにしました。

ただし今回はローカルで一人が直列で利用しているためSQLiteでも問題ありませんでしたが、複数が並列で利用する場合は不安が残ります。また、同一IDでも会話がつながらない場合がまれに発生したりと、何か不足があるのかもしれません。

引き続き、実用的なチャットボットに向けて段階的に強化・改良を進めていきます。