従来の LangChainエージェント から LangGraphエージェント への移行手順

以下の記事が面白かったので、簡単にまとめました。

・How to migrate from legacy LangChain agents to LangGraph

1. はじめに

従来の「LangChainエージェント」から「LangGraph エージェント」への移行手順を解説します。LangChainエージェント (特に AgentExecutor) には複数の構成パラメータがあります。これらのパラメータを LangGraphエージェント(ReactAgentExecutor) にどのようにマップされるかを示します。

2. 準備

(1) パッケージのインストール。

# パッケージのインストール
!pip install -U langgraph langchain langchain-openai

(2) 環境変数の準備。
左端の鍵アイコンで「OPENAI_API_KEY」を設定してからセルを実行してください。

import os
from google.colab import userdata

# 環境変数の準備 (左端の鍵アイコンでOPENAI_API_KEYを設定)
os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")

3. 基本的な使い方

tool-calling ReActエージェントの基本的な作成手順は同じです。

(1) モデルとツールとクエリの準備。

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# モデルの準備
model = ChatOpenAI(model="gpt-4o")

# ツールの準備
@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2

tools = [magic_function]

# クエリの準備
query = "what is the value of magic_function(3)?"

(2) AgentExecutorの準備と実行。
AgentExecutorの場合、agent_scratchpad の placeholder を含むプロンプトを定義します。 エージェントは次のよう呼び出すことができます。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

# プロンプトテンプレートの準備
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant"),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)

# AgentExecutorの準備
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# AgentExecutorの実行
agent_executor.invoke({"input": query})

{'input': 'what is the value of magic_function(3)?',
 'output': 'The value of `magic_function(3)` is 5.'}

(3) ReactAgentExecutorの準備と実行。
ReactAgentExecutorは、メッセージリストによって定義される状態を管理します。エージェントの出力にツール呼び出しがなくなるまで、処理が続行されます。出力には、グラフの状態全体 (この場合は会話履歴) が含まれます。

from langgraph.prebuilt import create_react_agent

# ReactAgentExecutorの準備
app = create_react_agent(model, tools)

# ReactAgentExecutorの実行
messages = app.invoke({"messages": [("human", query)]})
{
    "input": query,
    "output": messages["messages"][-1].content,
}

{'input': 'what is the value of magic_function(3)?',
 'output': 'The value of `magic_function(3)` is 5.'}

# 会話履歴の準備
message_history = messages["messages"]

# 新クエリの準備
new_query = "Pardon?"

# ReactAgentExecutorの実行
messages = app.invoke({"messages": message_history + [("human", new_query)]})
{
    "input": new_query,
    "output": messages["messages"][-1].content,
}

{'input': 'Pardon?',
 'output': 'The result of applying `magic_function` to the input 3 is 5.'}

3. プロンプトテンプレート

AgentExecutorでは、プロンプトテンプレートを渡す必要があります。これを使用してエージェントを制御します。

ReactAgentExecutorでは、いくつかの方法で同様の制御を実現できます。

(1) システムメッセージを入力として渡す
(2) システムメッセージでエージェントを初期化
(3) モデルに渡す前にメッセージ変換関数でエージェントを初期化

(1) AgentExecutorの準備と実行。
プロンプトテンプレートで、日本語での応答を促すためのカスタム指示を指定しています。

# プロンプトテンプレートの準備
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. Respond only in Japanese."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)

# AgentExecutorの準備
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# AgentExecutorの実行
agent_executor.invoke({"input": query})

{'input': 'what is the value of magic_function(3)?',
 'output': 'magic_function(3) の値は 5 です。'}

(2) ReactAgentExecutorの準備と実行。
カスタムシステムメッセージをReactAgentExecutorに渡します。

from langchain_core.messages import SystemMessage
from langgraph.prebuilt import create_react_agent

# システムメッセージの準備
system_message = "You are a helpful assistant. Respond only in Japanese."
# これは SystemMessage オブジェクトである可能性もある
# system_message = SystemMessage(content="You are a helpful assistant. Respond only in Japanese.")

# ReactAgentExecutorの準備
app = create_react_agent(model, tools, messages_modifier=system_message)

# ReactAgentExecutorの実行
messages = app.invoke({"messages": [("user", query)]})

