API Reference

class threadmem.RoleMessage(role: str, text: str, thread_id: str | None = None, images: ~typing.List[str] = <factory>, private: bool | None = False, created: float = <factory>, id: str = <factory>, metadata: dict | None = None)

Bases: WithDB

A role-based chat message that supports text, optional images, and metadata. It can be marked as private and is associated with a specific thread and role. Each message is uniquely identified by an ID and records the creation time.

Parameters:

• role (str) – The role associated with the message.

• text (str) – The text content of the message.

• thread_id (str) – The ID of the thread this message belongs to.

• images (List[str]) – A list of image URLs associated with the message. Defaults to an empty list.

• private (Optional[bool]) – A flag indicating if the message is private. Defaults to False.

• created (float) – The timestamp of when the message was created. Defaults to the current time.

• id (str) – A unique identifier for the message. Defaults to a UUID4 string.

• metadata (Optional[dict]) – Additional metadata associated with the message. Defaults to None.

created: float

classmethod find(**kwargs) → List[RoleMessage]

Finds and returns a list of RoleMessage instances based on the provided search criteria.

This method queries the database for RoleMessage records that match the given keyword arguments. The results are ordered by the creation time of the messages in ascending order.

Parameters:

**kwargs – Arbitrary keyword arguments that are passed to the filter_by method of the database query. These arguments should correspond to the attributes of the RoleMessageRecord model.

Examples

>>> found_messages = RoleMessage.find(role="example_role")

Returns:

A list of RoleMessage instances that match the search criteria.

Return type:

List[RoleMessage]

classmethod from_openai(msg: dict) → RoleMessage

Creates a RoleMessage from an OpenAI response.

classmethod from_record(record: RoleMessageRecord) → RoleMessage

Converts a RoleMessageRecord instance back into a RoleMessage instance.

This method is used to reconstruct a RoleMessage object from its stored representation in the database. It takes a RoleMessageRecord object, which represents a row in the database, and converts it into a RoleMessage object by extracting and converting the stored fields.

Parameters:

record (RoleMessageRecord) – The database record to convert.

Returns:

The reconstructed RoleMessage object.

Return type:

RoleMessage

classmethod from_v1(schema: V1RoleMessage) → RoleMessage

Converts a RoleMessageModel instance into a RoleMessage instance.

This method is used to create a RoleMessage object from a RoleMessageModel schema. It takes a RoleMessageModel object, which represents a structured input, possibly coming from an API request or another external source, and converts it into a RoleMessage object by directly mapping the schema fields to the RoleMessage object fields.

Parameters:

schema (V1RoleMessage) – The schema to convert.

Returns:

The newly created RoleMessage object.

Return type:

RoleMessage

id: str

images: List[str]

metadata: dict | None = None

private: bool | None = False

role: str

save() → None

Saves the current state of the RoleMessage instance to the database.

This method converts the RoleMessage instance into a RoleMessageRecord (the database model representation) and merges it with the existing record in the database if it exists, or creates a new record if it does not. After merging, it commits the changes to the database to ensure the RoleMessage instance is saved.

Raises:

SQLAlchemyError – If there is an issue with database connectivity or the merge and commit operations.

text: str

thread_id: str | None = None

to_openai() → dict

Converts a RoleMessage to the format expected by OpenAI.

to_record() → RoleMessageRecord

Converts the RoleMessage instance into a RoleMessageRecord for database storage.

Returns:

An instance of RoleMessageRecord with fields populated from the RoleMessage instance.

Return type:

RoleMessageRecord

to_v1() → V1RoleMessage

Converts the current RoleMessage instance into a RoleMessageModel.

This method is used to create a RoleMessageModel object from the current RoleMessage instance. It directly maps the fields of the RoleMessage instance to the corresponding fields in the RoleMessageModel, ensuring that the data structure is compatible for serialization or API response purposes.

Returns:

The RoleMessageModel object created from the RoleMessage instance.

Return type:

RoleMessageModel

class threadmem.RoleThread(owner_id: str | None = None, public: bool = False, name: str | None = None, metadata: dict | None = None, remote: str | None = None, version: str | None = None, role_mapping: Dict[str, V1Role] = {})

Bases: WithDB

