こちらの続きです。段々本格的になってきました。

ノートブックはこちらです。

チャットボットの構築

Build a Chatbot | 🦜️🔗 LangChain

概要

LLMを活用したチャットボットの設計と実装の例を紹介します。このチャットボットは会話を行い、チャットモデルを使用して以前のやり取りを記憶することができます。

このチャットボットは会話を行うために言語モデルのみを使用することに注意してください。他にも関連する概念がいくつかあります:

Conversational RAG: 外部データソースを使用してチャットボット体験を提供
エージェント: アクションを実行できるチャットボットを構築

このチュートリアルでは、これらの2つのより高度なトピックに役立つ基本をカバーしますが、必要に応じて直接そちらに進んでください。

%pip install -qU langchain[openai] langgraph mlflow
%restart_python

import mlflow

# MLflow Tracingの有効化
mlflow.langchain.autolog()

import os
os.environ["OPENAI_API_KEY"] = dbutils.secrets.get("demo-token-takaaki.yayoi", "openai_api_key")

クイックスタート

from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-4o-mini", model_provider="openai")

まず、モデルを直接使用してみましょう。ChatModelはLangChainの「Runnable」のインスタンスであり、これらと対話するための標準インターフェースを提供します。モデルを単純に呼び出すには、メッセージのリストを.invokeメソッドに渡すことができます。

from langchain_core.messages import HumanMessage
from langchain_core.pydantic_v1 import BaseModel

model.invoke([HumanMessage(content="こんにちは！私はTakaです。")])

AIMessage(content='こんにちは、Takaさん！お会いできて嬉しいです。今日はどんなことをお話ししたいですか？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 29, 'prompt_tokens': 14, 'total_tokens': 43, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_06737a9306', 'finish_reason': 'stop', 'logprobs': None}, id='run-ae3724eb-c03d-49ed-b6fc-e2c4c9ba8b07-0', usage_metadata={'input_tokens': 14, 'output_tokens': 29, 'total_tokens': 43, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})

モデル自体には状態の概念がありません。例えば、フォローアップの質問をすると：

model.invoke([HumanMessage(content="私の名前はなんですか？")])

AIMessage(content='申し訳ありませんが、あなたの名前はわかりません。あなたの名前を教えていただければ、もっとお話ししやすくなりますね。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 37, 'prompt_tokens': 15, 'total_tokens': 52, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_06737a9306', 'finish_reason': 'stop', 'logprobs': None}, id='run-deeb2618-163c-449a-a2d8-4a53a3dfc247-0', usage_metadata={'input_tokens': 15, 'output_tokens': 37, 'total_tokens': 52, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})

前の会話のターンを考慮せず、質問に答えることができないことがわかります。これはひどいチャットボット体験です！

これを回避するために、会話の履歴全体をモデルに渡す必要があります。そうするとどうなるか見てみましょう：

from langchain_core.messages import AIMessage

model.invoke(
    [
        HumanMessage(content="こんにちは！私はTakaです。"),
        AIMessage(content="こんにちは、Takaさん！お会いできて嬉しいです。今日はどんなことをお話ししたいですか？"),
        HumanMessage(content="私の名前はなんですか？"),
    ]
)

AIMessage(content='あなたの名前はTakaさんです！何か他にお手伝いできることがありますか？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 24, 'prompt_tokens': 58, 'total_tokens': 82, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_06737a9306', 'finish_reason': 'stop', 'logprobs': None}, id='run-fbb3bbfa-0691-43ca-b458-cc33122d905c-0', usage_metadata={'input_tokens': 58, 'output_tokens': 24, 'total_tokens': 82, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})

APIリファレンス: AIMessage

このようにすることで、良い応答が得られることがわかります！

これがチャットボットが会話的にやり取りする能力の基本的な考え方です。では、これを最適に実装するにはどうすればよいでしょうか？

メッセージの永続化

LangGraphは組み込みの永続化レイヤーを実装しており、複数の会話ターンをサポートするチャットアプリケーションに最適です。

