LLMs

ragbits.core.llms.LLM

LLM(model_name: str, default_options: LLMClientOptionsT | None = None)

Bases: ConfigurableComponent[LLMClientOptionsT], ABC

Abstract class for interaction with Large Language Model.

Constructs a new LLM instance.

PARAMETER	DESCRIPTION
model_name	Name of the model to be used. TYPE: str
default_options	Default options to be used. TYPE: LLMClientOptionsT | None DEFAULT: None

RAISES	DESCRIPTION
TypeError	If the subclass is missing the 'options_cls' attribute.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

def __init__(self, model_name: str, default_options: LLMClientOptionsT | None = None) -> None:
    """
    Constructs a new LLM instance.

    Args:
        model_name: Name of the model to be used.
        default_options: Default options to be used.

    Raises:
        TypeError: If the subclass is missing the 'options_cls' attribute.
    """
    super().__init__(default_options=default_options)
    self.model_name = model_name

default_options instance-attribute

default_options: OptionsT = default_options or options_cls()

options_cls instance-attribute

options_cls: type[LLMClientOptionsT]

default_module class-attribute instance-attribute

default_module: ClassVar = llms

configuration_key class-attribute instance-attribute

configuration_key: ClassVar = 'llm'

model_name instance-attribute

model_name = model_name

subclass_from_config classmethod

subclass_from_config(config: ObjectConstructionConfig) -> Self

Initializes the class with the provided configuration. May return a subclass of the class, if requested by the configuration.

PARAMETER	DESCRIPTION
config	A model containing configuration details for the class. TYPE: ObjectConstructionConfig

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided configuration.

RAISES	DESCRIPTION
InvalidConfigError	The class can't be found or is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def subclass_from_config(cls, config: ObjectConstructionConfig) -> Self:
    """
    Initializes the class with the provided configuration. May return a subclass of the class,
    if requested by the configuration.

    Args:
        config: A model containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.

    Raises:
        InvalidConfigError: The class can't be found or is not a subclass of the current class.
    """
    subclass = import_by_path(config.type, cls.default_module)
    if not issubclass(subclass, cls):
        raise InvalidConfigError(f"{subclass} is not a subclass of {cls}")

    return subclass.from_config(config.config)

subclass_from_factory classmethod

subclass_from_factory(factory_path: str) -> Self

Creates the class using the provided factory function. May return a subclass of the class, if requested by the factory. Supports both synchronous and asynchronous factory functions.

PARAMETER	DESCRIPTION
factory_path	A string representing the path to the factory function in the format of "module.submodule:factory_name". TYPE: str

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided factory function.

RAISES	DESCRIPTION
InvalidConfigError	The factory can't be found or the object returned is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def subclass_from_factory(cls, factory_path: str) -> Self:
    """
    Creates the class using the provided factory function. May return a subclass of the class,
    if requested by the factory. Supports both synchronous and asynchronous factory functions.

    Args:
        factory_path: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".

    Returns:
        An instance of the class initialized with the provided factory function.

    Raises:
        InvalidConfigError: The factory can't be found or the object returned
            is not a subclass of the current class.
    """
    factory = import_by_path(factory_path, cls.default_module)

    if asyncio.iscoroutinefunction(factory):
        try:
            loop = asyncio.get_running_loop()
            obj = asyncio.run_coroutine_threadsafe(factory, loop).result()
        except RuntimeError:
            obj = asyncio.run(factory())
    else:
        obj = factory()

    if not isinstance(obj, cls):
        raise InvalidConfigError(f"The object returned by factory {factory_path} is not an instance of {cls}")

    return obj

preferred_subclass classmethod

