LangChain v0.3 その15 ~LangMemでAgentを理解する２~

1. はじめに

前回のその14
ちょっと関数が長すぎてわかりにくい。
初心者向けを謳う永遠の初心者（俺）には長すぎてわかりにくい。
もっとシンプルにするのが今回の目的。
わかったよ！ってひとはすっ飛ばしがおすすめ。

2. 前回の記事

バージョン情報とかは変えてないので見てね

3. プログラムを作るべ

3-1. 前回と同じところ

変えていないところは解説もなくガシガシ飛ばします。

3-1-1. ライブラリインポート

ライブラリインポート

import os
from typing import Any, Dict, List
from dotenv import load_dotenv

from langgraph.store.memory import InMemoryStore
from langmem import create_manage_memory_tool, create_search_memory_tool
from langchain_openai import AzureChatOpenAI

3-1-2. 環境変数の用意

環境変数

load_dotenv()

os.environ['AZURE_OPENAI_API_KEY'] = os.getenv('API_KEY')
os.environ['AZURE_OPENAI_ENDPOINT'] = os.getenv('END_POINT')

3-1-3. モデルの用意

model

llm = AzureChatOpenAI(
    model="gpt-4o-mini",
    azure_endpoint=os.getenv("END_POINT"),
    api_version=os.getenv("API_VERSION"),
    api_key=os.getenv("API_KEY")
)

3-1-4. toolsの用意

tools

store = InMemoryStore(
            index={
                "dims": 1536,
                "embed": "azure_openai:text-embedding-ada-002",
            }
        )
memory_tools = [
    create_manage_memory_tool(namespace="memories", store=store),
    create_search_memory_tool(namespace="memories", store=store),
]

3-1-5. toolsを実行する関数

tools実行関数

def execute_tool(tools_by_name: Dict[str, Any], tool_call: Dict[str, Any]) -> str:
    """Execute a tool call and return the result"""
    tool_name = tool_call["name"]

    if tool_name not in tools_by_name:
        return f"Error: Tool {tool_name} not found"
    else:
        # print(tool_call)
        pass

    tool = tools_by_name[tool_name]
    # print(tool)
    try:
        result = tool.invoke(tool_call["args"])
        return str(result)
    except Exception as e:
        return f"Error executing {tool_name}: {str(e)}"

3-1-6. モデル実行に使う辞書

モデル実行に利用する辞書

tools_by_name = {tool.name: tool for tool in memory_tools}
tools_by_name

3-1-7. OpenAI形式のtools

OpenAI形式tools

openai_tools = [
    {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description,
            "parameters": tool.tool_call_schema.model_json_schema(),
        },
    }
    for tool in memory_tools
]
openai_tools

3-1-8. モデルへtoolsをバインド

binded model

llm_with_tools = llm.bind_tools(openai_tools)
llm_with_tools

3-2. Agentを回す関数を作る

はい。ここからですね！

def agent_run(messages: list)-> list:
    while True:
        # モデルに入れて推論して、出力をmessagesに追加する
        response = llm_with_tools.invoke(messages)
        tool_calls = response.tool_calls
        tool_calls_len = len(tool_calls)

        messages.append(
            {"role": "assistant", "content": response.content, "tool_calls": tool_calls}
        )
        
        # tool_callingされない時の処理はモデルの出力のmessagesを返す
        if not tool_calls:
            # No more tools to call, return the final response
            print(f"{response.content=}")
            return messages

        # tool_callingされたときにtoolsを実行してmessagesに追加し、返す
        for i, tool_call in enumerate(tool_calls):
            print(tool_call)
            tool_result = execute_tool(tools_by_name, tool_call)
            
            messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call["id"],
                    "content": tool_result,
            })

            # 複数のtoolsが呼ばれた時、すべて実行が終わったら返り値を返す
            if tool_calls_len == i + 1:
                return messages

agent_run関数を作りました。
3-1-1で作る予定のmessages(list)を受け取って、toolsをバインドしたモデルに入力します。
そして、その結果にtool_callsが空ならモデルの出力を返すことで、while文を抜け、最終出力をします。
tool_callsに何かがあるときは、execute_tool関数を実行します。
tool_callsが複数あるときはその回数分だけ実行し、最後にmessagesを返すことでwhile文を抜け、最終出力をします。

そんな関数。

3-3. 実行する

さて、この関数を実行してきましょう。

3-3-1. 最初の入力

最初の入力messagesを作る

user_input = "今日は8時から働いてつかれたな。"
messages = [{"role": "user", "content": user_input}]
messages

# [{'role': 'user', 'content': '今日は8時から働いてつかれたな。'}]

messagesを渡していくので、最初のmessagesを作ります。面倒なので、システムプロンプトは無しです。💦

最初の入力

messages = agent_run(messages)
print(messages)

# response.content='お疲れ様です！長い一日でしたね。何かリラックスする方法や楽しいことを計画していますか？'

# [{'role': 'user', 'content': '今日は8時から働いてつかれたな。'}, 
#  {'role': 'assistant', 'content': 'お疲れ様です！長い一日でしたね。何かリラックスする方法や楽しいことを計画していますか？', 'tool_calls': []}]