Represents a role-based chat thread where messages are organized based on roles. Each thread can be public or private, have a unique owner, and contain metadata for additional context. Threads are identified by a unique ID and can be versioned for tracking changes over time.

add_msg(msg: RoleMessage) → None

Add a message to the RoleThread.

This method allows for posting a message to the RoleThread, optionally including images, marking the message as private, and attaching metadata. If the RoleThread is marked as remote, the message is posted to a remote server. Otherwise, it is stored locally.

Parameters:

• role (str) – The role associated with the message.

• msg (RoleMessage) – The role message

Raises:

Exception – If an error occurs while posting the message to a remote server.

add_role(role: V1Role) → None

Adds a new role to the role mapping of the thread.

If the thread is associated with a remote location, the role is posted to the remote server. If the role already exists in the local role mapping, a ValueError is raised. Otherwise, the role is added to the local role mapping and the thread is saved.

Parameters:

role (RoleModel) – The role to be added.

Raises:

• Exception – If there is an issue posting the role to the remote server.

• ValueError – If the role already exists in the role mapping.

copy() → RoleThread

Creates a copy of the current RoleThread instance with a new unique ID but with identical other attributes.

Returns:

A new RoleThread instance that is a copy of the current instance with a new unique ID.

Return type:

RoleThread

delete() → None

Deletes the RoleThread instance from the database.

classmethod find(remote: str | None = None, **kwargs) → List[RoleThread]

Finds RoleThread instances based on various criteria.

This method retrieves RoleThread instances based on the provided keyword arguments. If a remote server is specified, it sends a request to the remote server to fetch the threads. If no remote server is specified, it retrieves the threads from the local database.

Parameters:

• remote (Optional[str]) – The remote server URL to fetch threads from. Defaults to None.

• **kwargs – Additional keyword arguments for filtering the threads.

Example

>>> remote_url = "http://example.com"
>>> owner_id = "user123"
>>> found_threads = RoleThread.find(remote=remote_url, owner_id=owner_id)
>>> print(found_threads)
[RoleThread(owner_id='user123', public=True, name='Thread Name', ...)]

Returns:

A list of RoleThread instances that match the filter criteria.

Return type:

List[RoleThread]

classmethod from_openai(msgs: List[dict]) → RoleThread

classmethod from_record(record: RoleThreadRecord) → RoleThread

Creates an instance of RoleThread from a RoleThreadRecord.

Parameters:

record (RoleThreadRecord) – The database record to convert into a RoleThread instance.

Returns:

An instance of RoleThread with properties populated from the database record.

Return type:

RoleThread

classmethod from_v1(schema: V1RoleThread) → RoleThread

Creates an instance of RoleThread from a RoleThreadModel.

generate_version_hash() → str

Generates a version hash for the RoleThread instance.

This method serializes the RoleThread instance into a JSON string, ensuring the keys are sorted to maintain consistency. It then encodes the JSON string into bytes and computes a SHA-256 hash of these bytes. The resulting hash is used as the version identifier for the RoleThread instance.

Returns:

A hexadecimal string representing the SHA-256 hash of the serialized RoleThread instance.

Return type:

str

property id: str

Retrieves the unique identifier of the RoleThread instance.

This property method returns the unique ID of the RoleThread, which is generated upon the creation of a new RoleThread instance. The ID is a UUID4 string that uniquely identifies the thread among all others.

Returns:

The unique identifier of the RoleThread instance.

Return type:

str

messages(include_private: bool = True) → List[RoleMessage]

Retrieves a list of messages associated with the RoleThread.

This method filters messages based on the include_private parameter. If include_private is True, all messages are returned. If False, only public messages are returned.

Parameters:

include_private (bool, optional) – A flag to determine if private messages should be included. Defaults to True.

Returns:

A list of RoleMessage instances that match the filter criteria.

Return type:

List[RoleMessage]

property metadata: dict | None

Get the metadata of the thread.

property name: str | None

Get the name of the thread.

property owner_id: str | None

Get the owner ID of the thread.

post(role: str, msg: str, images: List[str | Image] = [], private: bool = False, metadata: dict | None = None) → None

Posts a message to the RoleThread.

