Source code for langchain_community.callbacks.whylabs_callback

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, Optional

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.utils import get_from_env, guard_import

if TYPE_CHECKING:
    from whylogs.api.logger.logger import Logger

diagnostic_logger = logging.getLogger(__name__)


def import_langkit(
    sentiment: bool = False,
    toxicity: bool = False,
    themes: bool = False,
) -> Any:
    """导入langkit python包，如果未安装则引发错误。

参数：
    sentiment: 是否导入langkit.sentiment模块。默认为False。
    toxicity: 是否导入langkit.toxicity模块。默认为False。
    themes: 是否导入langkit.themes模块。默认为False。

返回：
    导入的langkit模块。
"""
    langkit = guard_import("langkit")
    guard_import("langkit.regexes")
    guard_import("langkit.textstat")
    if sentiment:
        guard_import("langkit.sentiment")
    if toxicity:
        guard_import("langkit.toxicity")
    if themes:
        guard_import("langkit.themes")
    return langkit


class WhyLabsCallbackHandler(BaseCallbackHandler):
    """用于记录到WhyLabs的回调处理程序。此回调处理程序利用`langkit`从与LLM交互时的提示和响应中提取特征。这些特征可用于隔离、评估和观察随时间发生的交互，以检测与幻觉、提示工程或输出验证相关的问题。LangKit是由WhyLabs开发的LLM监控工具包。

以下是使用LangKit可以监控的一些示例：
* 文本质量
  - 可读性分数
  - 复杂度和年级分数
* 文本相关性
  - 提示/响应之间的相似性分数
  - 与用户定义主题的相似性分数
  - 主题分类
* 安全性和隐私性
  - 模式 - 与用户定义的正则表达式模式组匹配的字符串计数
  - 越狱 - 与已知越狱尝试相关的相似性分数
  - 提示注入 - 与已知提示攻击相关的相似性分数
  - 拒绝 - 与已知LLM拒绝响应相关的相似性分数
* 情感和毒性
  - 情感分析
  - 毒性分析

有关更多信息，请参阅https://docs.whylabs.ai/docs/language-model-monitoring
或查看LangKit存储库：https://github.com/whylabs/langkit

---
参数：
    api_key（可选[str]）：WhyLabs API密钥。可选，因为指定API密钥的首选方式是使用环境变量WHYLABS_API_KEY。
    org_id（可选[str]）：要将配置文件写入的WhyLabs组织ID。可选，因为指定组织ID的首选方式是使用环境变量WHYLABS_DEFAULT_ORG_ID。
    dataset_id（可选[str]）：要将配置文件写入的WhyLabs数据集ID。可选，因为指定数据集ID的首选方式是使用环境变量WHYLABS_DEFAULT_DATASET_ID。
    sentiment（bool）：是否启用情感分析。默认为False。
    toxicity（bool）：是否启用毒性分析。默认为False。
    themes（bool）：是否启用主题分析。默认为False。"""

    def __init__(self, logger: Logger, handler: Any):
        """初始化滚动记录器。"""
        super().__init__()
        if hasattr(handler, "init"):
            handler.init(self)
        if hasattr(handler, "_get_callbacks"):
            self._callbacks = handler._get_callbacks()
        else:
            self._callbacks = dict()
            diagnostic_logger.warning("initialized handler without callbacks.")
        self._logger = logger

    def flush(self) -> None:
        """如果使用滚动记录器，请明确地写入当前配置文件。"""
        if self._logger and hasattr(self._logger, "_do_rollover"):
            self._logger._do_rollover()
            diagnostic_logger.info("Flushing WhyLabs logger, writing profile...")

    def close(self) -> None:
        """关闭任何记录器，以便在退出之前写出任何配置文件。"""
        if self._logger and hasattr(self._logger, "close"):
            self._logger.close()
            diagnostic_logger.info("Closing WhyLabs logger, see you next time!")

    def __enter__(self) -> WhyLabsCallbackHandler:
        return self

    def __exit__(
        self, exception_type: Any, exception_value: Any, traceback: Any
    ) -> None:
        self.close()

    @classmethod
    def from_params(
        cls,
        *,
        api_key: Optional[str] = None,
        org_id: Optional[str] = None,
        dataset_id: Optional[str] = None,
        sentiment: bool = False,
        toxicity: bool = False,
        themes: bool = False,
        logger: Optional[Logger] = None,
    ) -> WhyLabsCallbackHandler:
        """从参数中实例化whylogs Logger。

参数：
    api_key（可选[str]）：WhyLabs API密钥。可选，因为指定API密钥的首选方式是使用环境变量WHYLABS_API_KEY。
    org_id（可选[str]）：要将配置文件写入的WhyLabs组织ID。如果未设置，则必须在环境变量WHYLABS_DEFAULT_ORG_ID中指定。
    dataset_id（可选[str]）：此回调正在收集遥测数据的模型或数据集。如果未设置，则必须在环境变量WHYLABS_DEFAULT_DATASET_ID中指定。
    sentiment（bool）：如果为True，将初始化一个模型以执行情感分析复合分数。默认为False，不会收集此指标。
    toxicity（bool）：如果为True，将初始化一个模型以评分毒性。默认为False，不会收集此指标。
    themes（bool）：如果为True，将初始化一个模型以计算到配置主题的距离。默认为None，不会收集此指标。
    logger（可选[Logger]）：如果指定，将绑定配置的记录器作为遥测收集代理。默认为LangKit模式，带有定期的WhyLabs写入器。
"""
        # langkit library will import necessary whylogs libraries
        import_langkit(sentiment=sentiment, toxicity=toxicity, themes=themes)

        why = guard_import("whylogs")
        get_callback_instance = guard_import(
            "langkit.callback_handler"
        ).get_callback_instance
        WhyLabsWriter = guard_import("whylogs.api.writer.whylabs").WhyLabsWriter
        udf_schema = guard_import("whylogs.experimental.core.udf_schema").udf_schema

        if logger is None:
            api_key = api_key or get_from_env("api_key", "WHYLABS_API_KEY")
            org_id = org_id or get_from_env("org_id", "WHYLABS_DEFAULT_ORG_ID")
            dataset_id = dataset_id or get_from_env(
                "dataset_id", "WHYLABS_DEFAULT_DATASET_ID"
            )
            whylabs_writer = WhyLabsWriter(
                api_key=api_key, org_id=org_id, dataset_id=dataset_id
            )

            whylabs_logger = why.logger(
                mode="rolling", interval=5, when="M", schema=udf_schema()
            )

            whylabs_logger.append_writer(writer=whylabs_writer)
        else:
            diagnostic_logger.info("Using passed in whylogs logger {logger}")
            whylabs_logger = logger

        callback_handler_cls = get_callback_instance(logger=whylabs_logger, impl=cls)
        diagnostic_logger.info(
            "Started whylogs Logger with WhyLabsWriter and initialized LangKit. 📝"
        )
        return callback_handler_cls