Toggle Menu
Prev Up Next

# Source code for langchain_community.callbacks.utils

    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