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_textstat() -> Any:
"""导入textstat python包，并在未安装时引发错误。"""
return guard_import("textstat")
def _flatten_dict(
nested_dict: Dict[str, Any], parent_key: str = "", sep: str = "_"
) -> Iterable[Tuple[str, Any]]:
"""生成器，用于从嵌套字典中产生扁平化的项，以生成一个扁平的字典。
参数：
nested_dict（dict）：要扁平化的嵌套字典。
parent_key（str）：要添加到扁平字典键前面的前缀。
sep（str）：用于在父键和扁平字典键之间使用的分隔符。
产出：
（str，any）：扁平字典中的键值对。
"""
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]:
"""将嵌套字典展平为一个扁平的字典。
参数:
nested_dict (dict): 要展平的嵌套字典。
parent_key (str): 要添加到展平后字典键的前缀。
sep (str): 父键和展平后字典键之间要使用的分隔符。
返回:
(dict): 一个扁平的字典。
"""
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:
"""使用sha1对字符串进行哈希。
参数：
s（str）：要哈希的字符串。
返回：
（str）：哈希后的字符串。
"""
return hashlib.sha1(s.encode("utf-8")).hexdigest()
[docs]def load_json(json_path: Union[str, Path]) -> str:
"""将json文件加载到字符串中。
参数:
json_path (str): json文件的路径。
返回:
(str): json文件的字符串表示形式。
"""
with open(json_path, "r") as f:
data = f.read()
return data
[docs]class BaseMetadataCallbackHandler:
"""处理回调函数的元数据和相关函数状态。
属性:
step (int): 当前步骤。
starts (int): 调用start方法的次数。
ends (int): 调用end方法的次数。
errors (int): 调用error方法的次数。
text_ctr (int): 调用text方法的次数。
ignore_llm_ (bool): 是否忽略llm回调。
ignore_chain_ (bool): 是否忽略chain回调。
ignore_agent_ (bool): 是否忽略agent回调。
ignore_retriever_ (bool): 是否忽略retriever回调。
always_verbose_ (bool): 是否始终详细。
chain_starts (int): 调用chain start方法的次数。
chain_ends (int): 调用chain end方法的次数。
llm_starts (int): 调用llm start方法的次数。
llm_ends (int): 调用llm end方法的次数。
llm_streams (int): 调用text方法的次数。
tool_starts (int): 调用tool start方法的次数。
tool_ends (int): 调用tool end方法的次数。
agent_ends (int): 调用agent end方法的次数。
on_llm_start_records (list): on_llm_start方法的记录列表。
on_llm_token_records (list): on_llm_token方法的记录列表。
on_llm_end_records (list): on_llm_end方法的记录列表。
on_chain_start_records (list): on_chain_start方法的记录列表。
on_chain_end_records (list): on_chain_end方法的记录列表。
on_tool_start_records (list): on_tool_start方法的记录列表。
on_tool_end_records (list): on_tool_end方法的记录列表。
on_agent_finish_records (list): on_agent_end方法的记录列表。"""
[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:
"""即使 verbose 为 False，也要调用详细回调函数。"""
return self.always_verbose_
@property
def ignore_llm(self) -> bool:
"""是否忽略LLM回调。"""
return self.ignore_llm_
@property
def ignore_chain(self) -> bool:
"""是否忽略链式回调。"""
return self.ignore_chain_
@property
def ignore_agent(self) -> bool:
"""是否忽略代理回调。"""
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:
"""重置回调元数据。"""
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