preferred_subclass(config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at project's component preferences, either from YAML or from the factory. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
config	The CoreConfig instance containing preferred factory and configuration details. TYPE: CoreConfig
factory_path_override	A string representing the path to the factory function in the format of "module.submodule:factory_name". TYPE: str | None DEFAULT: None
yaml_path_override	A string representing the path to the YAML file containing the Ragstack instance configuration. TYPE: Path | None DEFAULT: None

RAISES	DESCRIPTION
InvalidConfigError	If the default factory or configuration can't be found.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def preferred_subclass(
    cls, config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at project's component preferences, either from YAML
    or from the factory. Takes optional overrides for both, which takes a higher precedence.

    Args:
        config: The CoreConfig instance containing preferred factory and configuration details.
        factory_path_override: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".
        yaml_path_override: A string representing the path to the YAML file containing
            the Ragstack instance configuration.

    Raises:
        InvalidConfigError: If the default factory or configuration can't be found.
    """
    if yaml_path_override:
        preferences = get_config_from_yaml(yaml_path_override)
        if type_config := preferences.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectConstructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if preferred_factory := config.component_preference_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(preferred_factory)

    if preferred_config := config.preferred_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectConstructionConfig.model_validate(preferred_config))

    raise NoPreferredConfigError(f"Could not find preferred factory or configuration for {cls.configuration_key}")

from_config classmethod

from_config(config: dict[str, Any]) -> Self

Initializes the class with the provided configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing configuration details for the class. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided configuration.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def from_config(cls, config: dict[str, Any]) -> Self:
    """
    Initializes the class with the provided configuration.

    Args:
        config: A dictionary containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.
    """
    default_options = config.pop("default_options", None)
    options = cls.options_cls(**default_options) if default_options else None
    return cls(**config, default_options=options)

get_model_id abstractmethod

get_model_id() -> str

Returns the model id.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

@abstractmethod
def get_model_id(self) -> str:
    """
    Returns the model id.
    """

get_estimated_cost abstractmethod

get_estimated_cost(prompt_tokens: int, completion_tokens: int) -> float

Returns the estimated cost of the LLM call.

PARAMETER	DESCRIPTION
prompt_tokens	The number of tokens in the prompt. TYPE: int
completion_tokens	The number of tokens in the completion. TYPE: int

RETURNS	DESCRIPTION
float	The estimated cost of the LLM call.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

@abstractmethod
def get_estimated_cost(self, prompt_tokens: int, completion_tokens: int) -> float:
    """
    Returns the estimated cost of the LLM call.

    Args:
        prompt_tokens: The number of tokens in the prompt.
        completion_tokens: The number of tokens in the completion.

    Returns:
        The estimated cost of the LLM call.
    """

count_tokens

count_tokens(prompt: BasePrompt) -> int

Counts tokens in the prompt.

PARAMETER	DESCRIPTION
prompt	Formatted prompt template with conversation and response parsing configuration. TYPE: BasePrompt

RETURNS	DESCRIPTION
int	Number of tokens in the prompt.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

def count_tokens(self, prompt: BasePrompt) -> int:  # noqa: PLR6301
    """
    Counts tokens in the prompt.

    Args:
        prompt: Formatted prompt template with conversation and response parsing configuration.

    Returns:
        Number of tokens in the prompt.
    """
    return sum(len(message["content"]) for message in prompt.chat)

get_token_id

get_token_id(token: str) -> int

Gets token id.

PARAMETER	DESCRIPTION
token	The token to encode. TYPE: str

RETURNS	DESCRIPTION
int	The id for the given token.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

def get_token_id(self, token: str) -> int:
    """
    Gets token id.

    Args:
        token: The token to encode.

    Returns:
        The id for the given token.
    """
    raise NotImplementedError("Token id lookup is not supported by this model")

generate_raw async

generate_raw(prompt: BasePrompt | str | ChatFormat, *, options: LLMClientOptionsT | None = None) -> dict

Prepares and sends a prompt to the LLM and returns the raw response (without parsing).

PARAMETER	DESCRIPTION
prompt	Can be one of: - BasePrompt instance: Formatted prompt template with conversation - str: Simple text prompt that will be sent as a user message - ChatFormat: List of message dictionaries in OpenAI chat format TYPE: BasePrompt | str | ChatFormat
options	Options to use for the LLM client. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
dict	Raw response from LLM.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

@deprecated("Use generate_with_metadata() instead")
async def generate_raw(
    self,
    prompt: BasePrompt | str | ChatFormat,
    *,
    options: LLMClientOptionsT | None = None,
) -> dict:
    """
    Prepares and sends a prompt to the LLM and returns the raw response (without parsing).

    Args:
        prompt: Can be one of:
            - BasePrompt instance: Formatted prompt template with conversation
            - str: Simple text prompt that will be sent as a user message
            - ChatFormat: List of message dictionaries in OpenAI chat format
        options: Options to use for the LLM client.

    Returns:
        Raw response from LLM.
    """
    merged_options = (self.default_options | options) if options else self.default_options

    if isinstance(prompt, str | list):
        prompt = SimplePrompt(prompt)

    response = (
        await self._call(
            prompt=[prompt],
            options=merged_options,
        )
    )[0]

    returned = {
        "response": response["response"],
        "throughput": response["throughput"],
    }
    for opt in ["tool_calls", "usage"]:
        if opt in response:
            returned[opt] = response[opt]

    return returned

generate async

generate(prompt: str | ChatFormat | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[ChatFormat | str] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]], *, tools: list[Tool] | None = None, options: LLMClientOptionsT | None = None) -> str | PromptOutputT | list[ToolCall] | list[list[ToolCall] | str] | list[str | PromptOutputT | list[ToolCall]]

Prepares and sends a prompt to the LLM and returns the parsed response.

PARAMETER	DESCRIPTION
prompt	Can be one of: - BasePrompt instance: Formatted prompt template with conversation - str: Simple text prompt that will be sent as a user message - ChatFormat: List of message dictionaries in OpenAI chat format - Iterable of any of the above (MutableSequence is only for typing purposes) TYPE: str | ChatFormat | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[ChatFormat | str] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]]
tools	Functions to be used as tools by the LLM. TYPE: list[Tool] | None DEFAULT: None
options	Options to use for the LLM client. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
str | PromptOutputT | list[ToolCall] | list[list[ToolCall] | str] | list[str | PromptOutputT | list[ToolCall]]	Parsed response(s) from LLM or list of tool calls.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

async def generate(
    self,
    prompt: str
    | ChatFormat
    | BasePrompt
    | BasePromptWithParser[PromptOutputT]
    | MutableSequence[ChatFormat | str]
    | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]],
    *,
    tools: list[Tool] | None = None,
    options: LLMClientOptionsT | None = None,
) -> str | PromptOutputT | list[ToolCall] | list[list[ToolCall] | str] | list[str | PromptOutputT | list[ToolCall]]:
    """
    Prepares and sends a prompt to the LLM and returns the parsed response.

    Args:
        prompt: Can be one of:
            - BasePrompt instance: Formatted prompt template with conversation
            - str: Simple text prompt that will be sent as a user message
            - ChatFormat: List of message dictionaries in OpenAI chat format
            - Iterable of any of the above (MutableSequence is only for typing purposes)
        tools: Functions to be used as tools by the LLM.
        options: Options to use for the LLM client.

    Returns:
        Parsed response(s) from LLM or list of tool calls.
    """
    response = await self.generate_with_metadata(prompt, tools=tools, options=options)
    if isinstance(response, list):
        return [r.tool_calls if tools and r.tool_calls else r.content for r in response]
    else:
        return response.tool_calls if tools and response.tool_calls else response.content

generate_with_metadata async

generate_with_metadata(prompt: str | ChatFormat | MutableSequence[str | ChatFormat] | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]], *, tools: list[Tool] | None = None, options: LLMClientOptionsT | None = None) -> LLMResponseWithMetadata[str] | list[LLMResponseWithMetadata[str]] | LLMResponseWithMetadata[PromptOutputT] | list[LLMResponseWithMetadata[PromptOutputT]]

Prepares and sends a prompt to the LLM and returns response parsed to the output type of the prompt (if available).

PARAMETER	DESCRIPTION
prompt	Can be one of: - BasePrompt instance: Formatted prompt template with conversation - str: Simple text prompt that will be sent as a user message - ChatFormat: List of message dictionaries in OpenAI chat format - Iterable of any of the above (MutableSequence is only for typing purposes) TYPE: str | ChatFormat | MutableSequence[str | ChatFormat] | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]]
tools	Functions to be used as tools by the LLM. TYPE: list[Tool] | None DEFAULT: None
options	Options to use for the LLM client. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
LLMResponseWithMetadata[str] | list[LLMResponseWithMetadata[str]] | LLMResponseWithMetadata[PromptOutputT] | list[LLMResponseWithMetadata[PromptOutputT]]	ResponseWithMetadata object(s) with text response, list of tool calls and metadata information.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

async def generate_with_metadata(
    self,
    prompt: str
    | ChatFormat
    | MutableSequence[str | ChatFormat]
    | BasePrompt
    | BasePromptWithParser[PromptOutputT]
    | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]],
    *,
    tools: list[Tool] | None = None,
    options: LLMClientOptionsT | None = None,
) -> (
    LLMResponseWithMetadata[str]
    | list[LLMResponseWithMetadata[str]]
    | LLMResponseWithMetadata[PromptOutputT]
    | list[LLMResponseWithMetadata[PromptOutputT]]
):
    """
    Prepares and sends a prompt to the LLM and returns response parsed to the
    output type of the prompt (if available).

    Args:
        prompt: Can be one of:
            - BasePrompt instance: Formatted prompt template with conversation
            - str: Simple text prompt that will be sent as a user message
            - ChatFormat: List of message dictionaries in OpenAI chat format
            - Iterable of any of the above (MutableSequence is only for typing purposes)
        tools: Functions to be used as tools by the LLM.
        options: Options to use for the LLM client.

    Returns:
        ResponseWithMetadata object(s) with text response, list of tool calls and metadata information.
    """
    single_prompt = False
    if isinstance(prompt, BasePrompt | str) or isinstance(prompt[0], dict):
        single_prompt = True
        prompt = [prompt]  # type: ignore

    parsed_tools = (
        [convert_function_to_function_schema(tool) if callable(tool) else tool for tool in tools] if tools else None
    )

    prompts: list[BasePrompt] = [SimplePrompt(p) if isinstance(p, str | list) else p for p in prompt]  # type: ignore

    merged_options = (self.default_options | options) if options else self.default_options

    with trace(name="generate", model_name=self.model_name, prompt=prompts, options=repr(options)) as outputs:
        results = await self._call(
            prompt=prompts,
            options=merged_options,
            tools=parsed_tools,
        )

        parsed_responses = []
        for prompt, response in zip(prompts, results, strict=True):
            tool_calls = (
                [ToolCall.model_validate(tool_call) for tool_call in _tool_calls]
                if (_tool_calls := response.pop("tool_calls", None)) and tools
                else None
            )

            usage = None
            if usage_data := response.pop("usage", None):
                usage = Usage.new(
                    llm=self,
                    prompt_tokens=cast(int, usage_data.get("prompt_tokens")),
                    completion_tokens=cast(int, usage_data.get("completion_tokens")),
                    total_tokens=cast(int, usage_data.get("total_tokens")),
                )

            content = response.pop("response")
            reasoning = response.pop("reasoning", None)

            if isinstance(prompt, BasePromptWithParser) and content:
                content = await prompt.parse_response(content)

            response_with_metadata = LLMResponseWithMetadata[type(content)](  # type: ignore
                content=content,
                reasoning=reasoning,
                tool_calls=tool_calls,
                metadata=response,
                usage=usage,
            )
            parsed_responses.append(response_with_metadata)
        outputs.response = parsed_responses

        prompt_tokens = sum(r.usage.prompt_tokens for r in parsed_responses if r.usage)
        outputs.prompt_tokens_batch = prompt_tokens
        record_metric(
            metric=LLMMetric.INPUT_TOKENS,
            value=prompt_tokens,
            metric_type=MetricType.HISTOGRAM,
            model=self.model_name,
        )

        total_throughput = sum(r["throughput"] for r in results if "throughput" in r)
        outputs.throughput_batch = total_throughput
        record_metric(
            metric=LLMMetric.PROMPT_THROUGHPUT,
            value=total_throughput,
            metric_type=MetricType.HISTOGRAM,
            model=self.model_name,
        )

        total_tokens = sum(r.usage.total_tokens for r in parsed_responses if r.usage)
        outputs.total_tokens_batch = total_tokens
        record_metric(
            metric=LLMMetric.TOKEN_THROUGHPUT,
            value=total_tokens / total_throughput,
            metric_type=MetricType.HISTOGRAM,
            model=self.model_name,
        )

    if single_prompt:
        return parsed_responses[0]

    return parsed_responses

generate_streaming

generate_streaming(prompt: str | ChatFormat | BasePrompt, *, tools: list[Tool] | None = None, options: LLMClientOptionsT | None = None) -> LLMResultStreaming

This method returns an LLMResultStreaming object that can be asynchronously iterated over. After the loop completes, metadata is available as metadata attribute.

PARAMETER	DESCRIPTION
prompt	Formatted prompt template with conversation. TYPE: str | ChatFormat | BasePrompt
tools	Functions to be used as tools by the LLM. TYPE: list[Tool] | None DEFAULT: None
options	Options to use for the LLM. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
LLMResultStreaming	Response stream from LLM or list of tool calls.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

def generate_streaming(
    self,
    prompt: str | ChatFormat | BasePrompt,
    *,
    tools: list[Tool] | None = None,
    options: LLMClientOptionsT | None = None,
) -> LLMResultStreaming:
    """
    This method returns an `LLMResultStreaming` object that can be asynchronously
    iterated over. After the loop completes, metadata is available as `metadata` attribute.

    Args:
        prompt: Formatted prompt template with conversation.
        tools: Functions to be used as tools by the LLM.
        options: Options to use for the LLM.

    Returns:
        Response stream from LLM or list of tool calls.
    """
    return LLMResultStreaming(self._stream_internal(prompt, tools=tools, options=options))

ragbits.core.llms.local.LocalLLM

LocalLLM(model_name: str, default_options: LocalLLMOptions | None = None, *, api_key: str | None = None, price_per_prompt_token: float = 0.0, price_per_completion_token: float = 0.0)

Bases: LLM[LocalLLMOptions]

Class for interaction with any LLM available in HuggingFace.

Note: Local implementation is not dedicated for production. Use it only in experiments / evaluation

Constructs a new local LLM instance.

PARAMETER	DESCRIPTION
model_name	Name of the model to use. This should be a model from the CausalLM class. TYPE: str
default_options	Default options for the LLM. TYPE: LocalLLMOptions | None DEFAULT: None
api_key	The API key for Hugging Face authentication. TYPE: str | None DEFAULT: None
price_per_prompt_token	The price per prompt token. TYPE: float DEFAULT: 0.0
price_per_completion_token	The price per completion token. TYPE: float DEFAULT: 0.0

RAISES	DESCRIPTION
ImportError	If the 'local' extra requirements are not installed.
ValueError	If the model was not trained as a chat model.

Source code in packages/ragbits-core/src/ragbits/core/llms/local.py

def __init__(
    self,
    model_name: str,
    default_options: LocalLLMOptions | None = None,
    *,
    api_key: str | None = None,
    price_per_prompt_token: float = 0.0,
    price_per_completion_token: float = 0.0,
) -> None:
    """
    Constructs a new local LLM instance.

    Args:
        model_name: Name of the model to use. This should be a model from the CausalLM class.
        default_options: Default options for the LLM.
        api_key: The API key for Hugging Face authentication.
        price_per_prompt_token: The price per prompt token.
        price_per_completion_token: The price per completion token.

    Raises:
        ImportError: If the 'local' extra requirements are not installed.
        ValueError: If the model was not trained as a chat model.
    """
    if not HAS_LOCAL_LLM:
        raise ImportError("You need to install the 'local' extra requirements to use local LLM models")

    super().__init__(model_name, default_options)
    self.model = AutoModelForCausalLM.from_pretrained(
        model_name, device_map="auto", torch_dtype=torch.bfloat16, token=api_key
    )
    self.tokenizer = AutoTokenizer.from_pretrained(model_name, token=api_key)
    try:
        self.tokenizer.get_chat_template()
    except ValueError as e:
        raise ValueError(
            f"{model_name} was not trained as a chat model - it doesn't support chat template. Select another model"
        ) from e
    self.api_key = api_key
    self._price_per_prompt_token = price_per_prompt_token
    self._price_per_completion_token = price_per_completion_token

default_module class-attribute instance-attribute

default_module: ClassVar = llms

configuration_key class-attribute instance-attribute

configuration_key: ClassVar = 'llm'

default_options instance-attribute

default_options: OptionsT = default_options or options_cls()

model_name instance-attribute

model_name = model_name

options_cls class-attribute instance-attribute

options_cls = LocalLLMOptions

model instance-attribute

model = from_pretrained(model_name, device_map='auto', torch_dtype=bfloat16, token=api_key)

tokenizer instance-attribute

tokenizer = from_pretrained(model_name, token=api_key)

api_key instance-attribute

api_key = api_key

subclass_from_config classmethod

subclass_from_config(config: ObjectConstructionConfig) -> Self

Initializes the class with the provided configuration. May return a subclass of the class, if requested by the configuration.

PARAMETER	DESCRIPTION
config	A model containing configuration details for the class. TYPE: ObjectConstructionConfig

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided configuration.

RAISES	DESCRIPTION
InvalidConfigError	The class can't be found or is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def subclass_from_config(cls, config: ObjectConstructionConfig) -> Self:
    """
    Initializes the class with the provided configuration. May return a subclass of the class,
    if requested by the configuration.

    Args:
        config: A model containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.

    Raises:
        InvalidConfigError: The class can't be found or is not a subclass of the current class.
    """
    subclass = import_by_path(config.type, cls.default_module)
    if not issubclass(subclass, cls):
        raise InvalidConfigError(f"{subclass} is not a subclass of {cls}")

    return subclass.from_config(config.config)

subclass_from_factory classmethod

subclass_from_factory(factory_path: str) -> Self

Creates the class using the provided factory function. May return a subclass of the class, if requested by the factory. Supports both synchronous and asynchronous factory functions.

PARAMETER	DESCRIPTION
factory_path	A string representing the path to the factory function in the format of "module.submodule:factory_name". TYPE: str

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided factory function.

RAISES	DESCRIPTION
InvalidConfigError	The factory can't be found or the object returned is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def subclass_from_factory(cls, factory_path: str) -> Self:
    """
    Creates the class using the provided factory function. May return a subclass of the class,
    if requested by the factory. Supports both synchronous and asynchronous factory functions.

    Args:
        factory_path: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".

    Returns:
        An instance of the class initialized with the provided factory function.

    Raises:
        InvalidConfigError: The factory can't be found or the object returned
            is not a subclass of the current class.
    """
    factory = import_by_path(factory_path, cls.default_module)

    if asyncio.iscoroutinefunction(factory):
        try:
            loop = asyncio.get_running_loop()
            obj = asyncio.run_coroutine_threadsafe(factory, loop).result()
        except RuntimeError:
            obj = asyncio.run(factory())
    else:
        obj = factory()

    if not isinstance(obj, cls):
        raise InvalidConfigError(f"The object returned by factory {factory_path} is not an instance of {cls}")

    return obj

preferred_subclass classmethod

preferred_subclass(config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at project's component preferences, either from YAML or from the factory. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
config	The CoreConfig instance containing preferred factory and configuration details. TYPE: CoreConfig
factory_path_override	A string representing the path to the factory function in the format of "module.submodule:factory_name". TYPE: str | None DEFAULT: None
yaml_path_override	A string representing the path to the YAML file containing the Ragstack instance configuration. TYPE: Path | None DEFAULT: None

RAISES	DESCRIPTION
InvalidConfigError	If the default factory or configuration can't be found.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def preferred_subclass(
    cls, config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at project's component preferences, either from YAML
    or from the factory. Takes optional overrides for both, which takes a higher precedence.

    Args:
        config: The CoreConfig instance containing preferred factory and configuration details.
        factory_path_override: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".
        yaml_path_override: A string representing the path to the YAML file containing
            the Ragstack instance configuration.

    Raises:
        InvalidConfigError: If the default factory or configuration can't be found.
    """
    if yaml_path_override:
        preferences = get_config_from_yaml(yaml_path_override)
        if type_config := preferences.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectConstructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if preferred_factory := config.component_preference_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(preferred_factory)

    if preferred_config := config.preferred_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectConstructionConfig.model_validate(preferred_config))

    raise NoPreferredConfigError(f"Could not find preferred factory or configuration for {cls.configuration_key}")

