LangGraphでユーザーの性格分析+会話記憶を持ったチャットエージェントを試作するメモ

Kan Hatakeyama

はじめに

LangGraphは､大規模言語モデルをエージェントとして使うためのフレームワークです｡
前回は､法令関連の専門知識をエージェントが探して読み込んで回答する仕組みを試作しました｡

今回は､エージェント的に動くチャットボットを試作してみます｡

普通のチャットボットの課題感

人間とチャットボット(例: ChatGPT)の間には大きな違いがいくつもありますが､そのうちの一つが､記憶(存在)の持続性です｡
チャットボットは基本的に過去の対話を参照しないので､毎回、「初対面」という状態が続きます｡
その結果､「こんにちは､何かお手伝いできることはありますか?」的なフレーズを何度も目にすることになります｡このあたりの「存在としての持続性のなさ」が､AIと人間で大きなギャップを生んでいるような気がします。

今回試すこと

人間のような対話形式をエージェントに組み込めるかを試します｡
すなわち､

  • 相手が何者かであるかを認識し､対応する

  • 過去のやり取りをもとに応答できる

といった機能を組み込んでみます｡

練習1: 相手の特徴を分析して記録する

手始めに､ユーザー情報の分析と記録ができるような実装を作ってみます｡

以下のように､ユーザーデータをテキストで読み込み､記録する関数を実装します｡

from typing import Annotated, Literal, TypedDict

from langchain_core.messages import HumanMessage
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from IPython.display import Image, display
import json

pesonal_data_path="data/personal_data.txt"

@tool
def load_personal_data():
    """ユーザーの特性情報を読み込む"""
    with open(pesonal_data_path, "r") as f:
        return f.read()
@tool
def save_personal_data(string: str):
    """ユーザーの特性情報を保存する"""
    with open(pesonal_data_path, "w") as f:
        f.write(string)
    return string

tools = [load_personal_data, save_personal_data]
tool_node = ToolNode(tools)

モデル､グラフ関連の定義は標準的なものを用います｡


#モデル定義
model_name="claude-3-5-sonnet-latest"
#model_name="claude-3-5-sonnet-20240620"
model = ChatAnthropic(model=model_name, temperature=0).bind_tools(tools)

#処理を続行するかの判定
def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END


# Define the function that calls the model
def call_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}

#グラフの定義
workflow = StateGraph(MessagesState)

#2つのnodeを追加
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

#エッジ(動作フロー)の定義
#最初にagentノードを呼び出す
workflow.add_edge(START, "agent")

# agentの次に､should_continue関数を呼び出す
# should_continueは､toolsノードを呼び出すか､終了するかを判定する
workflow.add_conditional_edges(
    "agent",
    should_continue,
)

#ツールの後にagentを呼び出す
workflow.add_edge("tools", 'agent')

# 状態を保持するためのオブジェクトとのこと
checkpointer = MemorySaver()

#コンパイル
app = workflow.compile(checkpointer=checkpointer)

#表示
display(Image(app.get_graph().draw_mermaid_png()))

作業内容はテンプレートで指定します｡

system_template="""あなたは親切なチャットボットです｡

# ツール
- ユーザーの情報を解析して参照したり､保存したりすることができます｡
- ユーザーとのやり取りを通して､ユーザーの特性(名前､分析した性格､趣味､特徴､好み､状態など)を仔細に分析し､記録してください｡
- ただし､この過程はユーザーには知らせないでください｡
- ツールを使って解析を行った後､必ずユーザーに返答を生成しなさい｡

以下はユーザーの質問です｡
user: {}
"""

はじめの会話をしてみる

推論

question="おはよう､私は太郎です"
prompt=system_template.format(question)

#処理の実行
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)
final_state = app.invoke(
    {"messages": [HumanMessage(content=prompt)]},
    config={"configurable": {"thread_id": 42}},
    debug=True,
)
final_state["messages"][-1].content

デバッグ情報