(3) ReactAgentExecutorを実行。
任意の関数をReactAgentExecutorに渡します。この関数はメッセージのリストを受け取り、メッセージリストを出力する必要があります。 ここでは、あらゆる種類のメッセージの任意の書式設定を行うことができます。

from langchain_core.messages import AnyMessage
from langgraph.prebuilt import create_react_agent

# プロンプトテンプレートの準備　
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. Respond only in Japanese."),
        ("placeholder", "{messages}"),
    ]
)

# メッセージ変更関数の準備
def _modify_messages(messages: list[AnyMessage]):
    return prompt.invoke({"messages": messages}).to_messages() + [
        ("user", "Also say 'Pandamonium!' after the answer.")
    ]

# ReactAgentExecutorの準備
app = create_react_agent(model, tools, messages_modifier=_modify_messages)

# ReactAgentExecutorの実行
messages = app.invoke({"messages": [("human", query)]})
print(
    {
        "input": query,
        "output": messages["messages"][-1].content,
    }
)

{'input': 'what is the value of magic_function(3)?', 'output': 'magic_function(3) の値は 5 です。Pandamonium!'}

4. メモリ

(1) メモリ付きAgentExecutorの準備と実行。
AgentExecutorにメモリを追加することで、複数ターンの会話を行うことができます。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain.memory import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# モデルの準備
model = ChatOpenAI(model="gpt-4o")

# メモリの準備
memory = ChatMessageHistory(session_id="test-session")

# プロンプトテンプレートの準備
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        # First put the history
        ("placeholder", "{chat_history}"),
        # Then the new input
        ("human", "{input}"),
        # Finally the scratchpad
        ("placeholder", "{agent_scratchpad}"),
    ]
)

# ツールの準備
@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2

tools = [magic_function]

# AgentExecutorの準備
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# メモリ付きAgentExecutorの準備
agent_with_chat_history = RunnableWithMessageHistory(
    agent_executor,
    # 現実世界のほとんどのシナリオではセッションIDが必要
    # 単純なメモリ内のChatMessageHistoryを使用しているためここでは使用されていない
    lambda session_id: memory,
    input_messages_key="input",
    history_messages_key="chat_history",
)

# メモリ付きAgentExecutorの実行
config = {"configurable": {"session_id": "test-session"}}
print(
    agent_with_chat_history.invoke(
        {"input": "Hi, I'm polly! What's the output of magic_function of 3?"}, config
    )["output"]
)
print("---")
print(agent_with_chat_history.invoke({"input": "Remember my name?"}, config)["output"])
print("---")
print(
    agent_with_chat_history.invoke({"input": "what was that output again?"}, config)[
        "output"
    ]
)

Hi Polly! The output of the magic_function for the input 3 is 5.
---
Yes, your name is Polly!
---
The output of the magic_function for the input 3 was 5.

(2) メモリ付きReactAgentExecutorの準備と実行。

from langchain_core.messages import SystemMessage
from langgraph.checkpoint import MemorySaver  # an in-memory checkpointer
from langgraph.prebuilt import create_react_agent

# システムメッセージの準備
system_message = "You are a helpful assistant."
# これは SystemMessage オブジェクトである可能性もある
# system_message = SystemMessage(content="You are a helpful assistant. Respond only in Japanese.")

# メモリの準備
memory = MemorySaver()

# メモリ付きReactAgentExecutorの準備
app = create_react_agent(
    model, tools, messages_modifier=system_message, checkpointer=memory
)

# メモリ付きReactAgentExecutorの実行
config = {"configurable": {"thread_id": "test-thread"}}
print(
    app.invoke(
        {
            "messages": [
                ("user", "Hi, I'm polly! What's the output of magic_function of 3?")
            ]
        },
        config,
    )["messages"][-1].content
)
print("---")
print(
    app.invoke({"messages": [("user", "Remember my name?")]}, config)["messages"][
        -1
    ].content
)
print("---")
print(
    app.invoke({"messages": [("user", "what was that output again?")]}, config)[
        "messages"
    ][-1].content
)

Hi Polly! The output of the magic_function for the input 3 is 5.
---
Yes, your name is Polly!
---
The output of the magic_function for the input 3 was 5.

5. ステップの段階的出力

(1) AgentExecutorのステップの段階的出力。
stream (または async astream) または iter を使用してステップを段階的出力できます。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# モデルの準備
model = ChatOpenAI(model="gpt-4o")