from_config classmethod

from_config(config: dict[str, Any]) -> Self

Initializes the class with the provided configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing configuration details for the class. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided configuration.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def from_config(cls, config: dict[str, Any]) -> Self:
    """
    Initializes the class with the provided configuration.

    Args:
        config: A dictionary containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.
    """
    default_options = config.pop("default_options", None)
    options = cls.options_cls(**default_options) if default_options else None
    return cls(**config, default_options=options)

get_token_id

get_token_id(token: str) -> int

Gets token id.

PARAMETER	DESCRIPTION
token	The token to encode. TYPE: str

RETURNS	DESCRIPTION
int	The id for the given token.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

def get_token_id(self, token: str) -> int:
    """
    Gets token id.

    Args:
        token: The token to encode.

    Returns:
        The id for the given token.
    """
    raise NotImplementedError("Token id lookup is not supported by this model")

generate_raw async

generate_raw(prompt: BasePrompt | str | ChatFormat, *, options: LLMClientOptionsT | None = None) -> dict

Prepares and sends a prompt to the LLM and returns the raw response (without parsing).

PARAMETER	DESCRIPTION
prompt	Can be one of: - BasePrompt instance: Formatted prompt template with conversation - str: Simple text prompt that will be sent as a user message - ChatFormat: List of message dictionaries in OpenAI chat format TYPE: BasePrompt | str | ChatFormat
options	Options to use for the LLM client. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
dict	Raw response from LLM.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

