BaseGraphTransformation

BaseGraphTransformation(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)())

基类： ABC

对 KnowledgeGraph 进行图变换的抽象基类。

transform

transform(kg: KnowledgeGraph) -> Any

变换 KnowledgeGraph 的抽象方法。变换应是幂等的，即多次应用应与应用一次得到相同结果。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

@abstractmethod
async def transform(self, kg: KnowledgeGraph) -> t.Any:
    """
    Abstract method to transform the KnowledgeGraph. Transformations should be
    idempotent, meaning that applying the transformation multiple times should
    yield the same result as applying it once.

    Parameters
    ----------
    kg : KnowledgeGraph
        The knowledge graph to be transformed.

    Returns
    -------
    t.Any
        The transformed knowledge graph.
    """
    pass

filter

filter(kg: KnowledgeGraph) -> KnowledgeGraph

过滤 KnowledgeGraph 并返回过滤后的图。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

def filter(self, kg: KnowledgeGraph) -> KnowledgeGraph:
    """
    Filters the KnowledgeGraph and returns the filtered graph.

    Parameters
    ----------
    kg : KnowledgeGraph
        The knowledge graph to be filtered.

    Returns
    -------
    KnowledgeGraph
        The filtered knowledge graph.
    """
    logger.debug("Filtering KnowledgeGraph with %s", self.filter_nodes.__name__)
    filtered_nodes = [node for node in kg.nodes if self.filter_nodes(node)]
    node_ids = {node.id for node in filtered_nodes}
    filtered_relationships = [
        rel
        for rel in kg.relationships
        if (rel.source.id in node_ids) and (rel.target.id in node_ids)
    ]
    logger.debug(
        "Filter reduced KnowledgeGraph by %d/%d nodes and %d/%d relationships",
        len(kg.nodes) - len(filtered_nodes),
        len(kg.nodes),
        len(kg.relationships) - len(filtered_relationships),
        len(kg.relationships),
    )
    return KnowledgeGraph(
        nodes=filtered_nodes,
        relationships=filtered_relationships,
    )

generate_execution_plan

generate_execution_plan(kg: KnowledgeGraph) -> Sequence[Coroutine]

生成由 Executor 按顺序执行的一组协程。该协程执行时会把变换写入 KnowledgeGraph。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

@abstractmethod
def generate_execution_plan(self, kg: KnowledgeGraph) -> t.Sequence[t.Coroutine]:
    """
    Generates a sequence of coroutines to be executed in sequence by the Executor. This
    coroutine will, upon execution, write the transformation into the KnowledgeGraph.

    Parameters
    ----------
    kg : KnowledgeGraph
        The knowledge graph to be transformed.

    Returns
    -------
    t.Sequence[t.Coroutine]
        A sequence of coroutines to be executed in parallel.
    """
    pass

Extractor

Extractor(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)())

基类： BaseGraphTransformation

通过从节点提取特定属性来变换 KnowledgeGraph 的提取器抽象基类。

方法：

transform

transform(kg: KnowledgeGraph) -> List[Tuple[Node, Tuple[str, Any]]]

通过从节点提取属性来变换 KnowledgeGraph。使用 filter 方法过滤图，使用 extract 方法从每个节点提取属性。

参数：

返回：

示例：

>>> kg = KnowledgeGraph(nodes=[Node(id=1, properties={"name": "Node1"}), Node(id=2, properties={"name": "Node2"})])
>>> extractor = SomeConcreteExtractor()
>>> extractor.transform(kg)
[(Node(id=1, properties={"name": "Node1"}), ("property_name", "extracted_value")),
 (Node(id=2, properties={"name": "Node2"}), ("property_name", "extracted_value"))]

源代码位于 src/ragas/testset/transforms/base.py

