"""使用嵌入向量比较两个模型输出的链条。"""
from enum import Enum
from typing import Any, Dict, List, Optional
import numpy as np
from langchain_core.callbacks.manager import (
AsyncCallbackManagerForChainRun,
CallbackManagerForChainRun,
Callbacks,
)
from langchain_core.embeddings import Embeddings
from langchain_core.pydantic_v1 import Field, root_validator
from langchain.chains.base import Chain
from langchain.evaluation.schema import PairwiseStringEvaluator, StringEvaluator
from langchain.schema import RUN_KEY
def _embedding_factory() -> Embeddings:
"""创建一个嵌入对象。
返回：
Embeddings：创建的嵌入对象。
"""
# Here for backwards compatibility.
# Generally, we do not want to be seeing imports from langchain community
# or partner packages in langchain.
try:
from langchain_openai import OpenAIEmbeddings
except ImportError:
try:
from langchain_community.embeddings.openai import OpenAIEmbeddings
except ImportError:
raise ImportError(
"Could not import OpenAIEmbeddings. Please install the "
"OpenAIEmbeddings package using `pip install langchain-openai`."
)
return OpenAIEmbeddings()
[docs]class EmbeddingDistance(str, Enum):
"""嵌入距离度量。
属性:
COSINE: 余弦距离度量。
EUCLIDEAN: 欧氏距离度量。
MANHATTAN: 曼哈顿距离度量。
CHEBYSHEV: 切比雪夫距离度量。
HAMMING: 汉明距离度量。"""
COSINE = "cosine"
EUCLIDEAN = "euclidean"
MANHATTAN = "manhattan"
CHEBYSHEV = "chebyshev"
HAMMING = "hamming"
class _EmbeddingDistanceChainMixin(Chain):
"""共享功能，用于嵌入距离评估器。
属性：
embeddings (Embeddings): 用于将输出向量化的嵌入对象。
distance_metric (EmbeddingDistance): 用于比较嵌入的距离度量标准。"""
embeddings: Embeddings = Field(default_factory=_embedding_factory)
distance_metric: EmbeddingDistance = Field(default=EmbeddingDistance.COSINE)
@root_validator(pre=False)
def _validate_tiktoken_installed(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""验证TikTok库是否已安装。
参数：
values（Dict[str, Any]）：要验证的值。
返回：
Dict[str, Any]：已验证的值。
"""
embeddings = values.get("embeddings")
types_ = []
try:
from langchain_openai import OpenAIEmbeddings
types_.append(OpenAIEmbeddings)
except ImportError:
pass
try:
from langchain_community.embeddings.openai import OpenAIEmbeddings
types_.append(OpenAIEmbeddings)
except ImportError:
pass
if not types_:
raise ImportError(
"Could not import OpenAIEmbeddings. Please install the "
"OpenAIEmbeddings package using `pip install langchain-openai`."
)
if isinstance(embeddings, tuple(types_)):
try:
import tiktoken # noqa: F401
except ImportError:
raise ImportError(
"The tiktoken library is required to use the default "
"OpenAI embeddings with embedding distance evaluators."
" Please either manually select a different Embeddings object"
" or install tiktoken using `pip install tiktoken`."
)
return values
class Config:
"""允许嵌入物不经验证。"""
arbitrary_types_allowed: bool = True
@property
def output_keys(self) -> List[str]:
"""返回链的输出键。
返回：
List[str]：输出键。
"""
return ["score"]
def _prepare_output(self, result: dict) -> dict:
parsed = {"score": result["score"]}
if RUN_KEY in result:
parsed[RUN_KEY] = result[RUN_KEY]
return parsed
def _get_metric(self, metric: EmbeddingDistance) -> Any:
"""获取给定度量名称的度量函数。
参数:
metric (EmbeddingDistance): 度量名称。
返回:
Any: 度量函数。
"""
metrics = {
EmbeddingDistance.COSINE: self._cosine_distance,
EmbeddingDistance.EUCLIDEAN: self._euclidean_distance,
EmbeddingDistance.MANHATTAN: self._manhattan_distance,
EmbeddingDistance.CHEBYSHEV: self._chebyshev_distance,
EmbeddingDistance.HAMMING: self._hamming_distance,
}
if metric in metrics:
return metrics[metric]
else:
raise ValueError(f"Invalid metric: {metric}")
@staticmethod
def _cosine_distance(a: np.ndarray, b: np.ndarray) -> np.ndarray:
"""计算两个向量之间的余弦距离。
参数:
a (np.ndarray): 第一个向量。
b (np.ndarray): 第二个向量。
返回:
np.ndarray: 余弦距离。
"""
try:
from langchain_community.utils.math import cosine_similarity
except ImportError:
raise ImportError(
"The cosine_similarity function is required to compute cosine distance."
" Please install the langchain-community package using"
" `pip install langchain-community`."
)
return 1.0 - cosine_similarity(a, b)
@staticmethod
def _euclidean_distance(a: np.ndarray, b: np.ndarray) -> np.floating:
"""计算两个向量之间的欧几里德距离。
参数:
a (np.ndarray): 第一个向量。
b (np.ndarray): 第二个向量。
返回:
np.floating: 欧几里德距离。
"""
return np.linalg.norm(a - b)
@staticmethod
def _manhattan_distance(a: np.ndarray, b: np.ndarray) -> np.floating:
"""计算两个向量之间的曼哈顿距离。
参数:
a (np.ndarray): 第一个向量。
b (np.ndarray): 第二个向量。
返回:
np.floating: 曼哈顿距离。
"""
return np.sum(np.abs(a - b))
@staticmethod
def _chebyshev_distance(a: np.ndarray, b: np.ndarray) -> np.floating:
"""计算两个向量之间的切比雪夫距离。
参数:
a (np.ndarray): 第一个向量。
b (np.ndarray): 第二个向量。
返回:
np.floating: 切比雪夫距离。
"""
return np.max(np.abs(a - b))
@staticmethod
def _hamming_distance(a: np.ndarray, b: np.ndarray) -> np.floating:
"""计算两个向量之间的汉明距离。
参数：
a（np.ndarray）：第一个向量。
b（np.ndarray）：第二个向量。
返回：
np.floating：汉明距离。
"""
return np.mean(a != b)
def _compute_score(self, vectors: np.ndarray) -> float:
"""根据距离度量计算得分。
参数:
vectors (np.ndarray): 输入向量。
返回:
float: 计算得分。
"""
metric = self._get_metric(self.distance_metric)
score = metric(vectors[0].reshape(1, -1), vectors[1].reshape(1, -1)).item()
return score
[docs]class EmbeddingDistanceEvalChain(_EmbeddingDistanceChainMixin, StringEvaluator):
"""使用嵌入距离来评分预测和参考之间的语义差异。
示例：
>>> chain = EmbeddingDistanceEvalChain()
>>> result = chain.evaluate_strings(prediction="Hello", reference="Hi")
>>> print(result)
{'score': 0.5}"""
@property
def requires_reference(self) -> bool:
"""返回链是否需要引用。
返回：
布尔值：如果需要引用，则为True，否则为False。
"""
return True
@property
def evaluation_name(self) -> str:
return f"embedding_{self.distance_metric.value}_distance"
@property
def input_keys(self) -> List[str]:
"""返回链的输入键。
返回：
List[str]：输入键。
"""
return ["prediction", "reference"]
def _call(
self,
inputs: Dict[str, Any],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
"""计算预测和参考之间的得分。
参数：
inputs（Dict[str, Any]）：输入数据。
run_manager（Optional[CallbackManagerForChainRun]，可选）：
回调管理器。
返回：
Dict[str, Any]：计算得分。
"""
vectors = np.array(
self.embeddings.embed_documents([inputs["prediction"], inputs["reference"]])
)
score = self._compute_score(vectors)
return {"score": score}
async def _acall(
self,
inputs: Dict[str, Any],
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
"""异步计算预测和参考值的得分。
参数：
inputs（Dict[str, Any]）：输入数据。
run_manager（AsyncCallbackManagerForChainRun，可选）：
回调管理器。
返回：
Dict[str, Any]：计算得分。
"""
embedded = await self.embeddings.aembed_documents(
[inputs["prediction"], inputs["reference"]]
)
vectors = np.array(embedded)
score = self._compute_score(vectors)
return {"score": score}
def _evaluate_strings(
self,
*,
prediction: str,
reference: Optional[str] = None,
callbacks: Callbacks = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
include_run_info: bool = False,
**kwargs: Any,
) -> dict:
"""评估预测和参考之间的嵌入距离。
参数:
prediction (str): 第一个模型的输出字符串。
reference (str): 参考字符串（必填）
callbacks (Callbacks, optional): 要使用的回调。
**kwargs (Any): 附加关键字参数。
返回:
dict: 包含以下内容的字典:
- score: 两个预测之间的嵌入距离。
"""
result = self(
inputs={"prediction": prediction, "reference": reference},
callbacks=callbacks,
tags=tags,
metadata=metadata,
include_run_info=include_run_info,
)
return self._prepare_output(result)
async def _aevaluate_strings(
self,
*,
prediction: str,
reference: Optional[str] = None,
callbacks: Callbacks = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
include_run_info: bool = False,
**kwargs: Any,
) -> dict:
"""异步评估预测和参考之间的嵌入距离。
参数：
prediction（str）：第一个模型的输出字符串。
reference（str）：第二个模型的输出字符串。
callbacks（Callbacks，可选）：要使用的回调。
**kwargs（Any）：额外的关键字参数。
返回：
dict：包含以下内容的字典：
- score：两个预测之间的嵌入距离。
"""
result = await self.acall(
inputs={"prediction": prediction, "reference": reference},
callbacks=callbacks,
tags=tags,
metadata=metadata,
include_run_info=include_run_info,
)
return self._prepare_output(result)
[docs]class PairwiseEmbeddingDistanceEvalChain(
_EmbeddingDistanceChainMixin, PairwiseStringEvaluator
):
"""使用嵌入距离来评估两个预测之间的语义差异。
示例：
>>> chain = PairwiseEmbeddingDistanceEvalChain()
>>> result = chain.evaluate_string_pairs(prediction="Hello", prediction_b="Hi")
>>> print(result)
{'score': 0.5}"""
@property
def input_keys(self) -> List[str]:
"""返回链的输入键。
返回：
List[str]：输入键。
"""
return ["prediction", "prediction_b"]
@property
def evaluation_name(self) -> str:
return f"pairwise_embedding_{self.distance_metric.value}_distance"
def _call(
self,
inputs: Dict[str, Any],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
"""计算两个预测的得分。
参数：
inputs（Dict[str, Any]）：输入数据。
run_manager（CallbackManagerForChainRun，可选）：
回调管理器。
返回：
Dict[str, Any]：计算得分。
"""
vectors = np.array(
self.embeddings.embed_documents(
[inputs["prediction"], inputs["prediction_b"]]
)
)
score = self._compute_score(vectors)
return {"score": score}
async def _acall(
self,
inputs: Dict[str, Any],
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
"""异步计算两个预测的得分。
参数：
inputs（Dict[str, Any]）：输入数据。
run_manager（AsyncCallbackManagerForChainRun，可选）：
回调管理器。
返回：
Dict[str, Any]：计算得分。
"""
embedded = await self.embeddings.aembed_documents(
[inputs["prediction"], inputs["prediction_b"]]
)
vectors = np.array(embedded)
score = self._compute_score(vectors)
return {"score": score}
def _evaluate_string_pairs(
self,
*,
prediction: str,
prediction_b: str,
callbacks: Callbacks = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
include_run_info: bool = False,
**kwargs: Any,
) -> dict:
"""评估两个预测之间的嵌入距离。
参数：
prediction (str): 第一个模型的输出字符串。
prediction_b (str): 第二个模型的输出字符串。
callbacks (Callbacks, optional): 要使用的回调。
tags (List[str], optional): 应用于跟踪的标签。
metadata (Dict[str, Any], optional): 应用于元数据的元数据。
**kwargs (Any): 其他关键字参数。
返回：
dict: 包含以下内容的字典：
- score: 两个预测之间的嵌入距离。
"""
result = self(
inputs={"prediction": prediction, "prediction_b": prediction_b},
callbacks=callbacks,
tags=tags,
metadata=metadata,
include_run_info=include_run_info,
)
return self._prepare_output(result)
async def _aevaluate_string_pairs(
self,
*,
prediction: str,
prediction_b: str,
callbacks: Callbacks = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
include_run_info: bool = False,
**kwargs: Any,
) -> dict:
"""异步评估两个预测之间的嵌入距离
参数:
prediction (str): 第一个模型的输出字符串。
prediction_b (str): 第二个模型的输出字符串。
callbacks (Callbacks, optional): 要使用的回调。
tags (List[str], optional): 应用于跟踪的标签
metadata (Dict[str, Any], optional): 应用于跟踪的元数据
**kwargs (Any): 附加关键字参数。
返回:
dict: 包含以下内容的字典:
- score: 两个预测之间的嵌入距离。
"""
result = await self.acall(
inputs={"prediction": prediction, "prediction_b": prediction_b},
callbacks=callbacks,
tags=tags,
metadata=metadata,
include_run_info=include_run_info,
)
return self._prepare_output(result)