@deprecated("Use generate_with_metadata() instead")
async def generate_raw(
    self,
    prompt: BasePrompt | str | ChatFormat,
    *,
    options: LLMClientOptionsT | None = None,
) -> dict:
    """
    Prepares and sends a prompt to the LLM and returns the raw response (without parsing).

    Args:
        prompt: Can be one of:
            - BasePrompt instance: Formatted prompt template with conversation
            - str: Simple text prompt that will be sent as a user message
            - ChatFormat: List of message dictionaries in OpenAI chat format
        options: Options to use for the LLM client.

    Returns:
        Raw response from LLM.
    """
    merged_options = (self.default_options | options) if options else self.default_options

    if isinstance(prompt, str | list):
        prompt = SimplePrompt(prompt)

    response = (
        await self._call(
            prompt=[prompt],
            options=merged_options,
        )
    )[0]

    returned = {
        "response": response["response"],
        "throughput": response["throughput"],
    }
    for opt in ["tool_calls", "usage"]:
        if opt in response:
            returned[opt] = response[opt]

    return returned

generate async

generate(prompt: str | ChatFormat | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[ChatFormat | str] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]], *, tools: list[Tool] | None = None, options: LLMClientOptionsT | None = None) -> str | PromptOutputT | list[ToolCall] | list[list[ToolCall] | str] | list[str | PromptOutputT | list[ToolCall]]

Prepares and sends a prompt to the LLM and returns the parsed response.

PARAMETER	DESCRIPTION
prompt	Can be one of: - BasePrompt instance: Formatted prompt template with conversation - str: Simple text prompt that will be sent as a user message - ChatFormat: List of message dictionaries in OpenAI chat format - Iterable of any of the above (MutableSequence is only for typing purposes) TYPE: str | ChatFormat | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[ChatFormat | str] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]]
tools	Functions to be used as tools by the LLM. TYPE: list[Tool] | None DEFAULT: None
options	Options to use for the LLM client. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
str | PromptOutputT | list[ToolCall] | list[list[ToolCall] | str] | list[str | PromptOutputT | list[ToolCall]]	Parsed response(s) from LLM or list of tool calls.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

async def generate(
    self,
    prompt: str
    | ChatFormat
    | BasePrompt
    | BasePromptWithParser[PromptOutputT]
    | MutableSequence[ChatFormat | str]
    | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]],
    *,
    tools: list[Tool] | None = None,
    options: LLMClientOptionsT | None = None,
) -> str | PromptOutputT | list[ToolCall] | list[list[ToolCall] | str] | list[str | PromptOutputT | list[ToolCall]]:
    """
    Prepares and sends a prompt to the LLM and returns the parsed response.

    Args:
        prompt: Can be one of:
            - BasePrompt instance: Formatted prompt template with conversation
            - str: Simple text prompt that will be sent as a user message
            - ChatFormat: List of message dictionaries in OpenAI chat format
            - Iterable of any of the above (MutableSequence is only for typing purposes)
        tools: Functions to be used as tools by the LLM.
        options: Options to use for the LLM client.

    Returns:
        Parsed response(s) from LLM or list of tool calls.
    """
    response = await self.generate_with_metadata(prompt, tools=tools, options=options)
    if isinstance(response, list):
        return [r.tool_calls if tools and r.tool_calls else r.content for r in response]
    else:
        return response.tool_calls if tools and response.tool_calls else response.content

generate_with_metadata async

generate_with_metadata(prompt: str | ChatFormat | MutableSequence[str | ChatFormat] | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]], *, tools: list[Tool] | None = None, options: LLMClientOptionsT | None = None) -> LLMResponseWithMetadata[str] | list[LLMResponseWithMetadata[str]] | LLMResponseWithMetadata[PromptOutputT] | list[LLMResponseWithMetadata[PromptOutputT]]

Prepares and sends a prompt to the LLM and returns response parsed to the output type of the prompt (if available).

