Standalone Message Compressor

A compressor that uses LLM to recontextualize the last message in the history, i.e. create a standalone version of the message that includes necessary context.

ragbits.conversations.history.compressors.llm.StandaloneMessageCompressor

StandaloneMessageCompressor(llm: LLM, history_len: int = 5, prompt: type[Prompt[LastMessageAndHistory, str]] | None = None)

Bases: ConversationHistoryCompressor

A compressor that uses LLM to recontextualize the last message in the history, i.e. create a standalone version of the message that includes necessary context.

Initialize the StandaloneMessageCompressor compressor with a LLM.

PARAMETER	DESCRIPTION
llm	A LLM instance to handle recontextualizing the last message. TYPE: LLM
history_len	The number of previous messages to include in the history. TYPE: int DEFAULT: 5
prompt	The prompt to use for recontextualizing the last message. TYPE: type[Prompt[LastMessageAndHistory, str]] | None DEFAULT: None

Source code in packages/ragbits-conversations/src/ragbits/conversations/history/compressors/llm.py

def __init__(self, llm: LLM, history_len: int = 5, prompt: type[Prompt[LastMessageAndHistory, str]] | None = None):
    """
    Initialize the StandaloneMessageCompressor compressor with a LLM.

    Args:
        llm: A LLM instance to handle recontextualizing the last message.
        history_len: The number of previous messages to include in the history.
        prompt: The prompt to use for recontextualizing the last message.
    """
    self._llm = llm
    self._history_len = history_len
    self._prompt = prompt or StandaloneMessageCompressorPrompt

default_module class-attribute instance-attribute

default_module: ClassVar = compressors

configuration_key class-attribute instance-attribute

configuration_key: ClassVar = 'history_compressor'

subclass_from_config classmethod

subclass_from_config(config: ObjectContructionConfig) -> 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: ObjectContructionConfig

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: ObjectContructionConfig) -> 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.

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.

    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)
    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

subclass_from_defaults classmethod