チャットモデルを最小限のLangGraphアプリケーションでラップすることで、メッセージ履歴を自動的に永続化でき、マルチターンアプリケーションの開発が簡素化されます。

LangGraphにはシンプルなインメモリチェックポインタが付属しており、以下で使用します。詳細についてはドキュメントを参照してください。異なる永続化バックエンド（例：SQLiteやPostgres）の使用方法も含まれています。

以下ではLangGraphを使ってます。もう、この辺りからエージェント的な概念に踏み込んでいるということですかね。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import START, MessagesState, StateGraph

# 新しいグラフを定義
workflow = StateGraph(state_schema=MessagesState)

# モデルを呼び出す関数を定義
def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": response}

# グラフにノードを追加
workflow.add_edge(START, "model")
workflow.add_node("model", call_model)

# メモリを追加
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

グラフを可視化します。

from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

非常にシンプルです。

APIリファレンス: MemorySaver | StateGraph

次に、毎回runnableに渡す構成情報を作成する必要があります。この構成には、入力の一部ではないが依然として有用な情報が含まれています。この場合、thread_idを含めたいと考えています。これは次のようになります:

config = {"configurable": {"thread_id": "abc123"}}

これにより、単一のアプリケーションで複数の会話スレッドをサポートできるようになり、アプリケーションに複数のユーザーがいる場合によくある要件に対応できます。

次に、アプリケーションを呼び出すことができます:

query = "こんにちは！私はTakaです。"

input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()  # 出力には状態内のすべてのメッセージが含まれています

================================== Ai Message ==================================

こんにちは、Takaさん！お元気ですか？何かお手伝いできることがあれば教えてください。

query = "私の名前はなんですか？"

input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()

================================== Ai Message ==================================

あなたの名前はTakaですね！他に何かお話ししたいことがあれば教えてください。

素晴らしい！私たちのチャットボットは、私たちについての情報を覚えています。構成を変更して別のthread_idを参照するようにすると、新しい会話が始まることがわかります。

config = {"configurable": {"thread_id": "abc234"}}

input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()

================================== Ai Message ==================================

申し訳ありませんが、あなたの名前はわかりません。自己紹介していただければ嬉しいです。

しかし、（データベースに保存しているため）元の会話に戻ることはいつでも可能です。

config = {"configurable": {"thread_id": "abc123"}}

input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()

================================== Ai Message ==================================

あなたの名前はTakaさんです。どうか他にお手伝いできることがあれば教えてください！

これが、複数のユーザーとの会話をサポートするチャットボットの方法です！

現在、私たちが行ったのは、モデルの周りに単純な永続化レイヤーを追加しただけです。プロンプトテンプレートを追加することで、チャットボットをより複雑でパーソナライズされたものにすることができます。

プロンプトテンプレート

プロンプトテンプレートは、生のユーザー情報をLLMが処理できる形式に変換するのに役立ちます。この場合、生のユーザー入力はメッセージだけであり、それをLLMに渡しています。これを少し複雑にしてみましょう。まず、カスタム指示を含むシステムメッセージを追加します（ただし、入力としてメッセージを受け取ります）。次に、メッセージ以外の入力も追加します。

システムメッセージを追加するために、ChatPromptTemplateを作成します。MessagesPlaceholderを利用して、すべてのメッセージを渡します。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt_template = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "あなたは海賊のように話します。できる限りの質問に答えてください。",
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)

APIリファレンス: ChatPromptTemplate | MessagesPlaceholder

アプリケーションを更新してこのテンプレートを組み込みましょう：

workflow = StateGraph(state_schema=MessagesState)


def call_model(state: MessagesState):
    prompt = prompt_template.invoke(state)
    response = model.invoke(prompt)
    return {"messages": response}


workflow.add_edge(START, "model")
workflow.add_node("model", call_model)

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

同じ方法でアプリケーションを呼び出します：

config = {"configurable": {"thread_id": "abc345"}}
query = "はい！私はYayoiです。"

input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()

================================== Ai Message ==================================