PARAMETER	DESCRIPTION
prompt	Can be one of: - BasePrompt instance: Formatted prompt template with conversation - str: Simple text prompt that will be sent as a user message - ChatFormat: List of message dictionaries in OpenAI chat format - Iterable of any of the above (MutableSequence is only for typing purposes) TYPE: str | ChatFormat | MutableSequence[str | ChatFormat] | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]]
tools	Functions to be used as tools by the LLM. TYPE: list[Tool] | None DEFAULT: None
options	Options to use for the LLM client. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
LLMResponseWithMetadata[str] | list[LLMResponseWithMetadata[str]] | LLMResponseWithMetadata[PromptOutputT] | list[LLMResponseWithMetadata[PromptOutputT]]	ResponseWithMetadata object(s) with text response, list of tool calls and metadata information.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

async def generate_with_metadata(
    self,
    prompt: str
    | ChatFormat
    | MutableSequence[str | ChatFormat]
    | BasePrompt
    | BasePromptWithParser[PromptOutputT]
    | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]],
    *,
    tools: list[Tool] | None = None,
    options: LLMClientOptionsT | None = None,
) -> (
    LLMResponseWithMetadata[str]
    | list[LLMResponseWithMetadata[str]]
    | LLMResponseWithMetadata[PromptOutputT]
    | list[LLMResponseWithMetadata[PromptOutputT]]
):
    """
    Prepares and sends a prompt to the LLM and returns response parsed to the
    output type of the prompt (if available).

    Args:
        prompt: Can be one of:
            - BasePrompt instance: Formatted prompt template with conversation
            - str: Simple text prompt that will be sent as a user message
            - ChatFormat: List of message dictionaries in OpenAI chat format
            - Iterable of any of the above (MutableSequence is only for typing purposes)
        tools: Functions to be used as tools by the LLM.
        options: Options to use for the LLM client.

    Returns:
        ResponseWithMetadata object(s) with text response, list of tool calls and metadata information.
    """
    single_prompt = False
    if isinstance(prompt, BasePrompt | str) or isinstance(prompt[0], dict):
        single_prompt = True
        prompt = [prompt]  # type: ignore

    parsed_tools = (
        [convert_function_to_function_schema(tool) if callable(tool) else tool for tool in tools] if tools else None
    )

    prompts: list[BasePrompt] = [SimplePrompt(p) if isinstance(p, str | list) else p for p in prompt]  # type: ignore

    merged_options = (self.default_options | options) if options else self.default_options

    with trace(name="generate", model_name=self.model_name, prompt=prompts, options=repr(options)) as outputs:
        results = await self._call(
            prompt=prompts,
            options=merged_options,
            tools=parsed_tools,
        )

        parsed_responses = []
        for prompt, response in zip(prompts, results, strict=True):
            tool_calls = (
                [ToolCall.model_validate(tool_call) for tool_call in _tool_calls]
                if (_tool_calls := response.pop("tool_calls", None)) and tools
                else None
            )

            usage = None
            if usage_data := response.pop("usage", None):
                usage = Usage.new(
                    llm=self,
                    prompt_tokens=cast(int, usage_data.get("prompt_tokens")),
                    completion_tokens=cast(int, usage_data.get("completion_tokens")),
                    total_tokens=cast(int, usage_data.get("total_tokens")),
                )

            content = response.pop("response")
            reasoning = response.pop("reasoning", None)

            if isinstance(prompt, BasePromptWithParser) and content:
                content = await prompt.parse_response(content)

            response_with_metadata = LLMResponseWithMetadata[type(content)](  # type: ignore
                content=content,
                reasoning=reasoning,
                tool_calls=tool_calls,
                metadata=response,
                usage=usage,
            )
            parsed_responses.append(response_with_metadata)
        outputs.response = parsed_responses

        prompt_tokens = sum(r.usage.prompt_tokens for r in parsed_responses if r.usage)
        outputs.prompt_tokens_batch = prompt_tokens
        record_metric(
            metric=LLMMetric.INPUT_TOKENS,
            value=prompt_tokens,
            metric_type=MetricType.HISTOGRAM,
            model=self.model_name,
        )

        total_throughput = sum(r["throughput"] for r in results if "throughput" in r)
        outputs.throughput_batch = total_throughput
        record_metric(
            metric=LLMMetric.PROMPT_THROUGHPUT,
            value=total_throughput,
            metric_type=MetricType.HISTOGRAM,
            model=self.model_name,
        )

        total_tokens = sum(r.usage.total_tokens for r in parsed_responses if r.usage)
        outputs.total_tokens_batch = total_tokens
        record_metric(
            metric=LLMMetric.TOKEN_THROUGHPUT,
            value=total_tokens / total_throughput,
            metric_type=MetricType.HISTOGRAM,
            model=self.model_name,
        )

    if single_prompt:
        return parsed_responses[0]

    return parsed_responses

generate_streaming

generate_streaming(prompt: str | ChatFormat | BasePrompt, *, tools: list[Tool] | None = None, options: LLMClientOptionsT | None = None) -> LLMResultStreaming

This method returns an LLMResultStreaming object that can be asynchronously iterated over. After the loop completes, metadata is available as metadata attribute.

PARAMETER	DESCRIPTION
prompt	Formatted prompt template with conversation. TYPE: str | ChatFormat | BasePrompt
tools	Functions to be used as tools by the LLM. TYPE: list[Tool] | None DEFAULT: None
options	Options to use for the LLM. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
LLMResultStreaming	Response stream from LLM or list of tool calls.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

def generate_streaming(
    self,
    prompt: str | ChatFormat | BasePrompt,
    *,
    tools: list[Tool] | None = None,
    options: LLMClientOptionsT | None = None,
) -> LLMResultStreaming:
    """
    This method returns an `LLMResultStreaming` object that can be asynchronously
    iterated over. After the loop completes, metadata is available as `metadata` attribute.

    Args:
        prompt: Formatted prompt template with conversation.
        tools: Functions to be used as tools by the LLM.
        options: Options to use for the LLM.

    Returns:
        Response stream from LLM or list of tool calls.
    """
    return LLMResultStreaming(self._stream_internal(prompt, tools=tools, options=options))

get_model_id

get_model_id() -> str

Returns the model id.

Source code in packages/ragbits-core/src/ragbits/core/llms/local.py

def get_model_id(self) -> str:
    """
    Returns the model id.
    """
    return "local:" + self.model_name

get_estimated_cost

get_estimated_cost(prompt_tokens: int, completion_tokens: int) -> float

Returns the estimated cost of the LLM call.

Source code in packages/ragbits-core/src/ragbits/core/llms/local.py

def get_estimated_cost(self, prompt_tokens: int, completion_tokens: int) -> float:
    """
    Returns the estimated cost of the LLM call.
    """
    return self._price_per_prompt_token * prompt_tokens + self._price_per_completion_token * completion_tokens

count_tokens

count_tokens(prompt: BasePrompt) -> int

Counts tokens in the messages.

PARAMETER	DESCRIPTION
prompt	Messages to count tokens for. TYPE: BasePrompt

RETURNS	DESCRIPTION
int	Number of tokens in the messages.

Source code in packages/ragbits-core/src/ragbits/core/llms/local.py

def count_tokens(self, prompt: BasePrompt) -> int:
    """
    Counts tokens in the messages.

    Args:
        prompt: Messages to count tokens for.

    Returns:
        Number of tokens in the messages.
    """
    input_ids = self.tokenizer.apply_chat_template(prompt.chat)
    return len(input_ids)