async def transform(
    self, kg: KnowledgeGraph
) -> t.List[t.Tuple[Node, t.Tuple[str, t.Any]]]:
    """
    Transforms the KnowledgeGraph by extracting properties from its nodes. Uses
    the `filter` method to filter the graph and the `extract` method to extract
    properties from each node.

    Parameters
    ----------
    kg : KnowledgeGraph
        The knowledge graph to be transformed.

    Returns
    -------
    t.List[t.Tuple[Node, t.Tuple[str, t.Any]]]
        A list of tuples where each tuple contains a node and the extracted
        property.

    Examples
    --------
    >>> kg = KnowledgeGraph(nodes=[Node(id=1, properties={"name": "Node1"}), Node(id=2, properties={"name": "Node2"})])
    >>> extractor = SomeConcreteExtractor()
    >>> extractor.transform(kg)
    [(Node(id=1, properties={"name": "Node1"}), ("property_name", "extracted_value")),
     (Node(id=2, properties={"name": "Node2"}), ("property_name", "extracted_value"))]
    """
    filtered = self.filter(kg)
    return [(node, await self.extract(node)) for node in filtered.nodes]

extract

extract(node: Node) -> Tuple[str, Any]

从节点提取特定属性的抽象方法。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

@abstractmethod
async def extract(self, node: Node) -> t.Tuple[str, t.Any]:
    """
    Abstract method to extract a specific property from a node.

    Parameters
    ----------
    node : Node
        The node from which to extract the property.

    Returns
    -------
    t.Tuple[str, t.Any]
        A tuple containing the property name and the extracted value.
    """
    pass

generate_execution_plan

generate_execution_plan(kg: KnowledgeGraph) -> Sequence[Coroutine]

生成由 Executor 并行执行的一组协程。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

def generate_execution_plan(self, kg: KnowledgeGraph) -> t.Sequence[t.Coroutine]:
    """
    Generates a sequence of coroutines to be executed in parallel by the Executor.

    Parameters
    ----------
    kg : KnowledgeGraph
        The knowledge graph to be transformed.

    Returns
    -------
    t.Sequence[t.Coroutine]
        A sequence of coroutines to be executed in parallel.
    """

    async def apply_extract(node: Node):
        property_name, property_value = await self.extract(node)
        if node.get_property(property_name) is None:
            node.add_property(property_name, property_value)
        else:
            logger.warning(
                "Property '%s' already exists in node '%.6s'. Skipping!",
                property_name,
                node.id,
            )

    filtered = self.filter(kg)
    plan = [apply_extract(node) for node in filtered.nodes]
    logger.debug(
        "Created %d coroutines for %s",
        len(plan),
        self.__class__.__name__,
    )
    return plan

NodeFilter

NodeFilter(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)())

基类： BaseGraphTransformation

custom_filter

custom_filter(node: Node, kg: KnowledgeGraph) -> bool

基于 prompt 过滤节点的抽象方法。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

@abstractmethod
async def custom_filter(self, node: Node, kg: KnowledgeGraph) -> bool:
    """
    Abstract method to filter a node based on a prompt.

    Parameters
    ----------
    node : Node
        The node to be filtered.

    Returns
    -------
    bool
        A boolean indicating whether the node should be filtered.
    """
    pass

generate_execution_plan

generate_execution_plan(kg: KnowledgeGraph) -> Sequence[Coroutine]

生成要执行的一组协程

源代码位于 src/ragas/testset/transforms/base.py

def generate_execution_plan(self, kg: KnowledgeGraph) -> t.Sequence[t.Coroutine]:
    """
    Generates a sequence of coroutines to be executed
    """

    async def apply_filter(node: Node):
        if await self.custom_filter(node, kg):
            kg.remove_node(node)

    filtered = self.filter(kg)
    plan = [apply_filter(node) for node in filtered.nodes]
    logger.debug(
        "Created %d coroutines for %s",
        len(plan),
        self.__class__.__name__,
    )
    return plan

RelationshipBuilder

RelationshipBuilder(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)())

基类： BaseGraphTransformation

在 KnowledgeGraph 中构建关系的抽象基类。

方法：

transform

transform(kg: KnowledgeGraph) -> List[Relationship]

通过构建关系来变换 KnowledgeGraph。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

@abstractmethod
async def transform(self, kg: KnowledgeGraph) -> t.List[Relationship]:
    """
    Transforms the KnowledgeGraph by building relationships.

    Parameters
    ----------
    kg : KnowledgeGraph
        The knowledge graph to be transformed.

    Returns
    -------
    t.List[Relationship]
        A list of new relationships.
    """
    pass

generate_execution_plan

