|
|
"""Functionality for loading agents.""" |
|
|
|
|
|
import json |
|
|
import logging |
|
|
from pathlib import Path |
|
|
from typing import Any, List, Optional, Union |
|
|
|
|
|
import yaml |
|
|
from langchain_core._api import deprecated |
|
|
from langchain_core.language_models import BaseLanguageModel |
|
|
from langchain_core.tools import Tool |
|
|
|
|
|
from langchain.agents.agent import BaseMultiActionAgent, BaseSingleActionAgent |
|
|
from langchain.agents.types import AGENT_TO_CLASS |
|
|
from langchain.chains.loading import load_chain, load_chain_from_config |
|
|
|
|
|
logger = logging.getLogger(__file__) |
|
|
|
|
|
URL_BASE = "https://raw.githubusercontent.com/hwchase17/langchain-hub/master/agents/" |
|
|
|
|
|
|
|
|
def _load_agent_from_tools( |
|
|
config: dict, llm: BaseLanguageModel, tools: List[Tool], **kwargs: Any |
|
|
) -> Union[BaseSingleActionAgent, BaseMultiActionAgent]: |
|
|
config_type = config.pop("_type") |
|
|
if config_type not in AGENT_TO_CLASS: |
|
|
raise ValueError(f"Loading {config_type} agent not supported") |
|
|
|
|
|
agent_cls = AGENT_TO_CLASS[config_type] |
|
|
combined_config = {**config, **kwargs} |
|
|
return agent_cls.from_llm_and_tools(llm, tools, **combined_config) |
|
|
|
|
|
|
|
|
@deprecated("0.1.0", removal="1.0") |
|
|
def load_agent_from_config( |
|
|
config: dict, |
|
|
llm: Optional[BaseLanguageModel] = None, |
|
|
tools: Optional[List[Tool]] = None, |
|
|
**kwargs: Any, |
|
|
) -> Union[BaseSingleActionAgent, BaseMultiActionAgent]: |
|
|
"""Load agent from Config Dict. |
|
|
|
|
|
Args: |
|
|
config: Config dict to load agent from. |
|
|
llm: Language model to use as the agent. |
|
|
tools: List of tools this agent has access to. |
|
|
kwargs: Additional keyword arguments passed to the agent executor. |
|
|
|
|
|
Returns: |
|
|
An agent executor. |
|
|
|
|
|
Raises: |
|
|
ValueError: If agent type is not specified in the config. |
|
|
""" |
|
|
if "_type" not in config: |
|
|
raise ValueError("Must specify an agent Type in config") |
|
|
load_from_tools = config.pop("load_from_llm_and_tools", False) |
|
|
if load_from_tools: |
|
|
if llm is None: |
|
|
raise ValueError( |
|
|
"If `load_from_llm_and_tools` is set to True, " |
|
|
"then LLM must be provided" |
|
|
) |
|
|
if tools is None: |
|
|
raise ValueError( |
|
|
"If `load_from_llm_and_tools` is set to True, " |
|
|
"then tools must be provided" |
|
|
) |
|
|
return _load_agent_from_tools(config, llm, tools, **kwargs) |
|
|
config_type = config.pop("_type") |
|
|
|
|
|
if config_type not in AGENT_TO_CLASS: |
|
|
raise ValueError(f"Loading {config_type} agent not supported") |
|
|
|
|
|
agent_cls = AGENT_TO_CLASS[config_type] |
|
|
if "llm_chain" in config: |
|
|
config["llm_chain"] = load_chain_from_config(config.pop("llm_chain")) |
|
|
elif "llm_chain_path" in config: |
|
|
config["llm_chain"] = load_chain(config.pop("llm_chain_path")) |
|
|
else: |
|
|
raise ValueError("One of `llm_chain` and `llm_chain_path` should be specified.") |
|
|
if "output_parser" in config: |
|
|
logger.warning( |
|
|
"Currently loading output parsers on agent is not supported, " |
|
|
"will just use the default one." |
|
|
) |
|
|
del config["output_parser"] |
|
|
|
|
|
combined_config = {**config, **kwargs} |
|
|
return agent_cls(**combined_config) |
|
|
|
|
|
|
|
|
@deprecated("0.1.0", removal="1.0") |
|
|
def load_agent( |
|
|
path: Union[str, Path], **kwargs: Any |
|
|
) -> Union[BaseSingleActionAgent, BaseMultiActionAgent]: |
|
|
"""Unified method for loading an agent from LangChainHub or local fs. |
|
|
|
|
|
Args: |
|
|
path: Path to the agent file. |
|
|
kwargs: Additional keyword arguments passed to the agent executor. |
|
|
|
|
|
Returns: |
|
|
An agent executor. |
|
|
|
|
|
Raises: |
|
|
RuntimeError: If loading from the deprecated github-based |
|
|
Hub is attempted. |
|
|
""" |
|
|
if isinstance(path, str) and path.startswith("lc://"): |
|
|
raise RuntimeError( |
|
|
"Loading from the deprecated github-based Hub is no longer supported. " |
|
|
"Please use the new LangChain Hub at https://smith.langchain.com/hub " |
|
|
"instead." |
|
|
) |
|
|
return _load_agent_from_file(path, **kwargs) |
|
|
|
|
|
|
|
|
def _load_agent_from_file( |
|
|
file: Union[str, Path], **kwargs: Any |
|
|
) -> Union[BaseSingleActionAgent, BaseMultiActionAgent]: |
|
|
"""Load agent from file.""" |
|
|
valid_suffixes = {"json", "yaml"} |
|
|
|
|
|
if isinstance(file, str): |
|
|
file_path = Path(file) |
|
|
else: |
|
|
file_path = file |
|
|
|
|
|
if file_path.suffix[1:] == "json": |
|
|
with open(file_path) as f: |
|
|
config = json.load(f) |
|
|
elif file_path.suffix[1:] == "yaml": |
|
|
with open(file_path, "r") as f: |
|
|
config = yaml.safe_load(f) |
|
|
else: |
|
|
raise ValueError(f"Unsupported file type, must be one of {valid_suffixes}.") |
|
|
|
|
|
return load_agent_from_config(config, **kwargs) |
|
|
|