subclass_from_defaults(defaults: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at default configuration file, and default factory function. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
defaults	The CoreConfig instance containing default 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 subclass_from_defaults(
    cls, defaults: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at default configuration file, and default factory function.
    Takes optional overrides for both, which takes a higher precedence.

    Args:
        defaults: The CoreConfig instance containing default 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:
        config = get_config_from_yaml(yaml_path_override)
        if type_config := config.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectContructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if default_factory := defaults.default_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(default_factory)

    if default_config := defaults.default_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectContructionConfig.model_validate(default_config))

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

from_config classmethod

from_config(config: dict) -> Self

Initializes the class with the provided configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing configuration details for the class. TYPE: dict

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) -> 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.
    """
    return cls(**config)

compress async

compress(conversation: ChatFormat) -> str

Contextualize the last message in the conversation history.

PARAMETER	DESCRIPTION
conversation	List of dicts with "role" and "content" keys, representing the chat history so far. The most recent message should be from the user. TYPE: ChatFormat

Source code in packages/ragbits-conversations/src/ragbits/conversations/history/compressors/llm.py

async def compress(self, conversation: ChatFormat) -> str:
    """
    Contextualize the last message in the conversation history.

    Args:
        conversation: List of dicts with "role" and "content" keys, representing the chat history so far.
            The most recent message should be from the user.
    """
    if len(conversation) == 0:
        raise ValueError("Conversation history is empty.")

    last_message = conversation[-1]
    if last_message["role"] != "user":
        raise ValueError("StandaloneMessageCompressor expects the last message to be from the user.")

    # Only include "user" and "assistant" messages in the history
    other_messages = [message for message in conversation[:-1] if message["role"] in ["user", "assistant"]]

    if not other_messages:
        # No history to use for recontextualization, simply return the user message
        return last_message["content"]

    history = [f"{message['role']}: {message['content']}" for message in other_messages[-self._history_len :]]

    input_data = LastMessageAndHistory(last_message=last_message["content"], history=history)
    prompt = self._prompt(input_data)
    response = await self._llm.generate(prompt)
    return response

ragbits.conversations.history.compressors.llm.LastMessageAndHistory

Bases: BaseModel

A class representing the last message and the history of messages.

last_message instance-attribute

last_message: str

history instance-attribute

history: list[str]

ragbits.conversations.history.compressors.llm.StandaloneMessageCompressorPrompt

StandaloneMessageCompressorPrompt(*args: Any, **kwargs: Any)

Bases: Prompt[LastMessageAndHistory, str]

A prompt for recontextualizing the last message in the history.

Source code in packages/ragbits-core/src/ragbits/core/prompt/prompt.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    input_data = args[0] if args else kwargs.get("input_data")
    if self.input_type and input_data is None:
        raise ValueError("Input data must be provided")

    self.rendered_system_prompt = (
        self._render_template(self.system_prompt_template, input_data) if self.system_prompt_template else None
    )
    self.rendered_user_prompt = self._render_template(self.user_prompt_template, input_data)
    self.images = self._get_images_from_input_data(input_data)

    # Additional few shot examples that can be added dynamically using methods
    # (in opposite to the static `few_shots` attribute which is defined in the class)
    self._instace_few_shots: list[FewShotExample[InputT, OutputT]] = []
    super().__init__()

chat property

chat: ChatFormat

Returns the conversation in the standard OpenAI chat format.

RETURNS	DESCRIPTION
ChatFormat	A list of dictionaries, each containing the role and content of a message. TYPE: ChatFormat

json_mode property

json_mode: bool

Returns whether the prompt should be sent in JSON mode.

RETURNS	DESCRIPTION
bool	Whether the prompt should be sent in JSON mode. TYPE: bool

few_shots class-attribute instance-attribute

few_shots: list[FewShotExample[InputT, OutputT]] = []

response_parser instance-attribute

response_parser: Callable[[str], OutputT]

input_type instance-attribute

input_type: type[InputT] | None

output_type instance-attribute

output_type: type[OutputT]

system_prompt_template instance-attribute

system_prompt_template: Template | None

user_prompt_template instance-attribute

user_prompt_template: Template

image_input_fields class-attribute instance-attribute

image_input_fields: list[str] | None = None

rendered_system_prompt instance-attribute

rendered_system_prompt = _render_template(system_prompt_template, input_data) if system_prompt_template else None

rendered_user_prompt instance-attribute

rendered_user_prompt = _render_template(user_prompt_template, input_data)

images instance-attribute

images = _get_images_from_input_data(input_data)

system_prompt class-attribute instance-attribute

system_prompt = '\n    Given a new message and a history of the conversation, create a standalone version of the message.\n    If the message references any context from history, it should be added to the message itself.\n    Return only the recontextualized message.\n    Do NOT return the history, do NOT answer the question, and do NOT add context irrelevant to the message.\n    '

user_prompt class-attribute instance-attribute

user_prompt = '\n    Message:\n    {{ last_message }}\n\n    History:\n    {% for message in history %}\n    * {{ message }}\n    {% endfor %}\n    '

output_schema

output_schema() -> dict | type[BaseModel] | None

Returns the schema of the desired output. Can be used to request structured output from the LLM API or to validate the output. Can return either a Pydantic model or a JSON schema.

RETURNS	DESCRIPTION
dict | type[BaseModel] | None	Optional[Dict | Type[BaseModel]]: The schema of the desired output or the model describing it.

Source code in packages/ragbits-core/src/ragbits/core/prompt/prompt.py

def output_schema(self) -> dict | type[BaseModel] | None:
    """
    Returns the schema of the desired output. Can be used to request structured output from the LLM API
    or to validate the output. Can return either a Pydantic model or a JSON schema.

    Returns:
        Optional[Dict | Type[BaseModel]]: The schema of the desired output or the model describing it.
    """
    return self.output_type if issubclass(self.output_type, BaseModel) else None

list_images

list_images() -> list[bytes | str]

Returns the schema of the list of images compatible with LLM APIs Returns: list of dictionaries

Source code in packages/ragbits-core/src/ragbits/core/prompt/prompt.py

def list_images(self) -> list[bytes | str]:
    """
    Returns the schema of the list of images compatible with LLM APIs
    Returns:
        list of dictionaries
    """
    return self.images

parse_response

parse_response(response: str) -> OutputT

Parse the response from the LLM to the desired output type.

PARAMETER	DESCRIPTION
response	The response from the LLM. TYPE: str

RETURNS	DESCRIPTION
OutputT	The parsed response. TYPE: OutputT

RAISES	DESCRIPTION
ResponseParsingError	If the response cannot be parsed.

Source code in packages/ragbits-core/src/ragbits/core/prompt/prompt.py

def parse_response(self, response: str) -> OutputT:
    """
    Parse the response from the LLM to the desired output type.

    Args:
        response (str): The response from the LLM.

    Returns:
        OutputT: The parsed response.

    Raises:
        ResponseParsingError: If the response cannot be parsed.
    """
    return self.response_parser(response)

add_few_shot

add_few_shot(user_message: str | InputT, assistant_message: str | OutputT) -> Prompt[InputT, OutputT]

Add a few-shot example to the conversation.

PARAMETER	DESCRIPTION
user_message	The raw user message or input data that will be rendered using the user prompt template. TYPE: str | InputT
assistant_message	The raw assistant response or output data that will be cast to a string or in case of a Pydantic model, to JSON. TYPE: str | OutputT

RETURNS	DESCRIPTION
Prompt[InputT, OutputT]	Prompt[InputT, OutputT]: The current prompt instance in order to allow chaining.

Source code in packages/ragbits-core/src/ragbits/core/prompt/prompt.py

def add_few_shot(self, user_message: str | InputT, assistant_message: str | OutputT) -> "Prompt[InputT, OutputT]":
    """
    Add a few-shot example to the conversation.

    Args:
        user_message (str | InputT): The raw user message or input data that will be rendered using the
            user prompt template.
        assistant_message (str | OutputT): The raw assistant response or output data that will be cast to a string
            or in case of a Pydantic model, to JSON.

    Returns:
        Prompt[InputT, OutputT]: The current prompt instance in order to allow chaining.
    """
    self._instace_few_shots.append((user_message, assistant_message))
    return self

list_few_shots

list_few_shots() -> ChatFormat

Returns the few shot examples in the standard OpenAI chat format.

RETURNS	DESCRIPTION
ChatFormat	A list of dictionaries, each containing the role and content of a message. TYPE: ChatFormat

Source code in packages/ragbits-core/src/ragbits/core/prompt/prompt.py

def list_few_shots(self) -> ChatFormat:
    """
    Returns the few shot examples in the standard OpenAI chat format.

    Returns:
        ChatFormat: A list of dictionaries, each containing the role and content of a message.
    """
    result: ChatFormat = []
    for user_message, assistant_message in self.few_shots + self._instace_few_shots:
        if not isinstance(user_message, str):
            user_content = self._render_template(self.user_prompt_template, user_message)
        else:
            user_content = user_message

        if isinstance(assistant_message, BaseModel):
            assistant_content = assistant_message.model_dump_json()
        else:
            assistant_content = str(assistant_message)

        result.append({"role": "user", "content": user_content})
        result.append({"role": "assistant", "content": assistant_content})
    return result

to_promptfoo classmethod

to_promptfoo(config: dict[str, Any]) -> ChatFormat

Generate a prompt in the promptfoo format from a promptfoo test configuration.

PARAMETER	DESCRIPTION
config	The promptfoo test configuration. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
ChatFormat	The prompt in the format used by promptfoo. TYPE: ChatFormat

Source code in packages/ragbits-core/src/ragbits/core/prompt/prompt.py

@classmethod
def to_promptfoo(cls, config: dict[str, Any]) -> ChatFormat:
    """
    Generate a prompt in the promptfoo format from a promptfoo test configuration.

    Args:
        config: The promptfoo test configuration.

    Returns:
        ChatFormat: The prompt in the format used by promptfoo.
    """
    return cls(cls.input_type.model_validate(config["vars"])).chat  # type: ignore