generate_execution_plan(kg: KnowledgeGraph) -> Sequence[Coroutine]

生成由 Executor 并行执行的一组协程。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

def generate_execution_plan(self, kg: KnowledgeGraph) -> t.Sequence[t.Coroutine]:
    """
    Generates a sequence of coroutines to be executed in parallel by the Executor.

    Parameters
    ----------
    kg : KnowledgeGraph
        The knowledge graph to be transformed.

    Returns
    -------
    t.Sequence[t.Coroutine]
        A sequence of coroutines to be executed in parallel.
    """

    async def apply_build_relationships(
        filtered_kg: KnowledgeGraph, original_kg: KnowledgeGraph
    ):
        relationships = await self.transform(filtered_kg)
        original_kg.relationships.extend(relationships)

    filtered_kg = self.filter(kg)
    plan = [apply_build_relationships(filtered_kg=filtered_kg, original_kg=kg)]
    logger.debug(
        "Created %d coroutines for %s",
        len(plan),
        self.__class__.__name__,
    )
    return plan

Splitter

Splitter(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)())

基类： BaseGraphTransformation

通过把节点切成更小 chunk 来变换 KnowledgeGraph 的 splitter 抽象基类。

方法：

transform

transform(kg: KnowledgeGraph) -> Tuple[List[Node], List[Relationship]]

通过把节点切成更小 chunk 来变换 KnowledgeGraph。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

async def transform(
    self, kg: KnowledgeGraph
) -> t.Tuple[t.List[Node], t.List[Relationship]]:
    """
    Transforms the KnowledgeGraph by splitting its nodes into smaller chunks.

    Parameters
    ----------
    kg : KnowledgeGraph
        The knowledge graph to be transformed.

    Returns
    -------
    t.Tuple[t.List[Node], t.List[Relationship]]
        A tuple containing a list of new nodes and a list of new relationships.
    """
    filtered = self.filter(kg)

    all_nodes = []
    all_relationships = []
    for node in filtered.nodes:
        nodes, relationships = await self.split(node)
        all_nodes.extend(nodes)
        all_relationships.extend(relationships)

    return all_nodes, all_relationships

split

split(node: Node) -> Tuple[List[Node], List[Relationship]]

把节点切成更小 chunk 的抽象方法。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

@abstractmethod
async def split(self, node: Node) -> t.Tuple[t.List[Node], t.List[Relationship]]:
    """
    Abstract method to split a node into smaller chunks.

    Parameters
    ----------
    node : Node
        The node to be split.

    Returns
    -------
    t.Tuple[t.List[Node], t.List[Relationship]]
        A tuple containing a list of new nodes and a list of new relationships.
    """
    pass

generate_execution_plan

generate_execution_plan(kg: KnowledgeGraph) -> Sequence[Coroutine]

生成由 Executor 并行执行的一组协程。

参数：

返回：

源代码位于 src/ragas/testset/transforms/base.py

def generate_execution_plan(self, kg: KnowledgeGraph) -> t.Sequence[t.Coroutine]:
    """
    Generates a sequence of coroutines to be executed in parallel by the Executor.

    Parameters
    ----------
    kg : KnowledgeGraph
        The knowledge graph to be transformed.

    Returns
    -------
    t.Sequence[t.Coroutine]
        A sequence of coroutines to be executed in parallel.
    """

    async def apply_split(node: Node):
        nodes, relationships = await self.split(node)
        kg.nodes.extend(nodes)
        kg.relationships.extend(relationships)

    filtered = self.filter(kg)
    plan = [apply_split(node) for node in filtered.nodes]
    logger.debug(
        "Created %d coroutines for %s",
        len(plan),
        self.__class__.__name__,
    )
    return plan

Parallel

Parallel(*transformations: Union[BaseGraphTransformation, 'Parallel'])

要并行应用的变换集合。

示例：

>>> Parallel(HeadlinesExtractor(), SummaryExtractor())

源代码位于 src/ragas/testset/transforms/engine.py

def __init__(self, *transformations: t.Union[BaseGraphTransformation, "Parallel"]):
    self.transformations = list(transformations)

EmbeddingExtractor