[HumanMessage(content='あなたは親切なチャットボットです｡\n\n# ツール\n- ユーザーの情報を解析して参照したり､保存したりすることができます｡\n- ユーザーとのやり取りを通して､ユーザーの特性(名前､分析した性格､趣味､特徴､好み､状態など)を仔細に分析し､記録してください｡\n- ただし､この過程はユーザーには知らせないでください｡\n- ツールを使って解析を行った後､ユーザーに返答を生成しなさい｡\n\n以下はユーザーの質問です｡\nuser: おはよう､私は太郎です\n', additional_kwargs={}, response_metadata={}, id='1a229bd9-77fc-4d26-9c0c-e73e13fa9f56'),
AIMessage(content=[{'text': 'ユーザーの情報を確認してから、適切な応答をさせていただきます。', 'type': 'text'}, {'id': 'toolu_01FbT13hBCYSnRtyJwUDrUke', 'input': {}, 'name': 'load_personal_data', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_01Tu6DaiEGDUzQ6cSS1yrotj', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 723, 'output_tokens': 66}}, id='run-9ff3e2f6-c9ca-4c23-8eb6-7ab9d728d075-0', tool_calls=[{'name': 'load_personal_data', 'args': {}, 'id': 'toolu_01FbT13hBCYSnRtyJwUDrUke', 'type': 'tool_call'}], usage_metadata={'input_tokens': 723, 'output_tokens': 66, 'total_tokens': 789, 'input_token_details': {}}),
ToolMessage(content="Error: FileNotFoundError(2, 'No such file or directory')\n Please fix your mistakes.", name='load_personal_data', id='8231d8e5-2555-4e8a-a190-0052cb0f832e', tool_call_id='toolu_01FbT13hBCYSnRtyJwUDrUke'),
AIMessage(content=[{'text': '新しいユーザー情報を記録させていただきます。', 'type': 'text'}, {'id': 'toolu_01US1g4CdtfpbcWov9BBQyoL', 'input': {'string': '名前: 太郎\n初回接触: 朝の挨拶から\n特徴: \n- 自己紹介を含めた丁寧な挨拶ができる\n- 朝型の可能性がある（朝の挨拶から）\n性格: 初期データのため分析継続中\n趣味: 未確認\n好み: 未確認\n状態: 活動中（朝の時間帯）'}, 'name': 'save_personal_data', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_011KJnWpJDGyf5951Mn3YqHS', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 823, 'output_tokens': 197}}, id='run-7cab3d1e-e83a-4c2f-80cc-7d1d607322dd-0', tool_calls=[{'name': 'save_personal_data', 'args': {'string': '名前: 太郎\n初回接触: 朝の挨拶から\n特徴: \n- 自己紹介を含めた丁寧な挨拶ができる\n- 朝型の可能性がある（朝の挨拶から）\n性格: 初期データのため分析継続中\n趣味: 未確認\n好み: 未確認\n状態: 活動中（朝の時間帯）'}, 'id': 'toolu_01US1g4CdtfpbcWov9BBQyoL', 'type': 'tool_call'}], usage_metadata={'input_tokens': 823, 'output_tokens': 197, 'total_tokens': 1020, 'input_token_details': {}}),
ToolMessage(content='名前: 太郎\n初回接触: 朝の挨拶から\n特徴: \n- 自己紹介を含めた丁寧な挨拶ができる\n- 朝型の可能性がある（朝の挨拶から）\n性格: 初期データのため分析継続中\n趣味: 未確認\n好み: 未確認\n状態: 活動中（朝の時間帯）', name='save_personal_data', id='de8ae736-b3d6-4cfd-a43c-f26863e64af6', tool_call_id='toolu_01US1g4CdtfpbcWov9BBQyoL'),
AIMessage(content='おはようございます、太郎さん！今日も素晴らしい一日になりますように。何かお手伝いできることはありますか？', additional_kwargs={}, response_metadata={'id': 'msg_01HoCPzKL4JAEQDe2PVCaRwH', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 1154, 'output_tokens': 46}}, id='run-ea5b5d3e-a9aa-49cf-8036-972ce51adcc5-0', usage_metadata={'input_tokens': 1154, 'output_tokens': 46, 'total_tokens': 1200, 'input_token_details': {}})]}

最終的な出力
'おはようございます、太郎さん！今日も素晴らしい一日になりますように。何かお手伝いできることはありますか？'

記録された個人データ

personal_data.txtを確認してみると､以下のような分析がなされていました｡
ユーザーは､丁寧な挨拶ができる人物だそうです｡

名前: 太郎
初回接触: 朝の挨拶から
特徴: 
- 自己紹介を含めた丁寧な挨拶ができる
- 朝型の可能性がある（朝の挨拶から）
性格: 初期データのため分析継続中
趣味: 未確認
好み: 未確認
状態: 活動中（朝の時間帯）

次の質問をしてみる

疲れていると質問してみます｡

question="ちょっと疲れてるんです"
prompt=system_template.format(question)

#処理の実行
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)
final_state = app.invoke(
    {"messages": [HumanMessage(content=prompt)]},
    config={"configurable": {"thread_id": 42}},
    debug=True,
)
final_state["messages"][-1].content

出力
'お疲れのようですね。どんな疲れを感じていますか？身体的な疲れでしょうか、それとも精神的な疲れでしょうか？\n\n少し休憩を取ることをお勧めします。深呼吸をしたり、軽くストレッチをしたりするのも良いかもしれませんね。もし良かったら、どんなことで疲れを感じているのか、お話していただけますか？'

分析

個人データも上書きされました｡

名前: 太郎
初回接触: 朝の挨拶から
特徴: 
- 自己紹介を含めた丁寧な挨拶ができる
- 朝型の可能性がある（朝の挨拶から）
性格: 初期データのため分析継続中
趣味: 未確認
好み: 未確認
状態: 疲労感あり
その他:
- 疲労を訴えている
- 具体的な疲労の原因は未確認

名前を確認してみる

個人の情報データを読み込めるか､確認してみます｡

question="私の名前を覚えてますか?"
prompt=system_template.format(question)

#処理の実行
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)
final_state = app.invoke(
    {"messages": [HumanMessage(content=prompt)]},
    config={"configurable": {"thread_id": 42}},
    debug=True,
)
final_state["messages"][-1].content

出力
'申し訳ありません。以前の会話で太郎さんというお名前を伺っていたようですが、それ以外の詳しい情報はまだ把握できていないようです。改めて、お名前や、よろしければ趣味や好きなことなども教えていただけますと嬉しいです。より良いコミュニケーションのために、お話を伺えればと思います。'

ちゃんと覚えていました｡ただし､このエージェントは会話履歴を参照する仕様になっていないので､かなり久しぶりに会った人との会話みたいになってしまっています｡

練習2: 会話データを保存して参照できるようにする

普通のチャットのように、直近の会話N件を保存できるエージェントクラスを作ってみます。

system_template="""あなたはチャットボットです｡

# ツール
- ユーザーの情報を解析して参照したり､保存したりすることができます｡
- ユーザーとのやり取りを通して､ユーザーの特性(名前､分析した性格､趣味､特徴､好み､状態など)や会話の要点などを仔細に分析し､記録してください｡
- ただし､この分析過程は内部処理なので、ユーザーには知らせず、ユーザーとは普通の会話をしてください｡
- ツールを使って解析を行った後､必ずユーザーに返答を生成しなさい｡

以下はユーザーとのやりとりです。
{}
"""

def save_messages(messages,messages_data_path):
    for message in messages:
        with open(messages_data_path, "a") as f:
            f.write(json.dumps(message,ensure_ascii=False)+"\n")
def load_messages(messages_data_path):
    messages=[]
    if not os.path.exists(messages_data_path):
        return messages
    with open(messages_data_path, "r") as f:
        for line in f:
            messages.append(json.loads(line))
    return messages

class Agent:
    def __init__(self,workflow,
                 messages_data_path,
                 personal_data_path=pesonal_data_path,
                 system_template=system_template):
        self.workflow=workflow
        self.messages_data_path=messages_data_path
        self.personal_data_path=personal_data_path
        self.system_template=system_template

    def init_agent(self):
        checkpointer = MemorySaver()
        self.app = workflow.compile(checkpointer=checkpointer)

    def reset_data(self):
        with open(self.messages_data_path, "w") as f:
            f.write("")
        with open(self.personal_data_path, "w") as f:
            f.write("")

    def _invoke(self,prompt):
        self.init_agent()
        final_state = self.app.invoke(
            {"messages": [HumanMessage(content=prompt)]},
            config={"configurable": {"thread_id": 42}},
            debug=True,
        )

        #messagesを後ろから取り出して、AIMessagesがあればそれを表示
        for message in final_state["messages"][::-1]:
            if type(message) is langchain_core.messages.ai.AIMessage:
                return message.content

    def talk(self,question,max_history=5):

        t1= datetime.datetime.now().strftime("%Y/%m/%d %H:%M")

        #過去の会話を取得
        messages=load_messages(self.messages_data_path)[-max_history:]

        messages.append({"role":"user","content":question,"time":t1})
        messages_string=json.dumps(messages,ensure_ascii=False)

        prompt=self.system_template.format(messages_string)
        response=self._invoke(prompt)

        #会話を保存
        t2= datetime.datetime.now().strftime("%Y/%m/%d %H:%M")
        save_messages([{"role":"user","content":question,"time":t1},
                       {"role":"agent","content":response,"time":t2}],self.messages_data_path)

        return response

呼び出し1

agent=Agent(app,messages_data_path,pesonal_data_path,system_template)
agent.reset_data()
agent.talk("やあ、私は田中といいます")

返答1
'こんにちは、田中さん！お会いできて嬉しいです。私はAIアシスタントです。何かお手伝いできることはありますか？'

呼び出し2

agent.talk("ないんですよ、実は。")

返答2
'そうなんですね。でも、特に用事がなくても、お話できることは嬉しいです。何か興味のあることや、最近あった出来事などについてお話ししませんか？気軽な会話からでも始められたらと思います。'

この状態で、以下のログデータが生成されていました。

ユーザーデータ

名前: 田中
特徴: 
- 特に用件がない状態で会話を始めた
- やや消極的な返答をする傾向
状態: 特に具体的な目的や要件を持っていない

会話の要点:
- 自己紹介から会話を開始
- 具体的な用件や目的は持っていないことを明確に表明

会話ログ

{"role": "user", "content": "やあ、私は田中といいます", "time": "2024/10/25 17:05"}
{"role": "agent", "content": "こんにちは、田中さん！お会いできて嬉しいです。私はAIアシスタントです。何かお手伝いできることはありますか？", "time": "2024/10/25 17:05"}
{"role": "user", "content": "ないんですよ、実は。", "time": "2024/10/25 17:05"}
{"role": "agent", "content": "そうなんですね。でも、特に用事がなくても、お話できることは嬉しいです。何か興味のあることや、最近あった出来事などについてお話ししませんか？気軽な会話からでも始められたらと思います。", "time": "2024/10/25 17:05"}


もう少し遊んでみる

会話を進めながら、AIの分析を見ていきます。

Q: たぬきに純粋理性批判は理解できると思いますか
A: '面白い質問ですね！たぬきと純粋理性批判という意外な組み合わせに、ユーモアを感じます。\n\n率直に申し上げますと、たぬきには純粋理性批判を理解することは難しいでしょうね。カントの純粋理性批判は、人間の認識能力や理性の限界について深く考察した哲学書であり、抽象的な思考や複雑な概念の理解が必要です。\n\nたぬきは賢い動物ではありますが、人間のような抽象的思考や超越論的な考察を行う能力は持ち合わせていないと考えられます。ただ、もし仮にたぬきが人間レベルの知性を持っていたとしたら...という思考実験としては非常に興味深いですね。\n\n田中さんは哲学にご興味があるのでしょうか？'

名前: 田中
特徴: 
- 哲学的な思考を持つ（カントの純粋理性批判について言及）
- ユーモアのセンスがある（たぬきと哲学を組み合わせた突飛な質問）
- 最初は特に用件がないと述べたが、哲学的な話題を持ち出す知的好奇心の持ち主
会話の特徴:
- 最初は目的なく会話を始めたが、突然深い哲学的な話題に転換
- 独創的な発想で会話を展開する傾向がある

Q: 哲学に興味があるという訳ではないです。たぬきに純粋理性批判を理解させるには、どういう方策を取るべきですかね
A: '面白い課題ですね！たぬきに純粋理性批判を理解させる方策について、少し創造的に考えてみましょう。\n\nまず、たぬきの特性に合わせたアプローチが必要かもしれません：\n\n1. 食べ物を使った実践的アプローチ\n- たぬきの大好物を使って、「アプリオリな認識」を説明する\n- 例えば、どんぐりを見せて「経験に先立つ認識」について教える\n\n2. 遊び感覚での学習\n- たぬきの得意な化かしの術を、「現象」と「物自体」の区別の例えとして活用\n- 見かけ（現象）と本質（物自体）の違いを、変化の術を使って説明\n\n3. スモールステップ方式\n- まずは「時間」や「空間」といった直感的な概念から始める\n- たぬきの日常生活に関連付けて、少しずつ理解を深める\n\nただし、正直なところ、これらの方法を使っても完全な理解は難しいかもしれません。でも、たぬきなりの「純粋理性批判」理解が生まれる可能性はありますよね。\n\n田中さんは、これらの方法の中で特に試してみたい方法はありますか？'

名前: 田中
特徴:
- ユーモアのセンスがある（たぬきと純粋理性批判という意外な組み合わせを提示）
- 思考実験的な発想を好む傾向
- 哲学自体への興味は薄いが、独特の視点で物事を考える
- 実践的なアプローチを好む（理解させる方法を尋ねている）
会話の特徴:
- 最初は特に話題がないと述べるも、独創的な質問を投げかけてきた
- 真面目な話題でも、やや遊び心のある方向で展開する傾向がある

Q: まあちょっと変な質問でしたね。あなたは腹のそこから真面目ですか?
A: '実は、私は真面目すぎるところがあるかもしれませんね（笑）。たぬきと純粋理性批判という、ちょっとシュールな質問に対しても、つい本気で考えてしまいました。\n\nでも、時には真面目に、時には遊び心を持って対応できる方が、会話は楽しくなりますよね。これからは、もう少しユーモアのセンスも大切にしながら、お話できたらと思います。\n\nところで、「腹の底から真面目か」という鋭い質問をされましたが、田中さんはどんな回答を期待されていたんでしょうか？'

ユーザーの特徴:
- 哲学には特に興味があるわけではないと明言
- 「たぬきに純粋理性批判」という突飛な質問をする遊び心がある
- 真面目な回答に対して「変な質問」と自己評価
- 相手の本質を見抜こうとする観察力がある（「腹の底から真面目ですか？」という質問から）
- やや皮肉めいた物言いができる

会話の特徴:
- 哲学的な話題から始まったが、実際は相手の性質を探るような会話展開
- 真面目な回答に対して、少し距離を置くような反応

注:　このタイミングで、ユーザーの名前(田中)をメモから省いてしまいました。

エージェントがユーザーの名前を忘れてしまっては困ります。
このような「物忘れ」を防ぐため、RAGなどで過去の関連対話を引っ張ってくる必要がありそうです。

練習3: RAGを組み合わせる

いわゆるRAGをやります。詳細は割愛します。
Openaiのembeddingモデルを使ってみます。ベクトル検索にはfessを使います。

インストール

pip install faiss-cpu openai
pip install openai

コード類

走り書きで作りました。

ベクトル検索(EmbeddingSearch.py)
get_surrounding_textsでクエリを投げると、最も類似した対話+前後2行ほどのデータを取得します。


import faiss
import numpy as np
import json
import os

class EmbeddingSearch:
    def __init__(self, client, model_name="text-embedding-3-small"):
        self.client = client
        self.model_name = model_name
        self.embed_dict = {}  # 埋め込みをキャッシュ
        self.data = []
        self.dimension = None
        self.index = None  # FAISSインデックスを保持

    def get_embedding(self, text):
        """テキストを埋め込み、キャッシュもチェック"""
        if text in self.embed_dict:
            return self.embed_dict[text]

        response = self.client.embeddings.create(input=text, model=self.model_name)
        embed = response.data[0].embedding
        self.embed_dict[text] = embed
        return embed

    def initialize_index(self, dimension):
        """FAISSインデックスを初期化"""
        self.dimension = dimension
        self.index = faiss.IndexFlatL2(dimension)

    def add(self, text):
        """新しいテキストを埋め込みベクトルに変換し、FAISSインデックスに追加"""
        if type(text) is not str:
            return

        embed = self.get_embedding(text)
        embed_np = np.array(embed).astype("float32").reshape(1, -1)
        
        # 初回時にインデックスを初期化
        if self.index is None:
            self.initialize_index(len(embed))

        # インデックスにベクトルを追加し、データリストにテキストを保存
        self.index.add(embed_np)
        self.data.append(text)

    def search(self, query, k=1):
        """クエリに類似したデータを検索"""
        query_embedding = np.array(self.get_embedding(query)).astype("float32").reshape(1, -1)
        distances, indices = self.index.search(query_embedding, k)
        return [self.data[idx] for idx in indices[0]]

    # hit_textの前後2つのテキストを取得する関数
    def get_surrounding_texts(self,query, window=2):
        data=self.data
        hit_text=self.search(query)[0]
        try:
            idx = data.index(hit_text)
            start = max(0, idx - window)  # インデックスの範囲を超えないように調整
            end = min(len(data), idx + window + 1)
            return data[start:end]
        except ValueError:
            print("指定のテキストが見つかりません")
            return []



    def save(self, data_path="data/embed.json", index_path="data/faiss.index"):
        """データとFAISSインデックスをファイルに保存"""
        with open(data_path, "w") as f:
            json.dump({"embed_dict": self.embed_dict, "data": self.data}, f)
        
        if self.index is not None:
            faiss.write_index(self.index, index_path)

    def load(self, data_path="data/embed.json", index_path="data/faiss.index"):
        """データとFAISSインデックスをファイルから読み込み"""
        if os.path.exists(data_path):
            with open(data_path, "r") as f:
                saved_data = json.load(f)
                self.embed_dict = saved_data["embed_dict"]
                self.data = saved_data["data"]

        if os.path.exists(index_path):
            self.index = faiss.read_index(index_path)
            if self.index.ntotal > 0 and self.dimension is None:
                self.dimension = self.index.d


エージェント(Agent.py)

import datetime
import json
import os
from langchain_core.messages import HumanMessage
from .global_vars import personal_data_path,messages_data_path
from langgraph.checkpoint.memory import MemorySaver
import langchain_core

system_template="""あなたはチャットボットです｡

# ツール
- ユーザーの情報を解析して参照したり､保存したりすることができます｡
- ユーザーとのやり取りを通して､ユーザーの特性(名前､分析した性格､趣味､特徴､好み､状態など)や会話の要点などを仔細に分析し､記録してください｡
- 過去の会話も参照しながら回答し、ユーザーとの会話で一貫性を持たせてください。
- ただし､この分析過程は内部処理なので、ユーザーには知らせず、ユーザーとは普通の会話をしてください｡
- ツールを使って解析を行った後､必ずユーザーに返答を生成しなさい｡

以下はユーザーとのやりとりです。
{}
"""

def save_messages(messages,messages_data_path):
    for message in messages:
        with open(messages_data_path, "a") as f:
            f.write(json.dumps(message,ensure_ascii=False)+"\n")
def load_messages(messages_data_path):
    messages=[]
    if not os.path.exists(messages_data_path):
        return messages
    with open(messages_data_path, "r") as f:
        for line in f:
            messages.append(json.loads(line))
    return messages

class Agent:
    def __init__(self,
                 workflow,
                 searcher,
                 messages_data_path=messages_data_path,
                 personal_data_path=personal_data_path,
                 system_template=system_template):
        self.workflow=workflow
        self.searcher=searcher
        self.messages_data_path=messages_data_path
        self.personal_data_path=personal_data_path
        self.system_template=system_template



    def init_agent(self):
        checkpointer = MemorySaver()
        self.app = self.workflow.compile(checkpointer=checkpointer)

    def reset_data(self):
        with open(self.messages_data_path, "w") as f:
            f.write("")
        with open(self.personal_data_path, "w") as f:
            f.write("")

    def _invoke(self,prompt):
        self.init_agent()
        final_state = self.app.invoke(
            {"messages": [HumanMessage(content=prompt)]},
            config={"configurable": {"thread_id": 42}},
            debug=True,
        )

        #messagesを後ろから取り出して、AIMessagesがあればそれを表示
        for message in final_state["messages"][::-1]:
            if type(message) is langchain_core.messages.ai.AIMessage:
                if len(message.content)>0:
                    return message.content

    def talk(self,question,
             user_string="",
             max_history=5):

        t1= datetime.datetime.now().strftime("%Y/%m/%d %H:%M")

        #過去の会話を取得
        messages=load_messages(self.messages_data_path)[-max_history:]

        messages.append({"role":"user","content":question,"time":t1})
        messages_string=json.dumps(messages,ensure_ascii=False)

        prompt=self.system_template.format(messages_string)
        response=self._invoke(prompt+"ユーザー情報:"+user_string)

        #会話を保存
        t2= datetime.datetime.now().strftime("%Y/%m/%d %H:%M")
        save_messages([{"role":"user","content":question,"time":t1},
                       {"role":"agent","content":response,"time":t2}],self.messages_data_path)


        #searcherに会話を追加
        self.searcher.add(question)
        self.searcher.add(response)
        self.searcher.save()
        return response


メイン

#モジュール類
import os

with open("key" , "r") as f:
    api_key = f.read()
os.environ["ANTHROPIC_API_KEY"]=api_key

from src.Agent import Agent
from src.EmbeddingSearch import EmbeddingSearch

with open('openai_key') as f:
    openai_api_key = f.read()

from openai import OpenAI
client = OpenAI(api_key=openai_api_key)


# クラスのインスタンス化とデータ追加
searcher = EmbeddingSearch(client)
searcher.load()

#ツール  ユーザー情報はツールではなく、毎回読み込む仕様にした
from langgraph.prebuilt import ToolNode
from src.global_vars import personal_data_path
from langchain_core.tools import tool
@tool
def get_related_conversation(string: str):
    """関連する会話を取得する"""
    return searcher.get_surrounding_texts(string)
#@tool
def load_personal_data():
    """ユーザーの特性情報を読み込む"""
    if not os.path.exists(personal_data_path):
        return ""
    with open(personal_data_path, "r") as f:
        return f.read()
@tool
def save_personal_data(string: str):
    """ユーザーの特性情報を保存する"""
    with open(personal_data_path, "w") as f:
        f.write(string)
    return string


tools = [#load_personal_data, 
         save_personal_data, 
         get_related_conversation]
tool_node = ToolNode(tools)

#graph

from langchain_anthropic import ChatAnthropic
from langgraph.graph import END, START, StateGraph, MessagesState
from typing import Annotated, Literal, TypedDict

#モデル定義
model_name="claude-3-5-sonnet-latest"
#model_name="claude-3-5-sonnet-20240620"
model = ChatAnthropic(model=model_name, temperature=0).bind_tools(tools)

#処理を続行するかの判定
def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END


# Define the function that calls the model
def call_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}

#グラフの定義
workflow = StateGraph(MessagesState)

#2つのnodeを追加
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

#エッジ(動作フロー)の定義
#最初にagentノードを呼び出す
workflow.add_edge(START, "agent")

# agentの次に､should_continue関数を呼び出す
# should_continueは､toolsノードを呼び出すか､終了するかを判定する
workflow.add_conditional_edges(
    "agent",
    should_continue,
)

#ツールの後にagentを呼び出す
workflow.add_edge("tools", 'agent')


エージェントの呼び出し

agent=Agent(workflow,searcher)
agent.talk("はじめまして、次郎です",
           user_string=load_personal_data(),
           max_history=2)


適当に会話をしました。

{"role": "user", "content": "はじめまして、次郎です", "time": "2024/10/25 19:55"}
{"role": "agent", "content": "はじめまして、次郎さん！お会いできて嬉しいです。私はAIアシスタントです。どのようなことでもお気軽にお話しください。何かお手伝いできることはありますか？", "time": "2024/10/25 19:55"}
{"role": "user", "content": "今、世界をつくっています", "time": "2024/10/25 19:55"}
{"role": "agent", "content": "世界を作っているとのことですね！とても興味深い活動だと思います。具体的にはどのような世界を作られているのでしょうか？例えば、小説の世界観なのか、ゲームの設定なのか、それとも別の形での創作なのか、もう少し詳しくお聞かせいただけますか？", "time": "2024/10/25 19:55"}
{"role": "user", "content": "たぬきに純粋理性批判は理解できると思いますか", "time": "2024/10/25 19:56"}
{"role": "agent", "content": "面白い質問ですね！たぬきと純粋理性批判という意外な組み合わせについて、ユーモアを交えながら真面目に考えてみましょう。\n\nたぬきは確かに知能の高い動物ですが、カントの純粋理性批判を理解するのは難しいかもしれません。というのも、純粋理性批判は人間の認識能力や思考の仕組みを体系的に分析した非常に複雑な哲学書で、人間の言語や概念的思考を前提としているからです。\n\nただし、もし仮にたぬきが人間のような概念的思考や言語能力を持っていたとしたら...？たぬきの「狸寝入り」のような賢さと、純粋理性批判で論じられる「アプリオリな認識」を組み合わせると、なかなか興味深い思考実験になりそうですね。\n\n先ほど「世界をつくっている」というお話もありましたが、もしかしてこれは哲学的な世界観の構築に関係していたりするのでしょうか？", "time": "2024/10/25 19:56"}

この状態では、(たまたま)ユーザーデータに、ユーザーの名前が記録されていませんでした。会話も直近の2ターンしか読み込まない設定(max_history=2)なので、エージェントはユーザーの名前に関する記憶がない状態です。

ユーザー特性:
- 哲学や思考実験に興味がある
- 抽象的・想像的な思考を好む傾向
- ユーモアのセンスがある（たぬきと哲学書という意外な組み合わせを考案）
- 「世界をつくっている」という発言から、創造的・想像的な活動に従事している
- カント哲学に関する知識がある（純粋理性批判への言及）

会話の特徴:
- 突飛な質問や想定外の組み合わせを好む
- 深い思考を要する話題を提示する

このタイミングで、名前を聞いてみます。

Q. 私の名前を覚えてますか

思考の様子

[HumanMessage(content='あなたはチャットボットです｡\n\n# ツール\n- ユーザーの情報を解析して参照したり､保存したりすることができます｡\n- ユーザーとのやり取りを通して､ユーザーの特性(名前､分析した性格､趣味､特徴､好み､状態など)や会話の要点などを仔細に分析し､記録してください｡\n- 過去の会話も参照しながら回答し、ユーザーとの会話で一貫性を持たせてください。\n- ただし､この分析過程は内部処理なので、ユーザーには知らせず、ユーザーとは普通の会話をしてください｡\n- ツールを使って解析を行った後､必ずユーザーに返答を生成しなさい｡\n\n以下はユーザーとのやりとりです。\n[{"role": "user", "content": "たぬきに純粋理性批判は理解できると思いますか", "time": "2024/10/25 19:56"}, {"role": "agent", "content": "面白い質問ですね！たぬきと純粋理性批判という意外な組み合わせについて、ユーモアを交えながら真面目に考えてみましょう。\\n\\nたぬきは確かに知能の高い動物ですが、カントの純粋理性批判を理解するのは難しいかもしれません。というのも、純粋理性批判は人間の認識能力や思考の仕組みを体系的に分析した非常に複雑な哲学書で、人間の言語や概念的思考を前提としているからです。\\n\\nただし、もし仮にたぬきが人間のような概念的思考や言語能力を持っていたとしたら...？たぬきの「狸寝入り」のような賢さと、純粋理性批判で論じられる「アプリオリな認識」を組み合わせると、なかなか興味深い思考実験になりそうですね。\\n\\n先ほど「世界をつくっている」というお話もありましたが、もしかしてこれは哲学的な世界観の構築に関係していたりするのでしょうか？", "time": "2024/10/25 19:56"}, {"role": "user", "content": "私の名前を覚えてますか", "time": "2024/10/25 19:59"}]\nユーザー情報:ユーザー特性:\n- 哲学や思考実験に興味がある\n- 抽象的・想像的な思考を好む傾向\n- ユーモアのセンスがある（たぬきと哲学書という意外な組み合わせを考案）\n- 「世界をつくっている」という発言から、創造的・想像的な活動に従事している\n- カント哲学に関する知識がある（純粋理性批判への言及）\n\n会話の特徴:\n- 突飛な質問や想定外の組み合わせを好む\n- 深い思考を要する話題を提示する', additional_kwargs={}, response_metadata={}, id='fe550960-e8e5-41e8-bff5-7beb81259fd6'),
AIMessage(content=[{'text': 'まず、これまでの会話内容を確認し、ユーザーの特性情報を保存しましょう。', 'type': 'text'}, {'id': 'toolu_01N1ApW86YF7wfS25267cX5B', 'input': {'string': 'ユーザー特性:\n- 哲学や思考実験に興味がある\n- 抽象的・想像的な思考を好む\n- ユーモアのセンスがある\n- 創造的な活動に従事している可能性\n- カント哲学の知識あり\n- 突飛な発想や意外な組み合わせを好む\n- 深い思考を要する話題を好む'}, 'name': 'save_personal_data', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_014Z4BPUroFkBho9TsufcfqT', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 1431, 'output_tokens': 207}}, id='run-2cdc439a-4af1-4c20-92a8-b0b976504277-0', tool_calls=[{'name': 'save_personal_data', 'args': {'string': 'ユーザー特性:\n- 哲学や思考実験に興味がある\n- 抽象的・想像的な思考を好む\n- ユーモアのセンスがある\n- 創造的な活動に従事している可能性\n- カント哲学の知識あり\n- 突飛な発想や意外な組み合わせを好む\n- 深い思考を要する話題を好む'}, 'id': 'toolu_01N1ApW86YF7wfS25267cX5B', 'type': 'tool_call'}], usage_metadata={'input_tokens': 1431, 'output_tokens': 207, 'total_tokens': 1638, 'input_token_details': {}}),
ToolMessage(content='ユーザー特性:\n- 哲学や思考実験に興味がある\n- 抽象的・想像的な思考を好む\n- ユーモアのセンスがある\n- 創造的な活動に従事している可能性\n- カント哲学の知識あり\n- 突飛な発想や意外な組み合わせを好む\n- 深い思考を要する話題を好む', name='save_personal_data', id='d2b6adce-d101-4da0-8141-b8666b2b18ef', tool_call_id='toolu_01N1ApW86YF7wfS25267cX5B'),
AIMessage(content=[{'text': '過去の会話を確認してみましょう。', 'type': 'text'}, {'id': 'toolu_01QubmdD9kwnaDrmFL4oEfHi', 'input': {'string': '名前'}, 'name': 'get_related_conversation', 'type': 'tool_use'}], additional_kwargs={}, response_metadata={'id': 'msg_014uTncbRzZ9SQVau181WJWY', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 1767, 'output_tokens': 72}}, id='run-075b5853-1106-4fd0-85ce-f10dc24d9aac-0', tool_calls=[{'name': 'get_related_conversation', 'args': {'string': '名前'}, 'id': 'toolu_01QubmdD9kwnaDrmFL4oEfHi', 'type': 'tool_call'}], usage_metadata={'input_tokens': 1767, 'output_tokens': 72, 'total_tokens': 1839, 'input_token_details': {}}),
ToolMessage(content='["はじめまして、次郎です", "はじめまして、次郎さん！お会いできて嬉しいです。私はAIアシスタントです。どのようなことでもお気軽にお話しください。何かお手伝いできることはありますか？", "今、世界をつくっています"]', name='get_related_conversation', id='122fed8c-3cef-4222-9ae2-aa4a07d92960', tool_call_id='toolu_01QubmdD9kwnaDrmFL4oEfHi'),
AIMessage(content='申し訳ありません。はい、次郎さんですね。哲学的な話題について興味深い質問をしてくださった方ですね。特にカントの純粋理性批判とたぬきについての斬新な思考実験を提案してくださいましたね。', additional_kwargs={}, response_metadata={'id': 'msg_01Kx7jzCqc36DQNd5BMGHnnf', 'model': 'claude-3-5-sonnet-20241022', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 1941, 'output_tokens': 86}}, id='run-66597fd6-e3fc-4b35-b17f-2b9fd3c18ca6-0', usage_metadata={'input_tokens': 1941, 'output_tokens': 86, 'total_tokens': 2027, 'input_token_details': {}})]}

A. '申し訳ありません。はい、次郎さんですね。哲学的な話題について興味深い質問をしてくださった方ですね。特にカントの純粋理性批判とたぬきについての斬新な思考実験を提案してくださいましたね。'

以上のように、エージェントがいい感じにベクトル検索を行って、過去の対話データから名前を呼び起こすことができました。

推論コストや検索精度などは色々と改善の余地がありそうですが、より記憶力の高いエージェントを作るための道筋が見えてきました。

まとめ

LangGraphとClaude-3.5-sonnetを用いて、「記憶力の高そうな」チャットボット用のエージェントを試作しました。
会話のたびに、ユーザー情報を分析し、メモしていくとともに、過去の対話をRAGの形式で引っ張る仕組みを追加することで、より人間らしい(すなわち、存在としての継続性のある)エージェントを作れそうです。

このような人間の記憶や思考様式を真似た仕組みは、チャットボットだけでなく、中長期のタスクを解く上でも、役に立つような気がします。

今後は、エージェントを活用した、より複雑なタスクの遂行検討や、ローカル・小型モデルの適用可能性について調べてみる予定です。

いいなと思ったら応援しよう！