ragbits.core.llms.litellm.LiteLLM

LiteLLM(model_name: str = 'gpt-3.5-turbo', default_options: LiteLLMOptions | None = None, *, api_base: str | None = None, base_url: str | None = None, api_key: str | None = None, api_version: str | None = None, use_structured_output: bool = False, router: Router | None = None, custom_model_cost_config: dict | None = None)

Bases: LLM[LiteLLMOptions]

Class for interaction with any LLM supported by LiteLLM API.

Constructs a new LiteLLM instance.

PARAMETER	DESCRIPTION
model_name	Name of the LiteLLM supported model to be used. Default is "gpt-3.5-turbo". TYPE: str DEFAULT: 'gpt-3.5-turbo'
default_options	Default options to be used. TYPE: LiteLLMOptions | None DEFAULT: None
api_base	Base URL of the LLM API. TYPE: str | None DEFAULT: None
base_url	Alias for api_base. If both are provided, api_base takes precedence. TYPE: str | None DEFAULT: None
api_key	API key to be used. API key to be used. If not specified, an environment variable will be used, for more information, follow the instructions for your specific vendor in the LiteLLM documentation. TYPE: str | None DEFAULT: None
api_version	API version to be used. If not specified, the default version will be used. TYPE: str | None DEFAULT: None
use_structured_output	Whether to request a structured output from the model. Default is False. Can only be combined with models that support structured output. TYPE: bool DEFAULT: False
router	Router to be used to route requests to different models. TYPE: Router | None DEFAULT: None
custom_model_cost_config	Custom cost and capabilities configuration for the model. Necessary for custom model cost and capabilities tracking in LiteLLM. See the LiteLLM documentation for more information. TYPE: dict | None DEFAULT: None

Source code in packages/ragbits-core/src/ragbits/core/llms/litellm.py

def __init__(
    self,
    model_name: str = "gpt-3.5-turbo",
    default_options: LiteLLMOptions | None = None,
    *,
    api_base: str | None = None,
    base_url: str | None = None,  # Alias for api_base
    api_key: str | None = None,
    api_version: str | None = None,
    use_structured_output: bool = False,
    router: litellm.Router | None = None,
    custom_model_cost_config: dict | None = None,
) -> None:
    """
    Constructs a new LiteLLM instance.

    Args:
        model_name: Name of the [LiteLLM supported model](https://docs.litellm.ai/docs/providers) to be used.\
            Default is "gpt-3.5-turbo".
        default_options: Default options to be used.
        api_base: Base URL of the LLM API.
        base_url: Alias for api_base. If both are provided, api_base takes precedence.
        api_key: API key to be used. API key to be used. If not specified, an environment variable will be used,
            for more information, follow the instructions for your specific vendor in the\
            [LiteLLM documentation](https://docs.litellm.ai/docs/providers).
        api_version: API version to be used. If not specified, the default version will be used.
        use_structured_output: Whether to request a
            [structured output](https://docs.litellm.ai/docs/completion/json_mode#pass-in-json_schema)
            from the model. Default is False. Can only be combined with models that support structured output.
        router: Router to be used to [route requests](https://docs.litellm.ai/docs/routing) to different models.
        custom_model_cost_config: Custom cost and capabilities configuration for the model.
            Necessary for custom model cost and capabilities tracking in LiteLLM.
            See the [LiteLLM documentation](https://docs.litellm.ai/docs/completion/token_usage#9-register_model)
            for more information.
    """
    super().__init__(model_name, default_options)
    self.api_base = api_base or base_url
    self.api_key = api_key
    self.api_version = api_version
    self.use_structured_output = use_structured_output
    self.router = router
    self.custom_model_cost_config = custom_model_cost_config
    if custom_model_cost_config:
        litellm.register_model(custom_model_cost_config)

default_module class-attribute instance-attribute

default_module: ClassVar = llms

configuration_key class-attribute instance-attribute

configuration_key: ClassVar = 'llm'

default_options instance-attribute

default_options: OptionsT = default_options or options_cls()

model_name instance-attribute

model_name = model_name

options_cls class-attribute instance-attribute

options_cls = LiteLLMOptions

api_base instance-attribute

api_base = api_base or base_url

api_key instance-attribute

api_key = api_key

api_version instance-attribute

api_version = api_version

use_structured_output instance-attribute

use_structured_output = use_structured_output

router instance-attribute

router = router

custom_model_cost_config instance-attribute

custom_model_cost_config = custom_model_cost_config

base_url property

base_url: str | None

Returns the base URL of the LLM API. Alias for api_base.

subclass_from_config classmethod

subclass_from_config(config: ObjectConstructionConfig) -> Self

Initializes the class with the provided configuration. May return a subclass of the class, if requested by the configuration.

PARAMETER	DESCRIPTION
config	A model containing configuration details for the class. TYPE: ObjectConstructionConfig

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided configuration.

RAISES	DESCRIPTION
InvalidConfigError	The class can't be found or is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def subclass_from_config(cls, config: ObjectConstructionConfig) -> Self:
    """
    Initializes the class with the provided configuration. May return a subclass of the class,
    if requested by the configuration.

    Args:
        config: A model containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.

    Raises:
        InvalidConfigError: The class can't be found or is not a subclass of the current class.
    """
    subclass = import_by_path(config.type, cls.default_module)
    if not issubclass(subclass, cls):
        raise InvalidConfigError(f"{subclass} is not a subclass of {cls}")

    return subclass.from_config(config.config)

subclass_from_factory classmethod

subclass_from_factory(factory_path: str) -> Self

Creates the class using the provided factory function. May return a subclass of the class, if requested by the factory. Supports both synchronous and asynchronous factory functions.

PARAMETER	DESCRIPTION
factory_path	A string representing the path to the factory function in the format of "module.submodule:factory_name". TYPE: str

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided factory function.

RAISES	DESCRIPTION
InvalidConfigError	The factory can't be found or the object returned is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def subclass_from_factory(cls, factory_path: str) -> Self:
    """
    Creates the class using the provided factory function. May return a subclass of the class,
    if requested by the factory. Supports both synchronous and asynchronous factory functions.

    Args:
        factory_path: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".

    Returns:
        An instance of the class initialized with the provided factory function.

    Raises:
        InvalidConfigError: The factory can't be found or the object returned
            is not a subclass of the current class.
    """
    factory = import_by_path(factory_path, cls.default_module)

    if asyncio.iscoroutinefunction(factory):
        try:
            loop = asyncio.get_running_loop()
            obj = asyncio.run_coroutine_threadsafe(factory, loop).result()
        except RuntimeError:
            obj = asyncio.run(factory())
    else:
        obj = factory()

    if not isinstance(obj, cls):
        raise InvalidConfigError(f"The object returned by factory {factory_path} is not an instance of {cls}")

    return obj

preferred_subclass classmethod