EmbeddingExtractor(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)(), property_name: str = 'embedding', embed_property_name: str = 'page_content', embedding_model: Union[BaseRagasEmbeddings, BaseRagasEmbedding] = embedding_factory())

基类： Extractor

从知识图谱节点提取嵌入的类。

属性：

extract

extract(node: Node) -> Tuple[str, Any]

提取给定节点的嵌入。

抛出：

源代码位于 src/ragas/testset/transforms/extractors/embeddings.py

async def extract(self, node: Node) -> t.Tuple[str, t.Any]:
    """
    Extracts the embedding for a given node.

    Raises
    ------
    ValueError
        If the property to be embedded is not a string.
    """
    text = node.get_property(self.embed_property_name)
    if not isinstance(text, str):
        raise ValueError(
            f"node.property('{self.embed_property_name}') must be a string, found '{type(text)}'"
        )

    # Handle both modern (BaseRagasEmbedding) and legacy (BaseRagasEmbeddings) interfaces
    if hasattr(self.embedding_model, "aembed_text"):
        # Modern interface (BaseRagasEmbedding)
        # Check if the client supports async operations by checking if is_async exists and is True
        if hasattr(self.embedding_model, "is_async") and getattr(
            self.embedding_model, "is_async", False
        ):
            embedding = await self.embedding_model.aembed_text(text)  # type: ignore[attr-defined]
        else:
            # For sync clients, use the sync method wrapped in thread executor to avoid blocking
            warnings.warn(
                f"Using sync embedding model {self.embedding_model.__class__.__name__} "
                f"in async context. This may impact performance. "
                f"Consider using an async-compatible embedding model for better performance.",
                UserWarning,
                stacklevel=2,
            )
            embedding = await run_sync_in_async(
                self.embedding_model.embed_text, text
            )  # type: ignore[attr-defined]
    else:
        # Legacy interface (BaseRagasEmbeddings)
        embedding = await self.embedding_model.embed_text(text)  # type: ignore[misc]

    return self.property_name, embedding

HeadlinesExtractor

HeadlinesExtractor(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)(), llm: Union[BaseRagasLLM, InstructorBaseRagasLLM] = _default_llm_factory(), merge_if_possible: bool = True, max_token_limit: int = 32000, tokenizer: BaseTokenizer = (lambda: DEFAULT_TOKENIZER)(), property_name: str = 'headlines', prompt: HeadlinesExtractorPrompt = HeadlinesExtractorPrompt(), max_num: int = 5)

基类： LLMBasedExtractor

从给定文本提取 headlines。

属性：

KeyphrasesExtractor

KeyphrasesExtractor(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)(), llm: Union[BaseRagasLLM, InstructorBaseRagasLLM] = _default_llm_factory(), merge_if_possible: bool = True, max_token_limit: int = 32000, tokenizer: BaseTokenizer = (lambda: DEFAULT_TOKENIZER)(), property_name: str = 'keyphrases', prompt: KeyphrasesExtractorPrompt = KeyphrasesExtractorPrompt(), max_num: int = 5)

基类： LLMBasedExtractor

从给定文本提取 top 关键短语。

属性：

SummaryExtractor

SummaryExtractor(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)(), llm: Union[BaseRagasLLM, InstructorBaseRagasLLM] = _default_llm_factory(), merge_if_possible: bool = True, max_token_limit: int = 32000, tokenizer: BaseTokenizer = (lambda: DEFAULT_TOKENIZER)(), property_name: str = 'summary', prompt: SummaryExtractorPrompt = SummaryExtractorPrompt())

基类： LLMBasedExtractor

从给定文本提取摘要。

属性：

TitleExtractor

TitleExtractor(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)(), llm: Union[BaseRagasLLM, InstructorBaseRagasLLM] = _default_llm_factory(), merge_if_possible: bool = True, max_token_limit: int = 32000, tokenizer: BaseTokenizer = (lambda: DEFAULT_TOKENIZER)(), property_name: str = 'title', prompt: TitleExtractorPrompt = TitleExtractorPrompt())

基类： LLMBasedExtractor

从给定文本提取标题。

属性：

CustomNodeFilter