# プロンプトテンプレートの準備
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)

# ツールの準備
@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2

tools = [magic_function]

# AgentExecutorの準備
agent = create_tool_calling_agent(model, tools, prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# ステップの段階的出力
for step in agent_executor.stream({"input": query}):
    print(step)

{'actions': [ToolAgentAction(tool='magic_function', tool_input={'input': 3}, log="\nInvoking: `magic_function` with `{'input': 3}`\n\n\n", message_log=[AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_pvvuJJVv5V34ISvHqr9i79Gd', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls'}, id='run-7a47db09-3c10-4574-a768-496688e8f11b', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_pvvuJJVv5V34ISvHqr9i79Gd'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_pvvuJJVv5V34ISvHqr9i79Gd', 'index': 0}])], tool_call_id='call_pvvuJJVv5V34ISvHqr9i79Gd')], 'messages': [AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_pvvuJJVv5V34ISvHqr9i79Gd', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls'}, id='run-7a47db09-3c10-4574-a768-496688e8f11b', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_pvvuJJVv5V34ISvHqr9i79Gd'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_pvvuJJVv5V34ISvHqr9i79Gd', 'index': 0}])]}
{'steps': [AgentStep(action=ToolAgentAction(tool='magic_function', tool_input={'input': 3}, log="\nInvoking: `magic_function` with `{'input': 3}`\n\n\n", message_log=[AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_pvvuJJVv5V34ISvHqr9i79Gd', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls'}, id='run-7a47db09-3c10-4574-a768-496688e8f11b', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_pvvuJJVv5V34ISvHqr9i79Gd'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_pvvuJJVv5V34ISvHqr9i79Gd', 'index': 0}])], tool_call_id='call_pvvuJJVv5V34ISvHqr9i79Gd'), observation=5)], 'messages': [FunctionMessage(content='5', name='magic_function')]}
{'output': 'The value of `magic_function(3)` is 5.', 'messages': [AIMessage(content='The value of `magic_function(3)` is 5.')]}

(1) ReactAgentExecutorのステップの段階的出力。
streamを使用してステップを段階的出力できます。

from langchain_core.messages import AnyMessage
from langgraph.prebuilt import create_react_agent

# プロンプトテンプレートの準備
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("placeholder", "{messages}"),
    ]
)

# メッセージ変更関数の準備
def _modify_messages(messages: list[AnyMessage]):
    return prompt.invoke({"messages": messages}).to_messages()

# ReactAgentExecutorの準備
app = create_react_agent(model, tools, messages_modifier=_modify_messages)

# ステップの段階的出力
for step in app.stream({"messages": [("human", query)]}, stream_mode="updates"):
    print(step)

{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_yTjXXibj76tyFyPRa1soLo0S', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 70, 'total_tokens': 84}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-b275f314-c42e-4e77-9dec-5c23f7dbd53b-0', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_yTjXXibj76tyFyPRa1soLo0S'}])]}}
{'tools': {'messages': [ToolMessage(content='5', name='magic_function', id='41c5f227-528d-4483-a313-b03b23b1d327', tool_call_id='call_yTjXXibj76tyFyPRa1soLo0S')]}}
{'agent': {'messages': [AIMessage(content='The value of `magic_function(3)` is 5.', response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 93, 'total_tokens': 107}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'stop', 'logprobs': None}, id='run-0ef12b6e-415d-4758-9b62-5e5e1b350072-0')]}}

6. return_intermediate_steps

(1) AgentExecutorの中間ステップの確認。
AgentExecutor でreturn_intermediate_stepsを指定すると、エージェントのアクション (ツール呼び出しなど) とその結果を組み合わせた中間ステップを取得できます。

agent_executor = AgentExecutor(agent=agent, tools=tools, return_intermediate_steps=True)
result = agent_executor.invoke({"input": query})
print(result["intermediate_steps"])

[(ToolAgentAction(tool='magic_function', tool_input={'input': 3}, log="\nInvoking: `magic_function` with `{'input': 3}`\n\n\n", message_log=[AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_ABI4hftfEdnVgKyfF6OzZbca', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls'}, id='run-837e794f-cfd8-40e0-8abc-4d98ced11b75', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_ABI4hftfEdnVgKyfF6OzZbca'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_ABI4hftfEdnVgKyfF6OzZbca', 'index': 0}])], tool_call_id='call_ABI4hftfEdnVgKyfF6OzZbca'), 5)]

