class langgraph_agent_toolkit.agents.agent_executor.AgentExecutor(*args)[source][source]

Bases: object

Handles the loading, execution and saving logic for different LangGraph agents.

Initialize the AgentExecutor by importing agents.

Parameters:

*args – Variable length strings specifying the agents to import, e.g., “langgraph_agent_toolkit.agents.blueprints.react.agent:react_agent”.

Raises:

ValueError – If no agents are provided.

__init__(*args)[source][source]

Initialize the AgentExecutor by importing agents.

Parameters:

*args – Variable length strings specifying the agents to import, e.g., “langgraph_agent_toolkit.agents.blueprints.react.agent:react_agent”.

Raises:

ValueError – If no agents are provided.

load_agents_from_imports(args)[source][source]

Dynamically imports agents based on the provided import strings.

Parameters:

args (tuple)

Return type:

None

get_agent(agent_id)[source][source]

Get an agent by its ID.

Parameters:

agent_id (str) – The ID of the agent to retrieve

Returns:

The requested Agent instance

Raises:

KeyError – If the agent_id is not found

Return type:

Agent

get_all_agent_info()[source][source]

Get information about all available agents.

Returns:

A list of AgentInfo objects containing agent IDs and descriptions

Return type:

list[AgentInfo]

add_agent(agent_id, agent)[source][source]

Add a new agent to the executor.

Parameters:

  • agent_id (str) – The ID to assign to the agent

  • agent (Agent) – The Agent instance to add

Return type:

None

static handle_agent_errors(func)[source][source]

Handle errors occurring during agent execution.

Specifically handles GraphRecursionError and other exceptions.

Parameters:

func (Callable[[...], T]) – The function to decorate

Returns:

The decorated function

Return type:

Callable[[…], T]

async invoke(agent_id, input, thread_id=None, user_id=None, model_name=None, model_provider=None, model_config_key=None, agent_config=None, recursion_limit=None)[source][source]

Invoke an agent with a message and return the response.

Parameters:

  • agent_id (str) – ID of the agent to invoke

  • input (Dict[str, Any]) – User message to send to the agent

  • thread_id (str | None) – Optional thread ID for conversation history

  • user_id (str | None) – Optional user ID for the agent

  • model_name (str | None) – Optional model name to override the default

  • model_provider (str | None) – Optional model provider to override the default

  • model_config_key (str | None) – Optional model config key to override the default

  • agent_config (Dict[str, Any] | None) – Optional additional configuration for the agent

  • recursion_limit (int | None) – Optional recursion limit for the agent

Returns:

The agent’s response

Return type:

ChatMessage

stream(agent_id, input, thread_id=None, user_id=None, model_name=None, model_provider=None, model_config_key=None, stream_tokens=True, agent_config=None, recursion_limit=None)[source][source]

Stream an agent’s response to a message, yielding either tokens or messages.

Parameters:

  • agent_id (str) – ID of the agent to invoke

  • input (Dict[str, Any]) – User message to send to the agent

  • thread_id (str | None) – Optional thread ID for conversation history

  • user_id (str | None) – Optional user ID for the agent

  • model_name (str | None) – Optional model name to override the default

  • model_provider (str | None) – Optional model provider to override the default

  • model_config_key (str | None) – Optional model config key to override the default

  • stream_tokens (bool) – Whether to stream individual tokens

  • agent_config (Dict[str, Any] | None) – Optional additional configuration for the agent

  • recursion_limit (int | None) – Optional recursion limit for the agent

Yields:

Either ChatMessage objects for full messages or strings for token chunks

Return type:

AsyncGenerator[str | ChatMessage, None]

save(path, agent_ids=None)[source][source]

Save agents to disk using joblib.

Parameters:

  • path (str) – Directory path where to save agents

  • agent_ids (List[str] | None) – List of agent IDs to save. If None, saves all agents.

Return type:

None

load_saved_agents(path)[source][source]

Load saved agents from disk using joblib.

Parameters:

path (str) – Directory path from which to load agents

Return type:

None