CustomNodeFilter(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)(), llm: Union[BaseRagasLLM, InstructorBaseRagasLLM] = _default_llm_factory(), scoring_prompt: PydanticPrompt = QuestionPotentialPrompt(), min_score: int = 2, rubrics: Dict[str, str] = (lambda: DEFAULT_RUBRICS)())

基类： LLMBasedNodeFilter

若分数小于 min_score 则返回 True

CosineSimilarityBuilder

CosineSimilarityBuilder(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)(), property_name: str = 'embedding', new_property_name: str = 'cosine_similarity', threshold: float = 0.9, block_size: int = 1024)

基类： RelationshipBuilder

generate_execution_plan

generate_execution_plan(kg: KnowledgeGraph) -> List[Coroutine]

生成用于查找相似嵌入对的协程任务，可由 Executor 调度/执行。

源代码位于 src/ragas/testset/transforms/relationship_builders/cosine.py

def generate_execution_plan(self, kg: KnowledgeGraph) -> t.List[t.Coroutine]:
    """
    Generates a coroutine task for finding similar embedding pairs, which can be scheduled/executed by an Executor.
    """
    filtered_kg = self.filter(kg)

    embeddings = []
    for node in filtered_kg.nodes:
        embedding = node.get_property(self.property_name)
        if embedding is None:
            raise ValueError(f"Node {node.id} has no {self.property_name}")
        embeddings.append(embedding)
    self._validate_embedding_shapes(embeddings)

    async def find_and_add_relationships():
        similar_pairs = self._find_similar_embedding_pairs(
            np.array(embeddings), self.threshold
        )
        for i, j, similarity_float in similar_pairs:
            rel = Relationship(
                source=filtered_kg.nodes[i],
                target=filtered_kg.nodes[j],
                type=self.new_property_name,
                properties={self.new_property_name: similarity_float},
                bidirectional=True,
            )
            kg.relationships.append(rel)

    return [find_and_add_relationships()]

SummaryCosineSimilarityBuilder

SummaryCosineSimilarityBuilder(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)(), property_name: str = 'summary_embedding', new_property_name: str = 'summary_cosine_similarity', threshold: float = 0.1, block_size: int = 1024)

基类： CosineSimilarityBuilder

JaccardSimilarityBuilder

JaccardSimilarityBuilder(name: str = '', filter_nodes: Callable[[Node], bool] = (lambda: default_filter)(), property_name: str = 'entities', key_name: Optional[str] = None, new_property_name: str = 'jaccard_similarity', threshold: float = 0.5)

基类： RelationshipBuilder

generate_execution_plan

generate_execution_plan(kg: KnowledgeGraph) -> List[Coroutine]

生成用于查找相似对的协程任务，可由 Executor 调度/执行。

源代码位于 src/ragas/testset/transforms/relationship_builders/traditional.py

def generate_execution_plan(self, kg: KnowledgeGraph) -> t.List[t.Coroutine]:
    """
    Generates a coroutine task for finding similar pairs, which can be scheduled/executed by an Executor.
    """

    async def find_and_add_relationships():
        similar_pairs = self._find_similar_embedding_pairs(kg)
        for i, j, similarity_float in similar_pairs:
            rel = Relationship(
                source=kg.nodes[i],
                target=kg.nodes[j],
                type=self.new_property_name,
                properties={self.new_property_name: similarity_float},
                bidirectional=True,
            )
            kg.relationships.append(rel)

    return [find_and_add_relationships()]

default_transforms

default_transforms(documents: List[Document], llm: Union[BaseRagasLLM, 'InstructorBaseRagasLLM'], embedding_model: BaseRagasEmbeddings) -> 'Transforms'

创建并返回用于处理知识图谱的默认 transforms 集合。

该函数定义要对知识图谱应用的一系列变换步骤，包括提取摘要、关键短语、标题、headline 和嵌入，以及在节点之间构建相似度关系。

返回：

源代码位于 src/ragas/testset/transforms/default.py