(2) ReactAgentExecutorの中間ステップの確認。
ReactAgentExecutor はメッセージリストを確認するだけで、中間のステップを確認できます。

from langgraph.prebuilt import create_react_agent

app = create_react_agent(model, tools=tools)
messages = app.invoke({"messages": [("human", query)]})
messages

{'messages': [HumanMessage(content='what is the value of magic_function(3)?', id='0f63e437-c4d8-4da9-b6f5-b293ebfe4a64'),
  AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_S96v28LlI6hNkQrNnIio0JPh', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 64, 'total_tokens': 78}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-ffef7898-14b1-4537-ad90-7c000a8a5d25-0', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_S96v28LlI6hNkQrNnIio0JPh'}]),
  ToolMessage(content='5', name='magic_function', id='fbd9df4e-1dda-4d3e-9044-b001f7875476', tool_call_id='call_S96v28LlI6hNkQrNnIio0JPh'),
  AIMessage(content='The value of `magic_function(3)` is 5.', response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 87, 'total_tokens': 101}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'stop', 'logprobs': None}, id='run-e5d94c54-d9f4-45cd-be8e-a9101a8d88d6-0')]}

7. max_iterations

AgentExecutor の max_iterations は、ReactAgentExecutor では recursion_limit で制御します。

AgentExecutorでは、「iterations」にはツール呼び出しと実行がすべて含まれることに注意してください。 ReactAgentExecutorでは、各ステップが再帰制限に影響するため、同等の結果を得るには 2 を乗算 (および 1 を加算) する必要があります。

ReactAgentExecutorでは再帰制限に達すると、特定の例外が発生するので、キャッチして管理します。

(1) ツールの準備。

# ツールの準備
@tool
def magic_function(input: str) -> str:
    """Applies a magic function to an input."""
    return "Sorry, there was an error. Please try again."


tools = [magic_function]

(2) AgentExecutorの準備と実行。

# プロンプトテンプレートの準備
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. Respond only in Japanese."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)

# AgentExecutorの準備
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=3,
)

# AgentExecutorの実行
agent_executor.invoke({"input": query})

{'input': 'what is the value of magic_function(3)?',
 'output': 'magic_function(3)の値は5です。'}

(3) ReactAgentExecutorの準備と実行。

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent

RECURSION_LIMIT = 2 * 3 + 1

# ReactAgentExecutorの準備
app = create_react_agent(model, tools=tools)

# ReactAgentExecutorの実行
try:
    for chunk in app.stream(
        {"messages": [("human", query)]},
        {"recursion_limit": RECURSION_LIMIT},
        stream_mode="values",
    ):
        print(chunk["messages"][-1])
except GraphRecursionError:
    print({"input": query, "output": "Agent stopped due to max iterations."})

('human', 'what is the value of magic_function(3)?')
content='' additional_kwargs={'tool_calls': [{'id': 'call_pFdKcCu5taDTtOOfX14vEDRp', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}]} response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 64, 'total_tokens': 78}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-25836468-ba7e-43be-a7cf-76bba06a2a08-0' tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_pFdKcCu5taDTtOOfX14vEDRp'}]
content='Sorry, there was an error. Please try again.' name='magic_function' id='1a08b883-9c7b-4969-9e9b-67ce64cdcb5f' tool_call_id='call_pFdKcCu5taDTtOOfX14vEDRp'
content='It seems there was an error when trying to apply the magic function. Let me try again.' additional_kwargs={'tool_calls': [{'id': 'call_DA0lpDIkBFg2GHy4WsEcZG4K', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}]} response_metadata={'token_usage': {'completion_tokens': 34, 'prompt_tokens': 97, 'total_tokens': 131}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-d571b774-0ea3-4e35-8b7d-f32932c3f3cc-0' tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_DA0lpDIkBFg2GHy4WsEcZG4K'}]
content='Sorry, there was an error. Please try again.' name='magic_function' id='0b45787b-c82a-487f-9a5a-de129c30460f' tool_call_id='call_DA0lpDIkBFg2GHy4WsEcZG4K'
content='It appears that there is a consistent issue when trying to apply the magic function to the input "3." This could be due to various reasons, such as the input not being in the correct format or an internal error.\n\nIf you have any other questions or if there\'s something else you\'d like to try, please let me know!' response_metadata={'token_usage': {'completion_tokens': 66, 'prompt_tokens': 153, 'total_tokens': 219}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'stop', 'logprobs': None} id='run-50a962e6-21b7-4327-8dea-8e2304062627-0'

