Synthesizers

BaseSynthesizer dataclass

BaseSynthesizer(name: str = '', llm: Union[BaseRagasLLM, 'InstructorBaseRagasLLM'] = _default_llm_factory(), llm_context: Optional[str] = None)

Bases: ABC, Generic[Scenario], PromptMixin

Base class for synthesizing scenarios and samples.

default_query_distribution

default_query_distribution(llm: Union[BaseRagasLLM, InstructorBaseRagasLLM], kg: Optional[KnowledgeGraph] = None, llm_context: Optional[str] = None) -> QueryDistribution

Source code in src/ragas/testset/synthesizers/__init__.py

def default_query_distribution(
    llm: t.Union[BaseRagasLLM, "InstructorBaseRagasLLM"],
    kg: t.Optional[KnowledgeGraph] = None,
    llm_context: t.Optional[str] = None,
) -> QueryDistribution:
    """ """
    default_queries = [
        SingleHopSpecificQuerySynthesizer(llm=llm, llm_context=llm_context),
        MultiHopAbstractQuerySynthesizer(llm=llm, llm_context=llm_context),
        MultiHopSpecificQuerySynthesizer(llm=llm, llm_context=llm_context),
    ]
    if kg is not None:
        available_queries = []
        for query in default_queries:
            try:
                if query.get_node_clusters(kg):
                    available_queries.append(query)
            except Exception as e:
                # Keep broad catch minimal for resilience; log and skip.
                logger.warning(
                    "Skipping %s due to unexpected error: %s",
                    getattr(query, "name", type(query).__name__),
                    e,
                )
                continue
        if not available_queries:
            raise ValueError(
                "No compatible query synthesizers for the provided KnowledgeGraph."
            )
    else:
        available_queries = default_queries

    return [(query, 1 / len(available_queries)) for query in available_queries]