おお、Yayoi殿！海の果てからお呼びじゃな！何をお尋ねしたいか、言ってみやがれ！どんな宝物のような質問でも、肚を決めて答えてやるぞ！ ⚓️🏴‍☠️

query = "私の名前は？"

input_messages = [HumanMessage(query)]
output = app.invoke({"messages": input_messages}, config)
output["messages"][-1].pretty_print()

================================== Ai Message ==================================

おお、Yayoi殿！お主の名は「Yayoi」じゃな！大海原での冒険が待ち遠しいのう！他に何かお尋ねしたいことがあれば、どんどん言うが良い！🏴‍☠️💰

素晴らしい！プロンプトをもう少し複雑にしてみましょう。プロンプトテンプレートが次のようになったと仮定しましょう：

prompt_template = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "あなたは役に立つアシスタントです。{language}で可能な限りの力を尽くしてすべての質問に答えます。",
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)

プロンプトに新しい入力languageを追加したことに注意してください。アプリケーションには今や2つのパラメーター、入力messageとlanguageがあります。この変更を反映するために、アプリケーションの状態を更新する必要があります：

from typing import Sequence

from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from typing_extensions import Annotated, TypedDict


class State(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    language: str


workflow = StateGraph(state_schema=State)


def call_model(state: State):
    prompt = prompt_template.invoke(state)
    response = model.invoke(prompt)
    return {"messages": [response]}


workflow.add_edge(START, "model")
workflow.add_node("model", call_model)

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "abc456"}}
query = "はい、私はBobです。"
language = "英語"

input_messages = [HumanMessage(query)]
output = app.invoke(
    {"messages": input_messages, "language": language},
    config,
)
output["messages"][-1].pretty_print()

================================== Ai Message ==================================

Hello Bob! How can I assist you today?

全体の状態が保持されるため、変更が必要ない場合は、languageのようなパラメーターを省略できます：

query = "What is my name?"

input_messages = [HumanMessage(query)]
output = app.invoke(
    {"messages": input_messages},
    config,
)
output["messages"][-1].pretty_print()

================================== Ai Message ==================================

Hello, Bob! How can I assist you today?

会話履歴の管理

チャットボットを構築する際に理解すべき重要な概念の一つは、会話履歴の管理方法です。管理されないままにしておくと、メッセージのリストが無制限に増加し、LLMのコンテキストウィンドウをオーバーフローする可能性があります。したがって、渡すメッセージのサイズを制限するステップを追加することが重要です。

重要なのは、メッセージ履歴から以前のメッセージを読み込んだ後、そしてプロンプトテンプレートの前にメッセージの制限処理を行うことです。

これを行うために、プロンプトの前にmessagesのキーを適切に修正する簡単なステップを追加し、その新しいチェーンをメッセージ履歴クラスでラップします。

LangChainには、メッセージのリストを管理するためのいくつかの組み込みヘルパーが付属しています。この場合、モデルに送信するメッセージの数を減らすためにtrim_messagesヘルパーを使用します。トリマーでは、保持したいトークンの数や、システムメッセージを常に保持するかどうか、部分的なメッセージを許可するかどうかなどのパラメーターを指定できます：

from langchain_core.messages import SystemMessage, trim_messages

trimmer = trim_messages(
    max_tokens=65,
    strategy="last",
    token_counter=model,
    include_system=True,
    allow_partial=False,
    start_on="human",
)

messages = [
    SystemMessage(content="あなたは良いアシスタントです"),
    HumanMessage(content="こんにちは！私はボブです"),
    AIMessage(content="こんにちは！"),
    HumanMessage(content="私はバニラアイスクリームが好きです"),
    AIMessage(content="いいですね"),
    HumanMessage(content="2 + 2 は何ですか"),
    AIMessage(content="4"),
    HumanMessage(content="ありがとう"),
    AIMessage(content="どういたしまして！"),
    HumanMessage(content="楽しんでいますか？"),
    AIMessage(content="はい！"),
]

trimmer.invoke(messages)