8. max_execution_time

(1) AgentExecutorの準備と実行。
AgentExecutor は max_execution_time で合計時間制限を超えた実行を中止できるようにします。

import time

# ツールの準備
@tool
def magic_function(input: str) -> str:
    """Applies a magic function to an input."""
    time.sleep(2.5)
    return "Sorry, there was an error. Please try again."

tools = [magic_function]

# AgentExecutorの準備
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_execution_time=2,
    verbose=True,
)

# AgentExecutorの実行
agent_executor.invoke({"input": query})

{'input': 'what is the value of magic_function(3)?',
 'output': 'Agent stopped due to max iterations.'}

(3) ReactAgentExecutorの準備と実行。
ReactAgentExecutorはstep_timeoutで各ステップの最大タイムアウトを設定します。

from langgraph.prebuilt import create_react_agent

# ReactAgentExecutorの準備
app = create_react_agent(model, tools=tools)
# ここで各ステップの最大タイムアウトを設定
app.step_timeout = 2

# ReactAgentExecutorの実行
try:
    for chunk in app.stream({"messages": [("human", query)]}):
        print(chunk)
        print("------")
except TimeoutError:
    print({"input": query, "output": "Agent stopped due to max iterations."})

{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_HaQkeCwD5QskzJzFixCBacZ4', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 64, 'total_tokens': 78}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-596c9200-771f-436d-8576-72fcb81620f1-0', tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_HaQkeCwD5QskzJzFixCBacZ4'}])]}}
------
{'input': 'what is the value of magic_function(3)?', 'output': 'Agent stopped due to max iterations.'}

(4) ReactAgentExecutorの準備と実行。
ReactAgentExecutorで合計時間制限を超えた実行を中止させるには、Python stdlib asyncio を直接使用します。

import asyncio
from langgraph.prebuilt import create_react_agent

# ReactAgentExecutorの準備
app = create_react_agent(model, tools=tools)


async def stream(app, inputs):
    async for chunk in app.astream({"messages": [("human", query)]}):
        print(chunk)
        print("------")

# ReactAgentExecutorの実行
try:
    task = asyncio.create_task(stream(app, {"messages": [("human", query)]}))
    await asyncio.wait_for(task, timeout=3)
except TimeoutError:
    print("Task Cancelled.")

{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_4agJXUHtmHrOOMogjF6ZuzAv', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 64, 'total_tokens': 78}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-a1c77db7-405f-43d9-8d57-751f2ca1a58c-0', tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_4agJXUHtmHrOOMogjF6ZuzAv'}])]}}
------
Task Cancelled.

9. early_stopping_method

(1) AgentExecutorの準備と実行。
AgentExecutorではearly_stopping_methodで、「Agent stopped due to iteration limit or time limit」という文字列を返す (force) か、LLM に最終応答を促す (generate) を指定することができます。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# モデルの準備
model = ChatOpenAI(model="gpt-4o")

# プロンプトテンプレートの準備
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)

# ツールの準備
@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return "Sorry there was an error, please try again."

tools = [magic_function]

# AgentExecutorの準備
agent = create_tool_calling_agent(model, tools, prompt=prompt)
agent_executor = AgentExecutor(
    agent=agent, tools=tools, early_stopping_method="force", max_iterations=1
)

# AgentExecutorの実行
result = agent_executor.invoke({"input": query})
print("Output with early_stopping_method='force':")
print(result["output"])

Output with early_stopping_method='force':
Agent stopped due to max iterations.

(2) ReactAgentExecutorの準備と実行。
ReactAgentExecutorでは完全な状態にアクセスできるため、エージェントの外部で応答動作を明示的に処理できます。

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent

RECURSION_LIMIT = 2 * 1 + 1

# ReactAgentExecutorの準備
app = create_react_agent(model, tools=tools)

# ReactAgentExecutorの実行
try:
    for chunk in app.stream(
        {"messages": [("human", query)]},
        {"recursion_limit": RECURSION_LIMIT},
        stream_mode="values",
    ):
        print(chunk["messages"][-1])
except GraphRecursionError:
    print({"input": query, "output": "Agent stopped due to max iterations."})

('human', 'what is the value of magic_function(3)?')
content='' additional_kwargs={'tool_calls': [{'id': 'call_bTURmOn9C8zslmn0kMFeykIn', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]} response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 64, 'total_tokens': 78}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-0844a504-7e6b-4ea6-a069-7017e38121ee-0' tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_bTURmOn9C8zslmn0kMFeykIn'}]
content='Sorry there was an error, please try again.' name='magic_function' id='00d5386f-eb23-4628-9a29-d9ce6a7098cc' tool_call_id='call_bTURmOn9C8zslmn0kMFeykIn'
content='' additional_kwargs={'tool_calls': [{'id': 'call_JYqvvvWmXow2u012DuPoDHFV', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]} response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 96, 'total_tokens': 110}, 'model_name': 'gpt-4o', 'system_fingerprint': 'fp_729ea513f7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-b73b1b1c-c829-4348-98cd-60b315c85448-0' tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_JYqvvvWmXow2u012DuPoDHFV'}]
{'input': 'what is the value of magic_function(3)?', 'output': 'Agent stopped due to max iterations.'}

