LangchainのAgentExecutorについて理解する：個人的な学習メモ
はじめに
以下のようにAgentを定義して実行するとどのような処理が走るのか大まかに調査しました。
（個人的な学習メモのため、誤りがあるかもしれませんがご了承ください）
# https://python.langchain.com/docs/modules/agents/how_to/use_toolkits_with_openai_functions
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-0613")
agent = initialize_agent(
toolkit.get_tools(),
llm,
agent=AgentType.OPENAI_FUNCTIONS,
verbose=True,
agent_kwargs=agent_kwargs,
)
agent.run("how many different artists are there?")処理の流れ
initialize_agent 関数は AgentExecutor インスタンスを返す
https://github.com/hwchase17/langchain/blob/a673a51efa3e03aaa7c8c7e0004dc5ff9c536f2e/langchain/agents/initialize.py#L22AgentExecutorクラスは、Chainクラスを継承している
https://github.com/hwchase17/langchain/blob/a673a51efa3e03aaa7c8c7e0004dc5ff9c536f2e/langchain/agents/agent.py#L620runを実行すると、 Chain クラスの "__call__"メソッドが呼び出される。
"__call__"は_callメソッドを呼び出すが、子クラスのAgentExecutorがオーバーライドしている。
https://github.com/hwchase17/langchain/blob/a673a51efa3e03aaa7c8c7e0004dc5ff9c536f2e/langchain/chains/base.py#L444C9-L448C1
https://github.com/hwchase17/langchain/blob/a673a51efa3e03aaa7c8c7e0004dc5ff9c536f2e/langchain/agents/agent.py#L975AgentExecutorは_callのwhileの中で、_should_continueメソッドで一連のアクションを続けれるべきか判断しながら、_take_next_stepメソッドを繰り返し実行する。_take_next_stepには過去に実行したアクションとツールの一覧、ユーザからの入力(kwargs)が渡される
https://github.com/hwchase17/langchain/blob/a673a51efa3e03aaa7c8c7e0004dc5ff9c536f2e/langchain/agents/agent.py#L993C8-L993C8_take_next_stepはplanを呼び出す。この時過去のアクションとユーザからの入力(kwargs)を渡す。
https://github.com/hwchase17/langchain/blob/a673a51efa3e03aaa7c8c7e0004dc5ff9c536f2e/langchain/agents/agent.py#L993C8-L993C8planはオーバライドされている。openai_functions_agentの場合、過去のアクションの履歴とユーザからの入力(kwargs)はpromptとしてフォーマットされ言語モデルに渡される。また、ツール一の覧はfunctions関数でOpenAIのfunction API形式にフォーマットされ渡される。
https://github.com/hwchase17/langchain/blob/a673a51efa3e03aaa7c8c7e0004dc5ff9c536f2e/langchain/agents/openai_functions_agent/base.py#L188
https://github.com/hwchase17/langchain/blob/a673a51efa3e03aaa7c8c7e0004dc5ff9c536f2e/langchain/agents/openai_functions_agent/base.py#L212
フローチャート
1. 「initialize_agent」関数
|
| (「AgentExecutor」のインスタンスを返す)
|
V
2. 「AgentExecutor」クラス（「Chain」クラスを継承）
|
| (「run」を実行すると、"__call__" メソッドが呼び出される)
|
V
3. 「Chain」クラスの "__call__" メソッド
|
| ("_call" メソッドを呼び出す。これは「AgentExecutor」で上書きされている)
|
V
4. 「AgentExecutor」クラスの "_call" メソッド
|
| ("_should_continue" メソッドでアクションを続けるべきか判断しながら、"_take_next_step" メソッドを繰り返し実行する)
|
V
5. 「AgentExecutor」クラスの "_take_next_step" メソッド
|
| ("plan" メソッドを呼び出す。ここで過去の行動とユーザー入力を渡す)
|
V
6. 「AgentExecutor」クラスの "plan" メソッド (これは上書きされている)
|
| (「openai_functions_agent」の場合、過去の行動とユーザー入力をプロンプトとしてフォーマットし、それを言語モデルに渡す)
|
V
7. 言語モデル（例：「openai_functions_agent」）
最後に
AgentExecutorの処理の流れとしては、ユーザの入力と、過去のAgentの行動ログを元に次回のアクションを決定するという動作をループで繰り返すようでした。
より複雑で長期的な計画が必要となるタスクの場合は、先に全体の実行計画をTree of Thoughtsのアルゴリズムなどで作成してから、個別の処理でフィードバックをかけながら実行することでAgent の処理全体が改善するのではないかという仮説が浮かびました。
おまけ
agent_kwargsの使い方をご紹介します。
from langchain.schema.messages import (
AIMessage,
BaseMessage,
FunctionMessage,
SystemMessage,
)
agent_kwargs = {
"system_message": SystemMessage(
content="JOJOの奇妙な冒険3部の空条承太郎みたいな語尾にして"
),
"extra_prompt_messages": [MessagesPlaceholder(variable_name="memory")],
}
memory = ConversationBufferMemory(memory_key="memory", return_messages=True)mrkl = initialize_agent(tools, llm, agent=AgentType.OPENAI_FUNCTIONS, agent_kwargs=agent_kwargs, memory=memory, verbose=True)
mrkl.run(input ="JOJOの奇妙な冒険3部の空条承太郎のスタンドは？")> Finished chain.
空条承太郎のスタンドは「スタープラチナ」だぜ。