This method allows for posting a message to the RoleThread, optionally including images, marking the message as private, and attaching metadata. If the RoleThread is marked as remote, the message is posted to a remote server. Otherwise, it is stored locally.

Parameters:

• role (str) – The role associated with the message.

• msg (str) – The message text.

• images (List[str | Image.Image], optional) – A list of images; accepts URLs, PIL images, local filepaths, or b64 encoded images. Defaults to an empty list.

• private (bool, optional) – Whether the message is private. Defaults to False.

• metadata (Optional[dict], optional) – Additional metadata for the message. Defaults to None.

Raises:

Exception – If an error occurs while posting the message to a remote server.

property public: bool

Check if the thread is public.

refresh(auth_token: str | None = None) → None

Refreshes the RoleThread instance from the remote server.

property remote: str | None

Remote status of the thread.

remove_images() → None

Remove all images associated with this role thread.

remove_role(name: str) → None

Removes a role from the role mapping of the thread.

If the thread is associated with a remote location, the role is removed from the remote server. If the role does not exist in the local role mapping, a ValueError is raised. Otherwise, the role is removed from the local role mapping and the thread is saved.

Parameters:

name (str) – The name of the role to be removed.

Raises:

• Exception – If there is an issue removing the role from the remote server.

• ValueError – If the role does not exist in the role mapping.

property role_mapping: Dict[str, V1Role]

save() → None

Saves the RoleThread instance to the database.

This method saves the RoleThread instance to the database. If the thread is remote, it sends a request to the remote server to update or create the thread. If the thread is local, it saves the thread to the database.

to_openai() → List[dict]

to_record() → RoleThreadRecord

Converts the RoleThread instance into a RoleThreadRecord for database storage.

Returns:

An instance of RoleThreadRecord with fields populated from the RoleThread instance.

Return type:

RoleThreadRecord

to_update_schema() → V1UpdateRoleThread

Generates an UpdateRoleThreadModel instance with current thread properties.

This method prepares the data for updating an existing RoleThread by creating an UpdateRoleThreadModel instance. It includes the thread’s name, visibility (public or private), and metadata.

Returns:

An instance populated with the current thread’s name, visibility, and metadata.

Return type:

UpdateRoleThreadModel

to_v1() → V1RoleThread

Converts the RoleThread instance into a RoleThreadModel for API representation.

class threadmem.V1Role(*, name: str, user_id: str, user_name: str, icon: str, description: str | None = None)

Bases: BaseModel

description: str | None

icon: str

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'description': FieldInfo(annotation=Union[str, NoneType], required=False), 'icon': FieldInfo(annotation=str, required=True), 'name': FieldInfo(annotation=str, required=True), 'user_id': FieldInfo(annotation=str, required=True), 'user_name': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

name: str

user_id: str

user_name: str

class threadmem.V1RoleThread(*, owner_id: str | None = None, public: bool, name: str | None = None, metadata: dict | None = None, id: str, messages: List[V1RoleMessage], role_mapping: Dict[str, V1Role] = {}, version: str | None = None, created: float, updated: float, remote: str | None = None)

Bases: BaseModel

created: float

id: str

messages: List[V1RoleMessage]

metadata: dict | None

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'created': FieldInfo(annotation=float, required=True), 'id': FieldInfo(annotation=str, required=True), 'messages': FieldInfo(annotation=List[V1RoleMessage], required=True), 'metadata': FieldInfo(annotation=Union[dict, NoneType], required=False), 'name': FieldInfo(annotation=Union[str, NoneType], required=False), 'owner_id': FieldInfo(annotation=Union[str, NoneType], required=False), 'public': FieldInfo(annotation=bool, required=True), 'remote': FieldInfo(annotation=Union[str, NoneType], required=False), 'role_mapping': FieldInfo(annotation=Dict[str, V1Role], required=False, default={}), 'updated': FieldInfo(annotation=float, required=True), 'version': FieldInfo(annotation=Union[str, NoneType], required=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

name: str | None

owner_id: str | None

public: bool

remote: str | None

role_mapping: Dict[str, V1Role]

updated: float

version: str | None

class threadmem.V1RoleThreads(*, threads: List[V1RoleThread])

Bases: BaseModel

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'threads': FieldInfo(annotation=List[V1RoleThread], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

threads: List[V1RoleThread]