mirror of
https://github.com/hwchase17/langchain
synced 2024-11-10 01:10:59 +00:00
259 lines
8.3 KiB
Python
259 lines
8.3 KiB
Python
import hashlib
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Iterable, Tuple, Union
|
|
|
|
|
|
def import_spacy() -> Any:
|
|
"""Import the spacy python package and raise an error if it is not installed."""
|
|
try:
|
|
import spacy
|
|
except ImportError:
|
|
raise ImportError(
|
|
"This callback manager requires the `spacy` python "
|
|
"package installed. Please install it with `pip install spacy`"
|
|
)
|
|
return spacy
|
|
|
|
|
|
def import_pandas() -> Any:
|
|
"""Import the pandas python package and raise an error if it is not installed."""
|
|
try:
|
|
import pandas
|
|
except ImportError:
|
|
raise ImportError(
|
|
"This callback manager requires the `pandas` python "
|
|
"package installed. Please install it with `pip install pandas`"
|
|
)
|
|
return pandas
|
|
|
|
|
|
def import_textstat() -> Any:
|
|
"""Import the textstat python package and raise an error if it is not installed."""
|
|
try:
|
|
import textstat
|
|
except ImportError:
|
|
raise ImportError(
|
|
"This callback manager requires the `textstat` python "
|
|
"package installed. Please install it with `pip install textstat`"
|
|
)
|
|
return textstat
|
|
|
|
|
|
def _flatten_dict(
|
|
nested_dict: Dict[str, Any], parent_key: str = "", sep: str = "_"
|
|
) -> Iterable[Tuple[str, Any]]:
|
|
"""
|
|
Generator that yields flattened items from a nested dictionary for a flat dict.
|
|
|
|
Parameters:
|
|
nested_dict (dict): The nested dictionary to flatten.
|
|
parent_key (str): The prefix to prepend to the keys of the flattened dict.
|
|
sep (str): The separator to use between the parent key and the key of the
|
|
flattened dictionary.
|
|
|
|
Yields:
|
|
(str, any): A key-value pair from the flattened dictionary.
|
|
"""
|
|
for key, value in nested_dict.items():
|
|
new_key = parent_key + sep + key if parent_key else key
|
|
if isinstance(value, dict):
|
|
yield from _flatten_dict(value, new_key, sep)
|
|
else:
|
|
yield new_key, value
|
|
|
|
|
|
def flatten_dict(
|
|
nested_dict: Dict[str, Any], parent_key: str = "", sep: str = "_"
|
|
) -> Dict[str, Any]:
|
|
"""Flatten a nested dictionary into a flat dictionary.
|
|
|
|
Parameters:
|
|
nested_dict (dict): The nested dictionary to flatten.
|
|
parent_key (str): The prefix to prepend to the keys of the flattened dict.
|
|
sep (str): The separator to use between the parent key and the key of the
|
|
flattened dictionary.
|
|
|
|
Returns:
|
|
(dict): A flat dictionary.
|
|
|
|
"""
|
|
flat_dict = {k: v for k, v in _flatten_dict(nested_dict, parent_key, sep)}
|
|
return flat_dict
|
|
|
|
|
|
def hash_string(s: str) -> str:
|
|
"""Hash a string using sha1.
|
|
|
|
Parameters:
|
|
s (str): The string to hash.
|
|
|
|
Returns:
|
|
(str): The hashed string.
|
|
"""
|
|
return hashlib.sha1(s.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def load_json(json_path: Union[str, Path]) -> str:
|
|
"""Load json file to a string.
|
|
|
|
Parameters:
|
|
json_path (str): The path to the json file.
|
|
|
|
Returns:
|
|
(str): The string representation of the json file.
|
|
"""
|
|
with open(json_path, "r") as f:
|
|
data = f.read()
|
|
return data
|
|
|
|
|
|
class BaseMetadataCallbackHandler:
|
|
"""Handle the metadata and associated function states for callbacks.
|
|
|
|
Attributes:
|
|
step (int): The current step.
|
|
starts (int): The number of times the start method has been called.
|
|
ends (int): The number of times the end method has been called.
|
|
errors (int): The number of times the error method has been called.
|
|
text_ctr (int): The number of times the text method has been called.
|
|
ignore_llm_ (bool): Whether to ignore llm callbacks.
|
|
ignore_chain_ (bool): Whether to ignore chain callbacks.
|
|
ignore_agent_ (bool): Whether to ignore agent callbacks.
|
|
ignore_retriever_ (bool): Whether to ignore retriever callbacks.
|
|
always_verbose_ (bool): Whether to always be verbose.
|
|
chain_starts (int): The number of times the chain start method has been called.
|
|
chain_ends (int): The number of times the chain end method has been called.
|
|
llm_starts (int): The number of times the llm start method has been called.
|
|
llm_ends (int): The number of times the llm end method has been called.
|
|
llm_streams (int): The number of times the text method has been called.
|
|
tool_starts (int): The number of times the tool start method has been called.
|
|
tool_ends (int): The number of times the tool end method has been called.
|
|
agent_ends (int): The number of times the agent end method has been called.
|
|
on_llm_start_records (list): A list of records of the on_llm_start method.
|
|
on_llm_token_records (list): A list of records of the on_llm_token method.
|
|
on_llm_end_records (list): A list of records of the on_llm_end method.
|
|
on_chain_start_records (list): A list of records of the on_chain_start method.
|
|
on_chain_end_records (list): A list of records of the on_chain_end method.
|
|
on_tool_start_records (list): A list of records of the on_tool_start method.
|
|
on_tool_end_records (list): A list of records of the on_tool_end method.
|
|
on_agent_finish_records (list): A list of records of the on_agent_end method.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.step = 0
|
|
|
|
self.starts = 0
|
|
self.ends = 0
|
|
self.errors = 0
|
|
self.text_ctr = 0
|
|
|
|
self.ignore_llm_ = False
|
|
self.ignore_chain_ = False
|
|
self.ignore_agent_ = False
|
|
self.ignore_retriever_ = False
|
|
self.always_verbose_ = False
|
|
|
|
self.chain_starts = 0
|
|
self.chain_ends = 0
|
|
|
|
self.llm_starts = 0
|
|
self.llm_ends = 0
|
|
self.llm_streams = 0
|
|
|
|
self.tool_starts = 0
|
|
self.tool_ends = 0
|
|
|
|
self.agent_ends = 0
|
|
|
|
self.on_llm_start_records: list = []
|
|
self.on_llm_token_records: list = []
|
|
self.on_llm_end_records: list = []
|
|
|
|
self.on_chain_start_records: list = []
|
|
self.on_chain_end_records: list = []
|
|
|
|
self.on_tool_start_records: list = []
|
|
self.on_tool_end_records: list = []
|
|
|
|
self.on_text_records: list = []
|
|
self.on_agent_finish_records: list = []
|
|
self.on_agent_action_records: list = []
|
|
|
|
@property
|
|
def always_verbose(self) -> bool:
|
|
"""Whether to call verbose callbacks even if verbose is False."""
|
|
return self.always_verbose_
|
|
|
|
@property
|
|
def ignore_llm(self) -> bool:
|
|
"""Whether to ignore LLM callbacks."""
|
|
return self.ignore_llm_
|
|
|
|
@property
|
|
def ignore_chain(self) -> bool:
|
|
"""Whether to ignore chain callbacks."""
|
|
return self.ignore_chain_
|
|
|
|
@property
|
|
def ignore_agent(self) -> bool:
|
|
"""Whether to ignore agent callbacks."""
|
|
return self.ignore_agent_
|
|
|
|
def get_custom_callback_meta(self) -> Dict[str, Any]:
|
|
return {
|
|
"step": self.step,
|
|
"starts": self.starts,
|
|
"ends": self.ends,
|
|
"errors": self.errors,
|
|
"text_ctr": self.text_ctr,
|
|
"chain_starts": self.chain_starts,
|
|
"chain_ends": self.chain_ends,
|
|
"llm_starts": self.llm_starts,
|
|
"llm_ends": self.llm_ends,
|
|
"llm_streams": self.llm_streams,
|
|
"tool_starts": self.tool_starts,
|
|
"tool_ends": self.tool_ends,
|
|
"agent_ends": self.agent_ends,
|
|
}
|
|
|
|
def reset_callback_meta(self) -> None:
|
|
"""Reset the callback metadata."""
|
|
self.step = 0
|
|
|
|
self.starts = 0
|
|
self.ends = 0
|
|
self.errors = 0
|
|
self.text_ctr = 0
|
|
|
|
self.ignore_llm_ = False
|
|
self.ignore_chain_ = False
|
|
self.ignore_agent_ = False
|
|
self.always_verbose_ = False
|
|
|
|
self.chain_starts = 0
|
|
self.chain_ends = 0
|
|
|
|
self.llm_starts = 0
|
|
self.llm_ends = 0
|
|
self.llm_streams = 0
|
|
|
|
self.tool_starts = 0
|
|
self.tool_ends = 0
|
|
|
|
self.agent_ends = 0
|
|
|
|
self.on_llm_start_records = []
|
|
self.on_llm_token_records = []
|
|
self.on_llm_end_records = []
|
|
|
|
self.on_chain_start_records = []
|
|
self.on_chain_end_records = []
|
|
|
|
self.on_tool_start_records = []
|
|
self.on_tool_end_records = []
|
|
|
|
self.on_text_records = []
|
|
self.on_agent_finish_records = []
|
|
self.on_agent_action_records = []
|
|
return None
|