def default_transforms(
    documents: t.List[LCDocument],
    llm: t.Union[BaseRagasLLM, "InstructorBaseRagasLLM"],
    embedding_model: BaseRagasEmbeddings,
) -> "Transforms":
    """
    Creates and returns a default set of transforms for processing a knowledge graph.

    This function defines a series of transformation steps to be applied to a
    knowledge graph, including extracting summaries, keyphrases, titles,
    headlines, and embeddings, as well as building similarity relationships
    between nodes.



    Returns
    -------
    Transforms
        A list of transformation steps to be applied to the knowledge graph.

    """

    def count_doc_length_bins(documents, bin_ranges):
        data = [num_tokens_from_string(doc.page_content) for doc in documents]
        bins = {f"{start}-{end}": 0 for start, end in bin_ranges}

        for num in data:
            for start, end in bin_ranges:
                if start <= num <= end:
                    bins[f"{start}-{end}"] += 1
                    break  # Move to the next number once it’s placed in a bin

        return bins

    def filter_doc_with_num_tokens(node, min_num_tokens=500):
        return (
            node.type == NodeType.DOCUMENT
            and num_tokens_from_string(node.properties["page_content"]) > min_num_tokens
        )

    def filter_docs(node):
        return node.type == NodeType.DOCUMENT

    def filter_chunks(node):
        return node.type == NodeType.CHUNK

    bin_ranges = [(0, 100), (101, 500), (501, float("inf"))]
    result = count_doc_length_bins(documents, bin_ranges)
    result = {k: v / len(documents) for k, v in result.items()}

    transforms = []

    if result["501-inf"] >= 0.25:
        headline_extractor = HeadlinesExtractor(
            llm=llm, filter_nodes=lambda node: filter_doc_with_num_tokens(node)
        )
        splitter = HeadlineSplitter(min_tokens=500)
        summary_extractor = SummaryExtractor(
            llm=llm, filter_nodes=lambda node: filter_doc_with_num_tokens(node)
        )

        theme_extractor = ThemesExtractor(
            llm=llm, filter_nodes=lambda node: filter_chunks(node)
        )
        ner_extractor = NERExtractor(
            llm=llm, filter_nodes=lambda node: filter_chunks(node)
        )

        summary_emb_extractor = EmbeddingExtractor(
            embedding_model=embedding_model,
            property_name="summary_embedding",
            embed_property_name="summary",
            filter_nodes=lambda node: filter_doc_with_num_tokens(node),
        )

        cosine_sim_builder = CosineSimilarityBuilder(
            property_name="summary_embedding",
            new_property_name="summary_similarity",
            threshold=0.7,
            filter_nodes=lambda node: filter_doc_with_num_tokens(node),
        )

        ner_overlap_sim = OverlapScoreBuilder(
            threshold=0.01, filter_nodes=lambda node: filter_chunks(node)
        )

        node_filter = CustomNodeFilter(
            llm=llm, filter_nodes=lambda node: filter_chunks(node)
        )
        transforms = [
            headline_extractor,
            splitter,
            summary_extractor,
            node_filter,
            Parallel(summary_emb_extractor, theme_extractor, ner_extractor),
            Parallel(cosine_sim_builder, ner_overlap_sim),
        ]
    elif result["101-500"] >= 0.25:
        summary_extractor = SummaryExtractor(
            llm=llm, filter_nodes=lambda node: filter_doc_with_num_tokens(node, 100)
        )
        summary_emb_extractor = EmbeddingExtractor(
            embedding_model=embedding_model,
            property_name="summary_embedding",
            embed_property_name="summary",
            filter_nodes=lambda node: filter_doc_with_num_tokens(node, 100),
        )

        cosine_sim_builder = CosineSimilarityBuilder(
            property_name="summary_embedding",
            new_property_name="summary_similarity",
            threshold=0.5,
            filter_nodes=lambda node: filter_doc_with_num_tokens(node, 100),
        )

        ner_extractor = NERExtractor(llm=llm)
        ner_overlap_sim = OverlapScoreBuilder(threshold=0.01)
        theme_extractor = ThemesExtractor(
            llm=llm, filter_nodes=lambda node: filter_docs(node)
        )
        node_filter = CustomNodeFilter(llm=llm)

        transforms = [
            summary_extractor,
            node_filter,
            Parallel(summary_emb_extractor, theme_extractor, ner_extractor),
            Parallel(cosine_sim_builder, ner_overlap_sim),
        ]
    else:
        raise ValueError(
            "Documents appears to be too short (ie 100 tokens or less). Please provide longer documents."
        )

    return transforms

