pyrit.memory.SQLiteMemory

class SQLiteMemory(*args, **kwargs)

Bases: MemoryInterface

A memory interface that uses SQLite as the backend database.

This class provides functionality to insert, query, and manage conversation data using SQLite. It supports both file-based and in-memory databases.

Note: this is replacing the old DuckDB implementation.

__init__(*, db_path: Path | str | None = None, verbose: bool = False)

Initialize the SQLiteMemory instance.

Parameters:

• db_path (Optional[Union[Path, str]]) – Path to the SQLite database file. Defaults to “pyrit.db”.

• verbose (bool) – Whether to enable verbose logging. Defaults to False.

Methods

__init__(*[, db_path, verbose])	Initialize the SQLiteMemory instance.
add_attack_results_to_memory(*, attack_results)	Insert a list of attack results into the memory storage.
add_attack_results_to_scenario(*, ...)	Add attack results to an existing scenario result in memory.
add_message_pieces_to_memory(*, message_pieces)	Insert a list of message pieces into the memory storage.
add_message_to_memory(*, request)	Insert a list of message pieces into the memory storage.
add_scenario_results_to_memory(*, ...)	Insert a list of scenario results into the memory storage.
add_scores_to_memory(*, scores)	Insert a list of scores into the memory storage.
add_seed_datasets_to_memory_async(*, ...)	Insert a list of seed datasets into the memory storage.
add_seed_groups_to_memory(*, prompt_groups)	Insert a list of seed groups into the memory storage.
add_seeds_to_memory_async(*, seeds[, added_by])	Insert a list of seeds into the memory storage.
cleanup()	Ensure cleanup on process exit.
disable_embedding()	Disable embedding functionality for the memory interface.
dispose_engine()	Dispose the engine and close all connections.
duplicate_conversation(*, conversation_id[, ...])	Duplicate a conversation for reuse.
duplicate_conversation_excluding_last_turn(*, ...)	Duplicate a conversation, excluding the last turn.
enable_embedding([embedding_model])	Enable embedding functionality for the memory interface.
export_all_tables(*[, export_type])	Export all table data using the specified exporter.
export_conversations(*[, attack_id, ...])	Export conversations and their associated scores from the database to a specified file.
get_all_embeddings()	Fetch all entries from the specified table and returns them as model instances.
get_all_table_models()	Return a list of all table models used in the database by inspecting the Base registry.
get_attack_results(*[, attack_result_ids, ...])	Retrieve a list of AttackResult objects based on the specified filters.
get_chat_messages_with_conversation_id(*, ...)	Return the memory for a given conversation_id.
get_conversation(*, conversation_id)	Retrieve a list of Message objects that have the specified conversation ID.
get_message_pieces(*[, attack_id, role, ...])	Retrieve a list of MessagePiece objects based on the specified filters.
get_prompt_scores(*[, attack_id, role, ...])	Retrieve scores attached to message pieces based on the specified filters.
get_request_from_response(*, response)	Retrieve the request that produced the given response.
get_scenario_results(*[, ...])	Retrieve a list of ScenarioResult objects based on the specified filters.
get_scores(*[, score_ids, score_type, ...])	Retrieve a list of Score objects based on the specified filters.
get_seed_dataset_names()	Return a list of all seed dataset names in the memory storage.
get_seed_groups(*[, value, value_sha256, ...])	Retrieve groups of seed prompts based on the provided filtering criteria.
get_seeds(*[, value, value_sha256, ...])	Retrieve a list of seed prompts based on the specified filters.
get_session()	Provide a SQLAlchemy session for transactional operations.
print_schema()	Print the schema of all tables in the SQLite database.
reset_database()	Drop and recreates all tables in the database.
update_labels_by_conversation_id(*, ...)	Update the labels of prompt entries in memory for a given conversation ID.
update_prompt_entries_by_conversation_id(*, ...)	Update prompt entries for a given conversation ID with the specified field values.
update_prompt_metadata_by_conversation_id(*, ...)	Update the metadata of prompt entries in memory for a given conversation ID.
update_scenario_run_state(*, ...)	Update the run state of an existing scenario result.

Attributes

DEFAULT_DB_FILE_NAME
engine
memory_embedding
results_path
results_storage_io

DEFAULT_DB_FILE_NAME = 'pyrit.db'

add_message_pieces_to_memory(*, message_pieces: Sequence[MessagePiece]) → None

Insert a list of message pieces into the memory storage.

dispose_engine() → None

Dispose the engine and close all connections.

export_all_tables(*, export_type: str = 'json')

Export all table data using the specified exporter.

Iterate over all tables, retrieves their data, and exports each to a file named after the table.

Parameters:

export_type (str) – The format to export the data in (defaults to “json”).

export_conversations(*, attack_id: str | UUID | None = None, conversation_id: str | UUID | None = None, prompt_ids: Sequence[str] | Sequence[UUID] | None = None, labels: dict[str, str] | None = None, sent_after: datetime | None = None, sent_before: datetime | None = None, original_values: Sequence[str] | None = None, converted_values: Sequence[str] | None = None, data_type: str | None = None, not_data_type: str | None = None, converted_value_sha256: Sequence[str] | None = None, file_path: Path | None = None, export_type: str = 'json') → Path

Export conversations and their associated scores from the database to a specified file.

Returns:

The path to the exported file.

Return type:

Path

get_all_embeddings() → Sequence[EmbeddingDataEntry]

Fetch all entries from the specified table and returns them as model instances.

Returns:

A sequence of EmbeddingDataEntry instances representing all stored embeddings.

Return type:

Sequence[EmbeddingDataEntry]

get_all_table_models() → list[type[Base]]

Return a list of all table models used in the database by inspecting the Base registry.

Returns:

A list of SQLAlchemy model classes.

Return type:

list[Base]

get_session() → Session

Provide a SQLAlchemy session for transactional operations.

Returns:

A SQLAlchemy session bound to the engine.

Return type:

Session

print_schema()

Print the schema of all tables in the SQLite database.

reset_database() → None

Drop and recreates all tables in the database.