概要
AgentExecutorはLangChainのエージェント実行クラスです。Chainを継承し、Runnableを実装しています。エージェントの行動シーケンスを生成・評価しつつ、ツール呼び出しを制御します。
Runnableの説明はLangChain公式に譲ります。
主な特徴
- ツール呼び出し: エージェントが使用するツール一覧を管理し、ツール呼び出しを実行。
- 実行制御: ユーザー入力→計画生成(この部分は使用するエージェントの構成による)→ツール実行→結果生成のループを提供。
-
同期/非同期対応: 同期（
run/invoke）・非同期（arun/ainvoke）双方で実行可能。(これはRunnableであればどのようなものでも可能) -
ストリーミング: 出力を逐次取得できる（
stream/astream/astream_events）。(これもRunnableであればどのようなものでも可能) -
バッチ実行: 複数入力を一括実行できる（
batch/abatch/batch_as_completed/abatch_as_completed）。 - 拡張性: リトライ、フォールバック、リスナー、型バインドなどRunnableの付加メソッドを利用可能。
引数
| 引数名 | 説明 |
|---|---|
| agent |
BaseSingleActionAgent/BaseMultiActionAgent/RunnableAgent。実行の主体となるエージェント。 |
| callbacks | コールバックハンドラのリスト。on_chain_start、on_chain_end、on_chain_errorなどのライフサイクルフック。 |
| early_stopping_method | ループ停止時の動作。"force"（即時終了）または"generate"（最終ステップで生成）。 |
| handle_parsing_errors | 出力パーサーの例外処理方法。False（例外スロー）、True（LLMに観察結果として渡す）、文字列/コールバックも指定可。 |
| max_execution_time | 実行ループに費やす最大時間（秒）。 |
| max_iterations | 実行ループの最大ステップ数。Noneで制限なし（無限ループに注意）。 |
| memory |
BaseMemory実装。開始前後に状態を読み書き。 |
| metadata | 呼び出し毎に紐づく任意のメタデータ。 |
| return_intermediate_steps | 中間ステップを結果に含めるか。 |
| tags | 実行呼び出しに紐づくタグ一覧。 |
主なメソッド
実行関連
-
run(inputs)/invoke(inputs)/__call__(inputs)
同期でエージェントを実行し、最終結果を返します。 -
arun(inputs)/acall(inputs)
非同期実行。awaitで結果を取得。 -
stream(inputs)/astream(inputs)/astream_events(inputs)
出力をストリーミングで受け取り、部分結果やイベントを逐次処理可能。
webアプリケーションでstreamする場合はエージェントの実行をgeneratorとして、出力を整形する関数でラップすると良いです。
(今後の記事でsseでの送信についてもまとめたいです。)
utils
-
bind(**fixed_args)
固定引数をバインドした新しいRunnableを生成。 -
lookup_tool(tool_name)
設定されたツール情報を参照。 -
save(path)/save_agent(path)
エージェント設定をファイル等に永続化。(設定の仕方次第でローカルでも何でも対応はできます。)
Runnable（付加可能なメソッド）
AgentExecutorはRunnableを継承し、以下のメソッドで動作をカスタマイズ可能です。
-
with_config(**config)
設定を変更した新しいRunnableを生成。 -
with_retry(stop_after_attempt=3, retry_if_exception_type=(Exception,), ...)
例外発生時のリトライ処理を追加。 -
with_listeners(on_start, on_end, on_error)
実行ライフサイクルのリスナーをバインド。 -
with_fallbacks(fallbacks, exceptions_to_handle=(Exception,), exception_key=None)
fallback時のランナブルを定義し、失敗時にを実行。 -
with_types(input_type, output_type)
入力/出力の型バインド。
使用例
from langchain.agents import initialize_agent, AgentExecutor, Tool
from langchain_openai import AzureChatOpenAI
llm = AzureChatOpenAI(model="gpt-4.1")
# プロンプトはよく考える必要あり
prompt = ChatPromptTemplate.from_template(
"""ConversationHistory: {history}
Question: {input}
Thought: Let's think step by step.
制約:
* 必要に応じて登録済みのツールを組み合わせて利用してください。
* ツール利用の必要がない場合は直接回答してください。
* 回答生成に必要な情報が足りない場合は聞き返すこと。
Answer: {agent_scratchpad}"""
)
# ツール定義
tools = [Tool(name="search", func=search_func)] # 適当に定義
# エージェント初期化
agent = create_tool_calling_agent(
llm=llm, # langchain_openai等
tools=tools,
prompt=prompt
)
# Executor生成
executor = AgentExecutor(agent=agent, tools=tools)
# 会話ログ等を入れておくことを想定
history = []
# ループを作成して、継続会話にして会話ログを入れていくとそれっぽくなる
result = executor.invoke({"history": history, "input": "〜〜について教えて"})
まとめ
- AgentExecutorはツール連携と実行ループを一元管理するエージェント実行基盤です。
- 同期/非同期、ストリーミング、バッチ処理など複数の実行モードに対応しています。
- Runnableにより、リトライ、フォールバック、リスナー、型バインドなどの拡張が可能です。
- metadataやcallbacksを活用して実行コンテキストやログ取得を柔軟に管理できます。
- シンプルなAPIでエージェントの構築・保存・再利用が容易に行えます。