Source code for langchain_community.callbacks.utils

import hashlib
from pathlib import Path
from typing import Any, Dict, Iterable, Tuple, Union

from langchain_core.utils import guard_import


[docs] def import_spacy() -> Any: """Import the spacy python package and raise an error if it is not installed.""" return guard_import("spacy")
[docs] def import_pandas() -> Any: """Import the pandas python package and raise an error if it is not installed.""" return guard_import("pandas")
[docs] 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
[docs] 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
[docs] 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()
[docs] 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
[docs] 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. """
[docs] 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_
[docs] 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, }
[docs] 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