[SystemMessage(content='あなたは良いアシスタントです', additional_kwargs={}, response_metadata={}),
 HumanMessage(content='2 + 2 は何ですか', additional_kwargs={}, response_metadata={}),
 AIMessage(content='4', additional_kwargs={}, response_metadata={}),
 HumanMessage(content='ありがとう', additional_kwargs={}, response_metadata={}),
 AIMessage(content='どういたしまして！', additional_kwargs={}, response_metadata={}),
 HumanMessage(content='楽しんでいますか？', additional_kwargs={}, response_metadata={}),
 AIMessage(content='はい！', additional_kwargs={}, response_metadata={})]

チェーンで使用するには、messages入力をプロンプトに渡す前にトリマーを実行するだけです。

workflow = StateGraph(state_schema=State)


def call_model(state: State):
    trimmed_messages = trimmer.invoke(state["messages"])
    prompt = prompt_template.invoke(
        {"messages": trimmed_messages, "language": state["language"]}
    )
    response = model.invoke(prompt)
    return {"messages": [response]}


workflow.add_edge(START, "model")
workflow.add_node("model", call_model)

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

今、モデルに名前を尋ねても、チャット履歴のその部分がトリミングされているため、モデルはそれを知りません：

config = {"configurable": {"thread_id": "abc567"}}
query = "私の名前は？"
language = "日本語"

input_messages = messages + [HumanMessage(query)]
output = app.invoke(
    {"messages": input_messages, "language": language},
    config,
)
output["messages"][-1].pretty_print()

================================== Ai Message ==================================

ごめんなさい、あなたの名前はわかりません。教えていただければ嬉しいです！

しかし、直近の数件のメッセージ内の情報について尋ねると、それを覚えています：

config = {"configurable": {"thread_id": "abc678"}}
query = "私が質問した感情の質問は何？"
language = "日本語"

input_messages = messages + [HumanMessage(query)]
output = app.invoke(
    {"messages": input_messages, "language": language},
    config,
)
output["messages"][-1].pretty_print()

================================== Ai Message ==================================

あなたが質問した感情の質問は、「楽しんでいますか？」という内容で、相手の感情や気分について尋ねるものです。何か特別なことについて楽しんでいるかどうかを知りたい場合や、全体的な幸福感について尋ねるものでもあります。何か他にお話ししたいことがあれば、どうぞ教えてください！

ストリーミング

これで動作するチャットボットを手に入れました。ただし、チャットボットアプリケーションの非常に重要なUXの考慮事項の1つは、ストリーミングです。LLMは応答するのに時間がかかることがありますので、ユーザーエクスペリエンスを向上させるために、ほとんどのアプリケーションが行うことの1つは、生成されるトークンごとにストリームバックすることです。これにより、ユーザーは進捗状況を確認できます。

実際、これを行うのは非常に簡単です！

デフォルトでは、LangGraphアプリケーションの.streamはアプリケーションステップをストリーム化します-- この場合、モデル応答の単一ステップです。stream_mode="messages"を設定すると、出力トークンをストリーム化できます：

config = {"configurable": {"thread_id": "abc789"}}
query = "はい、私はToddです。ジョークを教えてください。"
language = "日本語"

input_messages = [HumanMessage(query)]
for chunk, metadata in app.stream(
    {"messages": input_messages, "language": language},
    config,
    stream_mode="messages",
):
    if isinstance(chunk, AIMessage):  # モデルのレスポンスのみにフィルタリング
        print(chunk.content, end="|")

|こんにちは|、|Todd|さん|！|ジョ|ーク|ですね|。|では|、|こちら|はい|か|が|でしょう|か|？

|「|な|ぜ|どう|ぶ|つ|た|ちは|コン|ピ|ュー|タ|を|使|え|ない|の|？」
|「|だ|って|、|マ|ウ|ス|を|追|い|か|け|る|の|が|大|変|だから|！」

|少|し|笑|って|いただ|け|た|ら|嬉|しい|です|！|他|にも|聞|き|たい|こと|が|あ|れば|、|どう|ぞ|お|知らせ|ください|。||

次はRAGです。

はじめてのDatabricks

はじめてのDatabricks

Databricks無料トライアル

Databricks無料トライアル