Spaces:
Runtime error
Runtime error
| import hashlib | |
| from pathlib import Path | |
| from typing import Any, Dict, Iterable, Tuple, Union | |
| from langchain_core.utils import guard_import | |
| def import_spacy() -> Any: | |
| """Import the spacy python package and raise an error if it is not installed.""" | |
| return guard_import("spacy") | |
| def import_pandas() -> Any: | |
| """Import the pandas python package and raise an error if it is not installed.""" | |
| return guard_import("pandas") | |
| def import_textstat() -> Any: | |
| """Import the textstat python package and raise an error if it is not installed.""" | |
| return guard_import("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 = [] | |
| def always_verbose(self) -> bool: | |
| """Whether to call verbose callbacks even if verbose is False.""" | |
| return self.always_verbose_ | |
| def ignore_llm(self) -> bool: | |
| """Whether to ignore LLM callbacks.""" | |
| return self.ignore_llm_ | |
| def ignore_chain(self) -> bool: | |
| """Whether to ignore chain callbacks.""" | |
| return self.ignore_chain_ | |
| 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 | |