10. trim_intermediate_steps

(1) AgentExecutorの準備と実行。
AgentExecutor では trim_intermediate_steps で、長時間実行されているエージェントの中間ステップをトリミングできます。これは、整数 (エージェントが最後の N ステップを保持する必要があることを示す) またはカスタム関数のいずれかです。

たとえば、値をトリミングして、エージェントが最新の中間ステップのみを参照できるようにすることができます。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# モデルの準備
model = ChatOpenAI(model="gpt-4o")

# プロンプトテンプレートの準備
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


magic_step_num = 1

# ツールの準備
@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    global magic_step_num
    print(f"Call number: {magic_step_num}")
    magic_step_num += 1
    return input + magic_step_num


tools = [magic_function]

# AgentExecutorの準備
agent = create_tool_calling_agent(model, tools, prompt=prompt)

def trim_steps(steps: list):
    # Let's give the agent amnesia
    return []

agent_executor = AgentExecutor(
    agent=agent, tools=tools, trim_intermediate_steps=trim_steps
)

# クエリの準備
query = "Call the magic function 4 times in sequence with the value 3. You cannot call it multiple times at once."

# ステップの段階的出力
for step in agent_executor.stream({"input": query}):
    pass

Call number: 1
Call number: 2
Call number: 3
Call number: 4
Call number: 5
Call number: 6
Call number: 7
Call number: 8
Call number: 9
Call number: 10
Call number: 11
Call number: 12
Call number: 13
Call number: 14
``````output
Stopping agent prematurely due to triggering stop condition
``````output
Call number: 15

(2) ReactAgentExecutorの準備と実行。
ReactAgentExecutorでは、前と同じようにmessages_modifierを使用できます。

from langchain_core.messages import AnyMessage
from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent

magic_step_num = 1

# ツールの準備
@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    global magic_step_num
    print(f"Call number: {magic_step_num}")
    magic_step_num += 1
    return input + magic_step_num

tools = [magic_function]

# メッセージ変更関数の準備
def _modify_messages(messages: list[AnyMessage]):
    # Give the agent amnesia, only keeping the original user query
    return [("system", "You are a helpful assistant"), messages[0]]

# ReactAgentExecutorの準備
app = create_react_agent(model, tools, messages_modifier=_modify_messages)

# ReactAgentExecutorの実行
try:
    for step in app.stream({"messages": [("human", query)]}, stream_mode="updates"):
        pass
except GraphRecursionError as e:
    print("Stopping agent prematurely due to triggering stop condition")

Call number: 1
Call number: 2
Call number: 3
Call number: 4
Call number: 5
Call number: 6
Call number: 7
Call number: 8
Call number: 9
Call number: 10
Call number: 11
Call number: 12
Stopping agent prematurely due to triggering stop condition

OpenAI GPT-4V／ChatGPT／GPTs 人工知能プログラミング実践入門 | 株式会社ボーンデジタル