BaseSample
基类: BaseModel
评测样本的基类。
to_dict
to_dict() -> Dict
获取样本的字典表示,不含值为 None 的属性。
源代码位于 src/ragas/dataset_schema.py
def to_dict(self) -> t.Dict:
"""
Get the dictionary representation of the sample without attributes that are None.
"""
return self.model_dump(exclude_none=True)
get_features
get_features() -> List[str]
获取样本中不为 None 的特征。
源代码位于 src/ragas/dataset_schema.py
def get_features(self) -> t.List[str]:
"""
Get the features of the sample that are not None.
"""
return list(self.to_dict().keys())
to_string
to_string() -> str
获取样本的字符串表示。
源代码位于 src/ragas/dataset_schema.py
def to_string(self) -> str:
"""
Get the string representation of the sample.
"""
sample_dict = self.to_dict()
return "".join(f"\n{key}:\n\t{val}\n" for key, val in sample_dict.items())
SingleTurnSample
基类: BaseSample
表示单轮交互的评测样本。
属性:
| 名称 | 类型 | 说明 |
|---|---|---|
user_input |
Optional[str] |
用户的输入查询。 |
retrieved_contexts |
Optional[List[str]] |
为该查询检索到的上下文列表。 |
reference_contexts |
Optional[List[str]] |
该查询的参考上下文列表。 |
retrieved_context_ids |
Optional[List[Union[str, int]]] |
检索上下文的 ID 列表。 |
reference_context_ids |
Optional[List[Union[str, int]]] |
参考上下文的 ID 列表。 |
response |
Optional[str] |
该查询的生成响应。 |
multi_responses |
Optional[List[str]] |
为该查询生成的多个响应列表。 |
reference |
Optional[str] |
该查询的参考答案。 |
rubric |
Optional[Dict[str, str]] |
该样本的评测量表。 |
persona_name |
Optional[str] |
查询生成中使用的 persona 名称。 |
query_style |
Optional[str] |
生成查询的风格(例如 formal、casual)。 |
query_length |
Optional[str] |
查询的长度类别(例如 short、medium、long)。 |
MultiTurnSample
基类: BaseSample
表示多轮交互的评测样本。
属性:
| 名称 | 类型 | 说明 |
|---|---|---|
user_input |
List[Union[HumanMessage, AIMessage, ToolMessage]] |
表示对话轮次的 messages 列表。 |
reference |
(Optional[str], optional) |
对话的参考答案或期望结果。 |
reference_tool_calls |
(Optional[List[ToolCall]], optional) |
对话期望的工具调用列表。 |
rubrics |
(Optional[Dict[str, str]], optional) |
对话的评测量表。 |
reference_topics |
(Optional[List[str]], optional) |
对话的参考主题列表。 |
validate_user_input
validate_user_input(messages: List[Union[HumanMessage, AIMessage, ToolMessage]]) -> List[Union[HumanMessage, AIMessage, ToolMessage]]
校验用户输入 messages。
源代码位于 src/ragas/dataset_schema.py
@field_validator("user_input")
@classmethod
def validate_user_input(
cls,
messages: t.List[t.Union[HumanMessage, AIMessage, ToolMessage]],
) -> t.List[t.Union[HumanMessage, AIMessage, ToolMessage]]:
"""Validates the user input messages."""
if not all(
isinstance(m, (HumanMessage, AIMessage, ToolMessage)) for m in messages
):
raise ValueError(
"All inputs must be instances of HumanMessage, AIMessage, or ToolMessage."
)
has_seen_ai_message = False
for i, m in enumerate(messages):
if isinstance(m, AIMessage):
has_seen_ai_message = True
elif isinstance(m, ToolMessage):
# Rule 1: ToolMessage must be preceded by an AIMessage somewhere in the conversation
if not has_seen_ai_message:
raise ValueError(
"ToolMessage must be preceded by an AIMessage somewhere in the conversation."
)
# Rule 2: ToolMessage must follow an AIMessage or another ToolMessage
if i > 0:
prev_message = messages[i - 1]
if isinstance(prev_message, AIMessage):
# Rule 3: If following AIMessage, that message must have tool_calls
if not prev_message.tool_calls:
raise ValueError(
"ToolMessage must follow an AIMessage where tools were called."
)
elif not isinstance(prev_message, ToolMessage):
# Not following AIMessage or ToolMessage
raise ValueError(
"ToolMessage must follow an AIMessage or another ToolMessage."
)
return messages
to_messages
to_messages()
将用户输入 messages 转换为字典列表。
源代码位于 src/ragas/dataset_schema.py
def to_messages(self):
"""Converts the user input messages to a list of dictionaries."""
return [m.model_dump() for m in self.user_input]
pretty_repr
pretty_repr()
返回对话的美观字符串表示。
源代码位于 src/ragas/dataset_schema.py
def pretty_repr(self):
"""Returns a pretty string representation of the conversation."""
lines = []
for m in self.user_input:
lines.append(m.pretty_repr())
return "\n".join(lines)
RagasDataset
RagasDataset(samples: List[Sample])
基类: ABC, Generic[Sample]
to_list
to_list() -> List[Dict]
将样本转换为字典列表。
源代码位于 src/ragas/dataset_schema.py
@abstractmethod
def to_list(self) -> t.List[t.Dict]:
"""Converts the samples to a list of dictionaries."""
pass
from_list
from_list(data: List[Dict]) -> T
从字典列表创建 RagasDataset。
源代码位于 src/ragas/dataset_schema.py
@classmethod
@abstractmethod
def from_list(cls: t.Type[T], data: t.List[t.Dict]) -> T:
"""Creates an RagasDataset from a list of dictionaries."""
pass
validate_samples
validate_samples(samples: List[Sample]) -> List[Sample]
校验所有样本类型相同。
源代码位于 src/ragas/dataset_schema.py
def validate_samples(self, samples: t.List[Sample]) -> t.List[Sample]:
"""Validates that all samples are of the same type."""
if len(samples) == 0:
return samples
first_sample_type = type(samples[0])
for i, sample in enumerate(samples):
if not isinstance(sample, first_sample_type):
raise ValueError(
f"Sample at index {i} is of type {type(sample)}, expected {first_sample_type}"
)
return samples
get_sample_type
get_sample_type() -> Type[Sample]
返回数据集中样本的类型。
源代码位于 src/ragas/dataset_schema.py
def get_sample_type(self) -> t.Type[Sample]:
"""Returns the type of the samples in the dataset."""
return type(self.samples[0])
to_hf_dataset
to_hf_dataset() -> Dataset
将数据集转换为 Hugging Face Dataset。
源代码位于 src/ragas/dataset_schema.py
def to_hf_dataset(self) -> HFDataset:
"""Converts the dataset to a Hugging Face Dataset."""
try:
from datasets import Dataset as HFDataset
except ImportError:
raise ImportError(
"datasets is not installed. Please install it to use this function."
)
return HFDataset.from_list(self.to_list())
from_hf_dataset
from_hf_dataset(dataset: Dataset) -> T
从 Hugging Face Dataset 创建 EvaluationDataset。
源代码位于 src/ragas/dataset_schema.py
@classmethod
def from_hf_dataset(cls: t.Type[T], dataset: HFDataset) -> T:
"""Creates an EvaluationDataset from a Hugging Face Dataset."""
return cls.from_list(dataset.to_list())
to_pandas
to_pandas() -> DataFrame
将数据集转换为 pandas DataFrame。
源代码位于 src/ragas/dataset_schema.py
def to_pandas(self) -> PandasDataframe:
"""Converts the dataset to a pandas DataFrame."""
try:
import pandas as pd
except ImportError:
raise ImportError(
"pandas is not installed. Please install it to use this function."
)
data = self.to_list()
return pd.DataFrame(data)
from_pandas
from_pandas(dataframe: DataFrame)
从 pandas DataFrame 创建 EvaluationDataset。
源代码位于 src/ragas/dataset_schema.py
@classmethod
def from_pandas(cls, dataframe: PandasDataframe):
"""Creates an EvaluationDataset from a pandas DataFrame."""
return cls.from_list(dataframe.to_dict(orient="records"))
features
features()
返回样本的特征。
源代码位于 src/ragas/dataset_schema.py
def features(self):
"""Returns the features of the samples."""
return self.samples[0].get_features()
from_dict
from_dict(mapping: Dict) -> T
从字典创建 EvaluationDataset。
源代码位于 src/ragas/dataset_schema.py
@classmethod
def from_dict(cls: t.Type[T], mapping: t.Dict) -> T:
"""Creates an EvaluationDataset from a dictionary."""
samples = []
if all(
"user_input" in item and isinstance(mapping[0]["user_input"], list)
for item in mapping
):
samples.extend(MultiTurnSample(**sample) for sample in mapping)
else:
samples.extend(SingleTurnSample(**sample) for sample in mapping)
return cls(samples=samples)
to_csv
to_csv(path: Union[str, Path])
将数据集转换为 CSV 文件。
源代码位于 src/ragas/dataset_schema.py
def to_csv(self, path: t.Union[str, Path]):
"""Converts the dataset to a CSV file."""
import csv
data = self.to_list()
if not data:
return
fieldnames = data[0].keys()
with open(path, "w", newline="") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for row in data:
writer.writerow(row)
to_jsonl
to_jsonl(path: Union[str, Path])
将数据集转换为 JSONL 文件。
源代码位于 src/ragas/dataset_schema.py
def to_jsonl(self, path: t.Union[str, Path]):
"""Converts the dataset to a JSONL file."""
with open(path, "w") as jsonlfile:
for sample in self.to_list():
jsonlfile.write(json.dumps(sample, ensure_ascii=False) + "\n")
from_jsonl
from_jsonl(path: Union[str, Path]) -> T
从 JSONL 文件创建 EvaluationDataset。
源代码位于 src/ragas/dataset_schema.py
@classmethod
def from_jsonl(cls: t.Type[T], path: t.Union[str, Path]) -> T:
"""Creates an EvaluationDataset from a JSONL file."""
with open(path, "r") as jsonlfile:
data = [json.loads(line) for line in jsonlfile]
return cls.from_list(data)
EvaluationDataset
EvaluationDataset(samples: List[Sample], backend: Optional[str] = None, name: Optional[str] = None)
基类: RagasDataset[SingleTurnSampleOrMultiTurnSample]
表示评测样本数据集。
属性:
| 名称 | 类型 | 说明 |
|---|---|---|
samples |
List[BaseSample] |
评测样本列表。 |
backend |
Optional[str] |
用于存储数据集的后端(例如 "local/csv")。默认为 None。 |
name |
Optional[str] |
数据集名称。默认为 None。 |
方法:
| Name | Description |
|---|---|
validate_samples |
校验所有样本类型相同。 |
get_sample_type |
返回数据集中样本的类型。 |
to_hf_dataset |
将数据集转换为 Hugging Face Dataset。 |
to_pandas |
将数据集转换为 pandas DataFrame。 |
features |
返回样本的特征。 |
from_list |
从字典列表创建 EvaluationDataset。 |
from_dict |
从字典创建 EvaluationDataset。 |
to_csv |
将数据集转换为 CSV 文件。 |
to_jsonl |
将数据集转换为 JSONL 文件。 |
from_jsonl |
从 JSONL 文件创建 EvaluationDataset。 |
EvaluationResult
EvaluationResult(scores: List[Dict[str, Any]], dataset: EvaluationDataset, binary_columns: List[str] = list(), cost_cb: Optional[CostCallbackHandler] = None, traces: List[Dict[str, Any]] = list(), ragas_traces: Dict[str, ChainRun] = dict(), run_id: Optional[UUID] = None)
用于存储和处理评测结果的类。
属性:
| 名称 | 类型 | 说明 |
|---|---|---|
scores |
Dataset |
包含评测分数的数据集。 |
dataset |
(Dataset, optional) |
用于评测的原始数据集。默认为 None。 |
binary_columns |
list of str, optional |
作为二元指标的列列表。默认为空列表。 |
cost_cb |
(CostCallbackHandler, optional) |
用于成本计算的 callback handler。默认为 None。 |
to_pandas
to_pandas(batch_size: int | None = None, batched: bool = False)
将结果转换为 pandas DataFrame。
参数:
| 名称 | 类型 | 说明 | 默认值 |
|---|---|---|---|
batch_size |
int |
转换的批次大小。默认为 None。 | None |
batched |
bool |
是否分批转换。默认为 False。 | False |
返回:
| 类型 | 说明 |
|---|---|
DataFrame |
作为 pandas DataFrame 的结果。 |
抛出:
| 类型 | 说明 |
|---|---|
ValueError |
若未提供数据集。 |
源代码位于 src/ragas/dataset_schema.py
def to_pandas(self, batch_size: int | None = None, batched: bool = False):
"""
Convert the result to a pandas DataFrame.
Parameters
----------
batch_size : int, optional
The batch size for conversion. Default is None.
batched : bool, optional
Whether to convert in batches. Default is False.
Returns
-------
pandas.DataFrame
The result as a pandas DataFrame.
Raises
------
ValueError
If the dataset is not provided.
"""
try:
import pandas as pd
except ImportError:
raise ImportError(
"pandas is not installed. Please install it to use this function."
)
if self.dataset is None:
raise ValueError("dataset is not provided for the results class")
assert len(self.scores) == len(self.dataset)
# convert both to pandas dataframes and concatenate
scores_df = pd.DataFrame(self.scores)
dataset_df = self.dataset.to_pandas()
return pd.concat([dataset_df, scores_df], axis=1)
total_tokens
total_tokens() -> Union[List[TokenUsage], TokenUsage]
计算评测中使用的总 token 数。
返回:
| 类型 | 说明 |
|---|---|
list of TokenUsage or TokenUsage |
使用的总 token 数。 |
抛出:
| 类型 | 说明 |
|---|---|
ValueError |
若未提供成本 callback handler。 |
源代码位于 src/ragas/dataset_schema.py
def total_tokens(self) -> t.Union[t.List[TokenUsage], TokenUsage]:
"""
Compute the total tokens used in the evaluation.
Returns
-------
list of TokenUsage or TokenUsage
The total tokens used.
Raises
------
ValueError
If the cost callback handler is not provided.
"""
if self.cost_cb is None:
raise ValueError(
"The evaluate() run was not configured for computing cost. Please provide a token_usage_parser function to evaluate() to compute cost."
)
return self.cost_cb.total_tokens()
total_cost
total_cost(cost_per_input_token: Optional[float] = None, cost_per_output_token: Optional[float] = None, per_model_costs: Dict[str, Tuple[float, float]] = {}) -> float
计算评测总成本。
参数:
| 名称 | 类型 | 说明 | 默认值 |
|---|---|---|---|
cost_per_input_token |
float |
每个输入 token 的成本。默认为 None。 | None |
cost_per_output_token |
float |
每个输出 token 的成本。默认为 None。 | None |
per_model_costs |
dict of str to tuple of float |
各模型成本。默认为空字典。 | {} |
返回:
| 类型 | 说明 |
|---|---|
float |
评测总成本。 |
抛出:
| 类型 | 说明 |
|---|---|
ValueError |
若未提供成本 callback handler。 |
源代码位于 src/ragas/dataset_schema.py
def total_cost(
self,
cost_per_input_token: t.Optional[float] = None,
cost_per_output_token: t.Optional[float] = None,
per_model_costs: t.Dict[str, t.Tuple[float, float]] = {},
) -> float:
"""
Compute the total cost of the evaluation.
Parameters
----------
cost_per_input_token : float, optional
The cost per input token. Default is None.
cost_per_output_token : float, optional
The cost per output token. Default is None.
per_model_costs : dict of str to tuple of float, optional
The per model costs. Default is an empty dictionary.
Returns
-------
float
The total cost of the evaluation.
Raises
------
ValueError
If the cost callback handler is not provided.
"""
if self.cost_cb is None:
raise ValueError(
"The evaluate() run was not configured for computing cost. Please provide a token_usage_parser function to evaluate() to compute cost."
)
return self.cost_cb.total_cost(
cost_per_input_token, cost_per_output_token, per_model_costs
)
MetricAnnotation
基类: BaseModel
from_json
from_json(path: str, metric_name: Optional[str]) -> 'MetricAnnotation'
从 JSON 文件加载标注
源代码位于 src/ragas/dataset_schema.py
@classmethod
def from_json(cls, path: str, metric_name: t.Optional[str]) -> "MetricAnnotation":
"""Load annotations from a JSON file"""
dataset = json.load(open(path))
return cls._process_dataset(dataset, metric_name)
SingleMetricAnnotation
基类: BaseModel
train_test_split
train_test_split(test_size: float = 0.2, seed: int = 42, stratify: Optional[List[Any]] = None) -> Tuple['SingleMetricAnnotation', 'SingleMetricAnnotation']
将数据集拆分为训练集和测试集。
参数: test_size (float): 测试集应包含的数据集比例。 seed (int): 用于可复现性的随机种子。 stratify (list): 用于分层拆分的列值。
源代码位于 src/ragas/dataset_schema.py
def train_test_split(
self,
test_size: float = 0.2,
seed: int = 42,
stratify: t.Optional[t.List[t.Any]] = None,
) -> t.Tuple["SingleMetricAnnotation", "SingleMetricAnnotation"]:
"""
Split the dataset into training and testing sets.
Parameters:
test_size (float): The proportion of the dataset to include in the test split.
seed (int): Random seed for reproducibility.
stratify (list): The column values to stratify the split on.
"""
raise NotImplementedError
sample
sample(n: int, stratify_key: Optional[str] = None) -> 'SingleMetricAnnotation'
创建数据集的子集。
参数: n (int): 子集中要包含的样本数。 stratify_key (str): 用于分层子集的列。
返回: SingleMetricAnnotation: 包含 n 个样本的数据集子集。
源代码位于 src/ragas/dataset_schema.py
def sample(
self, n: int, stratify_key: t.Optional[str] = None
) -> "SingleMetricAnnotation":
"""
Create a subset of the dataset.
Parameters:
n (int): The number of samples to include in the subset.
stratify_key (str): The column to stratify the subset on.
Returns:
SingleMetricAnnotation: A subset of the dataset with `n` samples.
"""
if n > len(self.samples):
raise ValueError(
"Requested sample size exceeds the number of available samples."
)
if stratify_key is None:
# Simple random sampling
sampled_indices = random.sample(range(len(self.samples)), n)
sampled_samples = [self.samples[i] for i in sampled_indices]
else:
# Stratified sampling
class_groups = defaultdict(list)
for idx, sample in enumerate(self.samples):
key = sample[stratify_key]
class_groups[key].append(idx)
# Determine the proportion of samples to take from each class
total_samples = sum(len(indices) for indices in class_groups.values())
proportions = {
cls: len(indices) / total_samples
for cls, indices in class_groups.items()
}
sampled_indices = []
for cls, indices in class_groups.items():
cls_sample_count = int(np.round(proportions[cls] * n))
cls_sample_count = min(
cls_sample_count, len(indices)
) # Don't oversample
sampled_indices.extend(random.sample(indices, cls_sample_count))
# Handle any rounding discrepancies to ensure exactly `n` samples
while len(sampled_indices) < n:
remaining_indices = set(range(len(self.samples))) - set(sampled_indices)
if not remaining_indices:
break
sampled_indices.append(random.choice(list(remaining_indices)))
sampled_samples = [self.samples[i] for i in sampled_indices]
return SingleMetricAnnotation(name=self.name, samples=sampled_samples)
batch
batch(batch_size: int, drop_last_batch: bool = False)
创建批次迭代器。
参数: batch_size (int): 每批样本数。 stratify (str): 用于分层批次的列。 drop_last_batch (bool): 若最后一批小于指定批次大小是否丢弃。
源代码位于 src/ragas/dataset_schema.py
def batch(
self,
batch_size: int,
drop_last_batch: bool = False,
):
"""
Create a batch iterator.
Parameters:
batch_size (int): The number of samples in each batch.
stratify (str): The column to stratify the batches on.
drop_last_batch (bool): Whether to drop the last batch if it is smaller than the specified batch size.
"""
samples = self.samples[:]
random.shuffle(samples)
all_batches = [
samples[i : i + batch_size]
for i in range(0, len(samples), batch_size)
if len(samples[i : i + batch_size]) == batch_size or not drop_last_batch
]
return all_batches
stratified_batches
stratified_batches(batch_size: int, stratify_key: str, drop_last_batch: bool = False, replace: bool = False) -> List[List[SampleAnnotation]]
基于指定键创建分层批次,确保比例代表性。
参数: batch_size (int): 每批样本数。 stratify_key (str): metric_input 中用于分层的键(例如类别标签)。 drop_last_batch (bool): 若为 True,当最后一批样本少于 batch_size 时丢弃。 replace (bool): If True, allows reusing samples from the same class to fill a batch if necessary.
返回: List[List[SampleAnnotation]]: 分层批次列表,每批是 SampleAnnotation 对象列表。
源代码位于 src/ragas/dataset_schema.py
def stratified_batches(
self,
batch_size: int,
stratify_key: str,
drop_last_batch: bool = False,
replace: bool = False,
) -> t.List[t.List[SampleAnnotation]]:
"""
Create stratified batches based on a specified key, ensuring proportional representation.
Parameters:
batch_size (int): Number of samples per batch.
stratify_key (str): Key in `metric_input` used for stratification (e.g., class labels).
drop_last_batch (bool): If True, drops the last batch if it has fewer samples than `batch_size`.
replace (bool): If True, allows reusing samples from the same class to fill a batch if necessary.
Returns:
List[List[SampleAnnotation]]: A list of stratified batches, each batch being a list of SampleAnnotation objects.
"""
# Group samples based on the stratification key
class_groups = defaultdict(list)
for sample in self.samples:
key = sample[stratify_key]
class_groups[key].append(sample)
# Shuffle each class group for randomness
for group in class_groups.values():
random.shuffle(group)
# Determine the number of batches required
total_samples = len(self.samples)
num_batches = (
np.ceil(total_samples / batch_size).astype(int)
if drop_last_batch
else np.floor(total_samples / batch_size).astype(int)
)
samples_per_class_per_batch = {
cls: max(1, len(samples) // num_batches)
for cls, samples in class_groups.items()
}
# Create stratified batches
all_batches = []
while len(all_batches) < num_batches:
batch = []
for cls, samples in list(class_groups.items()):
# Determine the number of samples to take from this class
count = min(
samples_per_class_per_batch[cls],
len(samples),
batch_size - len(batch),
)
if count > 0:
# Add samples from the current class
batch.extend(samples[:count])
class_groups[cls] = samples[count:] # Remove used samples
elif replace and len(batch) < batch_size:
# Reuse samples if `replace` is True
batch.extend(random.choices(samples, k=batch_size - len(batch)))
# Shuffle the batch to mix classes
random.shuffle(batch)
if len(batch) == batch_size or not drop_last_batch:
all_batches.append(batch)
return all_batches
get_prompt_annotations
get_prompt_annotations() -> Dict[str, List[PromptAnnotation]]
以列表形式获取每个 prompt 的全部 prompt 标注。
源代码位于 src/ragas/dataset_schema.py
def get_prompt_annotations(self) -> t.Dict[str, t.List[PromptAnnotation]]:
"""
Get all the prompt annotations for each prompt as a list.
"""
prompt_annotations = defaultdict(list)
for sample in self.samples:
if sample.is_accepted:
for prompt_name, prompt_annotation in sample.prompts.items():
prompt_annotations[prompt_name].append(prompt_annotation)
return prompt_annotations
Message
基类: BaseModel
表示通用 message。
属性:
| 名称 | 类型 | 说明 |
|---|---|---|
content |
str |
message 的内容。 |
metadata |
(Optional[Dict[str, Any]], optional) |
与该 message 关联的额外元数据。 |
ToolCall
基类: BaseModel
表示带有名称和参数的工具调用。
参数:
| 名称 | 类型 | 说明 | 默认值 |
|---|---|---|---|
name |
str |
被调用工具的名称。 | required |
args |
Dict[str, Any] |
工具调用的参数字典,键为参数名,值可以是字符串、整数或浮点数。 | required |
HumanMessage
基类: Message
表示来自人类用户的 message。
属性:
| 名称 | 类型 | 说明 |
|---|---|---|
type |
Literal[human] |
message 类型,始终为 "human"。 |
方法:
| Name | Description |
|---|---|
pretty_repr |
返回 human message 的格式化字符串表示。 |
pretty_repr
pretty_repr()
返回 human message 的格式化字符串表示。
源代码位于 src/ragas/messages.py
def pretty_repr(self):
"""Returns a formatted string representation of the human message."""
return f"Human: {self.content}"
ToolMessage
基类: Message
表示来自工具的 message。
属性:
| 名称 | 类型 | 说明 |
|---|---|---|
type |
Literal[tool] |
message 类型,始终为 "tool"。 |
方法:
| Name | Description |
|---|---|
pretty_repr |
返回 tool message 的格式化字符串表示。 |
pretty_repr
pretty_repr()
返回 tool message 的格式化字符串表示。
源代码位于 src/ragas/messages.py
def pretty_repr(self):
"""Returns a formatted string representation of the tool message."""
return f"ToolOutput: {self.content}"
AIMessage
基类: Message
表示来自 AI 的 message。
属性:
| 名称 | 类型 | 说明 |
|---|---|---|
type |
Literal[ai] |
message 类型,始终为 "ai"。 |
tool_calls |
Optional[List[ToolCall]] |
AI 发出的工具调用列表(若有)。 |
metadata |
Optional[Dict[str, Any]] |
与 AI message 关联的额外元数据。 |
方法:
| Name | Description |
|---|---|
dict |
返回 AI message 的字典表示。 |
pretty_repr |
返回 AI message 的格式化字符串表示。 |
to_dict
to_dict(**kwargs)
返回 AI message 的字典表示。
源代码位于 src/ragas/messages.py
def to_dict(self, **kwargs):
"""
Returns a dictionary representation of the AI message.
"""
content = (
self.content
if self.tool_calls is None
else {
"text": self.content,
"tool_calls": [tc.dict() for tc in self.tool_calls],
}
)
return {"content": content, "type": self.type}
pretty_repr
pretty_repr()
返回 AI message 的格式化字符串表示。
源代码位于 src/ragas/messages.py
def pretty_repr(self):
"""
Returns a formatted string representation of the AI message.
"""
lines = []
if self.content != "":
lines.append(f"AI: {self.content}")
if self.tool_calls is not None:
lines.append("Tools:")
for tc in self.tool_calls:
lines.append(f" {tc.name}: {tc.args}")
return "\n".join(lines)
ragas.evaluation.EvaluationResult
EvaluationResult(scores: List[Dict[str, Any]], dataset: EvaluationDataset, binary_columns: List[str] = list(), cost_cb: Optional[CostCallbackHandler] = None, traces: List[Dict[str, Any]] = list(), ragas_traces: Dict[str, ChainRun] = dict(), run_id: Optional[UUID] = None)
用于存储和处理评测结果的类。
属性:
| 名称 | 类型 | 说明 |
|---|---|---|
scores |
Dataset |
包含评测分数的数据集。 |
dataset |
(Dataset, optional) |
用于评测的原始数据集。默认为 None。 |
binary_columns |
list of str, optional |
作为二元指标的列列表。默认为空列表。 |
cost_cb |
(CostCallbackHandler, optional) |
用于成本计算的 callback handler。默认为 None。 |
to_pandas
to_pandas(batch_size: int | None = None, batched: bool = False)
将结果转换为 pandas DataFrame。
参数:
| 名称 | 类型 | 说明 | 默认值 |
|---|---|---|---|
batch_size |
int |
转换的批次大小。默认为 None。 | None |
batched |
bool |
是否分批转换。默认为 False。 | False |
返回:
| 类型 | 说明 |
|---|---|
DataFrame |
作为 pandas DataFrame 的结果。 |
抛出:
| 类型 | 说明 |
|---|---|
ValueError |
若未提供数据集。 |
源代码位于 src/ragas/dataset_schema.py
def to_pandas(self, batch_size: int | None = None, batched: bool = False):
"""
Convert the result to a pandas DataFrame.
Parameters
----------
batch_size : int, optional
The batch size for conversion. Default is None.
batched : bool, optional
Whether to convert in batches. Default is False.
Returns
-------
pandas.DataFrame
The result as a pandas DataFrame.
Raises
------
ValueError
If the dataset is not provided.
"""
try:
import pandas as pd
except ImportError:
raise ImportError(
"pandas is not installed. Please install it to use this function."
)
if self.dataset is None:
raise ValueError("dataset is not provided for the results class")
assert len(self.scores) == len(self.dataset)
# convert both to pandas dataframes and concatenate
scores_df = pd.DataFrame(self.scores)
dataset_df = self.dataset.to_pandas()
return pd.concat([dataset_df, scores_df], axis=1)
total_tokens
total_tokens() -> Union[List[TokenUsage], TokenUsage]
计算评测中使用的总 token 数。
返回:
| 类型 | 说明 |
|---|---|
list of TokenUsage or TokenUsage |
使用的总 token 数。 |
抛出:
| 类型 | 说明 |
|---|---|
ValueError |
若未提供成本 callback handler。 |
源代码位于 src/ragas/dataset_schema.py
def total_tokens(self) -> t.Union[t.List[TokenUsage], TokenUsage]:
"""
Compute the total tokens used in the evaluation.
Returns
-------
list of TokenUsage or TokenUsage
The total tokens used.
Raises
------
ValueError
If the cost callback handler is not provided.
"""
if self.cost_cb is None:
raise ValueError(
"The evaluate() run was not configured for computing cost. Please provide a token_usage_parser function to evaluate() to compute cost."
)
return self.cost_cb.total_tokens()
total_cost
total_cost(cost_per_input_token: Optional[float] = None, cost_per_output_token: Optional[float] = None, per_model_costs: Dict[str, Tuple[float, float]] = {}) -> float
计算评测总成本。
参数:
| 名称 | 类型 | 说明 | 默认值 |
|---|---|---|---|
cost_per_input_token |
float |
每个输入 token 的成本。默认为 None。 | None |
cost_per_output_token |
float |
每个输出 token 的成本。默认为 None。 | None |
per_model_costs |
dict of str to tuple of float |
各模型成本。默认为空字典。 | {} |
返回:
| 类型 | 说明 |
|---|---|
float |
评测总成本。 |
抛出:
| 类型 | 说明 |
|---|---|
ValueError |
若未提供成本 callback handler。 |
源代码位于 src/ragas/dataset_schema.py
def total_cost(
self,
cost_per_input_token: t.Optional[float] = None,
cost_per_output_token: t.Optional[float] = None,
per_model_costs: t.Dict[str, t.Tuple[float, float]] = {},
) -> float:
"""
Compute the total cost of the evaluation.
Parameters
----------
cost_per_input_token : float, optional
The cost per input token. Default is None.
cost_per_output_token : float, optional
The cost per output token. Default is None.
per_model_costs : dict of str to tuple of float, optional
The per model costs. Default is an empty dictionary.
Returns
-------
float
The total cost of the evaluation.
Raises
------
ValueError
If the cost callback handler is not provided.
"""
if self.cost_cb is None:
raise ValueError(
"The evaluate() run was not configured for computing cost. Please provide a token_usage_parser function to evaluate() to compute cost."
)
return self.cost_cb.total_cost(
cost_per_input_token, cost_per_output_token, per_model_costs
)