preferred_subclass(config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at project's component preferences, either from YAML or from the factory. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
config	The CoreConfig instance containing preferred factory and configuration details. TYPE: CoreConfig
factory_path_override	A string representing the path to the factory function in the format of "module.submodule:factory_name". TYPE: str | None DEFAULT: None
yaml_path_override	A string representing the path to the YAML file containing the Ragstack instance configuration. TYPE: Path | None DEFAULT: None

RAISES	DESCRIPTION
InvalidConfigError	If the default factory or configuration can't be found.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def preferred_subclass(
    cls, config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at project's component preferences, either from YAML
    or from the factory. Takes optional overrides for both, which takes a higher precedence.

    Args:
        config: The CoreConfig instance containing preferred factory and configuration details.
        factory_path_override: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".
        yaml_path_override: A string representing the path to the YAML file containing
            the Ragstack instance configuration.

    Raises:
        InvalidConfigError: If the default factory or configuration can't be found.
    """
    if yaml_path_override:
        preferences = get_config_from_yaml(yaml_path_override)
        if type_config := preferences.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectConstructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if preferred_factory := config.component_preference_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(preferred_factory)

    if preferred_config := config.preferred_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectConstructionConfig.model_validate(preferred_config))

    raise NoPreferredConfigError(f"Could not find preferred factory or configuration for {cls.configuration_key}")

generate_raw async

generate_raw(prompt: BasePrompt | str | ChatFormat, *, options: LLMClientOptionsT | None = None) -> dict

Prepares and sends a prompt to the LLM and returns the raw response (without parsing).

PARAMETER	DESCRIPTION
prompt	Can be one of: - BasePrompt instance: Formatted prompt template with conversation - str: Simple text prompt that will be sent as a user message - ChatFormat: List of message dictionaries in OpenAI chat format TYPE: BasePrompt | str | ChatFormat
options	Options to use for the LLM client. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
dict	Raw response from LLM.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

@deprecated("Use generate_with_metadata() instead")
async def generate_raw(
    self,
    prompt: BasePrompt | str | ChatFormat,
    *,
    options: LLMClientOptionsT | None = None,
) -> dict:
    """
    Prepares and sends a prompt to the LLM and returns the raw response (without parsing).

    Args:
        prompt: Can be one of:
            - BasePrompt instance: Formatted prompt template with conversation
            - str: Simple text prompt that will be sent as a user message
            - ChatFormat: List of message dictionaries in OpenAI chat format
        options: Options to use for the LLM client.

    Returns:
        Raw response from LLM.
    """
    merged_options = (self.default_options | options) if options else self.default_options

    if isinstance(prompt, str | list):
        prompt = SimplePrompt(prompt)

    response = (
        await self._call(
            prompt=[prompt],
            options=merged_options,
        )
    )[0]

    returned = {
        "response": response["response"],
        "throughput": response["throughput"],
    }
    for opt in ["tool_calls", "usage"]:
        if opt in response:
            returned[opt] = response[opt]

    return returned

generate async

generate(prompt: str | ChatFormat | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[ChatFormat | str] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]], *, tools: list[Tool] | None = None, options: LLMClientOptionsT | None = None) -> str | PromptOutputT | list[ToolCall] | list[list[ToolCall] | str] | list[str | PromptOutputT | list[ToolCall]]

Prepares and sends a prompt to the LLM and returns the parsed response.

PARAMETER	DESCRIPTION
prompt	Can be one of: - BasePrompt instance: Formatted prompt template with conversation - str: Simple text prompt that will be sent as a user message - ChatFormat: List of message dictionaries in OpenAI chat format - Iterable of any of the above (MutableSequence is only for typing purposes) TYPE: str | ChatFormat | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[ChatFormat | str] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]]
tools	Functions to be used as tools by the LLM. TYPE: list[Tool] | None DEFAULT: None
options	Options to use for the LLM client. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
str | PromptOutputT | list[ToolCall] | list[list[ToolCall] | str] | list[str | PromptOutputT | list[ToolCall]]	Parsed response(s) from LLM or list of tool calls.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

async def generate(
    self,
    prompt: str
    | ChatFormat
    | BasePrompt
    | BasePromptWithParser[PromptOutputT]
    | MutableSequence[ChatFormat | str]
    | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]],
    *,
    tools: list[Tool] | None = None,
    options: LLMClientOptionsT | None = None,
) -> str | PromptOutputT | list[ToolCall] | list[list[ToolCall] | str] | list[str | PromptOutputT | list[ToolCall]]:
    """
    Prepares and sends a prompt to the LLM and returns the parsed response.

    Args:
        prompt: Can be one of:
            - BasePrompt instance: Formatted prompt template with conversation
            - str: Simple text prompt that will be sent as a user message
            - ChatFormat: List of message dictionaries in OpenAI chat format
            - Iterable of any of the above (MutableSequence is only for typing purposes)
        tools: Functions to be used as tools by the LLM.
        options: Options to use for the LLM client.

    Returns:
        Parsed response(s) from LLM or list of tool calls.
    """
    response = await self.generate_with_metadata(prompt, tools=tools, options=options)
    if isinstance(response, list):
        return [r.tool_calls if tools and r.tool_calls else r.content for r in response]
    else:
        return response.tool_calls if tools and response.tool_calls else response.content

generate_with_metadata async

generate_with_metadata(prompt: str | ChatFormat | MutableSequence[str | ChatFormat] | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]], *, tools: list[Tool] | None = None, options: LLMClientOptionsT | None = None) -> LLMResponseWithMetadata[str] | list[LLMResponseWithMetadata[str]] | LLMResponseWithMetadata[PromptOutputT] | list[LLMResponseWithMetadata[PromptOutputT]]

Prepares and sends a prompt to the LLM and returns response parsed to the output type of the prompt (if available).

PARAMETER	DESCRIPTION
prompt	Can be one of: - BasePrompt instance: Formatted prompt template with conversation - str: Simple text prompt that will be sent as a user message - ChatFormat: List of message dictionaries in OpenAI chat format - Iterable of any of the above (MutableSequence is only for typing purposes) TYPE: str | ChatFormat | MutableSequence[str | ChatFormat] | BasePrompt | BasePromptWithParser[PromptOutputT] | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]]
tools	Functions to be used as tools by the LLM. TYPE: list[Tool] | None DEFAULT: None
options	Options to use for the LLM client. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
LLMResponseWithMetadata[str] | list[LLMResponseWithMetadata[str]] | LLMResponseWithMetadata[PromptOutputT] | list[LLMResponseWithMetadata[PromptOutputT]]	ResponseWithMetadata object(s) with text response, list of tool calls and metadata information.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

async def generate_with_metadata(
    self,
    prompt: str
    | ChatFormat
    | MutableSequence[str | ChatFormat]
    | BasePrompt
    | BasePromptWithParser[PromptOutputT]
    | MutableSequence[BasePrompt | BasePromptWithParser[PromptOutputT]],
    *,
    tools: list[Tool] | None = None,
    options: LLMClientOptionsT | None = None,
) -> (
    LLMResponseWithMetadata[str]
    | list[LLMResponseWithMetadata[str]]
    | LLMResponseWithMetadata[PromptOutputT]
    | list[LLMResponseWithMetadata[PromptOutputT]]
):
    """
    Prepares and sends a prompt to the LLM and returns response parsed to the
    output type of the prompt (if available).

    Args:
        prompt: Can be one of:
            - BasePrompt instance: Formatted prompt template with conversation
            - str: Simple text prompt that will be sent as a user message
            - ChatFormat: List of message dictionaries in OpenAI chat format
            - Iterable of any of the above (MutableSequence is only for typing purposes)
        tools: Functions to be used as tools by the LLM.
        options: Options to use for the LLM client.

    Returns:
        ResponseWithMetadata object(s) with text response, list of tool calls and metadata information.
    """
    single_prompt = False
    if isinstance(prompt, BasePrompt | str) or isinstance(prompt[0], dict):
        single_prompt = True
        prompt = [prompt]  # type: ignore

    parsed_tools = (
        [convert_function_to_function_schema(tool) if callable(tool) else tool for tool in tools] if tools else None
    )

    prompts: list[BasePrompt] = [SimplePrompt(p) if isinstance(p, str | list) else p for p in prompt]  # type: ignore

    merged_options = (self.default_options | options) if options else self.default_options

    with trace(name="generate", model_name=self.model_name, prompt=prompts, options=repr(options)) as outputs:
        results = await self._call(
            prompt=prompts,
            options=merged_options,
            tools=parsed_tools,
        )

        parsed_responses = []
        for prompt, response in zip(prompts, results, strict=True):
            tool_calls = (
                [ToolCall.model_validate(tool_call) for tool_call in _tool_calls]
                if (_tool_calls := response.pop("tool_calls", None)) and tools
                else None
            )

            usage = None
            if usage_data := response.pop("usage", None):
                usage = Usage.new(
                    llm=self,
                    prompt_tokens=cast(int, usage_data.get("prompt_tokens")),
                    completion_tokens=cast(int, usage_data.get("completion_tokens")),
                    total_tokens=cast(int, usage_data.get("total_tokens")),
                )

            content = response.pop("response")
            reasoning = response.pop("reasoning", None)

            if isinstance(prompt, BasePromptWithParser) and content:
                content = await prompt.parse_response(content)

            response_with_metadata = LLMResponseWithMetadata[type(content)](  # type: ignore
                content=content,
                reasoning=reasoning,
                tool_calls=tool_calls,
                metadata=response,
                usage=usage,
            )
            parsed_responses.append(response_with_metadata)
        outputs.response = parsed_responses

        prompt_tokens = sum(r.usage.prompt_tokens for r in parsed_responses if r.usage)
        outputs.prompt_tokens_batch = prompt_tokens
        record_metric(
            metric=LLMMetric.INPUT_TOKENS,
            value=prompt_tokens,
            metric_type=MetricType.HISTOGRAM,
            model=self.model_name,
        )

        total_throughput = sum(r["throughput"] for r in results if "throughput" in r)
        outputs.throughput_batch = total_throughput
        record_metric(
            metric=LLMMetric.PROMPT_THROUGHPUT,
            value=total_throughput,
            metric_type=MetricType.HISTOGRAM,
            model=self.model_name,
        )

        total_tokens = sum(r.usage.total_tokens for r in parsed_responses if r.usage)
        outputs.total_tokens_batch = total_tokens
        record_metric(
            metric=LLMMetric.TOKEN_THROUGHPUT,
            value=total_tokens / total_throughput,
            metric_type=MetricType.HISTOGRAM,
            model=self.model_name,
        )

    if single_prompt:
        return parsed_responses[0]

    return parsed_responses

generate_streaming

generate_streaming(prompt: str | ChatFormat | BasePrompt, *, tools: list[Tool] | None = None, options: LLMClientOptionsT | None = None) -> LLMResultStreaming

This method returns an LLMResultStreaming object that can be asynchronously iterated over. After the loop completes, metadata is available as metadata attribute.

PARAMETER	DESCRIPTION
prompt	Formatted prompt template with conversation. TYPE: str | ChatFormat | BasePrompt
tools	Functions to be used as tools by the LLM. TYPE: list[Tool] | None DEFAULT: None
options	Options to use for the LLM. TYPE: LLMClientOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
LLMResultStreaming	Response stream from LLM or list of tool calls.

Source code in packages/ragbits-core/src/ragbits/core/llms/base.py

def generate_streaming(
    self,
    prompt: str | ChatFormat | BasePrompt,
    *,
    tools: list[Tool] | None = None,
    options: LLMClientOptionsT | None = None,
) -> LLMResultStreaming:
    """
    This method returns an `LLMResultStreaming` object that can be asynchronously
    iterated over. After the loop completes, metadata is available as `metadata` attribute.

    Args:
        prompt: Formatted prompt template with conversation.
        tools: Functions to be used as tools by the LLM.
        options: Options to use for the LLM.

    Returns:
        Response stream from LLM or list of tool calls.
    """
    return LLMResultStreaming(self._stream_internal(prompt, tools=tools, options=options))

get_model_id

get_model_id() -> str

Returns the model id.

Source code in packages/ragbits-core/src/ragbits/core/llms/litellm.py

def get_model_id(self) -> str:
    """
    Returns the model id.
    """
    return "litellm:" + self.model_name

get_estimated_cost

get_estimated_cost(prompt_tokens: int, completion_tokens: int) -> float

Returns the estimated cost of the LLM call.

PARAMETER	DESCRIPTION
prompt_tokens	The number of tokens in the prompt. TYPE: int
completion_tokens	The number of tokens in the completion. TYPE: int

RETURNS	DESCRIPTION
float	The estimated cost of the LLM call.

Source code in packages/ragbits-core/src/ragbits/core/llms/litellm.py

def get_estimated_cost(self, prompt_tokens: int, completion_tokens: int) -> float:
    """
    Returns the estimated cost of the LLM call.

    Args:
        prompt_tokens: The number of tokens in the prompt.
        completion_tokens: The number of tokens in the completion.

    Returns:
        The estimated cost of the LLM call.
    """
    response_cost = litellm.get_model_info(self.model_name)
    response_cost_input = prompt_tokens * response_cost["input_cost_per_token"]
    response_cost_output = completion_tokens * response_cost["output_cost_per_token"]
    return response_cost_input + response_cost_output

count_tokens

count_tokens(prompt: BasePrompt) -> int

Counts tokens in the prompt.

PARAMETER	DESCRIPTION
prompt	Formatted prompt template with conversation and response parsing configuration. TYPE: BasePrompt

RETURNS	DESCRIPTION
int	Number of tokens in the prompt.

Source code in packages/ragbits-core/src/ragbits/core/llms/litellm.py

def count_tokens(self, prompt: BasePrompt) -> int:
    """
    Counts tokens in the prompt.

    Args:
        prompt: Formatted prompt template with conversation and response parsing configuration.

    Returns:
        Number of tokens in the prompt.
    """
    return sum(
        litellm.token_counter(model=self.model_name, text=message.get("content") or "") for message in prompt.chat
    )

get_token_id

get_token_id(token: str) -> int

Gets token id.

PARAMETER	DESCRIPTION
token	The token to encode. TYPE: str

RETURNS	DESCRIPTION
int	The id for the given token.

Source code in packages/ragbits-core/src/ragbits/core/llms/litellm.py

def get_token_id(self, token: str) -> int:
    """
    Gets token id.

    Args:
        token: The token to encode.

    Returns:
        The id for the given token.
    """
    try:
        tokenizer = tiktoken.encoding_for_model(self.model_name)
        return tokenizer.encode_single_token(token)
    except KeyError:
        return litellm.encode(model=self.model_name, text=token)[0]

from_config classmethod

from_config(config: dict[str, Any]) -> Self

Creates and returns a LiteLLM instance.

PARAMETER	DESCRIPTION
config	A configuration object containing the configuration for initializing the LiteLLM instance. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
LiteLLM	An initialized LiteLLM instance. TYPE: Self

Source code in packages/ragbits-core/src/ragbits/core/llms/litellm.py

@classmethod
def from_config(cls, config: dict[str, Any]) -> Self:
    """
    Creates and returns a LiteLLM instance.

    Args:
        config: A configuration object containing the configuration for initializing the LiteLLM instance.

    Returns:
        LiteLLM: An initialized LiteLLM instance.
    """
    if "router" in config:
        router = litellm.Router(model_list=config["router"])
        config["router"] = router

    # Map base_url to api_base if present
    if "base_url" in config and "api_base" not in config:
        config["api_base"] = config.pop("base_url")

    return super().from_config(config)