messagesをagent_runに渡します。
tool_callsには何も入っていないので、そのまま出力されました。

3-3-2. 覚えてほしいことを入力（2回目の入力）

2回目の入力

user_input = "朝はメールを20件書いた。覚えといて！"
messages.append({"role": "user", "content": user_input})
messages = agent_run(messages)
print(messages[-1])

# {'name': 'manage_memory', 'args': {'content': 'User wrote 20 emails in the morning.', 'action': 'create'}, 'id': 'call_MBK0WYChIcJyPkpSD5QhHKsv', 'type': 'tool_call'}

# {'role': 'tool',
#  'tool_call_id': 'call_MBK0WYChIcJyPkpSD5QhHKsv',
#  'content': 'created memory 6994456e-489e-47d1-a5c2-0fe601c70721'}

さてここからです。
あえて「覚えといて！」とtoolsを意識させます。せこっ
argsのactionにcreateってありますね。情報を追加するための引数ですね。

3-3-3. 2回目の入力を上書き（3回目の入力）

3回目の入力

user_input = "いや、メールは25件だな。覚えといて！"
messages.append({"role": "user", "content": user_input})
messages = agent_run(messages)
print(messages[-1])

# {'name': 'manage_memory', 'args': {'content': 'User wrote 25 emails in the morning.', 'action': 'update', 'id': '6994456e-489e-47d1-a5c2-0fe601c70721'}, 'id': 'call_KjGdXZxOLePRIRrIGEFj7lnR', 'type': 'tool_call'}

# {'role': 'tool',
#  'tool_call_id': 'call_KjGdXZxOLePRIRrIGEFj7lnR',
#  'content': 'updated memory 6994456e-489e-47d1-a5c2-0fe601c70721'}

間違ったフリして訂正させます。今度はactionにupdateになりました。情報の上書きですね。

3-3-4. メモリー機能を用いる（4回目の入力）

4回目の入力

user_input = "朝、メールは何件書いたんだっけ？"
messages.append({"role": "user", "content": user_input})
messages = agent_run(messages)
print(messages[-1])

# response.content='あなたは朝に25件のメールを書いたと覚えています。'

# {'role': 'assistant', 'content': 'あなたは朝に25件のメールを書いたと覚えています。', 'tool_calls': []}

そして聞いてみます。tool_callsは空なので、回答を生成します。
25件と訂正された数字が返ってきました。

3-3-5. 情報の追加（5回目の入力）

5回目の入力

user_input = "昼にはメールを5件読んだな。"
messages.append({"role": "user", "content": user_input})
messages = agent_run(messages)
print(messages[-1])

# {'name': 'manage_memory', 'args': {'content': 'User read 5 emails at noon.', 'action': 'create'}, 'id': 'call_KKYSSdLq19B9fx0rRuxdXMAK', 'type': 'tool_call'}

# {'role': 'tool',
#  'tool_call_id': 'call_KKYSSdLq19B9fx0rRuxdXMAK',
#  'content': 'created memory b920feb0-61e7-493c-adf0-83095f4f2d48'}

さらにコメントを入れます。
この人、メールしか書いていないんですね。営業さんのアポ取り日？？？

3-3-6. メモリー機能を用いる（6回目の入力）

6回目の入力

user_input = "今日の昼までに何件のメールを処理したんだっけ？"
messages.append({"role": "user", "content": user_input})
messages = agent_run(messages)
print(messages[-1])

# response.content='今日の昼までに、あなたは25件のメールを書いて、5件のメールを読んだので、合計で30件のメールを処理しました。'

# {'role': 'assistant',
#  'content': '今日の昼までに、あなたは25件のメールを書いて、5件のメールを読んだので、合計で30件のメールを処理しました。',
#  'tool_calls': []}

おお、ちゃんと朝と昼のメール処理数を考えて30件と回答してくれました。
簡単な計算でもあるので、計算用のtoolsは使っていません。難しい計算なら計算表のtoolsを作るといいでしょうね。

3-4. ベクトルストアの状態の確認

ベクトルストアの確認

print(store.search(("memories", )))

# [Item(namespace=['memories'], key='6994456e-489e-47d1-a5c2-0fe601c70721', value={'content': 'User wrote 25 emails in the morning.'}, created_at='2025-03-04T08:48:52.141668+00:00', updated_at='2025-03-04T22:48:52.141668+00:00', score=None),
#  Item(namespace=['memories'], key='b920feb0-61e7-493c-adf0-83095f4f2d48', value={'content': 'User read 5 emails at noon.'}, created_at='2025-03-04T08:48:53.220401+00:00', updated_at='2025-03-04T22:48:53.220401+00:00', score=None)]

ベクトルストアを確認すると二つの情報が入っています。

• ユーザーは朝に25件のメールを書いた
• ユーザーは昼に5件のメールを読んだ

スンバラしい。

4. 終わりに

僕にはこういう使い方がイメージできました。皆さんはいかがでしょうか？
prebuilt-agentは何種類かあるのでこれから少しずつ試して行こうと思います。（ネタができたw）
日本語記事も読みながらではありますが、公式ドキュメントを頑張って読むことで少しずつ理解できるようになってきました。英語を読むのも早くなったし。

さぁ頑張るべ