# flake8: noqa
import os
import warnings
from typing import Any, Dict, List, Optional, Union
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.outputs import LLMResult
[docs]class DeepEvalCallbackHandler(BaseCallbackHandler):
"""回调处理程序，用于登录到deepeval。
参数:
implementation_name: deepeval中`implementation`的名称
metrics: 一个指标列表
异常:
ImportError: 如果未安装`deepeval`包。
示例:
>>> from langchain_community.llms import OpenAI
>>> from langchain_community.callbacks import DeepEvalCallbackHandler
>>> from deepeval.metrics import AnswerRelevancy
>>> metric = AnswerRelevancy(minimum_score=0.3)
>>> deepeval_callback = DeepEvalCallbackHandler(
... implementation_name="exampleImplementation",
... metrics=[metric],
... )
>>> llm = OpenAI(
... temperature=0,
... callbacks=[deepeval_callback],
... verbose=True,
... openai_api_key="API_KEY_HERE",
... )
>>> llm.generate([
... "What is the best evaluation tool out there? (no bias at all)",
... ])
"Deepeval，毫无疑问。" """
REPO_URL: str = "https://github.com/confident-ai/deepeval"
ISSUES_URL: str = f"{REPO_URL}/issues"
BLOG_URL: str = "https://docs.confident-ai.com" # noqa: E501
[docs] def __init__(
self,
metrics: List[Any],
implementation_name: Optional[str] = None,
) -> None:
"""初始化`deepevalCallbackHandler`。
参数：
implementation_name：您想要的实现名称。
metrics：您想要跟踪的指标是什么？
引发：
ImportError：如果未安装`deepeval`包。
ConnectionError：如果连接到deepeval失败。
"""
super().__init__()
# Import deepeval (not via `import_deepeval` to keep hints in IDEs)
try:
import deepeval # ignore: F401,I001
except ImportError:
raise ImportError(
"""To use the deepeval callback manager you need to have the
`deepeval` Python package installed. Please install it with
`pip install deepeval`"""
)
if os.path.exists(".deepeval"):
warnings.warn(
"""You are currently not logging anything to the dashboard, we
recommend using `deepeval login`."""
)
# Set the deepeval variables
self.implementation_name = implementation_name
self.metrics = metrics
warnings.warn(
(
"The `DeepEvalCallbackHandler` is currently in beta and is subject to"
" change based on updates to `langchain`. Please report any issues to"
f" {self.ISSUES_URL} as an `integration` issue."
),
)
[docs] def on_llm_start(
self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
) -> None:
"""存储提示"""
self.prompts = prompts
[docs] def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
"""当生成一个新的令牌时不执行任何操作。"""
pass
[docs] def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
"""当LLM结束时，将日志记录到deepeval。"""
from deepeval.metrics.answer_relevancy import AnswerRelevancy
from deepeval.metrics.bias_classifier import UnBiasedMetric
from deepeval.metrics.metric import Metric
from deepeval.metrics.toxic_classifier import NonToxicMetric
for metric in self.metrics:
for i, generation in enumerate(response.generations):
# Here, we only measure the first generation's output
output = generation[0].text
query = self.prompts[i]
if isinstance(metric, AnswerRelevancy):
result = metric.measure(
output=output,
query=query,
)
print(f"Answer Relevancy: {result}") # noqa: T201
elif isinstance(metric, UnBiasedMetric):
score = metric.measure(output)
print(f"Bias Score: {score}") # noqa: T201
elif isinstance(metric, NonToxicMetric):
score = metric.measure(output)
print(f"Toxic Score: {score}") # noqa: T201
else:
raise ValueError(
f"""Metric {metric.__name__} is not supported by deepeval
callbacks."""
)
[docs] def on_llm_error(self, error: BaseException, **kwargs: Any) -> None:
"""LLM 输出错误时不执行任何操作。"""
pass
[docs] def on_chain_start(
self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
) -> None:
"""当链条开始时不执行任何操作"""
pass
[docs] def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
"""当链结束时不执行任何操作。"""
pass
[docs] def on_chain_error(self, error: BaseException, **kwargs: Any) -> None:
"""当LLM链输出错误时不执行任何操作。"""
pass
[docs] def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
"""当代理执行特定动作时不执行任何操作。"""
pass
[docs] def on_text(self, text: str, **kwargs: Any) -> None:
"""什么都不做"""
pass
[docs] def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> None:
"""什么都不做"""
pass