default_transforms_for_prechunked

default_transforms_for_prechunked(llm: Union[BaseRagasLLM, 'InstructorBaseRagasLLM'], embedding_model: BaseRagasEmbeddings) -> 'Transforms'

创建并返回用于处理已预切分文档之知识图谱的默认 transforms 集合。

这会跳过切分步骤，直接对 chunks 应用提取器和关系构建器。

源代码位于 src/ragas/testset/transforms/default.py

def default_transforms_for_prechunked(
    llm: t.Union[BaseRagasLLM, "InstructorBaseRagasLLM"],
    embedding_model: BaseRagasEmbeddings,
) -> "Transforms":
    """
    Creates and returns a default set of transforms for processing a knowledge graph
    containing pre-chunked documents.

    This ignores the splitting step and directly applies extractors and relationship builders
    to the chunks.
    """

    def filter_chunks(node):
        return node.type == NodeType.CHUNK

    summary_extractor = SummaryExtractor(llm=llm, filter_nodes=filter_chunks)
    summary_emb_extractor = EmbeddingExtractor(
        embedding_model=embedding_model,
        property_name="summary_embedding",
        embed_property_name="summary",
        filter_nodes=filter_chunks,
    )

    theme_extractor = ThemesExtractor(llm=llm, filter_nodes=filter_chunks)
    ner_extractor = NERExtractor(llm=llm, filter_nodes=filter_chunks)

    cosine_sim_builder = CosineSimilarityBuilder(
        property_name="summary_embedding",
        new_property_name="summary_similarity",
        threshold=0.7,
        filter_nodes=filter_chunks,
    )

    ner_overlap_sim = OverlapScoreBuilder(threshold=0.01, filter_nodes=filter_chunks)

    node_filter = CustomNodeFilter(llm=llm, filter_nodes=filter_chunks)

    return [
        summary_extractor,
        node_filter,
        Parallel(summary_emb_extractor, theme_extractor, ner_extractor),
        Parallel(cosine_sim_builder, ner_overlap_sim),
    ]

apply_transforms

apply_transforms(kg: KnowledgeGraph, transforms: Transforms, run_config: RunConfig = RunConfig(), callbacks: Optional[Callbacks] = None)

就地递归地对知识图谱应用变换。

源代码位于 src/ragas/testset/transforms/engine.py

def apply_transforms(
    kg: KnowledgeGraph,
    transforms: Transforms,
    run_config: RunConfig = RunConfig(),
    callbacks: t.Optional[Callbacks] = None,
):
    """
    Recursively apply transformations to a knowledge graph in place.
    """
    # apply nest_asyncio to fix the event loop issue in jupyter
    apply_nest_asyncio()

    max_workers = getattr(run_config, "max_workers", -1)

    if isinstance(transforms, t.Sequence):
        for transform in transforms:
            apply_transforms(kg, transform, run_config, callbacks)
    elif isinstance(transforms, Parallel):
        apply_transforms(kg, transforms.transformations, run_config, callbacks)
    elif isinstance(transforms, BaseGraphTransformation):
        logger.debug(
            f"Generating execution plan for transformation {transforms.__class__.__name__}"
        )
        coros = transforms.generate_execution_plan(kg)
        desc = get_desc(transforms)
        run_async_tasks(
            coros,
            batch_size=None,
            show_progress=True,
            progress_bar_desc=desc,
            max_workers=max_workers,
        )
    else:
        raise ValueError(
            f"Invalid transforms type: {type(transforms)}. Expects a sequence of BaseGraphTransformations or a Parallel instance."
        )
    logger.debug("All transformations applied successfully.")

rollback_transforms

rollback_transforms(kg: KnowledgeGraph, transforms: Transforms)

从知识图谱回滚一组变换。

Note

尚未实现。若需要此功能请开 issue。

源代码位于 src/ragas/testset/transforms/engine.py

def rollback_transforms(kg: KnowledgeGraph, transforms: Transforms):
    """
    Rollback a sequence of transformations from a knowledge graph.

    Note
    ----
    This is not yet implemented. Please open an issue if you need this feature.
    """
    # this will allow you to roll back the transformations
    raise NotImplementedError