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 MemoryInterface.

Parameters:

embedding_model – If set, this includes embeddings in the memory entries which are extremely useful for comparing chat messages and similarities, but also includes overhead.

Methods

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

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

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

Exports all table data using the specified exporter.

Iterates 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

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

get_all_embeddings() → Sequence[EmbeddingDataEntry]

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

get_all_table_models() → list[type[Base]]

Returns 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

Provides a SQLAlchemy session for transactional operations.

Returns:

A SQLAlchemy session bound to the engine.

Return type:

Session

print_schema()

Prints the schema of all tables in the SQLite database.

reset_database() → None

Drops and recreates all tables in the database.